diff --git a/bbot/core/event/base.py b/bbot/core/event/base.py index 2602f4debe..751d254a67 100644 --- a/bbot/core/event/base.py +++ b/bbot/core/event/base.py @@ -958,6 +958,11 @@ def json(self, mode="json"): web_spider_distance = getattr(self, "web_spider_distance", None) if web_spider_distance is not None: j["web_spider_distance"] = web_spider_distance + # provenance for events whose evidence is an archived snapshot rather than the live host. + # kept out of `data` so it doesn't become part of the event's identity. + archive_url = self.archive_url + if archive_url: + j["archive_url"] = archive_url # scope distance j["scope_distance"] = self.scope_distance # scan @@ -1959,7 +1964,9 @@ def _pretty_string(self): confidence_str = f"[\033[1m{confidence}\033[0m]" else: confidence_str = f"[{confidence}]" - return f"Severity: [{severity}] Confidence: {confidence_str} {description}" + # the evidence came out of an archive, so the severity describes a snapshot, not the live host + archived_str = "[ARCHIVED] " if self.archive_url else "" + return f"{archived_str}Severity: [{severity}] Confidence: {confidence_str} {description}" def _data_human(self): parts = [] @@ -1972,6 +1979,10 @@ def _data_human(self): cves = self.data.get("cves", []) if cves: parts.append(f"[{', '.join(cves)}]") + # the evidence came out of an archive, so the severity describes a snapshot, not the live host + archive_url = self.archive_url + if archive_url: + parts.append(f"(archived: {archive_url})") return " ".join(parts) diff --git a/bbot/core/helpers/diff.py b/bbot/core/helpers/diff.py index 126f122e68..53237332ec 100644 --- a/bbot/core/helpers/diff.py +++ b/bbot/core/helpers/diff.py @@ -326,12 +326,19 @@ async def compare( else: return (False, diff_reasons, reflection, subject_response) - def _compare_sync(self, subject_response, subject): - """CPU-bound comparison work offloaded from the event loop.""" + @staticmethod + def parse_body(text): + """Parse a response body into the structure compare_body() expects: a document tree + when the body is well-formed XML, otherwise a list of lines. Passing raw text to + compare_body() works but bypasses the ddiff_filters that mask dynamic content.""" try: - subject_json = xmltodict.parse(subject_response.text) + return xmltodict.parse(text) except ExpatError: - subject_json = subject_response.text.split("\n") + return text.split("\n") + + def _compare_sync(self, subject_response, subject): + """CPU-bound comparison work offloaded from the event loop.""" + subject_json = self.parse_body(subject_response.text) diff_reasons = [] diff --git a/bbot/modules/lightfuzz/submodules/crypto.py b/bbot/modules/lightfuzz/submodules/crypto.py index 085589c1d5..00172f7edb 100644 --- a/bbot/modules/lightfuzz/submodules/crypto.py +++ b/bbot/modules/lightfuzz/submodules/crypto.py @@ -41,20 +41,20 @@ def _ascii_xor_score(b): def _is_structured_id_pair(bytes_a, bytes_b, xored, zero_run): """True when the pair looks like structured identifiers (MongoDB ObjectIds, - hex timestamps, sequential counters, time-ordered UUIDs) rather than - reused-keystream ciphertexts. + hex timestamps, sequential counters, record GUIDs) rather than reused-keystream + ciphertexts. - Structured IDs share a long byte prefix (timestamp, process ID, ...) and - differ only in a short counter/random suffix. Their XOR has a long leading- - zero run followed by a sparse or random tail -- superficially identical to - keystream reuse with shared-prefix plaintexts, but distinguishable by what - the tail looks like. + Structured IDs share a byte prefix (timestamp, process ID, ...) and differ only + in a short counter/random suffix. Their XOR has a leading-zero run followed by + a sparse or random tail -- superficially identical to keystream reuse with + shared-prefix plaintexts, but distinguishable by what the tail looks like. Real many-time-pad: the tail is XOR of diverging ASCII plaintexts -- dense, diverse bytes mostly in [0x00, 0x60]. - Structured IDs: the tail is either a small counter delta (sparse zeros with - one nonzero byte) or random machine/process bytes (fails ASCII-XOR check). + Structured IDs: the tail is a small counter delta (sparse zeros with one nonzero + byte), random machine/process bytes (fails the ASCII-XOR check), or a fixed-field + template whose matching segments zero out scattered through the tail. """ tail = xored[zero_run:] @@ -68,6 +68,15 @@ def _is_structured_id_pair(bytes_a, bytes_b, xored, zero_run): if len(bytes_a) == len(bytes_b) and len(tail) <= 4: return True + # Zeros scattered *through* the tail mean the two values keep re-converging at + # fixed offsets -- a shared field template (an instance/table/timestamp segment + # in a record GUID). Two ciphertexts under a reused keystream zero out only where + # their plaintexts still coincide, which is the leading run, or a shared suffix. + # Trailing zeros are stripped first so shared-suffix plaintexts aren't caught here. + core = tail.rstrip(b"\x00") + if sum(1 for b in core if b == 0) >= 2: + return True + # For longer tails: real many-time-pad XOR reveals XOR of diverging ASCII # plaintexts -- dense nonzero bytes mostly in [0x00, 0x60]. Random # suffixes (UUID v7 random bits, different-process machine bytes) and @@ -369,11 +378,10 @@ def detect_keystream_reuse(self, probe_value): # At least 2 leading zero bytes OR ≥90% of bytes in ASCII-XOR-ASCII range if zero_run < 2 and ascii_score < 0.9: continue - # A leading-zero run alone is the hallmark of structured hex - # identifiers (MongoDB ObjectIds, hex timestamps, sequential - # counters, time-ordered UUIDs) sharing a byte prefix -- not - # keystream reuse. Filter those out by examining the tail. - if zero_run >= 2 and _is_structured_id_pair(bytes_a, bytes_b, xored, zero_run): + # Both entry conditions are met just as easily by structured hex identifiers + # as by real ciphertexts, so every candidate pair goes through the same + # discrimination on what the diverging region actually looks like. + if _is_structured_id_pair(bytes_a, bytes_b, xored, zero_run): continue pair_score = (zero_run, ascii_score) if best is None or pair_score > (best[0], best[1]): diff --git a/bbot/modules/lightfuzz/submodules/serial.py b/bbot/modules/lightfuzz/submodules/serial.py index 0e67bec69b..e9e3beff91 100644 --- a/bbot/modules/lightfuzz/submodules/serial.py +++ b/bbot/modules/lightfuzz/submodules/serial.py @@ -207,6 +207,29 @@ def payload_language(payload_name): """Extract the language family from a payload name (e.g. 'java_base64_string_error' -> 'java').""" return payload_name.split("_")[0] + @staticmethod + def corrupt_payload(payload, encoding): + """Return a twin of ``payload`` with the same encoding, length and trailing bytes but a + scrambled magic/type header: still parses like the original, deserializes under nothing. + Returns None when no distinguishable twin can be built. + """ + if not payload: + return None + if encoding == "php_raw": + # PHP serialized data leads with a single type character + return f"z{payload[1:]}" if payload[0] != "z" else f"q{payload[1:]}" + try: + data = bytes.fromhex(payload) if encoding == "hex" else base64.b64decode(payload) + except Exception: + return None + header_length = min(4, len(data)) + if not header_length: + return None + corrupted = bytes((b + 0x55) % 256 for b in data[:header_length]) + data[header_length:] + if encoding == "hex": + return corrupted.hex().upper() if payload.isupper() else corrupted.hex() + return base64.b64encode(corrupted).decode() + async def confirm_baseline(self, control_payload, cookies): """Re-send the control payload to confirm the baseline error state is stable (not transient).""" confirmation = await self.standard_probe(self.event.data["type"], cookies, control_payload) @@ -248,13 +271,13 @@ async def fuzz(self): # Map each payload set to its control payload for baseline confirmation payload_sets = [ - (base64_serialization_payloads, http_compare_base64, control_payload_base64), - (hex_serialization_payloads, http_compare_hex, control_payload_hex), - (php_raw_serialization_payloads, http_compare_php_raw, control_payload_php_raw), + (base64_serialization_payloads, http_compare_base64, control_payload_base64, "base64"), + (hex_serialization_payloads, http_compare_hex, control_payload_hex, "hex"), + (php_raw_serialization_payloads, http_compare_php_raw, control_payload_php_raw, "php_raw"), ] # Proceed with payload probes - for payload_set, payload_baseline, control_payload in payload_sets: + for payload_set, payload_baseline, control_payload, encoding in payload_sets: for payload_type, payload in payload_set.items(): try: matches_baseline, diff_reasons, reflection, response = await self.compare_probe( @@ -315,6 +338,22 @@ async def fuzz(self): ) continue + # Deserialization is a claim about the payload's content, so a same-shape twin + # with a scrambled header must not resolve the error too. If it does, the value + # is only being parsed (e.g. as a URL/host), not deserialized. + corrupted_payload = self.corrupt_payload(payload, encoding) + if corrupted_payload is not None: + corrupted_response = await self.standard_probe( + self.event.data["type"], cookies, corrupted_payload + ) + corrupted_status = getattr(corrupted_response, "status_code", None) + if corrupted_status == status_code: + self.debug( + f"Corrupted twin of {payload_type} also returned {corrupted_status}, " + "outcome is independent of payload content, skipping" + ) + continue + def get_title(text): soup = self.lightfuzz.helpers.beautifulsoup(text, "html.parser") if soup and soup.title and soup.title.string: diff --git a/bbot/modules/lightfuzz/submodules/sqli.py b/bbot/modules/lightfuzz/submodules/sqli.py index 545d54a494..12029e85e8 100644 --- a/bbot/modules/lightfuzz/submodules/sqli.py +++ b/bbot/modules/lightfuzz/submodules/sqli.py @@ -1,3 +1,5 @@ +from urllib.parse import quote + from .base import BaseLightfuzz from bbot.errors import HttpCompareError @@ -13,6 +15,10 @@ class sqli(BaseLightfuzz): - Tests quote escape sequence variations - Matches against known SQL error patterns + * Code-change Detection: + - Compares the status code of a single-quote probe against a doubled-quote probe + - Requires a positive boolean (TRUE/FALSE) content differential to confirm + * Time-based Blind Detection: - Uses vendor-specific time delay payloads - Confirms delays with statistical analysis @@ -41,6 +47,14 @@ class sqli(BaseLightfuzz): "string not properly terminated", ] + # TRUE/FALSE payload pairs used to confirm the value reaches a SQL query. Both halves of a + # pair are the same length and differ by a single character, so a reflected payload can be + # stripped cleanly and any surviving body difference comes from the query result set. + BOOLEAN_PROBE_PAIRS = [ + ("' AND '1'='1", "' AND '1'='2"), + (" AND 1=1", " AND 1=2"), + ] + DELAY_PROBE_TEMPLATES = [ "'||pg_sleep({d})--", "' OR (SELECT TRUE FROM pg_sleep({d})) LIMIT 1-- -", @@ -115,6 +129,106 @@ async def _confirm_code_change(self, probe_value, cookies, initial_status_codes, return True + @staticmethod + def _strip_payload(text, payload): + """Remove reflected copies of a payload from a response body, in raw and encoded form.""" + for variant in (payload, quote(payload), payload.replace(" ", "+")): + text = text.replace(variant, "") + return text + + async def _probe_body(self, http_compare, payload, cookies): + """Send ``payload`` and return its parsed response body, reflections of the payload + stripped, ready for ``http_compare.compare_body()``. + + Returns None when the probe fails, or when the response is 403/429 (WAF or rate limit, + which tells us nothing about the query behind the parameter). + """ + try: + probe = await self.compare_probe( + http_compare, + self.event.data["type"], + payload, + cookies, + additional_params_populate_empty=True, + ) + except HttpCompareError as e: + self.debug(f"Boolean probe [{payload}] failed: {e}") + return None + if not probe[3]: + return None + if probe[3].status_code in (403, 429): + self.debug(f"Boolean probe [{payload}] returned {probe[3].status_code}, cannot confirm") + return None + return http_compare.parse_body(self._strip_payload(probe[3].text, payload)) + + async def confirm_boolean_differential(self, http_compare, probe_value, cookies): + """Require positive SQL-logic evidence before asserting injection from a status change. + + A bare status flip is not SQL: a WAF signature match, or an envelope whose structural + validity changes when the payload is repacked, both produce one. Only a TRUE/FALSE pair + that changes the response *content* shows the value is reaching a query. + + Returns the confirming ``(true_payload, false_payload)`` pair, or None. + """ + for true_suffix, false_suffix in self.BOOLEAN_PROBE_PAIRS: + true_payload = f"{probe_value}{true_suffix}" + false_payload = f"{probe_value}{false_suffix}" + + true_body = await self._probe_body(http_compare, true_payload, cookies) + if true_body is None: + continue + false_body = await self._probe_body(http_compare, false_payload, cookies) + if false_body is None: + continue + + if http_compare.compare_body(true_body, false_body) is not False: + self.debug(f"No boolean differential for [{true_suffix}] / [{false_suffix}]") + continue + + # A page that renders differently on every request produces a differential on its + # own. Re-send the TRUE payload; the body must reproduce for the pair to mean anything. + repeat_body = await self._probe_body(http_compare, true_payload, cookies) + if repeat_body is None: + continue + if http_compare.compare_body(true_body, repeat_body) is False: + self.debug("Response body is not deterministic, discarding boolean differential") + continue + + self.verbose(f"Boolean differential confirmed for {self.event.url}: [{true_suffix}] vs [{false_suffix}]") + return true_payload, false_payload + return None + + async def is_quote_specific(self, http_compare, probe_value, cookies, status_codes): + """Verify the status flip tracks the quote characters and not the payload's shape. + + Appending one vs. two benign characters mirrors the `'`/`''` pair in length while + carrying no SQL meaning. If that benign pair reproduces the same status triplet, the + flip tracks value length or envelope validity rather than quoting. + """ + control_codes = [] + for suffix in ("a", "aa"): + try: + control = await self.compare_probe( + http_compare, + self.event.data["type"], + f"{probe_value}{suffix}", + cookies, + additional_params_populate_empty=True, + ) + except HttpCompareError as e: + self.debug(f"Quote-specificity control probe failed: {e}") + return True + if not control[3]: + return True + control_codes.append(control[3].status_code) + + if (status_codes[0], *control_codes) == status_codes: + self.debug( + f"Benign control pair reproduced the status triplet {status_codes}, the change is not quote-specific" + ) + return False + return True + async def fuzz(self): cookies = self.event.data.get("assigned_cookies", {}) probe_value = self.incoming_probe_value(populate_empty=True) @@ -165,19 +279,14 @@ async def fuzz(self): if "code" in single_quote[1] and ( single_quote[3].status_code != double_single_quote[3].status_code ): - # Check if the status code change is due to a WAF, not SQL injection - is_waf = False - if single_quote[3].status_code == 403: - waf_matches = await self.lightfuzz.helpers.yara.match( - self.lightfuzz.waf_yara_rules, single_quote[3].text + # A transition into 403 is access control (usually a WAF matching its managed + # SQLi signature on the bare quote), not a code change driven by the query. + if 403 in (single_quote[3].status_code, double_single_quote[3].status_code): + self.debug( + "Quote probe transitioned into 403 (access control/WAF), " + "suppressing SQL injection finding" ) - if waf_matches: - self.debug( - "Single quote probe returned 403 with WAF signature, " - "suppressing SQL injection finding" - ) - is_waf = True - if not is_waf: + else: # Confirmation loop: require 2 additional rounds with fresh baselines # to confirm the status-code triplet is stable and not a transient CDN/server flap. # TODO: apply this same confirmation pattern to other submodules that use compare_probe-based detection. @@ -187,15 +296,28 @@ async def fuzz(self): double_single_quote[3].status_code, ) confirmed = await self._confirm_code_change(probe_value, cookies, initial_status_codes) - if confirmed: + quote_specific = confirmed and await self.is_quote_specific( + http_compare, probe_value, cookies, initial_status_codes + ) + boolean_pair = ( + await self.confirm_boolean_differential(http_compare, probe_value, cookies) + if quote_specific + else None + ) + if boolean_pair: self.results.append( { "name": "Possible SQL Injection", "severity": "HIGH", "confidence": "MEDIUM", - "description": f"Possible SQL Injection. {self.metadata()} Detection Method: [Single Quote/Two Single Quote, Code Change ({initial_status_codes[0]}->{initial_status_codes[1]}->{initial_status_codes[2]})]", + "description": f"Possible SQL Injection. {self.metadata()} Detection Method: [Single Quote/Two Single Quote, Code Change ({initial_status_codes[0]}->{initial_status_codes[1]}->{initial_status_codes[2]})] Boolean Confirmation: [{boolean_pair[0]}] vs [{boolean_pair[1]}]", } ) + elif confirmed and quote_specific: + self.verbose( + f"Discarding code change {initial_status_codes} for {self.event.url}: " + "no boolean differential, the value does not reach a query" + ) else: self.debug("Failed to get responses for both single_quote and double_single_quote") except HttpCompareError as e: diff --git a/bbot/test/test_step_1/test_events.py b/bbot/test/test_step_1/test_events.py index 95e3d5a1f5..c91ce0015a 100644 --- a/bbot/test/test_step_1/test_events.py +++ b/bbot/test/test_step_1/test_events.py @@ -1027,6 +1027,75 @@ async def test_event_web_spider_distance(bbot_scanner): assert "spider-max" not in url_event_5.tags +@pytest.mark.asyncio +async def test_event_archived_provenance(): + """A finding whose evidence is an archived snapshot renders with an [ARCHIVED] marker, so the + severity is never read as a claim about the live host.""" + scan = Scanner() + await scan._prep() + archived_response = scan.make_event( + { + "method": "GET", + "url": "http://www.evilcorp.com/asdf", + "hash": {"header_mmh3": "1", "body_mmh3": "2"}, + "raw_header": "HTTP/1.1 200 OK\r\n\r\n", + "archive_url": "http://web.archive.org/web/20190101000000/http://www.evilcorp.com/asdf", + }, + "HTTP_RESPONSE", + parent=scan.root_event, + tags=["from-wayback", "archived"], + ) + + finding = scan.make_event( + { + "host": "www.evilcorp.com", + "description": "test", + "severity": "HIGH", + "confidence": "HIGH", + "name": "Test Finding", + }, + "FINDING", + parent=archived_response, + ) + archive_url = "http://web.archive.org/web/20190101000000/http://www.evilcorp.com/asdf" + assert finding.archive_url == archive_url + assert finding.pretty_string.startswith("[ARCHIVED] Severity: [HIGH]") + # output.txt / stdout render data_human, which carries the snapshot URL itself + assert finding.data_human.startswith("Severity: [HIGH]") + assert finding.data_human.endswith(f"(archived: {archive_url})") + # output.json serializes json(), and the snapshot URL must not become part of the finding's identity + assert finding.json()["archive_url"] == archive_url + assert "archive_url" not in finding.json()["data_json"] + + live_response = scan.make_event( + { + "method": "GET", + "url": "http://www.evilcorp.com/qwerty", + "hash": {"header_mmh3": "3", "body_mmh3": "4"}, + "raw_header": "HTTP/1.1 200 OK\r\n\r\n", + }, + "HTTP_RESPONSE", + parent=scan.root_event, + ) + live_finding = scan.make_event( + { + "host": "www.evilcorp.com", + "description": "test", + "severity": "HIGH", + "confidence": "HIGH", + "name": "Live Finding", + }, + "FINDING", + parent=live_response, + ) + assert live_finding.archive_url is None + assert live_finding.pretty_string.startswith("Severity: [HIGH]") + assert live_finding.data_human.startswith("Severity: [HIGH]") + assert "archive_url" not in live_finding.json() + + await scan._cleanup() + + @pytest.mark.asyncio async def test_event_closest_host(): scan = Scanner() 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 e9cc38b87d..6c84b6c5bc 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 @@ -22,6 +22,25 @@ def _make_base_lightfuzz(url): return BaseLightfuzz(lightfuzz, event) +def sqli_injectable_response(value, empty_block): + """Emulate a parameter whose value lands unquoted inside a SQL string literal. + + Covers everything the sqli code-change branch must observe on a genuinely injectable + parameter: an unbalanced quote errors, a doubled quote recovers, and a TRUE/FALSE + boolean pair changes the result set. Benign suffixes behave like any other search term, + so the flip stays quote-specific. + """ + if value is None: + return Response(empty_block, status=200) + if value.endswith("' AND '1'='1"): + return Response(f"

