| import asyncio |
| import json |
| from pathlib import Path |
| from unittest.mock import patch |
|
|
| import pytest |
| from fastapi.testclient import TestClient |
| from pydantic import ValidationError |
|
|
| from app import final_modules as modules, main, template_builder |
| from app.module_inputs import ModuleInputs, clean_sbom |
| from test_worker_pipeline import run_job |
|
|
|
|
| BOM = json.dumps({"bomFormat": "CycloneDX", "components": [ |
| {"type": "library", "name": "jquery", "version": "1.6.1", "purl": "pkg:npm/jquery@1.6.1", |
| "externalReferences": [{"url": "http://127.0.0.1/secret"}]}]}) |
|
|
|
|
| @pytest.mark.parametrize("name", ["trufflehog", "depscan", "trivy", "osv-scanner", "kiterunner", "garak"]) |
| def test_missing_required_context_skips_without_process_or_database(name): |
| with patch.object(modules, "execute") as execute, patch.object(modules, "prepare_database") as prepare: |
| result = asyncio.run(modules.run(name, "https://example.org/", "example.org", "Test", 1.5)) |
| assert result["status"] == "skipped" |
| assert not execute.called and not prepare.called |
|
|
|
|
| def test_sbom_is_sanitized_and_external_urls_are_removed(): |
| result = clean_sbom(BOM) |
| assert "externalReferences" not in result and "127.0.0.1" not in result |
| assert json.loads(result)["components"][0]["purl"] == "pkg:npm/jquery@1.6.1" |
|
|
|
|
| @pytest.mark.parametrize("text", ['{}', '{"bomFormat":"SPDX"}', '{"bomFormat":"CycloneDX","components":[]}']) |
| def test_incompatible_sboms_are_rejected(text): |
| with pytest.raises(ValueError): |
| clean_sbom(text) |
|
|
|
|
| @pytest.mark.parametrize("path", ["//other.org/api", "https://outside.org/", "/api?token=secret", "/../api"]) |
| def test_llm_endpoint_cannot_change_origin_or_embed_credentials(path): |
| with pytest.raises(ValidationError): |
| ModuleInputs(llm_path=path) |
|
|
|
|
| def test_no_source_or_sbom_content_in_public_options(): |
| inputs = ModuleInputs(source_text="my-private-source-value", sbom_text=BOM) |
| summary = json.dumps(inputs.public_summary()) |
| assert "my-private-source-value" not in summary |
| assert "jquery" not in summary |
| assert inputs.public_summary()["sbom_components"] == 1 |
|
|
|
|
| def test_trufflehog_redacts_secret_and_context_everywhere(): |
| secret = "super-secret-material-do-not-return" |
| rows = [json.dumps({"DetectorName": "MockDetector", "Raw": secret, "RawV2": secret, |
| "ExtraData": {"credential": secret}, "Redacted": secret})] |
| async def execute(args, timeout): |
| assert "--no-verification" in args and "--no-update" in args |
| assert "--force-skip-archives" in args |
| root = Path(args[args.index("filesystem") + 1]) |
| assert (root / "source.txt").read_text() == secret |
| return rows, "completed" |
| with patch.object(modules, "execute", execute): |
| result = asyncio.run(modules.run("trufflehog", "https://example.org/", "example.org", "Test", 1.5, |
| ModuleInputs(source_text=secret))) |
| assert result["status"] == "findings" |
| assert secret not in json.dumps(result) |
| assert len(result["alerts"]) == 1 |
|
|
|
|
| @pytest.mark.parametrize("name", ["depscan", "trivy"]) |
| def test_database_failure_is_unavailable_not_clean(name): |
| async def prepare(module): return False, "Database download timeout" |
| with patch.object(modules, "prepare_database", prepare), patch.object(modules, "execute") as execute: |
| result = asyncio.run(modules.run(name, "https://example.org/", "example.org", "Test", 1.5, |
| ModuleInputs(sbom_text=BOM))) |
| assert result["status"] == "unavailable" |
| assert not execute.called |
|
|
|
|
| def test_amass_inventory_filters_unrelated_domains_without_expanding_scope(): |
| async def execute(args, timeout): |
| assert "-passive" in args and "-active" not in args |
| Path(args[args.index("-o") + 1]).write_text("a.example.org\nexample.org.evil.org\nexample.org\n") |
| return [], "completed" |
| with patch.object(modules, "execute", execute): |
| result = asyncio.run(modules.run("amass", "https://example.org/", "example.org", "Test", 1.5)) |
| assert result["hosts"] == ["a.example.org", "example.org"] |
| assert result["endpoints"] == [] |
|
|
|
|
| def test_kiterunner_uses_only_small_get_dictionary_and_parses_upstream_hit_shape(): |
| async def execute(args, timeout): |
| assert args[args.index("--max-redirects") + 1] == "0" |
| assert args[args.index("--force-method") + 1] == "GET" |
| assert args[args.index("--delay") + 1] == "1.5s" |
| assert "127.0.0.1" not in " ".join(args) |
| assert len(Path(args[args.index("-w") + 1]).read_text().splitlines()) == 8 |
| return [json.dumps({"method": "GET", "target": "https://example.org", "path": "/api/v1", "responses": [{"sc": 200}]}), |
| json.dumps({"method": "GET", "target": "https://outside.org", "path": "/api"})], "completed" |
| with patch.object(modules, "execute", execute): |
| result = asyncio.run(modules.run("kiterunner", "https://example.org/", "example.org", "Test", 1.5, |
| ModuleInputs(api_target=True))) |
| assert result["endpoints"] == ["https://example.org/api/v1"] |
|
|
|
|
| @pytest.mark.parametrize("name,payload", [ |
| ("trivy", {"SchemaVersion": 2, "Results": [{"Vulnerabilities": [ |
| {"VulnerabilityID": "CVE-TEST", "PkgName": "package", "Severity": "HIGH"}]}]}), |
| ("osv-scanner", {"results": [{"packages": [{"package": {"name": "package", "version": "1"}, |
| "vulnerabilities": [{"id": "GHSA-test"}]}]}]}), |
| ("depscan", {"bomFormat": "CycloneDX", "vulnerabilities": [{"id": "CVE-TEST", "ratings": [{"severity": "high"}]}]}), |
| ]) |
| def test_dependency_json_is_normalized_and_raw_artifacts_not_returned(name, payload): |
| async def prepare(*args): return True, "" |
| async def execute(args, timeout): |
| if name == "depscan": |
| directory = Path(args[args.index("--reports-dir") + 1]) |
| directory.mkdir() |
| (directory / "depscan-universal.vdr.json").write_text(json.dumps(payload)) |
| else: |
| Path(args[args.index("--output") + 1]).write_text(json.dumps(payload)) |
| return [], "completed" |
| with patch.object(modules, "prepare_database", prepare), patch.object(modules, "execute", execute): |
| result = asyncio.run(modules.run(name, "https://example.org/", "example.org", "Test", 1.5, |
| ModuleInputs(sbom_text=BOM))) |
| assert result["status"] == "findings" |
| assert len(result["alerts"]) == 1 |
| assert "externalReferences" not in json.dumps(result) |
|
|
|
|
| def test_invalid_report_cannot_claim_clean_scan(): |
| async def execute(args, timeout): |
| Path(args[args.index("--output") + 1]).write_text("{}") |
| return [], "completed" |
| with patch.object(modules, "execute", execute): |
| result = asyncio.run(modules.run("osv-scanner", "https://example.org/", "example.org", "Test", 1.5, |
| ModuleInputs(sbom_text=BOM))) |
| assert result["status"] == "failed" |
|
|
|
|
| def test_worker_keeps_amass_names_in_inventory_only(): |
| run, job = run_job(["amass", "httpx"]) |
| async def module(*args): |
| return {"status": "completed", "alerts": [], "hosts": ["a.example.com"], "output": "inventory"} |
| async def probe(urls, *args): |
| assert all("a.example.com" not in url for url in urls) |
| return {"status": "completed", "probes": []} |
| with patch.object(main, "public_host"), patch.object(main, "run_web_module", module), \ |
| patch.object(main, "probe", probe): |
| asyncio.run(run()) |
| assert job["discovered_hosts"] == ["a.example.com"] |
| assert job["status"] == "completed" |
|
|
|
|
| def test_artifact_input_is_not_returned_in_job_status_and_does_not_reuse_stale_job(): |
| main.assets.clear() |
| main.jobs.clear() |
| main.queue = asyncio.Queue() |
| async def no_scan(*args): |
| return None |
| with TestClient(main.app) as client, patch.object(main, "public_host"), patch.object(main, "run_scan_job", no_scan): |
| owner = client.get("/api/health").headers["x-asfb-session"] |
| main.assets["a"] = {"id": "a", "owner": owner, "name": "https://example.org/", "compatibility": "web"} |
| body = {"asset_id": "a", "policy_confirmed": True, "active_confirmed": True, |
| "modules": ["trufflehog"], "options": {"inputs": {"source_text": "secret-first"}}} |
| response = client.post("/api/scans", json=body) |
| assert response.status_code == 200 |
| first = response.json()["id"] |
| body["options"]["inputs"]["source_text"] = "secret-other" |
| second = client.post("/api/scans", json=body).json()["id"] |
| assert first != second |
| assert "secret-first" not in json.dumps(main.jobs) |
| assert "secret-other" not in json.dumps(main.jobs) |
|
|
|
|
| def test_keyless_builder_cannot_create_payloads_or_external_hosts(): |
| from app.nuclei_catalog import validate_template |
| draft = template_builder.build_draft("Test observation", "/status", "healthy") |
| validate_template(draft) |
| assert "{{BaseURL}}/status" in draft |
| for path in ("//outside.org", "/../secret", "/x?cmd=1"): |
| with pytest.raises(ValueError): |
| template_builder.build_draft("Test observation", path, "ok") |
|
|
|
|
| def test_optional_gemini_without_credentials_is_unavailable_not_required(monkeypatch): |
| monkeypatch.delenv("GEMINI_API_KEY", raising=False) |
| monkeypatch.delenv("GEMINI_MODEL", raising=False) |
| with pytest.raises(RuntimeError): |
| asyncio.run(template_builder.gemini_draft("Observation")) |
|
|
|
|
| def test_draft_endpoint_validates_but_never_launches_scan(): |
| async def validate(content): |
| return content |
| with TestClient(main.app) as client, patch.object(template_builder, "validate_draft", validate): |
| response = client.post("/api/nuclei/draft", json={"name": "Observation", "path": "/", "word": "hello"}) |
| assert response.status_code == 200 |
| assert response.json()["executed"] is False |
|
|
|
|
| |
| def native_secret(name="Github", raw="fixture-private-key", **fields): |
| return json.dumps({"DetectorName": name, "Raw": raw, "Redacted": raw, |
| "ExtraData": {"account": "fixture-private-identity"}, |
| "SourceMetadata": {"source": "fixture-private-source"}, |
| "SecretParts": {"key": raw}, **fields}) |
|
|
|
|
| def run_secret_fixture(rows, enabled=True, native_status="completed"): |
| calls, paths = [], [] |
| source = "database=https://fixture-private-host/\npassword=fixture-private-key\noriginal context" |
| async def execute(args, timeout): |
| calls.append((args, timeout)) |
| assert "--no-update" in args and "--force-skip-archives" in args |
| assert "--force-skip-binaries" in args |
| assert args[args.index("--results") + 1] == "verified,unverified,unknown" |
| assert args[args.index("--concurrency") + 1] == "1" |
| |
| assert not {"--include-detectors", "--exclude-detectors", "--verifier", "--config", |
| "--allow-verification-overlap", "--no-verification-cache"} & set(args) |
| assert ("--no-verification" in args) is (not enabled) |
| assert timeout == (120 if enabled else 40) |
| root = Path(args[args.index("filesystem") + 1]) |
| path = root / "source.txt" |
| assert path.read_text() == source |
| assert (path.stat().st_mode & 0o777) == 0o600 |
| assert sorted(p.name for p in root.iterdir()) == ["source.txt"] |
| assert "fixture-private" not in " ".join(args) |
| paths.append(path) |
| return rows, native_status |
| with patch.object(modules, "execute", execute): |
| result = asyncio.run(modules.run("trufflehog", "https://example.org/", "example.org", "Test", 1.5, |
| ModuleInputs(source_text=source, trufflehog_verify=enabled))) |
| assert all(not path.exists() for path in paths) |
| assert len(calls) == 1 |
| assert "fixture-private" not in json.dumps(result) |
| assert "attempted" not in result["verification"] |
| assert "limit" not in result["verification"] |
| return result |
|
|
|
|
| @pytest.mark.parametrize("fields,status,expected", [ |
| ({"Verified": True}, "completed", "verified"), |
| ({"Verified": False}, "completed", "unverified"), |
| ({"Verified": False, "VerificationError": "fixture-private-provider-error"}, "completed", "unknown"), |
| ({"Verified": False}, "timeout", "unverified"), |
| ({"Verified": True}, "timeout", "verified"), |
| ({"Verified": False, "VerificationSkipped": True}, "completed", "not_attempted"), |
| ({}, "completed", "unknown"), |
| ]) |
| def test_native_verification_labels_and_redaction(fields, status, expected): |
| result = run_secret_fixture([native_secret(**fields)], native_status=status) |
| finding = result["alerts"][0] |
| assert finding["verification_status"] == expected |
| assert finding["confidence"] == ("High" if expected == "verified" else "Low") |
| assert finding["risk"] == ("High" if expected == "verified" else "Medium") |
| assert result["verification"][expected] == 1 |
| assert result["verification"]["mode"] == "native" |
| assert result["verification"]["detectors"] == "all" |
| assert result["status"] == (status if status != "completed" else |
| "partial" if expected == "unknown" else "findings") |
|
|
|
|
| def test_detection_default_disables_native_verification_even_if_output_claims_verified(): |
| result = run_secret_fixture([native_secret(Verified=True)], enabled=False) |
| assert result["alerts"][0]["verification_status"] == "not_attempted" |
| assert result["verification"]["enabled"] is False |
|
|
|
|
| @pytest.mark.parametrize("name", ["AWS", "Github", "Slack", "HuggingFace", "OpenAI", |
| "Stripe", "Postgres", "Azure", "PrivateKey"]) |
| def test_native_verification_accepts_all_detector_results_without_application_allowlist(name): |
| result = run_secret_fixture([native_secret(name, Verified=True)]) |
| assert result["verification"]["verified"] == 1 |
| assert result["alerts"][0]["name"] == "Verified " + name + " secret" |
|
|
|
|
| def test_more_than_ten_credentials_remain_in_single_native_scan(): |
| rows = [native_secret(raw="fixture-private-" + str(i), Verified=True) for i in range(15)] |
| result = run_secret_fixture(rows) |
| assert result["verification"]["verified"] == 15 |
| assert result["verification"]["candidates_reported"] == 15 |
| assert result["alerts"][0]["candidate_count"] == 15 |
|
|
|
|
| def test_partial_native_output_retains_findings_and_timeout_status(): |
| result = run_secret_fixture([native_secret(Verified=True), |
| "[diagnostic] Deadline exceeded; fixture-private-error"], |
| native_status="timeout") |
| assert result["status"] == "timeout" |
| assert result["verification"]["verified"] == 1 |
| assert result["alerts"][0]["verification_status"] == "verified" |
|
|
|
|
| def test_empty_timeout_cannot_claim_clean_scan_or_completed_verification(): |
| result = run_secret_fixture(["[diagnostic] Deadline exceeded"], native_status="timeout") |
| assert result["status"] == "timeout" and result["alerts"] == [] |
| assert result["verification"]["candidates_reported"] == 0 |
|
|
|
|
| def test_unverified_does_not_claim_invalid_or_known_verification_attempt(): |
| result = run_secret_fixture([native_secret("UnsupportedExample", Verified=False)]) |
| assert result["alerts"][0]["verification_status"] == "unverified" |
| assert "may not support" in result["alerts"][0]["description"] |
| assert "does not prove" in result["alerts"][0]["description"] |
|
|
|
|
| def test_native_verification_deduplicates_candidates_and_keeps_strongest_evidence(): |
| rows = [native_secret(raw="fixture-private-good", Verified=False), |
| native_secret(raw="fixture-private-good", Verified=True), |
| native_secret(raw="fixture-private-bad", Verified=False)] |
| result = run_secret_fixture(rows) |
| assert result["verification"]["verified"] == result["verification"]["unverified"] == 1 |
| assert result["verification"]["candidates_reported"] == 2 |
| assert {a["verification_status"] for a in result["alerts"]} == {"verified", "unverified"} |
|
|
|
|
| def test_native_verification_summary_counts_all_candidates_beyond_alert_group_cap(): |
| rows = [native_secret("Detector" + str(i), Verified=True) for i in range(101)] |
| result = run_secret_fixture(rows) |
| assert result["verification"]["verified"] == 101 |
| assert len(result["alerts"]) == 100 |
|
|
|
|
| def test_verification_flag_is_visible_but_source_stays_private_in_job_options(): |
| summary = ModuleInputs(source_text="fixture-private-source", trufflehog_verify=True).public_summary() |
| assert summary["trufflehog_verify"] is True |
| assert "fixture-private" not in json.dumps(summary) |
|
|
