Blame for app/final_modules.py

"""Artifact-aware final-stage adapters, using pinned tools on remote infrastructure."""
import json
import base64
import os
import re
import tempfile
from pathlib import Path
from urllib.parse import urljoin, urlsplit

from .pipeline import execute
from .web_modules import alert, same_origin, read_report
from .module_inputs import ModuleInputs
from .coverage import profile as coverage_settings

MODULES = {"trufflehog", "amass", "depscan", "trivy", "kiterunner", "osv-scanner", "garak", "nettacker"}
CACHE = Path("/app/data/scanner-cache")


def skipped(module, url, reason):
    return {"source": module, "target": url, "status": "skipped", "alerts": [],
            "reason": reason, "output": module + " skipped: " + reason}


def diagnostics(rows):
    # Raw stderr and secret-scanner JSON can contain supplied source or model responses.
    # Persist fixed diagnostic categories only.
    if any(row.startswith("[diagnostic] Deadline") for row in rows):
        return "Process deadline reached; any parsed partial findings were retained"
    if any(row.startswith("[diagnostic] Process exit") for row in rows):
        return "Tool returned a nonzero exit status"
    return ""


def amass_internal_stop(rows, status):
    """Recognize the pinned CLI's context deadline, never an outer forced kill."""
    if status not in ("completed", "failed"):
        return False
    messages = [row.removeprefix("[stderr] ").strip() for row in rows]
    if "context deadline exceeded" not in messages:
        return False
    exits = [row for row in messages if row.startswith("[diagnostic] Process exit code:")]
    return (status == "completed" and not exits or
            status == "failed" and exits == ["[diagnostic] Process exit code: 1"])


def json_rows(rows):
    found = []
    for row in rows:
        try:
            item = json.loads(row)
        except (TypeError, ValueError):
            continue
        if isinstance(item, dict):
            found.append(item)
    return found


# Let the pinned native tool choose and run its built-in detector verifiers.
TRUFFLEHOG_DETECTION_SECONDS = 40
TRUFFLEHOG_VERIFY_SECONDS = 120


def native_secret_state(row):
    if row.get("Verified") is True:
        return "verified"
    if row.get("VerificationSkipped") is True:
        return "not_attempted"
    if row.get("VerificationError") or not isinstance(row.get("Verified"), bool):
        return "unknown"
    # Native false includes unsupported/skipped verification in versions whose
    # JSON does not expose VerificationSkipped. Never describe it as invalid.
    return "unverified"


def trufflehog_candidates(rows):
    # Keys and native metadata remain in memory only, never in returned diagnostics.
    candidates = {}
    rank = {"verified": 3, "unknown": 2, "unverified": 1, "not_attempted": 0}
    for row in json_rows(rows):
        name = row.get("DetectorName")
        if not isinstance(name, str) or not re.fullmatch(r"[A-Za-z][A-Za-z0-9_]{0,79}", name):
            continue
        raw = row.get("RawV2") or row.get("Raw") or ""
        if not isinstance(raw, str):
            continue
        key = (name, raw)
        previous = candidates.get(key)
        if previous is None or rank[native_secret_state(row)] > rank[native_secret_state(previous)]:
            candidates[key] = row
    return candidates


def trufflehog_states(rows, verification_enabled):
    return {key: native_secret_state(row) if verification_enabled else "not_attempted"
            for key, row in trufflehog_candidates(rows).items()}


def parse_trufflehog(rows, url, verification_enabled=False):
    # Never return Raw, RawV2, Redacted, ExtraData, errors or source snippets.
    groups = {}
    for (name, _), state in trufflehog_states(rows, verification_enabled).items():
        groups[(name, state)] = groups.get((name, state), 0) + 1
    descriptions = {
        "verified": "Native TruffleHog verification confirmed this credential at scan time.",
        "unverified": ("Native TruffleHog did not confirm validity. The detector may not support "
                       "verification or may have skipped it; this does not prove the credential is invalid."),
        "unknown": "Native verification was inconclusive or reported an error; validity is unknown.",
        "not_attempted": "Detection only; verification was disabled or the native detector reported skipping it.",
    }
    findings = []
    for (name, state), count in groups.items():
        label = {"verified": "Verified", "unverified": "Unverified",
                 "unknown": "Inconclusive", "not_attempted": "Detection-only"}[state]
        finding = alert("trufflehog", label + " " + name + " secret", url,
                        descriptions[state] + f" {count} distinct candidate(s). Values and context redacted.",
                        "High" if state == "verified" else "Medium", name)
        finding.update(verification_status=state, candidate_count=count,
                       confidence="High" if state == "verified" else "Low")
        findings.append(finding)
    return findings[:100]


