Sigma Rule Validator

#!/usr/bin/env python3
"""
Sigma Rule Validator

READ THE DOCUMENTATION BEFORE YOU RUN THIS:
    https://permitany.com/scripts/sigma-rule-validator

It covers what the script does, whether it is safe to run, the
inventory file and credentials it needs, how to read the output,
and what to do when something fails.

    pip install "pyyaml>=6.0"
"""

import sys

# A Windows console is cp1252: an arrow or a check mark raises
# UnicodeEncodeError and kills the run partway through a library scan.
if hasattr(sys.stdout, "reconfigure"):
    sys.stdout.reconfigure(encoding="utf-8", errors="replace")
    sys.stderr.reconfigure(encoding="utf-8", errors="replace")

import argparse
import json
import os
import re
import uuid
from collections import defaultdict

try:
    import yaml
except ImportError:
    print("PyYAML is required:  pip install pyyaml", file=sys.stderr)
    sys.exit(2)


# --------------------------------------------------------------------------
# Reference data
# --------------------------------------------------------------------------
REQUIRED_TOP = ("title", "logsource", "detection")
RECOMMENDED_TOP = ("id", "status", "description", "author", "date",
                   "level", "falsepositives", "tags", "references")

VALID_STATUS = {"stable", "test", "experimental", "deprecated", "unsupported"}
VALID_LEVEL = {"informational", "low", "medium", "high", "critical"}

# The Sigma logsource taxonomy. A category outside this list is not wrong by
# definition — back-ends accept custom ones — but it is almost always a typo,
# and a typo here means the rule queries nothing.
KNOWN_CATEGORIES = {
    "process_creation", "process_access", "process_termination",
    "image_load", "driver_load", "file_event", "file_change", "file_delete",
    "file_rename", "file_access", "file_executable_detected",
    "registry_event", "registry_add", "registry_set", "registry_delete",
    "registry_rename",
    "network_connection", "dns_query", "dns", "firewall", "proxy", "webserver",
    "pipe_created", "wmi_event", "ps_module", "ps_script", "ps_classic_start",
    "create_remote_thread", "create_stream_hash", "raw_access_thread",
    "sysmon_error", "sysmon_status", "clipboard_capture",
    "antivirus", "application", "database", "vpn",
}
KNOWN_PRODUCTS = {
    "windows", "linux", "macos", "aws", "azure", "gcp", "m365", "okta",
    "onelogin", "github", "gworkspace", "kubernetes", "cisco", "paloalto",
    "fortinet", "juniper", "sql", "zeek", "qualys", "velocity", "rpc_firewall",
    "python", "django", "spring", "nodejs", "opencanary", "bitwarden",
}

# Field names that exist per category. Used to catch the single most costly
# typo class. Not exhaustive — an unknown field is a WARNING, never an ERROR.
KNOWN_FIELDS = {
    "process_creation": {
        "Image", "OriginalFileName", "CommandLine", "CurrentDirectory", "User",
        "LogonGuid", "LogonId", "TerminalSessionId", "IntegrityLevel", "Hashes",
        "md5", "sha1", "sha256", "imphash", "ParentProcessId", "ParentImage",
        "ParentCommandLine", "ParentUser", "ProcessId", "ProcessGuid",
        "Description", "Product", "Company", "FileVersion", "RuleName",
    },
    "process_access": {
        "SourceImage", "TargetImage", "GrantedAccess", "CallTrace",
        "SourceProcessId", "TargetProcessId", "SourceProcessGUID",
        "TargetProcessGUID", "SourceUser", "TargetUser", "SourceThreadId",
    },
    "image_load": {
        "Image", "ImageLoaded", "Hashes", "md5", "sha1", "sha256", "imphash",
        "Signed", "Signature", "SignatureStatus", "OriginalFileName",
        "Description", "Product", "Company", "FileVersion", "User", "ProcessId",
    },
    "network_connection": {
        "Image", "User", "Protocol", "Initiated", "SourceIp", "SourceHostname",
        "SourcePort", "SourcePortName", "DestinationIp", "DestinationHostname",
        "DestinationPort", "DestinationPortName", "DestinationIsIpv6",
        "SourceIsIpv6", "ProcessId", "ProcessGuid",
    },
    "dns_query": {"Image", "QueryName", "QueryStatus", "QueryResults",
                  "ProcessId", "ProcessGuid", "User", "record_type"},
    "file_event": {"Image", "TargetFilename", "CreationUtcTime", "User",
                   "ProcessId", "ProcessGuid", "Hashes"},
    "registry_event": {"Image", "TargetObject", "Details", "EventType",
                       "NewName", "User", "ProcessId", "ProcessGuid"},
    "registry_set": {"Image", "TargetObject", "Details", "EventType", "User",
                     "ProcessId", "ProcessGuid"},
    "pipe_created": {"Image", "PipeName", "ProcessId", "ProcessGuid", "User"},
    "proxy": {"c-uri", "c-uri-query", "c-uri-stem", "c-uri-extension",
              "c-useragent", "cs-method", "cs-host", "cs-bytes", "sc-bytes",
              "sc-status", "r-dns", "cs-referrer", "cs-cookie-var"},
    "webserver": {"c-uri", "c-uri-query", "c-useragent", "cs-method",
                  "cs-referrer", "sc-status", "cs-host", "c-ip"},
}

