fix(scanner): reject URLs whose path repeats a path parameter

Servers that append a session id to the path on every redirect emit an
unbounded supply of URLs that dedup treats as distinct. Ingress now drops them
past url_max_path_param_repeats (default 10). Only ";key=value" path
parameters count; ordinary path segments do not, since a deep path that
repeats a directory name is finite and depth is bounded by web_spider_depth.

Author: liquidsec

bbot/core/config/models.py

@@ -395,6 +395,7 @@ class BBOTConfig(BaseModel):
     # URL handling
     url_querystring_remove: Optional[bool] = None
     url_querystring_collapse: Optional[bool] = None
+    url_max_path_param_repeats: Optional[int] = None
     url_extension_blacklist: Optional[list[str]] = None
     url_extension_special: Optional[list[str]] = None
     url_extension_static: Optional[list[str]] = None

bbot/core/helpers/url.py

@@ -1,5 +1,6 @@
 import uuid
 import logging
+from collections import Counter
 from contextlib import suppress
 from urllib.parse import urlparse, parse_qs, urlencode, ParseResult
 
@@ -211,6 +212,46 @@ def hash_url(url):
     return hash(tuple(to_hash))
 
 
+def max_path_param_repeats(url):
+    """
+    Count how many times the most-repeated path parameter appears in a URL's path.
+
+    Path (matrix) parameters are the ";key=value" pieces a server may append to a path
+    segment; some reissue one on every redirect, so the path grows without bound. Keys are
+    compared case-insensitively. Ordinary path segments are deliberately not counted: a deep
+    path that repeats a directory name is finite, and depth is bounded by web_spider_depth.
+
+    Args:
+        url (Union[str, ParseResult]): The URL to inspect.
+
+    Returns:
+        int: The highest number of times any single path parameter appears.
+
+    Examples:
+        >>> max_path_param_repeats('https://www.evilcorp.com/foo/bar')
+        0
+
+        >>> max_path_param_repeats('https://www.evilcorp.com/;JSESSIONID=a;JSESSIONID=b')
+        2
+
+        >>> max_path_param_repeats('https://www.evilcorp.com/app;a=1/b;c=2')
+        1
+    """
+    parsed = url if hasattr(url, "path") else parse_url(url)
+    # urlparse peels the final segment's matrix parameters off into .params
+    path = parsed.path
+    trailing_params = getattr(parsed, "params", "")
+    if trailing_params:
+        path = f"{path};{trailing_params}"
+    counts = Counter()
+    for segment in path.split("/"):
+        # the first ";"-delimited piece is the segment itself, not a parameter
+        for token in segment.split(";")[1:]:
+            if token:
+                counts[token.split("=", 1)[0].lower()] += 1
+    return max(counts.values(), default=0)
+
+
 def url_depth(url):
     """
     Calculate the depth of the given URL based on its path components.

bbot/defaults.yml

@@ -191,6 +191,8 @@ python: True
 url_querystring_remove: True
 # When query string is retained, by default collapse parameter values down to a single value per parameter
 url_querystring_collapse: True
+# Reject URLs whose path repeats the same path parameter this many times (e.g. a session id appended on every redirect)
+url_max_path_param_repeats: 10
 
 # Completely ignore URLs with these extensions
 url_extension_blacklist:

bbot/scanner/manager.py

@@ -2,6 +2,7 @@
 from contextlib import suppress
 from radixtarget import host_size_key
 
+from bbot.core.helpers.url import max_path_param_repeats
 from bbot.modules.base import BaseInterceptModule
 
 
@@ -104,6 +105,17 @@ async def handle_event(self, event, **kwargs):
                 )
                 event.add_tag("blacklisted")
 
+        # reject degenerate URL paths, which servers generate in unbounded numbers
+        parsed_url = getattr(event, "parsed_url", None)
+        if parsed_url is not None and self.scan.url_max_path_param_repeats > 0:
+            repeats = max_path_param_repeats(parsed_url)
+            if repeats >= self.scan.url_max_path_param_repeats:
+                self.debug(
+                    f"Blacklisting {event} because its path repeats a parameter {repeats:,} times "
+                    f"(url_max_path_param_repeats={self.scan.url_max_path_param_repeats:,})"
+                )
+                event.add_tag("blacklisted")
+
         # main scan blacklist
         host_filterable = getattr(event, "host_filterable", None)
         event_blacklisted = False

bbot/scanner/scanner.py

@@ -285,6 +285,9 @@ def __init__(
         # url querystring behavior
         self.url_querystring_remove = self.config.get("url_querystring_remove", True)
 
+        # reject URLs whose path repeats the same path parameter this many times
+        self.url_max_path_param_repeats = self.config.get("url_max_path_param_repeats", 10)
+
         # blob inclusion
         self._file_blobs = self.config.get("file_blobs", False)
         self._folder_blobs = self.config.get("folder_blobs", False)

bbot/test/test_step_1/test_helpers.py

@@ -45,6 +45,18 @@ async def test_helpers_misc(helpers, scan, bbot_scanner, bbot_httpserver):
     with pytest.raises(ValueError):
         helpers.validators.clean_url("http://evilcorp,com")
 
+    # path parameters accumulate inside a path segment, and are keyed by name
+    assert helpers.max_path_param_repeats("http://evilcorp.com/;JSESSIONID=a") == 1
+    assert helpers.max_path_param_repeats("http://evilcorp.com/;JSESSIONID=a;JSESSIONID=b;jsessionid=c") == 3
+    assert helpers.max_path_param_repeats("http://evilcorp.com/app;a=1;b=2") == 1
+    assert helpers.max_path_param_repeats("http://evilcorp.com/app;a=1/b;c=2") == 1
+    # ordinary path segments are not parameters, however deep or repetitive the path
+    assert helpers.max_path_param_repeats("http://evilcorp.com/") == 0
+    assert helpers.max_path_param_repeats("http://evilcorp.com/foo/bar") == 0
+    assert helpers.max_path_param_repeats("http://evilcorp.com/" + "/".join(["a"] * 30)) == 0
+    # the query string is not part of the path
+    assert helpers.max_path_param_repeats("http://evilcorp.com/foo?x=1&x=2&x=3") == 0
+
     assert helpers.url_depth("http://evilcorp.com/asdf/user/") == 2
     assert helpers.url_depth("http://evilcorp.com/asdf/user") == 2
     assert helpers.url_depth("http://evilcorp.com/asdf/") == 1

bbot/test/test_step_1/test_scan.py

@@ -520,3 +520,47 @@ def mock_memory_status():
         assert m.num_incoming_events == 0
     scan.modules_status(_log=False)
     assert scan._ingress_delay == 0.0, "throttle must clear in drain mode (no producers, ingress is the drain)"
+
+
+@pytest.mark.asyncio
+async def test_url_max_path_param_repeats():
+    """Servers that append a session id on every redirect, and spider traps, generate an
+    unbounded supply of URLs whose path repeats one key. Ingress drops them."""
+    from bbot.scanner import Scanner
+
+    def jsessionid_url(n):
+        return "http://evilcorp.com/" + "".join(f";JSESSIONID={i:08x}-2ab9-49eb" for i in range(n))
+
+    scan = Scanner("evilcorp.com", config={"url_max_path_param_repeats": 10})
+    await scan._prep()
+    ingress = scan.ingress_module
+
+    # handle_event returns None when it accepts, and (False, reason) when it rejects
+
+    # below the limit the URL is scanned normally
+    shallow = scan.make_event(jsessionid_url(9), "URL_UNVERIFIED", parent=scan.root_event)
+    assert await ingress.handle_event(shallow) is None
+    assert "blacklisted" not in shallow.tags
+
+    # at the limit it is rejected before reaching any module's queue
+    degenerate = scan.make_event(jsessionid_url(10), "URL_UNVERIFIED", parent=scan.root_event)
+    assert await ingress.handle_event(degenerate) == (False, "event is blacklisted")
+    assert "blacklisted" in degenerate.tags
+
+    # ordinary URLs are untouched, however deep or repetitive the path
+    for url in (
+        "http://evilcorp.com/api/v1/users/1",
+        "http://evilcorp.com/" + "/".join(["blacklanternsecurity"] * 20) + ".txt",
+    ):
+        normal = scan.make_event(url, "URL_UNVERIFIED", parent=scan.root_event)
+        assert await ingress.handle_event(normal) is None
+        assert "blacklisted" not in normal.tags
+    await scan._cleanup()
+
+    # the guard can be disabled outright
+    scan = Scanner("evilcorp.com", config={"url_max_path_param_repeats": 0})
+    await scan._prep()
+    off = scan.make_event(jsessionid_url(50), "URL_UNVERIFIED", parent=scan.root_event)
+    await scan.ingress_module.handle_event(off)
+    assert "blacklisted" not in off.tags
+    await scan._cleanup()