diff --git a/src/kambo/models.py b/src/kambo/models.py index 7d68b8d..506db4f 100644 --- a/src/kambo/models.py +++ b/src/kambo/models.py @@ -86,8 +86,24 @@ def confidence(self) -> Confidence: @property def confidence_pct(self) -> int: - """0-100 percentage representing confidence strength.""" - return min(100, int(self.total_weight * 40)) + """0-100 percentage representing confidence strength within tier. + + Mapping: + TENTATIVE (weight 0.0–0.99): 10–39% — single signal, high FP risk + FIRM (weight 1.0–1.99): 50–79% — multiple signals, reportable with caveats + CONFIRMED (weight 2.0+): 85–99% — exploit-grade, ready to submit + """ + w = self.total_weight + if w >= 2.0: + # Confirmed tier: starts at 85, caps at 99 + return min(99, 85 + int((w - 2.0) * 7)) + if w >= 1.0: + # Firm tier: 50–79, scales within the 1.0–2.0 range + return 50 + int((w - 1.0) * 29) + # Tentative tier: 10–39 (0.0 weight = 0% — no evidence at all) + if w == 0: + return 0 + return max(10, min(39, 10 + int(w * 29))) def add(self, signal: str, source: str, raw_data: str = "", weight: float = 1.0) -> EvidenceChain: """Return a new chain with the evidence item appended (immutable).""" diff --git a/src/kambo/tools/scanning.py b/src/kambo/tools/scanning.py index b9565b3..f941ca9 100644 --- a/src/kambo/tools/scanning.py +++ b/src/kambo/tools/scanning.py @@ -6,6 +6,7 @@ from kambo.models import Phase from kambo.parsers import parse_ffuf, parse_nmap from kambo.parsers.generic_parser import parse_json_output +from kambo.rate_limiter import get_rate_limiter from kambo.scope import validate_scope @@ -90,6 +91,10 @@ async def scan_directories( ) -> dict: """Web directory and file fuzzing with ffuf. + Integrates WAF detection and adaptive rate limiting — if a WAF is detected + from a probe response, the scan rate is automatically reduced to avoid + triggering blocks. + Args: target: Base URL to fuzz (e.g., https://example.com) wordlist: Path to wordlist inside container @@ -98,15 +103,51 @@ async def scan_directories( """ validate_scope(target) runner = get_runner() + limiter = get_rate_limiter() url = target if target.startswith("http") else f"https://{target}" + + # WAF probe: check a known-good path before fuzzing at full speed + probe_cmd = f'curl -s -D - "{url}/" 2>/dev/null | head -30' + probe_result = await runner.run(probe_cmd, "scan_directories_waf_probe", target, Phase.SCANNING, timeout=10) + + # Record the probe response through the rate limiter to detect WAF/blocking + probe_analysis = limiter.record_request( + target, + response_body=probe_result.raw_output, + headers=probe_result.raw_output[:500], # headers are in the first portion + ) + + # Adapt threading based on WAF detection + threads = 10 + extra_flags = "" + waf_name = probe_analysis.get("waf_detected") or "" + if waf_name: + threads = 2 + extra_flags = "-p 0.5" # 500ms delay between requests — be polite to WAF + elif probe_analysis.get("is_blocked"): + threads = 3 + extra_flags = "-p 0.2" + ext_flag = f"-e {extensions}" if extensions else "" - cmd = f"ffuf -u {url}/FUZZ -w {wordlist} -mc {match_codes} {ext_flag} -o /tmp/ffuf_out.json -of json -s 2>/dev/null && cat /tmp/ffuf_out.json" + cmd = ( + f"ffuf -u {url}/FUZZ -w {wordlist} -mc {match_codes} " + f"{ext_flag} -t {threads} {extra_flags} " + f"-o /tmp/ffuf_out.json -of json -s 2>/dev/null && cat /tmp/ffuf_out.json" + ) result = await runner.run(cmd, "scan_directories", target, Phase.SCANNING, timeout=120) parsed = parse_ffuf(result.raw_output) - return {"target": target, "scan_type": "directory_fuzz", **parsed} + return { + "target": target, + "scan_type": "directory_fuzz", + "waf_detected": bool(waf_name), + "waf_name": waf_name, + "rate_adapted": threads < 10, + "evasion_tips": probe_analysis.get("evasion_tips", []), + **parsed, + } async def scan_vhosts( diff --git a/src/kambo/validation.py b/src/kambo/validation.py index 06cc21f..e5e9ebc 100644 --- a/src/kambo/validation.py +++ b/src/kambo/validation.py @@ -213,8 +213,18 @@ def validate_xss( if payload_idx < 0: return chain # Payload was encoded after earlier check passed before_payload = raw_output[:payload_idx] - in_script = " tags — a closed tag means the payload is NOT inside it + script_opens = len(re.findall(r"", window, re.IGNORECASE)) + in_script = script_opens > script_closes + + # Comment detection: check if most recent + comment_window = before_payload[max(0, len(before_payload) - 200):] + last_open = comment_window.rfind("") + in_comment = last_open > last_close if in_script: chain = chain.add( @@ -342,10 +352,12 @@ def validate_ssrf( if re.search(r"(invalid url|bad request|url not allowed|blocked)", body_lower): chain = chain.add_fp_check("Server rejected the URL — input validation in place") - # Status code check (weak signal alone) - if status_code in ("200", "301", "302"): + # Status code check (weak signal alone) — 307/308 are permanent/temporary redirects + # that can indicate the server is fetching and forwarding the internal resource + if status_code in ("200", "301", "302", "307", "308"): + redirect_note = " (redirect — may indicate server-side fetch)" if status_code in ("301", "302", "307", "308") else "" chain = chain.add( - signal=f"HTTP {status_code} returned for internal payload: {payload[:100]}", + signal=f"HTTP {status_code} returned for internal payload: {payload[:100]}{redirect_note}", source="ssrf_probe", weight=0.2, ) @@ -1478,3 +1490,208 @@ def validate_deserialization( ) return chain + + +# --------------------------------------------------------------------------- +# validate_graphql — GraphQL Introspection & Injection +# --------------------------------------------------------------------------- + +_GRAPHQL_INTROSPECTION_SIGNATURES: list[str] = [ + r'"__schema"\s*:', # introspection response root + r'"types"\s*:\s*\[', # types array under __schema (real GQL responses) + r'"queryType"\s*:.*"name"', # schema root types + r'"mutationType"\s*:', # mutation support exposed + r'"directives"\s*:\s*\[', # directives listing +] + +_GRAPHQL_ERROR_SIGNATURES: list[tuple[str, str, float]] = [ + (r'"errors"\s*:.*"extensions"\s*:.*"code"', "GraphQL structured error with code — endpoint confirmed", 0.4), + # Use \W+ to handle both plain quotes and JSON-escaped \" sequences in raw HTTP bodies + (r"Cannot query field\W+(\w+)\W+on type", "Field-level error leaks schema info", 0.6), + (r"Unknown argument\W+(\w+)", "Argument error reveals accepted fields", 0.5), + (r"Variable.*must not be null", "Null variable error leaks input type", 0.3), + (r"Did you mean\W+(\w+)", "Suggestion error fully leaks field names", 0.8), +] + +_GRAPHQL_INJECTION_SIGNATURES: list[tuple[str, str, float]] = [ + (r"syntax error.*unexpected", "GraphQL syntax error from injected chars", 0.3), + (r"Expected.*got.*", "Incomplete query — injection point confirmed", 0.5), + (r"(sql|mongo|redis).*error", "Database error via GraphQL injection", 1.2), + (r"Internal\s+Server\s+Error.*graphql", "Internal error exposed via GraphQL", 0.6), +] + + +def validate_graphql( + output: str, + introspection_response: str = "", + error_response: str = "", + injection_response: str = "", + baseline_body: str = "", +) -> EvidenceChain: + """Validate GraphQL security issues: introspection enabled, info disclosure, injection. + + Three separate attack vectors: + 1. Introspection enabled — leaks full schema, used for targeted attacks + 2. Error verbosity — field names, types, and suggestions leak from errors + 3. Query injection — injected chars cause parse/DB errors + + Args: + output: General tool output (combines all responses if not split) + introspection_response: Response to an __schema introspection query + error_response: Response to an intentionally malformed query + injection_response: Response to an injection payload (e.g., ' OR 1=1) + baseline_body: Normal query response for comparison + """ + chain = EvidenceChain() + + # Combine all available response data + all_output = "\n".join(filter(None, [output, introspection_response, error_response, injection_response])) + + if not all_output.strip(): + return chain.add_fp_check("No GraphQL response data — endpoint may not exist") + + # FP check: not a GraphQL endpoint + if re.search(r"(404 not found|cannot (get|post)|method not allowed)", all_output, re.IGNORECASE): + if not re.search(r'"data"\s*:|"errors"\s*:', all_output): + return chain.add_fp_check("No GraphQL response structure — likely not a GraphQL endpoint") + + # FP check: introspection explicitly disabled + if re.search(r"(introspection.*disabled|introspection.*not.*allowed|GraphQL introspection is not allowed)", all_output, re.IGNORECASE): + chain = chain.add_fp_check("Introspection explicitly disabled — properly configured") + + # --- Introspection detection --- + if introspection_response: + for pattern in _GRAPHQL_INTROSPECTION_SIGNATURES: + if re.search(pattern, introspection_response, re.IGNORECASE): + chain = chain.add( + signal="GraphQL introspection enabled — full schema exposed", + source="graphql_introspection", + raw_data=introspection_response[:500], + weight=1.0, + ) + # Count types exposed — more types = bigger attack surface + type_count_match = re.findall(r'"name"\s*:\s*"[A-Z]\w+"', introspection_response) + if len(type_count_match) > 5: + chain = chain.add( + signal=f"Schema exposes {len(type_count_match)} named types — large attack surface", + source="graphql_introspection", + weight=0.5, + ) + break + + # --- Error verbosity detection --- + if error_response: + for pattern, description, weight in _GRAPHQL_ERROR_SIGNATURES: + if re.search(pattern, error_response, re.IGNORECASE): + chain = chain.add( + signal=description, + source="graphql_error_analysis", + raw_data=_extract_context(error_response, pattern), + weight=weight, + ) + + # --- Injection detection --- + if injection_response: + for pattern, description, weight in _GRAPHQL_INJECTION_SIGNATURES: + if re.search(pattern, injection_response, re.IGNORECASE): + chain = chain.add( + signal=description, + source="graphql_injection", + raw_data=_extract_context(injection_response, pattern), + weight=weight, + ) + break + + # Baseline comparison + if baseline_body and output and output.strip() != baseline_body.strip(): + if abs(len(output) - len(baseline_body)) > 200: + chain = chain.add( + signal="Response differs significantly from baseline", + source="graphql_baseline_comparison", + weight=0.2, + ) + + return chain + + +# --------------------------------------------------------------------------- +# validate_race_condition — TOCTOU / Race Condition +# --------------------------------------------------------------------------- + +def validate_race_condition( + responses: list[tuple[str, int, str]], + expected_unique: int = 1, + action: str = "", + baseline_count: int | None = None, +) -> EvidenceChain: + """Validate race condition / TOCTOU findings. + + A race condition is confirmed when parallel requests cause a state change + that should only happen once (e.g., coupon used twice, balance deducted once + but credit applied multiple times). + + Args: + responses: List of (status_code_str, body_hash_int, body_preview) tuples + from parallel requests sent simultaneously + expected_unique: How many unique success responses are expected (usually 1) + action: Description of the action being raced (for signal context) + baseline_count: How many times the action should succeed (usually 1) + """ + chain = EvidenceChain() + + if not responses: + return chain.add_fp_check("No responses — race condition test produced no data") + + total = len(responses) + success_responses = [r for r in responses if r[0] in ("200", "201", "204")] + error_responses = [r for r in responses if r[0] in ("409", "429", "400", "403")] + + # FP check: all requests failed (server properly serialized) + if not success_responses: + chain = chain.add_fp_check( + f"All {total} parallel requests failed — server properly handles concurrent requests" + ) + return chain + + # FP check: exactly expected successes, rest failed — proper behavior + if len(success_responses) <= expected_unique and len(error_responses) >= total - expected_unique - 1: + chain = chain.add_fp_check( + f"Only {len(success_responses)}/{total} requests succeeded — concurrency handled correctly" + ) + return chain + + # Primary signal: more successes than expected + extra_successes = len(success_responses) - expected_unique + if extra_successes > 0: + chain = chain.add( + signal=f"Race condition: {len(success_responses)}/{total} requests succeeded " + f"(expected max {expected_unique}){' for: ' + action if action else ''}", + source="race_condition_analysis", + weight=1.0 + min(1.0, extra_successes * 0.2), # more extras = stronger signal + ) + + # Check response body variance — identical success bodies = same resource hit twice + body_hashes = [r[1] for r in success_responses] + unique_hashes = set(body_hashes) + if len(unique_hashes) == 1 and len(success_responses) > 1: + chain = chain.add( + signal=f"All {len(success_responses)} success responses are identical — same state applied multiple times", + source="race_condition_identity_check", + weight=0.8, + ) + elif len(unique_hashes) > 1: + chain = chain.add( + signal=f"Varied responses across {len(success_responses)} successes — concurrent state mutation confirmed", + source="race_condition_variance_check", + weight=0.5, + ) + + # Baseline comparison: more successes than baseline implies race exploit + if baseline_count is not None and len(success_responses) > baseline_count: + chain = chain.add( + signal=f"Parallel requests produced {len(success_responses)} successes vs baseline {baseline_count}", + source="race_condition_baseline", + weight=0.5, + ) + + return chain diff --git a/tests/test_models.py b/tests/test_models.py index 5f04f23..bef9205 100644 --- a/tests/test_models.py +++ b/tests/test_models.py @@ -55,11 +55,14 @@ def test_confirmed_threshold(self) -> None: assert chain.confidence == Confidence.CONFIRMED assert chain.total_weight == 2.0 - def test_confidence_pct_capped_at_100(self) -> None: + def test_confidence_pct_confirmed_never_reaches_100(self) -> None: + # Certainty is never absolute — CONFIRMED caps at 99% (not 100) chain = EvidenceChain() chain = chain.add("s1", "src", weight=1.5) chain = chain.add("s2", "src", weight=1.5) - assert chain.confidence_pct == 100 # capped + pct = chain.confidence_pct + assert pct >= 85, f"CONFIRMED should be >= 85%, got {pct}%" + assert pct <= 99, f"CONFIRMED should never reach 100%, got {pct}%" def test_add_fp_check_returns_new_chain(self) -> None: chain = EvidenceChain() diff --git a/tests/test_refine_improvements.py b/tests/test_refine_improvements.py new file mode 100644 index 0000000..a02487f --- /dev/null +++ b/tests/test_refine_improvements.py @@ -0,0 +1,501 @@ +"""Tests for the six improvements identified in the refine pass. + +1. confidence_pct formula — correct tier mapping +2. validate_xss — closed

