diff --git a/README.md b/README.md index 1f6bbc4e1f..c04d8f9ae1 100644 --- a/README.md +++ b/README.md @@ -387,7 +387,7 @@ For details, see [Configuration](https://www.blacklanternsecurity.com/bbot/Stabl - [List of Modules](https://www.blacklanternsecurity.com/bbot/Stable/modules/list_of_modules) - [Nuclei](https://www.blacklanternsecurity.com/bbot/Stable/modules/nuclei) - [Custom YARA Rules](https://www.blacklanternsecurity.com/bbot/Stable/modules/custom_yara_rules) - - [Lightfuzz](https://www.blacklanternsecurity.com/bbot/Stable/modules/lightfuzz) + - [Lightfuzz (DAST)](https://www.blacklanternsecurity.com/bbot/Stable/modules/lightfuzz) - **Misc** - [Contribution](https://www.blacklanternsecurity.com/bbot/Stable/contribution) - [Release History](https://www.blacklanternsecurity.com/bbot/Stable/release_history) diff --git a/bbot/core/event/base.py b/bbot/core/event/base.py index 1f47c0d65a..a00fd59bf5 100644 --- a/bbot/core/event/base.py +++ b/bbot/core/event/base.py @@ -685,7 +685,7 @@ def parent(self, parent): self.web_spider_distance = getattr(parent, "web_spider_distance", 0) event_has_url = getattr(self, "parsed_url", None) is not None for t in parent.tags: - if t in ("affiliate",): + if t in ("affiliate", "from-lightfuzz"): self.add_tag(t) elif t.startswith("mutation-"): self.add_tag(t) @@ -1135,10 +1135,20 @@ def sanitize_data(self, data): class DictEvent(BaseEvent): + __slots__ = ["url_extension"] + def sanitize_data(self, data): url = data.get("url", "") if url: self.parsed_url = self.validators.validate_url_parsed(url) + # extract url_extension from any dict event with a URL + url_path = self.parsed_url.path + if url_path: + parsed_path_lower = str(url_path).lower() + extension = get_file_extension(parsed_path_lower) + if extension: + self.url_extension = extension + self.add_tag(f"extension-{extension}") return data def _data_load(self, data): @@ -1366,7 +1376,6 @@ class URL_UNVERIFIED(DictHostEvent): __slots__ = [ "web_spider_distance", - "url_extension", "num_redirects", ] @@ -1567,6 +1576,7 @@ def _minimize(self): self._data.pop("original_value", None) self._data.pop("additional_params", None) self._data.pop("assigned_cookies", None) + self._data.pop("same_param_values", None) @property def children(self): diff --git a/bbot/core/helpers/diff.py b/bbot/core/helpers/diff.py index 73aebfa137..9c6e36920b 100644 --- a/bbot/core/helpers/diff.py +++ b/bbot/core/helpers/diff.py @@ -21,6 +21,7 @@ def __init__( headers=None, cookies=None, timeout=10, + on_baseline_ready=None, ): self.parent_helper = parent_helper self.baseline_url = baseline_url @@ -33,6 +34,8 @@ def __init__( self.headers = headers self.cookies = cookies self.timeout = 10 + # Optional async callback fired once with baseline_1 after the baseline is established. + self.on_baseline_ready = on_baseline_ready @staticmethod def merge_dictionaries(headers1, headers2): @@ -130,6 +133,12 @@ async def _baseline(self): self.baseline_ignore_headers += [x.lower() for x in dynamic_headers] self._baselined = True + if self.on_baseline_ready is not None: + try: + await self.on_baseline_ready(baseline_1) + except Exception as e: + log.debug(f"on_baseline_ready callback raised: {e}") + def gen_cache_buster(self): return {self.parent_helper.rand_string(6): "1"} diff --git a/bbot/core/helpers/helper.py b/bbot/core/helpers/helper.py index 0d2a1dbb6b..26840d8211 100644 --- a/bbot/core/helpers/helper.py +++ b/bbot/core/helpers/helper.py @@ -157,6 +157,7 @@ def http_compare( data=None, json=None, timeout=10, + on_baseline_ready=None, ): return HttpCompare( url, @@ -169,6 +170,7 @@ def http_compare( method=method, data=data, json=json, + on_baseline_ready=on_baseline_ready, ) def temp_filename(self, extension=None): diff --git a/bbot/core/helpers/misc.py b/bbot/core/helpers/misc.py index 8ae4238ce7..d84c1d5df8 100644 --- a/bbot/core/helpers/misc.py +++ b/bbot/core/helpers/misc.py @@ -2754,6 +2754,12 @@ def get_waf_strings(): return [ "The requested URL was rejected", "This content has been blocked", + "You don't have permission to access ", + "The URL you requested has been blocked", + "Request unsuccessful. Incapsula incident", + "Access Denied - Sucuri Website Firewall", + "Attention Required! | Cloudflare", + "Microsoft-Azure-Application-Gateway", ] diff --git a/bbot/core/helpers/regexes.py b/bbot/core/helpers/regexes.py index f50ea2519c..ad3cf14780 100644 --- a/bbot/core/helpers/regexes.py +++ b/bbot/core/helpers/regexes.py @@ -157,7 +157,14 @@ re.DOTALL, ) post_form_regex_noaction = re.compile( - r"]*(?:\baction=[\"']?([^\s\"'<>]+)[\"']?)?[^>]*\bmethod=[\"']?[pP][oO][sS][tT][\"']?[^>]*>([\s\S]*?)<\/form>", + # Negative lookahead rejects forms that already carry action= (those are + # handled by post_form_regex / post_form_regex2 with the real action URL). + # Without it, the action group's `?` makes this regex over-eagerly match + # forms with action, capturing '' and emitting a phantom WEB_PARAMETER + # whose url falls back to event.url — racing the legitimate emission. + # The empty `()` keeps findall's tuple shape (form_action, form_content) + # consistent with the other form regexes consumed by GetForm.extract(). + r"]*\baction=)()[^>]*\bmethod=[\"']?[pP][oO][sS][tT][\"']?[^>]*>([\s\S]*?)<\/form>", re.DOTALL, ) generic_form_regex = re.compile( @@ -166,9 +173,15 @@ ) select_tag_regex = re.compile( - r"]+?name=[\"\']?([_\-\.\w]+)[\"\']?[^>]*>(?:\s*]*?value=[\"\']?([_\.\-\w]*)[\"\']?[^>]*>)?", + r"]+?\bname=[\"\']?([_\-\.\w]+)[\"\']?[^>]*>((?:\s*]*>(?:[^<]*(?:)?)?\s*)*)", re.IGNORECASE | re.DOTALL, ) +option_tag_regex = re.compile(r"]*)>", re.IGNORECASE) +option_value_regex = re.compile( + r"""\bvalue\s*=\s*(?:"([^"]*)"|'([^']*)'|([^\s>'"]+))""", + re.IGNORECASE, +) +option_selected_regex = re.compile(r"(?]*?\sname=[\"\']?([\-\._=+\/\w]+)[\"\']?[^>]*?\svalue=[\"\']?([:%\-\._=+\/\w]*)[\"\']?[^>]*?>" diff --git a/bbot/core/helpers/web/response_event.py b/bbot/core/helpers/web/response_event.py new file mode 100644 index 0000000000..16dafd4695 --- /dev/null +++ b/bbot/core/helpers/web/response_event.py @@ -0,0 +1,87 @@ +"""Build HTTP_RESPONSE event data dicts from blasthttp.Response objects. + +Materializes hash and cert_info eagerly, so only call when about to emit an +HTTP_RESPONSE event (otherwise blasthttp.Response stays lazy). +""" + +import re +from urllib.parse import urlparse + + +_TITLE_REGEX = re.compile(r"]*>(.*?)", re.IGNORECASE | re.DOTALL) + + +def response_to_event_dict(response, url_input, method="GET"): + """Convert a blasthttp.Response into the dict format HTTP_RESPONSE events expect. + + Args: + response: a blasthttp.Response (native 0.5+ object with .hash, .cert_info) + url_input: the original scan-target identifier (host:port or full URL) + method: HTTP method used to fetch the response + + Returns: + dict with keys url, input, status_code, method, path, host, raw_header, + header, content_type, content_length, title, body, location, hash. Adds + cert_info when the response carried a TLS certificate. + """ + parsed = urlparse(response.url) + path = parsed.path or "/" + + status_line = f"HTTP/1.1 {response.status} \r\n" + raw_header = f"{status_line}{response.raw_headers}\r\n\r\n" + + header_dict = {} + for k, v in response.headers.items(): + key = k.lower().replace("-", "_") + if key in header_dict: + header_dict[key] += f", {v}" + else: + header_dict[key] = v + + content_type = header_dict.get("content_type", "") + content_length = int(header_dict.get("content_length", len(response.body_bytes))) + location = header_dict.get("location", "") + + body = response.body + title = "" + title_match = _TITLE_REGEX.search(body) + if title_match: + title = title_match.group(1).strip() + + j = { + "url": response.url, + "input": url_input, + "status_code": response.status, + "method": method, + "path": path, + "host": parsed.hostname or "", + "raw_header": raw_header, + "header": header_dict, + "content_type": content_type, + "content_length": content_length, + "title": title, + "body": body, + "location": location, + "hash": { + "body_md5": response.hash.body_md5, + "body_mmh3": response.hash.body_mmh3, + "body_sha256": response.hash.body_sha256, + "header_md5": response.hash.header_md5, + "header_mmh3": response.hash.header_mmh3, + "header_sha256": response.hash.header_sha256, + }, + } + + ci = response.cert_info + if ci is not None: + j["cert_info"] = { + "common_name": ci.common_name, + "sans": ci.sans, + "emails": ci.emails, + "issuer": ci.issuer, + "not_before": ci.not_before, + "not_after": ci.not_after, + "fingerprint_sha256": ci.fingerprint_sha256, + } + + return j diff --git a/bbot/defaults.yml b/bbot/defaults.yml index 4eed9c9124..db93de5cdd 100644 --- a/bbot/defaults.yml +++ b/bbot/defaults.yml @@ -287,8 +287,23 @@ parameter_blacklist: - cf_clearance - _abck - bm_sz + - bm_sv - ak_bmsc - f5_cspm + # CSRF / Anti-Forgery tokens + - authenticity_token + - csrfmiddlewaretoken + - __RequestVerificationToken + - antiforgerytoken + - __csrf_magic + - _wpnonce + # ASP.NET session/identity cookies + - .ASPXANONYMOUS + - .ASPXAUTH + # PKCE (Proof Key for Code Exchange) + - code_verifier + - code_challenge + # Analytics - _ga - _gid - _gat diff --git a/bbot/modules/base.py b/bbot/modules/base.py index 50274cf936..f6d94d027e 100644 --- a/bbot/modules/base.py +++ b/bbot/modules/base.py @@ -868,13 +868,13 @@ def _event_precheck(self, event): if self.target_only and "target" not in event.tags: return False, "it did not meet target_only filter criteria" - # limit js URLs to modules that opt in to receive them - if (not self.accept_url_special) and event.type.startswith("URL"): + # limit events with special URL extensions (e.g. .js) to modules that opt in + if not self.accept_url_special: extension = getattr(event, "url_extension", "") if extension in self.scan.url_extension_special: return ( False, - f"it is a special URL (extension {extension}) but the module does not opt in to receive special URLs", + f"it has a special URL extension ({extension}) but the module does not opt in to receive special URLs", ) return True, "precheck succeeded" diff --git a/bbot/modules/bypass403.py b/bbot/modules/bypass403.py index 65cae080b4..4a4b0ec473 100644 --- a/bbot/modules/bypass403.py +++ b/bbot/modules/bypass403.py @@ -1,5 +1,6 @@ from bbot.errors import HttpCompareError from bbot.modules.base import BaseModule +from bbot.core.helpers.misc import get_waf_strings """ Port of https://github.com/iamj0ker/bypass-403/ and https://portswigger.net/bappstore/444407b96d9c4de0adb7aed89e826122 @@ -80,6 +81,10 @@ class bypass403(BaseModule): meta = {"description": "Check 403 pages for common bypasses", "created_date": "2022-07-05", "author": "@liquidsec"} in_scope_only = True + async def setup(self): + self.waf_yara_rules = self.helpers.yara.compile_strings(get_waf_strings(), nocase=True) + return True + async def do_checks(self, compare_helper, event, collapse_threshold): results = set() error_count = 0 @@ -105,10 +110,10 @@ async def do_checks(self, compare_helper, event, collapse_threshold): # In some cases WAFs will respond with a 200 code which causes a false positive if subject_response is not None: - for waf_string in self.helpers.get_waf_strings(): - if waf_string in subject_response.text: - self.debug("Rejecting result based on presence of WAF string") - return + waf_matches = await self.helpers.yara.match(self.waf_yara_rules, subject_response.text) + if waf_matches: + self.debug("Rejecting result based on presence of WAF string") + return if match is False: if str(subject_response.status_code)[0] != "4": diff --git a/bbot/modules/http.py b/bbot/modules/http.py index 4bbb04a275..5a3d6d866f 100644 --- a/bbot/modules/http.py +++ b/bbot/modules/http.py @@ -1,10 +1,10 @@ -import re from http.cookies import SimpleCookie from urllib.parse import urlparse import blasthttp from bbot.core.helpers.web.web import iter_batch_results +from bbot.core.helpers.web.response_event import response_to_event_dict from bbot.modules.base import BaseModule from bbot.core.config.models import BaseModuleConfig, Field @@ -100,75 +100,7 @@ def _build_headers(self): def _response_to_json(self, url_input, response): """Convert a blasthttp Response to a dict for HTTP_RESPONSE events.""" - parsed = urlparse(response.url) - path = parsed.path or "/" - - # Build raw_header string (required by HTTP_RESPONSE validation). - # blasthttp already builds the canonical "Name: Value\r\n..." form - # — reuse it instead of rebuilding. - status_line = f"HTTP/1.1 {response.status} \r\n" - raw_header = f"{status_line}{response.raw_headers}\r\n\r\n" - - # Build header dict (lowercase keys, comma-joined for dupes) - header_dict = {} - for k, v in response.headers.items(): - key = k.lower().replace("-", "_") - if key in header_dict: - header_dict[key] += f", {v}" - else: - header_dict[key] = v - - content_type = header_dict.get("content_type", "") - content_length = int(header_dict.get("content_length", len(response.body_bytes))) - - # Location header for redirects (excavate uses event.redirect_location) - location = header_dict.get("location", "") - - # Extract title from HTML - title = "" - body = response.body - title_match = re.search(r"]*>(.*?)", body, re.IGNORECASE | re.DOTALL) - if title_match: - title = title_match.group(1).strip() - - j = { - "url": response.url, - "input": url_input, - "status_code": response.status, - "method": "GET", - "path": path, - "host": parsed.hostname or "", - "raw_header": raw_header, - "header": header_dict, - "content_type": content_type, - "content_length": content_length, - "title": title, - "body": body, - "location": location, - "hash": { - "body_md5": response.hash.body_md5, - "body_mmh3": response.hash.body_mmh3, - "body_sha256": response.hash.body_sha256, - "header_md5": response.hash.header_md5, - "header_mmh3": response.hash.header_mmh3, - "header_sha256": response.hash.header_sha256, - }, - } - - # Include TLS certificate info when available (HTTPS responses) - ci = response.cert_info - if ci is not None: - j["cert_info"] = { - "common_name": ci.common_name, - "sans": ci.sans, - "emails": ci.emails, - "issuer": ci.issuer, - "not_before": ci.not_before, - "not_after": ci.not_after, - "fingerprint_sha256": ci.fingerprint_sha256, - } - - return j + return response_to_event_dict(response, url_input, method="GET") async def _process_result(self, result, parent_event): """Emit URL + HTTP_RESPONSE events for one batch result. Returns True if status was usable.""" diff --git a/bbot/modules/internal/excavate.py b/bbot/modules/internal/excavate.py index 68c5facdb1..c825267afe 100644 --- a/bbot/modules/internal/excavate.py +++ b/bbot/modules/internal/excavate.py @@ -41,6 +41,39 @@ def find_subclasses(obj, base_class): return subclasses +def _pick_select_value(options_html): + """Choose the best ' for i in range(option_count)) + return ( + "" + '
' + '' + '' + f'' + '' + "
" + "" + ) + + async def setup_after_prep(self, module_test): + body = self._build_giant_form_html(self.GIANT_OPTION_COUNT) + # Sanity check the fixture really is past the YARA cliff. + assert len(body) > 100_000, f"giant-form fixture not large enough: {len(body)} bytes" + module_test.set_expect_requests( + expect_args={"method": "GET", "uri": "/"}, + respond_args={"response_data": body, "status": 200, "headers": {"Content-Type": "text/html"}}, + ) + # Stash a start time so check() can do a coarse wall-clock assertion. + module_test._giantform_start = time.monotonic() + + def check(self, module_test, events): + elapsed = time.monotonic() - module_test._giantform_start + # Coarse upper bound: full scan including HTTP fetch should stay snappy on + # a normal dev box even with a 200 KB form. The spec calls for sub-200ms + # processing of the giant-form HTTP_RESPONSE itself; we allow more slack + # here because the wall-clock includes the rest of the scan harness. + assert elapsed < 30, f"giant-form scan took {elapsed:.1f}s — likely a pathological regex regression" + + web_params = {e.data.get("name") for e in events if e.type == "WEB_PARAMETER" and isinstance(e.data, dict)} + expected = {"provider_id", "csrf", "search_term"} + missing = expected - web_params + assert not missing, f"WEB_PARAMETERs missing for giant-form fields: {missing}. Got: {sorted(web_params)}" + + +# Same shape as the giant-form test but with a form body that exceeds +# max_form_bytes. The bound is a defensive cap: forms that don't fit are +# skipped entirely (no partial extraction). The test asserts excavate doesn't +# hang or OOM on a multi-hundred-KB form fixture. +class TestExcavateGiantFormExceedsMaxBytes(ModuleTestBase): + targets = ["http://127.0.0.1:8888/"] + modules_overrides = ["http", "excavate", "hunt"] + + # Constrain max_form_bytes so we don't have to generate a literal 2 MB + # fixture to prove the bound holds — 32 KB is well below the body size + # the next fixture generates (~240 KB). + config_overrides = {"modules": {"excavate": {"max_form_bytes": 32768}}} + + OPTION_COUNT = 6000 # ~240 KB form body — 7.5× past max_form_bytes + + @classmethod + def _build_giant_form_html(cls, option_count): + options = "".join(f'' for i in range(option_count)) + return ( + "" + '
' + '' + '' + f'' + '' + "
" + "" + ) + + async def setup_after_prep(self, module_test): + body = self._build_giant_form_html(self.OPTION_COUNT) + # Sanity check we built something that actually exceeds the configured bound. + assert len(body) > 32768 * 5, f"oversized fixture too small: {len(body)} bytes" + module_test.set_expect_requests( + expect_args={"method": "GET", "uri": "/"}, + respond_args={"response_data": body, "status": 200, "headers": {"Content-Type": "text/html"}}, + ) + module_test._oversized_start = time.monotonic() + + def check(self, module_test, events): + # Hang/OOM guard: a runaway extractor would push wall-clock well past + # this. A correctly-bounded extractor handles even a 240 KB form in seconds. + elapsed = time.monotonic() - module_test._oversized_start + assert elapsed < 60, f"oversized-form scan took {elapsed:.1f}s — bound likely not enforced" + + # No form-field WEB_PARAMETERs should be emitted for the oversized form + # — extraction regex anchors on `` which falls outside the bound. + # Other WEB_PARAMETERs (e.g. global `test` header from scan config) are fine. + form_fields = {"csrf", "search_term", "provider_id"} + emitted = {e.data.get("name") for e in events if e.type == "WEB_PARAMETER" and isinstance(e.data, dict)} + leaked = form_fields & emitted + assert not leaked, ( + f"oversized form's fields should be skipped entirely under the bound, but excavate emitted: {leaked}" + ) + + +# Verifies the opening-tag YARA regex fires regardless of attribute order. +# Both `
` and `` should +# extract the form's fields. +class TestExcavateFormAttributeOrder(ModuleTestBase): + targets = ["http://127.0.0.1:8888/"] + modules_overrides = ["http", "excavate", "hunt"] + + html = """ + + + +
+
+ +
+ + """ + + async def setup_after_prep(self, module_test): + module_test.set_expect_requests( + expect_args={"method": "GET", "uri": "/"}, + respond_args={"response_data": self.html, "status": 200, "headers": {"Content-Type": "text/html"}}, + ) + + def check(self, module_test, events): + names = {e.data.get("name") for e in events if e.type == "WEB_PARAMETER" and isinstance(e.data, dict)} + assert "action_first" in names, f"missing action-first form field; got {sorted(names)}" + assert "method_first" in names, f"missing method-first form field; got {sorted(names)}" diff --git a/bbot/test/test_step_2/module_tests/test_module_lightfuzz.py b/bbot/test/test_step_2/module_tests/test_module_lightfuzz.py index 9e0fa18d37..d610cc38a8 100644 --- a/bbot/test/test_step_2/module_tests/test_module_lightfuzz.py +++ b/bbot/test/test_step_2/module_tests/test_module_lightfuzz.py @@ -1,6 +1,8 @@ import json import re import base64 +from types import SimpleNamespace +from urllib.parse import urlparse, parse_qs from .base import ModuleTestBase, tempwordlist from werkzeug.wrappers import Response @@ -8,9 +10,51 @@ import xml.etree.ElementTree as ET +from bbot.core.helpers.url import add_get_params +from bbot.modules.lightfuzz.submodules.base import BaseLightfuzz + from .test_module_paramminer_headers import helper +def _make_base_lightfuzz(url): + event = SimpleNamespace(url=url, data={"name": "p"}) + lightfuzz = SimpleNamespace(helpers=SimpleNamespace(add_get_params=add_get_params)) + return BaseLightfuzz(lightfuzz, event) + + +def test_lightfuzz_build_query_string_no_existing_qs(): + bl = _make_base_lightfuzz("https://x.test/path") + assert bl.build_query_string("PROBE", "p") == "https://x.test/path?p=PROBE" + + +def test_lightfuzz_build_query_string_unrelated_existing_param(): + bl = _make_base_lightfuzz("https://x.test/path?init=true") + result = bl.build_query_string("PROBE", "p") + assert result.count("?") == 1 + assert parse_qs(urlparse(result).query) == {"init": ["true"], "p": ["PROBE"]} + + +def test_lightfuzz_build_query_string_probe_overrides_existing_same_param(): + bl = _make_base_lightfuzz("https://x.test/path?p=original&init=true") + result = bl.build_query_string("PROBE", "p") + assert result.count("?") == 1 + assert parse_qs(urlparse(result).query) == {"p": ["PROBE"], "init": ["true"]} + + +def test_lightfuzz_build_query_string_merges_additional_params(): + bl = _make_base_lightfuzz("https://x.test/path?init=true") + result = bl.build_query_string("PROBE", "p", additional_params={"unlock": "tok"}) + assert result.count("?") == 1 + assert parse_qs(urlparse(result).query) == {"init": ["true"], "p": ["PROBE"], "unlock": ["tok"]} + + +def test_lightfuzz_build_query_string_preserves_fragment(): + bl = _make_base_lightfuzz("https://x.test/path?init=true#frag") + result = bl.build_query_string("PROBE", "p") + assert result.count("?") == 1 + assert result.endswith("#frag") + + # Path Traversal single dot tolerance class Test_Lightfuzz_path_singledot(ModuleTestBase): targets = ["http://127.0.0.1:8888"] @@ -47,9 +91,20 @@ def request_handler(self, request): + """ + # Real path-traversal payload: a successful fetch of a different + # resource (200 + distinct body). This is what a vulnerable server + # actually returns — it's the signal the detector relies on. + traversed_block = """ + + + +SECRET_FROM_PARENT_DIR """ if value == "%2F.%2Fa%2F..%2Fdefault.jpg" or value == "default.jpg": return Response(block, status=200) + if value == "%2F..%2Fa%2F..%2Fdefault.jpg": + return Response(traversed_block, status=200) return Response("file not found", status=500) def check(self, module_test, events): @@ -71,6 +126,111 @@ def check(self, module_test, events): assert pathtraversal_finding_emitted, "Path Traversal single dot tolerance FINDING not emitted" +# Path Traversal single-dot tolerance, strict path resolver (Python/Go/Rust style) +# Simulates a server that opens files via a raw OS path walk rather than by +# string-normalizing first (as PHP include() does). Only the "simple" dot probes +# — `./X` and `/./X` — resolve cleanly; the `a/../X` variants fail on the missing +# `a/` component. Guards against regression of the simple-probe coverage path +# that was previously missed on non-PHP stacks. +class Test_Lightfuzz_path_singledot_strict(Test_Lightfuzz_path_singledot): + def request_handler(self, request): + qs = str(request.query_string.decode()) + if "filename=" in qs: + value = qs.split("=")[1] + if "&" in value: + value = value.split("&")[0] + + block = """ + + + + """ + # Real strict-resolver traversal: `../X` reads a different file + # successfully (200 + distinct body). The detector requires a + # successful fetch on the doubledot probe to flag. + traversed_block = """ + + + +SECRET_FROM_PARENT_DIR + """ + # Only exact-file or simple-dot-prefixed reads succeed. Any path + # containing `a/../` fails because `a/` does not exist, mirroring + # Python's open() / Go's os.Open behavior on the filesystem. + if value in ( + "default.jpg", + "./default.jpg", + "/./default.jpg", + "%2E%2Fdefault.jpg", + ): + return Response(block, status=200) + if value == "../default.jpg": + return Response(traversed_block, status=200) + return Response("file not found", status=500) + + def check(self, module_test, events): + web_parameter_emitted = False + simple_pathtraversal_finding_emitted = False + for e in events: + if e.type == "WEB_PARAMETER": + if "HTTP Extracted Parameter [filename]" in e.data["description"]: + web_parameter_emitted = True + + if e.type == "FINDING": + desc = e.data["description"] + if ( + "POSSIBLE Path Traversal" in desc + and "Parameter: [filename]" in desc + and "single-dot traversal tolerance (simple" in desc + ): + simple_pathtraversal_finding_emitted = True + + assert web_parameter_emitted, "WEB_PARAMETER was not emitted" + assert simple_pathtraversal_finding_emitted, ( + "Simple single-dot path traversal FINDING not emitted — strict-resolver detection regression." + ) + + +# Negative regression test for the JSF-style path-traversal FP: a server +# that normalizes `./X` (returns the canonical resource for the no-op +# prefix) but strictly REJECTS any `..` segment with a 4xx/empty body. +# That's the opposite of a vulnerability — no file from a different path +# is being delivered. The detector must not emit a finding. +class Test_Lightfuzz_path_singledot_rejection_fp(Test_Lightfuzz_path_singledot): + def request_handler(self, request): + qs = str(request.query_string.decode()) + if "filename=" in qs: + value = qs.split("=")[1] + if "&" in value: + value = value.split("&")[0] + + # Strict rejection of any `..` segment in any encoding (raw, + # url-encoded, double-url-encoded). Empty body — no resource + # exfiltrated. + decoded = unquote(unquote(value)) + if ".." in decoded: + return Response("", status=404) + + block = """ + + + + """ + # Server normalizes `./` (and url-encoded variants) to a no-op + # and returns the canonical resource — singledot tolerance only. + if "default.jpg" in decoded: + return Response(block, status=200) + return Response("", status=404) + + def check(self, module_test, events): + for e in events: + if e.type == "FINDING": + desc = e.data["description"] + assert "Possible Path Traversal" not in desc and "POSSIBLE Path Traversal" not in desc, ( + f"Path Traversal false positive emitted when server rejects `..`: {desc}" + ) + + # Path Traversal Absolute path class Test_Lightfuzz_path_absolute(Test_Lightfuzz_path_singledot): etc_passwd = """ @@ -170,6 +330,100 @@ def check(self, module_test, events): assert ssti_finding_emitted, "SSTI integer multiply FINDING not emitted" +# SSTI direct {{A*B}} probe — simulates a template engine (e.g. Jinja2 with +# StrictUndefined) that raises on `{{undefined_var}}` so the comma-collapse +# trick fails, but still evaluates an arithmetic expression like +# `{{1337*1337}}` because arithmetic has no undefined variables. +class Test_Lightfuzz_ssti_direct_jinja(Test_Lightfuzz_ssti_multiply): + def request_handler(self, request): + qs = str(request.query_string.decode()) + from urllib.parse import unquote as _unquote + + value = qs.split("data=")[1] if "data=" in qs else "" + if "&" in value: + value = value.split("&")[0] + decoded = _unquote(value) + # `{{z}}` raises in StrictUndefined engines — the comma-collapse + # probe produces a 500 instead of the expected `1,787,569`. + if "{{z}}" in decoded: + return Response("Template error: 'z' is undefined", status=500) + # `{{A*B}}` evaluates normally when A and B are literal ints. + if decoded.startswith("{{") and decoded.endswith("}}"): + inner = decoded[2:-2] + if "*" in inner: + a, b = inner.split("*") + try: + return Response(f"
{int(a) * int(b)}
", status=200) + except ValueError: + pass + # Anything else: baseline-ish echo, no template evaluation + return Response(f"
Hi, {decoded}
", status=200) + + def check(self, module_test, events): + ssti_finding = False + for e in events: + if e.type == "FINDING": + desc = e.data["description"] + if "Server-side Template Injection" in desc and "[{{1337*1337}}]" in desc: + ssti_finding = True + assert ssti_finding, "Direct {{1337*1337}} SSTI FINDING not emitted" + + +# SSTI baseline FP suppression — simulates an endpoint whose static content +# happens to include the canary product value (1787569 or 1,787,569). The +# baseline check should suppress all ssti findings on this endpoint. +class Test_Lightfuzz_ssti_baseline_fp(Test_Lightfuzz_ssti_multiply): + def request_handler(self, request): + # Every response contains the canary literal as static content. + # No template evaluation anywhere — pure coincidental number. + return Response( + "Total units sold: 1,787,569", + status=200, + ) + + def check(self, module_test, events): + ssti_finding = False + for e in events: + if e.type == "FINDING": + if "Server-side Template Injection" in e.data.get("description", ""): + ssti_finding = True + assert not ssti_finding, ( + "SSTI finding emitted on a page whose baseline already contains " + "the canary number — baseline-check guard failed." + ) + + +# SSTI Apache Velocity — `#set($x=A*B)$x` syntax. None of the pre-existing +# probes use this shape so a Velocity-backed endpoint would slip past. +class Test_Lightfuzz_ssti_velocity(Test_Lightfuzz_ssti_multiply): + def request_handler(self, request): + qs = str(request.query_string.decode()) + from urllib.parse import unquote as _unquote + + value = qs.split("data=")[1] if "data=" in qs else "" + if "&" in value: + value = value.split("&")[0] + decoded = _unquote(value) + # Recognize a Velocity `#set($x=A*B)$x` probe and render the product. + m = re.match(r"#set\(\$\w+=(\d+)\*(\d+)\)\$\w+", decoded) + if m: + return Response( + f"
{int(m.group(1)) * int(m.group(2))}
", + status=200, + ) + # Anything else: plain echo with no template evaluation + return Response(f"
Hi, {decoded}
", status=200) + + def check(self, module_test, events): + velocity_finding = False + for e in events: + if e.type == "FINDING": + desc = e.data["description"] + if "Server-side Template Injection" in desc and "%23set" in desc: + velocity_finding = True + assert velocity_finding, "Velocity SSTI FINDING not emitted" + + # Between Tags XSS Detection class Test_Lightfuzz_xss(ModuleTestBase): targets = ["http://127.0.0.1:8888"] @@ -208,7 +462,9 @@ def request_handler(self, request): return Response(parameter_block, status=200) async def setup_after_prep(self, module_test): - module_test.scan.modules["lightfuzz"].helpers.rand_string = lambda *args, **kwargs: "AAAAAAAAAAAAAA" + module_test.scan.modules["lightfuzz"].helpers.rand_string = lambda *args, **kwargs: ( + "1234567890" if kwargs.get("numeric_only") else "AAAAAAAAAAAAAA" + ) expect_args = re.compile("/") module_test.set_expect_requests_handler(expect_args=expect_args, request_handler=self.request_handler) @@ -228,6 +484,241 @@ def check(self, module_test, events): assert xss_finding_emitted, "Between Tags XSS FINDING not emitted" +# Tag Attribute XSS with single-quoted value. Verifies that the attribute- +# context probes trigger against `` — which was +# previously missed because the regex only matched double-quoted attributes +# and the breakout probe hardcoded `"` as the breakout char. +class Test_Lightfuzz_xss_single_quote_attribute(Test_Lightfuzz_xss): + def request_handler(self, request): + qs = str(request.query_string.decode()) + parameter_block = """ + + Link + + """ + if "foo=" in qs: + value = qs.split("=")[1] + if "&" in value: + value = value.split("&")[0] + # Single-quoted attribute; only `<` and `>` are stripped so the + # attacker can still close the quote and inject event handlers. + safe = unquote(value).replace("<", "").replace(">", "") + xss_block = f""" +
+
hi
+
+
+ """ + return Response(xss_block, status=200) + return Response(parameter_block, status=200) + + async def setup_after_prep(self, module_test): + module_test.scan.modules["lightfuzz"].helpers.rand_string = lambda *args, **kwargs: ( + "1234567890" if kwargs.get("numeric_only") else "AAAAAAAAAAAAAA" + ) + expect_args = re.compile("/") + module_test.set_expect_requests_handler(expect_args=expect_args, request_handler=self.request_handler) + expect_args = re.compile("/otherpage.php") + module_test.set_expect_requests_handler(expect_args=expect_args, request_handler=self.request_handler) + + def check(self, module_test, events): + web_parameter_emitted = False + single_quote_finding = False + for e in events: + if e.type == "WEB_PARAMETER": + if "HTTP Extracted Parameter [foo]" in e.data["description"]: + web_parameter_emitted = True + if e.type == "FINDING": + desc = e.data["description"] + if ( + "Possible Reflected XSS" in desc + and "Parameter: [foo]" in desc + and "Tag Attribute (' quoted)" in desc + ): + single_quote_finding = True + + assert web_parameter_emitted, "WEB_PARAMETER was not emitted" + assert single_quote_finding, "Single-quote attribute XSS FINDING not emitted" + + +# HTML-comment-context XSS. Verifies that a reflection inside `` +# is detected when the `-->` sequence survives reflection (attacker can +# close the comment and inject markup afterward). +class Test_Lightfuzz_xss_html_comment(Test_Lightfuzz_xss): + def request_handler(self, request): + qs = str(request.query_string.decode()) + parameter_block = """ + + Link + + """ + if "foo=" in qs: + value = qs.split("=")[1] + if "&" in value: + value = value.split("&")[0] + decoded = unquote(value) + return Response(f"", status=200) + return Response(parameter_block, status=200) + + async def setup_after_prep(self, module_test): + module_test.scan.modules["lightfuzz"].helpers.rand_string = lambda *args, **kwargs: ( + "1234567890" if kwargs.get("numeric_only") else "AAAAAAAAAAAAAA" + ) + expect_args = re.compile("/") + module_test.set_expect_requests_handler(expect_args=expect_args, request_handler=self.request_handler) + expect_args = re.compile("/otherpage.php") + module_test.set_expect_requests_handler(expect_args=expect_args, request_handler=self.request_handler) + + def check(self, module_test, events): + html_comment_finding = False + for e in events: + if e.type == "FINDING": + desc = e.data["description"] + if "Possible Reflected XSS" in desc and "Context: [HTML Comment]" in desc: + html_comment_finding = True + assert html_comment_finding, "HTML Comment XSS FINDING not emitted" + + +# JS-template-literal-context XSS. Verifies detection of reflection inside +# a backtick-wrapped string in a closure, but + # leave `${`, `}`, and backtick alone. + safe = decoded.replace("<", "").replace(">", "") + return Response( + f"", + status=200, + ) + return Response(parameter_block, status=200) + + async def setup_after_prep(self, module_test): + module_test.scan.modules["lightfuzz"].helpers.rand_string = lambda *args, **kwargs: ( + "1234567890" if kwargs.get("numeric_only") else "AAAAAAAAAAAAAA" + ) + expect_args = re.compile("/") + module_test.set_expect_requests_handler(expect_args=expect_args, request_handler=self.request_handler) + expect_args = re.compile("/otherpage.php") + module_test.set_expect_requests_handler(expect_args=expect_args, request_handler=self.request_handler) + + def check(self, module_test, events): + backtick_finding = False + for e in events: + if e.type == "FINDING": + desc = e.data["description"] + if "Possible Reflected XSS" in desc and "Context: [JS Template Literal]" in desc: + backtick_finding = True + assert backtick_finding, "JS Template Literal XSS FINDING not emitted" + + +# Negative test: param echoed in BOTH an HTML comment (with `-->` HTML-encoded, +# so no breakout there) AND in plain body text (verbatim). determine_context +# fires on the comment reflection, but the breakout match lands only in the +# body — outside any comment. _verify_match_context must reject this so no +# HTML Comment finding fires. +class Test_Lightfuzz_xss_html_comment_fp(Test_Lightfuzz_xss): + def request_handler(self, request): + qs = str(request.query_string.decode()) + parameter_block = """ + + Link + + """ + if "foo=" in qs: + value = qs.split("=")[1] + if "&" in value: + value = value.split("&")[0] + decoded = unquote(value) + # Comment: HTML-encode `-->` so the breakout is defanged. + comment_safe = decoded.replace("-->", "-->") + # Body: reflect verbatim (the differential-escaping FP case). + return Response( + f"

You searched for {decoded}

", + status=200, + ) + return Response(parameter_block, status=200) + + async def setup_after_prep(self, module_test): + module_test.scan.modules["lightfuzz"].helpers.rand_string = lambda *args, **kwargs: ( + "1234567890" if kwargs.get("numeric_only") else "AAAAAAAAAAAAAA" + ) + expect_args = re.compile("/") + module_test.set_expect_requests_handler(expect_args=expect_args, request_handler=self.request_handler) + expect_args = re.compile("/otherpage.php") + module_test.set_expect_requests_handler(expect_args=expect_args, request_handler=self.request_handler) + + def check(self, module_test, events): + html_comment_findings = [ + e for e in events if e.type == "FINDING" and "Context: [HTML Comment]" in e.data.get("description", "") + ] + assert not html_comment_findings, ( + f"HTML Comment XSS FINDING falsely emitted despite match landing outside comment context: " + f"{[f.data.get('description') for f in html_comment_findings]}" + ) + + +# Negative test: param echoed in BOTH a backtick string inside " + f"

You searched for {decoded}

", + status=200, + ) + return Response(parameter_block, status=200) + + async def setup_after_prep(self, module_test): + module_test.scan.modules["lightfuzz"].helpers.rand_string = lambda *args, **kwargs: ( + "1234567890" if kwargs.get("numeric_only") else "AAAAAAAAAAAAAA" + ) + expect_args = re.compile("/") + module_test.set_expect_requests_handler(expect_args=expect_args, request_handler=self.request_handler) + expect_args = re.compile("/otherpage.php") + module_test.set_expect_requests_handler(expect_args=expect_args, request_handler=self.request_handler) + + def check(self, module_test, events): + backtick_findings = [ + e + for e in events + if e.type == "FINDING" and "Context: [JS Template Literal]" in e.data.get("description", "") + ] + assert not backtick_findings, ( + f"JS Template Literal XSS FINDING falsely emitted despite match landing outside backtick context: " + f"{[f.data.get('description') for f in backtick_findings]}" + ) + + # Form Action Injection Detection class Test_Lightfuzz_xss_formaction(Test_Lightfuzz_xss): def request_handler(self, request): @@ -268,14 +759,64 @@ def check(self, module_test, events): web_parameter_emitted = True if e.type == "FINDING": + desc = e.data["description"] if ( - "Possible Reflected XSS. Parameter: [func] Context: [Form Action Injection] Parameter Type: [POSTPARAM]" - in e.data["description"] + 'Possible Reflected XSS. Parameter: [func] Context: [URL-scheme Injection (" quoted)] Parameter Type: [POSTPARAM]' + in desc ): xss_finding_emitted = True assert web_parameter_emitted, "WEB_PARAMETER was not emitted" - assert xss_finding_emitted, "Form Action XSS FINDING not emitted" + assert xss_finding_emitted, "URL-scheme Injection XSS FINDING not emitted" + + +# Negative regression test for the FMCSA-style URL-scheme false positive: +# the parameter is reflected only into a non-URL-bearing attribute +# (``), with `<`, `>`, `"` HTML-encoded everywhere else. +# `javascript:RAND` survives into the value attribute, but no browser will +# navigate to it from there. The URL-scheme Injection probe must NOT fire. +class Test_Lightfuzz_xss_url_scheme_value_attr_fp(Test_Lightfuzz_xss): + def request_handler(self, request): + qs = str(request.query_string.decode()) + parameter_block = """ + + Link + + """ + if "Keyword=" in qs: + value = qs.split("=")[1] + if "&" in value: + value = value.split("&")[0] + decoded = unquote(value) + # Mimic the FMCSA pattern: `"`, `<`, `>` are HTML-encoded + # outside the value attribute, and the only place the raw + # token survives is inside ``. + safe = decoded.replace("&", "&").replace("<", "<").replace(">", ">").replace('"', """) + return Response( + f"" + f"{safe}" + f'' + f"", + status=200, + ) + return Response(parameter_block, status=200) + + async def setup_after_prep(self, module_test): + module_test.scan.modules["lightfuzz"].helpers.rand_string = lambda *args, **kwargs: ( + "1234567890" if kwargs.get("numeric_only") else "AAAAAAAAAAAAAA" + ) + expect_args = re.compile("/") + module_test.set_expect_requests_handler(expect_args=expect_args, request_handler=self.request_handler) + expect_args = re.compile("/otherpage.php") + module_test.set_expect_requests_handler(expect_args=expect_args, request_handler=self.request_handler) + + def check(self, module_test, events): + for e in events: + if e.type == "FINDING": + desc = e.data["description"] + assert "URL-scheme Injection" not in desc, ( + f"URL-scheme Injection false positive emitted for non-URL-bearing attribute: {desc}" + ) # Base64 Envelope XSS Detection @@ -517,7 +1058,9 @@ def request_handler(self, request): return Response(parameter_block, status=200) async def setup_after_prep(self, module_test): - module_test.scan.modules["lightfuzz"].helpers.rand_string = lambda *args, **kwargs: "AAAAAAAAAAAAAA" + module_test.scan.modules["lightfuzz"].helpers.rand_string = lambda *args, **kwargs: ( + "1234567890" if kwargs.get("numeric_only") else "AAAAAAAAAAAAAA" + ) expect_args = re.compile("/") module_test.set_expect_requests_handler(expect_args=expect_args, request_handler=self.request_handler) expect_args = re.compile("/otherpage.php") @@ -535,12 +1078,13 @@ def check(self, module_test, events): original_value_captured = True if e.type == "FINDING": - if "Possible Reflected XSS. Parameter: [foo] Context: [Tag Attribute]" in e.data["description"]: + desc = e.data["description"] + if "Possible Reflected XSS. Parameter: [foo] Context: [Tag Attribute" in desc and "quoted)" in desc: xss_finding_emitted = True assert web_parameter_emitted, "WEB_PARAMETER was not emitted" assert original_value_captured, "original_value not captured" - assert xss_finding_emitted, "Between Tags XSS FINDING not emitted" + assert xss_finding_emitted, "Tag Attribute XSS FINDING not emitted" # In Javascript XSS Detection @@ -576,7 +1120,9 @@ def request_handler(self, request): return Response(self.parameter_block, status=200) async def setup_after_prep(self, module_test): - module_test.scan.modules["lightfuzz"].helpers.rand_string = lambda *args, **kwargs: "AAAAAAAAAAAAAA" + module_test.scan.modules["lightfuzz"].helpers.rand_string = lambda *args, **kwargs: ( + "1234567890" if kwargs.get("numeric_only") else "AAAAAAAAAAAAAA" + ) expect_args = re.compile("/") module_test.set_expect_requests_handler(expect_args=expect_args, request_handler=self.request_handler) expect_args = re.compile("/otherpage.php") @@ -688,7 +1234,9 @@ def request_handler(self, request): return Response(parameter_block, status=200) async def setup_after_prep(self, module_test): - module_test.scan.modules["lightfuzz"].helpers.rand_string = lambda *args, **kwargs: "AAAAAAAAAAAAAA" + module_test.scan.modules["lightfuzz"].helpers.rand_string = lambda *args, **kwargs: ( + "1234567890" if kwargs.get("numeric_only") else "AAAAAAAAAAAAAA" + ) expect_args = re.compile("/") module_test.set_expect_requests_handler(expect_args=expect_args, request_handler=self.request_handler) @@ -756,7 +1304,9 @@ def request_handler(self, request): return Response(parameter_block, status=200) async def setup_after_prep(self, module_test): - module_test.scan.modules["lightfuzz"].helpers.rand_string = lambda *args, **kwargs: "AAAAAAAAAAAAAA" + module_test.scan.modules["lightfuzz"].helpers.rand_string = lambda *args, **kwargs: ( + "1234567890" if kwargs.get("numeric_only") else "AAAAAAAAAAAAAA" + ) expect_args = re.compile("/") module_test.set_expect_requests_handler(expect_args=expect_args, request_handler=self.request_handler) @@ -813,7 +1363,9 @@ def check(self, module_test, events): # SQLI Single Quote/Two Single Quote (headers) class Test_Lightfuzz_sqli_headers(Test_Lightfuzz_sqli): async def setup_after_prep(self, module_test): - module_test.scan.modules["lightfuzz"].helpers.rand_string = lambda *args, **kwargs: "AAAAAAAAAAAAAA" + module_test.scan.modules["lightfuzz"].helpers.rand_string = lambda *args, **kwargs: ( + "1234567890" if kwargs.get("numeric_only") else "AAAAAAAAAAAAAA" + ) expect_args = re.compile("/") module_test.set_expect_requests_handler(expect_args=expect_args, request_handler=self.request_handler) @@ -881,7 +1433,9 @@ def check(self, module_test, events): # SQLI Single Quote/Two Single Quote (cookies) class Test_Lightfuzz_sqli_cookies(Test_Lightfuzz_sqli): async def setup_after_prep(self, module_test): - module_test.scan.modules["lightfuzz"].helpers.rand_string = lambda *args, **kwargs: "AAAAAAAAAAAAAA" + module_test.scan.modules["lightfuzz"].helpers.rand_string = lambda *args, **kwargs: ( + "1234567890" if kwargs.get("numeric_only") else "AAAAAAAAAAAAAA" + ) expect_args = re.compile("/") module_test.set_expect_requests_handler(expect_args=expect_args, request_handler=self.request_handler) @@ -975,8 +1529,10 @@ def request_handler(self, request):
""" - if "' AND (SLEEP(5)) AND '" in unquote(value): - sleep(5) + decoded = unquote(value) + m = re.search(r"AND \(SLEEP\((\d+)\)\) AND", decoded) + if m: + sleep(int(m.group(1))) return Response(sql_block, status=200) return Response(parameter_block, status=200) @@ -989,9 +1545,11 @@ def check(self, module_test, events): web_parameter_emitted = True if e.type == "FINDING": + desc = e.data["description"] if ( - "Possible Blind SQL Injection. Parameter: [search] Parameter Type: [GETPARAM] Detection Method: [Delay Probe (1' AND (SLEEP(5)) AND ')]" - in e.data["description"] + "Possible Blind SQL Injection" in desc + and "Delay Probe" in desc + and "1' AND (SLEEP(8)) AND '" in desc ): sqldelay_finding_emitted = True @@ -999,6 +1557,123 @@ def check(self, module_test, events): assert sqldelay_finding_emitted, "SQLi Delay FINDING not emitted" +# Blind SQLi where only row-independent one-shot SLEEP payloads trigger a delay. +# Simulates the common real-world case where the injected value does not match +# any row in the target table, causing row-scoped AND-based SLEEP payloads to +# short-circuit and never execute. Also guards against regression of a missing +# list comma that previously fused Oracle and MSSQL probes into one malformed +# concatenated string. +class Test_Lightfuzz_sqli_delay_or_rowindependent(Test_Lightfuzz_sqli): + received_payloads = [] + + def request_handler(self, request): + from time import sleep + + qs = str(request.query_string.decode()) + parameter_block = """ +
+
+ + +
+
+ """ + if "search=" in qs: + value = qs.split("=")[1] + if "&" in value: + value = value.split("&")[0] + decoded = unquote(value) + self.__class__.received_payloads.append(decoded) + + sql_block = """ +
+

