Skip to content

Auth

Auth Backend

fastapi_admin_kit.auth.backend.AuthBackend

Bases: ABC

Abstract authentication backend — verify credentials & load users.

Source code in fastapi_admin_kit/auth/backend.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
class AuthBackend(ABC):
    """Abstract authentication backend — verify credentials & load users."""

    def __init__(self, auth_model: type | None = None) -> None:
        self._auth_model = auth_model

    @abstractmethod
    async def authenticate(
        self,
        credential: str,
        password: str,
        session: Any,
        login_field: str = "email",
    ) -> AdminUserProtocol | None:
        """Verify credentials. Return user object if valid, ``None`` otherwise."""
        ...

    @abstractmethod
    async def get_user(self, user_id: int | str, session: Any) -> AdminUserProtocol | None:
        """Load user by PK. Return ``None`` if not found or inactive."""
        ...

    async def on_logout(self, user_id: int | str | None = None) -> None:
        """Called after a user logs out. Override to perform cleanup."""
        # Default implementation does nothing
        return None

authenticate(credential, password, session, login_field='email') abstractmethod async

Verify credentials. Return user object if valid, None otherwise.

Source code in fastapi_admin_kit/auth/backend.py 
20
21
22
23
24
25
26
27
28
29
@abstractmethod
async def authenticate(
    self,
    credential: str,
    password: str,
    session: Any,
    login_field: str = "email",
) -> AdminUserProtocol | None:
    """Verify credentials. Return user object if valid, ``None`` otherwise."""
    ...

get_user(user_id, session) abstractmethod async

Load user by PK. Return None if not found or inactive.

Source code in fastapi_admin_kit/auth/backend.py 
31
32
33
34
@abstractmethod
async def get_user(self, user_id: int | str, session: Any) -> AdminUserProtocol | None:
    """Load user by PK. Return ``None`` if not found or inactive."""
    ...

on_logout(user_id=None) async

Called after a user logs out. Override to perform cleanup.

Source code in fastapi_admin_kit/auth/backend.py 
36
37
38
39
async def on_logout(self, user_id: int | str | None = None) -> None:
    """Called after a user logs out. Override to perform cleanup."""
    # Default implementation does nothing
    return None

Permission Checker

fastapi_admin_kit.auth.permissions.PermissionChecker

Per-request permission checker.

Instantiated once per request via Depends(get_permission_checker). Caches permission results in-memory for the lifetime of the request.

Merges permissions from: 1. All assigned roles (M2M via admin_user_roles) 2. Direct per-user overrides (UserPermission)

Role permissions are OR'd together, then direct overrides are OR'd on top.