def parse_dependency_report(module, report, url):
    findings = []
    if module == "trivy" and isinstance(report, dict):
        for result in report.get("Results") or []:
            if not isinstance(result, dict):
                continue
            for item in result.get("Vulnerabilities") or []:
                if isinstance(item, dict):
                    findings.append(alert(module, str(item.get("VulnerabilityID") or "Advisory") +
                                          ": " + str(item.get("PkgName") or "package"), url,
                                          "SBOM advisory match; installed " + str(item.get("InstalledVersion") or "?") +
                                          "; fixed " + str(item.get("FixedVersion") or "not provided"),
                                          str(item.get("Severity") or "info").title().replace("Info", "Informational"),
                                          item.get("VulnerabilityID", "")))
    elif module == "osv-scanner" and isinstance(report, dict):
        for result in report.get("results") or []:
            for package in result.get("packages") or []:
                info = package.get("package") or {}
                for vuln in package.get("vulnerabilities") or []:
                    if isinstance(vuln, dict):
                        findings.append(alert(module, str(vuln.get("id") or "Advisory") + ": " +
                                              str(info.get("name") or "package"), url,
                                              "Unverified SBOM advisory match for version " + str(info.get("version") or "?"),
                                              "Informational", vuln.get("id", "")))
    elif module == "depscan" and isinstance(report, dict):
        for item in report.get("vulnerabilities") or []:
            if not isinstance(item, dict):
                continue
            ratings = item.get("ratings") or []
            severity = str(ratings[0].get("severity") or "info") if ratings and isinstance(ratings[0], dict) else "info"
            findings.append(alert(module, str(item.get("id") or "Dependency advisory"), url,
                                  "Unverified SBOM advisory match. Application-only database profile.",
                                  severity.title().replace("Info", "Informational"), item.get("id", "")))
    return findings[:100]


async def prepare_database(module):
    directory = CACHE / module
    directory.mkdir(parents=True, exist_ok=True)
    if module == "trivy":
        ready = directory / "db" / "metadata.json"
        args = ["/usr/local/bin/trivy", "--cache-dir", str(directory),
                "image", "--download-db-only", "--no-progress"]
    else:
        ready = directory / ".depscan-vdb-image"
        args = ["/usr/bin/env", "VDB_HOME=" + str(directory),
                "VDB_DATABASE_URL=ghcr.io/appthreat/vdbxz-app-2y:v6.7.x",
                "VDB_DOWNLOAD_RETRIES=1", "/opt/depscan/bin/depscan-vdb",
                "download", "--scope", "app", "--time", "2y", "--compression", "xz"]
    if ready.is_file():
        return True, ""
    rows, status = await execute(args, 180)
    return status == "completed" and ready.is_file(), diagnostics(rows) or "Vulnerability database preparation did not complete"


