dependency_scanner.py

"""
dependency_scanner.py — Dependency Vulnerability Scanner
===========================================================
Additive module. Scans manifest files (requirements.txt, package.json,
package-lock.json, Pipfile, poetry.lock) for outdated or known-vulnerable
package versions using a built-in advisory table.

No network calls — fully offline, deterministic, fast. This keeps repo
scans within the existing time budget and avoids external API rate limits.

Public API:
  scan_dependencies(base_dir)  → list of dependency Finding dicts
  count_dependencies(base_dir) → total dependency count across all manifests
"""

import os
import re
import json
import logging

logger = logging.getLogger("safeaiscan.dependency_scanner")


# ══════════════════════════════════════════════════════════════
#  KNOWN-VULNERABLE PACKAGE TABLE
#  Format: package_name (lowercase) -> list of (max_vulnerable_version, severity, cve, description)
#  "max_vulnerable_version" means: any version <= this is considered vulnerable.
#  This is a curated subset covering high-impact, widely-known CVEs —
#  not a full NVD mirror, but enough to demonstrate real dependency risk
#  without a network dependency.
# ══════════════════════════════════════════════════════════════

def _v(s: str) -> tuple:
    """Parse a version string into a tuple of ints for comparison. Best-effort."""
    parts = re.findall(r'\d+', s)
    return tuple(int(p) for p in parts) if parts else (0,)


_VULN_DB: dict = {
    # Python — requirements.txt / Pipfile / poetry.lock
    "django": [
        ("2.2.27",  "CRITICAL", "CVE-2022-28346", "SQL injection via QuerySet.annotate(), aggregate(), and extra() on PostgreSQL."),
        ("3.2.12",  "HIGH",     "CVE-2022-28347", "SQL injection via QuerySet.explain() on PostgreSQL/SQLite/Oracle."),
        ("4.0.3",   "HIGH",     "CVE-2022-34265", "SQL injection via Trunc() and Extract() with crafted kind/lookup_name."),
    ],
    "flask": [
        ("0.12.2", "HIGH", "CVE-2018-1000656", "Denial of service via crafted JSON in request."),
    ],
    "requests": [
        ("2.19.1", "MEDIUM", "CVE-2018-18074", "Authorization header leaked on cross-domain redirects."),
        ("2.31.0", "MEDIUM", "CVE-2024-35195", "Session verify=False persists across requests after redirect."),
    ],
    "urllib3": [
        ("1.26.17", "HIGH", "CVE-2023-45803", "Cookie header not stripped on cross-origin redirect."),
        ("2.0.6",   "HIGH", "CVE-2023-43804", "Cookie header leaked via redirect to different origin."),
    ],
    "pyyaml": [
        ("5.3.1", "CRITICAL", "CVE-2020-1747", "yaml.load() allows arbitrary code execution via crafted YAML."),
        ("5.4",   "HIGH",     "CVE-2020-14343", "Full execution via yaml.full_load on untrusted input."),
    ],
    "jinja2": [
        ("2.11.2", "HIGH", "CVE-2020-28493", "ReDoS via crafted email address in urlize filter."),
        ("3.1.2",  "MEDIUM", "CVE-2024-22195", "XSS via xmlattr filter with crafted keys."),
    ],
    "cryptography": [
        ("3.3.1", "HIGH", "CVE-2020-36242", "Bleichenbacher timing oracle attack on RSA decryption."),
        ("41.0.2", "MEDIUM", "CVE-2023-38325", "Memory exhaustion via crafted PKCS7 / X509."),
    ],
    "pillow": [
        ("9.0.0",  "CRITICAL", "CVE-2022-22817", "Arbitrary code execution via ImageMath.eval()."),
        ("9.3.0",  "HIGH",     "CVE-2022-45198", "DoS via crafted FLI file leading to large memory allocation."),
        ("10.0.0", "HIGH",     "CVE-2023-44271", "Heap buffer overflow in path drawing."),
    ],
    "fastapi": [
        ("0.65.2", "MEDIUM", "CVE-2021-32677", "ReDoS in regex used for CORS handling."),
    ],
    "sqlalchemy": [
        ("1.3.24", "HIGH", "CVE-2019-7164", "SQL injection via crafted order_by clause."),
    ],
    "paramiko": [
        ("2.10.1", "CRITICAL", "CVE-2022-24302", "Race condition allows privilege escalation via agent forwarding."),
    ],
    "lxml": [
        ("4.6.5", "HIGH", "CVE-2022-2309", "NULL pointer dereference via crafted XML."),
    ],
    "pyjwt": [
        ("1.7.1", "CRITICAL", "CVE-2022-29217", "Algorithm confusion allows forged tokens via 'none' or mismatched alg."),
    ],
    "werkzeug": [
        ("2.2.3", "HIGH", "CVE-2023-25577", "Multipart form data DoS via large file uploads."),
        ("2.3.7", "MEDIUM", "CVE-2023-46136", "Resource exhaustion via crafted multipart boundaries."),
    ],
    "aiohttp": [
        ("3.9.0", "HIGH", "CVE-2023-49081", "HTTP request smuggling via inconsistent chunked encoding parsing."),
    ],

    # JavaScript / Node — package.json / package-lock.json
    "lodash": [
        ("4.17.20", "HIGH", "CVE-2020-8203", "Prototype pollution via zipObjectDeep."),
        ("4.17.21", "CRITICAL", "CVE-2021-23337", "Command injection via template function."),
    ],
    "express": [
        ("4.17.2", "MEDIUM", "CVE-2022-24999", "DoS via crafted query string parsing (qs dependency)."),
    ],
    "axios": [
        ("0.21.1", "HIGH", "CVE-2021-3749", "ReDoS via crafted trim regex."),
        ("1.5.1",  "MEDIUM", "CVE-2023-45857", "Cross-site request forgery via baseURL absolute URL bypass."),
    ],
    "minimist": [
        ("1.2.5", "CRITICAL", "CVE-2021-44906", "Prototype pollution via crafted argument keys."),
    ],
    "node-fetch": [
        ("2.6.6", "MEDIUM", "CVE-2022-0235", "Information exposure via redirect to different protocol."),
    ],
    "moment": [
        ("2.29.3", "HIGH", "CVE-2022-24785", "Path traversal via crafted locale string."),
    ],
    "jsonwebtoken": [
        ("8.5.1", "CRITICAL", "CVE-2022-23529", "Arbitrary code execution via crafted secret/key argument."),
        ("9.0.0", "HIGH",     "CVE-2022-23540", "Algorithm confusion allows signature forgery."),
    ],
    "next": [
        ("12.1.0", "HIGH", "CVE-2022-21703", "Server-side request forgery in image optimisation."),
        ("13.4.1", "MEDIUM", "CVE-2023-46298", "DoS via crafted cache-control headers."),
    ],
    "ws": [
        ("7.4.6", "HIGH", "CVE-2021-32640", "DoS via crafted Sec-WebSocket-Extensions header."),
    ],
    "semver": [
        ("7.5.1", "HIGH", "CVE-2022-25883", "ReDoS via crafted version range string."),
    ],
    "tar": [
        ("6.1.11", "HIGH", "CVE-2021-37713", "Path traversal allows writing outside extraction directory."),
    ],
    "follow-redirects": [
        ("1.15.3", "MEDIUM", "CVE-2023-26159", "Improper handling of absolute URL in redirect."),
    ],
    "json5": [
        ("1.0.1", "HIGH", "CVE-2022-46175", "Prototype pollution via crafted JSON5 input."),
    ],
}