Source code in fastapi_admin_kit/auth/permissions.py 
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
class PermissionChecker:
    """Per-request permission checker.

    Instantiated once per request via ``Depends(get_permission_checker)``.
    Caches permission results in-memory for the lifetime of the request.

    Merges permissions from:
    1. All assigned roles (M2M via admin_user_roles)
    2. Direct per-user overrides (UserPermission)

    Role permissions are OR'd together, then direct overrides are OR'd on top.
    """

    def __init__(
        self,
        session: AsyncSession,
        user: AdminUserProtocol,
        *,
        user_snapshot: dict[str, object] | None = None,
    ) -> None:
        self.session = as_session_backend(session)
        self.user = user
        snap = user_snapshot or {}
        self._is_superuser: bool = (
            bool(snap["is_superuser"]) if "is_superuser" in snap else bool(user.is_superuser)
        )
        self._role_ids: list[int] = (
            snap["role_ids"] if "role_ids" in snap else getattr(user, "role_ids", [])
        )
        self._user_id: int | str | None = snap.get("id") if snap else getattr(user, "id", None)
        self._role_cache: dict[str, PermissionSet | None] | None = None
        self._direct_cache: dict[str, PermissionSet | None] | None = None
        self._cache: dict[tuple[str, str], bool] = {}
        self._field_cache: dict[tuple[str, str], set[str] | None] = {}

    async def _load_role_permissions(self) -> dict[str, PermissionSet | None]:
        """Load and cache all role-based permissions, merged with OR logic."""
        if self._role_cache is not None:
            return self._role_cache

        self._role_cache = {}
        if not self._role_ids:
            return self._role_cache

        from sqlalchemy import select

        from fastapi_admin_kit.auth.models import admin_role_permissions

        for perm in await self.session.all(
            select(Permission)
            .join(
                admin_role_permissions,
                Permission.id == admin_role_permissions.c.permission_id,
            )
            .where(admin_role_permissions.c.role_id.in_(self._role_ids))
        ):
            table = perm.table_name
            if table not in self._role_cache:
                self._role_cache[table] = PermissionSet()
            ps = self._role_cache[table]
            if perm.can_view:
                ps.can_view = True
            if perm.can_create:
                ps.can_create = True
            if perm.can_edit:
                ps.can_edit = True
            if perm.can_delete:
                ps.can_delete = True
            if perm.can_export:
                ps.can_export = True
            if perm.can_import:
                ps.can_import = True

        return self._role_cache

    async def _load_direct_permissions(self) -> dict[str, PermissionSet | None]:
        """Load and cache direct per-user permission overrides."""
        if self._direct_cache is not None:
            return self._direct_cache

        self._direct_cache = {}
        if self._user_id is None:
            return self._direct_cache

        from sqlalchemy import select

        from fastapi_admin_kit.auth.models import Permission

        result = await self.session.execute(
            select(UserPermission, Permission)
            .join(Permission, UserPermission.permission_id == Permission.id)
            .where(UserPermission.user_id == self._user_id)
        )
        for up, perm in result:
            table = perm.table_name
            if table not in self._direct_cache:
                self._direct_cache[table] = PermissionSet()
            ps = self._direct_cache[table]
            if perm.can_view:
                ps.can_view = True
            if perm.can_create:
                ps.can_create = True
            if perm.can_edit:
                ps.can_edit = True
            if perm.can_delete:
                ps.can_delete = True
            if perm.can_export:
                ps.can_export = True
            if perm.can_import:
                ps.can_import = True

        return self._direct_cache

    async def _get_merged_permission(self, table_name: str, action: str) -> bool:
        """Get merged permission for a table+action across roles and direct overrides."""
        role_perms = await self._load_role_permissions()
        direct_perms = await self._load_direct_permissions()

        attr = f"can_{action}"

        role_ps = role_perms.get(table_name)
        direct_ps = direct_perms.get(table_name)

        role_val = getattr(role_ps, attr, False) if role_ps else False
        direct_val = getattr(direct_ps, attr, False) if direct_ps else False

        return role_val or direct_val

    async def has_permission(self, table_name: str, action: str) -> bool:
        """Return True if the current user may perform *action* on *table_name*.

        Actions: ``"view"`` | ``"create"`` | ``"edit"`` | ``"delete"`` | ``"export"`` | ``"import"``

        Superusers always return True. Results are cached per-request.
        """
        if self._is_superuser:
            return True

        cache_key = (table_name, action)
        if cache_key in self._cache:
            return self._cache[cache_key]

        result_bool = await self._get_merged_permission(table_name, action)
        self._cache[cache_key] = result_bool
        return result_bool

    async def get_allowed_fields(self, table_name: str, mode: str) -> set[str] | None:
        """Return the set of field names the user may access, or ``None``.

        mode: ``"view"`` | ``"edit"``

        Semantics:
        - ``None`` → no field-level restrictions exist → all fields allowed.
        - Empty ``set()`` → restriction rows exist but none grant access → no fields.
        - Non-empty ``set()`` → only those field names are permitted.
        """
        if self._is_superuser:
            return None

        cache_key = (table_name, mode)
        if cache_key in self._field_cache:
            return self._field_cache[cache_key]

        # No role and no direct permissions → all fields restricted
        if not self._role_ids and self._user_id is None:
            self._field_cache[cache_key] = set()
            return set()

        # No field-level restrictions in this system anymore
        self._field_cache[cache_key] = None
        return None

    def permission_set(self, table_name: str) -> PermissionSet:
        """Return a :class:`PermissionSet` for convenient template / UI use.

        Warning: This reads from the async-populated cache. If ``load_permissions()``
        was not called first, all values will be ``False`` (except for superusers).
        Always call ``await load_permissions(table_name)`` before using this method.
        """
        if self._is_superuser:
            return PermissionSet(
                can_view=True,
                can_create=True,
                can_edit=True,
                can_delete=True,
                can_export=True,
                can_import=True,
            )
        # Check if cache has been populated for this table
        cache_populated = any(key[0] == table_name for key in self._cache)
        if not cache_populated:
            import logging

            logging.getLogger(__name__).warning(
                "permission_set('%s') called before load_permissions() — "
                "returning default (all-False). Always call await load_permissions() first.",
                table_name,
            )
        return PermissionSet(
            can_view=self._cache.get((table_name, "view"), False),
            can_create=self._cache.get((table_name, "create"), False),
            can_edit=self._cache.get((table_name, "edit"), False),
            can_delete=self._cache.get((table_name, "delete"), False),
            can_export=self._cache.get((table_name, "export"), False),
            can_import=self._cache.get((table_name, "import"), False),
        )

    async def load_permissions(self, table_name: str) -> PermissionSet:
        """Async method to load and cache all permissions for a table.

        Call this before using ``permission_set()`` to ensure the cache is populated.
        """
        await self.has_permission(table_name, "view")
        await self.has_permission(table_name, "create")
        await self.has_permission(table_name, "edit")
        await self.has_permission(table_name, "delete")
        await self.has_permission(table_name, "export")
        await self.has_permission(table_name, "import")
        return self.permission_set(table_name)

