fix: event.pop("_timestamp", None)

Author: scc-tw

linux_edr/app.py

@@ -332,8 +332,30 @@ def _process_event(self, evt: str) -> None:
             if self.verbose_debug:
                 self._log_parsed_event(evt)
 
-        # Add event to aggregator
-        self.agg.add(evt)
+        # If event is already a dict (e.g., when injected by tests or future extensions),
+        # we assume it's been validated and directly buffer it.
+        if isinstance(evt, dict):
+            self.agg.add(evt)
+            return
+
+        # Otherwise, treat it as raw text from trace_pipe and try to parse/validate.
+        parsed = parse_execve(evt)
+
+        if not parsed:
+            # Not an execve line – skip buffering
+            return
+
+        # Validate with Pydantic schema (ensures correct types/structure)
+        try:
+            from .domain.models.event_models import ExecveEvent as ExecveEventModel  # Local import to avoid cycles
+
+            model_event = ExecveEventModel.from_namedtuple(parsed)
+
+            # Buffer as plain dict (safer for serialization & downstream processing)
+            self.agg.add(model_event.model_dump())
+        except Exception as e:
+            # Any validation or conversion error – log and drop the event
+            logging.warning("Invalid event skipped: %s", e)
 
     def _log_parsed_event(self, evt: str) -> None:
         """

linux_edr/domain/models/event_models.py

@@ -0,0 +1,21 @@
+from pydantic import BaseModel, Field
+from typing import List, Any
+
+class ExecveEvent(BaseModel):
+    """Validated execve syscall event."""
+
+    timestamp: str = Field(..., description="Trace timestamp string as reported by ftrace")
+    pid: int = Field(..., description="Process ID of the task that executed the syscall")
+    command: str = Field(..., description="Executable invoked (basename)")
+    args: List[str] = Field(default_factory=list, description="Arguments supplied to the executable")
+
+    # Helper constructors ---------------------------------------------------
+    @classmethod
+    def from_namedtuple(cls, nt: Any) -> "ExecveEvent":
+        """Build ExecveEvent from an ExecveEvent NamedTuple instance."""
+        return cls(timestamp=nt.timestamp, pid=nt.pid, command=nt.command, args=list(nt.args))
+
+    def __str__(self) -> str:  # pragma: no cover – convenience only
+        """Human-readable representation useful in logs."""
+        cmd_line = " ".join([self.command, *self.args])
+        return f"[{self.timestamp}] pid={self.pid} cmd={cmd_line}" 

tests/test_app.py

@@ -8,10 +8,11 @@
     parse_execve,
     setup_logging,
     LinuxEDRApp,
-    ExecveEvent,
     SyscallTracer,
 )
 
+from linux_edr.domain.models.event_models import ExecveEvent
+
 
 class TestApp(unittest.TestCase):
 
@@ -259,7 +260,7 @@ def test_process_event(self, mock_parse_execve, mock_log_info, mock_log_debug):
         app.agg = MagicMock()
 
         # Setup mock for parse_execve
-        parsed_event = ExecveEvent("12345.6789", 1000, "test_cmd", ["-a", "-b"])
+        parsed_event = ExecveEvent(timestamp="12345.6789", pid=1000, command="test_cmd", args=["-a", "-b"])
         mock_parse_execve.return_value = parsed_event
 
         # Import the method to test it independently
@@ -268,23 +269,28 @@ def test_process_event(self, mock_parse_execve, mock_log_info, mock_log_debug):
         # Call the method directly
         LinuxEDRApp._process_event(app, "test_event")
 
-        # Verify debug logging
-        app.agg.add.assert_called_once_with("test_event")
-        self.assertTrue(mock_log_debug.called)
+        # The aggregator should receive a validated dict version of the parsed event
+        expected_dict = {
+            "timestamp": "12345.6789",
+            "pid": 1000,
+            "command": "test_cmd",
+            "args": ["-a", "-b"],
+        }
+        app.agg.add.assert_called_once_with(expected_dict)
 
         # Reset mock and test with verbose_debug=False
         mock_log_debug.reset_mock()
         app.verbose_debug = False
 
         LinuxEDRApp._process_event(app, "test_event2")
-        app.agg.add.assert_called_with("test_event2")
+        app.agg.add.assert_called_with(expected_dict)
 
         # Test with debug=False
         mock_log_debug.reset_mock()
         app.debug = False
 
         LinuxEDRApp._process_event(app, "test_event3")
-        app.agg.add.assert_called_with("test_event3")
+        app.agg.add.assert_called_with(expected_dict)
         mock_log_debug.assert_not_called()