diff options
Diffstat (limited to 'venv/lib/python3.11/site-packages/litestar/middleware/session')
8 files changed, 742 insertions, 0 deletions
diff --git a/venv/lib/python3.11/site-packages/litestar/middleware/session/__init__.py b/venv/lib/python3.11/site-packages/litestar/middleware/session/__init__.py new file mode 100644 index 0000000..1ca9c17 --- /dev/null +++ b/venv/lib/python3.11/site-packages/litestar/middleware/session/__init__.py @@ -0,0 +1,3 @@ +from .base import SessionMiddleware + +__all__ = ("SessionMiddleware",) diff --git a/venv/lib/python3.11/site-packages/litestar/middleware/session/__pycache__/__init__.cpython-311.pyc b/venv/lib/python3.11/site-packages/litestar/middleware/session/__pycache__/__init__.cpython-311.pyc Binary files differnew file mode 100644 index 0000000..8748ce3 --- /dev/null +++ b/venv/lib/python3.11/site-packages/litestar/middleware/session/__pycache__/__init__.cpython-311.pyc diff --git a/venv/lib/python3.11/site-packages/litestar/middleware/session/__pycache__/base.cpython-311.pyc b/venv/lib/python3.11/site-packages/litestar/middleware/session/__pycache__/base.cpython-311.pyc Binary files differnew file mode 100644 index 0000000..68a8b9c --- /dev/null +++ b/venv/lib/python3.11/site-packages/litestar/middleware/session/__pycache__/base.cpython-311.pyc diff --git a/venv/lib/python3.11/site-packages/litestar/middleware/session/__pycache__/client_side.cpython-311.pyc b/venv/lib/python3.11/site-packages/litestar/middleware/session/__pycache__/client_side.cpython-311.pyc Binary files differnew file mode 100644 index 0000000..692f54c --- /dev/null +++ b/venv/lib/python3.11/site-packages/litestar/middleware/session/__pycache__/client_side.cpython-311.pyc diff --git a/venv/lib/python3.11/site-packages/litestar/middleware/session/__pycache__/server_side.cpython-311.pyc b/venv/lib/python3.11/site-packages/litestar/middleware/session/__pycache__/server_side.cpython-311.pyc Binary files differnew file mode 100644 index 0000000..bd2373c --- /dev/null +++ b/venv/lib/python3.11/site-packages/litestar/middleware/session/__pycache__/server_side.cpython-311.pyc diff --git a/venv/lib/python3.11/site-packages/litestar/middleware/session/base.py b/venv/lib/python3.11/site-packages/litestar/middleware/session/base.py new file mode 100644 index 0000000..a823848 --- /dev/null +++ b/venv/lib/python3.11/site-packages/litestar/middleware/session/base.py @@ -0,0 +1,256 @@ +from __future__ import annotations + +from abc import ABC, abstractmethod +from typing import ( + TYPE_CHECKING, + Any, + Awaitable, + Callable, + Generic, + Literal, + TypeVar, + cast, +) + +from litestar.connection import ASGIConnection +from litestar.enums import ScopeType +from litestar.middleware.base import AbstractMiddleware, DefineMiddleware +from litestar.serialization import decode_json, encode_json +from litestar.utils import get_serializer_from_scope + +__all__ = ("BaseBackendConfig", "BaseSessionBackend", "SessionMiddleware") + + +if TYPE_CHECKING: + from litestar.types import ASGIApp, Message, Receive, Scope, Scopes, ScopeSession, Send + +ONE_DAY_IN_SECONDS = 60 * 60 * 24 + +ConfigT = TypeVar("ConfigT", bound="BaseBackendConfig") +BaseSessionBackendT = TypeVar("BaseSessionBackendT", bound="BaseSessionBackend") + + +class BaseBackendConfig(ABC, Generic[BaseSessionBackendT]): # pyright: ignore + """Configuration for Session middleware backends.""" + + _backend_class: type[BaseSessionBackendT] # pyright: ignore + + key: str + """Key to use for the cookie inside the header, e.g. ``session=<data>`` where ``session`` is the cookie key and + ``<data>`` is the session data. + + Notes: + - If a session cookie exceeds 4KB in size it is split. In this case the key will be of the format + ``session-{segment number}``. + + """ + max_age: int + """Maximal age of the cookie before its invalidated.""" + scopes: Scopes = {ScopeType.HTTP, ScopeType.WEBSOCKET} + """Scopes for the middleware - options are ``http`` and ``websocket`` with the default being both""" + path: str + """Path fragment that must exist in the request url for the cookie to be valid. + + Defaults to ``'/'``. + """ + domain: str | None + """Domain for which the cookie is valid.""" + secure: bool + """Https is required for the cookie.""" + httponly: bool + """Forbids javascript to access the cookie via 'Document.cookie'.""" + samesite: Literal["lax", "strict", "none"] + """Controls whether or not a cookie is sent with cross-site requests. + + Defaults to ``lax``. + """ + exclude: str | list[str] | None + """A pattern or list of patterns to skip in the session middleware.""" + exclude_opt_key: str + """An identifier to use on routes to disable the session middleware for a particular route.""" + + @property + def middleware(self) -> DefineMiddleware: + """Use this property to insert the config into a middleware list on one of the application layers. + + Examples: + .. code-block:: python + + from os import urandom + + from litestar import Litestar, Request, get + from litestar.middleware.sessions.cookie_backend import CookieBackendConfig + + session_config = CookieBackendConfig(secret=urandom(16)) + + + @get("/") + def my_handler(request: Request) -> None: ... + + + app = Litestar(route_handlers=[my_handler], middleware=[session_config.middleware]) + + + Returns: + An instance of DefineMiddleware including ``self`` as the config kwarg value. + """ + return DefineMiddleware(SessionMiddleware, backend=self._backend_class(config=self)) + + +class BaseSessionBackend(ABC, Generic[ConfigT]): + """Abstract session backend defining the interface between a storage mechanism and the application + :class:`SessionMiddleware`. + + This serves as the base class for all client- and server-side backends + """ + + __slots__ = ("config",) + + def __init__(self, config: ConfigT) -> None: + """Initialize ``BaseSessionBackend`` + + Args: + config: A instance of a subclass of ``BaseBackendConfig`` + """ + self.config = config + + @staticmethod + def serialize_data(data: ScopeSession, scope: Scope | None = None) -> bytes: + """Serialize data into bytes for storage in the backend. + + Args: + data: Session data of the current scope. + scope: A scope, if applicable, from which to extract a serializer. + + Notes: + - The serializer will be extracted from ``scope`` or fall back to + :func:`default_serializer <.serialization.default_serializer>` + + Returns: + ``data`` serialized as bytes. + """ + serializer = get_serializer_from_scope(scope) if scope else None + return encode_json(data, serializer) + + @staticmethod + def deserialize_data(data: Any) -> dict[str, Any]: + """Deserialize data into a dictionary for use in the application scope. + + Args: + data: Data to be deserialized + + Returns: + Deserialized data as a dictionary + """ + return cast("dict[str, Any]", decode_json(value=data)) + + @abstractmethod + def get_session_id(self, connection: ASGIConnection) -> str | None: + """Try to fetch session id from connection ScopeState. If one does not exist, generate one. + + Args: + connection: Originating ASGIConnection containing the scope + + Returns: + Session id str or None if the concept of a session id does not apply. + """ + + @abstractmethod + async def store_in_message(self, scope_session: ScopeSession, message: Message, connection: ASGIConnection) -> None: + """Store the necessary information in the outgoing ``Message`` + + Args: + scope_session: Current session to store + message: Outgoing send-message + connection: Originating ASGIConnection containing the scope + + Returns: + None + """ + + @abstractmethod + async def load_from_connection(self, connection: ASGIConnection) -> dict[str, Any]: + """Load session data from a connection and return it as a dictionary to be used in the current application + scope. + + Args: + connection: An ASGIConnection instance + + Returns: + The session data + + Notes: + - This should not modify the connection's scope. The data returned by this + method will be stored in the application scope by the middleware + + """ + + +class SessionMiddleware(AbstractMiddleware, Generic[BaseSessionBackendT]): + """Litestar session middleware for storing session data.""" + + def __init__(self, app: ASGIApp, backend: BaseSessionBackendT) -> None: + """Initialize ``SessionMiddleware`` + + Args: + app: An ASGI application + backend: A :class:`BaseSessionBackend` instance used to store and retrieve session data + """ + + super().__init__( + app=app, + exclude=backend.config.exclude, + exclude_opt_key=backend.config.exclude_opt_key, + scopes=backend.config.scopes, + ) + self.backend = backend + + def create_send_wrapper(self, connection: ASGIConnection) -> Callable[[Message], Awaitable[None]]: + """Create a wrapper for the ASGI send function, which handles setting the cookies on the outgoing response. + + Args: + connection: ASGIConnection + + Returns: + None + """ + + async def wrapped_send(message: Message) -> None: + """Wrap the ``send`` function. + + Declared in local scope to make use of closure values. + + Args: + message: An ASGI message. + + Returns: + None + """ + if message["type"] != "http.response.start": + await connection.send(message) + return + + scope_session = connection.scope.get("session") + + await self.backend.store_in_message(scope_session, message, connection) + await connection.send(message) + + return wrapped_send + + async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: + """ASGI-callable. + + Args: + scope: The ASGI connection scope. + receive: The ASGI receive function. + send: The ASGI send function. + + Returns: + None + """ + + connection = ASGIConnection[Any, Any, Any, Any](scope, receive=receive, send=send) + scope["session"] = await self.backend.load_from_connection(connection) + connection._connection_state.session_id = self.backend.get_session_id(connection) # pyright: ignore [reportGeneralTypeIssues] + + await self.app(scope, receive, self.create_send_wrapper(connection)) diff --git a/venv/lib/python3.11/site-packages/litestar/middleware/session/client_side.py b/venv/lib/python3.11/site-packages/litestar/middleware/session/client_side.py new file mode 100644 index 0000000..f709410 --- /dev/null +++ b/venv/lib/python3.11/site-packages/litestar/middleware/session/client_side.py @@ -0,0 +1,264 @@ +from __future__ import annotations + +import binascii +import contextlib +import re +import time +from base64 import b64decode, b64encode +from dataclasses import dataclass, field +from os import urandom +from typing import TYPE_CHECKING, Any, Literal + +from litestar.datastructures import MutableScopeHeaders +from litestar.datastructures.cookie import Cookie +from litestar.enums import ScopeType +from litestar.exceptions import ( + ImproperlyConfiguredException, + MissingDependencyException, +) +from litestar.serialization import decode_json, encode_json +from litestar.types import Empty, Scopes +from litestar.utils.dataclass import extract_dataclass_items + +from .base import ONE_DAY_IN_SECONDS, BaseBackendConfig, BaseSessionBackend + +__all__ = ("ClientSideSessionBackend", "CookieBackendConfig") + + +try: + from cryptography.exceptions import InvalidTag + from cryptography.hazmat.primitives.ciphers.aead import AESGCM +except ImportError as e: + raise MissingDependencyException("cryptography") from e + +if TYPE_CHECKING: + from litestar.connection import ASGIConnection + from litestar.types import Message, Scope, ScopeSession + +NONCE_SIZE = 12 +CHUNK_SIZE = 4096 - 64 +AAD = b"additional_authenticated_data=" + + +class ClientSideSessionBackend(BaseSessionBackend["CookieBackendConfig"]): + """Cookie backend for SessionMiddleware.""" + + __slots__ = ("aesgcm", "cookie_re") + + def __init__(self, config: CookieBackendConfig) -> None: + """Initialize ``ClientSideSessionBackend``. + + Args: + config: SessionCookieConfig instance. + """ + super().__init__(config) + self.aesgcm = AESGCM(config.secret) + self.cookie_re = re.compile(rf"{self.config.key}(?:-\d+)?") + + def dump_data(self, data: Any, scope: Scope | None = None) -> list[bytes]: + """Given serializable data, including pydantic models and numpy types, dump it into a bytes string, encrypt, + encode and split it into chunks of the desirable size. + + Args: + data: Data to serialize, encrypt, encode and chunk. + scope: The ASGI connection scope. + + Notes: + - The returned list is composed of a chunks of a single base64 encoded + string that is encrypted using AES-CGM. + + Returns: + List of encoded bytes string of a maximum length equal to the ``CHUNK_SIZE`` constant. + """ + serialized = self.serialize_data(data, scope) + associated_data = encode_json({"expires_at": round(time.time()) + self.config.max_age}) + nonce = urandom(NONCE_SIZE) + encrypted = self.aesgcm.encrypt(nonce, serialized, associated_data=associated_data) + encoded = b64encode(nonce + encrypted + AAD + associated_data) + return [encoded[i : i + CHUNK_SIZE] for i in range(0, len(encoded), CHUNK_SIZE)] + + def load_data(self, data: list[bytes]) -> dict[str, Any]: + """Given a list of strings, decodes them into the session object. + + Args: + data: A list of strings derived from the request's session cookie(s). + + Returns: + A deserialized session value. + """ + decoded = b64decode(b"".join(data)) + nonce = decoded[:NONCE_SIZE] + aad_starts_from = decoded.find(AAD) + associated_data = decoded[aad_starts_from:].replace(AAD, b"") if aad_starts_from != -1 else None + if associated_data and decode_json(value=associated_data)["expires_at"] > round(time.time()): + encrypted_session = decoded[NONCE_SIZE:aad_starts_from] + decrypted = self.aesgcm.decrypt(nonce, encrypted_session, associated_data=associated_data) + return self.deserialize_data(decrypted) + return {} + + def get_cookie_keys(self, connection: ASGIConnection) -> list[str]: + """Return a list of cookie-keys from the connection if they match the session-cookie pattern. + + Args: + connection: An ASGIConnection instance + + Returns: + A list of session-cookie keys + """ + return sorted(key for key in connection.cookies if self.cookie_re.fullmatch(key)) + + def _create_session_cookies(self, data: list[bytes], cookie_params: dict[str, Any] | None = None) -> list[Cookie]: + """Create a list of cookies containing the session data. + If the data is split into multiple cookies, the key will be of the format ``session-{segment number}``, + however if only one cookie is needed, the key will be ``session``. + """ + if cookie_params is None: + cookie_params = dict( + extract_dataclass_items( + self.config, + exclude_none=True, + include={f for f in Cookie.__dict__ if f not in ("key", "secret")}, + ) + ) + + if len(data) == 1: + return [ + Cookie( + value=data[0].decode("utf-8"), + key=self.config.key, + **cookie_params, + ) + ] + + return [ + Cookie( + value=datum.decode("utf-8"), + key=f"{self.config.key}-{i}", + **cookie_params, + ) + for i, datum in enumerate(data) + ] + + async def store_in_message(self, scope_session: ScopeSession, message: Message, connection: ASGIConnection) -> None: + """Store data from ``scope_session`` in ``Message`` in the form of cookies. If the contents of ``scope_session`` + are too large to fit a single cookie, it will be split across several cookies, following the naming scheme of + ``<cookie key>-<n>``. If the session is empty or shrinks, cookies will be cleared by setting their value to + ``"null"`` + + Args: + scope_session: Current session to store + message: Outgoing send-message + connection: Originating ASGIConnection containing the scope + + Returns: + None + """ + + scope = connection.scope + headers = MutableScopeHeaders.from_message(message) + cookie_keys = self.get_cookie_keys(connection) + + if scope_session and scope_session is not Empty: + data = self.dump_data(scope_session, scope=scope) + cookie_params = dict( + extract_dataclass_items( + self.config, + exclude_none=True, + include={f for f in Cookie.__dict__ if f not in ("key", "secret")}, + ) + ) + for cookie in self._create_session_cookies(data, cookie_params): + headers.add("Set-Cookie", cookie.to_header(header="")) + # Cookies with the same key overwrite the earlier cookie with that key. To expire earlier session + # cookies, first check how many session cookies will not be overwritten in this upcoming response. + # If leftover cookies are greater than or equal to 1, that means older session cookies have to be + # expired and their names are in cookie_keys. + cookies_to_clear = cookie_keys[len(data) :] if len(cookie_keys) - len(data) > 0 else [] + else: + cookies_to_clear = cookie_keys + + for cookie_key in cookies_to_clear: + cookie_params = dict( + extract_dataclass_items( + self.config, + exclude_none=True, + include={f for f in Cookie.__dict__ if f not in ("key", "secret", "max_age")}, + ) + ) + headers.add( + "Set-Cookie", + Cookie(value="null", key=cookie_key, expires=0, **cookie_params).to_header(header=""), + ) + + async def load_from_connection(self, connection: ASGIConnection) -> dict[str, Any]: + """Load session data from a connection's session-cookies and return it as a dictionary. + + Args: + connection: Originating ASGIConnection + + Returns: + The session data + """ + if cookie_keys := self.get_cookie_keys(connection): + data = [connection.cookies[key].encode("utf-8") for key in cookie_keys] + # If these exceptions occur, the session must remain empty so do nothing. + with contextlib.suppress(InvalidTag, binascii.Error): + return self.load_data(data) + return {} + + def get_session_id(self, connection: ASGIConnection) -> str | None: + return None + + +@dataclass +class CookieBackendConfig(BaseBackendConfig[ClientSideSessionBackend]): # pyright: ignore + """Configuration for [SessionMiddleware] middleware.""" + + _backend_class = ClientSideSessionBackend + + secret: bytes + """A secret key to use for generating an encryption key. + + Must have a length of 16 (128 bits), 24 (192 bits) or 32 (256 bits) characters. + """ + key: str = field(default="session") + """Key to use for the cookie inside the header, e.g. ``session=<data>`` where ``session`` is the cookie key and + ``<data>`` is the session data. + + Notes: + - If a session cookie exceeds 4KB in size it is split. In this case the key will be of the format + ``session-{segment number}``. + + """ + max_age: int = field(default=ONE_DAY_IN_SECONDS * 14) + """Maximal age of the cookie before its invalidated.""" + scopes: Scopes = field(default_factory=lambda: {ScopeType.HTTP, ScopeType.WEBSOCKET}) + """Scopes for the middleware - options are ``http`` and ``websocket`` with the default being both""" + path: str = field(default="/") + """Path fragment that must exist in the request url for the cookie to be valid. + + Defaults to ``'/'``. + """ + domain: str | None = field(default=None) + """Domain for which the cookie is valid.""" + secure: bool = field(default=False) + """Https is required for the cookie.""" + httponly: bool = field(default=True) + """Forbids javascript to access the cookie via 'Document.cookie'.""" + samesite: Literal["lax", "strict", "none"] = field(default="lax") + """Controls whether or not a cookie is sent with cross-site requests. + + Defaults to ``lax``. + """ + exclude: str | list[str] | None = field(default=None) + """A pattern or list of patterns to skip in the session middleware.""" + exclude_opt_key: str = field(default="skip_session") + """An identifier to use on routes to disable the session middleware for a particular route.""" + + def __post_init__(self) -> None: + if len(self.key) < 1 or len(self.key) > 256: + raise ImproperlyConfiguredException("key must be a string with a length between 1-256") + if self.max_age < 1: + raise ImproperlyConfiguredException("max_age must be greater than 0") + if len(self.secret) not in {16, 24, 32}: + raise ImproperlyConfiguredException("secret length must be 16 (128 bit), 24 (192 bit) or 32 (256 bit)") diff --git a/venv/lib/python3.11/site-packages/litestar/middleware/session/server_side.py b/venv/lib/python3.11/site-packages/litestar/middleware/session/server_side.py new file mode 100644 index 0000000..91708ac --- /dev/null +++ b/venv/lib/python3.11/site-packages/litestar/middleware/session/server_side.py @@ -0,0 +1,219 @@ +from __future__ import annotations + +import secrets +from dataclasses import dataclass, field +from typing import TYPE_CHECKING, Any, Literal + +from litestar.datastructures import Cookie, MutableScopeHeaders +from litestar.enums import ScopeType +from litestar.exceptions import ImproperlyConfiguredException +from litestar.middleware.session.base import ONE_DAY_IN_SECONDS, BaseBackendConfig, BaseSessionBackend +from litestar.types import Empty, Message, Scopes, ScopeSession +from litestar.utils.dataclass import extract_dataclass_items + +__all__ = ("ServerSideSessionBackend", "ServerSideSessionConfig") + + +if TYPE_CHECKING: + from litestar import Litestar + from litestar.connection import ASGIConnection + from litestar.stores.base import Store + + +class ServerSideSessionBackend(BaseSessionBackend["ServerSideSessionConfig"]): + """Base class for server-side backends. + + Implements :class:`BaseSessionBackend` and defines and interface which subclasses can + implement to facilitate the storage of session data. + """ + + def __init__(self, config: ServerSideSessionConfig) -> None: + """Initialize ``ServerSideSessionBackend`` + + Args: + config: A subclass of ``ServerSideSessionConfig`` + """ + super().__init__(config=config) + + async def get(self, session_id: str, store: Store) -> bytes | None: + """Retrieve data associated with ``session_id``. + + Args: + session_id: The session-ID + store: Store to retrieve the session data from + + Returns: + The session data, if existing, otherwise ``None``. + """ + max_age = int(self.config.max_age) if self.config.max_age is not None else None + return await store.get(session_id, renew_for=max_age if self.config.renew_on_access else None) + + async def set(self, session_id: str, data: bytes, store: Store) -> None: + """Store ``data`` under the ``session_id`` for later retrieval. + + If there is already data associated with ``session_id``, replace + it with ``data`` and reset its expiry time + + Args: + session_id: The session-ID + data: Serialized session data + store: Store to save the session data in + + Returns: + None + """ + expires_in = int(self.config.max_age) if self.config.max_age is not None else None + await store.set(session_id, data, expires_in=expires_in) + + async def delete(self, session_id: str, store: Store) -> None: + """Delete the data associated with ``session_id``. Fails silently if no such session-ID exists. + + Args: + session_id: The session-ID + store: Store to delete the session data from + + Returns: + None + """ + await store.delete(session_id) + + def get_session_id(self, connection: ASGIConnection) -> str: + """Try to fetch session id from the connection. If one does not exist, generate one. + + If a session ID already exists in the cookies, it is returned. + If there is no ID in the cookies but one in the connection state, then the session exists but has not yet + been returned to the user. + Otherwise, a new session must be created. + + Args: + connection: Originating ASGIConnection containing the scope + Returns: + Session id str or None if the concept of a session id does not apply. + """ + session_id = connection.cookies.get(self.config.key) + if not session_id or session_id == "null": + session_id = connection.get_session_id() + if not session_id: + session_id = self.generate_session_id() + return session_id + + def generate_session_id(self) -> str: + """Generate a new session-ID, with + n=:attr:`session_id_bytes <ServerSideSessionConfig.session_id_bytes>` random bytes. + + Returns: + A session-ID + """ + return secrets.token_hex(self.config.session_id_bytes) + + async def store_in_message(self, scope_session: ScopeSession, message: Message, connection: ASGIConnection) -> None: + """Store the necessary information in the outgoing ``Message`` by setting a cookie containing the session-ID. + + If the session is empty, a null-cookie will be set. Otherwise, the serialised + data will be stored using :meth:`set <ServerSideSessionBackend.set>`, under the current session-id. If no session-ID + exists, a new ID will be generated using :meth:`generate_session_id <ServerSideSessionBackend.generate_session_id>`. + + Args: + scope_session: Current session to store + message: Outgoing send-message + connection: Originating ASGIConnection containing the scope + + Returns: + None + """ + scope = connection.scope + store = self.config.get_store_from_app(scope["app"]) + headers = MutableScopeHeaders.from_message(message) + session_id = self.get_session_id(connection) + + cookie_params = dict(extract_dataclass_items(self.config, exclude_none=True, include=Cookie.__dict__.keys())) + + if scope_session is Empty: + await self.delete(session_id, store=store) + headers.add( + "Set-Cookie", + Cookie(value="null", key=self.config.key, expires=0, **cookie_params).to_header(header=""), + ) + else: + serialised_data = self.serialize_data(scope_session, scope) + await self.set(session_id=session_id, data=serialised_data, store=store) + headers.add( + "Set-Cookie", Cookie(value=session_id, key=self.config.key, **cookie_params).to_header(header="") + ) + + async def load_from_connection(self, connection: ASGIConnection) -> dict[str, Any]: + """Load session data from a connection and return it as a dictionary to be used in the current application + scope. + + The session-ID will be gathered from a cookie with the key set in + :attr:`BaseBackendConfig.key`. If a cookie is found, its value will be used as the session-ID and data associated + with this ID will be loaded using :meth:`get <ServerSideSessionBackend.get>`. + If no cookie was found or no data was loaded from the store, this will return an + empty dictionary. + + Args: + connection: An ASGIConnection instance + + Returns: + The current session data + """ + if session_id := connection.cookies.get(self.config.key): + store = self.config.get_store_from_app(connection.scope["app"]) + data = await self.get(session_id, store=store) + if data is not None: + return self.deserialize_data(data) + return {} + + +@dataclass +class ServerSideSessionConfig(BaseBackendConfig[ServerSideSessionBackend]): # pyright: ignore + """Base configuration for server side backends.""" + + _backend_class = ServerSideSessionBackend + + session_id_bytes: int = field(default=32) + """Number of bytes used to generate a random session-ID.""" + renew_on_access: bool = field(default=False) + """Renew expiry times of sessions when they're being accessed""" + key: str = field(default="session") + """Key to use for the cookie inside the header, e.g. ``session=<data>`` where ``session`` is the cookie key and + ``<data>`` is the session data. + + Notes: + - If a session cookie exceeds 4KB in size it is split. In this case the key will be of the format + ``session-{segment number}``. + + """ + max_age: int = field(default=ONE_DAY_IN_SECONDS * 14) + """Maximal age of the cookie before its invalidated.""" + scopes: Scopes = field(default_factory=lambda: {ScopeType.HTTP, ScopeType.WEBSOCKET}) + """Scopes for the middleware - options are ``http`` and ``websocket`` with the default being both""" + path: str = field(default="/") + """Path fragment that must exist in the request url for the cookie to be valid. + + Defaults to ``'/'``. + """ + domain: str | None = field(default=None) + """Domain for which the cookie is valid.""" + secure: bool = field(default=False) + """Https is required for the cookie.""" + httponly: bool = field(default=True) + """Forbids javascript to access the cookie via 'Document.cookie'.""" + samesite: Literal["lax", "strict", "none"] = field(default="lax") + """Controls whether or not a cookie is sent with cross-site requests. Defaults to ``lax``.""" + exclude: str | list[str] | None = field(default=None) + """A pattern or list of patterns to skip in the session middleware.""" + exclude_opt_key: str = field(default="skip_session") + """An identifier to use on routes to disable the session middleware for a particular route.""" + store: str = "sessions" + """Name of the :class:`Store <.stores.base.Store>` to use""" + + def __post_init__(self) -> None: + if len(self.key) < 1 or len(self.key) > 256: + raise ImproperlyConfiguredException("key must be a string with a length between 1-256") + if self.max_age < 1: + raise ImproperlyConfiguredException("max_age must be greater than 0") + + def get_store_from_app(self, app: Litestar) -> Store: + """Get the store defined in :attr:`store` from an :class:`Litestar <.app.Litestar>` instance""" + return app.stores.get(self.store) |