Merge origin/nowafpls (#3336) into bleeding-edge

Author: liquidsec

bbot/core/helpers/diff.py

@@ -373,6 +373,7 @@ async def compare(
         json=None,
         allow_redirects=False,
         timeout=None,
+        none_is_match=True,
     ):
         """
         Compares a URL with the baseline, with optional headers or cookies added
@@ -381,6 +382,8 @@ async def compare(
             where "match" is whether the content matched against the baseline, and
                 "reason" is the location of the change ("code", "body", "header", or None), and
                 "reflection" is whether the value was reflected in the HTTP response
+
+        When the request fails outright, "match" is `none_is_match` and subject_response is None.
         """
 
         await self._baseline()
@@ -406,8 +409,10 @@ async def compare(
         )
 
         if subject_response is None:
-            # this can be caused by a WAF not liking the header, so we really aren't interested in it
-            return (True, "403", reflection, subject_response)
+            # A dead request usually just means the probe wasn't interesting (a WAF not liking a
+            # fuzzed header, etc). Consumers that need to read it as interference, or as "unknown"
+            # rather than a verdict, pass none_is_match=False.
+            return (bool(none_is_match), ["request_failed"], reflection, None)
 
         if check_reflection:
             for arg in (headers, cookies):

bbot/core/helpers/nowafpls.py

@@ -41,6 +41,16 @@ class BypassResult:
     def bypassed(self) -> bool:
         return self.status == self.STATUS_BYPASSED
 
+    @property
+    def summary(self) -> str:
+        """One-line verdict for logging, shared by every consumer of the probe."""
+        parts = [f"status={self.status}", f"provider={self.waf_provider or 'unknown'}"]
+        if self.error:
+            parts.append(f"error={self.error}")
+        if self.diff_reasons:
+            parts.append(f"diff_reasons={','.join(str(r) for r in self.diff_reasons)}")
+        return " ".join(parts)
+
 
 class NowafplsHelper:
     """
@@ -55,6 +65,9 @@ class NowafplsHelper:
       * unpadded differs, padded matches    -> bypass works
       * both differ from baseline           -> gate held; padding did not help
 
+    A request the WAF kills outright (timeout / connection reset) counts as "differs",
+    not as a match, so a dropped connection reads as interference rather than acceptance.
+
     Results are memoized per host for the scan's lifetime. Concurrent callers hit
     the same in-flight `asyncio.Task`, so exactly one probe runs per host.
     """
@@ -105,6 +118,7 @@ async def pad_json(self, event, data):
     async def _probe(self, event, padding_size: int, payload: str) -> BypassResult:
         url = event.url
         provider = self._identify_provider(event)
+        log.debug(f"nowafpls: probing {url} with {padding_size} bytes of padding")
         encoded_payload = _urlquote(payload, safe="")
         benign_body = "q=hello"
         unpadded_body = f"q={encoded_payload}"
@@ -121,7 +135,7 @@ async def _probe(self, event, padding_size: int, payload: str) -> BypassResult:
 
         try:
             match_unpadded, reasons_unpadded, *_ = await compare.compare(
-                url, method="POST", data=unpadded_body, headers=headers
+                url, method="POST", data=unpadded_body, headers=headers, none_is_match=False
             )
         except HttpCompareError as e:
             return BypassResult(
@@ -142,7 +156,7 @@ async def _probe(self, event, padding_size: int, payload: str) -> BypassResult:
 
         try:
             match_padded, reasons_padded, *_ = await compare.compare(
-                url, method="POST", data=padded_body, headers=headers
+                url, method="POST", data=padded_body, headers=headers, none_is_match=False
             )
         except HttpCompareError as e:
             return BypassResult(

bbot/core/helpers/web/web.py

@@ -723,10 +723,14 @@ async def _probe_wildcard_host(self, scheme, host, port):
             return "retry"
         root_url = f"{scheme}://{host}:{port}/"
         try:
-            root_match, root_reasons, _, _ = await compare.compare(root_url)
+            root_match, root_reasons, _, root_response = await compare.compare(root_url, none_is_match=False)
         except HttpCompareError as e:
             log.debug(f"is_http_wildcard_host: root probe failed for {host}:{port}: {e}")
             return "retry"
+        if root_response is None:
+            # a dead probe is unknown, not a wildcard verdict
+            log.debug(f"is_http_wildcard_host: root probe to {host}:{port} returned no response")
+            return "retry"
         if not root_match:
             log.debug(
                 f"is_http_wildcard_host: {host}:{port} root distinct from random-path baseline ({root_reasons}); not a wildcard"

bbot/modules/base.py

@@ -980,9 +980,12 @@ async def _is_http_wildcard_host(self, event):
                     target_url, {name: value if value is not None else ""}
                 ).geturl()
         try:
-            match, _, _, _ = await result.compare(target_url)
+            match, _, _, subject_response = await result.compare(target_url, none_is_match=False)
         except HttpCompareError:
             return None
+        if subject_response is None:
+            # a dead probe is unknown, not a wildcard verdict
+            return None
         return match
 
     def _scope_distance_check(self, event):

bbot/modules/lightfuzz/lightfuzz.py

@@ -10,6 +10,7 @@
 
 from bbot.core.config.models import BaseModuleConfig, Field
 from bbot.core.helpers.misc import get_waf_strings
+from bbot.core.helpers.nowafpls import BypassResult
 from bbot.core.helpers.web.response_event import response_to_event_dict
 from bbot.modules.lightfuzz.submodules.base import BaseLightfuzz
 from bbot.modules.lightfuzz.submodules.serial import serial as _serial_submodule
@@ -445,16 +446,18 @@ async def filter_event(self, event):
                 self.debug(f"Skipping {event.type} (avoid_wafs=always). URL: {url}")
                 return False
             if self.avoid_wafs == "try_bypasses":
-                # Ask nowafpls if body padding gets us through. Only worth asking on events
-                # that can fire a POST probe — GET/COOKIE/HEADER can't be padded.
-                if not self._post_capable(event):
-                    return False, "WAF-tagged event has no POST-style probe to pad"
+                # The "waf" tag comes from the CDN/WAF provider's identity, not from observed
+                # blocking, so ask nowafpls what the host actually does with a payload.
                 result = await self.helpers.nowafpls.is_bypassable(event)
-                if not result.bypassed:
+                if result.status in (BypassResult.STATUS_BLOCKED, BypassResult.STATUS_ERROR):
                     parsed_url = getattr(event, "parsed_url", None)
                     url = parsed_url.geturl() if parsed_url else "unknown"
-                    self.debug(f"Skipping {event.type} because WAF is not bypassable. URL: {url}")
+                    self.debug(f"Skipping {event.type} ({result.summary}). URL: {url}")
                     return False
+                if result.status == BypassResult.STATUS_BYPASSED and not self._post_capable(event):
+                    # padding is body-only, so a GET/COOKIE/HEADER probe has no bypass to apply
+                    return False, "WAF is bypassable via body padding, but this event has no POST-style probe"
+                # STATUS_NO_INTERFERENCE: nothing is gating the payload, so fuzz normally
             # avoid_wafs == "never": fall through and fuzz raw
 
         # Skip WEB_PARAMETERs on static-asset URLs (pdf, doc, xml, etc.) — fuzzing them is pointless

bbot/modules/nowafpls.py

@@ -1,6 +1,6 @@
 from bbot.modules.base import BaseModule
 from bbot.core.config.models import BaseModuleConfig, Field
-from bbot.core.helpers.nowafpls import BypassResult, DEFAULT_PADDING_SIZE, DEFAULT_PAYLOAD
+from bbot.core.helpers.nowafpls import DEFAULT_PADDING_SIZE, DEFAULT_PAYLOAD
 
 
 class nowafpls(BaseModule):
@@ -40,10 +40,7 @@ async def handle_event(self, event):
             payload=self.config.get("payload") or DEFAULT_PAYLOAD,
         )
         if not result.bypassed:
-            self.debug(
-                f"No bypass finding for {event.url}: status={result.status} "
-                f"provider={result.waf_provider or 'unknown'}"
-            )
+            self.info(f"No bypass finding for {event.url}: {result.summary}")
             return
 
         provider = result.waf_provider or "WAF/inspection layer"
@@ -67,7 +64,3 @@ async def handle_event(self, event):
                 f"indicating the payload reached the application past the inspection layer."
             ),
         )
