Skip wildcard routes that do not compile

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

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"

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")