# Valid Sigma field modifiers. A typo here changes the match semantics or is
# ignored, depending on the backend.
VALID_MODIFIERS = {
    "contains", "startswith", "endswith", "all", "base64", "base64offset",
    "utf16", "utf16le", "utf16be", "wide", "re", "cidr", "lt", "lte", "gt",
    "gte", "windash", "expand", "fieldref", "exists", "cased",
}

CONDITION_KEYWORDS = {"and", "or", "not", "of", "them", "all", "1", "any"}

# Strings so short or so common that a rule matching on them alone will fire
# constantly. Flagged, not blocked.
WEAK_VALUES = {"cmd", "exe", "dll", "ps", "powershell.exe", "cmd.exe",
               "svchost.exe", "rundll32.exe", "explorer.exe", "a", "1", "test"}


class Finding:
    __slots__ = ("level", "code", "message", "where")

    def __init__(self, level, code, message, where=""):
        self.level = level
        self.code = code
        self.message = message
        self.where = where


# --------------------------------------------------------------------------
# Checks
# --------------------------------------------------------------------------
def check_metadata(rule, f):
    for key in REQUIRED_TOP:
        if key not in rule:
            f.append(Finding("ERROR", "missing-required",
                             f"missing required key '{key}'"))

    for key in RECOMMENDED_TOP:
        if key not in rule:
            f.append(Finding("INFO", "missing-recommended",
                             f"no '{key}' — recommended by the Sigma spec"))

    title = rule.get("title", "")
    if title:
        if len(title) > 120:
            f.append(Finding("WARNING", "title-long",
                             f"title is {len(title)} chars; it has to be readable "
                             f"in an alert queue"))
        if re.fullmatch(r"(?i)\s*(suspicious|malicious|possible)\s+activity\s*", title):
            f.append(Finding("WARNING", "title-vague",
                             f"title '{title}' says nothing to the analyst who "
                             f"receives the alert"))

    rid = rule.get("id")
    if rid is not None:
        try:
            uuid.UUID(str(rid))
        except (ValueError, AttributeError, TypeError):
            f.append(Finding("ERROR", "bad-id",
                             f"id '{rid}' is not a valid UUID"))

    status = rule.get("status")
    if status is not None and str(status).lower() not in VALID_STATUS:
        f.append(Finding("ERROR", "bad-status",
                         f"status '{status}' is not one of "
                         f"{', '.join(sorted(VALID_STATUS))}"))

    level = rule.get("level")
    if level is not None and str(level).lower() not in VALID_LEVEL:
        f.append(Finding("ERROR", "bad-level",
                         f"level '{level}' is not one of "
                         f"{', '.join(sorted(VALID_LEVEL))}"))

    date = rule.get("date")
    if date is not None:
        # The Sigma spec uses YYYY/MM/DD; YAML silently parses an unquoted
        # YYYY-MM-DD into a date object, which is why both shapes appear.
        text = date.isoformat() if hasattr(date, "isoformat") else str(date)
        if not re.fullmatch(r"\d{4}[/-]\d{2}[/-]\d{2}", text):
            f.append(Finding("WARNING", "bad-date",
                             f"date '{text}' is not YYYY/MM/DD"))

    fps = rule.get("falsepositives")
    if fps is not None:
        items = fps if isinstance(fps, list) else [fps]
        if not items or all(str(x).strip().lower() in ("unknown", "none", "")
                            for x in items):
            f.append(Finding("WARNING", "fp-unknown",
                             "falsepositives is 'Unknown' — the rule has not "
                             "been run against real data yet"))

    tags = rule.get("tags") or []
    if isinstance(tags, list):
        has_technique = any(re.match(r"(?i)attack\.t\d{4}", str(t)) for t in tags)
        has_tactic = any(re.match(r"(?i)attack\.[a-z_]+$", str(t)) for t in tags)
        if tags and not has_technique:
            f.append(Finding("INFO", "no-technique-tag",
                             "no attack.tXXXX technique tag — coverage mapping "
                             "will not pick this rule up"))
        if tags and not has_tactic:
            f.append(Finding("INFO", "no-tactic-tag", "no attack.<tactic> tag"))

    if str(rule.get("status", "")).lower() == "stable" and not rule.get("falsepositives"):
        f.append(Finding("WARNING", "stable-untested",
                         "status is 'stable' but there are no documented false "
                         "positives; stable should mean it has run in production"))


