From 6cf3214a2503cb90931a1a691bce6e756d6e301e Mon Sep 17 00:00:00 2001 From: Dan Cohen Vaxman Date: Wed, 15 Jul 2026 15:02:32 +0200 Subject: [PATCH] Add Kibana JSON and Sigma exporters, with an example rule --- .../windows_suspicious_scheduled_task.yaml | 24 ++++++ src/detection_as_code/exporters.py | 77 +++++++++++++++++++ tests/test_exporters.py | 59 ++++++++++++++ 3 files changed, 160 insertions(+) create mode 100644 examples/rules/windows_suspicious_scheduled_task.yaml create mode 100644 src/detection_as_code/exporters.py create mode 100644 tests/test_exporters.py diff --git a/examples/rules/windows_suspicious_scheduled_task.yaml b/examples/rules/windows_suspicious_scheduled_task.yaml new file mode 100644 index 0000000..430fadd --- /dev/null +++ b/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 diff --git a/src/detection_as_code/exporters.py b/src/detection_as_code/exporters.py new file mode 100644 index 0000000..336485b --- /dev/null +++ b/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) diff --git a/tests/test_exporters.py b/tests/test_exporters.py new file mode 100644 index 0000000..d6a0e3d --- /dev/null +++ b/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)