# Outdated-but-not-CVE table: packages flagged as "should upgrade" even
# without a specific known CVE, because the major version is end-of-life
# or significantly behind current.
_EOL_VERSIONS: dict = {
    "django":  (2, 2),     # Django 2.x is EOL
    "flask":   (1, 1),     # Flask 1.x is old
    "react":   (16, 0),    # React 16 — consider 18+
    "vue":     (2, 0),     # Vue 2 EOL Dec 2023
    "angular": (10, 0),
    "node":    (14, 0),    # Node 14 EOL
    "python":  (3, 8),     # informational
}


# ══════════════════════════════════════════════════════════════
#  MANIFEST PARSERS
# ══════════════════════════════════════════════════════════════

def _parse_requirements_txt(path: str) -> dict:
    """Parse requirements.txt — returns {package_name_lower: version_str}."""
    deps = {}
    try:
        with open(path, "r", encoding="utf-8", errors="replace") as fh:
            for line in fh:
                line = line.strip()
                if not line or line.startswith("#") or line.startswith("-"):
                    continue
                # Match: package==1.2.3 / package>=1.2.3 / package~=1.2
                m = re.match(r'^([A-Za-z0-9_.\-]+)\s*([=<>~!]{1,2})\s*([0-9][A-Za-z0-9.\-]*)', line)
                if m:
                    name, _op, ver = m.groups()
                    deps[name.lower().replace("_", "-")] = ver
                else:
                    # Package with no version pin
                    m2 = re.match(r'^([A-Za-z0-9_.\-]+)', line)
                    if m2:
                        deps[m2.group(1).lower().replace("_", "-")] = ""
    except Exception as e:
        logger.debug(f"_parse_requirements_txt({path}): {e}")
    return deps