0 search results found

+
+
+ """ + # Only the one-shot row-independent MySQL payload triggers a delay. + # The original AND-based mysql probe does not fire here, simulating + # a context where the injected value does not match any row. + m = re.search(r"OR SLEEP\((\d+)\) IS NOT NULL", decoded) + if m: + sleep(int(m.group(1))) + return Response(sql_block, status=200) + return Response(parameter_block, status=200) + + def check(self, module_test, events): + web_parameter_emitted = False + one_shot_delay_finding = False + for e in events: + if e.type == "WEB_PARAMETER": + if "HTTP Extracted Parameter [search]" in e.data["description"]: + web_parameter_emitted = True + if e.type == "FINDING": + desc = e.data["description"] + if ( + "Possible Blind SQL Injection" in desc + and "Delay Probe" in desc + and "OR SLEEP(8) IS NOT NULL LIMIT 1-- -" in desc + ): + one_shot_delay_finding = True + + # Guard against regression of the missing-comma bug: Python string-literal + # concatenation of adjacent list entries would produce a payload containing + # both DBMS_LOCK.SLEEP and WAITFOR, which is never a valid single probe. + garbled = [p for p in self.received_payloads if "DBMS_LOCK.SLEEP" in p and "WAITFOR" in p] + assert not garbled, ( + f"Garbled Oracle+MSSQL concatenated probe was sent ({len(garbled)} times): " + f"{garbled[:1]}. This indicates the missing-comma bug has regressed." + ) + + assert web_parameter_emitted, "WEB_PARAMETER was not emitted" + assert one_shot_delay_finding, ( + "One-shot row-independent SLEEP finding not emitted - row-independent blind sqli detection regression." + ) + + +class Test_Lightfuzz_sqli_delay_jitter_fp(Test_Lightfuzz_sqli): + """Payload-independent jitter must not produce a blind SQLi finding.""" + + _jitter_idx = 0 + + def request_handler(self, request): + from time import sleep + + qs = str(request.query_string.decode()) + parameter_block = """ +
+
+ + +
+
+ """ + if "search=" in qs: + sql_block = """ +
+

