app/tools/dependency_scanner.py

"""
Dependency vulnerability scanner using pip-audit.
Scans requirements.txt / pyproject.toml for known CVEs.
Best-effort: returns empty list if pip-audit is not installed or fails.
"""
import json
import subprocess  # nosec B404
import os
from app.core.logger import get_logger

logger = get_logger(__name__)


def scan_dependencies(local_path: str) -> list[dict]:
    """
    Run pip-audit on the repo and return a list of vulnerability dicts.
    Each dict has: package, version, id, description, fix_versions.
    Returns empty list if pip-audit unavailable or no manifest found.
    """
    # Find dependency manifest
    manifest = None
    for candidate in ["requirements.txt", "pyproject.toml", "setup.cfg"]:
        full = os.path.join(local_path, candidate)
        if os.path.exists(full):
            manifest = full
            break

    if manifest is None:
        logger.info("No dependency manifest found — skipping vulnerability scan")
        return []

    try:
        result = subprocess.run(  # nosec B603 B607
            ["pip-audit", "--requirement", manifest, "--format", "json", "--no-deps"],
            capture_output=True,
            text=True,
            timeout=120,
        )
    except FileNotFoundError:
        logger.warning("pip-audit not installed — skipping dependency scan")
        return []
    except subprocess.TimeoutExpired:
        logger.warning("pip-audit timed out")
        return []
    except Exception as exc:
        logger.warning("pip-audit failed", extra={"error": str(exc)})
        return []

    if result.returncode not in (0, 1):  # 0=clean, 1=vulns found, others=error
        logger.warning("pip-audit error", extra={"stderr": result.stderr[:200]})
        return []

    try:
        data = json.loads(result.stdout)
    except json.JSONDecodeError:
        return []

    findings = []
    for dep in data.get("dependencies", []):
        for vuln in dep.get("vulns", []):
            findings.append({
                "package": dep.get("name", "unknown"),
                "version": dep.get("version", "unknown"),
                "id": vuln.get("id", ""),
                "description": vuln.get("description", ""),
                "fix_versions": vuln.get("fix_versions", []),
            })

    logger.info("Dependency scan complete", extra={"vulnerabilities": len(findings)})
    return findings