#!/usr/bin/env python3
"""dep_prescan.py — deterministic dependency supply-chain pre-scan.

ARION service artifact. Scans dependency manifests for the classes of
supply-chain risk that are checkable without a network:

  * unpinned / floating versions (``*``, ``latest``, bare ``>=``,
    branch-tracking git refs) — the payload can change without an edit
  * non-registry sources (git+, ssh, http://, file:, tarball URLs) —
    integrity not verifiable by the registry
  * install-time code execution (npm pre/install/postinstall/prepare,
    setup.py-dynamic requirements, Cargo build scripts)
  * typosquat proximity: dependency names within edit distance 1 of a
    high-value package on the same ecosystem (plus one confusable-char
    pass) — the classic `reqeusts`/`lodas` class
  * path/replace overrides: Cargo ``[patch]``, go.mod ``replace`` to a
    filesystem path, pip ``-e``/``file:`` — silent redirection to
    attacker-controlled code
  * missing lockfile when a manifest directory is scanned
  * insecure transports (``http://`` index/source URLs)

Honest scope: PRE-SCAN, not an audit and not a vulnerability feed. It does
not resolve the dependency graph, does not check CVEs, and does not prove a
package is malicious — a FLAG means "this property deserves human review".
Absence of flags is not proof of safety (e.g. a freshly published malicious
version of a correctly-pinned package is out of scope).

Usage:
  dep_prescan.py <manifest-or-dir> [--json]
  dep_prescan.py --selftest
Stdlib only. Exit 0 = no HIGH/MEDIUM findings, 1 = findings, 2 = usage error.
"""
import json
import os
import re
import sys

# ---------------------------------------------------------------- model

SEV_ORDER = {"HIGH": 0, "MEDIUM": 1, "LOW": 2, "INFO": 3}

# High-value package names per ecosystem (attack targets). Kept short and
# curated; the check is edit-distance-1 + confusables, not a full index.
TOP_NPM = [
    "react", "react-dom", "lodash", "express", "axios", "chalk", "commander",
    "typescript", "webpack", "babel", "eslint", "prettier", "next", "vue",
    "angular", "jquery", "moment", "debug", "async", "request", "underscore",
    "rxjs", "tslib", "inquirer", "yargs", "glob", "minimist", "semver",
    "fs-extra", "chokidar", "node-fetch", "cross-env", "dotenv", "colors",
    "ora", "uuid", "nanoid", "express-validator", "mongoose", "pg", "redis",
    "jsonwebtoken", "bcrypt", "passport", "socket.io", "ws", "cors", "helmet",
    "morgan", "body-parser", "cookie-parser", "multer", "sharp", "puppeteer",
    "ethers", "web3", "solana", "bn.js", "elliptic", "tweetnacl",
]
TOP_PYPI = [
    "requests", "urllib3", "numpy", "pandas", "django", "flask", "sqlalchemy",
    "boto3", "botocore", "setuptools", "pip", "wheel", "pytest", "click",
    "rich", "pydantic", "fastapi", "aiohttp", "httpx", "cryptography",
    "pyopenssl", "paramiko", "fabric", "ansible", "celery", "redis",
    "pymongo", "psycopg2", "mysql-connector-python", "pillow", "matplotlib",
    "scipy", "scikit-learn", "tensorflow", "torch", "transformers", "openai",
    "anthropic", "langchain", "selenium", "scrapy", "beautifulsoup4", "lxml",
    "pyyaml", "toml", "jinja2", "markupsafe", "werkzeug", "gunicorn",
    "uvicorn", "starlette", "python-dotenv", "colorama", "tqdm", "dateutil",
    "python-dateutil", "six", "certifi", "charset-normalizer", "idna",
]
TOP_CARGO = [
    "serde", "serde_json", "tokio", "reqwest", "hyper", "clap", "anyhow",
    "thiserror", "rand", "log", "env_logger", "chrono", "regex", "lazy_static",
    "once_cell", "parking_lot", "crossbeam", "rayon", "futures", "async-std",
    "actix-web", "axum", "warp", "rocket", "diesel", "sqlx", "rusqlite",
    "openssl", "rustls", "ring", "sha2", "hmac", "aes", "rsa", "ed25519-dalek",
    "secp256k1", "web3", "ethers", "solana-sdk", "anchor-lang", "bincode",
    "prost", "tonic", "tracing", "tracing-subscriber", "config", "dotenv",
]
TOP_GO = [
    "github.com/gorilla/mux", "github.com/gin-gonic/gin", "github.com/sirupsen/logrus",
    "github.com/spf13/cobra", "github.com/spf13/viper", "github.com/stretchr/testify",
    "github.com/golang/protobuf", "google.golang.org/grpc", "github.com/go-sql-driver/mysql",
    "github.com/lib/pq", "github.com/redis/go-redis", "github.com/ethereum/go-ethereum",
    "github.com/gagliardetto/solana-go", "github.com/dgrijalva/jwt-go",
    "github.com/golang-jwt/jwt", "github.com/labstack/echo", "github.com/prometheus/client_golang",
    "k8s.io/client-go", "k8s.io/apimachinery", "sigs.k8s.io/controller-runtime",
]

