feat(sigma): ship the fifteen sequence detections as Sigma rules

.github/workflows/ci.yml

@@ -56,7 +56,9 @@ jobs:
       - name: Install orchestrator
         run: pip install -e ".[dev]"
       - name: Lint
-        run: ruff check agentmetry tests
+        # tools/ is linted too: the Sigma pack generator lives there and
+        # emits rules a SOC routes on, which is not a place for unlinted code.
+        run: ruff check agentmetry tests tools
       - name: Secret scan
         # The gitleaks action ships a Linux container only.
         if: matrix.os == 'ubuntu-latest'

apps/orchestrator/tests/test_sigma_pack.py

@@ -0,0 +1,126 @@
+"""The Sigma pack ships severities and MITRE ids. Those rot.
+
+`docs/integrations/sigma/agentmetry_rule_*.yml` is generated by
+`tools/generate_sigma_pack.py` from the real rule engine. A SOC that imports the
+pack routes on the `level` and tags in these files, so a severity that changes
+in `rules.py` and not here sends a critical to a low-priority queue and nobody
+finds out until an incident.
+
+`test_readme_claims.py` already solves this shape for the corpus counts. This
+does the same for the rules a customer actually alerts on.
+"""
+
+from __future__ import annotations
+
+import re
+from pathlib import Path
+
+import pytest
+import yaml
+
+from agentmetry.core.audit.detection.rules import BUILTIN_RULE_IDS
+
+SIGMA_DIR = Path(__file__).resolve().parents[3] / "docs" / "integrations" / "sigma"
+
+
+def _generated() -> dict[tuple[str, str], dict]:
+    """Keyed on (rule_id, level).
+
+    One rule id is deliberately two detections: `encoded-command-download` is
+    critical for remote code fetched and executed and low for local content
+    piped into an interpreter. Those get one Sigma rule each, pinned on
+    `action.outcome`, so keying on rule_id alone would silently drop one.
+    """
+    if not SIGMA_DIR.is_dir():
+        pytest.skip(f"sigma pack not found at {SIGMA_DIR}")
+    out = {}
+    for path in SIGMA_DIR.glob("agentmetry_rule_*.yml"):
+        doc = yaml.safe_load(path.read_text(encoding="utf-8"))
+        out[(doc["detection"]["selection"]["detection.rule_id"], doc["level"])] = doc
+    return out
+
+
+def _rule_ids(generated) -> set[str]:
+    return {rid for rid, _ in generated}
+
+
+def test_every_builtin_rule_has_a_sigma_rule():
+    """A rule with no Sigma rule is invisible to every SIEM that imports the pack."""
+    generated = _generated()
+    missing = sorted(BUILTIN_RULE_IDS - _rule_ids(generated))
+    assert not missing, (
+        f"no Sigma rule for: {missing}. Run tools/generate_sigma_pack.py"
+    )
+
+
+def test_no_sigma_rule_for_a_rule_that_does_not_exist():
+    """The reverse: a deleted rule leaving a Sigma file that never fires."""
+    generated = _generated()
+    orphans = sorted(_rule_ids(generated) - BUILTIN_RULE_IDS)
+    assert not orphans, (
+        f"Sigma rules for ids the engine does not emit: {orphans}. "
+        "Delete the file, or the pack promises a detection nobody produces."
+    )
+
+
+def test_severities_match_the_engine():
+    """The number a SOC routes on has to be the number the engine emits.
+
+    Harvested the same way the generator harvests it, so this fails when the
+    generator has not been re-run after a severity change.
+    """
+    import json
+
+    from agentmetry.core.audit.detection.benchmark import DEFAULT_CORPUS_DIR
+    from agentmetry.core.audit.detection.engine import run_detections, run_host_detections
+
+    generated = _generated()
+    for path in sorted(Path(DEFAULT_CORPUS_DIR).glob("*.jsonl")):
+        events = [
+            json.loads(line)
+            for line in path.read_text(encoding="utf-8").splitlines()
+            if line.strip()
+        ]
+        for det in (*run_detections(events), *run_host_detections(events)):
+            doc = generated.get((det.rule_id, det.severity))
+            assert doc is not None, (
+                f"{det.rule_id} fires at {det.severity} and no Sigma rule carries that "
+                "level. Run tools/generate_sigma_pack.py"
+            )
+
+
+def test_every_rule_is_valid_sigma():
+    """Enough of the Sigma shape that an import does not fail on our side."""
+    for (rule_id, _level), doc in _generated().items():
+        assert doc.get("title", "").startswith("Agentmetry - "), rule_id
+        assert doc.get("logsource", {}).get("product") == "agentmetry", rule_id
+        assert doc.get("condition") is None, f"{rule_id}: condition belongs under detection"
+        assert doc["detection"]["condition"] == "selection", rule_id
+        assert doc.get("level") in {"critical", "high", "medium", "low"}, rule_id
+        # Sigma requires a UUID and consumers key on it.
+        assert re.fullmatch(
+            r"[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}",
+            str(doc.get("id", "")),
+        ), f"{rule_id}: id is not a UUID"
+
+
+def test_rule_ids_are_stable_across_regeneration():
+    """Regenerating must not churn every UUID.
+
+    The ids are derived from the rule id rather than randomly, so a consumer who
+    pinned a Sigma id keeps working after somebody reruns the generator.
+    """
+    import hashlib
+    import uuid
+
+    generated = _generated()
+    counts = {}
+    for rid, _ in generated:
+        counts[rid] = counts.get(rid, 0) + 1
+    for (rule_id, level), doc in generated.items():
+        # Severity joins the seed only for a split rule, so unsplit ids never move.
+        seed = f"agentmetry.sigma.{rule_id}" + (f".{level}" if counts[rule_id] > 1 else "")
+        expected = str(uuid.UUID(bytes=hashlib.sha256(seed.encode()).digest()[:16], version=5))
+        assert doc["id"] == expected, (
+            f"{rule_id}: id is not the derived one. Regenerating would change it."
+        )

