From 7206d6302cc719aec3bd91959e8adf995258633a Mon Sep 17 00:00:00 2001 From: Daniel Iulian Iacob Date: Fri, 23 Oct 2026 11:24:12 +0200 Subject: [PATCH] Skip wildcard routes that do not compile --- aikido_zen/helpers/match_endpoints.py | 19 +++++++-- aikido_zen/helpers/match_endpoints_test.py | 47 ++++++++++++++++++++++ aikido_zen/vulnerabilities/init_test.py | 14 +++++++ 3 files changed, 76 insertions(+), 4 deletions(-) diff --git a/aikido_zen/helpers/match_endpoints.py b/aikido_zen/helpers/match_endpoints.py index df1a3b708..fb1c6eb82 100644 --- a/aikido_zen/helpers/match_endpoints.py +++ b/aikido_zen/helpers/match_endpoints.py @@ -2,10 +2,23 @@ Helper function file, see function docstring """ +from functools import lru_cache + import regex as re +from aikido_zen.helpers.logging import logger from .try_parse_url_path import try_parse_url_path +@lru_cache +def compile_wildcard_route(route): + try: + return re.compile(f"^{route.replace('*', '(.*)')}/?$", re.IGNORECASE) + except re.error as e: + # Raising here would turn off protection for every request. + logger.warning("Route regex is not compiling: %s (%s)", route, e) + return None + + def match_endpoints(route_metadata, endpoints): """ Based on the context's url this tries to find a match in the list of endpoints @@ -44,10 +57,8 @@ def match_endpoints(route_metadata, endpoints): ) for wildcard in wildcards: - route = wildcard["route"] - regex = re.compile(f"^{route.replace('*', '(.*)')}/?$", re.IGNORECASE) - - if regex.match(path): + regex = compile_wildcard_route(wildcard["route"]) + if regex and regex.match(path): results.append(wildcard) if len(results) > 0: return results diff --git a/aikido_zen/helpers/match_endpoints_test.py b/aikido_zen/helpers/match_endpoints_test.py index 0c72a200c..6097a4225 100644 --- a/aikido_zen/helpers/match_endpoints_test.py +++ b/aikido_zen/helpers/match_endpoints_test.py @@ -345,3 +345,50 @@ def test_it_prefers_specific_method_over_wildcard_second_case(): } assert result[0] == expected + + +@pytest.mark.parametrize("route", ["/api/[/*", "/api/(/*", "/api/v1+?+/*"]) +def test_invalid_wildcard_route_is_skipped(route): + endpoints = [ + {"method": "POST", "route": route, "forceProtectionOff": True}, + {"method": "POST", "route": "/api/*", "forceProtectionOff": False}, + ] + assert match_endpoints( + sample_route_metadata(route=None, url="/api/[/x"), endpoints + ) == [endpoints[1]] + + +def test_wildcard_route_keeps_regex_characters_that_compile(): + endpoints = [{"method": "POST", "route": "/v1.0/*", "forceProtectionOff": False}] + for url in ["/v1.0/abc", "/v1x0/abc"]: + assert ( + match_endpoints(sample_route_metadata(route=None, url=url), endpoints) + == endpoints + ) + + +def test_wildcard_route_is_compiled_once(monkeypatch): + endpoints = [ + {"method": "POST", "route": "/compiled-once/*", "forceProtectionOff": False} + ] + route_metadata = sample_route_metadata(route=None, url="/compiled-once/x") + assert match_endpoints(route_metadata, endpoints) == endpoints + + def fail_to_compile(*args): + raise AssertionError("compiled the route again") + + monkeypatch.setattr( + "aikido_zen.helpers.match_endpoints.re.compile", fail_to_compile + ) + for _ in range(3): + assert match_endpoints(route_metadata, endpoints) == endpoints + + +def test_invalid_wildcard_route_is_logged_once(caplog): + endpoints = [{"method": "POST", "route": "/logged-once/[/*"}] + route_metadata = sample_route_metadata(route=None, url="/logged-once/x") + for _ in range(3): + assert match_endpoints(route_metadata, endpoints) is None + warnings = [r for r in caplog.records if "/logged-once/[/*" in r.getMessage()] + assert len(warnings) == 1 + assert warnings[0].levelname == "WARNING" diff --git a/aikido_zen/vulnerabilities/init_test.py b/aikido_zen/vulnerabilities/init_test.py index a24c5a4d4..0f28e2137 100644 --- a/aikido_zen/vulnerabilities/init_test.py +++ b/aikido_zen/vulnerabilities/init_test.py @@ -118,6 +118,20 @@ def test_sql_injection_but_blocking_off(caplog, get_context, monkeypatch): assert get_cache().stats.get_record()["requests"]["attacksDetected"]["blocked"] == 0 +def test_sql_injection_with_invalid_wildcard_route(caplog, get_context, monkeypatch): + get_context.set_as_current_context() + monkeypatch.setenv("AIKIDO_BLOCK", "1") + get_cache().config.set_endpoints( + [{"method": "GET", "route": "/api/[/*", "forceProtectionOff": False}] + ) + with pytest.raises(AikidoSQLInjection): + run_vulnerability_scan( + kind="sql_injection", + op="test_op", + args=("INSERT * INTO VALUES ('doggoss2', TRUE);", "mysql"), + ) + + def test_sql_injection_with_route_params(caplog, get_context, monkeypatch): get_context.set_as_current_context() monkeypatch.setenv("AIKIDO_BLOCK", "1")