Add test-events command to run detections on JSON inputs

Author: illera88

README.md

@@ -41,14 +41,15 @@ Show available commands and their options:
 
 ```bash
 $ panther_analysis_tool -h
-usage: panther_analysis_tool [-h] [--version] [--debug] [--skip-version-check] {release,test,publish,upload,delete,update-custom-schemas,test-lookup-table,validate,zip,check-connection,benchmark,enrich-test-data} ...
+usage: panther_analysis_tool [-h] [--version] [--debug] [--skip-version-check] {release,test,test-events,publish,upload,delete,update-custom-schemas,test-lookup-table,validate,zip,check-connection,benchmark,enrich-test-data} ...
 
 Panther Analysis Tool: A command line tool for managing Panther policies and rules.
 
 positional arguments:
-  {release,test,publish,upload,delete,update-custom-schemas,test-lookup-table,validate,zip,check-connection,benchmark,enrich-test-data}
+  {release,test,test-events,publish,upload,delete,update-custom-schemas,test-lookup-table,validate,zip,check-connection,benchmark,enrich-test-data}
     release             Create release assets for repository containing panther detections. Generates a file called panther-analysis-all.zip and optionally generates panther-analysis-all.sig
     test                Validate analysis specifications and run policy and rule tests.
+    test-events         Run selected rules/policies against raw event JSON files. Each EVENT_FILE may contain a single JSON object, a JSON array of objects, or NDJSON/JSONL.
     debug               Run a single rule test in a debug environment, which allows you to see print statements and use breakpoints.
     publish             Publishes a new release, generates the release assets, and uploads them. Generates a file called panther-analysis-all.zip and optionally generates panther-analysis-all.sig
     upload              Upload specified policies and rules to a Panther deployment.
@@ -105,6 +106,16 @@ $ panther_analysis_tool test --filter RuleID=AWS.CloudTrail.Stopped --test-names
                 [PASS] [rule] false
 ```
 
+### Test Events
+
+Run a rule/policy against one-or-more raw event JSON files (JSON object, JSON array, or NDJSON/JSONL):
+
+```bash
+$ panther_analysis_tool test-events --path path/to/analysis --filter RuleID=My.RuleID --log-type AWS.CloudTrail ./event.json ./more_events.jsonl
+Event 1 (p_log_type=AWS.CloudTrail)
+  [ALERT] My.RuleID
+```
+
 ### Debug
 
 Run a specific unit test in debug mode:

panther_analysis_tool/event_input.py

@@ -0,0 +1,87 @@
+import json
+import sys
+from dataclasses import dataclass
+from typing import Any, Dict, List, Optional
+
+
+@dataclass(frozen=True)
+class EventSource:
+    """Identifies where events were loaded from (for error reporting)."""
+
+    path: str
+
+
+class EventInputError(ValueError):
+    pass
+
+
+def _ensure_event_dict(obj: Any, source: EventSource) -> Dict[str, Any]:
+    if not isinstance(obj, dict):
+        raise EventInputError(f"{source.path}: expected JSON object, got {type(obj).__name__}")
+    return obj
+
+
+def load_events_from_text(text: str, source: EventSource) -> List[Dict[str, Any]]:
+    """
+    Load one-or-more events from JSON text.
+
+    Supported formats:
+    - A single JSON object: {"k": "v"}
+    - A JSON array of objects: [{"k": "v"}, ...]
+    - NDJSON / JSON Lines: one JSON object per line
+    """
+    stripped = text.strip()
+    if not stripped:
+        return []
+
+    # First try "normal" JSON: object or array.
+    try:
+        parsed = json.loads(stripped)
+        if isinstance(parsed, list):
+            return [_ensure_event_dict(item, source) for item in parsed]
+        return [_ensure_event_dict(parsed, source)]
+    except json.JSONDecodeError:
+        pass
+
+    # Fall back to NDJSON / JSON Lines
+    events: List[Dict[str, Any]] = []
+    for line_no, line in enumerate(text.splitlines(), start=1):
+        if not line.strip():
+            continue
+        try:
+            parsed_line = json.loads(line)
+        except json.JSONDecodeError as exc:
+            raise EventInputError(
+                f"{source.path}:{line_no}: invalid JSON line ({exc.msg})"
+            ) from exc
+        events.append(_ensure_event_dict(parsed_line, source))
+    return events
+
+
+def load_events_from_path(path: str, encoding: str = "utf-8") -> List[Dict[str, Any]]:
+    """
+    Load events from a path, or from stdin when path == "-".
+    """
+    source = EventSource(path=path)
+    if path == "-":
+        return load_events_from_text(sys.stdin.read(), source)
+
+    try:
+        with open(path, "r", encoding=encoding) as handle:
+            return load_events_from_text(handle.read(), source)
+    except OSError as exc:
+        raise EventInputError(f"{path}: unable to read file ({exc})") from exc
+
+
+def apply_log_type(
+    events: List[Dict[str, Any]],
+    log_type: Optional[str],
+    overwrite: bool,
+) -> List[Dict[str, Any]]:
+    if not log_type:
+        return events
+    for event in events:
+        if overwrite or "p_log_type" not in event:
+            event["p_log_type"] = log_type
+    return events
+