Enable high-level interface for reading mmaps and tasks

1 files changed, 159 insertions, 5 deletions

diff --git a/src/focaccia/deterministic.py b/src/focaccia/deterministic.py
index 5a2b411..dfc48a3 100644
--- a/src/focaccia/deterministic.py
+++ b/src/focaccia/deterministic.py
@@ -1,8 +1,7 @@
 """Parsing of JSON files containing snapshot data."""
 
 import os
-import itertools
-from typing import Union, Iterable
+from typing import Union
 
 import brotli
 
@@ -138,6 +137,109 @@ class Event:
 
         return repr_str
 
+class MemoryMapping:
+    def __init__(self,
+                 event_count: int,
+                 start_address: int,
+                 end_address: int,
+                 source: str,
+                 offset: int,
+                 mmap_prot: int,
+                 mmap_flags: int):
+        self.event_count = event_count
+        self.start_address = start_address
+        self.length = end_address - self.start_address
+        self.source = source
+        self.offset = offset
+        self.mmap_prot = mmap_prot
+        self.mmap_flags = mmap_flags
+
+    def __repr__(self) -> str:
+        return f'Memory mapping at event {self.event_count}\n' \
+               f'start = {hex(self.start_address)}\n' \
+               f'length = {self.length}\n' \
+               f'source = {self.source}\n' \
+               f'offset = {self.offset}\n' \
+               f'mmap_prot = {hex(self.mmap_prot)}\n' \
+               f'mmap_flags = {hex(self.mmap_flags)}'
+
+class Task:
+    def __init__(self,
+                 event_count: int,
+                 tid: int):
+        self.event_count = event_count
+        self.tid = tid
+
+    def __repr__(self) -> str:
+        return f'For event index {self.event_count} at tid = {hex(self.tid)}'
+
+class CloneTask(Task):
+    def __init__(self,
+                 event_count: int,
+                 tid: int,
+                 parent_tid: int,
+                 clone_flags: int,
+                 own_namespace_tid: int):
+        super().__init__(event_count, tid)
+        self.parent_tid = parent_tid
+        self.clone_flags = clone_flags
+        self.own_namespace_tid = own_namespace_tid
+
+    def __repr__(self) -> str:
+        repr_str  = f'Clone task\n{super().__repr__()}\n'
+        repr_str += f'parent tid = {hex(self.parent_tid)}\n' \
+                    f'clone flags = {hex(self.clone_flags)}\n' \
+                    f'own namespace tid = {hex(self.own_namespace_tid)}'
+        return repr_str
+
+class ExecTask(Task):
+    def __init__(self,
+                 event_count: int,
+                 tid: int,
+                 filename: str,
+                 commandline: list[str],
+                 execution_base_address: int,
+                 interpreter_base_address: int,
+                 interpreter_name: str):
+        super().__init__(event_count, tid)
+        self.filename = filename
+        self.commandline = commandline
+        self.execution_base_address = execution_base_address
+        self.interpreter_base_address = interpreter_base_address
+        self.interpreter_name = interpreter_name
+
+    def __repr__(self):
+        repr_str  = f'Exec task\n{super().__repr__()}\n'
+        repr_str += f'filename = {self.filename}\n' \
+                    f'command-line = {self.commandline}\n' \
+                    f'execution base address = {hex(self.execution_base_address)}\n' \
+                    f'interpereter base address = {hex(self.interpreter_base_address)}\n' \
+                    f'interpreter name = {self.interpreter_name}'
+        return repr_str
+
+class ExitTask(Task):
+    def __init__(self,
+                 event_count: int,
+                 tid: int,
+                 exit_status: int):
+        super().__init__(event_count, tid)
+        self.exit_status = exit_status
+
+    def __repr__(self):
+        repr_str = f'Exit task\n{super().__repr__()}\n'
+        repr_str += f'exit status = {hex(self.exit_status)}'
+        return repr_str
+
+class DetachTask(Task):
+    def __init__(self,
+                 event_count: int,
+                 tid: int):
+        super().__init__(event_count, tid)
+
+    def __repr__(self):
+        repr_str = f'Detach task\n{super().__repr__()}'
+        return repr_str
+
 class DeterministicLog:
     def __init__(self, log_dir: str):
         self.base_directory = log_dir
@@ -176,13 +278,13 @@ class DeterministicLog:
                 objects.append(deser)
             return objects
 
-    def raw_events(self) -> list[SerializedObject]:
+    def raw_events(self) -> list[Frame]:
         return self._read(self.events_file(), Frame)
 
-    def raw_tasks(self) -> list[SerializedObject]:
+    def raw_tasks(self) -> list[TaskEvent]:
         return self._read(self.tasks_file(), TaskEvent)
 
-    def raw_mmaps(self) -> list[SerializedObject]:
+    def raw_mmaps(self) -> list[MMap]:
         return self._read(self.mmaps_file(), MMap)
 
     def events(self) -> list[Event]:
@@ -224,3 +326,55 @@ class DeterministicLog:
 
         return events
 
+    def tasks(self) -> list[Task]:
+        tasks = []
+        raw_tasks = self.raw_tasks()
+        for raw_task in raw_tasks:
+            task_type = raw_task.which()
+
+            task = None
+            if task_type == 'clone':
+                task = CloneTask(raw_task.frameTime,
+                                 raw_task.tid,
+                                 raw_task.clone.parentTid,
+                                 raw_task.clone.flags,
+                                 raw_task.clone.ownNsTid)
+            if task_type == 'exec':
+                task = ExecTask(raw_task.frameTime,
+                                raw_task.tid,
+                                raw_task.exec.fileName,
+                                raw_task.exec.cmdLine,
+                                raw_task.exec.exeBase,
+                                raw_task.exec.interpBase,
+                                raw_task.exec.interpName)
+            if task_type == 'exit':
+                task = ExitTask(raw_task.frameTime, raw_task.tid, raw_task.exit.exitStatus)
+            if task_type == 'detach':
+                task = DetachTask(raw_task.frameTime, raw_task.tid)
+            tasks.append(task)
+        return tasks
+
+    def mmaps(self) -> list[MemoryMapping]:
+        def mapping_source(mmap: MMap) -> str:
+            source_type = mmap.source.which()
+            if source_type == 'zero' or source_type == 'trace':
+                return source_type
+            elif source_type == 'file':
+                return mmap.source.file.backingFileName
+            else:
+                raise NotImplementedError(f'Unable to handle memory mappings from source type:'
+                                          f' {source_type}')
+
+        mmaps = []
+        raw_mmaps = self.raw_mmaps()
+        for raw_mmap in raw_mmaps:
+            mmap = MemoryMapping(raw_mmap.frameTime,
+                                 raw_mmap.start,
+                                 raw_mmap.end,
+                                 mapping_source(raw_mmap),
+                                 raw_mmap.fileOffsetBytes,
+                                 raw_mmap.prot,
+                                 raw_mmap.flags)
+            mmaps.append(mmap)
+        return mmaps
+