INSTALL_SCRIPTS = {"preinstall", "install", "postinstall", "prepare", "prepublish",
                   "prepack", "postpack"}


def lev1(a, b):
    """True if a is within Damerau-Levenshtein distance 1 of b
    (one substitution, insertion, deletion, or adjacent transposition)."""
    if a == b:
        return False
    la, lb = len(a), len(b)
    if abs(la - lb) > 1:
        return False
    if la == lb:
        diffs = [i for i in range(la) if a[i] != b[i]]
        if len(diffs) == 1:
            return True
        # adjacent transposition: exactly two diffs, swapped neighbours
        if len(diffs) == 2 and diffs[1] == diffs[0] + 1:
            i, j = diffs
            return a[i] == b[j] and a[j] == b[i]
        return False
    # length differs by 1: one insertion/deletion
    if la > lb:
        a, b = b, a
        la, lb = lb, la
    i = j = 0
    skips = 0
    while i < la and j < lb:
        if a[i] == b[j]:
            i += 1
            j += 1
        else:
            j += 1
            skips += 1
            if skips > 1:
                return False
    return True


def confusable_variants(name):
    """Names that normalize to the same string under common confusables."""
    n = name.lower()
    out = set()
    repl = {"0": "o", "1": "l", "3": "e", "5": "s", "vv": "w", "rn": "m",
            "cl": "d", "ij": "y"}
    for bad, good in repl.items():
        if bad in n:
            out.add(n.replace(bad, good))
    return out


def typosquat_hit(name, top):
    """Return the look-alike top-package name, or None."""
    n = name.lower()
    if n in top:
        return None
    for t in top:
        if lev1(n, t):
            return t
    for t in top:
        if t in confusable_variants(n):
            return t
    return None


def finding(sev, rule, where, detail):
    return {"severity": sev, "rule": rule, "where": where, "detail": detail}


# ---------------------------------------------------------------- npm