has_permission(table_name, action) async

Return True if the current user may perform action on table_name.

Actions: "view" | "create" | "edit" | "delete" | "export" | "import"

Superusers always return True. Results are cached per-request.

Source code in fastapi_admin_kit/auth/permissions.py 
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
async def has_permission(self, table_name: str, action: str) -> bool:
    """Return True if the current user may perform *action* on *table_name*.

    Actions: ``"view"`` | ``"create"`` | ``"edit"`` | ``"delete"`` | ``"export"`` | ``"import"``

    Superusers always return True. Results are cached per-request.
    """
    if self._is_superuser:
        return True

    cache_key = (table_name, action)
    if cache_key in self._cache:
        return self._cache[cache_key]

    result_bool = await self._get_merged_permission(table_name, action)
    self._cache[cache_key] = result_bool
    return result_bool

get_allowed_fields(table_name, mode) async

Return the set of field names the user may access, or None.

mode: "view" | "edit"

Semantics: - None → no field-level restrictions exist → all fields allowed. - Empty set() → restriction rows exist but none grant access → no fields. - Non-empty set() → only those field names are permitted.

Source code in fastapi_admin_kit/auth/permissions.py 
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
async def get_allowed_fields(self, table_name: str, mode: str) -> set[str] | None:
    """Return the set of field names the user may access, or ``None``.

    mode: ``"view"`` | ``"edit"``

    Semantics:
    - ``None`` → no field-level restrictions exist → all fields allowed.
    - Empty ``set()`` → restriction rows exist but none grant access → no fields.
    - Non-empty ``set()`` → only those field names are permitted.
    """
    if self._is_superuser:
        return None

    cache_key = (table_name, mode)
    if cache_key in self._field_cache:
        return self._field_cache[cache_key]

    # No role and no direct permissions → all fields restricted
    if not self._role_ids and self._user_id is None:
        self._field_cache[cache_key] = set()
        return set()

    # No field-level restrictions in this system anymore
    self._field_cache[cache_key] = None
    return None