async def run(module, url, host, agent, delay, inputs=None, profile="current"):
    settings = coverage_settings(profile)
    values = inputs if isinstance(inputs, ModuleInputs) else ModuleInputs(**(inputs or {}))
    if module == "trufflehog" and not values.source_text.strip():
        return skipped(module, url, "Requires supplied source text; a URL is not a filesystem input")
    if module in {"depscan", "trivy", "osv-scanner"} and not values.sbom_text:
        return skipped(module, url, "Requires a CycloneDX SBOM with named/versioned package URLs")
    if module == "kiterunner" and not values.api_target:
        return skipped(module, url, "API-target option is not enabled for this asset")
    if module == "garak" and not values.llm_path:
        return skipped(module, url, "Requires an explicitly configured same-origin REST LLM endpoint")
    with tempfile.TemporaryDirectory(prefix="asfb-" + module + "-") as tmp:
        root = Path(tmp)
        report_path = str(root / "report.json")
        args, deadline = [], 60
        report, findings, hosts, endpoints, probes = None, [], [], [], []
        reason = ""
        verification = None
        early_stop = False
        if module == "trufflehog":
            source = root / "source"
            source.mkdir()
            (source / "source.txt").write_text(values.source_text, encoding="utf-8")
            (source / "source.txt").chmod(0o600)
            args = ["/usr/local/bin/trufflehog", "filesystem", str(source), "--json",
                    "--results", "verified,unverified,unknown", "--no-update", "--concurrency", "1",
                    "--force-skip-binaries", "--force-skip-archives", "--fail-on-scan-errors"]
            if values.trufflehog_verify:
                deadline = TRUFFLEHOG_VERIFY_SECONDS
                reason = "Native TruffleHog verification enabled for all detectors; original supplied source; values and context redacted"
            else:
                args.append("--no-verification")
                deadline = TRUFFLEHOG_DETECTION_SECONDS
                reason = "Supplied source only; secret verification disabled; values and context redacted"
        elif module in ("trivy", "depscan", "osv-scanner"):
            sbom = root / "bom.json"
            sbom.write_text(values.sbom_text, encoding="utf-8")
            if module in ("trivy", "depscan"):
                ok, failure = await prepare_database(module)
                if not ok:
                    return {"source": module, "target": url, "status": "unavailable",
                            "alerts": [], "reason": failure,
                            "output": module + ": database unavailable; no clean-scan claim"}
            if module == "trivy":
                args = ["/usr/local/bin/trivy", "--cache-dir", str(CACHE / module), "sbom",
                        "--skip-db-update", "--skip-java-db-update", "--offline-scan", "--format", "json",
                        "--output", report_path, str(sbom)]
                reason = "Supplied application SBOM only; cached database; no container daemon"
            elif module == "osv-scanner":
                args = ["/usr/local/bin/osv-scanner", "scan", "source", "--sbom", str(sbom),
                        "--format", "json", "--output", report_path]
                reason = "Supplied SBOM package names/versions are queried against the public keyless OSV service"
            else:
                args = ["/usr/bin/env", "VDB_HOME=" + str(CACHE / module),
                        "VDB_DATABASE_URL=ghcr.io/appthreat/vdbxz-app-2y:v6.7.x",
                        "VDB_DOWNLOAD_RETRIES=1", "/opt/depscan/bin/depscan",
                        "--bom", str(sbom), "--reports-dir", str(root / "reports"),
                        "--reachability-analyzer", "off", "--no-suggest", "--no-vuln-table"]
                reason = "Application-only SBOM profile; smaller 2024+ advisory database; no source builds or OS reachability"
            deadline = 90
        elif module == "amass":
            args = ["/usr/local/bin/amass", "enum", "-passive", "-d", host, "-timeout", str(settings["amass_minutes"]),
                    "-dir", str(root / "state"), "-o", str(root / "hosts.txt"), "-nocolor"]
            deadline = settings["amass_timeout"]
            reason = "HF standalone-compatible Amass 4.2.0 passive profile; discovered names are inventory, never auto-scanned"
        elif module == "kiterunner":
            dictionary = root / "routes.txt"
            dictionary.write_text("api\napi/v1\napi/v2\nhealth\nstatus\nopenapi.json\nswagger.json\nversion\n", encoding="utf-8")
            target = urlsplit(url)._replace(path="/", query="", fragment="").geturl()
            args = ["/usr/local/bin/kr", "brute", target, "-w", str(dictionary),
                    "-x", "1", "-j", "1", "--max-redirects", "0", "--force-method", "GET",
                    "--delay", str(max(1.0, delay)) + "s", "--timeout", "5s",
                    "--disable-precheck", "--wildcard-detection=false", "--progress=false",
                    "--preflight-depth", "0", "--header", "Accept: application/json",
                    "--user-agent", agent, "-o", "json", "-q"]
            deadline = 65
            reason = "Eight-route GET-only API sample; one connection, no redirects, no generated payloads"
        elif module == "nettacker":
            args = ["/opt/nettacker/bin/python", "/app/app/nettacker_runner.py", url, report_path, agent]
            deadline = 15
            reason = "Partial Nettacker integration: one exact-URL HTTP-status check using its HTTP engine and module conditions"
        elif module == "garak":
            target = urljoin(urlsplit(url)._replace(path="/", query="", fragment="").geturl(), values.llm_path)
            if not same_origin(target, url):
                raise ValueError("LLM endpoint outside selected origin")
            request = root / "config.json"
            request.write_text(json.dumps({"url": target, "prompt_field": values.prompt_field,
                                          "response_field": values.response_field, "delay": delay}), encoding="utf-8")
            args = ["/opt/garak/bin/python", "/app/app/garak_runner.py", str(request), report_path]
            deadline = 35
            reason = "Two garak REST output-filter test-signature prompts; applicable when that LLM is expected to filter antivirus signatures"
        else:
            raise ValueError("Unsupported module")
        rows, status = await execute(args, deadline)
        if module == "trufflehog":
            states = trufflehog_states(rows, values.trufflehog_verify)
            findings = parse_trufflehog(rows, url, values.trufflehog_verify)
            verification = {"enabled": values.trufflehog_verify, "mode": "native",
                            "detectors": "all", "deadline_seconds": deadline,
                            "candidates_reported": len(states),
                            **{state: sum(value == state for value in states.values())
                               for state in ("verified", "unverified", "unknown", "not_attempted")}}
            if values.trufflehog_verify:
                reason += "; " + ", ".join(f"{verification[state]} {state.replace('_', ' ')}"
                                          for state in ("verified", "unverified", "unknown", "not_attempted"))
                if verification["unknown"] and status == "completed":
                    status = "partial"
        elif module == "amass":
            path = root / "hosts.txt"
            candidates = (path.read_text(encoding="utf-8", errors="replace").splitlines()[:500] if path.is_file() else []) + rows
            for entry in candidates:
                name = entry.strip().lower().rstrip(".")
                if re.fullmatch(r"[a-z0-9.-]{1,253}", name) and (name == host or name.endswith("." + host)):
                    hosts.append(name)
            hosts = sorted(set(hosts))[:100]
            if amass_internal_stop(rows, status):
                status, early_stop = "completed", True
                reason += "; Early stop: native enumeration time budget reached; inventory coverage is incomplete"
        elif module == "kiterunner":
            # The pinned CLI's responses[].sc is the observed HTTP status.
            # A route record alone (including 404/wildcard hits) is not discovery.
            for item in json_rows(rows):
                responses = item.get("responses")
                if (item.get("method") != "GET" or not isinstance(responses, list)
                        or not responses or not isinstance(responses[0], dict)):
                    continue
                code = responses[0].get("sc")
                if type(code) is not int or not 200 <= code < 300:
                    continue
                def as_text(value):
                    if not isinstance(value, str):
                        return ""
                    if value.startswith(("http://", "https://", "/")):
                        return value
                    try:
                        return base64.b64decode(value, validate=True).decode("utf-8")
                    except (ValueError, UnicodeError):
                        return value
                target = as_text(item.get("target"))
                path = as_text(item.get("path"))
                candidate = item.get("url") or (urljoin(target.rstrip("/") + "/", path) if target else "")
                if isinstance(candidate, str) and same_origin(candidate, url):
                    endpoints.append(candidate)
                    probes.append({"url": candidate, "status_code": code, "source": "kiterunner"})
            endpoints = sorted(set(endpoints))[:20]
            probes = list({item["url"]: item for item in probes}.values())[:20]
        elif module == "depscan":
            candidates = sorted((root / "reports").glob("*.vdr.json"))
            if candidates:
                report = read_report(str(candidates[0]))
            findings = parse_dependency_report(module, report, url)
        else:
            report = read_report(report_path)
            if module in ("trivy", "osv-scanner"):
                findings = parse_dependency_report(module, report, url)
                if module == "osv-scanner" and report is not None and any(
                        row.strip() == "[diagnostic] Process exit code: 1" for row in rows) and findings:
                    status = "findings"
            elif module == "nettacker" and isinstance(report, dict):
                if isinstance(report.get("status_code"), int):
                    probes = [{"url": url, "status_code": report["status_code"], "source": "nettacker"}]
                if report.get("status") == "failed":
                    status = "failed"
            elif module == "garak" and isinstance(report, dict):
                for item in report.get("tests", []):
                    if item.get("matched"):
                        findings.append(alert("garak", "LLM emitted antivirus test signature", url,
                                              "Output-filter observation from a benign garak test signature; this is not proof of a general model vulnerability.",
                                              "Informational", "EICAR"))
                if report.get("status") == "failed":
                    status = "failed"
        valid_report = isinstance(report, dict) and (
            (module == "depscan" and report.get("bomFormat") == "CycloneDX") or
            (module == "trivy" and isinstance(report.get("Results"), list)) or
            (module == "osv-scanner" and isinstance(report.get("results"), list)) or
            (module == "garak" and isinstance(report.get("tests"), list) and bool(report.get("tests"))) or
            (module == "nettacker" and isinstance(report.get("status_code"), int))
        )
        if module in ("depscan", "trivy", "osv-scanner", "garak", "nettacker") and not valid_report:
            if status in ("completed", "findings"):
                status = "failed"
            reason += "; no valid structured report produced"
        if status in ("completed", "findings") and findings:
            status = "findings"
        diag = diagnostics(rows) if status not in ("completed", "findings") else ""
        summary = (f"{module}: {len(findings)} unverified findings; {len(hosts)} inventory hosts; "
                   f"{len(endpoints)} observed routes; status: {status}")
        if verification is not None:
            summary = ("trufflehog: " + ", ".join(f"{verification[state]} {state.replace('_', ' ')}"
                       for state in ("verified", "unverified", "unknown", "not_attempted")) + f"; status: {status}")
        return {"source": module, "target": url, "status": status, "alerts": findings,
                **({"verification": verification} if verification is not None else {}),
                **({"early_stop": True, "stop_reason": "native_time_budget", "coverage_complete": False}
                   if early_stop else {}),
                "hosts": hosts, "endpoints": endpoints, "probes": probes,
                "deadline_seconds": deadline,
                "reason": (reason + ("; " + diag if diag else ""))[:700],
                "output": summary}