def check_logsource(rule, f):
    ls = rule.get("logsource")
    if ls is None:
        return None
    if not isinstance(ls, dict):
        f.append(Finding("ERROR", "logsource-shape", "logsource must be a mapping"))
        return None
    if not any(k in ls for k in ("category", "product", "service")):
        f.append(Finding("ERROR", "logsource-empty",
                         "logsource has none of category/product/service — "
                         "this converts to a query against nothing"))
        return None

    cat = ls.get("category")
    if cat and str(cat) not in KNOWN_CATEGORIES:
        f.append(Finding("WARNING", "unknown-category",
                         f"logsource category '{cat}' is not in the Sigma "
                         f"taxonomy — a typo here makes the rule query the "
                         f"wrong index and return nothing, silently"))
    prod = ls.get("product")
    if prod and str(prod) not in KNOWN_PRODUCTS:
        f.append(Finding("INFO", "unknown-product",
                         f"logsource product '{prod}' is not a common value"))
    return str(cat) if cat else None


def walk_fields(node, out):
    """Collect every field name used anywhere in a detection selection."""
    if isinstance(node, dict):
        for key, value in node.items():
            out.append(key)
            walk_fields(value, out)
    elif isinstance(node, list):
        for item in node:
            walk_fields(item, out)


def walk_values(node, out):
    if isinstance(node, dict):
        for value in node.values():
            walk_values(value, out)
    elif isinstance(node, list):
        for item in node:
            walk_values(item, out)
    elif node is not None:
        out.append(node)