-
-    # expose the status constants for consumers/tests that want to reason about
-    # the helper's verdict without importing the helper module directly
-    RESULT = BypassResult

bbot/test/test_step_1/test_web.py

@@ -1,4 +1,5 @@
 import re
+from types import SimpleNamespace
 
 from blasthttp import HTTPStatusError
 
@@ -825,8 +826,11 @@ async def compare(self, url, **kwargs):
             self.calls.append(url)
             if self.behaviour == "raise":
                 raise HttpCompareError("boom")
+            if self.behaviour == "dead":
+                # compare() only hands back a None response when the request itself failed
+                return (False, ["request_failed"], False, None)
             match = self.behaviour == "match"
-            return (match, [], False, None)
+            return (match, [], False, SimpleNamespace(status_code=200))
 
     fake_compare = FakeCompare()
 
@@ -878,21 +882,27 @@ def make_web_param_event(url, name, value, ptype="GETPARAM"):
     assert result is False
     assert fake_compare.calls == ["https://wildcardhost.test/index.php"]
 
-    # 5) HttpCompareError from the compare -> None
+    # 5) probe request died: unknown, not a wildcard verdict
+    fake_compare.behaviour = "dead"
+    fake_compare.calls.clear()
+    result = await module._is_http_wildcard_host(make_url_event("https://wildcardhost.test/dead.php"))
+    assert result is None
+
+    # 6) HttpCompareError from the compare -> None
     fake_compare.behaviour = "raise"
     fake_compare.calls.clear()
     result = await module._is_http_wildcard_host(make_url_event("https://wildcardhost.test/broken.php"))
     assert result is None
 