def _parse_package_json(path: str) -> dict:
    """Parse package.json dependencies + devDependencies."""
    deps = {}
    try:
        with open(path, "r", encoding="utf-8", errors="replace") as fh:
            data = json.load(fh)
        for section in ("dependencies", "devDependencies"):
            for name, ver in (data.get(section) or {}).items():
                # Strip ^ ~ >= etc.
                clean = re.sub(r'^[\^~>=<\s]+', '', str(ver))
                deps[name.lower()] = clean
    except Exception as e:
        logger.debug(f"_parse_package_json({path}): {e}")
    return deps


def _parse_package_lock_json(path: str) -> dict:
    """Parse package-lock.json (v1, v2, v3 formats) for resolved versions."""
    deps = {}
    try:
        with open(path, "r", encoding="utf-8", errors="replace") as fh:
            data = json.load(fh)
        # v2/v3: "packages" key with "node_modules/x" entries
        packages = data.get("packages")
        if packages:
            for pkg_path, info in packages.items():
                if not pkg_path or "node_modules/" not in pkg_path:
                    continue
                name = pkg_path.split("node_modules/")[-1]
                ver  = info.get("version", "")
                if name and ver:
                    deps[name.lower()] = ver
        else:
            # v1: "dependencies" key, recursive
            def _walk(d):
                for name, info in (d or {}).items():
                    ver = info.get("version", "")
                    if ver:
                        deps[name.lower()] = ver
                    if "dependencies" in info:
                        _walk(info["dependencies"])
            _walk(data.get("dependencies", {}))
    except Exception as e:
        logger.debug(f"_parse_package_lock_json({path}): {e}")
    return deps


def _parse_pipfile_lock(path: str) -> dict:
    """Parse Pipfile.lock for default + develop packages."""
    deps = {}
    try:
        with open(path, "r", encoding="utf-8", errors="replace") as fh:
            data = json.load(fh)
        for section in ("default", "develop"):
            for name, info in (data.get(section) or {}).items():
                ver = (info.get("version") or "").lstrip("=")
                if ver:
                    deps[name.lower()] = ver
    except Exception as e:
        logger.debug(f"_parse_pipfile_lock({path}): {e}")
    return deps


def _parse_pipfile(path: str) -> dict:
    """Parse a Pipfile (TOML-ish) — best-effort regex, no toml dependency."""
    deps = {}
    try:
        with open(path, "r", encoding="utf-8", errors="replace") as fh:
            text = fh.read()
        # crude: lines like   requests = "==2.25.1"  or  requests = "*"
        for section_name in ("packages", "dev-packages"):
            m = re.search(rf'\[{section_name}\](.*?)(\n\[|\Z)', text, re.S)
            if not m:
                continue
            block = m.group(1)
            for line in block.splitlines():
                line = line.strip()
                m2 = re.match(r'^([A-Za-z0-9_.\-]+)\s*=\s*"([^"]*)"', line)
                if m2:
                    name, ver = m2.groups()
                    ver = ver.lstrip("=~^<>")
                    deps[name.lower().replace("_", "-")] = ver if ver != "*" else ""
    except Exception as e:
        logger.debug(f"_parse_pipfile({path}): {e}")
    return deps


def _parse_poetry_lock(path: str) -> dict:
    """Parse poetry.lock for [[package]] entries — best-effort regex."""
    deps = {}
    try:
        with open(path, "r", encoding="utf-8", errors="replace") as fh:
            text = fh.read()
        # Each package block: name = "x" \n version = "y"
        for block in re.split(r'\[\[package\]\]', text)[1:]:
            name_m = re.search(r'^name\s*=\s*"([^"]+)"', block, re.M)
            ver_m  = re.search(r'^version\s*=\s*"([^"]+)"', block, re.M)
            if name_m and ver_m:
                deps[name_m.group(1).lower()] = ver_m.group(1)
    except Exception as e:
        logger.debug(f"_parse_poetry_lock({path}): {e}")
    return deps


_MANIFEST_PARSERS = {
    "requirements.txt":    _parse_requirements_txt,
    "package.json":        _parse_package_json,
    "package-lock.json":   _parse_package_lock_json,
    "Pipfile.lock":        _parse_pipfile_lock,
    "Pipfile":             _parse_pipfile,
    "poetry.lock":         _parse_poetry_lock,
}


