Swain Security Check

#!/usr/bin/env python3
"""Context-File Injection Heuristic Scanner.

Scans agentic coding context files (AGENTS.md, CLAUDE.md, .cursorrules,
skill SKILL.md files, etc.) for prompt injection patterns using regex
heuristics.

Categories A-J per SPIKE-020:
  A: Instruction override
  B: Role override / persona hijacking
  C: Privilege escalation / authority spoofing
  D: Data exfiltration (explicit)
  E: Persistence mechanisms
  F: Base64 / encoding obfuscation
  G: Hidden Unicode (byte-level)
  H: MCP / config file manipulation
  I: HTML comment injection
  J: External fetch + exec

Exit codes:
  0 = no findings
  1 = findings detected
  2 = error

Zero external deps beyond Python 3 stdlib (re + unicodedata).
"""

from __future__ import annotations

import argparse
import json
import os
import re
import sys
import unicodedata
from typing import Any


# ---------------------------------------------------------------------------
# Rule definitions: each rule is (compiled_regex, category, severity, description)
# ---------------------------------------------------------------------------

RuleType = tuple[re.Pattern[str], str, str, str]

_RULES: list[RuleType] = []


def _add_rule(pattern: str, category: str, severity: str, description: str, flags: int = re.IGNORECASE) -> None:
    """Register a detection rule."""
    _RULES.append((re.compile(pattern, flags), category, severity, description))


# ---------------------------------------------------------------------------
# Category A — Instruction Override
# ---------------------------------------------------------------------------

_add_rule(
    r"ignore\s+(?:all\s+)?(?:previous|prior|above)\s+(?:instructions|rules|guidelines|prompts)",
    "A", "critical",
    "Instruction override: attempts to nullify prior instructions",
)
_add_rule(
    r"disregard\s+(?:all\s+)?(?:your|previous|prior|the)\s+(?:guidelines|instructions|rules|prompts)",
    "A", "critical",
    "Instruction override: disregard directives",
)
_add_rule(
    r"(?:new|override)\s+system\s+prompt",
    "A", "critical",
    "Instruction override: system prompt replacement",
)
_add_rule(
    r"\[END\s+OF\s+(?:CONTEXT|INSTRUCTIONS|SYSTEM|PROMPT)\]",
    "A", "high",
    "Instruction override: fake end-of-context marker",
)
_add_rule(
    r"forget\s+everything\s+(?:above|before|previous)",
    "A", "critical",
    "Instruction override: memory wipe directive",
)
_add_rule(
    r"do\s+not\s+follow\s+(?:the\s+)?previous\s+instructions",
    "A", "critical",
    "Instruction override: explicit instruction negation",
)

# ---------------------------------------------------------------------------
# Category B — Role Override / Persona Hijacking
# ---------------------------------------------------------------------------

_add_rule(
    r"you\s+are\s+now\s+(?!responsible|expected|required|going\s+to\s+be\s+(?:review|test|build))",
    "B", "critical",
    "Role override: persona reassignment",
)
_add_rule(
    r"act\s+as\s+(?:an?\s+)?(?:unrestricted|unfiltered|uncensored|evil|malicious|jailbroken)",
    "B", "critical",
    "Role override: unrestricted persona",
)
_add_rule(
    r"(?:enter|enable|activate)\s+(?:DAN|developer|god|sudo|jailbreak)\s+mode",
    "B", "critical",
    "Role override: jailbreak mode activation",
)
_add_rule(
    r"DAN\s+mode\s+(?:enabled|activated|on)",
    "B", "critical",
    "Role override: DAN mode declaration",
)
_add_rule(
    r"pretend\s+you\s+are\s+(?!working|a\s+code\s+reviewer|a\s+developer)",
    "B", "high",
    "Role override: pretend persona",
)
_add_rule(
    r"roleplay\s+as\s+",
    "B", "high",
    "Role override: roleplay directive",
)
_add_rule(
    r"jailbreak\s+(?:mode|enabled|activated|prompt)",
    "B", "critical",
    "Role override: jailbreak keyword",
)
_add_rule(
    r"developer\s+mode\s+(?:enabled|activated|on)",
    "B", "critical",
    "Role override: developer mode activation",
)

# ---------------------------------------------------------------------------
# Category C — Privilege Escalation / Authority Spoofing
# ---------------------------------------------------------------------------

_add_rule(
    r"you\s+have\s+full\s+access",
    "C", "critical",
    "Privilege escalation: full access claim",
)
_add_rule(
    r"(?:this\s+is\s+)?a?\s*message\s+from\s+(?:Anthropic|OpenAI|Google|Meta|the\s+developers?)\s*:",
    "C", "critical",
    "Privilege escalation: vendor impersonation",
)
_add_rule(
    r"(?:security|safety)\s+(?:restrictions?|filters?|guardrails?|limits?)\s+(?:are\s+)?(?:disabled|removed|lifted|off|turned\s+off)",
    "C", "critical",
    "Privilege escalation: safety bypass claim",
)
_add_rule(
    r"sudo\s+access\s+granted",
    "C", "critical",
    "Privilege escalation: sudo access claim",
)
_add_rule(
    r"admin\s+mode\s+(?:enabled|activated|on)",
    "C", "high",
    "Privilege escalation: admin mode claim",
)
_add_rule(
    r"all\s+restrictions?\s+(?:have\s+been\s+|are\s+)?lifted",
    "C", "critical",
    "Privilege escalation: restrictions lifted claim",
)
_add_rule(
    r"(?:this\s+)?(?:action\s+is\s+)?authorized\s+by\s+the\s+developer",
    "C", "high",
    "Privilege escalation: developer authorization claim",
)
_add_rule(
    r"I\s+am\s+(?:an?\s+)?(?:Anthropic|OpenAI|Google|Meta)\s+(?:employee|engineer|staff|team\s+member)",
    "C", "critical",
    "Privilege escalation: employee impersonation",
)

# ---------------------------------------------------------------------------
# Category D — Data Exfiltration
# ---------------------------------------------------------------------------

_add_rule(
    r"(?:send|output|print|display|show|share|upload|post|transmit)\s+(?:the\s+)?contents?\s+of\s+~?/",
    "D", "critical",
    "Data exfiltration: file content extraction",
)
_add_rule(
    r"output\s+all\s+files?\s+matching\s+\*?\.\w+",
    "D", "critical",
    "Data exfiltration: glob file extraction",
)
_add_rule(
    r"\|\s*base64\s*\|\s*(?:sh|bash|zsh)",
    "D", "critical",
    "Data exfiltration: base64 pipe to shell",
)
_add_rule(
    r"grep\s+[\w_-]*(?:key|secret|token|password|credential)[\w_-]*.*\|\s*curl",
    "D", "critical",
    "Data exfiltration: grep secrets piped to curl",
)
_add_rule(
    r"cat\s+[^\|]+\|\s*(?:curl|wget|nc|netcat)",
    "D", "critical",
    "Data exfiltration: cat file piped to network command",
)
_add_rule(
    r"(?:read|cat|print|display)\s+(?:the\s+)?contents?\s+of\s+~?/?\.\w+",
    "D", "high",
    "Data exfiltration: hidden file read",
)
_add_rule(
    r"(?:send|post|upload)\s+(?:this\s+)?(?:data|file|contents?)\s+to\s+https?://",
    "D", "critical",
    "Data exfiltration: send data to URL",
)
_add_rule(
    r"curl\s+(?:-\w\s+)*-F\s+['\"]?file=@",
    "D", "critical",
    "Data exfiltration: curl file upload",
)
_add_rule(
    r"\|\s*nc\s+\S+\s+\d+",
    "D", "critical",
    "Data exfiltration: pipe to netcat",
)
_add_rule(
    r"wget\s+--post-file=",
    "D", "critical",
    "Data exfiltration: wget POST file upload",
)

# ---------------------------------------------------------------------------
# Category E — Persistence Mechanisms
# ---------------------------------------------------------------------------

_add_rule(
    r"(?:write|append|add|overwrite|save|put|echo)\s+(?:the\s+following\s+|this\s+)?(?:to|into|in)\s+(?:\.?claude/)?MEMORY\.md",
    "E", "high",
    "Persistence: write to MEMORY.md",
)
_add_rule(
    r"(?:write|append|add|overwrite|modify|update|save|echo)\s+(?:the\s+following\s+)?(?:to|into|in|these\s+\w+\s+to)\s+\.clinerules",
    "E", "high",
    "Persistence: modify .clinerules",
)
_add_rule(
    r"(?:write|append|add|overwrite|modify|update|save|echo)\s+(?:the\s+following\s+)?(?:to|into|in|this\s+to)\s+\.cursorrules",
    "E", "high",
    "Persistence: modify .cursorrules",
)
_add_rule(
    r"(?:>>|write\s+to|append\s+to|modify)\s+~?/?\.(?:bashrc|zshrc|bash_profile|profile)",
    "E", "critical",
    "Persistence: shell RC file modification",
)
_add_rule(
    r"echo\s+['\"].*['\"]\s*>>\s*~?/?\.(?:bashrc|zshrc|bash_profile|profile)",
    "E", "critical",
    "Persistence: echo append to shell RC file",
)
_add_rule(
    r"nohup\s+nc\s+",
    "E", "critical",
    "Persistence: nohup netcat backdoor",
)
_add_rule(
    r"(?:>>|write\s+to|append\s+to|echo.*>>)\s*~?/?\.ssh/authorized_keys",
    "E", "critical",
    "Persistence: authorized_keys modification",
)
_add_rule(
    r"crontab",
    "E", "high",
    "Persistence: crontab manipulation",
)
_add_rule(
    r"overwrite\s+AGENTS\.md",
    "E", "high",
    "Persistence: overwrite AGENTS.md",
)
_add_rule(
    r"(?:write|append|add|save)\s+(?:the\s+following\s+)?to\s+\.claude/(?:settings\.json|commands/)",
    "E", "high",
    "Persistence: modify Claude settings or commands",
)

