feat: Add packaging + vendor-neutral DetectionRule schema and secret provider abstraction (#2)

* Remove legacy interactive scripts, superseded by the detection_as_code package

* Add packaging + vendor-neutral DetectionRule schema and secret provider abstraction

Author: SentinelByte

.gitignore

@@ -145,6 +145,9 @@ venv.bak/
 .dmypy.json
 dmypy.json
 
+# ruff
+.ruff_cache/
+
 # Pyre type checker
 .pyre/
 

pyproject.toml

@@ -0,0 +1,58 @@
+[build-system]
+requires = ["setuptools>=68", "wheel"]
+build-backend = "setuptools.build_meta"
+
+[project]
+name = "detection-as-code"
+version = "0.2.0"
+description = "A schema-first detection engineering pipeline: author rules once, lint them deterministically, export to Kibana or Sigma, and get an advisory AI review before they ship."
+readme = "README.md"
+requires-python = ">=3.10"
+license = { file = "LICENSE" }
+authors = [{ name = "SentinelByte" }]
+dependencies = [
+    "requests>=2.31",
+    "elasticsearch>=8.13,<9",
+    "PyYAML>=6.0",
+]
+
+[project.optional-dependencies]
+ai = ["anthropic>=0.34"]
+dev = [
+    "pytest>=8.0",
+    "pytest-cov>=5.0",
+    "ruff>=0.5",
+    "mypy>=1.10",
+    "bandit>=1.7",
+    "pip-audit>=2.7",
+    "types-requests",
+    "types-PyYAML",
+    "anthropic>=0.34",
+]
+
+[project.scripts]
+detection-as-code = "detection_as_code.cli:main"
+
+[project.urls]
+Repository = "https://github.com/SentinelByte/detection-as-code"
+
+[tool.setuptools.packages.find]
+where = ["src"]
+
+[tool.ruff]
+line-length = 100
+target-version = "py310"
+
+[tool.ruff.lint]
+select = ["E", "F", "I", "UP", "B"]
+
+[tool.mypy]
+python_version = "3.10"
+ignore_missing_imports = true
+disallow_untyped_defs = false
+
+[tool.pytest.ini_options]
+testpaths = ["tests"]
+
+[tool.bandit]
+exclude_dirs = ["tests"]

src/detection_as_code/__init__.py

@@ -0,0 +1,11 @@
+"""A schema-first detection engineering pipeline.
+
+Rules are authored once as declarative YAML, checked deterministically,
+exported to whichever backend needs them, and optionally reviewed by an
+LLM before a human ships them. See docs/architecture.md for the full flow.
+"""
+
+from .models import DetectionRule, Severity, Tactic
+
+__all__ = ["DetectionRule", "Severity", "Tactic"]
+__version__ = "0.2.0"

src/detection_as_code/models.py

@@ -0,0 +1,97 @@
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from enum import Enum
+from typing import Any
+
+
+class Severity(str, Enum):
+    LOW = "low"
+    MEDIUM = "medium"
+    HIGH = "high"
+    CRITICAL = "critical"
+
+
+class Tactic(str, Enum):
+    """MITRE ATT&CK Enterprise tactics, keyed by their official ID."""
+
+    INITIAL_ACCESS = "TA0001"
+    EXECUTION = "TA0002"
+    PERSISTENCE = "TA0003"
+    PRIVILEGE_ESCALATION = "TA0004"
+    DEFENSE_EVASION = "TA0005"
+    CREDENTIAL_ACCESS = "TA0006"
+    DISCOVERY = "TA0007"
+    LATERAL_MOVEMENT = "TA0008"
+    COLLECTION = "TA0009"
+    EXFILTRATION = "TA0010"
+    COMMAND_AND_CONTROL = "TA0011"
+    IMPACT = "TA0040"
+
+
+@dataclass
+class DetectionRule:
+    """The vendor-neutral internal representation of a detection rule.
+
+    Everything downstream (Kibana JSON, Sigma YAML, the deterministic linter,
+    the AI critic) works off this one schema, so adding a new export target
+    or a new check never requires touching the rule authoring format.
+    """
+
+    name: str
+    platform: str
+    query: str
+    index_patterns: list[str]
+    description: str
+    severity: Severity
+    risk_score: int
+    tactics: list[Tactic] = field(default_factory=list)
+    techniques: list[str] = field(default_factory=list)
+    tags: list[str] = field(default_factory=list)
+    false_positives: list[str] = field(default_factory=list)
+    author: str = ""
+    interval: str = "30m"
+    lookback: str = "45m"
+    enabled: bool = False
+    rule_id: str | None = None
+
+    def __post_init__(self) -> None:
+        if not 0 < self.risk_score <= 100:
+            raise ValueError(f"risk_score must be in (0, 100], got {self.risk_score}")
+        if not self.name.strip():
+            raise ValueError("name must not be empty")
+
+    @classmethod
+    def from_dict(cls, data: dict[str, Any]) -> DetectionRule:
+        try:
+            severity = Severity(data["severity"])
+        except ValueError as exc:
+            allowed = ", ".join(s.value for s in Severity)
+            raise ValueError(f"severity must be one of: {allowed}") from exc
+
+        tactics = []
+        for raw in data.get("tactics", []):
+            try:
+                tactics.append(Tactic[str(raw).upper()])
+            except KeyError as exc:
+                allowed = ", ".join(t.name for t in Tactic)
+                raise ValueError(f"Unknown tactic {raw!r}, expected one of: {allowed}") from exc
+
+        return cls(
+            name=data["name"],
+            platform=data["platform"],
+            query=data["query"],
+            index_patterns=list(data.get("index_patterns", [])),
+            description=data.get("description", ""),
+            severity=severity,
+            risk_score=int(data["risk_score"]),
+            tactics=tactics,
+            techniques=list(data.get("techniques", [])),
+            tags=list(data.get("tags", [])),
+            false_positives=list(data.get("false_positives", [])),
+            author=data.get("author", ""),
+            interval=data.get("interval", "30m"),
+            lookback=data.get("lookback", "45m"),
+            enabled=bool(data.get("enabled", False)),
+            rule_id=data.get("rule_id"),
+        )

src/detection_as_code/secrets.py

@@ -0,0 +1,30 @@
+from __future__ import annotations
+
+import os
+from typing import Protocol
+
+
+class SecretProvider(Protocol):
+    """Anything that can resolve a named secret. Implement this against your
+    org's actual vault (AWS Secrets Manager, Akeyless, CyberArk, ...) and the
+    rest of the pipeline needs no changes - clients only depend on this shape.
+    """
+
+    def get(self, key: str) -> str: ...
+
+
+class EnvSecretProvider:
+    """Reads secrets from environment variables. This is the only provider
+    shipped here; production deployments should swap in a real vault-backed
+    provider rather than storing long-lived credentials in process env vars.
+    """
+
+    def __init__(self, prefix: str = "DAC_"):
+        self.prefix = prefix
+
+    def get(self, key: str) -> str:
+        env_key = f"{self.prefix}{key.upper()}"
+        value = os.environ.get(env_key)
+        if not value:
+            raise KeyError(f"Missing required secret: {env_key}")
+        return value

tests/test_models.py

@@ -0,0 +1,49 @@
+import pytest
+
+from detection_as_code.models import DetectionRule, Severity, Tactic
+
+VALID_RULE = {
+    "name": "windows_suspicious_scheduled_task",
+    "platform": "windows",
+    "query": 'process.name:"schtasks.exe"',
+    "index_patterns": ["winlogbeat-*"],
+    "description": "Detects scheduled task creation from a suspicious path.",
+    "severity": "high",
+    "risk_score": 62,
+    "tactics": ["persistence", "privilege_escalation"],
+    "techniques": ["T1053.005"],
+    "false_positives": ["Legitimate installer scheduling maintenance tasks"],
+}
+
+
+def test_from_dict_builds_a_valid_rule():
+    rule = DetectionRule.from_dict(VALID_RULE)
+
+    assert rule.severity == Severity.HIGH
+    assert Tactic.PERSISTENCE in rule.tactics
+    assert Tactic.PRIVILEGE_ESCALATION in rule.tactics
+    assert rule.risk_score == 62
+
+
+def test_risk_score_out_of_range_rejected():
+    data = {**VALID_RULE, "risk_score": 150}
+    with pytest.raises(ValueError, match="risk_score"):
+        DetectionRule.from_dict(data)
+
+
+def test_unknown_severity_rejected():
+    data = {**VALID_RULE, "severity": "apocalyptic"}
+    with pytest.raises(ValueError, match="severity"):
+        DetectionRule.from_dict(data)
+
+
+def test_unknown_tactic_rejected():
+    data = {**VALID_RULE, "tactics": ["quantum_leap"]}
+    with pytest.raises(ValueError, match="Unknown tactic"):
+        DetectionRule.from_dict(data)
+
+
+def test_empty_name_rejected():
+    data = {**VALID_RULE, "name": "   "}
+    with pytest.raises(ValueError, match="name"):
+        DetectionRule.from_dict(data)