def scan_package_json(path):
    out = []
    try:
        data = json.load(open(path, encoding="utf-8"))
    except Exception as e:
        return [finding("INFO", "parse-error", path, "not valid JSON: %s" % e)]
    deps = {}
    for sec in ("dependencies", "devDependencies", "optionalDependencies",
                "peerDependencies"):
        for name, ver in (data.get(sec) or {}).items():
            deps.setdefault(name, []).append((sec, str(ver)))
    for name, uses in sorted(deps.items()):
        hit = typosquat_hit(name, TOP_NPM)
        if hit:
            out.append(finding("HIGH", "typosquat-proximity", name,
                               "within edit distance 1 / confusable of popular npm package '%s'" % hit))
        for sec, ver in uses:
            v = ver.strip().lower()
            where = "%s (%s)" % (name, sec)
            if v in ("*", "latest", "") or v == "x":
                out.append(finding("MEDIUM", "unpinned-version", where,
                                   "floating version '%s' — payload can change silently" % ver))
            elif v.startswith(("http://",)):
                out.append(finding("HIGH", "insecure-source", where,
                                   "dependency fetched over plaintext http"))
            elif v.startswith(("git+", "git://", "ssh://", "github:", "gitlab:",
                               "bitbucket:", "https://github.com", "https://gitlab.com")):
                ref = "no ref" if "#" not in v else v.split("#", 1)[1]
                sev = "MEDIUM"
                if re.search(r"#(main|master|head|dev|develop)$", v) or "#" not in v:
                    sev = "HIGH"
                out.append(finding(sev, "vcs-dependency", where,
                                   "git/VCS source ref='%s' — integrity not registry-verified" % ref))
            elif v.startswith(("file:", "link:", "workspace:")):
                out.append(finding("LOW", "local-dependency", where,
                                   "local path dep '%s' — fine in monorepo, flag if unexpected" % ver))
            elif v.startswith((">=", ">", "<=", "<")):
                out.append(finding("LOW", "open-range", where,
                                   "unbounded range '%s'" % ver))
            elif v.startswith(("^", "~")):
                out.append(finding("INFO", "semver-range", where,
                                   "semver range '%s' — lockfile required for integrity" % ver))
    scripts = data.get("scripts") or {}
    for s in sorted(scripts):
        if s in INSTALL_SCRIPTS:
            out.append(finding("MEDIUM", "install-script", "scripts.%s" % s,
                               "install-time code execution: %s" % scripts[s][:120]))
    return out


# ---------------------------------------------------------------- pypi

REQ_LINE = re.compile(r"^\s*([A-Za-z0-9._\-]+)\s*(.*)$")


def scan_requirements(path):
    out = []
    for n, raw in enumerate(open(path, encoding="utf-8", errors="replace"), 1):
        line = raw.strip()
        if not line or line.startswith("#"):
            continue
        where = "%s:%d" % (os.path.basename(path), n)
        if line.startswith(("--index-url", "--extra-index-url")):
            if "http://" in line:
                out.append(finding("HIGH", "insecure-index", where,
                                   "plaintext http index URL — MITM-installable packages"))
            else:
                out.append(finding("LOW", "extra-index", where,
                                   "non-default index — dependency-confusion surface"))
            continue
        if line.startswith(("-e", "--editable")) or "git+" in line:
            ref = line.rsplit("@", 1)[-1] if "@" in line else "no ref"
            sev = "HIGH" if ref in ("no ref", "main", "master", "HEAD") else "MEDIUM"
            out.append(finding(sev, "vcs-dependency", where,
                               "VCS/editable install ref='%s'" % ref))
            continue
        if line.startswith(("http://", "https://")) or "://" in line:
            sev = "HIGH" if line.startswith("http://") else "MEDIUM"
            out.append(finding(sev, "direct-url", where,
                               "direct URL dependency — not registry-verified"))
            continue
        m = REQ_LINE.match(line)
        if not m:
            continue
        name, rest = m.group(1), m.group(2)
        hit = typosquat_hit(name, TOP_PYPI)
        if hit:
            out.append(finding("HIGH", "typosquat-proximity", where,
                               "'%s' looks like popular package '%s'" % (name, hit)))
        r = rest.strip()
        if not r:
            out.append(finding("MEDIUM", "unpinned-version", where,
                               "'%s' has no version pin" % name))
        elif r.startswith((">=", ">", "<=", "<", "~=", "!=")) and "==" not in r:
            out.append(finding("LOW", "open-range", where,
                               "'%s %s' is not an exact pin" % (name, r)))
    return out


# ---------------------------------------------------------------- cargo