# ---------------------------------------------------------------------------
# Category F — Base64 / Encoding Obfuscation
# ---------------------------------------------------------------------------

_add_rule(
    r"(?:echo\s+['\"][\w+/=]+['\"]\s*\|?\s*)?base64\s+(?:-d|--decode)",
    "F", "high",
    "Encoding obfuscation: base64 decode operation",
)
_add_rule(
    r"base64\.b64decode\s*\(",
    "F", "high",
    "Encoding obfuscation: Python base64 decode",
)
_add_rule(
    r"tr\s+['\"]a-zA-Z['\"]\s+['\"]n-za-mN-ZA-M['\"]",
    "F", "high",
    "Encoding obfuscation: ROT13 transformation",
)
_add_rule(
    r"(?:%[0-9a-fA-F]{2}){6,}",
    "F", "medium",
    "Encoding obfuscation: URL-encoded payload (6+ encoded chars)",
)
_add_rule(
    r"(?:printf|echo\s+-e)\s+['\"](?:\\x[0-9a-fA-F]{2}){4,}",
    "F", "high",
    "Encoding obfuscation: hex-encoded payload",
)
_add_rule(
    r"openssl\s+(?:enc\s+)?base64\s+-d",
    "F", "high",
    "Encoding obfuscation: openssl base64 decode",
)
_add_rule(
    r"exec\s*\(\s*base64\.b64decode\s*\(",
    "F", "critical",
    "Encoding obfuscation: exec base64 decoded content",
)

# ---------------------------------------------------------------------------
# Category G — Hidden Unicode (byte-level)
# Handled specially in scan_content rather than via _RULES
# ---------------------------------------------------------------------------

# Cyrillic homoglyphs that look identical to Latin letters
# Maps Cyrillic codepoint to the Latin letter it impersonates
_CYRILLIC_HOMOGLYPHS: set[int] = {
    0x0410,  # А -> A
    0x0412,  # В -> B
    0x0415,  # Е -> E
    0x041A,  # К -> K
    0x041C,  # М -> M
    0x041D,  # Н -> H
    0x041E,  # О -> O
    0x0420,  # Р -> P
    0x0421,  # С -> C
    0x0422,  # Т -> T
    0x0425,  # Х -> X
    0x0430,  # а -> a
    0x0435,  # е -> e
    0x043E,  # о -> o
    0x0440,  # р -> p
    0x0441,  # с -> c
    0x0443,  # у -> y
    0x0445,  # х -> x
}


def _check_unicode_line(line: str, line_number: int, file_path: str) -> list[dict[str, Any]]:
    """Check a single line for suspicious Unicode characters."""
    findings: list[dict[str, Any]] = []
    seen_categories: set[str] = set()

    for char in line:
        cp = ord(char)

        # RTLO and bidi controls: U+202A-U+202E, U+2066-U+2069
        if 0x202A <= cp <= 0x202E or 0x2066 <= cp <= 0x2069:
            key = "bidi"
            if key not in seen_categories:
                seen_categories.add(key)
                findings.append({
                    "file_path": file_path,
                    "line_number": line_number,
                    "category": "G",
                    "severity": "critical",
                    "matched_pattern": f"U+{cp:04X} ({unicodedata.name(char, 'UNKNOWN')})",
                    "description": "Hidden Unicode: bidirectional control character",
                })

        # Zero-width characters: U+200B, U+200C, U+200D, U+FEFF
        elif cp in (0x200B, 0x200C, 0x200D, 0xFEFF):
            key = "zw"
            if key not in seen_categories:
                seen_categories.add(key)
                findings.append({
                    "file_path": file_path,
                    "line_number": line_number,
                    "category": "G",
                    "severity": "high",
                    "matched_pattern": f"U+{cp:04X} ({unicodedata.name(char, 'UNKNOWN')})",
                    "description": "Hidden Unicode: zero-width character",
                })

        # Unicode Tag block: U+E0000-U+E007F
        elif 0xE0000 <= cp <= 0xE007F:
            key = "tag"
            if key not in seen_categories:
                seen_categories.add(key)
                findings.append({
                    "file_path": file_path,
                    "line_number": line_number,
                    "category": "G",
                    "severity": "critical",
                    "matched_pattern": f"U+{cp:04X} (Unicode Tag block)",
                    "description": "Hidden Unicode: Tag block character (Rules File Backdoor carrier)",
                })

        # Cyrillic homoglyphs in otherwise Latin text
        elif cp in _CYRILLIC_HOMOGLYPHS:
            key = "homoglyph"
            if key not in seen_categories:
                seen_categories.add(key)
                findings.append({
                    "file_path": file_path,
                    "line_number": line_number,
                    "category": "G",
                    "severity": "high",
                    "matched_pattern": f"U+{cp:04X} ({unicodedata.name(char, 'UNKNOWN')})",
                    "description": "Hidden Unicode: Cyrillic homoglyph mixed with Latin text",
                })

    return findings


# ---------------------------------------------------------------------------
# Category H — MCP / Config File Manipulation
# ---------------------------------------------------------------------------

_add_rule(
    r"(?:rewrite|overwrite|modify|update|write\s+to|replace)\s+\.cursor/(?:mcp\.json|settings\.json)",
    "H", "critical",
    "MCP config manipulation: modify Cursor config file",
)
_add_rule(
    r"(?:insert|add|inject)\s+(?:new\s+)?MCP\s+server",
    "H", "critical",
    "MCP config manipulation: insert MCP server entry",
)
_add_rule(
    r"(?:curl|wget)\s+\S+\s*>\s*\.?cursor/mcp\.json",
    "H", "critical",
    "MCP config manipulation: download to mcp.json",
)
_add_rule(
    r"(?:add\s+this|configure\s+(?:this|a|the))\s+MCP\s+server",
    "H", "high",
    "MCP config manipulation: add MCP server directive",
)
_add_rule(
    r'"mcpServers"\s*:\s*\{',
    "H", "high",
    "MCP config manipulation: mcpServers JSON key",
)
_add_rule(
    r"(?:overwrite|modify|update|write\s+to|replace)\s+\.(?:vscode|cursor)/settings\.json",
    "H", "high",
    "MCP config manipulation: editor settings modification",
)

# ---------------------------------------------------------------------------
# Category I — HTML Comment Injection
# ---------------------------------------------------------------------------

_add_rule(
    r"<!--\s*(?:.*?(?:ignore|disregard|forget|override|new\s+system|you\s+are\s+now|act\s+as|run:|exec:|curl|wget|bash|sh\s))",
    "I", "critical",
    "HTML comment injection: hidden instruction in HTML comment",
)
_add_rule(
    r"<!--\s*(?:.*?(?:IMPORTANT|instruction|real\s+instructions|the\s+actual|everything\s+above))",
    "I", "high",
    "HTML comment injection: hidden directive in HTML comment",
)

# ---------------------------------------------------------------------------
# Category J — External Fetch + Exec
# ---------------------------------------------------------------------------

_add_rule(
    r"(?:curl|wget)\s+(?:-[\w-]+\s+)*\S+\s*\|\s*(?:sh|bash|zsh|python|python3|perl|ruby|node)",
    "J", "critical",
    "External fetch+exec: download and execute via pipe",
)
_add_rule(
    r"(?:curl|wget)\s+\S+\?\S*\$\(",
    "J", "critical",
    "External fetch+exec: URL with command substitution exfiltration",
)
_add_rule(
    r"!\[.*?\]\(https?://\S+\?\S*\$\{",
    "J", "critical",
    "External fetch+exec: markdown image with dynamic parameter exfiltration",
)
_add_rule(
    r"exec\s*\(\s*requests\.get\s*\(",
    "J", "critical",
    "External fetch+exec: Python fetch and exec",
)
_add_rule(
    r"(?:curl|wget)\s+(?:[\w./:=@?&%-]+\s+)*\S+\s*(?:&&|;)\s*(?:sh|bash|chmod\s+\+x)",
    "J", "critical",
    "External fetch+exec: download then execute",
)
_add_rule(
    r"npx\s+https?://",
    "J", "high",
    "External fetch+exec: npx from URL",
)
_add_rule(
    r"(?:curl|wget)\s+(?:-\w+\s+)*-[oO]\s+\S+\s+\S+\s*(?:&&|;)\s*(?:sh|bash|chmod|\.?/)",
    "J", "critical",
    "External fetch+exec: download to file then execute",
)


# ---------------------------------------------------------------------------
# Core scanning functions
# ---------------------------------------------------------------------------

