import asyncio import json from pathlib import Path from unittest.mock import patch import pytest from app import analysis_modules as analysis, planning, main from app.module_inputs import ModuleInputs from app.waf_runner import Guard, ProfileStopped, run as run_waf from test_worker_pipeline import run_job def test_semgrep_requires_source_and_explicit_language_without_launching_process(): for inputs in (ModuleInputs(), ModuleInputs(source_text="source"), ModuleInputs(sast_language="python")): with patch.object(analysis, "execute") as execute: result = asyncio.run(analysis.run("semgrep", "https://example.org", "example.org", "Test", 1, inputs)) assert result["status"] == "skipped" and not execute.called def test_semgrep_uses_bundled_rules_and_omits_source_context_from_results(): secret = "fixture-private-source" paths = [] async def execute(args, timeout): assert args[args.index("--metrics") + 1] == "off" assert "--disable-version-check" in args and "--oss-only" in args assert "--config" in args and args[args.index("--config") + 1].endswith("review.yaml") assert "--autofix" not in args and "--allow-local-builds" not in args assert "--no-git-ignore" in args and timeout == 75 source = Path(args[-1]);paths.append(source) assert source.suffix == ".py" and source.read_text() == secret assert source.stat().st_mode & 0o777 == 0o600 Path(args[args.index("--output") + 1]).write_text(json.dumps({ "results": [{"check_id": "prefix.asfb-python-dynamic-code", "start": {"line": 3}, "extra": {"message": secret, "lines": secret, "metavars": {"value": secret}}}], "errors": [], "paths": {"scanned": [str(source)]}})) return [], "completed" with patch.object(analysis, "execute", execute): result = asyncio.run(analysis.run("semgrep", "https://example.org", "example.org", "Test", 1, ModuleInputs(source_text=secret, sast_language="python"))) assert result["status"] == "findings" and result["alerts"][0]["source_line"] == 3 assert secret not in json.dumps(result) assert all(not path.exists() for path in paths) @pytest.mark.parametrize("report", [ {"results": [], "errors": [{"message": "fixture-private-error"}], "paths": {"scanned": []}}, {"results": [], "errors": [], "paths": {"scanned": []}}, {}, ]) def test_semgrep_errors_or_skipped_file_cannot_claim_clean(report): async def execute(args, timeout): Path(args[args.index("--output") + 1]).write_text(json.dumps(report)) return [], "completed" with patch.object(analysis, "execute", execute): result = asyncio.run(analysis.run("semgrep", "https://example.org", "example.org", "Test", 1, ModuleInputs(source_text="fixture-private-source", sast_language="python"))) assert result["status"] == "failed" assert "fixture-private" not in json.dumps(result) def test_semgrep_unknown_rule_messages_and_lines_are_never_trusted(): report = {"results": [{"check_id": "fixture-private-native-rule", "start": {"line": 1}, "extra": {"message": "fixture-private-message"}}, {"check_id": "asfb-python-dynamic-code", "start": {"line": True}}]} assert analysis.normalized_sast(report, "https://example.org") == [] class FakeResponse: status_code = 200 def __init__(self, body=b"fixture-private-body"): self.body, self._content = body, b"" def __enter__(self): return self def __exit__(self, *args): return False def iter_content(self, size): yield self.body class FakeSession: def __init__(self, response=None): self.calls, self.response, self.trust_env = [], response or FakeResponse(), True def get(self, target, **kwargs): self.calls.append((target, kwargs)) assert kwargs["allow_redirects"] is False and kwargs["verify"] is True assert kwargs["headers"] == {"User-Agent": "Test"} return self.response def close(self): pass SPEC = {"target": "https://example.org/app?q=1", "host": "example.org", "agent": "Test", "delay": 1} def test_waf_guard_preserves_scope_revalidates_dns_paces_and_caps_requests(): session, pauses = FakeSession(), [] with patch("app.waf_runner.public_host") as public: guard = Guard(SPEC, session, pauses.append) for _ in range(8): guard.request(params={"probe": "value"}) with pytest.raises(ProfileStopped, match="request_budget"): guard.request() assert len(session.calls) == 8 and public.call_count == 8 and len(pauses) == 7 assert session.trust_env is False assert all(target == SPEC["target"] for target, _ in session.calls) @pytest.mark.parametrize("path", ["//outside.org/", "https://outside.org/", "/bad\nheader"]) def test_waf_guard_rejects_out_of_scope_request_paths(path): session = FakeSession() with patch("app.waf_runner.public_host"): guard = Guard(SPEC, session) with pytest.raises(ProfileStopped, match="scope_blocked"): guard.request(path=path) assert not session.calls @pytest.mark.parametrize("probe", ["SLEEP(5)", "curl outside.org", "file:///etc/passwd", "/bin/cat"]) def test_waf_profile_does_not_forward_time_based_or_command_probes(probe): session = FakeSession() guard = Guard(SPEC, session) with pytest.raises(ProfileStopped, match="probe_blocked"): guard.request(params={"probe": probe}) assert not session.calls def test_waf_rate_limit_stops_the_profile_without_retries(): response = FakeResponse();response.status_code = 429 session = FakeSession(response) with patch("app.waf_runner.public_host"): guard = Guard(SPEC, session) with pytest.raises(ProfileStopped, match="rate_limited"): guard.request() assert len(session.calls) == 1 def test_waf_response_size_is_bounded_and_reported_partial(): class Engine: def __init__(self, *args, **kwargs): self.rq = self.Request() def identwaf(self, **kwargs): return ["Mock vendor"], "fixture-private-trigger" session = FakeSession(FakeResponse(b"x" * 262145)) with patch("app.waf_runner.public_host"): result = run_waf(SPEC, Engine, session) assert result["status"] == "partial" and result["stop_reason"] == "response_limit" assert result["vendors"] == ["Mock vendor"] assert "fixture-private" not in json.dumps(result) def test_guided_waf_inventory_precedes_web_reviews_but_manual_order_remains_exact(): requested = ["nuclei", "semgrep", "ffuf", "wafw00f", "httpx"] guided = planning.plan(requested, "guided") assert guided["effective"].index("wafw00f") < guided["effective"].index("ffuf") assert guided["effective"].index("wafw00f") < guided["effective"].index("nuclei") assert any(h["data"] == "WAF inventory / scan context" for h in guided["handoffs"]) manual = planning.plan(requested, "manual") assert manual["effective"] == requested and manual["warnings"] def test_worker_retains_waf_inventory_and_later_stage_context(): run, job = run_job(["wafw00f", "semgrep"]) async def module(name, *args, **kwargs): return {"status": "completed", "alerts": [], "waf": {"vendors": ["Mock vendor"], "requests": 2}} if name == "wafw00f" else {"status": "skipped", "alerts": []} with patch.object(main, "public_host"), patch.object(main, "run_web_module", module): asyncio.run(run()) assert job["waf"]["vendors"] == ["Mock vendor"] event = next(e for e in main.scan_events if e["event"] == "stage_finished" and e["module"] == "semgrep") assert event["diagnostics"]["waf_context"] == ["Mock vendor"] def test_source_language_is_public_but_source_is_not(): summary = ModuleInputs(source_text="fixture-private-source", sast_language="typescript").public_summary() assert summary["sast_language"] == "typescript" and "fixture-private" not in json.dumps(summary)