permission_set(table_name)

Return a :class:PermissionSet for convenient template / UI use.

Warning: This reads from the async-populated cache. If load_permissions() was not called first, all values will be False (except for superusers). Always call await load_permissions(table_name) before using this method.

Source code in fastapi_admin_kit/auth/permissions.py 
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
def permission_set(self, table_name: str) -> PermissionSet:
    """Return a :class:`PermissionSet` for convenient template / UI use.

    Warning: This reads from the async-populated cache. If ``load_permissions()``
    was not called first, all values will be ``False`` (except for superusers).
    Always call ``await load_permissions(table_name)`` before using this method.
    """
    if self._is_superuser:
        return PermissionSet(
            can_view=True,
            can_create=True,
            can_edit=True,
            can_delete=True,
            can_export=True,
            can_import=True,
        )
    # Check if cache has been populated for this table
    cache_populated = any(key[0] == table_name for key in self._cache)
    if not cache_populated:
        import logging

        logging.getLogger(__name__).warning(
            "permission_set('%s') called before load_permissions() — "
            "returning default (all-False). Always call await load_permissions() first.",
            table_name,
        )
    return PermissionSet(
        can_view=self._cache.get((table_name, "view"), False),
        can_create=self._cache.get((table_name, "create"), False),
        can_edit=self._cache.get((table_name, "edit"), False),
        can_delete=self._cache.get((table_name, "delete"), False),
        can_export=self._cache.get((table_name, "export"), False),
        can_import=self._cache.get((table_name, "import"), False),
    )

load_permissions(table_name) async

Async method to load and cache all permissions for a table.

Call this before using permission_set() to ensure the cache is populated.

Source code in fastapi_admin_kit/auth/permissions.py 
224
225
226
227
228
229
230
231
232
233
234
235
async def load_permissions(self, table_name: str) -> PermissionSet:
    """Async method to load and cache all permissions for a table.

    Call this before using ``permission_set()`` to ensure the cache is populated.
    """
    await self.has_permission(table_name, "view")
    await self.has_permission(table_name, "create")
    await self.has_permission(table_name, "edit")
    await self.has_permission(table_name, "delete")
    await self.has_permission(table_name, "export")
    await self.has_permission(table_name, "import")
    return self.permission_set(table_name)

Models

The built-in admin models are materialized at runtime from the schemas in fastapi_admin_kit.schemas.builtin. They live in fastapi_admin_kit.migrations.models (the deprecated fastapi_admin_kit.auth.models module re-exports the same classes and will be removed in v3.0).

User

fastapi_admin_kit.migrations.models.User = _backend.materialize(USER_SCHEMA, base=Base, schemas=BUILTIN_SCHEMAS) module-attribute

Role

fastapi_admin_kit.migrations.models.Role = _backend.materialize(ROLE_SCHEMA, base=Base, schemas=BUILTIN_SCHEMAS) module-attribute

Permission

fastapi_admin_kit.migrations.models.Permission = _backend.materialize(PERMISSION_SCHEMA, base=Base, schemas=BUILTIN_SCHEMAS) module-attribute

UserPermission

fastapi_admin_kit.migrations.models.UserPermission = _backend.materialize(USER_PERMISSION_SCHEMA, base=Base, schemas=BUILTIN_SCHEMAS) module-attribute

Dependencies

fastapi_admin_kit.auth.dependencies

FastAPI dependencies — session, current user, permission checker.

get_session(request)

Read and decode the admin session cookie from the current request.

Source code in fastapi_admin_kit/auth/dependencies.py 
45
46
47
48
49
def get_session(request: Request) -> dict[str, Any] | None:
    """Read and decode the admin session cookie from the current request."""
    session_backend = _get_session_backend(request)
    token = request.cookies.get(session_backend.cookie_name)
    return session_backend.decode(token)