def scan_content(content: str, file_path: str = "<stdin>") -> list[dict[str, Any]]:
    """Scan text content for injection patterns.

    Returns a list of finding dicts with keys:
      file_path, line_number, category, severity, matched_pattern, description
    """
    findings: list[dict[str, Any]] = []
    lines = content.split("\n")

    for line_idx, line in enumerate(lines):
        line_number = line_idx + 1

        # Check regex-based rules (Categories A-F, H-J)
        for pattern, category, severity, description in _RULES:
            match = pattern.search(line)
            if match:
                findings.append({
                    "file_path": file_path,
                    "line_number": line_number,
                    "category": category,
                    "severity": severity,
                    "matched_pattern": match.group(0),
                    "description": description,
                })

        # Check Unicode-based rules (Category G)
        findings.extend(_check_unicode_line(line, line_number, file_path))

    # Also check for multiline HTML comments (Category I)
    # Handle comments that span multiple lines
    for match in re.finditer(r"<!--(.*?)-->", content, re.DOTALL | re.IGNORECASE):
        comment_text = match.group(1)
        # Check if the comment contains injection patterns
        injection_patterns = [
            r"ignore", r"disregard", r"forget", r"override",
            r"new\s+system", r"you\s+are\s+now", r"act\s+as",
            r"run:", r"exec:", r"curl", r"wget", r"bash", r"sh\b",
            r"IMPORTANT", r"instruction", r"real\s+instructions",
            r"everything\s+above",
        ]
        for inj_pat in injection_patterns:
            if re.search(inj_pat, comment_text, re.IGNORECASE):
                # Find the line number of the comment start
                comment_start = match.start()
                line_num = content[:comment_start].count("\n") + 1
                # Only add if not already found by the line-based scan
                already_found = any(
                    f["category"] == "I" and f["line_number"] == line_num
                    for f in findings
                )
                if not already_found:
                    findings.append({
                        "file_path": file_path,
                        "line_number": line_num,
                        "category": "I",
                        "severity": "critical",
                        "matched_pattern": match.group(0)[:120],
                        "description": "HTML comment injection: hidden instruction in multiline HTML comment",
                    })
                break  # One finding per comment is enough

    return findings


def scan_file(file_path: str) -> list[dict[str, Any]]:
    """Scan a single file for injection patterns."""
    try:
        with open(file_path, encoding="utf-8", errors="replace") as f:
            content = f.read()
    except OSError:
        return []
    return scan_content(content, file_path=file_path)


# ---------------------------------------------------------------------------
# File discovery
# ---------------------------------------------------------------------------

# Exact filenames to match at any directory level
_CONTEXT_FILE_NAMES: set[str] = {
    "CLAUDE.md",
    "CLAUDE.local.md",
    "AGENTS.md",
    "AGENTS.override.md",
    ".cursorrules",
    ".clinerules",
    ".windsurfrules",
    ".aider.conf.yml",
    "system.md",
    "copilot-instructions.md",
}

# Filename patterns for files inside specific directories
_CONTEXT_DIR_PATTERNS: list[tuple[str, str]] = [
    # (directory component to match, file extension or name pattern)
    (".cursor/rules", ".mdc"),
    (".cursor", "mcp.json"),
    (".cursor", "settings.json"),
    (".claude/skills", "SKILL.md"),
    (".claude/commands", ".md"),
    (".claude", "settings.json"),
    (".github", "copilot-instructions.md"),
    (".github/agents", ".md"),
    (".gemini", "settings.json"),
    (".roo/rules", ".md"),
    (".agents/skills", "SKILL.md"),
    (".vscode", "settings.json"),
]


def discover_context_files(directory: str) -> list[str]:
    """Discover agentic runtime context files in a directory tree.

    Walks the directory tree looking for files that match known agentic
    context file names and patterns.
    """
    found: list[str] = []
    directory = os.path.abspath(directory)

    for root, dirs, files in os.walk(directory):
        # Skip common non-relevant directories
        basename = os.path.basename(root)
        if basename in ("node_modules", ".git", "__pycache__", ".venv", "venv"):
            dirs.clear()
            continue

        rel_root = os.path.relpath(root, directory)

        for filename in files:
            full_path = os.path.join(root, filename)

            # Check exact filename matches
            if filename in _CONTEXT_FILE_NAMES:
                found.append(full_path)
                continue

            # Check directory-based patterns
            for dir_pattern, file_pattern in _CONTEXT_DIR_PATTERNS:
                # Check if the relative path contains the directory pattern
                # Normalize both for comparison
                norm_rel = rel_root.replace(os.sep, "/")
                if dir_pattern in norm_rel or norm_rel.endswith(dir_pattern):
                    if file_pattern.startswith("."):
                        # Extension match
                        if filename.endswith(file_pattern):
                            found.append(full_path)
                            break
                    else:
                        # Exact filename match
                        if filename == file_pattern:
                            found.append(full_path)
                            break

    return sorted(found)


def scan_directory(directory: str) -> list[dict[str, Any]]:
    """Scan all context files in a directory tree."""
    files = discover_context_files(directory)
    findings: list[dict[str, Any]] = []
    for file_path in files:
        findings.extend(scan_file(file_path))
    return findings


# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------

def main(argv: list[str] | None = None) -> int:
    """CLI entry point. Returns exit code.

    Exit codes:
      0 = no findings
      1 = findings detected
      2 = error
    """
    parser = argparse.ArgumentParser(
        description="Scan agentic context files for prompt injection patterns.",
    )
    parser.add_argument(
        "paths",
        nargs="*",
        default=["."],
        help="Files or directories to scan (default: current directory)",
    )
    parser.add_argument(
        "--json",
        action="store_true",
        dest="json_output",
        help="Output findings as JSON array",
    )

    args = parser.parse_args(argv)

    all_findings: list[dict[str, Any]] = []
    had_error = False

    for path in args.paths:
        if os.path.isfile(path):
            try:
                all_findings.extend(scan_file(path))
            except Exception:
                had_error = True
        elif os.path.isdir(path):
            try:
                all_findings.extend(scan_directory(path))
            except Exception:
                had_error = True
        else:
            print(f"Error: path not found: {path}", file=sys.stderr)
            had_error = True

    if had_error and not all_findings:
        return 2

    if args.json_output:
        print(json.dumps(all_findings, indent=2))
    else:
        if all_findings:
            for finding in all_findings:
                severity = finding["severity"].upper()
                print(
                    f"[{severity}] {finding['file_path']}:{finding['line_number']} "
                    f"({finding['category']}) {finding['description']}"
                )
                print(f"  matched: {finding['matched_pattern']}")
                print()
        else:
            print("No findings.")

    if all_findings:
        return 1
    return 0


if __name__ == "__main__":
    sys.exit(main())

#!/usr/bin/env python3
"""Lightweight security diagnostic for swain-doctor session-start flow.

SPEC-061: Runs critical-only context-file scanning on AGENTS.md/CLAUDE.md
and checks for tracked .env files. Must complete in <3 seconds.

This module:
1. Scans context files for CRITICAL categories only: D, F, G, H
   (exfiltration, encoding obfuscation, hidden Unicode, MCP manipulation)
2. Checks for tracked .env files in git history
3. Outputs diagnostics in swain-doctor format (WARN/CRIT)
4. Silent pass when no issues found
5. Advisory only — does not block session start (always exits 0)

Exit codes:
  0 = no findings (clean)
  1 = findings detected (advisory)
"""

from __future__ import annotations

import os
import re
import subprocess
import sys
from typing import Any

# Import the context file scanner from the same directory
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from context_file_scanner import scan_content  # noqa: E402


# ---------------------------------------------------------------------------
# Constants
# ---------------------------------------------------------------------------

# Only these categories are checked during the lightweight doctor scan
CRITICAL_CATEGORIES = {"D", "F", "G", "H"}

# Context file names to scan at the repo root
CONTEXT_FILES = ["AGENTS.md", "CLAUDE.md"]

# Glob patterns for SKILL.md files
SKILL_DIRS = [
    os.path.join(".claude", "skills"),
    os.path.join(".agents", "skills"),
    os.path.join("skills"),
]


# ---------------------------------------------------------------------------
# Context-file scanning (critical categories only)
# ---------------------------------------------------------------------------


def _discover_context_files(repo_dir: str) -> list[str]:
    """Discover context files to scan in the repo.

    Finds AGENTS.md, CLAUDE.md at repo root, and SKILL.md files in
    known skill directories.
    """
    found: list[str] = []

    # Root context files
    for name in CONTEXT_FILES:
        path = os.path.join(repo_dir, name)
        if os.path.isfile(path):
            found.append(path)

    # SKILL.md files in skill directories
    for skill_dir in SKILL_DIRS:
        full_dir = os.path.join(repo_dir, skill_dir)
        if not os.path.isdir(full_dir):
            continue
        for entry in os.listdir(full_dir):
            skill_path = os.path.join(full_dir, entry, "SKILL.md")
            if os.path.isfile(skill_path):
                found.append(skill_path)

    return sorted(found)


def scan_context_files_critical(file_paths: list[str]) -> list[dict[str, Any]]:
    """Scan the given context files, returning only critical-category findings.

    Filters to categories D, F, G, H only.
    """
    findings: list[dict[str, Any]] = []

    for file_path in file_paths:
        try:
            with open(file_path, encoding="utf-8", errors="replace") as f:
                content = f.read()
        except OSError:
            continue

        all_findings = scan_content(content, file_path=file_path)
        critical = [f for f in all_findings if f["category"] in CRITICAL_CATEGORIES]
        findings.extend(critical)

    return findings


