Merge pull request #183 from EC-DIGIT-CSIRC/event_dataclass

wip: [datamodel] datamodel has 'data' subkey

Author: dario-br

pyproject.toml

@@ -35,6 +35,7 @@ dependencies = [
   "python-magic==0.4.27",
   "jinja2==3.1.6",
   "matplotlib==3.10.1",
+  "orjsonl==1.0.0"
 ]
 
 

src/sysdiagnose/analysers/apps.py

@@ -4,7 +4,7 @@
 # Author: Emiliern Le Jamtel
 
 import re
-from sysdiagnose.utils.base import BaseAnalyserInterface, logger
+from sysdiagnose.utils.base import BaseAnalyserInterface, SysdiagnoseConfig, logger
 from sysdiagnose.parsers.accessibility_tcc import AccessibilityTccParser
 from sysdiagnose.parsers.brctl import BrctlParser
 from sysdiagnose.parsers.itunesstore import iTunesStoreParser
@@ -15,7 +15,7 @@ class AppsAnalyser(BaseAnalyserInterface):
     description = 'Get list of Apps installed on the device'
     format = 'json'
 
-    def __init__(self, config: dict, case_id: str):
+    def __init__(self, config: SysdiagnoseConfig, case_id: str):
         super().__init__(__file__, config, case_id)
 
     # this code is quite slow, but that's due to logarchive.jsonl being slow to parse
@@ -26,7 +26,7 @@ def execute(self):
         apps = {}
         json_data = AccessibilityTccParser(self.config, self.case_id).get_result()
         for entry in json_data:
-            apps[entry['client']] = {'found': ['accessibility-tcc'], 'services': [entry['service']]}
+            apps[entry['data']['client']] = {'found': ['accessibility-tcc'], 'services': [entry['data']['service']]}
 
         json_data = BrctlParser(self.config, self.case_id).get_result()
         if json_data and not json_data.get('error'):
@@ -55,7 +55,8 @@ def execute(self):
         re_bundle_id_pattern = r'(([a-zA-Z0-9-_]+\.)+[a-zA-Z0-9-_]+)'
         # list files in here
         json_entries = LogarchiveParser(self.config, self.case_id).get_result()
-        for entry in json_entries:
+        for events in json_entries:
+            entry = events['data']
             try:
                 # skip empty entries
                 if entry['subsystem'] == '':

src/sysdiagnose/analysers/coverage.py

@@ -3,15 +3,15 @@
 import importlib
 import os
 import magic
-from sysdiagnose.utils.base import BaseAnalyserInterface, BaseParserInterface, logger
+from sysdiagnose.utils.base import BaseAnalyserInterface, BaseParserInterface, SysdiagnoseConfig, logger
 from sysdiagnose.parsers.remotectl_dumpstate import RemotectlDumpstateParser
 
 
 class CoverageAnalyser(BaseAnalyserInterface):
     description = "Provides parser coverage information"
     format = "html"
 
-    def __init__(self, config: dict, case_id: str):
+    def __init__(self, config: SysdiagnoseConfig, case_id: str):
         super().__init__(__file__, config, case_id)
 
     def execute(self):

src/sysdiagnose/analysers/demo_analyser.py

@@ -3,14 +3,14 @@
 # For Python3
 # DEMO - Skeleton
 
-from sysdiagnose.utils.base import BaseAnalyserInterface, logger
+from sysdiagnose.utils.base import BaseAnalyserInterface, SysdiagnoseConfig, logger
 
 
 class DemoAnalyser(BaseAnalyserInterface):
     description = "Do something useful (DEMO)"
     # format = "json"  # by default json
 
-    def __init__(self, config: dict, case_id: str):
+    def __init__(self, config: SysdiagnoseConfig, case_id: str):
         super().__init__(__file__, config, case_id)
 
     def execute(self):

src/sysdiagnose/analysers/file_stats.py

@@ -1,14 +1,14 @@
 import os
 import magic
 from sysdiagnose.parsers.remotectl_dumpstate import RemotectlDumpstateParser