1 result for: {value}

Row Alpha

", status=200) + if value.endswith("' AND '1'='2"): + return Response(f"

0 results for: {value}

", status=200) + if value.endswith("'") and not value.endswith("''"): + return Response("

Found error in SQL query

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

0 results for: {value}

", status=200) + + 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" @@ -1198,40 +1217,17 @@ class Test_Lightfuzz_sqli(ModuleTestBase): }, } - def request_handler(self, request): - qs = str(request.query_string.decode()) - parameter_block = """ + parameter_block = """ - """ - if "search=" in qs: - value = qs.split("=")[1] - - if "&" in value: - value = value.split("&")[0] - - sql_block_normal = f""" -
-

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

-
-
""" - sql_block_error = """ -
-

Found error in SQL query

-
-
- """ - if value.endswith("'"): - if value.endswith("''"): - return Response(sql_block_normal, status=200) - return Response(sql_block_error, status=500) - return Response(parameter_block, status=200) + def request_handler(self, request): + return sqli_injectable_response(request.args.get("search"), self.parameter_block) async def setup_after_prep(self, module_test): module_test.scan.modules["lightfuzz"].helpers.rand_string = lambda *args, **kwargs: ( @@ -1271,8 +1267,7 @@ class Test_Lightfuzz_sqli_post(ModuleTestBase): }, } - def request_handler(self, request): - parameter_block = """ + parameter_block = """ """ - if "search" in request.form.keys(): - value = request.form["search"] - - sql_block_normal = f""" -
-

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

-
-
- """ - - sql_block_error = """ -
-

Found error in SQL query

-
-
- """ - if value.endswith("'"): - if value.endswith("''"): - return Response(sql_block_normal, status=200) - return Response(sql_block_error, status=500) - return Response(parameter_block, status=200) + def request_handler(self, request): + return sqli_injectable_response(request.form.get("search"), self.parameter_block) async def setup_after_prep(self, module_test): module_test.scan.modules["lightfuzz"].helpers.rand_string = lambda *args, **kwargs: ( @@ -1391,32 +1367,14 @@ async def setup_after_prep(self, module_test): for event in seed_events: await module_test.scan.ingress_module.incoming_event_queue.put(event) - def request_handler(self, request): - placeholder_block = """ + placeholder_block = """

placeholder

""" - if request.headers.get("testheader") is not None: - header_value = request.headers.get("testheader") - - header_block_normal = f""" - -

placeholder

-

test: {header_value}

- - """ - header_block_error = """ - -

placeholder

-

Error!

- - """ - if header_value.endswith("'") and not header_value.endswith("''"): - return Response(header_block_error, status=500) - return Response(header_block_normal, status=200) - return Response(placeholder_block, status=200) + def request_handler(self, request): + return sqli_injectable_response(request.headers.get("testheader"), self.placeholder_block) def check(self, module_test, events): sqli_finding_emitted = False @@ -1461,33 +1419,14 @@ async def setup_after_prep(self, module_test): for event in seed_events: await module_test.scan.ingress_module.incoming_event_queue.put(event) - def request_handler(self, request): - placeholder_block = """ + placeholder_block = """

placeholder

""" - if request.cookies.get("test") is not None: - header_value = request.cookies.get("test") - - header_block_normal = f""" - -

placeholder

-

test: {header_value}

- - """ - - header_block_error = """ - -

placeholder

-

Error!

- - """ - if header_value.endswith("'") and not header_value.endswith("''"): - return Response(header_block_error, status=500) - return Response(header_block_normal, status=200) - return Response(placeholder_block, status=200) + def request_handler(self, request): + return sqli_injectable_response(request.cookies.get("test"), self.placeholder_block) def check(self, module_test, events): sqli_finding_emitted = False @@ -1815,6 +1754,36 @@ def check(self, module_test, events): assert no_finding_emitted, "False positive finding was emitted" +# Serialization Module: the parameter is parsed as a URI, not deserialized. .NET's Uri reads +# everything before the first "/" as the hostname, so the control payload (no "/", trailing "=") +# throws and the dotnet payload (leading token "AAEAAAD") parses -- a 500->200 flip that has +# nothing to do with deserialization. A structurally-corrupted twin parses just as well, which +# is what proves the outcome is independent of the payload's content. +class Test_Lightfuzz_serial_errorresolution_uri_parse_fp(Test_Lightfuzz_serial_errorresolution): + uri_parse_error = ( + "System.UriFormatException: Invalid URI: The hostname could not be parsed." + ) + + def request_handler(self, request): + post_params = request.form + if "TextBox1" not in post_params.keys(): + return Response(self.dotnet_serial_html, status=200) + if post_params["__VIEWSTATE"] != "/wEPDwULLTE5MTI4MzkxNjVkZNt7ICM+GixNryV6ucx+srzhXlwP": + return Response(self.dotnet_serial_error, status=500) + host = post_params["TextBox1"].split("/")[0] + if not host or re.fullmatch(r"[A-Za-z0-9.-]+", host) is None: + return Response(self.uri_parse_error, status=500) + return Response("Request accepted", status=200) + + def check(self, module_test, events): + findings = [ + e.data["description"] + for e in events + if e.type == "FINDING" and "Unsafe Deserialization" in e.data.get("description", "") + ] + assert not findings, f"URI parsing was misreported as unsafe deserialization: {findings}" + + class Test_Lightfuzz_serial_errorresolution_existingvalue_valid(Test_Lightfuzz_serial_errorresolution): dotnet_serial_html = """ @@ -3789,10 +3758,7 @@ class Test_Lightfuzz_try_post_as_get(ModuleTestBase): }, } - def request_handler(self, request): - qs = str(request.query_string.decode()) - - parameter_block = """ + parameter_block = """ """ - if "search=" in qs: - value = qs.split("=")[1] - if "&" in value: - value = value.split("&")[0] - - sql_block_normal = f""" -
-

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

-
-
- """ - - sql_block_error = """ -
-

Found error in SQL query

-
-
- """ - if value.endswith("'"): - if value.endswith("''"): - return Response(sql_block_normal, status=200) - return Response(sql_block_error, status=500) - return Response(parameter_block, status=200) + def request_handler(self, request): + # only the converted GETPARAM pass reaches a query here; the POST pass is disabled + return sqli_injectable_response(request.args.get("search"), self.parameter_block) async def setup_after_prep(self, module_test): module_test.scan.modules["lightfuzz"].helpers.rand_string = lambda *args, **kwargs: ( @@ -3871,8 +3817,7 @@ class Test_Lightfuzz_try_get_as_post(ModuleTestBase): }, } - def request_handler(self, request): - parameter_block = """ + parameter_block = """ """ - if request.method == "POST" and "search" in request.form.keys(): - value = request.form["search"] - - sql_block_normal = f""" -
-

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

-
-
- """ - - sql_block_error = """ -
-

Found error in SQL query

-
-
- """ - if value.endswith("'"): - if value.endswith("''"): - return Response(sql_block_normal, status=200) - return Response(sql_block_error, status=500) - return Response(parameter_block, status=200) + def request_handler(self, request): + # only the converted POSTPARAM pass reaches a query here + value = request.form.get("search") if request.method == "POST" else None + return sqli_injectable_response(value, self.parameter_block) async def setup_after_prep(self, module_test): module_test.scan.modules["lightfuzz"].helpers.rand_string = lambda *args, **kwargs: ( @@ -4147,6 +4075,66 @@ def check(self, module_test, events): ) +def _sqli_code_change_findings(events): + return [ + e.data["description"] for e in events if e.type == "FINDING" and "Code Change" in e.data.get("description", "") + ] + + +# SQLi negative test: the textbook 200->500->200 status flip on a parameter that never reaches a +# query. TRUE and FALSE boolean payloads come back byte-identical, so there is no SQL logic to +# confirm. This is the shape an envelope-wrapped parameter produces when the injected quote +# changes the envelope's structural validity rather than a query's. +class Test_Lightfuzz_sqli_no_boolean_differential_fp(Test_Lightfuzz_sqli): + def request_handler(self, request): + value = request.args.get("search") + if value is None: + return Response(self.parameter_block, status=200) + if value.endswith("'") and not value.endswith("''"): + return Response("

Bad Request

", status=500) + return Response("

0 results

", status=200) + + def check(self, module_test, events): + findings = _sqli_code_change_findings(events) + assert not findings, f"SQLi reported from a status flip with no boolean differential: {findings}" + + +# SQLi negative test: the single-quote probe is blocked by an access-control layer whose block +# page carries no recognizable WAF signature. The transition into 403 is what identifies it. +class Test_Lightfuzz_sqli_unsigned_403_fp(Test_Lightfuzz_sqli): + def request_handler(self, request): + value = request.args.get("search") + if value is None: + return Response(self.parameter_block, status=200) + if value.endswith("'") and not value.endswith("''"): + return Response( + "Forbidden

Forbidden

" + "

Reference: 0aB1cD2e

", + status=403, + ) + return Response("

0 results

", status=200) + + def check(self, module_test, events): + findings = _sqli_code_change_findings(events) + assert not findings, f"SQLi reported from a probe blocked with an unsigned 403: {findings}" + + +# SQLi negative test: the status is keyed on the payload's length, not its quoting. A benign +# one-vs-two-character pair reproduces the same status triplet, so the change isn't quote-specific. +class Test_Lightfuzz_sqli_length_keyed_fp(Test_Lightfuzz_sqli): + def request_handler(self, request): + value = request.args.get("search") + if value is None: + return Response(self.parameter_block, status=200) + if len(value) == 11: + return Response("

Bad Request

", status=500) + return Response("

0 results

", status=200) + + def check(self, module_test, events): + findings = _sqli_code_change_findings(events) + assert not findings, f"SQLi reported from a length-keyed status flip: {findings}" + + # 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"] @@ -4160,8 +4148,7 @@ class Test_Lightfuzz_sqli_post_additional_params(ModuleTestBase): }, } - def request_handler(self, request): - parameter_block = """ + 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) + + def request_handler(self, request): + return sqli_injectable_response(request.form.get("search"), self.parameter_block) async def setup_after_prep(self, module_test): module_test.scan.modules["lightfuzz"].helpers.rand_string = lambda *args, **kwargs: ( @@ -5027,6 +5009,20 @@ def test_keystream_fp_sibling_form_fields_incremental(): assert not c.results, f"FP on sibling sequential hex form fields: {c.results}" +def test_keystream_fp_record_guid_shared_template(): + """Two 32-hex record GUIDs from the same platform instance. They share no leading byte, so + the leading-zero test never applies, but they re-converge at fixed offsets where the shared + instance/table segment sits -- and the XOR of the diverging bytes lands almost entirely in + the ASCII-XOR range by coincidence, clearing the 0.9 score on its own.""" + ids = [ + "e42a5af4c700201072b211d4d8c2607c", + "c86a62e2c7022010099a308dc7c26022", + ] + c = _make_crypto_for_keystream(ids[0], additional_params={"app_sys_id": ids[1]}) + c.detect_keystream_reuse(ids[0]) + assert not c.results, f"FP on record GUIDs sharing a field template: {c.results}" + + def test_keystream_fp_decimal_account_numbers(): """Digit-only parameter values (account numbers, zip codes, etc.) are valid hex but are plain decimal IDs. Their decoded bytes XOR to small values that diff --git a/bbot/test/test_step_2/module_tests/test_module_wayback.py b/bbot/test/test_step_2/module_tests/test_module_wayback.py index b37d0d4d57..5c8ce484e4 100644 --- a/bbot/test/test_step_2/module_tests/test_module_wayback.py +++ b/bbot/test/test_step_2/module_tests/test_module_wayback.py @@ -133,6 +133,17 @@ def check(self, module_test, events): assert "web.archive.org" not in e.data["url"], ( f"FINDING url should NOT be an archive.org URL, got: {e.data['url']}" ) + # the evidence is a snapshot, not the live host, and that has to be visible + # without reading discovery_path + assert e.archive_url, f"FINDING from archived content has no archive_url: {e.tags}" + assert e.pretty_string.startswith("[ARCHIVED] "), ( + f"FINDING from archived content is not marked as archived: {e.pretty_string}" + ) + # the snapshot URL has to survive into output.txt and output.json + assert e.data_human.endswith(f"(archived: {e.archive_url})"), ( + f"output.txt is missing the archive URL: {e.data_human}" + ) + assert e.json()["archive_url"] == e.archive_url # web.archive.org should NOT appear as a DNS_NAME event assert not any(e.type == "DNS_NAME" and e.data == "web.archive.org" for e in events), ( "web.archive.org should not leak as a DNS_NAME event"