0 search results found

+
+
+ """ + jitter = [0.1, 0.4, 0.15, 0.5, 0.2, 0.35, 0.45, 0.1, 0.3, 0.25] + idx = self.__class__._jitter_idx + sleep(jitter[idx % len(jitter)]) + self.__class__._jitter_idx = idx + 1 + return Response(sql_block, status=200) + return Response(parameter_block, status=200) + + async def setup_after_prep(self, module_test): + self.__class__._jitter_idx = 0 + await super().setup_after_prep(module_test) + + def check(self, module_test, events): + for e in events: + if e.type == "FINDING" and "SQL Injection" in e.data.get("description", ""): + raise AssertionError( + f"False positive: finding emitted under jitter-only conditions: {e.data['description']}" + ) + + # Serialization Module (Error Resolution) class Test_Lightfuzz_serial_errorresolution(ModuleTestBase): targets = ["http://127.0.0.1:8888"] @@ -1097,8 +1772,10 @@ def check(self, module_test, events): excavate_extracted_form_parameter_details = True if e.type == "FINDING": if ( - e.data["description"] - == "POSSIBLE Unsafe Deserialization. Parameter: [TextBox1] Parameter Type: [POSTPARAM] Technique: [Error Resolution (Baseline: [500] -> Probe: [200] )] Serialization Payload: [dotnet_base64]" + "POSSIBLE Unsafe Deserialization. Parameter: [TextBox1] Parameter Type: [POSTPARAM]" + in e.data["description"] + and "Technique: [Error Resolution (Baseline: [500] -> Probe: [200] )] Serialization Payload: [dotnet_base64]" + in e.data["description"] ): lightfuzz_serial_detect_errorresolution = True @@ -1200,8 +1877,10 @@ def check(self, module_test, events): if e.data["description"] == "HTTP response (body) contains a possible serialized object (DOTNET)": excavate_detect_serialization_value = True if ( - e.data["description"] - == "POSSIBLE Unsafe Deserialization. Parameter: [TextBox1] Parameter Type: [POSTPARAM] Original Value: [AAEAAAD/////AQAAAAAAAAAGAQAAAAdndXN0YXZvCw==] Technique: [Error Resolution (Baseline: [500] -> Probe: [200] )] Serialization Payload: [dotnet_base64]" + "POSSIBLE Unsafe Deserialization. Parameter: [TextBox1] Parameter Type: [POSTPARAM] Original Value: [AAEAAAD/////AQAAAAAAAAAGAQAAAAdndXN0YXZvCw==]" + in e.data["description"] + and "Technique: [Error Resolution (Baseline: [500] -> Probe: [200] )] Serialization Payload: [dotnet_base64]" + in e.data["description"] ): lightfuzz_serial_detect_errorresolution = True @@ -1385,6 +2064,174 @@ def check(self, module_test, events): ) +class Test_Lightfuzz_serial_errorresolution_nonstandard_status(Test_Lightfuzz_serial_errorresolution): + """A baseline with a non-standard status code (>511, e.g. GlobalProtect's 512) + should not produce an Error Resolution finding even if the probe returns 200.""" + + def request_handler(self, request): + dotnet_serial_error_resolved = ( + "Deserialization successful! Object type: System.String" + ) + post_params = request.form + + if "TextBox1" not in post_params.keys(): + return Response(self.dotnet_serial_html, status=200) + + else: + if post_params["__VIEWSTATE"] != "/wEPDwULLTE5MTI4MzkxNjVkZNt7ICM+GixNryV6ucx+srzhXlwP": + # Non-standard status code (like GlobalProtect's 512) + return Response(self.dotnet_serial_error, status=512) + if post_params["TextBox1"] == "AAEAAAD/////AQAAAAAAAAAGAQAAAAdndXN0YXZvCw==": + return Response(dotnet_serial_error_resolved, status=200) + else: + return Response(self.dotnet_serial_error, status=512) + + def check(self, module_test, events): + no_finding_emitted = True + for e in events: + if e.type == "FINDING" and "Error Resolution" in e.data.get("description", ""): + no_finding_emitted = False + assert no_finding_emitted, ( + "False positive Error Resolution finding was emitted for non-standard baseline status code (>511)" + ) + + +# Python pickle Error Resolution — verifies the new python_pickle_base64 +# payload flips a baseline UnpicklingError 500 into a 200, which is the +# canonical detection path for all languages in the serial submodule. +class Test_Lightfuzz_serial_python_pickle(Test_Lightfuzz_serial_errorresolution): + modules_overrides = ["http", "lightfuzz", "excavate", "paramminer_getparams"] + config_overrides = { + "interactsh_disable": True, + "modules": { + "lightfuzz": {"enabled_submodules": ["serial"]}, + # Make paramminer discover the pklparam on /deser_pickle. + "paramminer_getparams": {"wordlist": "", "recycle_words": False, "skip_boring_words": True}, + }, + } + + PICKLE_BENIGN_B64 = "gASVCAAAAAAAAACMBHRlc3SULg==" + + # Seed value is `aTowOw==` — base64 of `i:0;` (PHP-serialized integer). + # bbot's envelope system detects the outer B64 envelope and would + # normally re-pack any outgoing probe as base64 again — which would + # double-encode serial's already-base64 payloads and break detection. + # Because serial sets `skip_envelopes = True`, the bypass kicks in and + # the raw outer value is forwarded to `is_possibly_serialized` (which + # accepts base64-looking strings) and probes are sent verbatim. + SEED_HREF = 'pkl' + + async def setup_after_prep(self, module_test): + # Seed the parameter directly so the scan doesn't depend on + # paramminer discovering it. + expect_args = {"method": "GET", "uri": "/"} + respond_args = {"response_data": self.SEED_HREF, "status": 200} + module_test.set_expect_requests(expect_args=expect_args, respond_args=respond_args) + expect_args = re.compile("/deser_pickle") + module_test.set_expect_requests_handler(expect_args=expect_args, request_handler=self.request_handler) + + def request_handler(self, request): + value = request.args.get("pklparam", "") + if value == self.PICKLE_BENIGN_B64: + return Response("ok", status=200) + return Response("UnpicklingError: invalid pickle", status=500) + + def check(self, module_test, events): + pickle_finding = False + for e in events: + if e.type == "FINDING": + desc = e.data["description"] + if ( + "POSSIBLE Unsafe Deserialization" in desc + and "Serialization Payload: [python_pickle_base64]" in desc + ): + pickle_finding = True + assert pickle_finding, "Python pickle Error Resolution FINDING not emitted" + + +# Python pickle OOB (blind RCE) — verifies that the new pickle OOB payload +# triggers an interactsh interaction when deserialized, giving a CONFIRMED +# blind-RCE finding without needing response-body inspection. +class Test_Lightfuzz_serial_pickle_interactsh(Test_Lightfuzz_serial_python_pickle): + config_overrides = { + "interactsh_disable": False, + "modules": { + "lightfuzz": {"enabled_submodules": ["serial"]}, + }, + } + + async def setup_before_prep(self, module_test): + self.interactsh_mock_instance = module_test.mock_interactsh("lightfuzz") + + def mock_interactsh_factory(*args, **kwargs): + return self.interactsh_mock_instance + + from bbot.core.helpers.helper import ConfigAwareHelper + + module_test.monkeypatch.setattr(ConfigAwareHelper, "interactsh", mock_interactsh_factory) + + def request_handler(self, request): + import base64 as _b64 + import re as _re + + value = request.args.get("pklparam", "") + try: + decoded = _b64.b64decode(value) + except Exception: + decoded = b"" + # Look for the interactsh subdomain tag embedded in the pickle bytes. + # The subdomain is UTF-8 in the pickle stream, so scan the raw bytes. + match = _re.search(rb"([a-z]+)\.fakedomain\.fakeinteractsh\.com", decoded) + if match: + tag = match.group(1).decode("ascii") + self.interactsh_mock_instance.mock_interaction(tag) + return Response("ok", status=200) + + def check(self, module_test, events): + oob_finding = False + for e in events: + if e.type == "FINDING": + desc = e.data["description"] + if "Python pickle OOB RCE (OOB Interaction)" in desc and "Payload: [python_pickle_oob]" in desc: + oob_finding = True + assert oob_finding, "Python pickle OOB interactsh FINDING not emitted" + + +# Java URLDNS OOB — verifies that the pure-Python URLDNS payload built +# from scratch triggers an interactsh DNS lookup callback when a mock +# "Java" endpoint "deserializes" it. Because the URLDNS gadget requires +# only java.util.HashMap + java.net.URL (both stdlib), it fires against +# any Java deserialization sink without depending on app-specific gadget +# classes. +class Test_Lightfuzz_serial_urldns_interactsh(Test_Lightfuzz_serial_pickle_interactsh): + def request_handler(self, request): + import base64 as _b64 + import re as _re + + value = request.args.get("pklparam", "") + try: + decoded = _b64.b64decode(value) + except Exception: + decoded = b"" + # The URLDNS payload embeds the host string as a Java-modified-UTF + # with a 2-byte length prefix. The host bytes appear verbatim, so + # we can grep for the interactsh subdomain directly. + match = _re.search(rb"([a-z]+)\.fakedomain\.fakeinteractsh\.com", decoded) + if match: + tag = match.group(1).decode("ascii") + self.interactsh_mock_instance.mock_interaction(tag) + return Response("ok", status=200) + + def check(self, module_test, events): + urldns_finding = False + for e in events: + if e.type == "FINDING": + desc = e.data["description"] + if "Java URLDNS OOB (OOB Interaction)" in desc and "Payload: [java_urldns_oob]" in desc: + urldns_finding = True + assert urldns_finding, "Java URLDNS OOB interactsh FINDING not emitted" + + # CMDi echo canary class Test_Lightfuzz_cmdi(ModuleTestBase): targets = ["http://127.0.0.1:8888"] @@ -1413,13 +2260,20 @@ def request_handler(self, request): value = qs.split("=")[1] if "&" in value: value = value.split("&")[0] - if "&& echo " in unquote(value): - cmdi_value = unquote(value).split("&& echo ")[1].split(" ")[0] + decoded = unquote(value) + # Simulate a Linux bash-family shell: evaluate $((A*B)) first so + # the arithmetic confirmation probe lands a product value, then + # fall back to plain-echo reflection for the generic probe. + arith = re.search(r"&& echo \$\(\((\d+)\*(\d+)\)\) &&", decoded) + if arith: + cmdi_value = str(int(arith.group(1)) * int(arith.group(2))) + elif "&& echo " in decoded: + cmdi_value = decoded.split("&& echo ")[1].split(" ")[0] else: - cmdi_value = value + cmdi_value = decoded cmdi_block = f"""
-