-from sysdiagnose.utils.base import BaseAnalyserInterface, logger
+from sysdiagnose.utils.base import BaseAnalyserInterface, SysdiagnoseConfig, logger
 
 
 class FileStatisticsAnalyser(BaseAnalyserInterface):
     description = "Obatins statistics about the files of the sysdiagnose"
     format = "json"  # Output format
 
-    def __init__(self, config: dict, case_id: str):
+    def __init__(self, config: SysdiagnoseConfig, case_id: str):
         super().__init__(__file__, config, case_id)
 
     def get_file_type(self, file_path: str) -> str:

src/sysdiagnose/analysers/mcstateshared.py

@@ -1,4 +1,4 @@
-from sysdiagnose.utils.base import BaseAnalyserInterface
+from sysdiagnose.utils.base import BaseAnalyserInterface, SysdiagnoseConfig
 from sysdiagnose.parsers.mcstate_shared_profile import McStateSharedProfileParser
 from sysdiagnose.utils import misc
 import csv
@@ -9,7 +9,7 @@ class MCStateSharedProfileAnalyser(BaseAnalyserInterface):
     description = "Exports MCState Shared Profile stub files to CSV for better analysis."
     format = "csv"
 
-    def __init__(self, config: dict, case_id: str):
+    def __init__(self, config: SysdiagnoseConfig, case_id: str):
         super().__init__(__file__, config, case_id)
 
     def execute(self):
@@ -19,11 +19,11 @@ def execute(self):
 
         for entry in mcstate_result:
             for key in ['SignerCerts', 'datetime', 'timestamp']:
-                entry.pop(key, None)
+                entry['data'].pop(key, None)
 
-            payload_contents = entry.pop('PayloadContent', None)
+            payload_contents = entry['data'].pop('PayloadContent', None)
 
-            entry_tpl = misc.flatten_dict(entry)
+            entry_tpl = misc.flatten_dict(entry['data'])
 
             for payload_content in payload_contents:
                 item = entry_tpl.copy()

src/sysdiagnose/analysers/plist.py

@@ -1,7 +1,8 @@
+from datetime import datetime
 import json
 from typing import Generator
 
-from sysdiagnose.utils.base import BaseAnalyserInterface, logger
+from sysdiagnose.utils.base import BaseAnalyserInterface, SysdiagnoseConfig, logger, Event
 from sysdiagnose.parsers.plists import PlistParser
 
 
@@ -14,7 +15,7 @@ class PListAnalyzer(BaseAnalyserInterface):
     description = 'Gathers information from a plist file.'
     format = 'jsonl'
 
-    def __init__(self, config: dict, case_id: str):
+    def __init__(self, config: SysdiagnoseConfig, case_id: str):
         super().__init__(__file__, config, case_id)
         self.parser = PlistParser(config, case_id)
 
@@ -53,14 +54,15 @@ def __extract_plist_mdm_data(self) -> Generator[dict, None, None]:
                 for line in f:
                     entry = json.loads(line)
 
-                    mdm_entry: dict[str, str] = {
-                        'ManagingProfileIdentifier': entry.get('ManagingProfileIdentifier'),
-                        'AccessRights': entry.get('AccessRights'),
-                        'LastPollingAttempt': entry.get('LastPollingAttempt'),
-                        'source': entity_type,
-                    }
+                    mdm_entry = Event(
+                        datetime=datetime.fromisoformat(entry.get('LastPollingAttempt')),
+                        message= f"MDM Profile: {entry.get('ManagingProfileIdentifier')} with access rights {entry.get('AccessRights')}",
+                        timestamp_desc='Last Polling Attempt',
+                        module=self.module_name,
+                        data={'source': entity_type}
+                    )
 
-                    yield mdm_entry
+                    yield mdm_entry.to_dict()
 
         except FileNotFoundError as e:
             logger.warning(f'{entity_type} not found for {self.case_id}. {e}')