-    # 6) Scalar True from a test mock (no .compare attribute) -> True (backward-compat)
+    # 7) Scalar True from a test mock (no .compare attribute) -> True (backward-compat)
     async def wildcard_returns_true(scheme, host, port):
         return True
 
     scan.helpers.web.is_http_wildcard_host = wildcard_returns_true
     result = await module._is_http_wildcard_host(make_url_event("https://wildcardhost.test/whatever"))
     assert result is True
 
-    # 7) False / None from the helper pass through unchanged
+    # 8) False / None from the helper pass through unchanged
     async def wildcard_returns_false(scheme, host, port):
         return False
 

bbot/test/test_step_2/module_tests/test_module_lightfuzz.py

@@ -13,6 +13,7 @@
 
 from bbot.core.helpers.url import add_get_params
 from bbot.modules.base import BaseModule
+from bbot.core.helpers.nowafpls import BypassResult
 from bbot.modules.lightfuzz.lightfuzz import lightfuzz
 from bbot.modules.lightfuzz.submodules.base import BaseLightfuzz
 
@@ -3823,6 +3824,95 @@ def check(self, module_test, events):
         pass
 
 
+class Test_Lightfuzz_filter_event_try_bypasses(ModuleTestBase):
+    """Under try_bypasses the nowafpls verdict decides, not the "waf" tag. The tag reflects the
+    CDN/WAF provider's identity, so a host that isn't gating payloads must still be fuzzed."""
+
+    targets = [HTTPSERVER_URL]
+    modules_overrides = ["http", "lightfuzz"]
+    config_overrides = {
+        "interactsh_disable": True,
+        "modules": {
+            "lightfuzz": {
+                "enabled_submodules": ["xss"],
+                "avoid_wafs": "try_bypasses",
+            }
+        },
+    }
+
+    def _web_param(self, module_test, param_type):
+        return module_test.scan.make_event(
+            {
+                "host": "127.0.0.1",
+                "type": param_type,
+                "name": "test",
+                "original_value": "value",
+                "url": f"{HTTPSERVER_URL}/",
+                "description": "Test parameter",
+            },
+            "WEB_PARAMETER",
+            module_test.scan.root_event,
+            module="excavate",
+            tags=["distance-0", "waf"],
+        )
+
+    async def setup_after_prep(self, module_test):
+        self.url_event = module_test.scan.make_event(
+            f"{HTTPSERVER_URL}/",
+            "URL",
+            module_test.scan.root_event,
+            module="http",
+            tags=["status-200", "distance-0", "waf"],
+        )
+        self.getparam_event = self._web_param(module_test, "GETPARAM")
+        self.postparam_event = self._web_param(module_test, "POSTPARAM")
+
+    @staticmethod
+    def _accepted(result):
+        # filter_event returns True to accept, or False / (False, reason) to reject
+        return result is True
+
+    def _set_verdict(self, module_test, status):
+        async def _stub(event, *args, **kwargs):
+            return BypassResult(status=status)
+
+        module_test.scan.helpers.nowafpls.is_bypassable = _stub
+
+    async def test_filter_event(self, module_test):
+        module = module_test.scan.modules["lightfuzz"]
+        all_events = (self.url_event, self.getparam_event, self.postparam_event)
+
+        # nothing is gating the payload, so there is no WAF to work around: fuzz every event type
+        self._set_verdict(module_test, BypassResult.STATUS_NO_INTERFERENCE)
+        for event in all_events:
+            result = await module.filter_event(event)
+            assert self._accepted(result), (
+                f"{event.type} should be fuzzed when the probe reports no interference, got {result}"
+            )
+
+        # padding is body-only, so a confirmed bypass only helps events that can fire a POST probe
+        self._set_verdict(module_test, BypassResult.STATUS_BYPASSED)
+        assert self._accepted(await module.filter_event(self.postparam_event)), (
+            "POSTPARAM should be accepted when the WAF is bypassable via body padding"
+        )
+        for event in (self.url_event, self.getparam_event):
+            result = await module.filter_event(event)
+            assert not self._accepted(result), (
+                f"{event.type} has no POST-style probe to pad and should be rejected, got {result}"
+            )
+
+        # the gate held, or we never got a verdict: reject everything
+        for status in (BypassResult.STATUS_BLOCKED, BypassResult.STATUS_ERROR):
+            self._set_verdict(module_test, status)
+            for event in all_events:
+                result = await module.filter_event(event)
+                assert not self._accepted(result), f"{event.type} should be rejected on status={status}, got {result}"
+
+    def check(self, module_test, events):
+        # assertions live in test_filter_event
+        pass
+
+
 class _NowafplsFuzzTestBase(ModuleTestBase):
     """Shared setup: dummy module emits a WAF-tagged POSTPARAM WEB_PARAMETER, and the mocked
     endpoint's callback records every POST body so tests can assert on what actually fired.
@@ -3833,6 +3923,8 @@ class _NowafplsFuzzTestBase(ModuleTestBase):
 
     targets = ["nowafpls-fuzz.test"]
     bypass_works = True
+    # when False the endpoint answers the malicious payload normally, i.e. nothing is gating it
+    waf_blocks = True
     avoid_wafs = "try_bypasses"
 
     class DummyModule(BaseModule):
@@ -3890,6 +3982,7 @@ async def setup_after_prep(self, module_test):
         await module_test.mock_dns({"nowafpls-fuzz.test": {"A": ["127.0.0.1"]}})
         self.post_bodies: list[bytes] = []
         bypass = self.bypass_works
+        blocks = self.waf_blocks
 
         def cb(request):
             body = request.content or b""
@@ -3902,7 +3995,7 @@ def cb(request):
             # Padded malicious is accepted iff bypass_works.
             has_pad = body.startswith(b"__nowafpls_pad=")
             has_malicious = b"%3Cscript" in body or b"<script" in body
-            if has_malicious and not has_pad:
+            if has_malicious and not has_pad and blocks:
                 return MockResponse(status_code=403, text="Attention Required! | Cloudflare\nRay ID: abcd")
             if has_malicious and has_pad and not bypass:
                 return MockResponse(status_code=403, text="Attention Required! | Cloudflare\nRay ID: abcd")
@@ -3953,6 +4046,31 @@ def check(self, module_test, events):
         )
 
 
+class Test_Nowafpls_try_bypasses_no_interference(_NowafplsFuzzTestBase):
+    """try_bypasses + the host isn't gating the payload: the probe reports no interference, so
+    lightfuzz fuzzes normally and unpadded. A verdict of "not bypassed" must not be read as
+    "skip this host" -- only an observed block should suppress fuzzing."""
+
+    waf_blocks = False
+    avoid_wafs = "try_bypasses"
+
+    # the only POST bodies nowafpls's own probe ever sends: two benign baselines and one
+    # unpadded malicious payload (it returns before testing padding when there's no interference)
+    PROBE_BODIES = {b"q=hello", b"q=%3Cscript%3Ealert%281%29%3C%2Fscript%3E"}
+
+    def check(self, module_test, events):
+        fuzz_bodies = [b for b in self.post_bodies if b not in self.PROBE_BODIES]
+        padded = [b for b in self.post_bodies if b.startswith(b"__nowafpls_pad=")]
+        # anything the probe didn't send is lightfuzz, which only reaches the wire if
+        # filter_event accepted the WEB_PARAMETER
+        assert fuzz_bodies, (
+            f"Expected lightfuzz to fuzz a host with no interference, but the only POST traffic "
+            f"was nowafpls's own probe: {self.post_bodies[:5]}"
+        )
+        # nothing is gating the payload, so there is nothing to pad around
+        assert not padded, f"No padding should be applied when nothing is gating the payload. Got: {padded[:3]}"
+
+
 class Test_Nowafpls_never_still_pads(_NowafplsFuzzTestBase):
     """avoid_wafs=never: no filter-time probe, but prepare_request still opportunistically pads POST
     when the helper reports bypassable. Padded fuzz bodies should still land on the wire."""