Return explicit origin in CORS response when credentials are allowed

starlette/middleware/cors.py

@@ -68,6 +68,7 @@ def __init__(
         self.allow_headers = [h.lower() for h in allow_headers]
         self.allow_all_origins = allow_all_origins
         self.allow_all_headers = allow_all_headers
+        self.allow_credentials = allow_credentials
         self.preflight_explicit_allow_origin = preflight_explicit_allow_origin
         self.allow_origin_regex = compiled_allow_origin_regex
         self.allow_private_network = allow_private_network
@@ -161,15 +162,12 @@ async def send(self, message: Message, send: Send, request_headers: Headers) ->
         headers = MutableHeaders(scope=message)
         headers.update(self.simple_headers)
         origin = request_headers["Origin"]
-        has_cookie = "cookie" in request_headers
 
-        # If request includes any cookie headers, then we must respond
-        # with the specific origin instead of '*'.
-        if self.allow_all_origins and has_cookie:
+        # If credentials are allowed, then we must respond with the specific origin instead of '*'.
+        if self.allow_all_origins and self.allow_credentials:
             self.allow_explicit_origin(headers, origin)
 
-        # If we only allow specific origins, then we have to mirror back
-        # the Origin header in the response.
+        # If we only allow specific origins, then we have to mirror back the Origin header in the response.
         elif not self.allow_all_origins and self.is_allowed_origin(origin=origin):
             self.allow_explicit_origin(headers, origin)
 

tests/middleware/test_cors.py

@@ -48,7 +48,7 @@ def homepage(request: Request) -> PlainTextResponse:
     response = client.get("/", headers=headers)
     assert response.status_code == 200
     assert response.text == "Homepage"
-    assert response.headers["access-control-allow-origin"] == "*"
+    assert response.headers["access-control-allow-origin"] == "https://example.org"
     assert response.headers["access-control-expose-headers"] == "X-Status"
     assert response.headers["access-control-allow-credentials"] == "true"
 
@@ -404,30 +404,7 @@ def homepage(request: Request) -> PlainTextResponse:
     assert "access-control-allow-origin" not in response.headers
 
 
-def test_cors_credentialed_requests_return_specific_origin(
-    test_client_factory: TestClientFactory,
-) -> None:
-    def homepage(request: Request) -> PlainTextResponse:
-        return PlainTextResponse("Homepage", status_code=200)
-
-    app = Starlette(
-        routes=[Route("/", endpoint=homepage)],
-        middleware=[Middleware(CORSMiddleware, allow_origins=["*"])],
-    )
-    client = test_client_factory(app)
-
-    # Test credentialed request
-    headers = {"Origin": "https://example.org", "Cookie": "star_cookie=sugar"}
-    response = client.get("/", headers=headers)
-    assert response.status_code == 200
-    assert response.text == "Homepage"
-    assert response.headers["access-control-allow-origin"] == "https://example.org"
-    assert "access-control-allow-credentials" not in response.headers
-
-
-def test_cors_vary_header_defaults_to_origin(
-    test_client_factory: TestClientFactory,
-) -> None:
+def test_cors_vary_header_defaults_to_origin(test_client_factory: TestClientFactory) -> None:
     def homepage(request: Request) -> PlainTextResponse:
         return PlainTextResponse("Homepage", status_code=200)
 
@@ -445,9 +422,7 @@ def homepage(request: Request) -> PlainTextResponse:
     assert response.headers["vary"] == "Origin"
 
 
-def test_cors_vary_header_is_not_set_for_non_credentialed_request(
-    test_client_factory: TestClientFactory,
-) -> None:
+def test_cors_vary_header_is_not_set_for_non_credentialed_request(test_client_factory: TestClientFactory) -> None:
     def homepage(request: Request) -> PlainTextResponse:
         return PlainTextResponse("Homepage", status_code=200, headers={"Vary": "Accept-Encoding"})
 
@@ -462,19 +437,17 @@ def homepage(request: Request) -> PlainTextResponse:
     assert response.headers["vary"] == "Accept-Encoding"
 
 
-def test_cors_vary_header_is_properly_set_for_credentialed_request(
-    test_client_factory: TestClientFactory,
-) -> None:
+def test_cors_vary_header_is_properly_set_for_credentialed_request(test_client_factory: TestClientFactory) -> None:
     def homepage(request: Request) -> PlainTextResponse:
         return PlainTextResponse("Homepage", status_code=200, headers={"Vary": "Accept-Encoding"})
 
     app = Starlette(
         routes=[Route("/", endpoint=homepage)],
-        middleware=[Middleware(CORSMiddleware, allow_origins=["*"])],
+        middleware=[Middleware(CORSMiddleware, allow_origins=["*"], allow_credentials=True)],
     )
     client = test_client_factory(app)
 
-    response = client.get("/", headers={"Cookie": "foo=bar", "Origin": "https://someplace.org"})
+    response = client.get("/", headers={"Origin": "https://someplace.org"})
     assert response.status_code == 200
     assert response.headers["vary"] == "Accept-Encoding, Origin"
 
@@ -498,38 +471,25 @@ def homepage(request: Request) -> PlainTextResponse:
     assert response.headers["vary"] == "Accept-Encoding, Origin"
 
 
-def test_cors_allowed_origin_does_not_leak_between_credentialed_requests(
-    test_client_factory: TestClientFactory,
-) -> None:
+def test_cors_allowed_origin_does_not_leak_between_requests(test_client_factory: TestClientFactory) -> None:
     def homepage(request: Request) -> PlainTextResponse:
         return PlainTextResponse("Homepage", status_code=200)
 
     app = Starlette(
-        routes=[
-            Route("/", endpoint=homepage),
-        ],
-        middleware=[
-            Middleware(
-                CORSMiddleware,
-                allow_origins=["*"],
-                allow_headers=["*"],
-                allow_methods=["*"],
-            )
-        ],
+        routes=[Route("/", endpoint=homepage)],
+        middleware=[Middleware(CORSMiddleware, allow_origins=["https://example.org"])],
     )
 
     client = test_client_factory(app)
-    response = client.get("/", headers={"Origin": "https://someplace.org"})
-    assert response.headers["access-control-allow-origin"] == "*"
-    assert "access-control-allow-credentials" not in response.headers
 
-    response = client.get("/", headers={"Cookie": "foo=bar", "Origin": "https://someplace.org"})
-    assert response.headers["access-control-allow-origin"] == "https://someplace.org"
-    assert "access-control-allow-credentials" not in response.headers
+    response = client.get("/", headers={"Origin": "https://example.org"})
+    assert response.headers["access-control-allow-origin"] == "https://example.org"
 
-    response = client.get("/", headers={"Origin": "https://someplace.org"})
-    assert response.headers["access-control-allow-origin"] == "*"
-    assert "access-control-allow-credentials" not in response.headers
+    response = client.get("/", headers={"Origin": "https://other.org"})
+    assert "access-control-allow-origin" not in response.headers
+
+    response = client.get("/", headers={"Origin": "https://example.org"})
+    assert response.headers["access-control-allow-origin"] == "https://example.org"
 
 
 def test_cors_private_network_access_allowed(test_client_factory: TestClientFactory) -> None: