feat: Add Kibana JSON and Sigma exporters, with an example rule

examples/rules/windows_suspicious_scheduled_task.yaml

@@ -0,0 +1,24 @@
+# Example rule definition. This is the unit of work in this repo: author one
+# of these, then run `detection-as-code lint|export|review|deploy` against it.
+name: windows_suspicious_scheduled_task_creation
+platform: windows
+description: >
+  Detects creation of a scheduled task via schtasks.exe with cmd.exe as the
+  parent process - a common persistence and privilege-escalation technique
+  used to survive reboots or execute with a different user's context.
+severity: high
+risk_score: 62
+index_patterns:
+  - "winlogbeat-*"
+query: 'process.name:"schtasks.exe" and process.parent.name:"cmd.exe"'
+tactics:
+  - persistence
+  - privilege_escalation
+techniques:
+  - T1053.005
+false_positives:
+  - Legitimate software installers registering scheduled maintenance tasks
+  - IT administration scripts run from approved deployment tooling
+author: detection-engineering
+tags:
+  - maturity:staging

src/detection_as_code/exporters.py

@@ -0,0 +1,77 @@
+from __future__ import annotations
+
+import json
+import re
+import uuid
+
+import yaml
+
+from .models import DetectionRule
+
+_SIMPLE_CLAUSE = re.compile(r'^\s*([\w.]+)\s*:\s*"?([^"]+?)"?\s*$')
+
+
+class UnsupportedQueryError(ValueError):
+    """Raised when a query is too complex for the mechanical Sigma converter.
+
+    Sigma's `detection` block expects structured field:value selections. Kuery
+    supports arbitrary boolean grouping this converter does not attempt to
+    reason about. Refusing loudly beats silently emitting a Sigma rule with
+    different match semantics than the source query.
+    """
+
+
+def _parse_simple_conjunction(query: str) -> dict[str, str]:
+    if " or " in query.lower() or "(" in query or ")" in query:
+        raise UnsupportedQueryError(
+            "Query uses OR/grouping - convert to Sigma by hand, "
+            "a mechanical conversion would risk changing match semantics"
+        )
+    fields: dict[str, str] = {}
+    for clause in re.split(r"\s+and\s+", query.strip(), flags=re.IGNORECASE):
+        match = _SIMPLE_CLAUSE.match(clause)
+        if not match:
+            raise UnsupportedQueryError(f"Cannot mechanically translate clause: '{clause}'")
+        fields[match.group(1)] = match.group(2)
+    return fields
+
+
+def to_sigma(rule: DetectionRule) -> str:
+    """Export to Sigma YAML. Only supports simple `field:value AND field:value`
+    Kuery - see UnsupportedQueryError for why anything richer is rejected
+    rather than guessed at.
+    """
+    selection = _parse_simple_conjunction(rule.query)
+    sigma_rule = {
+        "title": rule.name,
+        "id": rule.rule_id or str(uuid.uuid4()),
+        "status": "stable" if rule.enabled else "experimental",
+        "description": rule.description,
+        "logsource": {"category": rule.platform.lower()},
+        "detection": {"selection": selection, "condition": "selection"},
+        "falsepositives": rule.false_positives or ["Unknown"],
+        "level": rule.severity.value,
+        "tags": [f"attack.{tactic.name.lower()}" for tactic in rule.tactics] + rule.tags,
+    }
+    return yaml.safe_dump(sigma_rule, sort_keys=False)
+
+
+def to_kibana_json(rule: DetectionRule) -> str:
+    payload = {
+        "rule_id": rule.rule_id or str(uuid.uuid4()),
+        "name": rule.name,
+        "description": rule.description,
+        "risk_score": rule.risk_score,
+        "severity": rule.severity.value,
+        "type": "query",
+        "query": rule.query,
+        "language": "kuery",
+        "index": rule.index_patterns,
+        "interval": rule.interval,
+        "from": f"now-{rule.lookback}",
+        "tags": [f"tactic:{tactic.name.lower()}" for tactic in rule.tactics] + rule.tags,
+        "false_positives": rule.false_positives,
+        "author": [rule.author] if rule.author else [],
+        "enabled": rule.enabled,
+    }
+    return json.dumps(payload, indent=2)

tests/test_exporters.py

@@ -0,0 +1,59 @@
+import json
+
+import pytest
+import yaml
+
+from detection_as_code.exporters import UnsupportedQueryError, to_kibana_json, to_sigma
+from detection_as_code.models import DetectionRule
+
+SIMPLE_RULE = DetectionRule.from_dict(
+    {
+        "name": "windows_suspicious_scheduled_task",
+        "platform": "windows",
+        "query": 'process.name:"schtasks.exe" and process.parent.name:"cmd.exe"',
+        "index_patterns": ["winlogbeat-*"],
+        "description": "Detects scheduled task creation from a suspicious parent process.",
+        "severity": "high",
+        "risk_score": 62,
+        "tactics": ["persistence"],
+        "techniques": ["T1053.005"],
+        "false_positives": ["Legitimate installer scheduling maintenance tasks"],
+    }
+)
+
+
+def test_to_kibana_json_round_trips():
+    payload = json.loads(to_kibana_json(SIMPLE_RULE))
+
+    assert payload["name"] == SIMPLE_RULE.name
+    assert payload["risk_score"] == 62
+    assert payload["index"] == ["winlogbeat-*"]
+    assert "tactic:persistence" in payload["tags"]
+
+
+def test_to_sigma_converts_simple_conjunction():
+    sigma_yaml = to_sigma(SIMPLE_RULE)
+    parsed = yaml.safe_load(sigma_yaml)
+
+    assert parsed["detection"]["selection"] == {
+        "process.name": "schtasks.exe",
+        "process.parent.name": "cmd.exe",
+    }
+    assert parsed["logsource"]["category"] == "windows"
+    assert "attack.persistence" in parsed["tags"]
+
+
+def test_to_sigma_rejects_or_queries():
+    rule = DetectionRule.from_dict(
+        {
+            "name": "or_query_rule",
+            "platform": "windows",
+            "query": 'process.name:"a.exe" or process.name:"b.exe"',
+            "index_patterns": ["winlogbeat-*"],
+            "description": "Has an OR clause that cannot be mechanically translated.",
+            "severity": "low",
+            "risk_score": 20,
+        }
+    )
+    with pytest.raises(UnsupportedQueryError):
+        to_sigma(rule)