From 03f370f2ae296e5ecde59674c8c6f5828da31bd1 Mon Sep 17 00:00:00 2001 From: Matt Suiche Date: Thu, 22 Oct 2026 09:27:51 -0700 Subject: [PATCH 1/2] feat: add lightweight differential WAF check mode Add small control and exploit-shaped multipart requests to assess selective Cloudflare blocking without running memory-pressure probes. Keep WAF results separate from vulnerability status and support existing CLI, path, threading, and JSON options. Adapt the WAF-check technique and multipart layout from Matt Suiche's php2shell_waf_check. Add local HTTP integration tests, verdict tests, and usage documentation. --- README.md | 49 ++++++ scanner.py | 186 +++++++++++++++++--- tests/test_waf.py | 439 ++++++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 653 insertions(+), 21 deletions(-) create mode 100644 tests/test_waf.py diff --git a/README.md b/README.md index 279b75d..8ca4a50 100644 --- a/README.md +++ b/README.md @@ -18,6 +18,44 @@ The default `--sapi auto` mode applies both mod_php and PHP-FPM verdict rules, m Each tested path may receive a 42-byte control request and three probes totalling about 29 MiB. Probing stops as soon as the verdict is settled, and duplicate input URLs are scanned once. +### Lightweight WAF check + +Use `--waf-check` to check edge mitigation with small requests instead of the +memory-pressure probes: + +``` +python3 scanner.py -u https://example.com/upload.php --waf-check +python3 scanner.py -l hosts.txt --waf-check -o waf-results.json +``` + +Each tested path receives a benign multipart control request followed by an +approximately 9 KB shaped request containing `PHP_SESSION_UPLOAD_PROGRESS`, +three `t[]` fields, and two 4 KB file parts. Both requests use the same cookie +context; a custom `Cookie` header is preserved. No large probes are built or sent +in this mode. Control redirects and connection failures stop the check early. + +| WAF verdict | Meaning | +|---|---| +| `BLOCKED` | Control returned 2xx or 404; shaped request returned 403 with a Cloudflare block-page marker. | +| `NOT BLOCKED` | Control and shaped request returned 2xx or 404; no blocking was observed for this layout. | +| `INCONCLUSIVE` | The control was rejected or failed at the server, or the shaped response did not establish Cloudflare blocking. Includes non-Cloudflare 403 responses. | +| `REDIRECT` | A request redirected. Retest the intended endpoint; redirects are not followed. | +| `ERROR` | A request failed or the target URL was invalid. | + +These results describe only the tested request shape. A Cloudflare block page +does not identify the rule that fired or prove protection against other payloads. +An accepted request does not prove that PHP handled it. This mode does not assess +PHP patch status or exploitability; verify installed versions separately. + +The existing URL, path, header, timeout, TLS, and threading options apply. +`--sapi` applies only to the default vulnerability scan. JSON output includes all +WAF results without requiring `--all-results`, with `mode: "waf"`, a nullable +`waf_blocked` field, and `vulnerable: null`. With multiple paths, a host is reported +blocked only if every tested path is blocked; an observed `NOT BLOCKED` path takes +precedence over an inconclusive path. `--quiet` prints only checks that are not +confirmed blocked. Exit status is 0 only when every tested target/path is blocked, +1 otherwise, or 2 if the output file cannot be written. + ## Requirements - Python 3.9+ @@ -84,6 +122,7 @@ python3 scanner.py -u https://example.com --sapi fpm -v, --verbose Show verdict reasons and response snippets -q, --quiet Only output vulnerable hosts --no-color Disable colored output +--waf-check Check selective WAF blocking using only small multipart requests --sapi Verdict rules: auto, mod_php, or fpm (default: auto) --path Custom path to test (can be used multiple times) --path-file File containing paths to test (one per line) @@ -95,6 +134,10 @@ The first, original discovery of php2shell is attributed to the Calif team. You The scanner was built by the Assetnote Security Research Team using logic built for [Searchlight's Exposure Platform](https://www.slcyber.io/exposure). +The lightweight WAF-check technique and multipart layout are adapted from +[php2shell_waf_check](https://github.com/msuiche/php2shell_waf_check) by +Matt Suiche, Tolmo ([@msuiche](https://github.com/msuiche)). + ## Output Results are printed to the terminal. When using `-o`, vulnerable hosts are saved to a JSON file with the verdict reason and redacted HTTP request/response evidence. @@ -106,3 +149,9 @@ On specific PHP configurations, the scanner may report a false positive when its This was reproduced on patched PHP 8.4.27 with `memory_limit` values from **2M to 32M**, and on PHP 8.3.36 at **128M** when `max_multipart_body_parts` is set to **2785 or higher**. The exact threshold may vary between builds. When PHP does not disclose its version, the scanner cannot distinguish this safe post-patch OOM from the vulnerable pre-patch behavior. Results should therefore be verified against the installed PHP version and any vendor backports + +## Tests + +After installing the requirements, run `python3 -m unittest discover -s tests -v`. +The tests use mocked responses and a local HTTP server; they do not probe external +targets. diff --git a/scanner.py b/scanner.py index 041d14a..95ed5ed 100755 --- a/scanner.py +++ b/scanner.py @@ -59,6 +59,8 @@ ) PROBES = ((1, 4000), (8, 1020), (20, 1020)) +WAF_BOUNDARY = "php2shell-waf-check" +CF_BLOCK_MARKER = b"attention required! | cloudflare" REJECTED_STATUSES = {400, 403, 411, 413, 414, 417, 422, 429, 431} FIXED_VERSIONS = {(8, 2): 35, (8, 3): 36, (8, 4): 27, (8, 5): 12} VERSION_PATTERN = re.compile(r"PHP/(\d+)\.(\d+)\.(\d+)") @@ -757,6 +759,105 @@ def response_metadata(response: HTTPResult) -> dict: } +def build_waf_body(shaped: bool) -> bytes: + """Small multipart layout adapted from Matt Suiche's php2shell_waf_check.""" + fields = ( + [("PHP_SESSION_UPLOAD_PROGRESS", "probe")] + + [("t[]", letter * 32) for letter in "ABC"] + if shaped + else [("wafprobe", "benign")] + ) + chunks = [] + for name, value in fields: + chunks.append( + f'--{WAF_BOUNDARY}\r\nContent-Disposition: form-data; ' + f'name="{name}"\r\n\r\n{value}\r\n'.encode("ascii") + ) + if shaped: + for index, letter in enumerate((b"A", b"B"), start=1): + filename = "first.bin" if index == 1 else "second.bin" + chunks.append( + f'--{WAF_BOUNDARY}\r\nContent-Disposition: form-data; ' + f'name="file{index}"; filename="{filename}"\r\n' + "Content-Type: application/octet-stream\r\n\r\n".encode("ascii") + + letter * 4096 + + b"\r\n" + ) + chunks.append(f"--{WAF_BOUNDARY}--\r\n".encode("ascii")) + return b"".join(chunks) + + +def check_waf_url( + host: str, + target_url: str, + timeout: int, + verify_ssl: bool, + custom_headers: Dict[str, str], +) -> dict: + result = base_result(host) + result.update( + mode="waf", waf_blocked=None, tested_url=target_url, final_url=target_url + ) + headers = dict(custom_headers) + # Keep session context identical for the control and shaped requests. + if not has_header(headers, "Cookie"): + headers["Cookie"] = "PHPSESSID=wafprobe0001" + responses = [] + for shaped in (False, True): + body = build_waf_body(shaped) + response = send_post( + target_url, body, WAF_BOUNDARY, headers, timeout, verify_ssl + ) + responses.append(response) + result["request"] = request_dump( + target_url, + request_headers(headers, WAF_BOUNDARY), + len(body), + "small php2shell-shaped WAF probe" if shaped else "benign multipart control", + ) + apply_response(result, response) + if shaped: + result["probes"].append(response_metadata(response)) + else: + result["control"] = response_metadata(response) + if response.error or response.status_code is None: + result["error"] = response.error or "no HTTP status received" + result["reason"] = result["error"] + return result + if 300 <= response.status_code < 400: + result["verdict"] = "REDIRECT" + result["reason"] = "request redirected; rerun against the intended endpoint" + return result + + control, probe = responses + result["verdict"] = "INCONCLUSIVE" + if not (200 <= control.status_code < 300 or control.status_code == 404): + result["reason"] = ( + f"benign multipart returned HTTP {control.status_code}; " + "cannot establish selective blocking" + ) + elif probe.status_code == 403 and CF_BLOCK_MARKER in probe.body.lower(): + result["waf_blocked"] = True + result["verdict"] = "BLOCKED" + result["reason"] = ( + "exploit-shaped request selectively blocked with a Cloudflare block page; " + "this does not identify the rule or establish origin patch status" + ) + elif 200 <= probe.status_code < 300 or probe.status_code == 404: + result["waf_blocked"] = False + result["verdict"] = "NOT BLOCKED" + result["reason"] = ( + "no blocking observed for this exploit-shaped request; " + "this does not establish origin vulnerability or reachability" + ) + else: + result["reason"] = ( + f"shaped request returned HTTP {probe.status_code} without a confirmed " + "Cloudflare block; cannot establish WAF mitigation" + ) + return result + + def check_url( host: str, target_url: str, @@ -932,12 +1033,15 @@ def check_vulnerability( custom_headers: Optional[Dict[str, str]] = None, sapi: str = "auto", paths: Optional[List[str]] = None, + waf_check: bool = False, ) -> dict: display_host = safe_url_label(host) try: normalized = normalize_url(host) except ValueError as exc: result = base_result(display_host) + if waf_check: + result.update(mode="waf", waf_blocked=None) result["error"] = str(exc) result["reason"] = str(exc) return result @@ -947,14 +1051,18 @@ def check_vulnerability( ) fallback: Optional[dict] = None fallback_priority = {"NO SIGNAL": 0, "FIXED": 1, "INCONCLUSIVE": 2, "ERROR": 2} + if waf_check: + fallback_priority = { + "BLOCKED": 0, "NOT BLOCKED": 3, "INCONCLUSIVE": 2, + "ERROR": 2, "REDIRECT": 2, + } for target_url in targets: - result = check_url( - normalized, - target_url, - timeout, - verify_ssl, - custom_headers or {}, - sapi, + result = ( + check_waf_url(normalized, target_url, timeout, verify_ssl, custom_headers or {}) + if waf_check + else check_url( + normalized, target_url, timeout, verify_ssl, custom_headers or {}, sapi + ) ) if result["vulnerable"] is True: return result @@ -1028,7 +1136,10 @@ def save_results( def print_result(result: dict, verbose: bool = False) -> None: host = result["host"] verdict = result["verdict"] - if result["vulnerable"] is True: + if result.get("mode") == "waf": + color = Colors.GREEN if result.get("waf_blocked") is True else Colors.YELLOW + status = colorize(f"[WAF {verdict}]", color) + elif result["vulnerable"] is True: status = colorize("[VULNERABLE]", Colors.RED + Colors.BOLD) elif result["vulnerable"] is False: status = colorize("[NOT VULNERABLE]", Colors.GREEN) @@ -1090,6 +1201,11 @@ def build_parser() -> argparse.ArgumentParser: action="store_true", help="Save all results, not only vulnerable hosts", ) + parser.add_argument( + "--waf-check", + action="store_true", + help="Only send small multipart requests to check selective WAF blocking", + ) parser.add_argument( "-k", "--insecure", @@ -1111,7 +1227,9 @@ def build_parser() -> argparse.ArgumentParser: help="Show verdict reasons and response snippets", ) parser.add_argument( - "-q", "--quiet", action="store_true", help="Only print vulnerable hosts" + "-q", "--quiet", + action="store_true", + help="Only print vulnerable hosts (or WAF checks that are not confirmed blocked)", ) parser.add_argument( "--no-color", action="store_true", help="Disable colored output" @@ -1176,7 +1294,12 @@ def main() -> None: print(colorize(f"[*] Default path: {DEFAULT_PATH}", Colors.CYAN)) print(colorize(f"[*] Using {args.threads} thread(s)", Colors.CYAN)) print(colorize(f"[*] Timeout: {args.timeout}s", Colors.CYAN)) - print(colorize(f"[*] SAPI rules: {args.sapi}", Colors.CYAN)) + if args.waf_check: + print(colorize( + "[*] WAF check: small requests only; no patch assessment", Colors.CYAN + )) + else: + print(colorize(f"[*] SAPI rules: {args.sapi}", Colors.CYAN)) if args.insecure: print(colorize("[!] TLS verification disabled", Colors.YELLOW)) print() @@ -1195,21 +1318,31 @@ def scan(host: str) -> dict: custom_headers=custom_headers, sapi=args.sapi, paths=paths, + waf_check=args.waf_check, ) except Exception as exc: # Keep one bad target from aborting a scan. result = base_result(safe_url_label(host)) + if args.waf_check: + result.update(mode="waf", waf_blocked=None) result["error"] = f"unexpected error: {exc}" result["reason"] = result["error"] return result - if len(hosts) > 1: + def needs_attention(result: dict) -> bool: + return ( + result.get("waf_blocked") is not True + if args.waf_check + else result["vulnerable"] is True + ) + + if len(hosts) > 1 and not args.waf_check: for probe in PROBES: build_probe_body(*probe) if len(hosts) == 1: result = scan(hosts[0]) results.append(result) - if not args.quiet or result["vulnerable"] is True: + if not args.quiet or needs_attention(result): print_result(result, args.verbose) else: with ThreadPoolExecutor(max_workers=args.threads) as executor: @@ -1225,8 +1358,8 @@ def scan(host: str) -> dict: result = future.result() results.append(result) - should_print = result["vulnerable"] is True or ( - not args.quiet and args.verbose + should_print = needs_attention(result) or ( + not args.quiet and (args.waf_check or args.verbose) ) if should_print: if args.quiet: @@ -1244,12 +1377,21 @@ def scan(host: str) -> dict: print(colorize("SCAN SUMMARY", Colors.BOLD)) print(colorize("=" * 60, Colors.CYAN)) print(f" Total hosts scanned: {len(hosts)}") - vulnerable_label = f"Vulnerable: {vulnerable_count}" - if vulnerable_count: - vulnerable_label = colorize(vulnerable_label, Colors.RED + Colors.BOLD) - print(f" {vulnerable_label}") - print(f" Not vulnerable: {len(results) - vulnerable_count - error_count}") - print(f" Inconclusive/errors: {error_count}") + if args.waf_check: + blocked = sum(result.get("waf_blocked") is True for result in results) + not_blocked = sum(result.get("waf_blocked") is False for result in results) + print(f" Exploit-shaped request blocked: {blocked}") + print(f" Not blocked: {not_blocked}") + print( + f" Inconclusive/redirects/errors: {len(results) - blocked - not_blocked}" + ) + else: + vulnerable_label = f"Vulnerable: {vulnerable_count}" + if vulnerable_count: + vulnerable_label = colorize(vulnerable_label, Colors.RED + Colors.BOLD) + print(f" {vulnerable_label}") + print(f" Not vulnerable: {len(results) - vulnerable_count - error_count}") + print(f" Inconclusive/errors: {error_count}") print(colorize("=" * 60, Colors.CYAN)) output_saved = True @@ -1257,11 +1399,13 @@ def scan(host: str) -> dict: output_saved = save_results( results, args.output, - vulnerable_only=not args.all_results, + vulnerable_only=not (args.all_results or args.waf_check), announce=not args.quiet, ) if not output_saved: raise SystemExit(2) + if args.waf_check: + raise SystemExit(1 if any(needs_attention(result) for result in results) else 0) raise SystemExit(1 if vulnerable_count else 0) diff --git a/tests/test_waf.py b/tests/test_waf.py new file mode 100644 index 0000000..1c7da46 --- /dev/null +++ b/tests/test_waf.py @@ -0,0 +1,439 @@ +import contextlib +import io +import json +from email import policy +from email.parser import BytesParser +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path +import subprocess +import sys +import tempfile +import threading +import unittest +from unittest.mock import patch + +import scanner + + +URL = "https://example.test/upload.php" +CF_BLOCK = b"Attention Required! | Cloudflare" + + +def response(status=200, body=b"ok", **kwargs): + return scanner.HTTPResult(status_code=status, body=body, **kwargs) + + +class WafTests(unittest.TestCase): + def check(self, control, probe, **kwargs): + with patch.object(scanner, "send_post", side_effect=[control, probe]) as send: + result = scanner.check_waf_url(URL, URL, 5, True, kwargs.get("headers", {})) + self.assertIsNone(result["vulnerable"]) + self.assertEqual(result["mode"], "waf") + return result, send + + def test_payload_structure_and_size(self): + for shaped in (False, True): + body = scanner.build_waf_body(shaped) + message = BytesParser(policy=policy.default).parsebytes( + f"Content-Type: multipart/form-data; boundary={scanner.WAF_BOUNDARY}\r\n" + "MIME-Version: 1.0\r\n\r\n".encode() + + body + ) + parts = list(message.iter_parts()) + self.assertFalse(message.defects) + self.assertEqual(len(parts), 6 if shaped else 1) + self.assertTrue(body.endswith(f"--{scanner.WAF_BOUNDARY}--\r\n".encode())) + self.assertLess(len(body), 10 * 1024) + names = [ + part.get_param("name", header="content-disposition") for part in parts + ] + if shaped: + self.assertEqual( + names, + [ + "PHP_SESSION_UPLOAD_PROGRESS", + "t[]", + "t[]", + "t[]", + "file1", + "file2", + ], + ) + self.assertEqual( + [len(part.get_payload(decode=True)) for part in parts[-2:]], + [4096, 4096], + ) + self.assertEqual( + [part.get_filename() for part in parts[-2:]], + ["first.bin", "second.bin"], + ) + else: + self.assertEqual(names, ["wafprobe"]) + self.assertEqual(parts[0].get_payload(decode=True), b"benign") + + def test_selective_cloudflare_block(self): + result, send = self.check(response(), response(403, CF_BLOCK)) + self.assertEqual(result["verdict"], "BLOCKED") + self.assertIs(result["waf_blocked"], True) + self.assertEqual(send.call_count, 2) + self.assertEqual(len(result["probes"]), 1) + self.assertNotIn("wafprobe0001", result["request"]) + + def test_no_block_observed(self): + for status in (200, 201, 204, 404): + with self.subTest(status=status): + result, _ = self.check(response(404), response(status)) + self.assertEqual(result["verdict"], "NOT BLOCKED") + self.assertIs(result["waf_blocked"], False) + + def test_control_rejection_cannot_prove_selectivity(self): + for status in (401, 403, 405, 413, 429, 500, 502): + with self.subTest(status=status): + result, _ = self.check(response(status), response(403, CF_BLOCK)) + self.assertEqual(result["verdict"], "INCONCLUSIVE") + self.assertIsNone(result["waf_blocked"]) + + def test_ambiguous_probe_is_inconclusive(self): + for status in (401, 403, 405, 413, 429, 500, 502): + with self.subTest(status=status): + result, _ = self.check(response(), response(status)) + self.assertEqual(result["verdict"], "INCONCLUSIVE") + self.assertIsNone(result["waf_blocked"]) + + def test_marker_without_403_is_not_a_block(self): + result, _ = self.check(response(), response(200, CF_BLOCK)) + self.assertIs(result["waf_blocked"], False) + + def test_redirects_are_not_followed(self): + for shaped in (False, True): + with self.subTest(shaped=shaped): + result, send = self.check( + response() if shaped else response(302), response(307) + ) + self.assertEqual(result["verdict"], "REDIRECT") + self.assertEqual(send.call_count, 2 if shaped else 1) + + def test_network_errors_and_missing_status(self): + for shaped in (False, True): + for failure in ( + scanner.HTTPResult(error="request timed out"), + scanner.HTTPResult(), + ): + with self.subTest(shaped=shaped, error=failure.error): + result, send = self.check( + response() if shaped else failure, failure + ) + self.assertEqual(result["verdict"], "ERROR") + self.assertIsNone(result["waf_blocked"]) + self.assertEqual(send.call_count, 2 if shaped else 1) + + def test_matching_session_context_and_custom_headers(self): + for headers in ( + {}, + {"cookie": "PHPSESSID=custom", "Authorization": "Bearer secret"}, + ): + original = dict(headers) + result, send = self.check( + response(), response(403, CF_BLOCK), headers=headers + ) + first_headers, second_headers = [ + call.args[3] for call in send.call_args_list + ] + self.assertEqual(first_headers, second_headers) + self.assertEqual(headers, original) + if headers: + self.assertEqual(first_headers, headers) + self.assertNotIn("secret", result["request"]) + self.assertNotIn("PHPSESSID=custom", result["request"]) + else: + self.assertEqual(first_headers["Cookie"], "PHPSESSID=wafprobe0001") + + def test_path_aggregation_requires_every_path_blocked(self): + for verdicts, expected in ( + (["BLOCKED", "BLOCKED"], "BLOCKED"), + (["BLOCKED", "INCONCLUSIVE"], "INCONCLUSIVE"), + (["ERROR", "BLOCKED"], "ERROR"), + (["BLOCKED", "REDIRECT"], "REDIRECT"), + (["NOT BLOCKED", "BLOCKED"], "NOT BLOCKED"), + (["NOT BLOCKED", "INCONCLUSIVE"], "NOT BLOCKED"), + (["INCONCLUSIVE", "NOT BLOCKED"], "NOT BLOCKED"), + ): + with self.subTest(verdicts=verdicts): + results = [ + dict( + scanner.base_result(URL), + mode="waf", + verdict=v, + waf_blocked=v == "BLOCKED", + ) + for v in verdicts + ] + with patch.object( + scanner, "check_waf_url", side_effect=results + ) as check: + result = scanner.check_vulnerability( + URL, paths=["/a.php", "/b.php"], waf_check=True + ) + self.assertEqual(result["verdict"], expected) + self.assertEqual( + [call.args[1] for call in check.call_args_list], + ["https://example.test/a.php", "https://example.test/b.php"], + ) + + def test_invalid_url_stays_a_waf_error(self): + with patch.object(scanner, "send_post") as send: + result = scanner.check_vulnerability("ftp://example.test", waf_check=True) + self.assertEqual(result["mode"], "waf") + self.assertEqual(result["verdict"], "ERROR") + self.assertIsNone(result["waf_blocked"]) + send.assert_not_called() + + def test_default_mode_still_calls_original_scanner(self): + expected = scanner.base_result(URL) + with patch.object(scanner, "check_url", return_value=expected) as check: + with patch.object(scanner, "check_waf_url") as waf: + self.assertIs(scanner.check_vulnerability(URL), expected) + check.assert_called_once() + waf.assert_not_called() + + +class WafCliTests(unittest.TestCase): + @classmethod + def setUpClass(cls): + cls.requests = [] + requests = cls.requests + + class Handler(BaseHTTPRequestHandler): + def do_POST(self): + body = self.rfile.read(int(self.headers["Content-Length"])) + requests.append((self.path, dict(self.headers), body)) + shaped = b"PHP_SESSION_UPLOAD_PROGRESS" in body + status, content = 200, b"ok" + if self.path == "/blocked" and shaped: + status, content = 403, CF_BLOCK + elif self.path == "/blanket": + status, content = 403, CF_BLOCK + elif self.path == "/other" and shaped: + status, content = 403, b"Forbidden" + elif self.path == "/redirect": + status = 302 + self.send_response(status) + self.send_header("Content-Length", str(len(content))) + self.send_header("X-Powered-By", "PHP/8.4.27") + if status == 302: + self.send_header("Location", "/blocked") + self.end_headers() + self.wfile.write(content) + + def log_message(self, *args): + pass + + cls.server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) + cls.thread = threading.Thread(target=cls.server.serve_forever, daemon=True) + cls.thread.start() + cls.origin = f"http://127.0.0.1:{cls.server.server_port}" + + @classmethod + def tearDownClass(cls): + cls.server.shutdown() + cls.server.server_close() + cls.thread.join() + + def setUp(self): + self.requests.clear() + + def cli(self, *args): + return subprocess.run( + [sys.executable, str(Path(scanner.__file__)), "--no-color", *args], + capture_output=True, + text=True, + timeout=20, + ) + + def test_live_verdicts_exit_codes_and_json(self): + for path, verdict, code in ( + ("blocked", "BLOCKED", 0), + ("open", "NOT BLOCKED", 1), + ("blanket", "INCONCLUSIVE", 1), + ("other", "INCONCLUSIVE", 1), + ("redirect", "REDIRECT", 1), + ): + with self.subTest(path=path), tempfile.TemporaryDirectory() as directory: + output = Path(directory) / "results.json" + self.requests.clear() + completed = self.cli( + "--waf-check", "-u", f"{self.origin}/{path}", "-o", str(output) + ) + self.assertEqual(completed.returncode, code, completed.stderr) + self.assertIn(f"[WAF {verdict}]", completed.stdout) + self.assertNotIn("NOT VULNERABLE", completed.stdout) + result = json.loads(output.read_text())["results"][0] + self.assertIsNone(result["vulnerable"]) + self.assertEqual(result["verdict"], verdict) + self.assertEqual(len(self.requests), 1 if path == "redirect" else 2) + self.assertTrue(all(len(body) < 10240 for _, _, body in self.requests)) + self.assertTrue( + all( + headers["Cookie"] == "PHPSESSID=wafprobe0001" + for _, headers, _ in self.requests + ) + ) + + def test_quiet_mode_and_multihost_do_not_build_large_probes(self): + for paths, exit_code in ( + (["blocked", "blocked"], 0), + (["blocked", "open"], 1), + (["blocked", "blanket"], 1), + ): + with self.subTest(paths=paths), tempfile.TemporaryDirectory() as directory: + targets = Path(directory) / "hosts.txt" + targets.write_text("\n".join(f"{self.origin}/{path}" for path in paths)) + output = Path(directory) / "results.json" + argv = [ + "scanner.py", + "--waf-check", + "-q", + "-l", + str(targets), + "-o", + str(output), + ] + with ( + patch.object(sys, "argv", argv), + patch.object( + scanner, + "build_probe_body", + side_effect=AssertionError("large probe in WAF mode"), + ), + ): + with contextlib.redirect_stdout(io.StringIO()) as stdout: + with self.assertRaises(SystemExit) as raised: + scanner.main() + self.assertEqual(raised.exception.code, exit_code) + self.assertNotIn("[WAF BLOCKED]", stdout.getvalue()) + if exit_code: + self.assertIn("[WAF ", stdout.getvalue()) + self.assertEqual( + json.loads(output.read_text())["total_results"], len(set(paths)) + ) + + def test_legacy_mode_over_real_http(self): + completed = self.cli("-u", f"{self.origin}/legacy") + self.assertEqual(completed.returncode, 0, completed.stderr) + self.assertIn("[NOT VULNERABLE]", completed.stdout) + self.assertIn("FIXED", completed.stdout) + self.assertEqual(len(self.requests), 2) + self.assertGreater(len(self.requests[1][2]), 1_000_000) + + def test_custom_cookie_and_credentials_over_real_http(self): + with tempfile.TemporaryDirectory() as directory: + output = Path(directory) / "results.json" + completed = self.cli( + "--waf-check", + "-u", + f"{self.origin}/blocked", + "-v", + "-H", + "cookie: PHPSESSID=private", + "-H", + "Authorization: Bearer secret", + "-o", + str(output), + ) + evidence = output.read_text() + self.assertEqual(completed.returncode, 0, completed.stderr) + self.assertEqual(len(self.requests), 2) + for _, headers, _ in self.requests: + self.assertEqual(headers["cookie"], "PHPSESSID=private") + self.assertEqual(headers["Authorization"], "Bearer secret") + for secret in ("private", "Bearer secret"): + self.assertNotIn(secret, evidence) + self.assertNotIn(secret, completed.stdout) + + def test_waf_summary_and_path_file(self): + with tempfile.TemporaryDirectory() as directory: + paths = Path(directory) / "paths.txt" + paths.write_text("/blocked\n/open\n") + argv = [ + "scanner.py", + "--waf-check", + "-u", + self.origin, + "--path-file", + str(paths), + ] + with ( + patch.object(sys, "argv", argv), + contextlib.redirect_stdout(io.StringIO()) as stdout, + ): + with self.assertRaises(SystemExit) as raised: + scanner.main() + self.assertEqual(raised.exception.code, 1) + self.assertIn("Exploit-shaped request blocked: 0", stdout.getvalue()) + self.assertIn("Not blocked: 1", stdout.getvalue()) + self.assertNotIn("Not vulnerable:", stdout.getvalue()) + self.assertEqual( + [path for path, _, _ in self.requests], + ["/blocked", "/blocked", "/open", "/open"], + ) + + def test_threaded_mode_prints_blocked_hosts(self): + with tempfile.TemporaryDirectory() as directory: + hosts = Path(directory) / "hosts.txt" + hosts.write_text(f"{self.origin}/blocked\n{self.origin}/open\n") + completed = self.cli("--waf-check", "-l", str(hosts)) + self.assertEqual(completed.returncode, 1) + self.assertIn("[WAF BLOCKED]", completed.stdout) + self.assertIn("[WAF NOT BLOCKED]", completed.stdout) + + def test_unexpected_exception_preserves_mode(self): + argv = ["scanner.py", "--waf-check", "-u", self.origin] + with ( + patch.object(sys, "argv", argv), + patch.object( + scanner, + "check_vulnerability", + side_effect=RuntimeError("unexpected failure"), + ), + ): + with contextlib.redirect_stdout(io.StringIO()) as stdout: + with self.assertRaises(SystemExit) as raised: + scanner.main() + self.assertEqual(raised.exception.code, 1) + self.assertIn("[WAF ERROR]", stdout.getvalue()) + self.assertIn("Inconclusive/redirects/errors: 1", stdout.getvalue()) + + def test_legacy_summary_and_vulnerable_exit_code(self): + for vulnerable, verdict in ((True, "VULNERABLE"), (False, "FIXED")): + with self.subTest(verdict=verdict): + result = dict( + scanner.base_result(self.origin), + vulnerable=vulnerable, + verdict=verdict, + ) + with ( + patch.object(sys, "argv", ["scanner.py", "-u", self.origin]), + patch.object(scanner, "check_vulnerability", return_value=result), + ): + with contextlib.redirect_stdout(io.StringIO()) as stdout: + with self.assertRaises(SystemExit) as raised: + scanner.main() + self.assertEqual(raised.exception.code, 1 if vulnerable else 0) + self.assertIn(f"Vulnerable: {int(vulnerable)}", stdout.getvalue()) + + def test_quiet_single_host_and_output_failure(self): + blocked = self.cli("--waf-check", "-q", "-u", f"{self.origin}/blocked") + self.assertEqual(blocked.returncode, 0) + self.assertEqual(blocked.stdout, "") + opened = self.cli("--waf-check", "-q", "-u", f"{self.origin}/open") + self.assertEqual(opened.returncode, 1) + self.assertIn("[WAF NOT BLOCKED]", opened.stdout) + with tempfile.TemporaryDirectory() as directory: + failed = self.cli( + "--waf-check", "-u", f"{self.origin}/blocked", "-o", directory + ) + self.assertEqual(failed.returncode, 2) + + +if __name__ == "__main__": + unittest.main() From b5d095f18358a5e04bb8d845a8b2a6e0fb6c10b5 Mon Sep 17 00:00:00 2001 From: Matt Suiche Date: Thu, 22 Oct 2026 09:28:41 -0700 Subject: [PATCH 2/2] docs: clarify WAF quiet and JSON output behavior --- README.md | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/README.md b/README.md index 8ca4a50..9412467 100644 --- a/README.md +++ b/README.md @@ -120,7 +120,7 @@ python3 scanner.py -u https://example.com --sapi fpm -k, --insecure Disable TLS certificate verification -H, --header Custom header (can be used multiple times) -v, --verbose Show verdict reasons and response snippets --q, --quiet Only output vulnerable hosts +-q, --quiet Only output vulnerable hosts, or WAF checks not confirmed blocked --no-color Disable colored output --waf-check Check selective WAF blocking using only small multipart requests --sapi Verdict rules: auto, mod_php, or fpm (default: auto) @@ -140,7 +140,7 @@ Matt Suiche, Tolmo ([@msuiche](https://github.com/msuiche)). ## Output -Results are printed to the terminal. When using `-o`, vulnerable hosts are saved to a JSON file with the verdict reason and redacted HTTP request/response evidence. +Results are printed to the terminal. When using `-o`, vulnerable hosts are saved to a JSON file with the verdict reason and redacted HTTP request/response evidence. Use `--all-results` to save all vulnerability-scan results. In `--waf-check` mode, all WAF results are saved automatically. ## Limitations