0 search results for '{unquote(cmdi_value)}'

+

0 search results for '{cmdi_value}'

""" @@ -1428,13 +2282,21 @@ def request_handler(self, request): return Response(parameter_block, status=200) async def setup_after_prep(self, module_test): - module_test.scan.modules["lightfuzz"].helpers.rand_string = lambda *args, **kwargs: "AAAAAAAAAAAAAA" + # Respect the numeric_only=True contract so int() conversions in the + # arithmetic canary path succeed. Non-numeric calls still receive the + # original deterministic letter string. + def rand_string(*args, **kwargs): + if kwargs.get("numeric_only"): + return "1234567890" + return "AAAAAAAAAAAAAA" + + module_test.scan.modules["lightfuzz"].helpers.rand_string = rand_string expect_args = re.compile("/") module_test.set_expect_requests_handler(expect_args=expect_args, request_handler=self.request_handler) def check(self, module_test, events): web_parameter_emitted = False - cmdi_echocanary_finding_emitted = False + cmdi_posix_arith_finding = False for e in events: if e.type == "WEB_PARAMETER": if "HTTP Extracted Parameter [search]" in e.data["description"]: @@ -1442,16 +2304,144 @@ def check(self, module_test, events): if e.type == "FINDING": if ( - "POSSIBLE OS Command Injection. Parameter: [search] Parameter Type: [GETPARAM] Detection Method: [echo canary] CMD Probe Delimeters: [&&]" + "POSSIBLE OS Command Injection. Parameter: [search] Parameter Type: [GETPARAM] Detection Method: [arithmetic canary (POSIX)] CMD Probe Delimeters: [&&]" in e.data["description"] ): - cmdi_echocanary_finding_emitted = True + cmdi_posix_arith_finding = True + + assert web_parameter_emitted, "WEB_PARAMETER was not emitted" + assert cmdi_posix_arith_finding, "POSIX arithmetic canary CMDi FINDING not emitted" + + +# CMDi Windows `set /A` canary: simulates a cmd.exe target that evaluates +# `set /A A*B` and prints the product. The arithmetic confirmation cascade +# should land on the Windows arithmetic probe (after the POSIX probe returns +# the literal) and emit a HIGH-confidence finding labeled as cmd. +class Test_Lightfuzz_cmdi_windows(Test_Lightfuzz_cmdi): + def request_handler(self, request): + qs = str(request.query_string.decode()) + parameter_block = """ +
+
+ + +
+
+ """ + if "search=" in qs: + value = qs.split("=")[1] + if "&" in value: + value = value.split("&")[0] + decoded = unquote(value) + # Simulate cmd.exe: evaluate `set /A A*B` and print the product; + # reflect $((A*B)) literal (cmd doesn't expand POSIX arithmetic); + # reflect `echo X` args verbatim. + setA = re.search(r"&& set /A (\d+)\*(\d+) &&", decoded) + if setA: + result = int(setA.group(1)) * int(setA.group(2)) + return Response(f"

{result}

", status=200) + if "&& echo " in decoded: + cmdi_value = decoded.split("&& echo ")[1].split(" ")[0] + return Response( + f"

0 search results for '{cmdi_value}'

", + status=200, + ) + return Response(f"

0 search results for '{decoded}'

", status=200) + return Response(parameter_block, status=200) + def check(self, module_test, events): + web_parameter_emitted = False + windows_arith_finding = False + for e in events: + if e.type == "WEB_PARAMETER": + if "HTTP Extracted Parameter [search]" in e.data["description"]: + web_parameter_emitted = True + if e.type == "FINDING": + desc = e.data["description"] + if ( + "Detection Method: [arithmetic canary (cmd)]" in desc + and "Parameter: [search]" in desc + and "[&&]" in desc + ): + windows_arith_finding = True assert web_parameter_emitted, "WEB_PARAMETER was not emitted" - assert cmdi_echocanary_finding_emitted, "echo canary CMDi FINDING not emitted" + assert windows_arith_finding, "Windows cmd arithmetic canary HIGH-confidence FINDING was not emitted" + + +# CMDi parser-error reflection: simulates a parser (SQL, JSON, YAML) that +# reflects the probe's offending token back in its error. Under the three- +# stage cascade this should NOT produce a "Possible Command Injection" +# finding at all, because neither the POSIX `$((A*B))` nor the Windows +# `set /A A*B` confirmation probe can coax a shell product out of a text +# parser. Instead, a separate "Possible Parameter Reflection" finding is +# emitted to preserve the adjacent-vuln signal without overclaiming cmdi. +class Test_Lightfuzz_cmdi_parser_reflection_downgrade(Test_Lightfuzz_cmdi): + def request_handler(self, request): + qs = str(request.query_string.decode()) + parameter_block = """ +
+
+ + +
+
+ """ + if "search=" in qs: + value = qs.split("=")[1] + if "&" in value: + value = value.split("&")[0] + decoded = unquote(value) + # Real parsers only trip when a shell-style metachar breaks the + # surrounding syntax. Require a delimiter before the token so the + # AAAA false-positive probe does not accidentally reflect. + arith_match = re.search(r"[;|&]\s*echo\s+(\$\(\(\d+\*\d+\)\))", decoded) + if arith_match: + return Response( + f'{{"error":"parse error near {arith_match.group(1)}"}}', + status=500, + mimetype="application/json", + ) + setA_match = re.search(r"[;|&]\s*set\s*/A\s+(\d+\*\d+)", decoded) + if setA_match: + return Response( + f'{{"error":"parse error near {setA_match.group(1)}"}}', + status=500, + mimetype="application/json", + ) + generic_match = re.search(r"[;|&]\s*echo\s+(\d+)", decoded) + if generic_match: + return Response( + f'{{"error":"parse error near {generic_match.group(1)}"}}', + status=500, + mimetype="application/json", + ) + return Response('{"results":[]}', status=200, mimetype="application/json") + return Response(parameter_block, status=200) + + def check(self, module_test, events): + cmdi_finding = False + reflection_finding = False + for e in events: + if e.type == "FINDING": + name = e.data.get("name", "") + if "Command Injection" in name: + cmdi_finding = True + if "Parameter Reflection" in name: + reflection_finding = True + + assert not cmdi_finding, ( + "Parser-error reflection should NOT produce a Possible Command " + "Injection finding under the three-stage cascade (neither POSIX " + "nor Windows arithmetic probe can coax a product out of a text " + "parser)." + ) + assert reflection_finding, ( + "Expected a Possible Parameter Reflection finding for the " + "reflection-only detection path (preserves adjacent-vuln signal " + "without overclaiming cmdi)." + ) -# CMDi interactsh class Test_Lightfuzz_cmdi_interactsh(Test_Lightfuzz_cmdi): @staticmethod def extract_subdomain_tag(data): @@ -1643,7 +2633,9 @@ def request_handler(self, request): return Response(parameter_block, status=200, headers={"Content-Type": "application/json"}) async def setup_after_prep(self, module_test): - module_test.scan.modules["lightfuzz"].helpers.rand_string = lambda *args, **kwargs: "AAAAAAAAAAAAAA" + module_test.scan.modules["lightfuzz"].helpers.rand_string = lambda *args, **kwargs: ( + "1234567890" if kwargs.get("numeric_only") else "AAAAAAAAAAAAAA" + ) expect_args = re.compile("/") module_test.set_expect_requests_handler(expect_args=expect_args, request_handler=self.request_handler) @@ -1674,32 +2666,40 @@ class Test_Lightfuzz_crypto_error(ModuleTestBase): }, } + canonical_value = "08a5a2cea9c5a5576e6e5314edcba581d21c7111c9c0c06990327b9127058d67" + def request_handler(self, request): qs = str(request.query_string.decode()) - parameter_block = """ + parameter_block = f"""
- +
""" crypto_block = """
-

Access Denied!

+

Padding is invalid

""" if "secret=" in qs: - value = qs.split("=")[1] + value = qs.split("=", 1)[1].split("&")[0] + # Canonical ciphertext decrypts cleanly → normal page; any other value + # (truncation, mutation, garbage) → padding error. + if value == self.canonical_value: + return Response(parameter_block, status=200) if value: return Response(crypto_block, status=200) return Response(parameter_block, status=200) async def setup_after_prep(self, module_test): - module_test.scan.modules["lightfuzz"].helpers.rand_string = lambda *args, **kwargs: "AAAAAAAAAAAAAA" + module_test.scan.modules["lightfuzz"].helpers.rand_string = lambda *args, **kwargs: ( + "1234567890" if kwargs.get("numeric_only") else "AAAAAAAAAAAAAA" + ) expect_args = re.compile("/") module_test.set_expect_requests_handler(expect_args=expect_args, request_handler=self.request_handler) @@ -1744,7 +2744,9 @@ def request_handler(self, request): return Response(fp_block, status=200) async def setup_after_prep(self, module_test): - module_test.scan.modules["lightfuzz"].helpers.rand_string = lambda *args, **kwargs: "AAAAAAAAAAAAAA" + module_test.scan.modules["lightfuzz"].helpers.rand_string = lambda *args, **kwargs: ( + "1234567890" if kwargs.get("numeric_only") else "AAAAAAAAAAAAAA" + ) expect_args = re.compile("/") module_test.set_expect_requests_handler(expect_args=expect_args, request_handler=self.request_handler) @@ -2300,41 +3302,389 @@ def check(self, module_test, events): assert esi_finding_emitted, "ESI FINDING not emitted" -# Envelope state isolation: crypto error detection with all submodules enabled. -# Crypto runs after sqli/cmdi/xss/path/ssti. Each prior submodule calls outgoing_probe_value() -# which must not corrupt the envelope state that crypto reads via incoming_probe_value(). -class Test_Lightfuzz_envelope_isolation_crypto(Test_Lightfuzz_crypto_error): +# ESI remote-include OOB — verifies the new `` probe triggers +# an interactsh callback when the edge actually fetches the include URL, +# producing a CRITICAL CONFIRMED finding separate from the tag-strip one. +class Test_Lightfuzz_esi_interactsh(Test_Lightfuzz_esi): config_overrides = { - "interactsh_disable": True, + "interactsh_disable": False, "modules": { - "lightfuzz": { - "enabled_submodules": ["sqli", "cmdi", "xss", "path", "ssti", "crypto", "serial", "esi"], - } + "lightfuzz": {"enabled_submodules": ["esi"]}, }, } + async def setup_before_prep(self, module_test): + self.interactsh_mock_instance = module_test.mock_interactsh("lightfuzz") -# Envelope state isolation: padding oracle detection with all submodules enabled. -class Test_Lightfuzz_envelope_isolation_paddingoracle(Test_Lightfuzz_PaddingOracleDetection): - config_overrides = { - "interactsh_disable": True, - "modules": { - "lightfuzz": { - "enabled_submodules": ["sqli", "cmdi", "xss", "path", "ssti", "crypto", "serial", "esi"], - } - }, - } + def mock_interactsh_factory(*args, **kwargs): + return self.interactsh_mock_instance + from bbot.core.helpers.helper import ConfigAwareHelper -# Envelope state isolation: reflecting padding oracle detection with all submodules enabled. -class Test_Lightfuzz_envelope_isolation_paddingoracle_reflecting(Test_Lightfuzz_PaddingOracleDetection_Reflecting): - config_overrides = { - "interactsh_disable": True, - "modules": { - "lightfuzz": { - "enabled_submodules": ["sqli", "cmdi", "xss", "path", "ssti", "crypto", "serial", "esi"], - } - }, + module_test.monkeypatch.setattr(ConfigAwareHelper, "interactsh", mock_interactsh_factory) + + def request_handler(self, request): + import re as _re + + qs = str(request.query_string.decode()) + # Parameter block must include the search form so excavate can + # extract the `search` GETPARAM on the initial fetch; otherwise + # esi only fuzzes the HEADER test param and the include probe + # never reaches us. + parameter_block = """ +
+
+ + +
+
+ """ + if "search=" in qs: + # Split only at the FIRST `=` — the include payload's `src=` + # embeds unencoded `=` characters that would otherwise + # truncate the value. + value = qs.split("=", 1)[1] + if "&" in value: + value = value.split("&")[0] + decoded = unquote(value) + # Simulate an edge that fetches esi:include src URLs. Extract + # the interactsh subdomain from within an include payload and + # fire a mock interaction to prove the OOB detection path. + if "search: '{decoded}'", status=200) + return Response(parameter_block, status=200) + + def check(self, module_test, events): + remote_include_finding = False + for e in events: + if e.type == "FINDING": + desc = e.data["description"] + if "Edge Side Include Remote Fetch (OOB Interaction)" in desc and "Parameter: [search]" in desc: + remote_include_finding = True + assert remote_include_finding, "ESI remote-include OOB FINDING not emitted" + + +# Envelope state isolation: crypto error detection with all submodules enabled. +# Crypto runs after sqli/cmdi/xss/path/ssti. Each prior submodule calls outgoing_probe_value() +# which must not corrupt the envelope state that crypto reads via incoming_probe_value(). +class Test_Lightfuzz_envelope_isolation_crypto(Test_Lightfuzz_crypto_error): + config_overrides = { + "interactsh_disable": True, + "modules": { + "lightfuzz": { + "enabled_submodules": ["sqli", "cmdi", "xss", "path", "ssti", "crypto", "serial", "esi"], + } + }, + } + + +# Envelope state isolation: padding oracle detection with all submodules enabled. +class Test_Lightfuzz_envelope_isolation_paddingoracle(Test_Lightfuzz_PaddingOracleDetection): + config_overrides = { + "interactsh_disable": True, + "modules": { + "lightfuzz": { + "enabled_submodules": ["sqli", "cmdi", "xss", "path", "ssti", "crypto", "serial", "esi"], + } + }, + } + + +# Envelope state isolation: reflecting padding oracle detection with all submodules enabled. +class Test_Lightfuzz_envelope_isolation_paddingoracle_reflecting(Test_Lightfuzz_PaddingOracleDetection_Reflecting): + config_overrides = { + "interactsh_disable": True, + "modules": { + "lightfuzz": { + "enabled_submodules": ["sqli", "cmdi", "xss", "path", "ssti", "crypto", "serial", "esi"], + } + }, + } + + +# ECB Mode Detection: ciphertext with repeated 16-byte blocks (A+B+A+B pattern) +class Test_Lightfuzz_ECBDetection(ModuleTestBase): + targets = ["http://127.0.0.1:8888/"] + modules_overrides = ["http", "excavate", "lightfuzz"] + config_overrides = { + "interactsh_disable": True, + "modules": { + "lightfuzz": {"enabled_submodules": ["crypto"]}, + }, + } + + # Two 16-byte blocks with non-overlapping byte values (high entropy), repeated A+B+A+B + _block_a = bytes(range(16)).hex() # 000102...0f + _block_b = bytes(range(128, 144)).hex() # 808182...8f + ecb_ciphertext_hex = _block_a + _block_b + _block_a + _block_b + + def request_handler(self, request): + qs = str(request.query_string.decode()) + parameter_block = f""" +
+
+ + +
+
+ """ + if "token=" in qs: + return Response("OK", status=200) + return Response(parameter_block, status=200) + + async def setup_after_prep(self, module_test): + module_test.scan.modules["lightfuzz"].helpers.rand_string = lambda *args, **kwargs: ( + "1234567890" if kwargs.get("numeric_only") else "AAAAAAAAAAAAAA" + ) + expect_args = re.compile("/") + module_test.set_expect_requests_handler(expect_args=expect_args, request_handler=self.request_handler) + + def check(self, module_test, events): + ecb_detected = False + for e in events: + if e.type == "FINDING": + if "ECB Mode Encryption Detected" in e.data["description"]: + ecb_detected = True + assert ecb_detected, "ECB Mode Encryption FINDING not emitted" + + +# ECB Negative: all unique blocks, should NOT detect ECB +class Test_Lightfuzz_ECBDetection_Negative(ModuleTestBase): + targets = ["http://127.0.0.1:8888/"] + modules_overrides = ["http", "excavate", "lightfuzz"] + config_overrides = { + "interactsh_disable": True, + "modules": { + "lightfuzz": {"enabled_submodules": ["crypto"]}, + }, + } + + # 4 unique 16-byte blocks with non-overlapping byte values (high entropy, no repeats) + unique_ciphertext_hex = ( + bytes(range(0, 16)).hex() + + bytes(range(64, 80)).hex() + + bytes(range(128, 144)).hex() + + bytes(range(192, 208)).hex() + ) + + def request_handler(self, request): + qs = str(request.query_string.decode()) + parameter_block = f""" +
+
+ + +
+
+ """ + if "token=" in qs: + return Response("OK", status=200) + return Response(parameter_block, status=200) + + async def setup_after_prep(self, module_test): + module_test.scan.modules["lightfuzz"].helpers.rand_string = lambda *args, **kwargs: ( + "1234567890" if kwargs.get("numeric_only") else "AAAAAAAAAAAAAA" + ) + expect_args = re.compile("/") + module_test.set_expect_requests_handler(expect_args=expect_args, request_handler=self.request_handler) + + def check(self, module_test, events): + for e in events: + if e.type == "FINDING": + assert "ECB Mode Encryption Detected" not in e.data["description"], ( + "ECB falsely detected on unique blocks" + ) + + +# CBC Bit-Flipping Detection: server returns different responses for different byte-position mutations +class Test_Lightfuzz_CBCBitflipDetection(ModuleTestBase): + targets = ["http://127.0.0.1:8888"] + modules_overrides = ["http", "excavate", "lightfuzz"] + config_overrides = { + "interactsh_disable": True, + "modules": { + "lightfuzz": {"enabled_submodules": ["crypto"]}, + }, + } + + # 3 blocks of 16 bytes = 48 bytes, base64-encoded + original_b64 = base64.b64encode(bytes(range(48))).decode() + + def request_handler(self, request): + encrypted_value = quote(self.original_b64) + default_html = f""" + +
+ + +
+ + """ + if "/process" in request.url and request.method == "POST": + if request.form and request.form.get("cipher"): + cipher_val = request.form["cipher"] + try: + raw = base64.b64decode(cipher_val) + except Exception: + return Response("Invalid input", status=200) + # Penultimate block starts at byte 16 + # Check which byte in the penultimate block differs from original + original_raw = bytes(range(48)) + if len(raw) == len(original_raw): + penultimate_start = 16 + for i in range(penultimate_start, penultimate_start + 16): + if raw[i] != original_raw[i]: + # First byte (pos 16) vs middle byte (pos 24) produce different responses + if i < penultimate_start + 8: + return Response("Decryption result: type A", status=200) + else: + return Response("Decryption result: type B", status=200) + return Response("Decryption failed", status=200) + return Response("Decryption failed", status=200) + return Response(default_html, status=200) + + async def setup_after_prep(self, module_test): + module_test.set_expect_requests_handler(expect_args=re.compile(".*"), request_handler=self.request_handler) + + def check(self, module_test, events): + cbc_bitflip_detected = False + for e in events: + if e.type == "FINDING": + if "CBC Bit-Flipping Detected" in e.data["description"]: + cbc_bitflip_detected = True + assert cbc_bitflip_detected, "CBC Bit-Flipping Detected FINDING not emitted" + + +# CBC Bit-Flipping Negative: server returns identical response regardless of mutation position +class Test_Lightfuzz_CBCBitflipDetection_Negative(ModuleTestBase): + targets = ["http://127.0.0.1:8888"] + modules_overrides = ["http", "excavate", "lightfuzz"] + config_overrides = { + "interactsh_disable": True, + "modules": { + "lightfuzz": {"enabled_submodules": ["crypto"]}, + }, + } + + original_b64 = base64.b64encode(bytes(range(48))).decode() + + def request_handler(self, request): + encrypted_value = quote(self.original_b64) + default_html = f""" + +
+ + +
+ + """ + if "/process" in request.url and request.method == "POST": + return Response("Decryption failed", status=200) + return Response(default_html, status=200) + + async def setup_after_prep(self, module_test): + module_test.set_expect_requests_handler(expect_args=re.compile(".*"), request_handler=self.request_handler) + + def check(self, module_test, events): + for e in events: + if e.type == "FINDING": + assert "CBC Bit-Flipping Detected" not in e.data["description"], ( + "CBC Bit-Flipping falsely detected on identical responses" + ) + + +# CBC Bit-Flipping without Padding Oracle: server never fails decryption (OPENSSL_ZERO_PADDING equivalent). +# Every input produces a unique response derived from the submitted value — no padding validity leaked. +# Padding oracle sends ~254 probes that all get unique responses → differ_count >> block_size → not detected. +# CBC bit-flip probes still produce different responses → detected. +class Test_Lightfuzz_CBCBitflipDetection_NoPaddingOracle(ModuleTestBase): + targets = ["http://127.0.0.1:8888"] + modules_overrides = ["http", "excavate", "lightfuzz"] + config_overrides = { + "interactsh_disable": True, + "modules": { + "lightfuzz": {"enabled_submodules": ["crypto"]}, + }, + } + + # 3 blocks of 16 bytes = 48 bytes, base64-encoded + original_b64 = base64.b64encode(bytes(range(48))).decode() + + def request_handler(self, request): + encrypted_value = quote(self.original_b64) + default_html = f""" + +
+ + +
+ + """ + if "/process" in request.url and request.method == "POST": + if request.form and request.form.get("cipher"): + cipher_val = request.form["cipher"] + # No error paths — always return content derived from input. + # Simulates OPENSSL_ZERO_PADDING: decryption never fails. + import hashlib + + digest = hashlib.md5(cipher_val.encode()).hexdigest() + return Response(f"Session loaded: {digest}", status=200) + return Response("Session loaded: default", status=200) + return Response(default_html, status=200) + + async def setup_after_prep(self, module_test): + module_test.set_expect_requests_handler(expect_args=re.compile(".*"), request_handler=self.request_handler) + + def check(self, module_test, events): + cbc_bitflip_detected = False + padding_oracle_detected = False + for e in events: + if e.type == "FINDING": + if "CBC Bit-Flipping Detected" in e.data["description"]: + cbc_bitflip_detected = True + if "Padding Oracle Vulnerability" in e.data["description"]: + padding_oracle_detected = True + assert cbc_bitflip_detected, "CBC Bit-Flipping should be detected even without padding oracle" + assert not padding_oracle_detected, "Padding Oracle should NOT be detected when decryption never fails" + + +# Envelope state isolation: ECB detection with all submodules enabled +class Test_Lightfuzz_envelope_isolation_ecb(Test_Lightfuzz_ECBDetection): + config_overrides = { + "interactsh_disable": True, + "modules": { + "lightfuzz": { + "enabled_submodules": ["sqli", "cmdi", "xss", "path", "ssti", "crypto", "serial", "esi"], + } + }, + } + + +# Envelope state isolation: CBC bit-flip detection with all submodules enabled +class Test_Lightfuzz_envelope_isolation_cbc_bitflip(Test_Lightfuzz_CBCBitflipDetection): + config_overrides = { + "interactsh_disable": True, + "modules": { + "lightfuzz": { + "enabled_submodules": ["sqli", "cmdi", "xss", "path", "ssti", "crypto", "serial", "esi"], + } + }, + } + + +# Envelope state isolation: CBC bit-flip without padding oracle, all submodules enabled +class Test_Lightfuzz_envelope_isolation_cbc_bitflip_no_po(Test_Lightfuzz_CBCBitflipDetection_NoPaddingOracle): + config_overrides = { + "interactsh_disable": True, + "modules": { + "lightfuzz": { + "enabled_submodules": ["sqli", "cmdi", "xss", "path", "ssti", "crypto", "serial", "esi"], + } + }, } @@ -2476,7 +3826,9 @@ def request_handler(self, request): return Response(parameter_block, status=200) async def setup_after_prep(self, module_test): - module_test.scan.modules["lightfuzz"].helpers.rand_string = lambda *args, **kwargs: "AAAAAAAAAAAAAA" + module_test.scan.modules["lightfuzz"].helpers.rand_string = lambda *args, **kwargs: ( + "1234567890" if kwargs.get("numeric_only") else "AAAAAAAAAAAAAA" + ) expect_args = re.compile("/") module_test.set_expect_requests_handler(expect_args=expect_args, request_handler=self.request_handler) @@ -2552,7 +3904,9 @@ def request_handler(self, request): return Response(parameter_block, status=200) async def setup_after_prep(self, module_test): - module_test.scan.modules["lightfuzz"].helpers.rand_string = lambda *args, **kwargs: "AAAAAAAAAAAAAA" + module_test.scan.modules["lightfuzz"].helpers.rand_string = lambda *args, **kwargs: ( + "1234567890" if kwargs.get("numeric_only") else "AAAAAAAAAAAAAA" + ) expect_args = re.compile("/") module_test.set_expect_requests_handler(expect_args=expect_args, request_handler=self.request_handler) @@ -2575,3 +3929,1416 @@ def check(self, module_test, events): assert sqli_postparam_converted_finding_emitted, ( "SQLi POSTPARAM (converted from GETPARAM) FINDING not emitted (try_get_as_post failed)" ) + + +# Padding Oracle Jitter Stability Pre-Check +class Test_Lightfuzz_PaddingOracleDetection_JitterStability(Test_Lightfuzz_PaddingOracleDetection): + """Padding oracle negative test: the endpoint produces different response bodies for identical inputs + (e.g. ADFS with embedded timestamps/nonces). The stability pre-check should detect this and skip.""" + + jitter_counter = 0 + + def request_handler(self, request): + encrypted_value = quote( + "dplyorsu8VUriMW/8DqVDU6kRwL/FDk3Q+4GXVGZbo0CTh9YX1YvzZZJrYe4cHxvAICyliYtp1im4fWoOa54Zg==" + ) + default_html_response = f""" + + +
+ + +
+ + + """ + + if "/decrypt" in request.url and request.method == "POST": + # Every response is unique, simulating ADFS-style dynamic content + Test_Lightfuzz_PaddingOracleDetection_JitterStability.jitter_counter += 1 + response_content = f"Error correlation_id={Test_Lightfuzz_PaddingOracleDetection_JitterStability.jitter_counter} nonce=abc{Test_Lightfuzz_PaddingOracleDetection_JitterStability.jitter_counter}" + return Response(response_content, status=200) + else: + return Response(default_html_response, status=200) + + def check(self, module_test, events): + web_parameter_extracted = False + padding_oracle_detected = False + for e in events: + if e.type == "WEB_PARAMETER": + if "HTTP Extracted Parameter [encrypted_data] (POST Form" in e.data["description"]: + web_parameter_extracted = True + if e.type == "FINDING": + if "Padding Oracle" in e.data["description"]: + padding_oracle_detected = True + + assert web_parameter_extracted, "Web parameter was not extracted" + assert not padding_oracle_detected, ( + "Padding oracle should NOT be detected when endpoint has jittery responses (stability pre-check should abort)" + ) + + +# XSS Multi-Context Reflection False Positive +class Test_Lightfuzz_xss_multicontext(Test_Lightfuzz_xss): + """XSS negative test: parameter reflected in multiple contexts with different encoding. + Quote survives in text content but is encoded in tag attribute. Should NOT report Tag Attribute XSS.""" + + def request_handler(self, request): + qs = str(request.query_string.decode()) + + parameter_block = """ + +
+ + +
+ + """ + if "path=" in qs: + value = qs.split("path=")[1] + if "&" in value: + value = value.split("&")[0] + decoded = unquote(value) + # Tag attribute context: quotes are URL-encoded (safe) + attr_value = decoded.replace('"', "%22") + # Text content: raw reflection (quotes survive but harmless here) + text_value = decoded + # JS context: everything URL-encoded (safe) + js_value = value + + html = f""" + +
+ +
+