def check_detection(rule, category, f):
    det = rule.get("detection")
    if det is None:
        return
    if not isinstance(det, dict):
        f.append(Finding("ERROR", "detection-shape", "detection must be a mapping"))
        return

    condition = det.get("condition")
    if condition is None:
        f.append(Finding("ERROR", "no-condition",
                         "detection has no 'condition' — the rule cannot be "
                         "evaluated"))
        return

    conditions = condition if isinstance(condition, list) else [condition]
    cond_text = " ".join(str(c) for c in conditions)

    selections = [k for k in det if k != "condition"]
    if not selections:
        f.append(Finding("ERROR", "no-selection",
                         "detection has a condition but no selection blocks"))
        return

    # Identifiers referenced by the condition, minus the operators.
    referenced = {
        tok for tok in re.findall(r"[A-Za-z_][A-Za-z0-9_]*\*?", cond_text)
        if tok.lower() not in CONDITION_KEYWORDS
    }

    def matches_any(token, names):
        if token.endswith("*"):
            return [n for n in names if n.startswith(token[:-1])]
        return [n for n in names if n == token]

    # A condition referencing a selection that does not exist is the defect
    # that produces a rule which converts cleanly and matches nothing.
    for token in sorted(referenced):
        if not matches_any(token, selections):
            f.append(Finding("ERROR", "undefined-selection",
                             f"condition references '{token}', which is not a "
                             f"defined selection ({', '.join(selections)})"))

    for name in selections:
        if not any(matches_any(tok, [name]) for tok in referenced):
            f.append(Finding("WARNING", "unused-selection",
                             f"selection '{name}' is defined but never used in "
                             f"the condition — it has no effect"))

    # Field names and modifiers. Walk each selection block's CONTENTS: passing
    # the detection dict itself would collect the selection names as if they
    # were field names.
    fields = []
    for name in selections:
        walk_fields(det[name], fields)
    known = KNOWN_FIELDS.get(category or "", set())

    seen_bare = set()
    for raw in fields:
        if not isinstance(raw, str):
            continue
        parts = raw.split("|")
        bare = parts[0]
        for mod in parts[1:]:
            if mod not in VALID_MODIFIERS:
                f.append(Finding("ERROR", "bad-modifier",
                                 f"'{raw}' uses modifier '{mod}', which is not "
                                 f"a valid Sigma modifier"))
        if bare in seen_bare or not known or bare in known:
            seen_bare.add(bare)
            continue
        seen_bare.add(bare)
        # Case-insensitive near-miss is almost always the real cause.
        near = [k for k in known if k.lower() == bare.lower()]
        if near:
            f.append(Finding("ERROR", "field-case",
                             f"field '{bare}' should be '{near[0]}' — field "
                             f"names are case-sensitive in most backends"))
        else:
            f.append(Finding("WARNING", "unknown-field",
                             f"field '{bare}' is not a known field for "
                             f"logsource category '{category}'"))

    # Value quality
    values = []
    walk_values({k: v for k, v in det.items() if k != "condition"}, values)
    for v in values:
        if isinstance(v, str):
            if v.strip() == "":
                f.append(Finding("ERROR", "empty-value",
                                 "a detection value is an empty string, which "
                                 "matches unpredictably"))
            elif v.strip().lower() in WEAK_VALUES:
                f.append(Finding("WARNING", "weak-value",
                                 f"value '{v}' on its own is too common to be "
                                 f"a discriminator"))
            elif v.strip() == "*":
                f.append(Finding("WARNING", "wildcard-only",
                                 "a bare '*' value matches everything"))

    # Exclusions built on a bare path are an evasion handed to the attacker if
    # the directory is writable.
    for name in selections:
        if not name.lower().startswith(("filter", "exclu")):
            continue
        sub_fields = []
        walk_fields(det[name], sub_fields)
        path_only = [x for x in sub_fields
                     if isinstance(x, str) and x.split("|")[0] in
                     ("Image", "SourceImage", "ParentImage", "TargetFilename")]
        has_strong = any(isinstance(x, str) and x.split("|")[0] in
                         ("Hashes", "md5", "sha1", "sha256", "imphash",
                          "Signed", "Signature", "SignatureStatus")
                         for x in sub_fields)
        if path_only and not has_strong:
            f.append(Finding("WARNING", "weak-exclusion",
                             f"exclusion '{name}' matches on a path alone; if "
                             f"that directory is writable this is an evasion. "
                             f"Add a signature or hash constraint"))

    if "timeframe" in det and not re.search(r"\b(count|near|sum|min|max|avg)\b",
                                            cond_text, re.I):
        f.append(Finding("INFO", "unused-timeframe",
                         "a timeframe is set but the condition has no "
                         "aggregation that uses it"))
    if re.search(r"\b(count|near|sum|avg)\b", cond_text, re.I) and "timeframe" not in det:
        f.append(Finding("ERROR", "aggregation-no-timeframe",
                         "the condition aggregates but no 'timeframe' is set"))


def validate(rule, path, index):
    f = []
    if not isinstance(rule, dict):
        f.append(Finding("ERROR", "not-a-mapping",
                         "document is not a YAML mapping"))
        return f
    check_metadata(rule, f)
    category = check_logsource(rule, f)
    check_detection(rule, category, f)
    for item in f:
        item.where = f"{path}" + (f" [doc {index}]" if index else "")
    return f


# --------------------------------------------------------------------------
def collect_files(paths):
    out = []
    for p in paths:
        if os.path.isdir(p):
            for root, _dirs, files in os.walk(p):
                for name in sorted(files):
                    if name.lower().endswith((".yml", ".yaml")):
                        out.append(os.path.join(root, name))
        elif os.path.isfile(p):
            out.append(p)
        else:
            print(f"  ! not found: {p}", file=sys.stderr)
    return out


COLOUR = {"ERROR": "ERROR  ", "WARNING": "WARN   ", "INFO": "INFO   "}


