diff --git a/src/detection_as_code/cli.py b/src/detection_as_code/cli.py new file mode 100644 index 0000000..2c25b93 --- /dev/null +++ b/src/detection_as_code/cli.py @@ -0,0 +1,118 @@ +from __future__ import annotations + +import argparse +import json +import sys +from pathlib import Path + +import yaml + +from .exporters import to_kibana_json, to_sigma +from .models import DetectionRule +from .secrets import EnvSecretProvider +from .validators import has_blocking_issues, lint_rule + + +def _load_rule(path: Path) -> DetectionRule: + data = yaml.safe_load(path.read_text()) + return DetectionRule.from_dict(data) + + +def cmd_lint(args: argparse.Namespace) -> int: + rule = _load_rule(args.rule_file) + issues = lint_rule(rule) + for issue in issues: + print(f"[{issue.level.value.upper()}] {issue.message}") + if has_blocking_issues(issues): + print("\nLint failed: blocking issues found") + return 1 + print("\nLint passed") + return 0 + + +def cmd_export(args: argparse.Namespace) -> int: + rule = _load_rule(args.rule_file) + print(to_sigma(rule) if args.format == "sigma" else to_kibana_json(rule)) + return 0 + + +def cmd_deploy(args: argparse.Namespace) -> int: + from .elastic_client import ElasticClient + from .kibana_client import KibanaClient + + rule = _load_rule(args.rule_file) + issues = lint_rule(rule) + if has_blocking_issues(issues): + print("Refusing to deploy: rule has blocking lint issues", file=sys.stderr) + return 1 + + secrets = EnvSecretProvider() + elastic = ElasticClient.from_secrets(secrets) + if not elastic.validate_query(rule.index_patterns, rule.query): + print("Refusing to deploy: query failed Elasticsearch validation", file=sys.stderr) + return 1 + + kibana = KibanaClient.from_secrets(secrets) + kibana.create_rule(json.loads(to_kibana_json(rule))) + print("Deployed rule successfully") + return 0 + + +def cmd_review(args: argparse.Namespace) -> int: + from .ai_review import review_rule + + rule = _load_rule(args.rule_file) + review = review_rule(rule) + + print(f"Overall: {review.overall_comment}\n") + for label, items in ( + ("Coverage gaps", review.coverage_gaps), + ("False positive risks", review.false_positive_risks), + ("Missing context", review.missing_context), + ): + print(f"{label}:") + for item in items: + print(f" - {item}") + if not items: + print(" (none)") + + if not review.available: + print( + "\nNote: this is advisory only and does not affect lint/deploy gates.", + file=sys.stderr, + ) + return 0 + + +def _build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(prog="detection-as-code") + sub = parser.add_subparsers(dest="command", required=True) + + p_lint = sub.add_parser("lint", help="Run deterministic checks against a rule file") + p_lint.add_argument("rule_file", type=Path) + p_lint.set_defaults(func=cmd_lint) + + p_export = sub.add_parser("export", help="Export a rule to Sigma or Kibana JSON") + p_export.add_argument("rule_file", type=Path) + p_export.add_argument("--format", choices=["sigma", "kibana"], default="kibana") + p_export.set_defaults(func=cmd_export) + + p_deploy = sub.add_parser("deploy", help="Validate and deploy a rule to Kibana") + p_deploy.add_argument("rule_file", type=Path) + p_deploy.set_defaults(func=cmd_deploy) + + p_review = sub.add_parser("review", help="Get an advisory AI review of a rule (non-blocking)") + p_review.add_argument("rule_file", type=Path) + p_review.set_defaults(func=cmd_review) + + return parser + + +def main(argv: list[str] | None = None) -> int: + parser = _build_parser() + args = parser.parse_args(argv) + return args.func(args) + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/test_cli.py b/tests/test_cli.py new file mode 100644 index 0000000..f746f56 --- /dev/null +++ b/tests/test_cli.py @@ -0,0 +1,75 @@ +import json +from pathlib import Path + +import pytest +import yaml + +from detection_as_code.cli import main + +VALID_RULE = { + "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"], +} + + +@pytest.fixture +def rule_file(tmp_path: Path) -> Path: + path = tmp_path / "rule.yaml" + path.write_text(yaml.safe_dump(VALID_RULE)) + return path + + +def test_lint_passes_for_a_clean_rule(rule_file, capsys): + exit_code = main(["lint", str(rule_file)]) + out = capsys.readouterr().out + + assert exit_code == 0 + assert "Lint passed" in out + + +def test_lint_fails_for_a_broad_rule(tmp_path, capsys): + path = tmp_path / "broad.yaml" + path.write_text(yaml.safe_dump({**VALID_RULE, "query": "*"})) + + exit_code = main(["lint", str(path)]) + out = capsys.readouterr().out + + assert exit_code == 1 + assert "too broad" in out.lower() + + +def test_export_kibana_produces_valid_json(rule_file, capsys): + exit_code = main(["export", str(rule_file), "--format", "kibana"]) + out = capsys.readouterr().out + + assert exit_code == 0 + payload = json.loads(out) + assert payload["name"] == VALID_RULE["name"] + + +def test_export_sigma_produces_valid_yaml(rule_file, capsys): + exit_code = main(["export", str(rule_file), "--format", "sigma"]) + out = capsys.readouterr().out + + assert exit_code == 0 + parsed = yaml.safe_load(out) + assert parsed["title"] == VALID_RULE["name"] + + +def test_deploy_refuses_when_lint_has_blocking_issues(tmp_path, capsys): + path = tmp_path / "broad.yaml" + path.write_text(yaml.safe_dump({**VALID_RULE, "query": "*"})) + + exit_code = main(["deploy", str(path)]) + err = capsys.readouterr().err + + assert exit_code == 1 + assert "Refusing to deploy" in err