apps/orchestrator/tools/generate_sigma_pack.py

@@ -0,0 +1,198 @@
+"""Emit a Sigma rule per built-in sequence detection, into docs/integrations/sigma/.
+
+Run from the orchestrator:
+
+    .venv/Scripts/python.exe tools/generate_sigma_pack.py
+
+## Why generated rather than written
+
+The Sigma pack that already existed covers the recorder's own health: heartbeat
+silence, degraded coverage, MCP schema drift, denial bursts. Useful, and none of
+it is the product. The fifteen sequence detections, which are what a SOC would
+actually route on, had no Sigma representation at all, so a Splunk or Sentinel
+team wanting to alert on `credential-exfil` had to write the search themselves
+from a schema document.
+
+Hand-writing fifteen would put rule ids, severities and MITRE ids in a second
+place that drifts from the first. So the metadata is harvested by replaying the
+benchmark corpus through the real engine and reading the `Detection` objects it
+emits. A severity that changes in `rules.py` changes here on the next run, and
+`tests/test_sigma_pack.py` fails if somebody forgets to run it.
+
+## The two the corpus cannot produce
+
+`host-subagent-swarm-burst` is host-scoped and needs several sessions on one
+host inside a window. `off-hours-activity` is opt-in behind
+`AGENTMETRY_DETECT_OFF_HOURS` and needs an operator-set window. Neither has a
+corpus case (issue #36), so both carry an explicit entry below.
+
+That table is checked against `BUILTIN_RULE_IDS` rather than trusted: adding a
+sixteenth rule without either a corpus case or an entry here fails this script,
+rather than silently shipping fourteen rules in a pack that claims fifteen.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import sys
+import uuid
+from pathlib import Path
+
+from agentmetry.core.audit.detection.benchmark import DEFAULT_CORPUS_DIR
+from agentmetry.core.audit.detection.engine import run_detections, run_host_detections
+from agentmetry.core.audit.detection.rules import BUILTIN_RULE_IDS
+
+OUT = Path(__file__).resolve().parents[2].parent / "docs" / "integrations" / "sigma"
+
+# Sigma's `level` vocabulary happens to match ours one for one. Mapped
+# explicitly anyway, so a new severity fails here rather than emitting a level
+# no SIEM recognises.
+_LEVEL = {"critical": "critical", "high": "high", "medium": "medium", "low": "low"}
+
+# The two rules no corpus case can produce. Severity comes from
+# docs/detection-rules.md, the same table the README renders.
+_UNCOVERED: dict[str, dict] = {
+    "host-subagent-swarm-burst": {
+        "title": "Subagent swarm across sessions on one host",
+        "severity": "high",
+        "tactics": ["TA0002"],
+        "techniques": ["T1059"],
+        "note": "Host-scoped: needs several sessions on one host inside a window.",
+    },
+    "off-hours-activity": {
+        "title": "Autonomous impact action outside business hours",
+        "severity": "medium",
+        "tactics": ["TA0040"],
+        "techniques": ["T1485"],
+        "note": "Opt-in behind AGENTMETRY_DETECT_OFF_HOURS with an operator-set window.",
+    },
+}
+
+
+def _harvest() -> dict[tuple[str, str], dict]:
+    """Replay the corpus and read the Detection objects the engine emits.
+
+    Keyed on `(rule_id, severity)` rather than `rule_id`, because one rule id is
+    deliberately two detections. `encoded-command-download` returns `critical`
+    for remote code fetched and executed and `low` for local content piped into
+    an interpreter, with a comment in `rules.py` saying the low one exists "so
+    it stops drowning the criticals". A single Sigma `level` would either page
+    on the quiet variant or stay silent on the loud one, so each severity gets
+    its own rule and the selection pins `action.outcome`.
+    """
+    found: dict[tuple[str, str], dict] = {}
+    for path in sorted(Path(DEFAULT_CORPUS_DIR).glob("*.jsonl")):
+        events = [
+            json.loads(line)
+            for line in path.read_text(encoding="utf-8").splitlines()
+            if line.strip()
+        ]
+        for det in (*run_detections(events), *run_host_detections(events)):
+            found.setdefault(
+                (det.rule_id, det.severity),
+                {
+                    "title": det.title,
+                    "severity": det.severity,
+                    "tactics": sorted(set(det.tactic_ids)),
+                    "techniques": sorted(set(det.technique_ids)),
+                    "note": "",
+                },
+            )
+    return found
+
+
+def _stable_uuid(rule_id: str, severity: str = "") -> str:
+    """A deterministic UUID, so regenerating does not churn every id.
+
+    Sigma requires a UUID and consumers key on it, so running this twice must
+    produce no diff. Severity joins the seed only for a split rule, which keeps
+    the fourteen unsplit ids exactly where they are.
+    """
+    seed = f"agentmetry.sigma.{rule_id}" + (f".{severity}" if severity else "")
+    return str(uuid.UUID(bytes=hashlib.sha256(seed.encode()).digest()[:16], version=5))
+
+
+def _yaml(rule_id: str, meta: dict, *, split: bool) -> str:
+    severity = meta["severity"]
+    tags = [f"attack.{t.lower()}" for t in meta["tactics"] + meta["techniques"]]
+    tag_lines = "\n".join(f"  - {t}" for t in tags) or "  []"
+    note = f"\n  {meta['note']}" if meta.get("note") else ""
+    outcome = f"\n    action.outcome: {severity}" if split else ""
+    variant = (
+        f"\n  This rule id emits more than one severity. This is the `{severity}`"
+        " variant, pinned on `action.outcome`."
+        if split
+        else ""
+    )
+    return f"""title: Agentmetry - {meta["title"]}
+id: {_stable_uuid(rule_id, severity if split else "")}
+status: experimental
+description: |
+  Agentmetry sequence detection `{rule_id}` fired. The individual tool calls are
+  ordinary; the order is the finding, and `detection.event_ids` names the exact
+  events that produced it.{note}{variant}
+
+  GENERATED by tools/generate_sigma_pack.py. Do not hand-edit: severity and
+  MITRE ids are read from the rule engine, so an edit here is a claim that
+  disagrees with the code.
+references:
+  - https://github.com/blitzcrieg1/agentmetry/blob/master/docs/detection-rules.md
+author: Agentmetry
+logsource:
+  product: agentmetry
+  service: audit
+detection:
+  selection:
+    action.type: detection
+    detection.rule_id: {rule_id}{outcome}
+  condition: selection
+falsepositives:
+  - Legitimate automation that reproduces the sequence. Disposition it in the
+    trail rather than deleting the rule, so the decision is recorded.
+level: {_LEVEL[severity]}
+tags:
+{tag_lines}
+"""
+
+
+def main() -> int:
+    harvested = _harvest()
+    for rule_id, entry in _UNCOVERED.items():
+        harvested.setdefault((rule_id, entry["severity"]), entry)
+
+    covered = {rid for rid, _ in harvested}
+    missing = sorted(BUILTIN_RULE_IDS - covered)
+    if missing:
+        print("No corpus case and no _UNCOVERED entry for: " + ", ".join(missing), file=sys.stderr)
+        print("Add a corpus case (preferred, see issue #36) or an entry here.", file=sys.stderr)
+        return 1
+
+    extra = sorted(covered - BUILTIN_RULE_IDS)
+    if extra:
+        # A rule that fired in the corpus but is not built in means a YAML rule
+        # leaked into the run, which would ship an operator's local rule as ours.
+        print("Not a built-in rule: " + ", ".join(extra), file=sys.stderr)
+        return 1
+
+    severities: dict[str, set[str]] = {}
+    for rid, sev in harvested:
+        severities.setdefault(rid, set()).add(sev)
+
+    OUT.mkdir(parents=True, exist_ok=True)
+    for rule_id, severity in sorted(harvested):
+        split = len(severities[rule_id]) > 1
+        stem = rule_id.replace("-", "_") + (f"_{severity}" if split else "")
+        (OUT / f"agentmetry_rule_{stem}.yml").write_text(
+            _yaml(rule_id, harvested[(rule_id, severity)], split=split), encoding="utf-8"
+        )
+
+    split_ids = sorted(r for r, s in severities.items() if len(s) > 1)
+    print(f"wrote {len(harvested)} Sigma rules for {len(BUILTIN_RULE_IDS)} rule ids -> {OUT}")
+    if split_ids:
+        print(f"  split by severity: {', '.join(split_ids)}")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

apps/orchestrator/tools/mcp_audit_proxy.py

@@ -204,7 +204,10 @@ async def run_proxy(command: list[str], server_name: str) -> int:
         stdout=asyncio.subprocess.PIPE,
         stderr=asyncio.subprocess.PIPE,
     )
-    assert proc.stdin and proc.stdout and proc.stderr
+    # noqa S101: narrowing for the type checker, not a runtime check.
+    # create_subprocess_exec with PIPE on all three always sets them, and
+    # python -O would strip this without changing behaviour.
+    assert proc.stdin and proc.stdout and proc.stderr  # noqa: S101
 
     pending: dict[str, dict[str, str]] = {}
     list_buf = ToolsListBuffer()