# ══════════════════════════════════════════════════════════════
#  VULNERABILITY MATCHING
# ══════════════════════════════════════════════════════════════

def _check_package(name: str, version: str) -> list[dict]:
    """
    Check a single (name, version) pair against the vulnerability DB
    and EOL table. Returns a list of finding dicts (may be empty).
    """
    findings = []
    name_l = name.lower()

    # ── Known-CVE check ────────────────────────────────────────
    if name_l in _VULN_DB and version:
        cur = _v(version)
        for max_vuln, severity, cve, desc in _VULN_DB[name_l]:
            if cur and cur <= _v(max_vuln):
                findings.append({
                    "title":       "Vulnerable Dependency",
                    "type":        "Vulnerable Dependency",
                    "category":    "Dependency Vulnerability",
                    "package":     name,
                    "version":     version,
                    "severity":    severity,
                    "description": f"{desc} Affects versions <= {max_vuln}.",
                    "fix":         f"Upgrade {name} to a version newer than {max_vuln}.",
                    "cve":         cve,
                    "match":       f"{name}=={version}",
                })
                break  # one finding per package is enough for the dashboard

    # ── EOL / outdated major version check ─────────────────────
    if name_l in _EOL_VERSIONS and version:
        cur = _v(version)
        eol = _EOL_VERSIONS[name_l]
        if cur and cur[:len(eol)] <= eol:
            # Don't duplicate if we already flagged a CVE for this package
            if not any(f["package"] == name for f in findings):
                findings.append({
                    "title":       "Outdated Dependency",
                    "type":        "Outdated Dependency",
                    "category":    "Dependency Vulnerability",
                    "package":     name,
                    "version":     version,
                    "severity":    "LOW",
                    "description": f"{name} {version} is an end-of-life or significantly outdated major version.",
                    "fix":         f"Upgrade {name} to a currently supported major version.",
                    "cve":         "N/A",
                    "match":       f"{name}=={version}",
                })

    return findings


# ══════════════════════════════════════════════════════════════
#  PUBLIC API
# ══════════════════════════════════════════════════════════════

# Files to look for, in priority order (lockfiles preferred — exact versions)
_MANIFEST_PRIORITY = [
    "package-lock.json", "poetry.lock", "Pipfile.lock",
    "requirements.txt", "package.json", "Pipfile",
]


def _find_manifests(base_dir: str) -> list[str]:
    """Walk base_dir and return paths to all recognised manifest files."""
    found = []
    skip_dirs = {".git", "node_modules", "__pycache__", ".venv", "venv", "dist", "build"}
    for root, dirs, files in os.walk(base_dir):
        dirs[:] = [d for d in dirs if d not in skip_dirs]
        for fname in files:
            if fname in _MANIFEST_PARSERS:
                found.append(os.path.join(root, fname))
    return found


def scan_dependencies(base_dir: str) -> list[dict]:
    """
    Scan all recognised manifest files under base_dir for known-vulnerable
    or outdated dependencies.

    Returns:
        List of finding dicts, each with keys:
          title, type, category, package, version, severity,
          description, fix, cve, match

    Safe to call on any directory — returns [] if no manifests found
    or all parsers fail. Never raises.
    """
    findings = []
    seen_packages = set()

    manifests = _find_manifests(base_dir)
    if not manifests:
        return []

    # Prefer lockfiles (exact versions) over manifest files (version ranges)
    manifests.sort(key=lambda p: _MANIFEST_PRIORITY.index(os.path.basename(p))
                    if os.path.basename(p) in _MANIFEST_PRIORITY else 99)

    for path in manifests:
        fname  = os.path.basename(path)
        parser = _MANIFEST_PARSERS.get(fname)
        if not parser:
            continue
        try:
            deps = parser(path)
        except Exception as e:
            logger.debug(f"Manifest parse failed {path}: {e}")
            continue

        for name, version in deps.items():
            if name in seen_packages:
                continue
            seen_packages.add(name)
            findings.extend(_check_package(name, version))

    return findings


def count_dependencies(base_dir: str) -> int:
    """
    Count the total number of unique dependencies declared across all
    recognised manifest files under base_dir. Used for the repository
    health "dependency_count" field.
    """
    seen = set()
    for path in _find_manifests(base_dir):
        fname  = os.path.basename(path)
        parser = _MANIFEST_PARSERS.get(fname)
        if not parser:
            continue
        try:
            deps = parser(path)
            seen.update(deps.keys())
        except Exception as e:
            logger.debug(f"count_dependencies parse failed {path}: {e}")
    return len(seen)