Refine: Fix confidence tiers, XSS/SSRF validators, add GraphQL & race condition checks - #5
Merged
Merged
Conversation
Bugs fixed: - validate_xss: script context check now counts open/closed <script> tags instead of just checking presence — prevents FP when payload is after a closed </script> tag - EvidenceChain.confidence_pct: new tiered formula (TENTATIVE 10-39%, FIRM 50-79%, CONFIRMED 85-99%) — old flat formula showed CONFIRMED as 80% - validate_ssrf: added HTTP 307/308 as valid status codes for SSRF detection New validators: - validate_graphql: introspection detection, error verbosity analysis, and query injection detection (Did you mean / Cannot query field patterns handle raw JSON with escaped quotes via \W+ matching) - validate_race_condition: TOCTOU detection via parallel response analysis, scales weight with extra successes, detects identical bodies Integration: - scan_directories: first tool to use rate_limiter — WAF probe before fuzzing, adaptive thread count (10 → 2 on WAF, 10 → 3 on block) with ffuf -p delay flag when WAF detected Tests: +41 new tests (613 total, all passing) https://claude.ai/code/session_01QrWLmqz6sz4tMVvDWHJqrU
Reviewer's GuideRefines confidence scoring and several validators (XSS, SSRF), introduces new GraphQL and race-condition validators, and integrates WAF-aware rate limiting into directory scanning, with a dedicated regression test suite for all six improvements. Sequence diagram for WAF-aware scan_directories rate limitingsequenceDiagram
participant Scan as scan_directories
participant Runner as get_runner
participant Limiter as get_rate_limiter
participant Ffuf as ffuf
Scan->>Runner: get_runner()
Scan->>Limiter: get_rate_limiter()
Note over Scan: Build probe_cmd with curl to target root
Scan->>Runner: run(probe_cmd, "scan_directories_waf_probe", target, Phase.SCANNING)
Runner-->>Scan: probe_result
Scan->>Limiter: record_request(target, response_body, headers)
Limiter-->>Scan: probe_analysis
alt probe_analysis.waf_detected
Scan->>Scan: threads = 2, extra_flags = "-p 0.5"
else probe_analysis.is_blocked
Scan->>Scan: threads = 3, extra_flags = "-p 0.2"
else normal
Scan->>Scan: threads = 10, extra_flags = ""
end
Note over Scan: Build ffuf cmd with -t threads and extra_flags
Scan->>Runner: run(cmd, "scan_directories", target, Phase.SCANNING)
Runner-->>Scan: result
Scan->>Scan: parse_ffuf(result.raw_output)
Scan-->>Scan: return {target, scan_type,
Scan-->>Scan: waf_detected, waf_name,
Scan-->>Scan: rate_adapted, evasion_tips, **parsed}
File-Level Changes
Tips and commandsInteracting with Sourcery
Customizing Your ExperienceAccess your dashboard to:
Getting Help
|
Regression validator
No regressions or failures detected. |
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
There was a problem hiding this comment.
Hey - I've found 2 issues
Prompt for AI Agents
Please address the comments from this code review:
## Individual Comments
### Comment 1
<location path="src/kambo/validation.py" line_range="1499-1501" />
<code_context>
+# validate_graphql — GraphQL Introspection & Injection
+# ---------------------------------------------------------------------------
+
+_GRAPHQL_INTROSPECTION_SIGNATURES: list[str] = [
+ r'"__schema"\s*:', # introspection response
+ r'"__types"\s*:', # type listing
+ r'"queryType"\s*:.*"name"', # schema root types
+ r'"mutationType"\s*:', # mutation support exposed
</code_context>
<issue_to_address>
**issue (bug_risk):** The '__types' introspection signature looks incorrect and may never match real GraphQL responses.
GraphQL introspection responses expose types under `__schema.types`, e.g. `{"data": {"__schema": {"types": [...]}}}`. The current `"__types"` pattern doesn’t match real responses, so this signature likely never fires. Consider matching `"types"` under `"__schema"` (or a more specific path) instead to detect introspection reliably.
</issue_to_address>
### Comment 2
<location path="tests/test_refine_improvements.py" line_range="339-348" />
<code_context>
+class TestScanDirectoriesRateLimiterIntegration:
</code_context>
<issue_to_address>
**suggestion (testing):** The scan_directories/WAF integration tests only verify imports and helpers, not the adaptive threading behavior of scan_directories itself
Since `scan_directories` now adapts `ffuf` threads/delay and returns WAF metadata, the tests should exercise it directly rather than just its collaborators. Please add an async integration-style test that monkeypatches `get_runner` and `get_rate_limiter` to return fakes:
- Fake rate limiter: return different `probe_analysis` dicts (e.g., `{"waf_detected": "Cloudflare"}`, `{ "is_blocked": True }`, and a normal case) and assert the constructed `ffuf` command includes the expected `-t` and `-p` values.
- Fake runner: return canned `probe_result` and `scan` output and assert the `scan_directories` result contains `waf_detected`, `waf_name`, `rate_adapted`, and `evasion_tips` with correct values.
This will validate the end-to-end WAF detection and rate adaptation behavior.
Suggested implementation:
```python
class TestScanDirectoriesRateLimiterIntegration:
"""Verify rate_limiter integration and adaptive behavior in scan_directories."""
def test_rate_limiter_imported_in_scanning_module(self) -> None:
import kambo.tools.scanning as scanning_module
import kambo.rate_limiter as rl_module # noqa: F401
# The module should reference the rate limiter
import inspect
source = inspect.getsource(scanning_module)
assert "get_rate_limiter" in source, "get_rate_limiter not used in scanning.py"
assert "rate_limiter" in source.lower(), "rate_limiter not imported in scanning.py"
async def test_scan_directories_waf_rate_adaptation(self, monkeypatch, tmp_path) -> None:
"""
Integration-style test for scan_directories WAF detection and rate adaptation.
We monkeypatch get_rate_limiter and get_runner to inject fakes, then assert:
- ffuf command contains expected -t (threads) and -p (delay) based on probe_analysis
- the returned scan_directories result contains WAF-related metadata
"""
import kambo.tools.scanning as scanning_module
captured_commands = []
class FakeRunner:
def __init__(self, probe_output: str, scan_output: str) -> None:
self.probe_output = probe_output
self.scan_output = scan_output
async def run(self, cmd, *args, **kwargs):
# record the command so we can assert on -t and -p
captured_commands.append(list(cmd))
# emulate the structure scan_directories expects from the runner
return {
"probe_result": self.probe_output,
"scan_result": self.scan_output,
}
class FakeRateLimiter:
def __init__(self, probe_analysis):
self.probe_analysis = probe_analysis
async def analyze_probe(self, *_args, **_kwargs):
# emulate whatever scan_directories uses from the rate limiter
return self.probe_analysis
# Different WAF scenarios to exercise:
scenarios = [
{
"probe_analysis": {"waf_detected": "Cloudflare", "threads": 5, "delay": 0.3},
"expected_waf_name": "Cloudflare",
"expected_rate_adapted": True,
},
{
"probe_analysis": {"is_blocked": True, "threads": 1, "delay": 1.0},
"expected_waf_name": None,
"expected_rate_adapted": True,
},
{
"probe_analysis": {"threads": 25, "delay": 0.0},
"expected_waf_name": None,
"expected_rate_adapted": False,
},
]
async def run_scenario(scenario):
captured_commands.clear()
def fake_get_rate_limiter(*_args, **_kwargs):
# scan_directories should call this factory to get a rate limiter instance
return FakeRateLimiter(probe_analysis=scenario["probe_analysis"])
def fake_get_runner(*_args, **_kwargs):
# scan_directories should call this factory to get a runner
return FakeRunner(
probe_output="fake_probe_output",
scan_output="fake_scan_output",
)
monkeypatch.setattr(scanning_module, "get_rate_limiter", fake_get_rate_limiter)
monkeypatch.setattr(scanning_module, "get_runner", fake_get_runner)
# Minimal realistic call into scan_directories
# (tmp_path provides a concrete wordlist path)
wordlist = tmp_path / "words.txt"
wordlist.write_text("admin\nlogin\n")
result = await scanning_module.scan_directories(
url="https://example.com",
wordlist_path=str(wordlist),
)
# 1) Assert we actually invoked ffuf with a command that includes -t and -p
assert captured_commands, "Runner was not invoked by scan_directories"
ffuf_cmd = captured_commands[-1]
assert "-t" in ffuf_cmd, "ffuf command missing -t (threads) flag"
assert "-p" in ffuf_cmd, "ffuf command missing -p (delay) flag"
t_index = ffuf_cmd.index("-t") + 1
p_index = ffuf_cmd.index("-p") + 1
# Note: we only assert presence/consistency, not exact numeric values,
# because the mapping logic belongs to scan_directories.
assert ffuf_cmd[t_index], "threads value after -t must not be empty"
assert ffuf_cmd[p_index], "delay value after -p must not be empty"
# 2) Assert WAF / rate-limiter metadata is surfaced in the result
assert "waf_detected" in result
assert "waf_name" in result
assert "rate_adapted" in result
assert "evasion_tips" in result
expected_waf_name = scenario["expected_waf_name"]
if expected_waf_name:
assert result["waf_detected"] is True
assert result["waf_name"] == expected_waf_name
else:
# for non-Cloudflare scenarios we just assert the key exists;
# the implementation may use None/False/"" to indicate "no WAF"
assert result["waf_detected"] in (False, None)
if scenario["expected_rate_adapted"]:
assert result["rate_adapted"] is True
assert isinstance(result["evasion_tips"], (list, tuple))
else:
assert result["rate_adapted"] in (False, None)
# Run all scenarios
for scenario in scenarios:
await run_scenario(scenario)
```
The fake `FakeRunner` and `FakeRateLimiter` in this test assume:
- `scan_directories` imports `get_runner` and `get_rate_limiter` from `kambo.tools.scanning` and calls them as factories.
- The object returned by `get_runner` exposes an async `run(cmd, *args, **kwargs)` that returns a dict with `"probe_result"` and `"scan_result"` keys.
- The object returned by `get_rate_limiter` exposes an async `analyze_probe(...)` method that returns the `probe_analysis` dict used to derive threads/delay and WAF metadata.
- `scan_directories` is an async function taking at least `url` and `wordlist_path` parameters and returning a dict with `"waf_detected"`, `"waf_name"`, `"rate_adapted"`, and `"evasion_tips"` keys.
If your actual `scan_directories` / rate-limiter / runner APIs differ, you should:
1. Adjust `FakeRunner.run`’s signature and returned structure to match what `scan_directories` expects.
2. Adjust `FakeRateLimiter`’s method name and signature to align with how the real rate limiter is used (e.g., `analyze`, `analyze_result`, etc.).
3. Update the monkeypatch targets (`scanning_module.get_runner`, `scanning_module.get_rate_limiter`) if those helpers live under different names or modules.
4. If `scan_directories` is synchronous, remove `async`/`await` usage and `pytest.mark.asyncio` (if present around the test) and call it directly instead.
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
Comment on lines
+339
to
+348
| class TestScanDirectoriesRateLimiterIntegration: | ||
| """Verify rate_limiter is imported and used in scan_directories.""" | ||
|
|
||
| def test_rate_limiter_imported_in_scanning_module(self) -> None: | ||
| import kambo.tools.scanning as scanning_module | ||
| import kambo.rate_limiter as rl_module | ||
| # The module should reference the rate limiter | ||
| import inspect | ||
| source = inspect.getsource(scanning_module) | ||
| assert "get_rate_limiter" in source, "get_rate_limiter not used in scanning.py" |
There was a problem hiding this comment.
suggestion (testing): The scan_directories/WAF integration tests only verify imports and helpers, not the adaptive threading behavior of scan_directories itself
Since scan_directories now adapts ffuf threads/delay and returns WAF metadata, the tests should exercise it directly rather than just its collaborators. Please add an async integration-style test that monkeypatches get_runner and get_rate_limiter to return fakes:
- Fake rate limiter: return different
probe_analysisdicts (e.g.,{"waf_detected": "Cloudflare"},{ "is_blocked": True }, and a normal case) and assert the constructedffufcommand includes the expected-tand-pvalues. - Fake runner: return canned
probe_resultandscanoutput and assert thescan_directoriesresult containswaf_detected,waf_name,rate_adapted, andevasion_tipswith correct values.
This will validate the end-to-end WAF detection and rate adaptation behavior.
Suggested implementation:
class TestScanDirectoriesRateLimiterIntegration:
"""Verify rate_limiter integration and adaptive behavior in scan_directories."""
def test_rate_limiter_imported_in_scanning_module(self) -> None:
import kambo.tools.scanning as scanning_module
import kambo.rate_limiter as rl_module # noqa: F401
# The module should reference the rate limiter
import inspect
source = inspect.getsource(scanning_module)
assert "get_rate_limiter" in source, "get_rate_limiter not used in scanning.py"
assert "rate_limiter" in source.lower(), "rate_limiter not imported in scanning.py"
async def test_scan_directories_waf_rate_adaptation(self, monkeypatch, tmp_path) -> None:
"""
Integration-style test for scan_directories WAF detection and rate adaptation.
We monkeypatch get_rate_limiter and get_runner to inject fakes, then assert:
- ffuf command contains expected -t (threads) and -p (delay) based on probe_analysis
- the returned scan_directories result contains WAF-related metadata
"""
import kambo.tools.scanning as scanning_module
captured_commands = []
class FakeRunner:
def __init__(self, probe_output: str, scan_output: str) -> None:
self.probe_output = probe_output
self.scan_output = scan_output
async def run(self, cmd, *args, **kwargs):
# record the command so we can assert on -t and -p
captured_commands.append(list(cmd))
# emulate the structure scan_directories expects from the runner
return {
"probe_result": self.probe_output,
"scan_result": self.scan_output,
}
class FakeRateLimiter:
def __init__(self, probe_analysis):
self.probe_analysis = probe_analysis
async def analyze_probe(self, *_args, **_kwargs):
# emulate whatever scan_directories uses from the rate limiter
return self.probe_analysis
# Different WAF scenarios to exercise:
scenarios = [
{
"probe_analysis": {"waf_detected": "Cloudflare", "threads": 5, "delay": 0.3},
"expected_waf_name": "Cloudflare",
"expected_rate_adapted": True,
},
{
"probe_analysis": {"is_blocked": True, "threads": 1, "delay": 1.0},
"expected_waf_name": None,
"expected_rate_adapted": True,
},
{
"probe_analysis": {"threads": 25, "delay": 0.0},
"expected_waf_name": None,
"expected_rate_adapted": False,
},
]
async def run_scenario(scenario):
captured_commands.clear()
def fake_get_rate_limiter(*_args, **_kwargs):
# scan_directories should call this factory to get a rate limiter instance
return FakeRateLimiter(probe_analysis=scenario["probe_analysis"])
def fake_get_runner(*_args, **_kwargs):
# scan_directories should call this factory to get a runner
return FakeRunner(
probe_output="fake_probe_output",
scan_output="fake_scan_output",
)
monkeypatch.setattr(scanning_module, "get_rate_limiter", fake_get_rate_limiter)
monkeypatch.setattr(scanning_module, "get_runner", fake_get_runner)
# Minimal realistic call into scan_directories
# (tmp_path provides a concrete wordlist path)
wordlist = tmp_path / "words.txt"
wordlist.write_text("admin\nlogin\n")
result = await scanning_module.scan_directories(
url="https://example.com",
wordlist_path=str(wordlist),
)
# 1) Assert we actually invoked ffuf with a command that includes -t and -p
assert captured_commands, "Runner was not invoked by scan_directories"
ffuf_cmd = captured_commands[-1]
assert "-t" in ffuf_cmd, "ffuf command missing -t (threads) flag"
assert "-p" in ffuf_cmd, "ffuf command missing -p (delay) flag"
t_index = ffuf_cmd.index("-t") + 1
p_index = ffuf_cmd.index("-p") + 1
# Note: we only assert presence/consistency, not exact numeric values,
# because the mapping logic belongs to scan_directories.
assert ffuf_cmd[t_index], "threads value after -t must not be empty"
assert ffuf_cmd[p_index], "delay value after -p must not be empty"
# 2) Assert WAF / rate-limiter metadata is surfaced in the result
assert "waf_detected" in result
assert "waf_name" in result
assert "rate_adapted" in result
assert "evasion_tips" in result
expected_waf_name = scenario["expected_waf_name"]
if expected_waf_name:
assert result["waf_detected"] is True
assert result["waf_name"] == expected_waf_name
else:
# for non-Cloudflare scenarios we just assert the key exists;
# the implementation may use None/False/"" to indicate "no WAF"
assert result["waf_detected"] in (False, None)
if scenario["expected_rate_adapted"]:
assert result["rate_adapted"] is True
assert isinstance(result["evasion_tips"], (list, tuple))
else:
assert result["rate_adapted"] in (False, None)
# Run all scenarios
for scenario in scenarios:
await run_scenario(scenario)The fake FakeRunner and FakeRateLimiter in this test assume:
scan_directoriesimportsget_runnerandget_rate_limiterfromkambo.tools.scanningand calls them as factories.- The object returned by
get_runnerexposes an asyncrun(cmd, *args, **kwargs)that returns a dict with"probe_result"and"scan_result"keys. - The object returned by
get_rate_limiterexposes an asyncanalyze_probe(...)method that returns theprobe_analysisdict used to derive threads/delay and WAF metadata. scan_directoriesis an async function taking at leasturlandwordlist_pathparameters and returning a dict with"waf_detected","waf_name","rate_adapted", and"evasion_tips"keys.
If your actual scan_directories / rate-limiter / runner APIs differ, you should:
- Adjust
FakeRunner.run’s signature and returned structure to match whatscan_directoriesexpects. - Adjust
FakeRateLimiter’s method name and signature to align with how the real rate limiter is used (e.g.,analyze,analyze_result, etc.). - Update the monkeypatch targets (
scanning_module.get_runner,scanning_module.get_rate_limiter) if those helpers live under different names or modules. - If
scan_directoriesis synchronous, removeasync/awaitusage andpytest.mark.asyncio(if present around the test) and call it directly instead.
…n tests
Comment 1 — validate_graphql: fix wrong '__types' introspection signature
The pattern r'"__types"\s*:' never matched real GraphQL responses.
Real introspection responses expose types under __schema.types as:
{"data": {"__schema": {"types": [...]}}}
Fixed to r'"types"\s*:\s*\[' which reliably matches the types
array present in every introspection response.
Comment 2 — scan_directories: replace import-only checks with monkeypatched
integration tests that exercise the actual adaptive behavior:
- _fake_runner: stubs runner.run() — first call returns probe, second scan
- _fake_limiter: stubs record_request() to return controlled probe_analysis
- test_waf_detected_reduces_threads_to_2: Cloudflare → -t 2 -p 0.5
- test_blocked_without_waf_reduces_threads_to_3: blocked → -t 3 -p 0.2
- test_no_waf_uses_full_threads: clean → -t 10, no -p flag
- test_result_always_contains_waf_metadata_keys: dict shape contract
- test_probe_runs_before_ffuf: curl probe fires first, ffuf second
614 tests passing.
https://claude.ai/code/session_01QrWLmqz6sz4tMVvDWHJqrU
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Implements six improvements identified in the refine pass:
</script>tags no longer incorrectly trigger the "inside script block" context checkscan_directoriesnow probes for WAF presence and adapts rate limiting accordinglyKey Changes
src/kambo/models.pyconfidence_pctproperty to implement proper tier mapping:src/kambo/validation.pyvalidate_xss: Fixed script/comment context detection by counting open/close tags instead of simple substring checks</script>tags no longer block XSS detection<!--is actually closed by-->validate_ssrf: Added 307/308 to accepted redirect status codes (permanent/temporary redirects that indicate server-side fetch)validate_graphql(new): Three-vector GraphQL security validator__schema,__types, directivesvalidate_race_condition(new): TOCTOU/race condition analyzersrc/kambo/tools/scanning.pyscan_directories: Integrated WAF detection and adaptive rate limitingget_rate_limiter()to analyze probe response for WAF signaturestests/test_refine_improvements.py(new)tests/test_models.pyImplementation Details
https://claude.ai/code/session_01QrWLmqz6sz4tMVvDWHJqrU
Summary by Sourcery
Refine confidence scoring, enhance existing XSS/SSRF validators, add new GraphQL and race condition validators, and integrate WAF-aware rate limiting into directory scanning, with tests covering all changes.
New Features:
Bug Fixes:
Enhancements:
Tests: