From b6e053927e4db5a19eb332c12a794cdd974a75a6 Mon Sep 17 00:00:00 2001 From: Polina Date: Thu, 2 Jul 2026 20:47:33 +0000 Subject: [PATCH] =?UTF-8?q?=D0=97=D0=B0=D0=B3=D1=80=D1=83=D0=B7=D0=B8?= =?UTF-8?q?=D1=82=D1=8C=20=D1=84=D0=B0=D0=B9=D0=BB=D1=8B=20=D0=B2=20=C2=AB?= =?UTF-8?q?venv/Lib/site-packages/starlette/middleware=C2=BB?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../starlette/middleware/exceptions.py | 73 +++++++++ .../starlette/middleware/gzip.py | 145 ++++++++++++++++++ .../starlette/middleware/httpsredirect.py | 19 +++ .../starlette/middleware/sessions.py | 125 +++++++++++++++ 4 files changed, 362 insertions(+) create mode 100644 venv/Lib/site-packages/starlette/middleware/exceptions.py create mode 100644 venv/Lib/site-packages/starlette/middleware/gzip.py create mode 100644 venv/Lib/site-packages/starlette/middleware/httpsredirect.py create mode 100644 venv/Lib/site-packages/starlette/middleware/sessions.py diff --git a/venv/Lib/site-packages/starlette/middleware/exceptions.py b/venv/Lib/site-packages/starlette/middleware/exceptions.py new file mode 100644 index 0000000..6c98558 --- /dev/null +++ b/venv/Lib/site-packages/starlette/middleware/exceptions.py @@ -0,0 +1,73 @@ +from __future__ import annotations + +from collections.abc import Mapping +from typing import Any + +from starlette._exception_handler import ( + ExceptionHandlers, + StatusHandlers, + wrap_app_handling_exceptions, +) +from starlette.exceptions import HTTPException, WebSocketException +from starlette.requests import Request +from starlette.responses import PlainTextResponse, Response +from starlette.types import ASGIApp, ExceptionHandler, Receive, Scope, Send +from starlette.websockets import WebSocket + + +class ExceptionMiddleware: + def __init__( + self, + app: ASGIApp, + handlers: Mapping[Any, ExceptionHandler] | None = None, + debug: bool = False, + ) -> None: + self.app = app + self.debug = debug # TODO: We ought to handle 404 cases if debug is set. + self._status_handlers: StatusHandlers = {} + self._exception_handlers: ExceptionHandlers = { + HTTPException: self.http_exception, + WebSocketException: self.websocket_exception, + } + if handlers is not None: # pragma: no branch + for key, value in handlers.items(): + self.add_exception_handler(key, value) + + def add_exception_handler( + self, + exc_class_or_status_code: int | type[Exception], + handler: ExceptionHandler, + ) -> None: + if isinstance(exc_class_or_status_code, int): + self._status_handlers[exc_class_or_status_code] = handler + else: + assert issubclass(exc_class_or_status_code, Exception) + self._exception_handlers[exc_class_or_status_code] = handler + + async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: + if scope["type"] not in ("http", "websocket"): + await self.app(scope, receive, send) + return + + scope["starlette.exception_handlers"] = ( + self._exception_handlers, + self._status_handlers, + ) + + conn: Request | WebSocket + if scope["type"] == "http": + conn = Request(scope, receive, send) + else: + conn = WebSocket(scope, receive, send) + + await wrap_app_handling_exceptions(self.app, conn)(scope, receive, send) + + async def http_exception(self, request: Request, exc: Exception) -> Response: + assert isinstance(exc, HTTPException) + if exc.status_code in {204, 304}: + return Response(status_code=exc.status_code, headers=exc.headers) + return PlainTextResponse(exc.detail, status_code=exc.status_code, headers=exc.headers) + + async def websocket_exception(self, websocket: WebSocket, exc: Exception) -> None: + assert isinstance(exc, WebSocketException) + await websocket.close(code=exc.code, reason=exc.reason) # pragma: no cover diff --git a/venv/Lib/site-packages/starlette/middleware/gzip.py b/venv/Lib/site-packages/starlette/middleware/gzip.py new file mode 100644 index 0000000..abd898b --- /dev/null +++ b/venv/Lib/site-packages/starlette/middleware/gzip.py @@ -0,0 +1,145 @@ +import gzip +import io +from typing import NoReturn + +from starlette.datastructures import Headers, MutableHeaders +from starlette.types import ASGIApp, Message, Receive, Scope, Send + +DEFAULT_EXCLUDED_CONTENT_TYPES = ("text/event-stream",) + + +class GZipMiddleware: + def __init__(self, app: ASGIApp, minimum_size: int = 500, compresslevel: int = 9) -> None: + self.app = app + self.minimum_size = minimum_size + self.compresslevel = compresslevel + + async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: + if scope["type"] != "http": # pragma: no cover + await self.app(scope, receive, send) + return + + headers = Headers(scope=scope) + responder: ASGIApp + if "gzip" in headers.get("Accept-Encoding", ""): + responder = GZipResponder(self.app, self.minimum_size, compresslevel=self.compresslevel) + else: + responder = IdentityResponder(self.app, self.minimum_size) + + await responder(scope, receive, send) + + +class IdentityResponder: + content_encoding: str + + def __init__(self, app: ASGIApp, minimum_size: int) -> None: + self.app = app + self.minimum_size = minimum_size + self.send: Send = unattached_send + self.initial_message: Message = {} + self.started = False + self.content_encoding_set = False + self.content_type_is_excluded = False + + async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: + self.send = send + await self.app(scope, receive, self.send_with_compression) + + async def send_with_compression(self, message: Message) -> None: + message_type = message["type"] + if message_type == "http.response.start": + # Don't send the initial message until we've determined how to + # modify the outgoing headers correctly. + self.initial_message = message + headers = Headers(raw=self.initial_message["headers"]) + self.content_encoding_set = "content-encoding" in headers + self.content_type_is_excluded = headers.get("content-type", "").startswith(DEFAULT_EXCLUDED_CONTENT_TYPES) + elif message_type == "http.response.body" and (self.content_encoding_set or self.content_type_is_excluded): + if not self.started: + self.started = True + await self.send(self.initial_message) + await self.send(message) + elif message_type == "http.response.body" and not self.started: + self.started = True + body = message.get("body", b"") + more_body = message.get("more_body", False) + if len(body) < self.minimum_size and not more_body: + # Don't apply compression to small outgoing responses. + await self.send(self.initial_message) + await self.send(message) + elif not more_body: + # Standard response. + body = self.apply_compression(body, more_body=False) + + headers = MutableHeaders(raw=self.initial_message["headers"]) + headers.add_vary_header("Accept-Encoding") + if body != message["body"]: + headers["Content-Encoding"] = self.content_encoding + headers["Content-Length"] = str(len(body)) + message["body"] = body + + await self.send(self.initial_message) + await self.send(message) + else: + # Initial body in streaming response. + body = self.apply_compression(body, more_body=True) + + headers = MutableHeaders(raw=self.initial_message["headers"]) + headers.add_vary_header("Accept-Encoding") + if body != message["body"]: + headers["Content-Encoding"] = self.content_encoding + del headers["Content-Length"] + message["body"] = body + + await self.send(self.initial_message) + await self.send(message) + elif message_type == "http.response.body": + # Remaining body in streaming response. + body = message.get("body", b"") + more_body = message.get("more_body", False) + + message["body"] = self.apply_compression(body, more_body=more_body) + + await self.send(message) + elif message_type == "http.response.pathsend": # pragma: no branch + # Don't apply GZip to pathsend responses + await self.send(self.initial_message) + await self.send(message) + + def apply_compression(self, body: bytes, *, more_body: bool) -> bytes: + """Apply compression on the response body. + + If more_body is False, any compression file should be closed. If it + isn't, it won't be closed automatically until all background tasks + complete. + """ + return body + + +class GZipResponder(IdentityResponder): + content_encoding = "gzip" + + def __init__(self, app: ASGIApp, minimum_size: int, compresslevel: int = 9) -> None: + super().__init__(app, minimum_size) + + self.gzip_buffer = io.BytesIO() + self.gzip_file = gzip.GzipFile(mode="wb", fileobj=self.gzip_buffer, compresslevel=compresslevel) + + async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: + with self.gzip_buffer, self.gzip_file: + await super().__call__(scope, receive, send) + + def apply_compression(self, body: bytes, *, more_body: bool) -> bytes: + self.gzip_file.write(body) + if not more_body: + self.gzip_file.close() + + body = self.gzip_buffer.getvalue() + self.gzip_buffer.seek(0) + self.gzip_buffer.truncate() + + return body + + +async def unattached_send(message: Message) -> NoReturn: + raise RuntimeError("send awaitable not set") # pragma: no cover diff --git a/venv/Lib/site-packages/starlette/middleware/httpsredirect.py b/venv/Lib/site-packages/starlette/middleware/httpsredirect.py new file mode 100644 index 0000000..a835906 --- /dev/null +++ b/venv/Lib/site-packages/starlette/middleware/httpsredirect.py @@ -0,0 +1,19 @@ +from starlette.datastructures import URL +from starlette.responses import RedirectResponse +from starlette.types import ASGIApp, Receive, Scope, Send + + +class HTTPSRedirectMiddleware: + def __init__(self, app: ASGIApp) -> None: + self.app = app + + async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: + if scope["type"] in ("http", "websocket") and scope["scheme"] in ("http", "ws"): + url = URL(scope=scope) + redirect_scheme = {"http": "https", "ws": "wss"}[url.scheme] + netloc = url.hostname if url.port in (80, 443) else url.netloc + url = url.replace(scheme=redirect_scheme, netloc=netloc) + response = RedirectResponse(url, status_code=307) + await response(scope, receive, send) + else: + await self.app(scope, receive, send) diff --git a/venv/Lib/site-packages/starlette/middleware/sessions.py b/venv/Lib/site-packages/starlette/middleware/sessions.py new file mode 100644 index 0000000..aa0383f --- /dev/null +++ b/venv/Lib/site-packages/starlette/middleware/sessions.py @@ -0,0 +1,125 @@ +from __future__ import annotations + +import json +import typing +from base64 import b64decode, b64encode +from typing import Literal + +import itsdangerous +from itsdangerous.exc import BadSignature + +from starlette.datastructures import MutableHeaders, Secret +from starlette.requests import HTTPConnection +from starlette.types import ASGIApp, Message, Receive, Scope, Send + + +class SessionMiddleware: + def __init__( + self, + app: ASGIApp, + secret_key: str | Secret, + session_cookie: str = "session", + max_age: int | None = 14 * 24 * 60 * 60, # 14 days, in seconds + path: str = "/", + same_site: Literal["lax", "strict", "none"] = "lax", + https_only: bool = False, + domain: str | None = None, + ) -> None: + self.app = app + self.signer = itsdangerous.TimestampSigner(str(secret_key)) + self.session_cookie = session_cookie + self.max_age = max_age + self.path = path + self.security_flags = "httponly; samesite=" + same_site + if https_only: # Secure flag can be used with HTTPS only + self.security_flags += "; secure" + if domain is not None: + self.security_flags += f"; domain={domain}" + + async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: + if scope["type"] not in ("http", "websocket"): # pragma: no cover + await self.app(scope, receive, send) + return + + connection = HTTPConnection(scope) + initial_session_was_empty = True + + if self.session_cookie in connection.cookies: + data = connection.cookies[self.session_cookie].encode("utf-8") + try: + data = self.signer.unsign(data, max_age=self.max_age) + scope["session"] = Session(json.loads(b64decode(data))) + initial_session_was_empty = False + except BadSignature: + scope["session"] = Session() + else: + scope["session"] = Session() + + async def send_wrapper(message: Message) -> None: + if message["type"] == "http.response.start": + session: Session = scope["session"] + headers = MutableHeaders(scope=message) + if session.accessed: + headers.add_vary_header("Cookie") + if session.modified and session: + # We have session data to persist. + data = b64encode(json.dumps(session).encode("utf-8")) + data = self.signer.sign(data) + header_value = "{session_cookie}={data}; path={path}; {max_age}{security_flags}".format( + session_cookie=self.session_cookie, + data=data.decode("utf-8"), + path=self.path, + max_age=f"Max-Age={self.max_age}; " if self.max_age else "", + security_flags=self.security_flags, + ) + headers.append("Set-Cookie", header_value) + elif session.modified and not initial_session_was_empty: + # The session has been cleared. + header_value = "{session_cookie}={data}; path={path}; {expires}{security_flags}".format( + session_cookie=self.session_cookie, + data="null", + path=self.path, + expires="expires=Thu, 01 Jan 1970 00:00:00 GMT; ", + security_flags=self.security_flags, + ) + headers.append("Set-Cookie", header_value) + await send(message) + + await self.app(scope, receive, send_wrapper) + + +class Session(dict[str, typing.Any]): + accessed: bool = False + modified: bool = False + + def mark_accessed(self) -> None: + self.accessed = True + + def mark_modified(self) -> None: + self.accessed = True + self.modified = True + + def __setitem__(self, key: str, value: typing.Any) -> None: + self.mark_modified() + super().__setitem__(key, value) + + def __delitem__(self, key: str) -> None: + self.mark_modified() + super().__delitem__(key) + + def clear(self) -> None: + self.mark_modified() + super().clear() + + def pop(self, key: str, *args: typing.Any) -> typing.Any: + self.modified = self.modified or key in self + return super().pop(key, *args) + + def setdefault(self, key: str, default: typing.Any = None) -> typing.Any: + if key not in self: + self.mark_modified() + return super().setdefault(key, default) + + def update(self, *args: typing.Any, **kwargs: typing.Any) -> None: + self.mark_modified() + super().update(*args, **kwargs)