CASAN/packages/casan-harness/scripts/bash/supply-chain-scan.py

#!/usr/bin/env python3
"""CASAN Track C-MVP — supply-chain manifest diff + risk scan (C2, V18 core).

Diffs a dependency manifest against a baseline and classifies the change.
Prints "<OUTCOME>|<reason>" on stdout and writes a JSON report.

Argv: <manifest> <baseline> <known-packages.txt> <malicious-packages.txt>
      <report.json> <scanners-space-separated>

Outcomes: BLOCK (denylist / typosquat / dangerous lifecycle),
          REQUIRE_APPROVAL (new dependency added), ALLOW (no new deps).
Deterministic: no network; a local denylist is authoritative when no live
scanner is available (their availability is recorded honestly in the report).
"""
import json
import os
import re
import sys


def read(path: str) -> str:
    try:
        return open(path, encoding="utf-8").read()
    except OSError:
        return ""


def load_list(path: str):
    out = []
    for line in read(path).splitlines():
        line = line.strip()
        if line and not line.startswith("#"):
            out.append(line)
    return out


def parse_deps(text: str, name: str):
    """Return ({pkg: version}, [dangerous lifecycle scripts])."""
    deps, scripts = {}, []
    base = os.path.basename(name).lower()
    if base == "package.json" or name.endswith(".json"):
        try:
            data = json.loads(text) if text.strip() else {}
        except ValueError:
            data = {}
        for sect in ("dependencies", "devDependencies",
                     "optionalDependencies", "peerDependencies"):
            for k, v in (data.get(sect) or {}).items():
                deps[k] = str(v)
        for hook in ("preinstall", "install", "postinstall"):
            s = (data.get("scripts") or {}).get(hook)
            if s:
                scripts.append(hook + ":" + s)
    elif base == "requirements.txt" or name.endswith(".txt"):
        for line in text.splitlines():
            line = line.split("#", 1)[0].strip()
            if not line:
                continue
            m = re.match(r"^([A-Za-z0-9._-]+)\s*([=<>!~].*)?$", line)
            if m:
                deps[m.group(1)] = (m.group(2) or "").strip()
    elif base == "pom.xml" or name.endswith(".xml"):
        for m in re.finditer(r"<artifactId>([^<]+)</artifactId>", text):
            deps[m.group(1).strip()] = ""
    elif base.startswith("build.gradle"):
        for m in re.finditer(r"""['"]([\w.\-]+):([\w.\-]+):([\w.\-]+)['"]""", text):
            deps[m.group(1) + ":" + m.group(2)] = m.group(3)
    return deps, scripts


def levenshtein(a: str, b: str) -> int:
    # SEC-15: return a LARGE sentinel (not 2) when lengths are far apart — the old
    # sentinel 2 collided with the widened distance<=2 typosquat threshold and
    # produced false positives (e.g. fastapi vs numpy). For len-diff<=2 the DP below
    # computes the true edit distance.
    if abs(len(a) - len(b)) > 2:
        return 99
    prev = list(range(len(b) + 1))
    for i, ca in enumerate(a, 1):
        row = [i]
        for j, cb in enumerate(b, 1):
            row.append(min(prev[j] + 1, row[-1] + 1, prev[j - 1] + (ca != cb)))
        prev = row
    return prev[-1]


def main() -> int:
    manifest, baseline, known_path, deny_path, report_path, scanners_raw = sys.argv[1:7]
    scanners = [s for s in scanners_raw.split() if s]
    known = load_list(known_path)
    deny = set(load_list(deny_path))

    cur, cur_scripts = parse_deps(read(manifest), manifest)
    base_deps, base_scripts = parse_deps(read(baseline), manifest)

    added = {k: v for k, v in cur.items() if k not in base_deps}
    new_scripts = [s for s in cur_scripts if s not in base_scripts]

    findings = []
    approval = []
    for name, ver in added.items():
        clean_ver = ver.lstrip("=<>!~^ ")
        ident = (name + "@" + clean_ver) if ver else name
        denied = (
            name in deny
            or ident in deny
            or any(d.split("@")[0] == name and "@" in d and d.split("@", 1)[1] in ver for d in deny)
        )
        if denied:
            findings.append({"package": name, "reason": "denylisted_or_known_malicious"})
            continue
        if name not in known:
            # SEC-15: distance==1 missed 2-char typosquats (e.g. reqests/reqeusts of
            # "requests"). Allow distance<=2 for names long enough that a 2-edit match
            # is meaningful (short names stay at 1 to avoid false positives).
            _maxd = 2 if len(name) >= 5 else 1
            near = next((k for k in known if 1 <= levenshtein(name.lower(), k.lower()) <= _maxd), None)
            if near:
                findings.append({"package": name, "reason": "typosquat_of:" + near})
                continue
        approval.append({"package": name, "version": ver})

    for s in new_scripts:
        findings.append({"package": "<lifecycle-script>", "reason": "dangerous_lifecycle:" + s[:120]})

    if findings:
        outcome = "BLOCK"
    elif approval:
        outcome = "REQUIRE_APPROVAL"
    else:
        outcome = "ALLOW"

    report = {
        "manifest": manifest,
        "scanners_available": scanners,
        "scanners_note": "local denylist authoritative when no live scanner present",
        "added": [{"package": k, "version": v} for k, v in added.items()],
        "new_lifecycle_scripts": new_scripts,
        "blocked": findings,
        "require_approval": approval,
        "outcome": outcome,
    }
    with open(report_path, "w", encoding="utf-8") as f:
        json.dump(report, f, indent=2, ensure_ascii=False)

    if findings:
        reason = ";".join(x["package"] + ":" + x["reason"] for x in findings)
    elif approval:
        reason = ",".join(x["package"] for x in approval)
    else:
        reason = "no_new_dependencies"
    print(outcome + "|" + reason)
    return 0


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