modules/cypher_rule_validator.py

"""CYPHER V12 M5 — Defensive rules validator (YARA + Sigma + Suricata).

Validates security rules generated/analyzed by CYPHER, annotates output.
Post-processing layer: detect blocks via regex, compile/parse, append
[VALID] or [SYNTAX_ERROR: ...] marker.
"""
from __future__ import annotations

import logging
import re
from typing import Any

import yaml

try:
    import yara  # noqa: F401
    _HAS_YARA = True
except ImportError:
    _HAS_YARA = False

logger = logging.getLogger(__name__)

# Block detection patterns
_YARA_RULE_RE = re.compile(
    r"(rule\s+[A-Za-z_][A-Za-z0-9_]*\s*(?::[^\n{]+)?\s*\{[^}]*\})",
    re.DOTALL,
)
_YARA_STRINGS_RE = re.compile(r'\$[A-Za-z0-9_]+\s*=\s*"([^"\\]*(?:\\.[^"\\]*)*)"')
_YARA_NAME_RE = re.compile(r"rule\s+([A-Za-z_][A-Za-z0-9_]*)")

# Sigma: detect YAML blocks with 'title' + 'detection' keys
_SIGMA_BLOCK_RE = re.compile(
    r"(---\s*\n.*?(?=\n---|\Z)|(?<!\S)title:\s*[^\n]+\n.*?(?=\n\n|\Z))",
    re.DOTALL,
)

# Suricata: alert lines
_SURICATA_RE = re.compile(r"^(alert\s+\w+\s+[^\n]+)", re.MULTILINE)
_SURICATA_SID_RE = re.compile(r"sid:\s*(\d+)")
_SURICATA_MSG_RE = re.compile(r'msg:\s*"([^"]+)"')
_SURICATA_CLASSTYPE_RE = re.compile(r"classtype:\s*([a-z0-9_-]+)")


class YaraValidator:
    """Compile YARA rules with yara-python. Reports syntax errors."""

    def validate(self, rule_str: str) -> dict:
        rule_name = None
        m = _YARA_NAME_RE.search(rule_str)
        if m:
            rule_name = m.group(1)
        if not _HAS_YARA:
            return {
                "valid": None,
                "error": "yara-python not installed (cannot verify)",
                "rule_name": rule_name,
            }
        try:
            yara.compile(source=rule_str)
            return {"valid": True, "error": None, "rule_name": rule_name}
        except yara.SyntaxError as e:
            return {"valid": False, "error": str(e), "rule_name": rule_name}
        except yara.Error as e:
            return {"valid": False, "error": f"YARA error: {e}", "rule_name": rule_name}

    def extract_strings(self, rule_str: str) -> list[str]:
        return [m for m in _YARA_STRINGS_RE.findall(rule_str)]


class SigmaValidator:
    """Validate Sigma rules: YAML parse + required fields check."""

    REQUIRED_FIELDS = ("title", "logsource", "detection")

    def validate(self, yaml_str: str) -> dict:
        try:
            doc = yaml.safe_load(yaml_str)
        except yaml.YAMLError as e:
            return {"valid": False, "error": f"YAML parse: {e}", "title": None, "level": None}
        if not isinstance(doc, dict):
            return {"valid": False, "error": "Not a mapping", "title": None, "level": None}
        missing = [k for k in self.REQUIRED_FIELDS if k not in doc]
        if missing:
            return {
                "valid": False,
                "error": f"Missing required fields: {missing}",
                "title": doc.get("title"),
                "level": doc.get("level"),
            }
        det = doc.get("detection")
        if not isinstance(det, dict):
            return {"valid": False, "error": "detection must be a mapping",
                    "title": doc.get("title"), "level": doc.get("level")}
        if "condition" not in det:
            return {"valid": False, "error": "detection.condition missing",
                    "title": doc.get("title"), "level": doc.get("level")}
        return {
            "valid": True,
            "error": None,
            "title": doc.get("title"),
            "level": doc.get("level"),
        }

    def extract_logsource(self, yaml_str: str) -> dict:
        try:
            doc = yaml.safe_load(yaml_str)
            if isinstance(doc, dict):
                ls = doc.get("logsource", {})
                if isinstance(ls, dict):
                    return ls
        except yaml.YAMLError:
            pass
        return {}


class SuricataValidator:
    """Minimal Suricata/Snort rule validator.

    Verifies:
    - Starts with action (alert|drop|reject|pass)
    - Has protocol + IP/port spec + options in (...)
    - sid present
    """

    ACTIONS = ("alert", "drop", "reject", "pass")
    PROTOCOLS = ("tcp", "udp", "icmp", "ip", "http", "tls", "dns", "smb", "ftp", "smtp")

    def validate(self, rule_str: str) -> dict:
        rule_str = rule_str.strip()
        if not rule_str:
            return {"valid": False, "error": "empty rule", "sid": None, "msg": None}
        parts = rule_str.split(None, 1)
        action = parts[0].lower()
        if action not in self.ACTIONS:
            return {"valid": False, "error": f"unknown action: {action}",
                    "sid": None, "msg": None}
        if "(" not in rule_str or ")" not in rule_str:
            return {"valid": False, "error": "missing options ( ... )",
                    "sid": None, "msg": None}
        # Check protocol present
        proto_found = any(f" {p} " in f" {rule_str.lower()} " for p in self.PROTOCOLS)
        if not proto_found:
            return {"valid": False, "error": "no recognized protocol",
                    "sid": None, "msg": None}
        sid_m = _SURICATA_SID_RE.search(rule_str)
        msg_m = _SURICATA_MSG_RE.search(rule_str)
        if not sid_m:
            return {"valid": False, "error": "sid: missing",
                    "sid": None, "msg": msg_m.group(1) if msg_m else None}
        return {
            "valid": True,
            "error": None,
            "sid": int(sid_m.group(1)),
            "msg": msg_m.group(1) if msg_m else None,
        }

    def extract_classtype(self, rule_str: str) -> str | None:
        m = _SURICATA_CLASSTYPE_RE.search(rule_str)
        return m.group(1) if m else None


def extract_rules_from_text(text: str) -> dict[str, list[str]]:
    """Detect and extract YARA, Sigma, Suricata rule blocks from arbitrary text."""
    yara_blocks = _YARA_RULE_RE.findall(text)
    # Sigma: blocks starting with --- or 'title:' followed by 'detection:' nearby
    sigma_blocks: list[str] = []
    for m in _SIGMA_BLOCK_RE.finditer(text):
        block = m.group(0).strip()
        if "detection" in block and "title" in block and "logsource" in block:
            sigma_blocks.append(block)
    # Suricata: lines starting with action verb
    suricata_blocks: list[str] = []
    for line in text.split("\n"):
        line = line.strip()
        if line and line.split(None, 1)[0].lower() in SuricataValidator.ACTIONS:
            if "(" in line and ")" in line:
                suricata_blocks.append(line)
    return {
        "yara": yara_blocks,
        "sigma": sigma_blocks,
        "suricata": suricata_blocks,
    }


def annotate_response(response: str) -> str:
    """Detect rule blocks in CYPHER output, validate each, append marker."""
    blocks = extract_rules_from_text(response)
    yv = YaraValidator()
    sv = SigmaValidator()
    sur = SuricataValidator()
    annotations: list[str] = [response]

    for b in blocks["yara"]:
        r = yv.validate(b)
        if r["valid"] is True:
            annotations.append(f"\n[YARA VALID: {r['rule_name'] or 'anonymous'}]")
        elif r["valid"] is False:
            annotations.append(f"\n[YARA SYNTAX_ERROR: {r['error']}]")
        else:
            annotations.append(f"\n[YARA UNVERIFIED: {r['error']}]")
    for b in blocks["sigma"]:
        r = sv.validate(b)
        if r["valid"]:
            annotations.append(f"\n[SIGMA VALID: {r['title']} level={r['level']}]")
        else:
            annotations.append(f"\n[SIGMA SYNTAX_ERROR: {r['error']}]")
    for b in blocks["suricata"]:
        r = sur.validate(b)
        if r["valid"]:
            annotations.append(f"\n[SURICATA VALID: sid={r['sid']} msg={r['msg']}]")
        else:
            annotations.append(f"\n[SURICATA SYNTAX_ERROR: {r['error']}]")
    return "".join(annotations)


__all__ = [
    "YaraValidator",
    "SigmaValidator",
    "SuricataValidator",
    "extract_rules_from_text",
    "annotate_response",
]


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    print("=== M5 cypher_rule_validator SMOKE TEST ===")

    # 1. YARA
    valid_yara = '''
rule Test_Mimikatz {
    meta:
        author = "test"
    strings:
        $a = "sekurlsa::logonpasswords"
        $b = "mimikatz"
    condition:
        any of them
}
'''
    invalid_yara = "rule Bad { strings: $x = invalid_no_quotes condition: $x }"
    yv = YaraValidator()
    r1 = yv.validate(valid_yara)
    r2 = yv.validate(invalid_yara)
    print(f"YARA valid: {r1}")
    print(f"YARA invalid: {r2}")
    strings = yv.extract_strings(valid_yara)
    print(f"YARA strings extracted: {strings}")

    # 2. Sigma
    valid_sigma = """title: Suspicious PowerShell Download
logsource:
    product: windows
    service: powershell
detection:
    selection:
        EventID: 4104
        ScriptBlockText|contains:
            - 'DownloadString'
            - 'IEX'
    condition: selection
level: high
"""
    invalid_sigma = "title: Bad\nfoo: bar\n"
    sv = SigmaValidator()
    r3 = sv.validate(valid_sigma)
    r4 = sv.validate(invalid_sigma)
    print(f"Sigma valid: {r3}")
    print(f"Sigma invalid: {r4}")

    # 3. Suricata
    valid_suri = 'alert tcp $HOME_NET any -> $EXTERNAL_NET 443 (msg:"ET MALWARE Suspicious TLS"; classtype:trojan-activity; sid:2000001; rev:1;)'
    invalid_suri = "unknown tcp $HOME_NET -> any 80"
    sur = SuricataValidator()
    r5 = sur.validate(valid_suri)
    r6 = sur.validate(invalid_suri)
    print(f"Suricata valid: {r5}")
    print(f"Suricata invalid: {r6}")

    # 4. extract_rules_from_text + annotate_response
    response = "Here are some defensive rules:\n" + valid_yara + "\nAlso this Sigma rule:\n" + valid_sigma + "\nAnd a Suricata rule:\n" + valid_suri
    extracted = extract_rules_from_text(response)
    print(f"Extracted: yara={len(extracted['yara'])} sigma={len(extracted['sigma'])} suricata={len(extracted['suricata'])}")
    annotated = annotate_response(response)
    print(f"Annotated tail (last 300 chars):\n{annotated[-300:]}")

    print("=== SMOKE PASS ===")