get_current_admin_user(request, session_payload=Depends(get_session)) async

Resolve the logged-in admin user from the session cookie.

Raises 401 if the session is missing, invalid, the user is no longer active, or the password was changed after the session was issued.

Source code in fastapi_admin_kit/auth/dependencies.py 
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
async def get_current_admin_user(
    request: Request,
    session_payload: dict[str, Any] | None = Depends(get_session),
) -> AdminUserProtocol:
    """Resolve the logged-in admin user from the session cookie.

    Raises ``401`` if the session is missing, invalid, the user is
    no longer active, or the password was changed after the session was issued.
    """
    if session_payload is None:
        raise HTTPException(
            status_code=401,
            detail="Not authenticated. Please log in.",
        )

    user_id = session_payload.get("user_id")
    if user_id is None:
        raise HTTPException(status_code=401, detail="Invalid session payload.")

    # S18: every session token must carry an ``iat`` — without it the
    # password-change invalidation below would silently not apply.
    if session_payload.get("iat") is None:
        raise HTTPException(status_code=401, detail="Invalid session payload.")

    # request.state.admin_user is populated as a side effect.
    from fastapi_admin_kit.auth.identity import resolve_user

    user = await resolve_user(request, user_id)
    if user is None:
        raise HTTPException(status_code=401, detail="User not found or inactive.")

    # Check session invalidation: reject if password changed after session iat
    password_changed_at = getattr(user, "password_changed_at", None)
    session_iat = session_payload.get("iat")
    if password_changed_at is not None and session_iat is not None:
        from datetime import UTC, datetime

        if isinstance(password_changed_at, datetime):
            if password_changed_at.tzinfo is None:
                password_changed_at = password_changed_at.replace(tzinfo=UTC)
            if isinstance(session_iat, int | float):
                session_time = datetime.fromtimestamp(session_iat, tz=UTC)
                if password_changed_at > session_time:
                    raise HTTPException(
                        status_code=401,
                        detail="Session invalidated. Password was changed. Please log in again.",
                    )

    return user

get_permission_checker(request, user=Depends(get_current_admin_user), session=Depends(_get_db_session)) async

Build a PermissionChecker for the current user.

The concrete PermissionChecker class is imported here to avoid circular imports.

Source code in fastapi_admin_kit/auth/dependencies.py 
103
104
105
106
107
108
109
110
111
112
113
114
115
116
async def get_permission_checker(
    request: Request,
    user: AdminUserProtocol = Depends(get_current_admin_user),
    session: AsyncSession = Depends(_get_db_session),
) -> Any:
    """Build a ``PermissionChecker`` for the current user.

    The concrete ``PermissionChecker`` class is imported here to avoid
    circular imports.
    """
    from fastapi_admin_kit.auth.permissions import PermissionChecker

    snapshot = getattr(request.state, "admin_user_snapshot", None)
    return PermissionChecker(session=session, user=user, user_snapshot=snapshot)

require_permission(table_name, action)

Return a FastAPI dependency that enforces a permission check.

Usage::

@router.get("/")
async def list_view(_=Depends(require_permission("products", "view"))):
    ...
Source code in fastapi_admin_kit/auth/dependencies.py 
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
def require_permission(table_name: str, action: str):  # type: ignore[no-untyped-def]
    """Return a FastAPI dependency that enforces a permission check.

    Usage::

        @router.get("/")
        async def list_view(_=Depends(require_permission("products", "view"))):
            ...
    """

    async def _check(
        checker: Any = Depends(get_permission_checker),
    ) -> None:
        if not await checker.has_permission(table_name, action):
            raise HTTPException(
                status_code=403,
                detail=f"You do not have permission to {action} {table_name}.",
            )

    return _check

require_superuser(user=Depends(get_current_admin_user)) async

FastAPI dependency that enforces superuser access.