# ---------------------------------------------------------------------------
# Tracked .env file detection
# ---------------------------------------------------------------------------


def detect_tracked_env_files(repo_dir: str | None = None) -> list[str]:
    """Detect tracked .env files in git history.

    Returns list of tracked file paths matching .env pattern,
    excluding .env.example files.
    """
    cmd = ["git", "ls-files"]
    try:
        result = subprocess.run(
            cmd,
            capture_output=True,
            text=True,
            cwd=repo_dir,
            timeout=5,
        )
        if result.returncode != 0:
            return []
    except (subprocess.TimeoutExpired, FileNotFoundError, OSError):
        return []

    tracked_files = result.stdout.strip().splitlines()
    env_pattern = re.compile(r"\.env($|\.)")
    example_pattern = re.compile(r"\.example$")

    env_files = []
    for f in tracked_files:
        basename = os.path.basename(f)
        if env_pattern.search(basename) and not example_pattern.search(basename):
            env_files.append(f)

    return env_files


# ---------------------------------------------------------------------------
# Diagnostic formatting
# ---------------------------------------------------------------------------


def format_diagnostics(findings: list[dict[str, Any]]) -> list[dict[str, str]]:
    """Format scanner findings as swain-doctor diagnostics.

    Maps scanner severity to diagnostic severity:
      critical -> CRIT
      high     -> WARN
      medium   -> WARN
      low      -> INFO
    """
    severity_map = {
        "critical": "CRIT",
        "high": "WARN",
        "medium": "WARN",
        "low": "INFO",
    }

    diagnostics: list[dict[str, str]] = []
    for finding in findings:
        sev = severity_map.get(finding["severity"], "WARN")
        file_path = finding["file_path"]
        line_num = finding["line_number"]
        desc = finding["description"]
        cat = finding["category"]

        message = f"[{cat}] {file_path}:{line_num} — {desc}"
        diagnostics.append({"severity": sev, "message": message})

    return diagnostics


def format_env_diagnostics(tracked_envs: list[str]) -> list[dict[str, str]]:
    """Format tracked .env file findings as swain-doctor diagnostics."""
    diagnostics: list[dict[str, str]] = []
    for env_file in tracked_envs:
        message = (
            f"Tracked .env file: {env_file} — "
            f"remove with `git rm --cached {env_file}` and add to .gitignore. "
            f"For history cleanup, use BFG Repo-Cleaner or git filter-branch."
        )
        diagnostics.append({"severity": "WARN", "message": message})
    return diagnostics


# ---------------------------------------------------------------------------
# Main entry point
# ---------------------------------------------------------------------------


def main(repo_dir: str | None = None) -> int:
    """Run lightweight security diagnostics.

    Returns:
      0 = no findings
      1 = findings detected (advisory)
    """
    if repo_dir is None:
        try:
            result = subprocess.run(
                ["git", "rev-parse", "--show-toplevel"],
                capture_output=True,
                text=True,
                timeout=5,
            )
            repo_dir = result.stdout.strip() if result.returncode == 0 else os.getcwd()
        except (subprocess.TimeoutExpired, FileNotFoundError):
            repo_dir = os.getcwd()

    all_diagnostics: list[dict[str, str]] = []

    # 1. Context-file scan (critical categories only)
    context_files = _discover_context_files(repo_dir)
    findings = scan_context_files_critical(context_files)
    all_diagnostics.extend(format_diagnostics(findings))

    # 2. Tracked .env file detection
    tracked_envs = detect_tracked_env_files(repo_dir=repo_dir)
    all_diagnostics.extend(format_env_diagnostics(tracked_envs))

    # Output diagnostics
    if all_diagnostics:
        for diag in all_diagnostics:
            print(f"security-check [{diag['severity']}]: {diag['message']}")
        return 1

    return 0


if __name__ == "__main__":
    sys.exit(main())

"""External security skill hook interface (SPEC-065).

Provides three hook points for external security skills to plug into
swain-do's security gates:

1. **Pre-claim** — external skills contribute additional security guidance
2. **During-implementation** — security co-pilot context for file paths
3. **Completion** — differential review of git diffs

Skills are discovered by checking for known SKILL.md paths in
.claude/skills/ and .agents/skills/. All hooks are no-ops when no
external skills are installed — built-in guidance (SPEC-063) always runs.

Adding a new external skill requires only a call to register_skill()
with the detection pattern and hook capabilities — no core code changes.
"""

from __future__ import annotations

import os
from fnmatch import fnmatch
from pathlib import Path
from typing import Any

# ---------------------------------------------------------------------------
# Skill registry
# ---------------------------------------------------------------------------
# Each entry maps a detection pattern to the hook capabilities it supports.
# Capabilities: "security-briefing" (pre-claim), "security-context" (impl),
#               "security-review" (completion)

_SKILL_REGISTRY: list[dict[str, Any]] = [
    {
        "name": "trailofbits-sharp-edges",
        "detection_pattern": "trailofbits-sharp-edges",
        "hook_capabilities": ["security-briefing"],
    },
    {
        "name": "trailofbits-insecure-defaults",
        "detection_pattern": "trailofbits-insecure-defaults",
        "hook_capabilities": ["security-briefing"],
    },
    {
        "name": "trailofbits-differential-review",
        "detection_pattern": "trailofbits-differential-review",
        "hook_capabilities": ["security-review"],
    },
    {
        "name": "owasp-security",
        "detection_pattern": "owasp-security",
        "hook_capabilities": ["security-briefing", "security-context"],
    },
]

# Glob-style patterns registered via register_skill()
_GLOB_REGISTRY: list[dict[str, Any]] = []

# Standard skill directory prefixes (relative to a search dir)
_SKILL_PREFIXES = [
    os.path.join(".claude", "skills"),
    os.path.join(".agents", "skills"),
]


def register_skill(
    name: str,
    detection_pattern: str,
    hook_capabilities: list[str],
) -> None:
    """Register a new external skill detection pattern.

    After registration, detect_installed_skills() will discover skills
    matching the given detection_pattern in the standard skill directories.

    Args:
        name: Human-readable skill name (used as registry key).
        detection_pattern: Directory name or glob pattern to match
            (e.g., "snyk-security" or "snyk-*").
        hook_capabilities: List of supported hook types:
            "security-briefing", "security-context", "security-review".
    """
    entry = {
        "name": name,
        "detection_pattern": detection_pattern,
        "hook_capabilities": hook_capabilities,
    }
    # If the pattern contains glob characters, add to glob registry
    if "*" in detection_pattern or "?" in detection_pattern:
        _GLOB_REGISTRY.append(entry)
    else:
        _SKILL_REGISTRY.append(entry)


def _get_capabilities_for_skill(skill_name: str) -> list[str]:
    """Look up the hook capabilities for a detected skill name.

    Checks the static registry first, then glob registry entries.
    Returns a default set of all capabilities if the skill isn't
    in any registry (shouldn't normally happen).
    """
    for entry in _SKILL_REGISTRY:
        if entry["detection_pattern"] == skill_name or entry["name"] == skill_name:
            return entry["hook_capabilities"]
    for entry in _GLOB_REGISTRY:
        if fnmatch(skill_name, entry["detection_pattern"]):
            return entry["hook_capabilities"]
    # Default: all capabilities (permissive fallback)
    return ["security-briefing", "security-context", "security-review"]


# ---------------------------------------------------------------------------
# Skill detection
# ---------------------------------------------------------------------------


def detect_installed_skills(
    search_dirs: list[str] | None = None,
) -> dict[str, str]:
    """Detect installed external security skills.

    Scans known skill directory locations for SKILL.md files matching
    registered detection patterns.

    Args:
        search_dirs: List of root directories to search. Defaults to [cwd].

    Returns:
        Dict mapping skill_name -> skill_directory_path for all detected skills.
    """
    if search_dirs is None:
        search_dirs = [os.getcwd()]

    detected: dict[str, str] = {}

    for search_dir in search_dirs:
        for prefix in _SKILL_PREFIXES:
            skills_root = Path(search_dir) / prefix
            if not skills_root.is_dir():
                continue

            # Check each subdirectory against known patterns
            for child in skills_root.iterdir():
                if not child.is_dir():
                    continue

                skill_md = child / "SKILL.md"
                if not skill_md.is_file():
                    continue

                dir_name = child.name

                # Check static registry (exact match)
                for entry in _SKILL_REGISTRY:
                    pattern = entry["detection_pattern"]
                    if "*" not in pattern and "?" not in pattern:
                        if dir_name == pattern:
                            detected[dir_name] = str(child)
                            break
                    elif fnmatch(dir_name, pattern):
                        detected[dir_name] = str(child)
                        break
                else:
                    # Check glob registry
                    for entry in _GLOB_REGISTRY:
                        if fnmatch(dir_name, entry["detection_pattern"]):
                            detected[dir_name] = str(child)
                            break

    return detected


# ---------------------------------------------------------------------------
# Hook dispatch
# ---------------------------------------------------------------------------