def main():
    ap = argparse.ArgumentParser(
        description="Validate Sigma detection rules offline.",
        epilog="Exit 0 = clean, 1 = warnings only, 2 = errors (or could not run).",
    )
    ap.add_argument("paths", nargs="+", help="Rule files or directories")
    ap.add_argument("--strict", action="store_true",
                    help="Treat warnings as failures (exit 2)")
    ap.add_argument("--quiet", action="store_true",
                    help="Only show rules that have findings")
    ap.add_argument("--no-info", action="store_true",
                    help="Suppress INFO findings")
    ap.add_argument("--output", help="Write the full report to a .json file")
    args = ap.parse_args()

    files = collect_files(args.paths)
    if not files:
        print("No .yml/.yaml rule files found.", file=sys.stderr)
        return 2

    all_findings = []
    ids_seen = defaultdict(list)
    titles_seen = defaultdict(list)
    rules_ok = 0
    parse_failures = 0

    print()
    print("=" * 78)
    print("  SIGMA RULE VALIDATION")
    print("=" * 78)

    for path in files:
        try:
            with open(path, "r", encoding="utf-8") as fh:
                docs = list(yaml.safe_load_all(fh))
        except yaml.YAMLError as exc:
            parse_failures += 1
            finding = Finding("ERROR", "yaml-parse", f"invalid YAML: {exc}", path)
            all_findings.append(finding)
            print(f"\n  {path}")
            print(f"    ERROR    [yaml-parse] invalid YAML: "
                  f"{str(exc).splitlines()[0]}")
            continue
        except OSError as exc:
            parse_failures += 1
            print(f"  ! cannot read {path}: {exc}", file=sys.stderr)
            continue

        for i, doc in enumerate(d for d in docs if d is not None):
            findings = validate(doc, path, i if len(docs) > 1 else 0)
            if args.no_info:
                findings = [x for x in findings if x.level != "INFO"]

            if isinstance(doc, dict):
                if doc.get("id"):
                    ids_seen[str(doc["id"])].append(path)
                if doc.get("title"):
                    titles_seen[str(doc["title"]).strip().lower()].append(path)

            all_findings.extend(findings)
            if not findings:
                rules_ok += 1
                if not args.quiet:
                    print(f"\n  {path}")
                    print(f"    clean")
                continue

            print(f"\n  {path}")
            for item in sorted(findings, key=lambda x:
                               {"ERROR": 0, "WARNING": 1, "INFO": 2}[x.level]):
                print(f"    {COLOUR[item.level]}[{item.code}] {item.message}")

    # Library-wide checks: only meaningful across files.
    dupes = []
    for rid, paths_ in ids_seen.items():
        if len(paths_) > 1:
            dupes.append(Finding("ERROR", "duplicate-id",
                                 f"id {rid} is used by {len(paths_)} rules: "
                                 f"{', '.join(paths_)}"))
    for title, paths_ in titles_seen.items():
        if len(paths_) > 1:
            dupes.append(Finding("WARNING", "duplicate-title",
                                 f"title '{title}' is used by {len(paths_)} "
                                 f"rules: {', '.join(paths_)}"))
    if dupes:
        print(f"\n  LIBRARY-WIDE")
        for item in dupes:
            print(f"    {COLOUR[item.level]}[{item.code}] {item.message}")
        all_findings.extend(dupes)

    errors = [x for x in all_findings if x.level == "ERROR"]
    warnings = [x for x in all_findings if x.level == "WARNING"]
    infos = [x for x in all_findings if x.level == "INFO"]

    print()
    print("=" * 78)
    print(f"  {len(files)} file(s), {rules_ok} clean")
    print(f"  {len(errors)} error(s), {len(warnings)} warning(s), {len(infos)} info")
    if errors:
        by_code = defaultdict(int)
        for e in errors:
            by_code[e.code] += 1
        print(f"  errors by type: " +
              ", ".join(f"{k}={v}" for k, v in sorted(by_code.items())))
    print("=" * 78)
    print()

    if args.output:
        with open(args.output, "w", encoding="utf-8") as fh:
            json.dump({
                "files": len(files),
                "clean": rules_ok,
                "findings": [
                    {"level": x.level, "code": x.code,
                     "message": x.message, "where": x.where}
                    for x in all_findings
                ],
            }, fh, indent=2)
        print(f"  Report written to {args.output}\n")

    # Exit 2 on a broken rule so this can gate a detection-as-code pipeline.
    # Exiting 0 regardless would make the check decorative.
    if errors or parse_failures:
        return 2
    if warnings:
        return 2 if args.strict else 1
    return 0


if __name__ == "__main__":
    try:
        sys.exit(main())
    except KeyboardInterrupt:
        # Explicit, so Ctrl-C still works while walking a large rule library.
        print("\nInterrupted.", file=sys.stderr)
        sys.exit(130)