The single source of the "must be superuser" rule, shared by the roles and audit views (previously copy-pasted in each).

Source code in fastapi_admin_kit/auth/dependencies.py 
141
142
143
144
145
146
147
148
149
150
151
async def require_superuser(
    user: AdminUserProtocol = Depends(get_current_admin_user),
) -> AdminUserProtocol:
    """FastAPI dependency that enforces superuser access.

    The single source of the "must be superuser" rule, shared by the roles
    and audit views (previously copy-pasted in each).
    """
    if not getattr(user, "is_superuser", False):
        raise HTTPException(status_code=403, detail="Superuser access required.")
    return user

resolve_permission_checker(request) async

Resolve a PermissionChecker for the current request.

Returns None if the user is not authenticated or the checker cannot be built. This is the canonical implementation — import from here, not from renderers/factory.

Source code in fastapi_admin_kit/auth/dependencies.py 
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
async def resolve_permission_checker(request: Request) -> Any:
    """Resolve a ``PermissionChecker`` for the current request.

    Returns ``None`` if the user is not authenticated or the checker cannot be built.
    This is the canonical implementation — import from here, not from renderers/factory.
    """
    from fastapi_admin_kit.auth.identity import get_current_user_from_cookie
    from fastapi_admin_kit.auth.permissions import PermissionChecker

    user = await get_current_user_from_cookie(request)
    if user is None:
        return None

    from fastapi_admin_kit.db import get_db_session

    async_session = get_db_session(request)
    if async_session is None:
        return None

    snapshot = getattr(request.state, "admin_user_snapshot", None)
    return PermissionChecker(session=async_session, user=user, user_snapshot=snapshot)

Session

fastapi_admin_kit.auth.session

Session backend — ABC + signed-cookie implementation.

SessionBackend

Bases: ABC

Abstract session backend — encode/decode session payloads.

Source code in fastapi_admin_kit/auth/session.py 
52
53
54
55
56
57
58
59
60
61
62
63
class SessionBackend(ABC):
    """Abstract session backend — encode/decode session payloads."""

    @abstractmethod
    def encode(self, payload: dict[str, Any]) -> str:
        """Sign *payload* and return a token string suitable for a cookie."""
        ...

    @abstractmethod
    def decode(self, token: str | None) -> dict[str, Any] | None:
        """Verify *token* and return the payload dict, or ``None`` if invalid/expired."""
        ...

encode(payload) abstractmethod

Sign payload and return a token string suitable for a cookie.

Source code in fastapi_admin_kit/auth/session.py 
55
56
57
58
@abstractmethod
def encode(self, payload: dict[str, Any]) -> str:
    """Sign *payload* and return a token string suitable for a cookie."""
    ...

decode(token) abstractmethod

Verify token and return the payload dict, or None if invalid/expired.

Source code in fastapi_admin_kit/auth/session.py 
60
61
62
63
@abstractmethod
def decode(self, token: str | None) -> dict[str, Any] | None:
    """Verify *token* and return the payload dict, or ``None`` if invalid/expired."""
    ...

SignedCookieSessionBackend

Bases: SessionBackend

Session backend that signs a JSON payload using itsdangerous.