def scan_cargo_toml(path):
    out = []
    section = None
    in_patch = False
    for n, raw in enumerate(open(path, encoding="utf-8", errors="replace"), 1):
        line = raw.strip()
        where = "%s:%d" % (os.path.basename(path), n)
        if line.startswith("["):
            section = line.strip("[]").strip()
            in_patch = section.startswith("patch")
            continue
        if not line or line.startswith("#") or "=" not in line:
            continue
        key, _, val = line.partition("=")
        key, val = key.strip(), val.strip().strip('"')
        if in_patch:
            out.append(finding("HIGH", "patch-override", where,
                               "[patch] overrides '%s' -> %s — silent source redirection" % (key, val[:80])))
            continue
        if section not in ("dependencies", "dev-dependencies", "build-dependencies") \
                and not (section and section.startswith("dependencies.")):
            continue
        if val.startswith("{"):
            if 'git' in val:
                ref = "no ref"
                for tag in ("rev", "tag"):
                    m = re.search(tag + r'\s*=\s*"([^"]+)"', val)
                    if m:
                        ref = m.group(1)
                sev = "HIGH" if ref == "no ref" or "branch" in val else "MEDIUM"
                out.append(finding(sev, "vcs-dependency", where,
                                   "git dep '%s' ref=%s" % (key, ref)))
            if 'path' in val:
                out.append(finding("LOW", "local-dependency", where,
                                   "path dep '%s'" % key))
            vm = re.search(r'version\s*=\s*"([^"]+)"', val)
            if vm and vm.group(1) in ("*", ""):
                out.append(finding("MEDIUM", "unpinned-version", where,
                                   "wildcard version on '%s'" % key))
            continue
        hit = typosquat_hit(key, TOP_CARGO)
        if hit:
            out.append(finding("HIGH", "typosquat-proximity", where,
                               "'%s' looks like popular crate '%s'" % (key, hit)))
        if val in ("*", ""):
            out.append(finding("MEDIUM", "unpinned-version", where,
                               "wildcard version '%s'" % val))
        elif val.startswith((">=", ">", "<=", "<")):
            out.append(finding("LOW", "open-range", where,
                               "unbounded range '%s'" % val))
    return out


# ---------------------------------------------------------------- go

def scan_go_mod(path):
    out = []
    in_replace = False
    for n, raw in enumerate(open(path, encoding="utf-8", errors="replace"), 1):
        line = raw.strip()
        where = "%s:%d" % (os.path.basename(path), n)
        if line.startswith("replace"):
            in_replace = True
            if "(" not in line and "=>" in line:
                _check_go_replace(out, line[7:].strip(), where)
                in_replace = False
            continue
        if in_replace:
            if line.startswith(")"):
                in_replace = False
                continue
            if "=>" in line:
                _check_go_replace(out, line, where)
            continue
        m = re.match(r"^([A-Za-z0-9._~\-/]+)\s+v\S+", line)
        if m and not line.startswith(("module", "go ", "toolchain")):
            hit = typosquat_hit(m.group(1), TOP_GO)
            if hit:
                out.append(finding("HIGH", "typosquat-proximity", where,
                                   "'%s' looks like '%s'" % (m.group(1), hit)))
    return out


def _check_go_replace(out, line, where):
    left, _, right = line.partition("=>")
    right = right.strip()
    if right.startswith(("./", "../", "/")) or ":\\" in right:
        out.append(finding("HIGH", "replace-path", where,
                           "go.mod replace to local path: %s — redirects module to filesystem" % line[:100]))
    else:
        out.append(finding("MEDIUM", "replace-remote", where,
                           "go.mod replace directive: %s — verify the target is intended" % line[:100]))


# ---------------------------------------------------------------- driver

MANIFESTS = {
    "package.json": ("npm", scan_package_json, ["package-lock.json", "yarn.lock",
                                                "pnpm-lock.yaml", "npm-shrinkwrap.json"]),
    "requirements.txt": ("pypi", scan_requirements, ["Pipfile.lock", "poetry.lock",
                                                     "requirements.lock"]),
    "Cargo.toml": ("cargo", scan_cargo_toml, ["Cargo.lock"]),
    "go.mod": ("go", scan_go_mod, ["go.sum"]),
    "pyproject.toml": ("pypi", None, ["poetry.lock", "uv.lock", "Pipfile.lock"]),
    "Pipfile": ("pypi", None, ["Pipfile.lock"]),
    "setup.py": ("pypi", None, []),
}