Hello {payload}

' + chain = validate_xss(response, payload, "q") + signals = [item.signal.lower() for item in chain.items] + # The "inside " + chain = validate_xss(response, payload, "q") + has_script_signal = any( + "script" in item.signal.lower() and "inside" in item.signal.lower() + for item in chain.items + ) + assert has_script_signal, "Open " + response = f" {payload}" + chain = validate_xss(response, payload, "q") + fp_checks = " ".join(chain.false_positive_checks).lower() + assert "comment" not in fp_checks, ( + f"Closed HTML comment incorrectly blocked XSS. FP checks: {chain.false_positive_checks}" + ) + + def test_multiple_script_tags_with_one_closed(self) -> None: + payload = "alert(1)" + # Two opens, two closes before payload — balanced, payload is NOT in script + response = f"

{payload}

" + chain = validate_xss(response, payload, "q") + signals = [item.signal.lower() for item in chain.items] + assert not any("script" in s and "inside" in s for s in signals), ( + f"Balanced script tags incorrectly flagged. Signals: {chain.items}" + ) + + +# --------------------------------------------------------------------------- +# Fix 3: validate_ssrf — 307/308 status codes +# --------------------------------------------------------------------------- + +class TestSsrfRedirectCodes: + """307 and 308 must be accepted as valid SSRF indicators.""" + + def test_307_accepted(self) -> None: + chain = validate_ssrf( + payload="http://169.254.169.254/latest/meta-data/", + status_code="307", + response_body="ami-id: ami-12345678", + ) + # Should have evidence (cloud metadata matched) + assert chain.total_weight > 0 + + def test_308_accepted(self) -> None: + chain = validate_ssrf( + payload="http://metadata.google.internal/", + status_code="308", + response_body="project-id: my-project instance-id: 123", + ) + assert chain.total_weight > 0 + + def test_307_redirect_with_no_metadata_is_weak(self) -> None: + chain = validate_ssrf( + payload="http://internal-host/", + status_code="307", + response_body="redirect page", + ) + # Status alone gives 0.2 weight — tentative signal only + assert chain.total_weight < 1.0 + + def test_non_redirect_blocked(self) -> None: + chain = validate_ssrf( + payload="http://internal-host/", + status_code="403", + response_body="Access Denied", + ) + assert chain.total_weight == 0.0 + assert any("blocked" in c.lower() for c in chain.false_positive_checks) + + def test_200_still_works(self) -> None: + chain = validate_ssrf( + payload="http://169.254.169.254/", + status_code="200", + response_body="ami-id: ami-12345 iam: role-name security-credentials: key", + ) + assert chain.total_weight >= 1.5 + + +# --------------------------------------------------------------------------- +# Feature 4a: validate_graphql — introspection +# --------------------------------------------------------------------------- + +class TestGraphqlIntrospection: + """Introspection enabled = confirmed schema exposure.""" + + def test_introspection_enabled_detected(self) -> None: + introspection_resp = '{"data": {"__schema": {"queryType": {"name": "Query"}, "mutationType": {"name": "Mutation"}, "directives": []}}}' + chain = validate_graphql( + output="", + introspection_response=introspection_resp, + ) + assert chain.total_weight >= 1.0, f"Introspection should give FIRM+ confidence, got {chain.total_weight}" + + def test_disabled_introspection_is_fp(self) -> None: + chain = validate_graphql( + output='{"errors": [{"message": "GraphQL introspection is not allowed"}]}', + introspection_response='{"errors": [{"message": "GraphQL introspection is not allowed"}]}', + ) + fp_checks = " ".join(chain.false_positive_checks).lower() + assert "disabled" in fp_checks or "not allowed" in fp_checks + + def test_large_schema_gets_bonus_weight(self) -> None: + # Build a schema with >5 named types + types = [f'"name": "Type{i}"' for i in range(10)] + introspection_resp = '{' + ', '.join(types) + ', "__schema": {"queryType": {"name": "Query"}}}' + chain = validate_graphql( + output="", + introspection_response=introspection_resp, + ) + assert chain.total_weight >= 1.5 + + def test_not_graphql_endpoint_flagged(self) -> None: + chain = validate_graphql(output="404 Not Found") + assert chain.total_weight == 0 or len(chain.false_positive_checks) > 0 + + +class TestGraphqlErrors: + """Error verbosity exposes schema info.""" + + def test_suggestion_error_high_weight(self) -> None: + error_resp = '{"errors": [{"message": "Did you mean \\"userName\\"?"}]}' + chain = validate_graphql(output="", error_response=error_resp) + assert chain.total_weight >= 0.8 + + def test_field_error_medium_weight(self) -> None: + error_resp = '{"errors": [{"message": "Cannot query field \\"password\\" on type \\"User\\""}]}' + chain = validate_graphql(output="", error_response=error_resp) + assert chain.total_weight >= 0.5 + + def test_generic_error_no_schema_leak(self) -> None: + error_resp = '{"errors": [{"message": "An error occurred"}]}' + chain = validate_graphql(output="", error_response=error_resp) + # Generic error = no schema info leaked = low signal + assert chain.total_weight < 0.5 + + +class TestGraphqlInjection: + """Injection detection via parse errors.""" + + def test_db_error_via_graphql(self) -> None: + injection_resp = '{"errors": [{"message": "sql error: syntax error near \'injected\'"}]}' + chain = validate_graphql(output="", injection_response=injection_resp) + assert chain.total_weight >= 1.0 # DB error = high weight + + def test_parse_error_from_injection(self) -> None: + injection_resp = '{"errors": [{"message": "Expected Name, got "}]}' + chain = validate_graphql(output="", injection_response=injection_resp) + assert chain.total_weight >= 0.5 + + +# --------------------------------------------------------------------------- +# Feature 4b: validate_race_condition +# --------------------------------------------------------------------------- + +class TestRaceCondition: + """Race condition detection via parallel response analysis.""" + + def test_multiple_successes_detected(self) -> None: + # 5 parallel requests, all succeeded — expected only 1 + responses = [("200", hash("success_body"), "success") for _ in range(5)] + chain = validate_race_condition( + responses=responses, + expected_unique=1, + action="redeem_coupon", + ) + assert chain.total_weight >= 1.0 + assert chain.confidence in (Confidence.FIRM, Confidence.CONFIRMED) + + def test_single_success_is_not_vulnerable(self) -> None: + # Only 1 out of 10 parallel requests succeeded — proper serialization + responses = [("200", hash("success"), "ok")] + [("409", 0, "conflict") for _ in range(9)] + chain = validate_race_condition( + responses=responses, + expected_unique=1, + ) + assert chain.total_weight == 0 + assert any("correctly" in c.lower() or "properly" in c.lower() for c in chain.false_positive_checks) + + def test_all_failures_is_not_vulnerable(self) -> None: + responses = [("409", 0, "conflict") for _ in range(10)] + chain = validate_race_condition(responses=responses, expected_unique=1) + assert chain.total_weight == 0 + assert len(chain.false_positive_checks) > 0 + + def test_identical_success_bodies_strengthen_signal(self) -> None: + # Identical bodies = same resource applied multiple times + body_hash = hash("same_state") + responses = [("200", body_hash, "same") for _ in range(3)] + chain = validate_race_condition(responses=responses, expected_unique=1) + signal_texts = " ".join(item.signal.lower() for item in chain.items) + assert "identical" in signal_texts or "same" in signal_texts + + def test_varied_responses_add_variance_signal(self) -> None: + responses = [ + ("200", hash("response_a"), "state_a"), + ("200", hash("response_b"), "state_b"), + ("200", hash("response_c"), "state_c"), + ] + chain = validate_race_condition(responses=responses, expected_unique=1) + assert chain.total_weight > 0 + + def test_empty_responses_is_fp(self) -> None: + chain = validate_race_condition(responses=[], expected_unique=1) + assert chain.total_weight == 0 + assert len(chain.false_positive_checks) > 0 + + def test_baseline_count_comparison(self) -> None: + responses = [("200", hash("ok"), "ok") for _ in range(4)] + chain = validate_race_condition( + responses=responses, + expected_unique=1, + baseline_count=1, + ) + baseline_signals = [item.signal for item in chain.items if "baseline" in item.signal.lower()] + assert len(baseline_signals) > 0 + + def test_weight_scales_with_extra_successes(self) -> None: + # More extra successes = higher weight + responses_2 = [("200", hash("x"), "x") for _ in range(2)] + responses_5 = [("200", hash("x"), "x") for _ in range(5)] + chain_2 = validate_race_condition(responses=responses_2, expected_unique=1) + chain_5 = validate_race_condition(responses=responses_5, expected_unique=1) + assert chain_5.total_weight >= chain_2.total_weight + + +# --------------------------------------------------------------------------- +# Fix 6: scan_directories — integration tests with monkeypatched fakes +# --------------------------------------------------------------------------- + +from types import SimpleNamespace + + +def _make_tool_result(raw_output: str) -> SimpleNamespace: + """Create a minimal ToolResult-compatible stub.""" + return SimpleNamespace(raw_output=raw_output, exit_code=0) + + +def _fake_runner(probe_raw: str = "HTTP/1.1 200 OK", scan_raw: str = "") -> object: + """Return a runner stub: first run() call = probe, subsequent = scan.""" + + class _Runner: + def __init__(self) -> None: + self.commands: list[str] = [] + self._call = 0 + + async def run(self, cmd: str, *args, **kwargs) -> SimpleNamespace: + self.commands.append(cmd) + self._call += 1 + if self._call == 1: + return _make_tool_result(probe_raw) + return _make_tool_result(scan_raw or '{"results":[],"commandline":""}') + + return _Runner() + + +def _fake_limiter(probe_analysis: dict) -> object: + """Return a rate-limiter stub whose record_request returns probe_analysis.""" + + class _Limiter: + def record_request(self, target: str, **kwargs: object) -> dict: + return probe_analysis + + return _Limiter() + + +class TestScanDirectoriesRateLimiterIntegration: + """Integration tests: scan_directories WAF detection + adaptive rate limiting.""" + + # --- source-level checks (kept for fast regression) --- + + def test_rate_limiter_imported_in_scanning_module(self) -> None: + import inspect + import kambo.tools.scanning as scanning_module + + source = inspect.getsource(scanning_module) + assert "get_rate_limiter" in source, "get_rate_limiter not called in scanning.py" + assert "rate_limiter" in source.lower(), "rate_limiter not imported in scanning.py" + + # --- WAF-detected scenario --- + + async def test_waf_detected_reduces_threads_to_2(self, monkeypatch) -> None: + import kambo.tools.scanning as sm + + runner = _fake_runner() + analysis = { + "waf_detected": "Cloudflare", + "is_blocked": False, + "evasion_tips": [{"technique": "ip_rotation", "description": "rotate IPs"}], + "recommended_delay_seconds": 0.5, + } + monkeypatch.setattr(sm, "get_runner", lambda: runner) + monkeypatch.setattr(sm, "get_rate_limiter", lambda: _fake_limiter(analysis)) + monkeypatch.setattr(sm, "validate_scope", lambda t: None) + + result = await sm.scan_directories("https://example.com") + + # ffuf command should contain -t 2 and -p 0.5 + ffuf_cmd = runner.commands[1] # second call is the real scan + assert "-t 2" in ffuf_cmd, f"Expected '-t 2' in ffuf cmd, got: {ffuf_cmd}" + assert "-p 0.5" in ffuf_cmd, f"Expected '-p 0.5' in ffuf cmd, got: {ffuf_cmd}" + + # result metadata + assert result["waf_detected"] is True + assert result["waf_name"] == "Cloudflare" + assert result["rate_adapted"] is True + assert len(result["evasion_tips"]) > 0 + + # --- blocked (no named WAF) scenario --- + + async def test_blocked_without_waf_reduces_threads_to_3(self, monkeypatch) -> None: + import kambo.tools.scanning as sm + + runner = _fake_runner() + analysis = { + "waf_detected": None, + "is_blocked": True, + "evasion_tips": [], + "recommended_delay_seconds": 0.2, + } + monkeypatch.setattr(sm, "get_runner", lambda: runner) + monkeypatch.setattr(sm, "get_rate_limiter", lambda: _fake_limiter(analysis)) + monkeypatch.setattr(sm, "validate_scope", lambda t: None) + + result = await sm.scan_directories("https://example.com") + + ffuf_cmd = runner.commands[1] + assert "-t 3" in ffuf_cmd, f"Expected '-t 3' in blocked cmd, got: {ffuf_cmd}" + assert "-p 0.2" in ffuf_cmd, f"Expected '-p 0.2' in blocked cmd, got: {ffuf_cmd}" + + assert result["waf_detected"] is False + assert result["waf_name"] == "" + assert result["rate_adapted"] is True + + # --- clean (no WAF, no block) scenario --- + + async def test_no_waf_uses_full_threads(self, monkeypatch) -> None: + import kambo.tools.scanning as sm + + runner = _fake_runner() + analysis = { + "waf_detected": None, + "is_blocked": False, + "evasion_tips": [], + "recommended_delay_seconds": 0.1, + } + monkeypatch.setattr(sm, "get_runner", lambda: runner) + monkeypatch.setattr(sm, "get_rate_limiter", lambda: _fake_limiter(analysis)) + monkeypatch.setattr(sm, "validate_scope", lambda t: None) + + result = await sm.scan_directories("https://example.com") + + ffuf_cmd = runner.commands[1] + assert "-t 10" in ffuf_cmd, f"Expected '-t 10' (full threads), got: {ffuf_cmd}" + # No delay flag expected in the clean path + assert "-p " not in ffuf_cmd, f"Unexpected delay flag in clean cmd: {ffuf_cmd}" + + assert result["waf_detected"] is False + assert result["rate_adapted"] is False + assert result["evasion_tips"] == [] + + # --- result dict structure --- + + async def test_result_always_contains_waf_metadata_keys(self, monkeypatch) -> None: + import kambo.tools.scanning as sm + + runner = _fake_runner() + analysis = {"waf_detected": None, "is_blocked": False, "evasion_tips": []} + monkeypatch.setattr(sm, "get_runner", lambda: runner) + monkeypatch.setattr(sm, "get_rate_limiter", lambda: _fake_limiter(analysis)) + monkeypatch.setattr(sm, "validate_scope", lambda t: None) + + result = await sm.scan_directories("https://example.com") + + for key in ("waf_detected", "waf_name", "rate_adapted", "evasion_tips"): + assert key in result, f"Missing key in result: {key}" + + # --- probe fires before the ffuf scan --- + + async def test_probe_runs_before_ffuf(self, monkeypatch) -> None: + import kambo.tools.scanning as sm + + runner = _fake_runner(probe_raw="HTTP/1.1 200 OK cf-ray: abc") + analysis = {"waf_detected": None, "is_blocked": False, "evasion_tips": []} + monkeypatch.setattr(sm, "get_runner", lambda: runner) + monkeypatch.setattr(sm, "get_rate_limiter", lambda: _fake_limiter(analysis)) + monkeypatch.setattr(sm, "validate_scope", lambda t: None) + + await sm.scan_directories("https://example.com") + + assert len(runner.commands) == 2, "Expected exactly 2 runner calls (probe + scan)" + assert "curl" in runner.commands[0], "First call should be the curl WAF probe" + assert "ffuf" in runner.commands[1], "Second call should be the ffuf scan"