File size: 8,074 Bytes
54f5b77 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 | 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)
|