chg: [accessibility_tcc] jsonl event based output

analysers/apps.py

@@ -24,18 +24,17 @@ def execute(self):
         Go through all json files in the folder and generate the json list of apps
         '''
         apps = {}
-        # TODO add a check to see if the files exist, and if necessary, call the parsers (or ask the user to call them), or maybe using a flag in the function call
-
         json_data = AccessibilityTccParser(self.config, self.case_id).get_result()
-        if json_data and not json_data.get('error'):
-            for entry in json_data['access']:
-                if entry['client'] not in apps:
-                    apps[entry['client']] = {'found': ['accessibility-tcc'], 'services': [entry['service']]}
-                else:
-                    try:
-                        apps[entry['client']]['services'].append(entry['service'])
-                    except KeyError:
-                        apps[entry['client']]['services'] = [entry['service']]
+        for entry in json_data:
+            if entry['db_table'] != 'access':
+                continue
+            if entry['client'] not in apps:
+                apps[entry['client']] = {'found': ['accessibility-tcc'], 'services': [entry['service']]}
+            else:
+                try:
+                    apps[entry['client']]['services'].append(entry['service'])
+                except KeyError:
+                    apps[entry['client']]['services'] = [entry['service']]
 
         json_data = BrctlParser(self.config, self.case_id).get_result()
         if json_data and not json_data.get('error'):

analysers/timeliner.py

@@ -93,18 +93,20 @@ def __extract_ts_accessibility_tcc(self) -> Generator[dict, None, None]:
         try:
             p = AccessibilityTccParser(self.config, self.case_id)
             data = p.get_result()
-            if 'access' in data.keys():
-                for access in data['access']:
-                    # create timeline entry
-                    timestamp = datetime.fromtimestamp(access['last_modified'], tz=timezone.utc)
-                    ts_event = {
-                        'message': access['service'],
-                        'timestamp': timestamp.timestamp() * 1000000,
-                        'datetime': timestamp.isoformat(),
-                        'timestamp_desc': 'Accessibility TC Last Modified',
-                        'extra_field_1': f"client: {access['client']}"
-                    }
-                    yield ts_event
+
+            for item in data:
+                if item['db_table'] != 'access':
+                    continue
+                # create timeline entry
+                timestamp = datetime.fromtimestamp(item['last_modified'], tz=timezone.utc)
+                ts_event = {
+                    'message': item['service'],
+                    'timestamp': timestamp.timestamp() * 1000000,
+                    'datetime': timestamp.isoformat(),
+                    'timestamp_desc': 'Accessibility TC Last Modified',
+                    'extra_field_1': f"client: {item['client']}"
+                }
+                yield ts_event
         except Exception as e:
             print(f"ERROR while extracting timestamp from accessibility_tcc. Reason {str(e)}")
 

parsers/accessibility_tcc.py

@@ -9,10 +9,12 @@
 import os
 import utils.misc as misc
 from utils.base import BaseParserInterface
+from datetime import datetime, timezone
 
 
 class AccessibilityTccParser(BaseParserInterface):
-    description = "Parsing Accessibility TCC logs"
+    description = 'Parsing Accessibility TCC logs'
+    format = 'jsonl'
 
     def __init__(self, config: dict, case_id: str):
         super().__init__(__file__, config, case_id)
@@ -30,6 +32,27 @@ def get_log_files(self) -> list:
     def execute(self) -> list | dict:
         # only one file to parse
         try:
-            return misc.json_serializable(sqlite2json.sqlite2struct(self.get_log_files()[0]))
+            result = []
+            skipped = set()
+            json_db = misc.json_serializable(sqlite2json.sqlite2struct(self.get_log_files()[0]))
+            for key, values in json_db.items():
+                if 'sqlite_sequence' in key:
+                    continue
+                for value in values:
+                    if 'last_modified' not in value:
+                        skipped.add(key)
+                        continue
+
+                    try:
+                        value['db_table'] = key
+                        value['datetime'] = datetime.fromtimestamp(value['last_modified'], tz=timezone.utc).isoformat()
+                        value['timestamp'] = value['last_modified']
+                        result.append(value)
+                    except TypeError:
+                        # skip "None" values and such
+                        pass
+
+            return result
+
         except IndexError:
-            return {'error': 'No TCC.db file found in logs/Accessibility/ directory'}
+            return []

tests/test_parsers_accessibility_tcc.py

@@ -16,12 +16,9 @@ def test_get_accessibility_tcc(self):
             self.assertTrue(os.path.isfile(p.output_file))
 
             result = p.get_result()
-            self.assertTrue('admin' in result)
-            self.assertTrue('policies' in result)
-            self.assertTrue('active_policy' in result)
-            self.assertTrue('access_overrides' in result)
-            self.assertTrue('expired' in result)
-            self.assertTrue('access' in result)
+            for item in result:
+                self.assertTrue('db_table' in item)
+                self.assertTrue('datetime' in item)
 
 
 if __name__ == '__main__':