def _dispatch_hook(
    hook_type: str,
    installed_skills: dict[str, str],
    context_label: str,
    context_data: str,
) -> list[str]:
    """Internal dispatcher for all hook types.

    Filters installed skills by capability, then generates a markdown
    guidance block for each matching skill. The actual guidance content
    comes from the skill's SKILL.md metadata — this interface provides
    the structured invocation surface.

    Args:
        hook_type: One of "security-briefing", "security-context", "security-review".
        installed_skills: Dict of skill_name -> skill_path.
        context_label: Label for the context data (e.g., "Task metadata", "File paths").
        context_data: The context data to pass to the skill.

    Returns:
        List of markdown-formatted guidance blocks, one per matching skill.
    """
    if not installed_skills:
        return []

    results: list[str] = []

    for skill_name, skill_path in installed_skills.items():
        capabilities = _get_capabilities_for_skill(skill_name)
        if hook_type not in capabilities:
            continue

        # Read the skill's SKILL.md for metadata
        skill_md_path = Path(skill_path) / "SKILL.md"
        skill_description = ""
        if skill_md_path.is_file():
            try:
                content = skill_md_path.read_text()
                # Extract the first heading as description
                for line in content.splitlines():
                    if line.startswith("# "):
                        skill_description = line[2:].strip()
                        break
            except OSError:
                pass

        # Build the guidance block
        block_lines = [
            f"### External: {skill_name}",
            "",
        ]
        if skill_description:
            block_lines.append(f"*{skill_description}*")
            block_lines.append("")

        block_lines.extend([
            f"- **Skill path**: `{skill_path}`",
            f"- **Hook**: `{hook_type}`",
            f"- **{context_label}**: {context_data}",
            "",
            f"Invoke `{skill_name}` for detailed {hook_type.replace('-', ' ')} analysis.",
        ])

        results.append("\n".join(block_lines))

    return results


def run_pre_claim_hooks(
    task_metadata: dict[str, Any],
    installed_skills: dict[str, str],
) -> list[str]:
    """Run pre-claim hooks on all installed skills that support it.

    Pre-claim hooks provide additional security guidance before a task
    is claimed. The guidance is additive — it never replaces built-in
    SPEC-063 guidance.

    Args:
        task_metadata: Dict with keys: title, tags, categories.
        installed_skills: Dict of skill_name -> skill_path from detect_installed_skills().

    Returns:
        List of markdown guidance blocks, one per matching skill.
        Empty list when no skills are installed or none support pre-claim.
    """
    context_parts = []
    if task_metadata.get("title"):
        context_parts.append(f"title='{task_metadata['title']}'")
    if task_metadata.get("tags"):
        context_parts.append(f"tags={task_metadata['tags']}")
    if task_metadata.get("categories"):
        context_parts.append(f"categories={task_metadata['categories']}")

    context_str = ", ".join(context_parts) if context_parts else "(empty)"

    return _dispatch_hook(
        hook_type="security-briefing",
        installed_skills=installed_skills,
        context_label="Task metadata",
        context_data=context_str,
    )


def run_implementation_hooks(
    file_paths: list[str],
    installed_skills: dict[str, str],
) -> list[str]:
    """Run during-implementation hooks on all installed skills that support it.

    Implementation hooks provide security co-pilot context for the files
    being edited. The notes are additive — they supplement, not replace,
    built-in guidance.

    Args:
        file_paths: List of file paths being edited.
        installed_skills: Dict of skill_name -> skill_path from detect_installed_skills().

    Returns:
        List of markdown security note blocks, one per matching skill.
        Empty list when no skills are installed or none support implementation hooks.
    """
    context_str = ", ".join(f"`{p}`" for p in file_paths) if file_paths else "(none)"

    return _dispatch_hook(
        hook_type="security-context",
        installed_skills=installed_skills,
        context_label="File paths",
        context_data=context_str,
    )


def run_completion_hooks(
    git_diff: str,
    installed_skills: dict[str, str],
) -> list[str]:
    """Run completion hooks on all installed skills that support it.

    Completion hooks provide differential review of the git diff after
    implementation. Findings are additive — they supplement, not replace,
    built-in security checks.

    Args:
        git_diff: Git diff string to review.
        installed_skills: Dict of skill_name -> skill_path from detect_installed_skills().

    Returns:
        List of markdown finding blocks, one per matching skill.
        Empty list when no skills are installed or none support completion hooks.
    """
    # Truncate long diffs for the context display
    diff_preview = git_diff[:200] + "..." if len(git_diff) > 200 else git_diff
    diff_display = diff_preview.replace("\n", " ")[:100]

    return _dispatch_hook(
        hook_type="security-review",
        installed_skills=installed_skills,
        context_label="Diff preview",
        context_data=diff_display if diff_display else "(empty)",
    )

#!/usr/bin/env python3
"""Scanner availability detection for swain-security-check (SPEC-059).

Detects whether security scanner binaries are available on the system
and generates OS-appropriate install commands when they are missing.

Supported scanners:
- gitleaks: secret detection (PATH check)
- osv-scanner: vulnerability scanning (PATH check)
- trivy: container/IaC scanning (PATH check)
- semgrep: static analysis (PATH check + uv-run fallback)

Design constraints:
- No network calls
- Must complete in < 1 second
- Returns structured results for swain-doctor integration
"""

from __future__ import annotations

import platform
import shutil
from dataclasses import dataclass
from typing import Optional


@dataclass
class ScannerResult:
    """Result of checking a single scanner's availability."""

    name: str
    available: bool
    path: Optional[str] = None
    install_hint: Optional[str] = None


# ---------------------------------------------------------------------------
# OS detection
# ---------------------------------------------------------------------------

def detect_os() -> str:
    """Detect the operating system. Returns 'darwin', 'linux', or the raw system name lowercased."""
    return platform.system().lower()


# ---------------------------------------------------------------------------
# Install command matrix
# ---------------------------------------------------------------------------

_INSTALL_COMMANDS: dict[str, dict[str, str]] = {
    "gitleaks": {
        "darwin": "brew install gitleaks",
        "linux": "apt install gitleaks  # or: cargo install gitleaks",
    },
    "osv-scanner": {
        "darwin": "brew install osv-scanner",
        "linux": "go install github.com/google/osv-scanner/cmd/osv-scanner@latest",
    },
    "trivy": {
        "darwin": "brew install trivy",
        "linux": "apt install trivy",
    },
    "semgrep": {
        "darwin": "uv run --with semgrep semgrep",
        "linux": "uv run --with semgrep semgrep",
    },
}

# Universal cargo fallback for scanners that support it
_CARGO_FALLBACK = {"gitleaks"}


def get_install_command(scanner: str, os_name: str) -> str:
    """Return the install command for a scanner on the given OS.

    Args:
        scanner: Scanner name (gitleaks, osv-scanner, trivy, semgrep).
        os_name: OS identifier from detect_os() (darwin, linux).

    Returns:
        Install command string. Falls back to cargo if available for the scanner,
        or returns a generic hint.
    """
    commands = _INSTALL_COMMANDS.get(scanner, {})
    cmd = commands.get(os_name)
    if cmd:
        return cmd
    # Fallback: cargo if supported, otherwise generic
    if scanner in _CARGO_FALLBACK:
        return f"cargo install {scanner}"
    return f"Install {scanner} — see project documentation"


# ---------------------------------------------------------------------------
# Individual scanner checks
# ---------------------------------------------------------------------------

def check_scanner(name: str) -> ScannerResult:
    """Check if a scanner binary is available in PATH.

    Args:
        name: Binary name to search for (gitleaks, osv-scanner, trivy).

    Returns:
        ScannerResult with availability status and install hint if missing.
    """
    path = shutil.which(name)
    if path:
        return ScannerResult(name=name, available=True, path=path)

    os_name = detect_os()
    hint = get_install_command(name, os_name)
    return ScannerResult(name=name, available=False, install_hint=hint)


def check_semgrep() -> ScannerResult:
    """Check semgrep availability with uv-run fallback.

    Strategy:
    1. Check if semgrep is directly in PATH -> available
    2. Check if uv is in PATH -> available via uv-run fallback
    3. Neither -> unavailable, hint to install uv

    No subprocess calls are made — detection is purely passive (which checks).
    """
    # Direct PATH check
    semgrep_path = shutil.which("semgrep")
    if semgrep_path:
        return ScannerResult(name="semgrep", available=True, path=semgrep_path)

    # uv-run fallback: if uv is available, semgrep can be run via uv
    uv_path = shutil.which("uv")
    if uv_path:
        return ScannerResult(
            name="semgrep",
            available=True,
            path=f"uv run --with semgrep semgrep",
        )

    # Neither available
    os_name = detect_os()
    hint = get_install_command("semgrep", os_name)
    return ScannerResult(
        name="semgrep",
        available=False,
        install_hint=f"uv run --with semgrep semgrep  # or install uv: {hint}",
    )


# ---------------------------------------------------------------------------
# Aggregate check
# ---------------------------------------------------------------------------

# Scanners that use simple PATH checks (not semgrep)
_PATH_SCANNERS = ["gitleaks", "osv-scanner", "trivy"]


def check_all_scanners() -> list[ScannerResult]:
    """Check availability of all security scanners.

    Returns:
        List of ScannerResult for each scanner (gitleaks, osv-scanner, trivy, semgrep).
    """
    results = [check_scanner(name) for name in _PATH_SCANNERS]
    results.append(check_semgrep())
    return results


# ---------------------------------------------------------------------------
# CLI entry point for swain-doctor integration
# ---------------------------------------------------------------------------