The cookie value is a URLSafeTimedSerializer-serialized string — the JSON payload is base64-encoded so its output is safe for Cookie headers (unlike raw JSON, which contains {, }, " and other characters that break the http.cookies.SimpleCookie parser).

Source code in fastapi_admin_kit/auth/session.py 
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
class SignedCookieSessionBackend(SessionBackend):
    """Session backend that signs a JSON payload using ``itsdangerous``.

    The cookie value is a ``URLSafeTimedSerializer``-serialized string — the JSON
    payload is base64-encoded so its output is safe for ``Cookie`` headers (unlike
    raw JSON, which contains ``{``, ``}``, ``\"`` and other characters that break
    the ``http.cookies.SimpleCookie`` parser).
    """

    COOKIE_NAME = "admin_session"

    def __init__(
        self,
        secret_key: str,
        session_ttl: int = 28800,
        cookie_name: str = COOKIE_NAME,
        secure: bool = False,
    ) -> None:
        self._secret_key = secret_key
        # Custom JSON module lets us encode UUIDs, datetimes, Decimals, etc.
        # that may appear in a session payload when a project supplies a
        # custom ``auth_model`` whose ``id`` is a UUID (e.g. SQLModel /
        # SQLAlchemy ``Uuid`` PK). Without this, ``json.dumps`` raises
        # ``TypeError: Object of type UUID is not JSON serializable`` the
        # first time a user with a UUID PK tries to log in.
        #
        # ``itsdangerous`` forwards ``**serializer_kwargs`` to ``dumps``, so
        # passing ``default=_json_default`` is the supported way to register
        # a custom encoder without subclassing.
        self._serializer = URLSafeTimedSerializer(
            secret_key,
            salt="admin-session",
            serializer_kwargs={"default": _json_default},
        )
        # Separate salt + short TTL: a pending-2FA token can never be
        # replayed as a full session cookie (different signing domain).
        self._mfa_serializer = URLSafeTimedSerializer(
            secret_key,
            salt="admin-2fa",
            serializer_kwargs={"default": _json_default},
        )
        self._mfa_ttl = 300
        self._session_ttl = session_ttl
        self.cookie_name = cookie_name
        self.secure = secure

    @property
    def secret_key(self) -> str:
        """The signing key used by this backend.

        Public accessor so CSRF / JWT signing can share the same key without
        reaching into the (private) itsdangerous signer. Swapping the session
        backend no longer silently breaks CSRF.
        """
        return self._secret_key

    def encode(self, payload: dict[str, Any]) -> str:
        """Sign *payload* and return the signed token."""
        if "iat" not in payload:
            payload["iat"] = time.time()
        return self._serializer.dumps(payload)

    def decode(self, token: str | None) -> dict[str, Any] | None:
        """Verify *token* and return the decoded payload dict, or ``None``.

        Tokens without an ``iat`` claim are rejected outright (S18): the
        ``iat`` timestamp powers the password-change invalidation check, and
        a token minted without it would silently bypass that revocation.
        """
        if not token:
            return None
        try:
            payload = self._serializer.loads(token, max_age=self._session_ttl)
        except (BadSignature, SignatureExpired, ValueError):
            return None
        if not isinstance(payload, dict) or "iat" not in payload:
            return None
        return payload

    def should_secure(self, request: Any) -> bool:
        """Return True only when the configured secure flag is set AND the
        current request arrived over HTTPS.  This prevents browsers from
        dropping the cookie on plain-HTTP dev servers."""
        if not self.secure:
            return False
        return getattr(request.url, "scheme", "") == "https"

    def load(self, token: str | None) -> dict[str, Any] | None:
        """Alias for decode — used by flash message system."""
        return self.decode(token)

    def encode_pending_2fa(self, user_id: int | str) -> str:
        """Sign a short-lived token proving credentials were verified but 2FA is pending."""
        return self._mfa_serializer.dumps({"user_id": user_id})

    def decode_pending_2fa(self, token: str | None) -> int | str | None:
        """Verify a pending-2FA token and return the user_id, or ``None``."""
        if not token:
            return None
        try:
            payload = self._mfa_serializer.loads(token, max_age=self._mfa_ttl)
        except (BadSignature, SignatureExpired, ValueError):
            return None
        return payload.get("user_id")

    def save(self, response: Any, data: dict[str, Any], *, request: Any | None = None) -> None:
        """Encode *data* and set it as a signed cookie on *response*."""
        token = self.encode(data)
        if request is not None:
            secure = self.should_secure(request)
        else:
            secure = self.secure
        response.set_cookie(
            key=self.cookie_name,
            value=token,
            max_age=self._session_ttl,
            path="/",
            secure=secure,
            httponly=True,
            samesite="strict",
        )

secret_key property

The signing key used by this backend.

Public accessor so CSRF / JWT signing can share the same key without reaching into the (private) itsdangerous signer. Swapping the session backend no longer silently breaks CSRF.

encode(payload)

Sign payload and return the signed token.

Source code in fastapi_admin_kit/auth/session.py 
122
123
124
125
126
def encode(self, payload: dict[str, Any]) -> str:
    """Sign *payload* and return the signed token."""
    if "iat" not in payload:
        payload["iat"] = time.time()
    return self._serializer.dumps(payload)

decode(token)

Verify token and return the decoded payload dict, or None.

Tokens without an iat claim are rejected outright (S18): the iat timestamp powers the password-change invalidation check, and a token minted without it would silently bypass that revocation.

Source code in fastapi_admin_kit/auth/session.py 
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
def decode(self, token: str | None) -> dict[str, Any] | None:
    """Verify *token* and return the decoded payload dict, or ``None``.

    Tokens without an ``iat`` claim are rejected outright (S18): the
    ``iat`` timestamp powers the password-change invalidation check, and
    a token minted without it would silently bypass that revocation.
    """
    if not token:
        return None
    try:
        payload = self._serializer.loads(token, max_age=self._session_ttl)
    except (BadSignature, SignatureExpired, ValueError):
        return None
    if not isinstance(payload, dict) or "iat" not in payload:
        return None
    return payload

should_secure(request)

Return True only when the configured secure flag is set AND the current request arrived over HTTPS. This prevents browsers from dropping the cookie on plain-HTTP dev servers.

Source code in fastapi_admin_kit/auth/session.py 
145
146
147
148
149
150
151
def should_secure(self, request: Any) -> bool:
    """Return True only when the configured secure flag is set AND the
    current request arrived over HTTPS.  This prevents browsers from
    dropping the cookie on plain-HTTP dev servers."""
    if not self.secure:
        return False
    return getattr(request.url, "scheme", "") == "https"

load(token)

Alias for decode — used by flash message system.

Source code in fastapi_admin_kit/auth/session.py 
153
154
155
def load(self, token: str | None) -> dict[str, Any] | None:
    """Alias for decode — used by flash message system."""
    return self.decode(token)

encode_pending_2fa(user_id)

Sign a short-lived token proving credentials were verified but 2FA is pending.

Source code in fastapi_admin_kit/auth/session.py 
157
158
159
def encode_pending_2fa(self, user_id: int | str) -> str:
    """Sign a short-lived token proving credentials were verified but 2FA is pending."""
    return self._mfa_serializer.dumps({"user_id": user_id})

decode_pending_2fa(token)

Verify a pending-2FA token and return the user_id, or None.

Source code in fastapi_admin_kit/auth/session.py 
161
162
163
164
165
166
167
168
169
def decode_pending_2fa(self, token: str | None) -> int | str | None:
    """Verify a pending-2FA token and return the user_id, or ``None``."""
    if not token:
        return None
    try:
        payload = self._mfa_serializer.loads(token, max_age=self._mfa_ttl)
    except (BadSignature, SignatureExpired, ValueError):
        return None
    return payload.get("user_id")

save(response, data, *, request=None)

Encode data and set it as a signed cookie on response.

Source code in fastapi_admin_kit/auth/session.py 
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
def save(self, response: Any, data: dict[str, Any], *, request: Any | None = None) -> None:
    """Encode *data* and set it as a signed cookie on *response*."""
    token = self.encode(data)
    if request is not None:
        secure = self.should_secure(request)
    else:
        secure = self.secure
    response.set_cookie(
        key=self.cookie_name,
        value=token,
        max_age=self._session_ttl,
        path="/",
        secure=secure,
        httponly=True,
        samesite="strict",
    )