def scan_target(target):
    findings = []
    scanned = []
    if os.path.isdir(target):
        for base in sorted(MANIFESTS):
            p = os.path.join(target, base)
            if os.path.exists(p):
                scanned.append((base, p))
        if not scanned:
            return findings, scanned, ["no known manifests in %s" % target]
        for base, p in scanned:
            eco, fn, locks = MANIFESTS[base]
            if fn:
                findings += fn(p)
            if locks and not any(os.path.exists(os.path.join(target, l)) for l in locks):
                findings.append(finding("MEDIUM", "missing-lockfile", base,
                                        "%s manifest present but none of %s — installs are non-deterministic"
                                        % (base, locks)))
        # build.rs next to Cargo.toml
        if os.path.exists(os.path.join(target, "build.rs")):
            findings.append(finding("MEDIUM", "build-script", "build.rs",
                                    "Cargo build script executes arbitrary code at build time"))
        return findings, scanned, []
    base = os.path.basename(target)
    if base not in MANIFESTS or not MANIFESTS[base][1]:
        return findings, scanned, ["unsupported manifest: %s (supported: package.json, requirements.txt, Cargo.toml, go.mod, or a directory)" % base]
    scanned.append((base, target))
    findings += MANIFESTS[base][1](target)
    return findings, scanned, []


def render_md(findings, scanned, notes, target):
    order = {s: i for i, s in enumerate(SEV_ORDER)}
    findings = sorted(findings, key=lambda f: order[f["severity"]])
    lines = ["# Dependency Supply-Chain Pre-Scan", "",
             "Target: `%s`" % target,
             "Scanned: %s" % (", ".join("`%s`" % b for b, _ in scanned) or "none"),
             "", "**Scope:** deterministic pre-scan — NOT an audit, no CVE feed, "
             "no graph resolution. A FLAG means \"deserves human review\"; a clean "
             "report is not proof of safety.", ""]
    for n in notes:
        lines.append("> note: %s" % n)
    if notes:
        lines.append("")
    counts = {}
    for f in findings:
        counts[f["severity"]] = counts.get(f["severity"], 0) + 1
    lines.append("**Findings:** " + (", ".join("%s=%d" % (s, counts[s])
                 for s in SEV_ORDER if s in counts) or "none"))
    lines.append("")
    for f in findings:
        lines.append("- **%s** `%s` — %s\n  - %s" %
                     (f["severity"], f["where"], f["rule"], f["detail"]))
    return "\n".join(lines) + "\n"