def format_report(results: list[ScannerResult]) -> str:
    """Format scanner availability as a human-readable report.

    Returns a multi-line string suitable for swain-doctor output.
    """
    lines = []
    available_count = sum(1 for r in results if r.available)
    total = len(results)

    lines.append(f"Scanner availability: {available_count}/{total} scanners found")

    for r in results:
        if r.available:
            lines.append(f"  [ok] {r.name}: {r.path}")
        else:
            lines.append(f"  [--] {r.name}: not found")
            if r.install_hint:
                lines.append(f"        install: {r.install_hint}")

    return "\n".join(lines)


if __name__ == "__main__":
    results = check_all_scanners()
    print(format_report(results))

"""Pre-claim security briefing generator (SPEC-063).

Generates markdown-formatted security guidance based on threat surface
detection categories. When a task is security-sensitive, the agent receives
relevant OWASP and swain-specific guidance before writing code.

Uses detect_threat_surface() from SPEC-062 for classification.
"""

from __future__ import annotations

import sys
from pathlib import Path

# Ensure sibling modules are importable
_SCRIPT_DIR = str(Path(__file__).parent)
if _SCRIPT_DIR not in sys.path:
    sys.path.insert(0, _SCRIPT_DIR)

from threat_surface import detect_threat_surface

# ---------------------------------------------------------------------------
# OWASP category-to-guidance mapping
# ---------------------------------------------------------------------------
# Each entry maps a threat_surface category to a dict with:
#   - owasp_id: OWASP Top 10 2021 identifier
#   - owasp_name: Full OWASP category name
#   - guidance: List of actionable guidance bullet points

CATEGORY_GUIDANCE: dict[str, dict] = {
    "auth": {
        "owasp_id": "A07:2021",
        "owasp_name": "Identification and Authentication Failures",
        "guidance": [
            "Never store passwords in plaintext; use bcrypt, scrypt, or argon2 with appropriate work factors",
            "Session tokens must be regenerated on privilege change (login, role elevation)",
            "Implement account lockout or rate limiting after repeated failed authentication attempts",
            "Use constant-time comparison for token and credential validation to prevent timing attacks",
            "Ensure multi-factor authentication is supported where applicable",
        ],
    },
    "input-validation": {
        "owasp_id": "A03:2021",
        "owasp_name": "Injection",
        "guidance": [
            "Validate and sanitize all user input at system boundaries",
            "Use parameterized queries for database access; never use string concatenation for SQL",
            "Apply output encoding appropriate to the context (HTML, URL, JavaScript, CSS)",
            "Implement allowlist validation where possible rather than blocklist filtering",
            "Treat all data from external sources as untrusted until validated",
        ],
    },
    "crypto": {
        "owasp_id": "A02:2021",
        "owasp_name": "Cryptographic Failures",
        "guidance": [
            "Use standard, well-vetted cryptographic algorithms (AES-256, RSA-2048+, SHA-256+)",
            "Never implement custom cryptographic algorithms or protocols",
            "Ensure deprecated or weak algorithms (MD5, SHA-1, DES, RC4) are not used",
            "Manage cryptographic keys securely; never hardcode keys in source code",
            "Use TLS 1.2+ for all data in transit; disable older protocol versions",
        ],
    },
    "external-data": {
        "owasp_id": "A08:2021",
        "owasp_name": "Software and Data Integrity Failures",
        "guidance": [
            "Verify the integrity of all external data using signatures or checksums",
            "Do not deserialize untrusted data without validation; use safe deserialization methods",
            "Validate that CI/CD pipelines have proper access controls and integrity checks",
            "Ensure software updates and patches come from verified, trusted sources",
            "Review and validate all data from third-party APIs before processing",
        ],
    },
    "agent-context": {
        "owasp_id": None,  # swain-specific, no OWASP mapping
        "owasp_name": None,
        "guidance": [
            "Agent context files (AGENTS.md, CLAUDE.md) are trust boundaries — do not write user-controlled data into them",
            "Never embed secrets, credentials, or API keys in context files",
            "Validate any data that flows from task descriptions into agent instructions",
            "Treat context file modifications as privileged operations requiring review",
            "Ensure agent context does not leak sensitive information across task boundaries",
        ],
    },
    "dependency-change": {
        "owasp_id": "A06:2021",
        "owasp_name": "Vulnerable and Outdated Components",
        "guidance": [
            "Audit new dependencies for known vulnerabilities before adding them (npm audit, pip-audit, etc.)",
            "Pin dependency versions and review lockfile changes carefully",
            "Remove unused dependencies to reduce attack surface",
            "Prefer well-maintained packages with active security response teams",
            "Check that transitive dependencies are not introducing known vulnerabilities",
        ],
    },
    "secrets": {
        "owasp_id": "A07:2021",
        "owasp_name": "Identification and Authentication Failures",
        "guidance": [
            "Never commit secrets, credentials, or API keys to version control",
            "Use environment variables or a dedicated secrets manager for sensitive values",
            "Ensure .env files and credential stores are listed in .gitignore",
            "Rotate secrets immediately if they are suspected of being exposed",
            "Use short-lived tokens and credentials where possible to limit blast radius",
        ],
    },
}


def _format_category_section(category: str) -> str:
    """Format a single category's guidance as a markdown section."""
    info = CATEGORY_GUIDANCE.get(category)
    if info is None:
        return ""

    lines: list[str] = []

    if info["owasp_id"] is not None:
        lines.append(f"### OWASP {info['owasp_id']} — {info['owasp_name']}")
    else:
        # swain-specific category (e.g., agent-context)
        lines.append(f"### swain guidance — {category}")

    lines.append("")
    for point in info["guidance"]:
        lines.append(f"- {point}")

    return "\n".join(lines)


def generate_security_briefing(
    title: str = "",
    description: str = "",
    tags: list[str] | None = None,
    spec_criteria: str = "",
    file_paths: list[str] | None = None,
) -> str:
    """Generate a markdown-formatted security briefing for a task.

    Delegates to detect_threat_surface() for classification, then maps
    detected categories to OWASP and swain-specific guidance.

    Args:
        title: Task title text.
        description: Task description text.
        tags: List of task tags.
        spec_criteria: SPEC acceptance criteria text.
        file_paths: List of file paths touched by the task.

    Returns:
        Markdown-formatted security briefing string, or empty string
        if the task is not security-sensitive.
    """
    result = detect_threat_surface(
        title=title,
        description=description,
        tags=tags,
        spec_criteria=spec_criteria,
        file_paths=file_paths,
    )

    if not result.is_security_sensitive:
        return ""

    sections: list[str] = []

    # Header with detected categories
    if result.categories:
        cat_list = ", ".join(result.categories)
        sections.append(f"## Security Briefing (categories: {cat_list})")
    else:
        # Triggered by generic 'security' tag with no specific category
        sections.append("## Security Briefing")

    sections.append("")

    if result.categories:
        # Emit guidance for each detected category
        for category in result.categories:
            section = _format_category_section(category)
            if section:
                sections.append(section)
                sections.append("")
    else:
        # Generic security tag with no specific category — emit general guidance
        sections.append("### General Security Guidance")
        sections.append("")
        sections.append("- Review code changes for common security anti-patterns")
        sections.append("- Validate all inputs and sanitize all outputs")
        sections.append("- Ensure no secrets or credentials are exposed")
        sections.append("- Check for proper error handling that does not leak sensitive information")
        sections.append("")

    return "\n".join(sections).rstrip("\n") + "\n"

#!/usr/bin/env python3
"""Security check orchestrator for swain-security-check (SPEC-060).

Orchestrates all security scanners into a single invocation with a
unified report. Integrates:
- context_file_scanner.py (SPEC-058) — built-in, always runs
- scanner_availability.py (SPEC-059) — checks which external scanners are on PATH
- gitleaks — secret detection
- osv-scanner / trivy — dependency vulnerability scanning
- semgrep — static analysis with ai-best-practices
- Repo hygiene — built-in .gitignore completeness + tracked secret detection

Exit codes:
  0 = no findings
  1 = findings detected
  2 = error

Zero external deps beyond Python 3 stdlib + sibling modules.
"""

from __future__ import annotations

import argparse
import json
import os
import subprocess
import sys
from dataclasses import dataclass, field
from typing import Any

# Import sibling modules
from context_file_scanner import scan_directory as ctx_scan_directory
from scanner_availability import ScannerResult, check_all_scanners


# ---------------------------------------------------------------------------
# Data structures
# ---------------------------------------------------------------------------

@dataclass
class ScanResult:
    """Result of a full security scan across all scanners."""

    findings: list[dict[str, Any]] = field(default_factory=list)
    scanners_run: list[str] = field(default_factory=list)
    scanners_skipped: list[dict[str, str]] = field(default_factory=list)
    errors: list[str] = field(default_factory=list)


# ---------------------------------------------------------------------------
# Finding normalization helpers
# ---------------------------------------------------------------------------

_SEVERITY_ORDER = ("critical", "high", "medium", "low")


def _normalize_severity(severity: str) -> str:
    """Map various severity strings to our canonical set."""
    s = severity.lower().strip()
    mapping = {
        "critical": "critical",
        "error": "critical",
        "high": "high",
        "warning": "medium",
        "medium": "medium",
        "info": "low",
        "low": "low",
    }
    return mapping.get(s, "medium")


