summaryrefslogtreecommitdiff
path: root/venv/lib/python3.11/site-packages/litestar/middleware/session
diff options
context:
space:
mode:
authorcyfraeviolae <cyfraeviolae>2024-04-03 03:10:44 -0400
committercyfraeviolae <cyfraeviolae>2024-04-03 03:10:44 -0400
commit6d7ba58f880be618ade07f8ea080fe8c4bf8a896 (patch)
treeb1c931051ffcebd2bd9d61d98d6233ffa289bbce /venv/lib/python3.11/site-packages/litestar/middleware/session
parent4f884c9abc32990b4061a1bb6997b4b37e58ea0b (diff)
venv
Diffstat (limited to 'venv/lib/python3.11/site-packages/litestar/middleware/session')
-rw-r--r--venv/lib/python3.11/site-packages/litestar/middleware/session/__init__.py3
-rw-r--r--venv/lib/python3.11/site-packages/litestar/middleware/session/__pycache__/__init__.cpython-311.pycbin0 -> 287 bytes
-rw-r--r--venv/lib/python3.11/site-packages/litestar/middleware/session/__pycache__/base.cpython-311.pycbin0 -> 11124 bytes
-rw-r--r--venv/lib/python3.11/site-packages/litestar/middleware/session/__pycache__/client_side.cpython-311.pycbin0 -> 15608 bytes
-rw-r--r--venv/lib/python3.11/site-packages/litestar/middleware/session/__pycache__/server_side.cpython-311.pycbin0 -> 12648 bytes
-rw-r--r--venv/lib/python3.11/site-packages/litestar/middleware/session/base.py256
-rw-r--r--venv/lib/python3.11/site-packages/litestar/middleware/session/client_side.py264
-rw-r--r--venv/lib/python3.11/site-packages/litestar/middleware/session/server_side.py219
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
new file mode 100644
index 0000000..8748ce3
--- /dev/null
+++ b/venv/lib/python3.11/site-packages/litestar/middleware/session/__pycache__/__init__.cpython-311.pyc
Binary files differ
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
new file mode 100644
index 0000000..68a8b9c
--- /dev/null
+++ b/venv/lib/python3.11/site-packages/litestar/middleware/session/__pycache__/base.cpython-311.pyc
Binary files differ
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
new file mode 100644
index 0000000..692f54c
--- /dev/null
+++ b/venv/lib/python3.11/site-packages/litestar/middleware/session/__pycache__/client_side.cpython-311.pyc
Binary files differ
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
new file mode 100644
index 0000000..bd2373c
--- /dev/null
+++ b/venv/lib/python3.11/site-packages/litestar/middleware/session/__pycache__/server_side.cpython-311.pyc
Binary files differ
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)