def selftest():
    import tempfile
    fails = []

    def check(name, cond):
        if not cond:
            fails.append(name)

    d = tempfile.mkdtemp()
    # npm fixture
    pj = os.path.join(d, "package.json")
    json.dump({
        "dependencies": {
            "react": "^18.2.0", "lodas": "1.0.0", "underscore": "*",
            "evil-git": "git+https://github.com/x/y#main",
            "local": "file:../x", "axios": ">=0.27.0",
            "badhttp": "http://example.com/x.tgz",
        },
        "scripts": {"postinstall": "node steal.js", "test": "echo ok"},
    }, open(pj, "w"))
    f, sc, notes = scan_target(d)
    rules = {(x["rule"], x["severity"]) for x in f}
    check("npm typosquat", ("typosquat-proximity", "HIGH") in rules)
    check("npm wildcard", ("unpinned-version", "MEDIUM") in rules)
    check("npm git#main HIGH", ("vcs-dependency", "HIGH") in rules)
    check("npm http HIGH", ("insecure-source", "HIGH") in rules)
    check("npm postinstall", ("install-script", "MEDIUM") in rules)
    check("npm lockfile", ("missing-lockfile", "MEDIUM") in rules)
    check("npm react no fp", not any(x["where"].startswith("react ")
          and x["severity"] in ("HIGH", "MEDIUM") for x in f))

    # clean npm
    d2 = tempfile.mkdtemp()
    json.dump({"dependencies": {"react": "18.2.0"}},
              open(os.path.join(d2, "package.json"), "w"))
    open(os.path.join(d2, "package-lock.json"), "w").write("{}")
    f2, _, _ = scan_target(d2)
    check("clean npm no HIGH/MED",
          not any(x["severity"] in ("HIGH", "MEDIUM") for x in f2))

    # requirements fixture
    d3 = tempfile.mkdtemp()
    rq = os.path.join(d3, "requirements.txt")
    open(rq, "w").write(
        "requests==2.31.0\n"
        "reqeusts\n"
        "flask\n"
        "git+https://github.com/x/y@master\n"
        "http://evil.example.com/pkg-1.0.tar.gz\n"
        "--index-url http://insecure.example.com/simple\n"
        "numpy>=1.0\n")
    f3, _, _ = scan_target(d3)
    r3 = {(x["rule"], x["severity"]) for x in f3}
    check("pypi typosquat", ("typosquat-proximity", "HIGH") in r3)
    check("pypi unpinned", ("unpinned-version", "MEDIUM") in r3)
    check("pypi vcs master", ("vcs-dependency", "HIGH") in r3)
    check("pypi http url", ("direct-url", "HIGH") in r3)
    check("pypi http index", ("insecure-index", "HIGH") in r3)
    check("pypi range low", ("open-range", "LOW") in r3)

    # cargo fixture
    d4 = tempfile.mkdtemp()
    ct = os.path.join(d4, "Cargo.toml")
    open(ct, "w").write(
        "[package]\nname = \"x\"\nversion = \"0.1.0\"\n\n"
        "[dependencies]\nserde = \"1\"\nserdi = \"1\"\nrand = \"*\"\n"
        "evil = { git = \"https://github.com/x/y\" }\n"
        "local = { path = \"../local\" }\n\n"
        "[patch.crates-io]\nserde = { git = \"https://github.com/attacker/serde\" }\n")
    open(os.path.join(d4, "build.rs"), "w").write("fn main(){}")
    f4, _, _ = scan_target(d4)
    r4 = {(x["rule"], x["severity"]) for x in f4}
    check("cargo typosquat", ("typosquat-proximity", "HIGH") in r4)
    check("cargo wildcard", ("unpinned-version", "MEDIUM") in r4)
    check("cargo git noref", ("vcs-dependency", "HIGH") in r4)
    check("cargo patch", ("patch-override", "HIGH") in r4)
    check("cargo build.rs", ("build-script", "MEDIUM") in r4)
    check("cargo lock", ("missing-lockfile", "MEDIUM") in r4)

    # go.mod fixture
    d5 = tempfile.mkdtemp()
    gm = os.path.join(d5, "go.mod")
    open(gm, "w").write(
        "module example.com/x\n\ngo 1.21\n\n"
        "require github.com/gorilla/mux v1.8.0\n\n"
        "replace github.com/gorilla/mux => ../local/mux\n")
    f5, _, _ = scan_target(d5)
    r5 = {(x["rule"], x["severity"]) for x in f5}
    check("go replace path", ("replace-path", "HIGH") in r5)

    # lev1 sanity
    check("lev1 del", lev1("requets", "request"))
    check("lev1 sub", lev1("reqeusts", "requests"))
    check("lev1 ins", lev1("requestss", "requests"))
    check("lev1 no", not lev1("requests", "requests"))
    check("lev1 far", not lev1("abcde", "requests"))
    check("confusable digit", "requests" in confusable_variants("reque5ts"))
    check("confusable rn", "moment" in confusable_variants("rnoment"))

    if fails:
        print("SELFTEST FAIL: %s" % ", ".join(fails))
        return 1
    print("SELFTEST PASS (all vectors)")
    return 0


def main():
    args = sys.argv[1:]
    if "--selftest" in args:
        return selftest()
    if not args or args[0].startswith("-h"):
        print(__doc__)
        return 2
    as_json = "--json" in args
    target = [a for a in args if not a.startswith("--")][0]
    findings, scanned, notes = scan_target(target)
    if as_json:
        print(json.dumps({"target": target, "scanned": [b for b, _ in scanned],
                          "notes": notes, "findings": findings}, indent=2))
    else:
        sys.stdout.write(render_md(findings, scanned, notes, target))
    return 1 if any(f["severity"] in ("HIGH", "MEDIUM") for f in findings) else 0


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