Merge commit from fork

Author: provinzkraut

litestar/config/cors.py

@@ -1,10 +1,11 @@
 from __future__ import annotations
 
 import re
+import uuid
 from dataclasses import dataclass, field
 from functools import cached_property
 from re import Pattern
-from typing import TYPE_CHECKING, Literal
+from typing import TYPE_CHECKING, Final, Literal
 
 from litestar.constants import DEFAULT_ALLOWED_CORS_HEADERS
 
@@ -15,6 +16,11 @@
     from litestar.types import Method
 
 
+# this is just a UUID, so we can be sure it's not contained within the string we're
+# calling '.replace' on
+_RE_ESCAPE_PLACEHOLDER: Final = uuid.uuid4().hex
+
+
 @dataclass
 class CORSConfig:
     """Configuration for CORS (Cross-Origin Resource Sharing).
@@ -60,10 +66,14 @@ def allowed_origins_regex(self) -> Pattern[str]:
         Returns:
             A compiled regex of the allowed path.
         """
-        origins = self.allow_origins
+        # escape the allowed origins, while turning '*' into wildcard '.*' matches
+        origins = [
+            re.escape(o.replace("*", _RE_ESCAPE_PLACEHOLDER)).replace(_RE_ESCAPE_PLACEHOLDER, ".*")
+            for o in self.allow_origins
+        ]
         if self.allow_origin_regex:
             origins.append(self.allow_origin_regex)
-        return re.compile("|".join([origin.replace("*.", r".*\.") for origin in origins]))
+        return re.compile("|".join(origins))
 
     @cached_property
     def is_allow_all_origins(self) -> bool:

tests/unit/test_middleware/test_cors_middleware.py

@@ -121,3 +121,37 @@ def handler() -> dict[str, str]:
             assert response.headers.get("Access-Control-Allow-Origin") == origin
         else:
             assert not response.headers.get("Access-Control-Allow-Origin")
+
+
+@pytest.mark.parametrize(
+    "allow_origin,origin,host,should_allow",
+    [
+        ("httpx://good.example", "https://goodXexample", "example.com", False),
+        ("https://*good.example", "https://very.good.example", "very.good.example", True),
+        ("https://*good.example", "https://verygood.example", "vergood.example", True),
+        ("https://*good.example", "https://good.example", "good.example", True),
+        ("https://*good.example", "https://bad.example", "bad.example", False),
+        ("https://*.good.example", "https://very.good.example", "very.good.example", True),
+        ("https://*.good.example", "https://verygood.example", "verygood.example", False),
+        ("https://*.good.example", "https://some.verygood.example", "verygood.example", False),
+        ("https://*.good.example", "https://good.example", "good.example", False),
+    ],
+)
+def test_cors_test_regex_escape(allow_origin: str, origin: str, host: str, should_allow: bool) -> None:
+    @get("/")
+    async def handler() -> None:
+        return None
+
+    with create_test_client(
+        [handler],
+        cors_config=CORSConfig(
+            allow_origins=[allow_origin],
+            allow_credentials=True,
+        ),
+    ) as client:
+        res = client.get("/", headers={"Origin": origin, "Host": host})
+
+    if should_allow:
+        assert "Access-Control-Allow-Origin" in res.headers
+    else:
+        assert "Access-Control-Allow-Origin" not in res.headers