src/genro_asgi/middleware/cors.py¶
Source from this local checkout, regenerated when the reader rebuilds.
Line links use #L<number>; a GitHub line range opens its first line.
1 # Copyright 2025 Softwell S.r.l.2 #3 # Licensed under the Apache License, Version 2.0 (the "License");4 # you may not use this file except in compliance with the License.5 # You may obtain a copy of the License at6 #7 # https://www.apache.org/licenses/LICENSE-2.08 #9 # Unless required by applicable law or agreed to in writing, software10 # distributed under the License is distributed on an "AS IS" BASIS,11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.12 # See the License for the specific language governing permissions and13 # limitations under the License.14 15 """CORS (Cross-Origin Resource Sharing) middleware.16 17 Adds CORS headers to HTTP responses and answers preflight ``OPTIONS``18 requests. Preflight-only headers are precomputed once in ``__init__``.19 """20 21 from __future__ import annotations22 23 from collections.abc import MutableMapping24 from typing import TYPE_CHECKING, Any25 26 from .base import BaseMiddleware27 28 if TYPE_CHECKING:29 from ..types import ASGIApp, Receive, Scope, Send30 31 __all__ = ["CORSMiddleware"]32 33 34 def _split_and_strip(value: str | list[str] | None, default: list[str] | None = None) -> list[str]:35 """Split a comma-separated string into a stripped list; pass a list through."""36 if value is None:37 return list(default) if default is not None else []38 if isinstance(value, str):39 return [item.strip() for item in value.split(",")]40 return list(value)41 42 43 class CORSMiddleware(BaseMiddleware):44 """Answer CORS preflight requests and add CORS headers to responses."""45 46 middleware_order = 30047 middleware_default = False48 49 def __init__(50 self,51 app: ASGIApp,52 server: Any,53 allow_origins: str | list[str] | None = None,54 allow_methods: str | list[str] | None = None,55 allow_headers: str | list[str] | None = None,56 allow_credentials: bool = False,57 expose_headers: str | list[str] | None = None,58 max_age: int = 600,59 **options: Any,60 ) -> None:61 super().__init__(app, server, **options)62 self._allow_origins = _split_and_strip(allow_origins, ["*"])63 self._allow_methods = _split_and_strip(64 allow_methods, ["GET", "POST", "PUT", "DELETE", "OPTIONS", "PATCH", "HEAD"]65 )66 self._allow_headers = _split_and_strip(allow_headers, ["*"])67 self._allow_credentials = allow_credentials68 self._expose_headers = _split_and_strip(expose_headers)69 self._max_age = max_age70 self._allow_all_origins = "*" in self._allow_origins71 self._preflight_headers = self._build_preflight_headers()72 73 def _build_preflight_headers(self) -> list[tuple[bytes, bytes]]:74 headers = [75 (b"access-control-allow-methods", ", ".join(self._allow_methods).encode()),76 (b"access-control-max-age", str(self._max_age).encode()),77 ]78 if self._allow_headers:79 value = (80 b"*"81 if "*" in self._allow_headers82 else ", ".join(self._allow_headers).encode()83 )84 headers.append((b"access-control-allow-headers", value))85 if self._allow_credentials:86 headers.append((b"access-control-allow-credentials", b"true"))87 return headers88 89 def _cors_headers(self, origin: str | None) -> list[tuple[bytes, bytes]]:90 """CORS headers for one response, or ``[]`` when the origin is not allowed."""91 if not origin:92 return []93 if self._allow_all_origins:94 if self._allow_credentials:95 headers = [(b"access-control-allow-origin", origin.encode()), (b"vary", b"Origin")]96 else:97 headers = [(b"access-control-allow-origin", b"*")]98 elif origin in self._allow_origins:99 headers = [(b"access-control-allow-origin", origin.encode()), (b"vary", b"Origin")]100 else:101 return []102 if self._allow_credentials:103 headers.append((b"access-control-allow-credentials", b"true"))104 if self._expose_headers:105 headers.append(106 (b"access-control-expose-headers", ", ".join(self._expose_headers).encode())107 )108 return headers109 110 async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:111 """Short-circuit an OPTIONS preflight; otherwise wrap ``send`` to add CORS headers."""112 origin = None113 for name, value in scope.get("headers", []):114 if name == b"origin":115 origin = value.decode("latin-1")116 break117 118 if scope.get("method") == "OPTIONS" and origin:119 await self._respond_preflight(send, origin)120 return121 122 cors_headers = self._cors_headers(origin)123 124 async def send_with_cors(message: MutableMapping[str, Any]) -> None:125 if message["type"] == "http.response.start" and cors_headers:126 headers = list(message.get("headers", []))127 headers.extend(cors_headers)128 message = {**message, "headers": headers}129 await send(message)130 131 await self.app(scope, receive, send_with_cors)132 133 async def _respond_preflight(self, send: Send, origin: str) -> None:134 headers = self._cors_headers(origin)135 if not headers:136 await send({"type": "http.response.start", "status": 400, "headers": []})137 await send({"type": "http.response.body", "body": b""})138 return139 headers = headers + self._preflight_headers140 await send({"type": "http.response.start", "status": 200, "headers": headers})141 await send({"type": "http.response.body", "body": b""})