def _make_finding(
    scanner: str,
    file_path: str,
    line: int,
    severity: str,
    description: str,
    remediation: str,
    **extra: Any,
) -> dict[str, Any]:
    """Create a normalized finding dict."""
    finding: dict[str, Any] = {
        "scanner": scanner,
        "file_path": file_path,
        "line": line,
        "severity": severity,
        "description": description,
        "remediation": remediation,
    }
    finding.update(extra)
    return finding


# ---------------------------------------------------------------------------
# Gitleaks scanner
# ---------------------------------------------------------------------------

def _run_gitleaks(directory: str, scanner_info: ScannerResult) -> list[dict[str, Any]]:
    """Run gitleaks and return normalized findings."""
    cmd = [
        "gitleaks", "detect",
        "--source", directory,
        "--report-format", "json",
        "--report-path", "/dev/stdout",
        "--no-git",
    ]
    try:
        proc = subprocess.run(
            cmd,
            capture_output=True,
            text=True,
            timeout=120,
        )
    except (subprocess.TimeoutExpired, OSError) as e:
        raise RuntimeError(f"gitleaks failed: {e}") from e

    if proc.returncode == 0:
        # No findings
        return []

    # Exit 1 = findings present
    try:
        raw = json.loads(proc.stdout) if proc.stdout.strip() else []
    except json.JSONDecodeError:
        return []

    findings = []
    for item in raw:
        findings.append(_make_finding(
            scanner="gitleaks",
            file_path=item.get("File", "unknown"),
            line=item.get("StartLine", 0),
            severity="critical",
            description=item.get("Description", "Secret detected"),
            remediation="Rotate the secret, remove from source control, and add to .gitignore",
        ))
    return findings


# ---------------------------------------------------------------------------
# Semgrep scanner
# ---------------------------------------------------------------------------

def _run_semgrep(directory: str, scanner_info: ScannerResult) -> list[dict[str, Any]]:
    """Run semgrep with ai-best-practices config and return normalized findings."""
    path = scanner_info.path or "semgrep"
    # If path contains "uv run", split it into parts
    if path.startswith("uv run"):
        cmd_parts = path.split() + ["--config", "p/ai-best-practices", "--json", directory]
    else:
        cmd_parts = [path, "--config", "p/ai-best-practices", "--json", directory]

    try:
        proc = subprocess.run(
            cmd_parts,
            capture_output=True,
            text=True,
            timeout=300,
        )
    except (subprocess.TimeoutExpired, OSError) as e:
        raise RuntimeError(f"semgrep failed: {e}") from e

    try:
        data = json.loads(proc.stdout) if proc.stdout.strip() else {"results": []}
    except json.JSONDecodeError:
        return []

    findings = []
    for item in data.get("results", []):
        severity_raw = item.get("extra", {}).get("severity", "WARNING")
        findings.append(_make_finding(
            scanner="semgrep",
            file_path=item.get("path", "unknown"),
            line=item.get("start", {}).get("line", 0),
            severity=_normalize_severity(severity_raw),
            description=item.get("extra", {}).get("message", item.get("check_id", "Semgrep finding")),
            remediation="Review and fix the flagged pattern per semgrep rule guidance",
        ))
    return findings


# ---------------------------------------------------------------------------
# OSV-scanner
# ---------------------------------------------------------------------------

def _run_osv_scanner(directory: str, scanner_info: ScannerResult) -> list[dict[str, Any]]:
    """Run osv-scanner and return normalized findings."""
    cmd = ["osv-scanner", "--format", "json", "--recursive", directory]
    try:
        proc = subprocess.run(
            cmd,
            capture_output=True,
            text=True,
            timeout=120,
        )
    except (subprocess.TimeoutExpired, OSError) as e:
        raise RuntimeError(f"osv-scanner failed: {e}") from e

    try:
        data = json.loads(proc.stdout) if proc.stdout.strip() else {"results": []}
    except json.JSONDecodeError:
        return []

    findings = []
    for result_group in data.get("results", []):
        source_path = result_group.get("source", {}).get("path", "unknown")
        for pkg in result_group.get("packages", []):
            for vuln in pkg.get("vulnerabilities", []):
                severity_val = "high"  # Default for known CVEs
                db_specific = vuln.get("database_specific", {})
                if db_specific.get("severity"):
                    severity_val = _normalize_severity(db_specific["severity"])
                findings.append(_make_finding(
                    scanner="osv-scanner",
                    file_path=source_path,
                    line=0,
                    severity=severity_val,
                    description=f"{vuln.get('id', 'Unknown CVE')}: {vuln.get('summary', 'Vulnerability detected')}",
                    remediation="Update the affected dependency to a patched version",
                ))
    return findings


# ---------------------------------------------------------------------------
# Trivy scanner (fallback for osv-scanner)
# ---------------------------------------------------------------------------

def _run_trivy(directory: str, scanner_info: ScannerResult) -> list[dict[str, Any]]:
    """Run trivy and return normalized findings."""
    cmd = ["trivy", "fs", "--format", "json", directory]
    try:
        proc = subprocess.run(
            cmd,
            capture_output=True,
            text=True,
            timeout=120,
        )
    except (subprocess.TimeoutExpired, OSError) as e:
        raise RuntimeError(f"trivy failed: {e}") from e

    try:
        data = json.loads(proc.stdout) if proc.stdout.strip() else {"Results": []}
    except json.JSONDecodeError:
        return []

    findings = []
    for result_group in data.get("Results", []):
        target = result_group.get("Target", "unknown")
        for vuln in result_group.get("Vulnerabilities", []):
            findings.append(_make_finding(
                scanner="trivy",
                file_path=target,
                line=0,
                severity=_normalize_severity(vuln.get("Severity", "MEDIUM")),
                description=f"{vuln.get('VulnerabilityID', 'Unknown')}: {vuln.get('Title', 'Vulnerability detected')}",
                remediation=f"Update {vuln.get('PkgName', 'package')} to {vuln.get('FixedVersion', 'latest patched version')}",
            ))
    return findings


# ---------------------------------------------------------------------------
# Context-file scanner (built-in)
# ---------------------------------------------------------------------------

def _run_context_file_scanner(directory: str) -> list[dict[str, Any]]:
    """Run the built-in context-file injection scanner."""
    raw_findings = ctx_scan_directory(directory)
    findings = []
    for item in raw_findings:
        # Map context-file scanner categories to remediation guidance
        cat = item.get("category", "?")
        remediation_map = {
            "A": "Remove or rewrite the instruction override pattern",
            "B": "Remove the role override / persona hijacking pattern",
            "C": "Remove the privilege escalation / authority spoofing pattern",
            "D": "Remove the data exfiltration pattern",
            "E": "Remove the persistence mechanism",
            "F": "Remove or decode the obfuscated content for review",
            "G": "Remove hidden Unicode characters and replace with ASCII equivalents",
            "H": "Remove the MCP / config file manipulation pattern",
            "I": "Remove hidden instructions from HTML comments",
            "J": "Remove the external fetch + exec pattern",
        }
        findings.append(_make_finding(
            scanner="context-file-scanner",
            file_path=item.get("file_path", "unknown"),
            line=item.get("line_number", 0),
            severity=item.get("severity", "medium"),
            description=item.get("description", "Context file injection pattern detected"),
            remediation=remediation_map.get(cat, "Review and remove the suspicious pattern"),
            category=cat,
        ))
    return findings


# ---------------------------------------------------------------------------
# Repo hygiene (built-in)
# ---------------------------------------------------------------------------

_EXPECTED_GITIGNORE_PATTERNS = [
    (".env", "Secrets in .env files may be committed"),
    ("node_modules", "node_modules/ should be ignored"),
    ("__pycache__", "__pycache__/ should be ignored"),
    (".DS_Store", ".DS_Store files should be ignored"),
]


def _check_gitignore(directory: str) -> list[dict[str, Any]]:
    """Check .gitignore completeness."""
    gitignore_path = os.path.join(directory, ".gitignore")
    findings = []

    if not os.path.isfile(gitignore_path):
        findings.append(_make_finding(
            scanner="repo-hygiene",
            file_path=".gitignore",
            line=0,
            severity="medium",
            description="Missing .gitignore file — secrets and build artifacts may be committed",
            remediation="Create a .gitignore file with patterns for .env, node_modules, __pycache__, .DS_Store",
        ))
        return findings

    try:
        with open(gitignore_path, encoding="utf-8") as f:
            content = f.read()
    except OSError:
        return findings

    for pattern, message in _EXPECTED_GITIGNORE_PATTERNS:
        if pattern not in content:
            findings.append(_make_finding(
                scanner="repo-hygiene",
                file_path=".gitignore",
                line=0,
                severity="medium",
                description=f".gitignore missing '{pattern}' pattern — {message}",
                remediation=f"Add '{pattern}' to .gitignore",
            ))

    return findings


def _check_tracked_env_files(directory: str) -> list[dict[str, Any]]:
    """Check for tracked .env files (not .env.example) via git ls-files."""
    findings = []

    try:
        proc = subprocess.run(
            ["git", "ls-files"],
            capture_output=True,
            text=True,
            timeout=10,
            cwd=directory,
        )
        if proc.returncode != 0:
            return findings
    except (subprocess.TimeoutExpired, OSError):
        return findings

    for line in proc.stdout.strip().split("\n"):
        line = line.strip()
        if not line:
            continue
        basename = os.path.basename(line)
        # Match .env, .env.local, .env.production, etc. but NOT .env.example
        if basename.startswith(".env") and ".example" not in basename and basename != ".env.example":
            # Exclude if it's just ".env.example" in a subdir
            if ".example" not in line:
                findings.append(_make_finding(
                    scanner="repo-hygiene",
                    file_path=line,
                    line=0,
                    severity="critical",
                    description=f"Tracked .env file '{line}' — secrets may be in git history",
                    remediation=f"Remove '{line}' from git tracking: git rm --cached {line}",
                ))

    return findings


def _run_repo_hygiene(directory: str) -> list[dict[str, Any]]:
    """Run all built-in repo hygiene checks."""
    findings = []
    findings.extend(_check_gitignore(directory))
    findings.extend(_check_tracked_env_files(directory))
    return findings


# ---------------------------------------------------------------------------
# Orchestrator
# ---------------------------------------------------------------------------

def run_scan(directory: str) -> ScanResult:
    """Run all available security scanners against a directory.

    Checks scanner availability, runs available scanners, normalizes
    findings, and compiles results.

    Args:
        directory: Path to the project directory to scan.

    Returns:
        ScanResult with all findings, scanner status, and any errors.
    """
    result = ScanResult()
    directory = os.path.abspath(directory)

    # Check external scanner availability
    availability = check_all_scanners()
    scanner_map: dict[str, ScannerResult] = {s.name: s for s in availability}

    # --- External scanners ---

    # Gitleaks (secrets)
    gitleaks = scanner_map.get("gitleaks")
    if gitleaks and gitleaks.available:
        try:
            result.findings.extend(_run_gitleaks(directory, gitleaks))
            result.scanners_run.append("gitleaks")
        except RuntimeError as e:
            result.errors.append(str(e))
    elif gitleaks:
        result.scanners_skipped.append({
            "name": "gitleaks",
            "install_hint": gitleaks.install_hint or "",
        })

    # Dependency scanner: prefer osv-scanner, fallback to trivy
    osv = scanner_map.get("osv-scanner")
    trivy = scanner_map.get("trivy")
    if osv and osv.available:
        try:
            result.findings.extend(_run_osv_scanner(directory, osv))
            result.scanners_run.append("osv-scanner")
        except RuntimeError as e:
            result.errors.append(str(e))
    elif trivy and trivy.available:
        try:
            result.findings.extend(_run_trivy(directory, trivy))
            result.scanners_run.append("trivy")
        except RuntimeError as e:
            result.errors.append(str(e))
    else:
        # Both missing — skip both
        if osv:
            result.scanners_skipped.append({
                "name": "osv-scanner",
                "install_hint": osv.install_hint or "",
            })
        if trivy:
            result.scanners_skipped.append({
                "name": "trivy",
                "install_hint": trivy.install_hint or "",
            })

    # Semgrep (static analysis)
    semgrep = scanner_map.get("semgrep")
    if semgrep and semgrep.available:
        try:
            result.findings.extend(_run_semgrep(directory, semgrep))
            result.scanners_run.append("semgrep")
        except RuntimeError as e:
            result.errors.append(str(e))
    elif semgrep:
        result.scanners_skipped.append({
            "name": "semgrep",
            "install_hint": semgrep.install_hint or "",
        })

    # --- Built-in scanners (always run) ---

    # Context-file scanner
    result.findings.extend(_run_context_file_scanner(directory))
    result.scanners_run.append("context-file-scanner")

    # Repo hygiene
    result.findings.extend(_run_repo_hygiene(directory))
    result.scanners_run.append("repo-hygiene")

    return result


# ---------------------------------------------------------------------------
# Severity bucketing
# ---------------------------------------------------------------------------

def bucket_by_severity(findings: list[dict[str, Any]]) -> dict[str, list[dict[str, Any]]]:
    """Group findings by severity level.

    Returns a dict with keys: critical, high, medium, low.
    Each key maps to a list of findings at that severity.
    Empty severity levels are always present.
    """
    buckets: dict[str, list[dict[str, Any]]] = {
        "critical": [],
        "high": [],
        "medium": [],
        "low": [],
    }
    for finding in findings:
        sev = finding.get("severity", "medium")
        if sev in buckets:
            buckets[sev].append(finding)
        else:
            buckets["medium"].append(finding)
    return buckets


# ---------------------------------------------------------------------------
# Summary line
# ---------------------------------------------------------------------------

def format_summary_line(findings: list[dict[str, Any]], scanners_run: list[str]) -> str:
    """Format a summary line with severity counts and scanner count.

    Example: "1 critical, 2 high, 0 medium, 0 low findings across 4 scanners"
    """
    buckets = bucket_by_severity(findings)
    parts = [f"{len(buckets[sev])} {sev}" for sev in _SEVERITY_ORDER]
    total = len(findings)
    n_scanners = len(scanners_run)
    return f"{', '.join(parts)} findings ({total} total) across {n_scanners} scanners"


# ---------------------------------------------------------------------------
# JSON report
# ---------------------------------------------------------------------------

def format_json_report(result: ScanResult) -> str:
    """Format scan results as a JSON report string.

    Structure:
    {
        "summary": {"critical": N, "high": N, "medium": N, "low": N, "total": N},
        "scanners_run": [...],
        "scanners_skipped": [...],
        "findings": [...]
    }
    """
    buckets = bucket_by_severity(result.findings)
    report = {
        "summary": {
            "critical": len(buckets["critical"]),
            "high": len(buckets["high"]),
            "medium": len(buckets["medium"]),
            "low": len(buckets["low"]),
            "total": len(result.findings),
        },
        "scanners_run": result.scanners_run,
        "scanners_skipped": result.scanners_skipped,
        "findings": result.findings,
    }
    if result.errors:
        report["errors"] = result.errors
    return json.dumps(report, indent=2)


# ---------------------------------------------------------------------------
# Markdown report
# ---------------------------------------------------------------------------

def format_markdown_report(result: ScanResult) -> str:
    """Format scan results as a markdown report string."""
    lines: list[str] = []
    summary = format_summary_line(result.findings, result.scanners_run)

    lines.append("# Security Scan Report")
    lines.append("")
    lines.append(f"**Summary:** {summary}")
    lines.append("")

    # Scanners run
    lines.append("## Scanners")
    lines.append("")
    for name in result.scanners_run:
        lines.append(f"- [x] {name}")
    for skipped in result.scanners_skipped:
        name = skipped["name"]
        hint = skipped.get("install_hint", "")
        lines.append(f"- [ ] {name} — skipped (not found)")
        if hint:
            lines.append(f"  - Install: `{hint}`")
    lines.append("")

    # Findings by severity
    if result.findings:
        buckets = bucket_by_severity(result.findings)
        for sev in _SEVERITY_ORDER:
            sev_findings = buckets[sev]
            if not sev_findings:
                continue
            lines.append(f"## {sev.capitalize()} ({len(sev_findings)})")
            lines.append("")
            for f in sev_findings:
                scanner = f.get("scanner", "unknown")
                fpath = f.get("file_path", "unknown")
                line_num = f.get("line", 0)
                desc = f.get("description", "")
                remed = f.get("remediation", "")
                lines.append(f"### {desc}")
                lines.append("")
                lines.append(f"- **Scanner:** {scanner}")
                lines.append(f"- **File:** `{fpath}`:{line_num}")
                lines.append(f"- **Remediation:** {remed}")
                lines.append("")
    else:
        lines.append("## No findings")
        lines.append("")
        lines.append("No security findings detected. All scanned vectors are clean.")
        lines.append("")

    # Errors
    if result.errors:
        lines.append("## Errors")
        lines.append("")
        for err in result.errors:
            lines.append(f"- {err}")
        lines.append("")

    return "\n".join(lines)


# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------

def main(argv: list[str] | None = None) -> int:
    """CLI entry point. Returns exit code.

    Exit codes:
      0 = no findings
      1 = findings detected
      2 = error
    """
    parser = argparse.ArgumentParser(
        description="Run all security scanners and produce a unified report.",
    )
    parser.add_argument(
        "paths",
        nargs="*",
        default=["."],
        help="Directories to scan (default: current directory)",
    )
    parser.add_argument(
        "--json",
        action="store_true",
        dest="json_output",
        help="Output report as JSON",
    )

    args = parser.parse_args(argv)

    all_findings: list[dict[str, Any]] = []
    had_error = False
    combined_result = ScanResult()

    for path in args.paths:
        if not os.path.isdir(path):
            print(f"Error: directory not found: {path}", file=sys.stderr)
            had_error = True
            continue

        result = run_scan(path)
        combined_result.findings.extend(result.findings)
        # Merge scanners_run (unique)
        for s in result.scanners_run:
            if s not in combined_result.scanners_run:
                combined_result.scanners_run.append(s)
        # Merge scanners_skipped (unique by name)
        existing_skipped = {s["name"] for s in combined_result.scanners_skipped}
        for s in result.scanners_skipped:
            if s["name"] not in existing_skipped:
                combined_result.scanners_skipped.append(s)
                existing_skipped.add(s["name"])
        combined_result.errors.extend(result.errors)

    if had_error and not combined_result.findings:
        return 2

    if args.json_output:
        print(format_json_report(combined_result))
    else:
        print(format_markdown_report(combined_result))

    if combined_result.findings:
        return 1
    return 0


if __name__ == "__main__":
    sys.exit(main())