{text_value}

+ + + """ + return Response(html, status=200) + return Response(parameter_block, status=200) + + async def setup_after_prep(self, module_test): + module_test.scan.modules["lightfuzz"].helpers.rand_string = lambda *args, **kwargs: ( + "1234567890" if kwargs.get("numeric_only") else "AAAAAAAAAAAAAA" + ) + expect_args = re.compile("/") + module_test.set_expect_requests_handler(expect_args=expect_args, request_handler=self.request_handler) + + def check(self, module_test, events): + web_parameter_emitted = False + tag_attribute_xss_emitted = False + for e in events: + if e.type == "WEB_PARAMETER": + if "HTTP Extracted Parameter [path]" in e.data["description"]: + web_parameter_emitted = True + if e.type == "FINDING": + if "Possible Reflected XSS" in e.data["description"] and "Tag Attribute" in e.data["description"]: + tag_attribute_xss_emitted = True + + assert web_parameter_emitted, "WEB_PARAMETER was not emitted" + assert not tag_attribute_xss_emitted, ( + "Tag Attribute XSS should NOT be reported when the quote only survives in text content, not in tag attributes" + ) + + +# SQLi WAF False Positive (Akamai-style 403) +class Test_Lightfuzz_sqli_waf(Test_Lightfuzz_sqli): + """SQLi negative test: endpoint returns 403 with WAF signature when single quote is sent. + Should NOT report SQL injection.""" + + def request_handler(self, request): + qs = str(request.query_string.decode()) + parameter_block = """ +
+
+ + +
+
+ """ + if "search=" in qs: + value = qs.split("=")[1] + if "&" in value: + value = value.split("&")[0] + + if value.endswith("'") and not value.endswith("''"): + # WAF blocks the request with a known WAF string + waf_response = """ + + Access Denied + +

