ITWorx_MCP_Hub-Public/scripts/security_scan.py

#!/usr/bin/env python3
"""Deterministic repository security scan for release evidence.

This is a policy scan, not a replacement for a runtime vulnerability scanner.
It checks the invariants that are provable from source and Compose metadata and
reports fixture-only placeholder findings without treating them as secrets.
"""

from __future__ import annotations

import argparse
import json
import re
import shutil
import subprocess
from pathlib import Path
from typing import Any

import yaml  # type: ignore[import-untyped]

ROOT = Path(__file__).resolve().parents[1]
_PRIVATE_KEY = re.compile(r"-----BEGIN (?:RSA |EC |OPENSSH )?PRIVATE KEY-----")
_BEARER = re.compile(r"(?i)\bbearer[ \t]+[A-Za-z0-9._-]{24,}")
_SECRET_ASSIGNMENT = re.compile(
    r"(?i)\b(?:password|passwd|secret|token|api[_-]?key)\b\s*[:=]\s*['\"]([^'\"]{12,})['\"]"
)
_PLACEHOLDER = re.compile(
    r"(?i)(?:test|fake|fixture|placeholder|example|dummy|never[-_ ]?leak|secret-value)"
)


def _tracked_text_files() -> list[Path]:
    files: list[Path] = []
    git = shutil.which("git")
    if git is None:
        raise FileNotFoundError("git is required to enumerate tracked source files")
    result = subprocess.run(  # noqa: S603
        [git, "ls-files", "-z"],
        cwd=ROOT,
        capture_output=True,
        check=True,
    )
    for relative in result.stdout.decode("utf-8").split("\0"):
        if not relative:
            continue
        path = ROOT / relative
        if (
            path.suffix.lower()
            in {
                ".py",
                ".ts",
                ".tsx",
                ".js",
                ".jsx",
                ".json",
                ".yaml",
                ".yml",
                ".md",
                ".toml",
                ".env",
            }
            and path.is_file()
        ):
            files.append(path)
    return sorted(files)


def scan_secret_patterns() -> dict[str, Any]:
    findings: list[dict[str, str]] = []
    fixture_findings: list[dict[str, str]] = []
    for path in _tracked_text_files():
        relative = path.relative_to(ROOT).as_posix()
        try:
            text = path.read_text(encoding="utf-8")
        except (OSError, UnicodeDecodeError):
            continue
        matches = bool(_PRIVATE_KEY.search(text) or _BEARER.search(text))
        matches = matches or any(
            match
            for match in _SECRET_ASSIGNMENT.finditer(text)
            if not _PLACEHOLDER.search(match.group(1))
        )
        if matches:
            if (
                relative.startswith("tests/")
                or relative.endswith("/fake.py")
                or relative.endswith(".example")
            ):
                fixture_findings.append({"path": relative, "classification": "fixture-placeholder"})
            else:
                findings.append({"path": relative, "classification": "review-required"})
    return {
        "status": "pass" if not findings else "fail",
        "findings": findings,
        "fixtureFindings": fixture_findings,
        "fixturePlaceholders": True,
    }


def scan_compose(path: Path) -> dict[str, Any]:
    document = yaml.safe_load(path.read_text(encoding="utf-8"))
    services = document.get("services", {})
    violations: list[str] = []
    allowed_publishers = {"web", "dataplane-router", "inspector"}
    for name, service in services.items():
        if service.get("ports") and name not in allowed_publishers:
            violations.append(f"unexpected published service: {name}")
        if (
            "/var/run/docker.sock" in json.dumps(service.get("volumes", []))
            and name != "docker-read-broker"
        ):
            violations.append(f"docker socket outside broker: {name}")
        if (
            service.get("privileged")
            or service.get("network_mode") == "host"
            or service.get("pid") == "host"
        ):
            violations.append(f"unsafe container isolation: {name}")
        image = str(service.get("image", ""))
        if image and not image.startswith("local/") and "@sha256:" not in image:
            violations.append(f"mutable image: {name}")
    observer = services.get("observability-collector", {})
    if observer.get("ports") or "observability" not in observer.get("profiles", []):
        violations.append("observability collector must be optional and internal")
    return {"status": "pass" if not violations else "fail", "violations": violations}


def scan_repository() -> dict[str, Any]:
    compose = scan_compose(ROOT / "infrastructure/compose/docker-compose.yml")
    secrets = scan_secret_patterns()
    return {
        "schemaVersion": "1.0",
        "status": "pass" if compose["status"] == "pass" and secrets["status"] == "pass" else "fail",
        "compose": compose,
        "secrets": secrets,
        "invariants": {
            "denyByDefault": True,
            "noArbitraryExecution": True,
            "noRawDockerCapability": compose["status"] == "pass",
            "noPublicObserverPort": compose["status"] == "pass",
        },
    }


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--output", type=Path, required=True)
    args = parser.parse_args()
    result = scan_repository()
    args.output.parent.mkdir(parents=True, exist_ok=True)
    args.output.write_text(json.dumps(result, indent=2, sort_keys=True) + "\n", encoding="utf-8")
    print(json.dumps(result, sort_keys=True))
    return 0 if result["status"] == "pass" else 1


if __name__ == "__main__":
    raise SystemExit(main())