Access Denied

+

The requested URL was rejected. Please consult with your administrator.

+ + + """ + return Response(waf_response, status=403) + return Response(parameter_block, status=200) + return Response(parameter_block, status=200) + + def check(self, module_test, events): + web_parameter_emitted = False + sqli_finding_emitted = False + for e in events: + if e.type == "WEB_PARAMETER": + if "HTTP Extracted Parameter [search]" in e.data["description"]: + web_parameter_emitted = True + if e.type == "FINDING": + if "Possible SQL Injection" in e.data["description"] and "Code Change" in e.data["description"]: + sqli_finding_emitted = True + + assert web_parameter_emitted, "WEB_PARAMETER was not emitted" + assert not sqli_finding_emitted, ( + "SQLi should NOT be reported when single quote probe triggers a WAF 403 response" + ) + + +# SQLi flappy baseline: server alternates between status codes across rounds. +# The confirmation loop should detect the instability and suppress the finding. +class Test_Lightfuzz_sqli_flappy_baseline(Test_Lightfuzz_sqli): + """SQLi negative test: server flaps between 200 and 303 across requests. + The confirmation loop should detect unstable baselines and suppress the finding.""" + + request_count = 0 + + def request_handler(self, request): + self.__class__.request_count += 1 + qs = str(request.query_string.decode()) + parameter_block = """ +
+
+ + +
+
+ """ + if "search=" in qs: + value = qs.split("=")[1] + if "&" in value: + value = value.split("&")[0] + + # First round: return the classic 200->500->200 pattern to trigger initial detection + # Subsequent rounds: flap baseline to 303 to simulate CDN instability + if self.__class__.request_count > 12: + # After initial probes, start flapping the baseline + return Response("Redirecting", status=303, headers={"Location": "/"}) + + if value.endswith("'"): + if value.endswith("''"): + return Response("

normal

", status=200) + return Response("

error

", status=500) + return Response(parameter_block, status=200) + + def check(self, module_test, events): + web_parameter_emitted = False + sqli_finding_emitted = False + for e in events: + if e.type == "WEB_PARAMETER": + if "HTTP Extracted Parameter [search]" in e.data["description"]: + web_parameter_emitted = True + if e.type == "FINDING": + if "Possible SQL Injection" in e.data["description"] and "Code Change" in e.data["description"]: + sqli_finding_emitted = True + + assert web_parameter_emitted, "WEB_PARAMETER was not emitted" + assert not sqli_finding_emitted, ( + "SQLi code-change finding should NOT be emitted when baseline is flappy across confirmation rounds" + ) + + +# Verify that POST SQLi findings include additional_params in the description +class Test_Lightfuzz_sqli_post_additional_params(ModuleTestBase): + targets = ["http://127.0.0.1:8888"] + modules_overrides = ["http", "lightfuzz", "excavate"] + config_overrides = { + "interactsh_disable": True, + "modules": { + "lightfuzz": { + "enabled_submodules": ["sqli"], + } + }, + } + + def request_handler(self, request): + parameter_block = """ +
+
+ + + +
+
+ """ + if "search" in request.form.keys(): + value = request.form["search"] + if value.endswith("'"): + if value.endswith("''"): + return Response("

normal

", status=200) + return Response("

error

", status=500) + return Response("

results

", status=200) + return Response(parameter_block, status=200) + + async def setup_after_prep(self, module_test): + module_test.scan.modules["lightfuzz"].helpers.rand_string = lambda *args, **kwargs: ( + "1234567890" if kwargs.get("numeric_only") else "AAAAAAAAAAAAAA" + ) + expect_args = re.compile("/") + module_test.set_expect_requests_handler(expect_args=expect_args, request_handler=self.request_handler) + + def check(self, module_test, events): + sqli_finding_emitted = False + for e in events: + if e.type == "FINDING" and "Code Change" in e.data.get("description", ""): + desc = e.data["description"] + if "POSTPARAM" in desc: + # POST findings should include additional_params info + assert "Additional Params:" in desc, f"POST finding description missing additional_params: {desc}" + sqli_finding_emitted = True + assert sqli_finding_emitted, "SQLi POST code-change FINDING not emitted" + + +# Verify that lightfuzz rejects WEB_PARAMETER events on static-asset URLs (.pdf, .xml, etc.) +class Test_Lightfuzz_static_url_filter(ModuleTestBase): + targets = ["http://127.0.0.1:8888"] + modules_overrides = ["http", "lightfuzz", "excavate"] + config_overrides = { + "interactsh_disable": True, + "modules": { + "lightfuzz": { + "enabled_submodules": ["sqli"], + } + }, + } + + async def setup_after_prep(self, module_test): + module_test.scan.modules["lightfuzz"].helpers.rand_string = lambda *args, **kwargs: ( + "1234567890" if kwargs.get("numeric_only") else "AAAAAAAAAAAAAA" + ) + respond_args = {"response_data": "placeholder", "status": 200} + expect_args = {"method": "GET", "uri": "/"} + module_test.set_expect_requests(expect_args=expect_args, respond_args=respond_args) + + # Inject a WEB_PARAMETER event on a .pdf URL + seed_events = [] + parent_event = module_test.scan.make_event( + "http://127.0.0.1:8888/", + "URL", + module_test.scan.root_event, + module="http", + tags=["status-200", "distance-0"], + ) + data = { + "host": "127.0.0.1", + "type": "GETPARAM", + "name": "v", + "original_value": "1", + "url": "http://127.0.0.1:8888/document.pdf?v=1", + "description": "HTTP Extracted Parameter [v]", + } + seed_event = module_test.scan.make_event(data, "WEB_PARAMETER", parent_event, tags=["distance-0"]) + seed_events.append(seed_event) + for event in seed_events: + await module_test.scan.ingress_module.incoming_event_queue.put(event) + + def check(self, module_test, events): + sqli_finding_emitted = False + for e in events: + if e.type == "FINDING": + if "Possible SQL Injection" in e.data["description"]: + sqli_finding_emitted = True + + assert not sqli_finding_emitted, ( + "SQLi finding should NOT be emitted for WEB_PARAMETER on a static-asset URL (.pdf)" + ) + + +# End-to-end test for the baseline → HTTP_RESPONSE → excavate chain. +# +# Validates two features in one pass: +# 1. excavate's whose first option is empty +# and second option is `