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 
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
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 
38
39
40
41
42
43
44
45
46
47
@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 
49
50
51
52
@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 
54
55
56
57
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 
 16
 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
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 = 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

        result = await self.session.execute(
            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))
        )
        for perm in result.scalars():
            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

        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

        result = await self.session.execute(
            select(UserPermission).where(UserPermission.user_id == self._user_id)
        )
        for perm in result.scalars():
            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

        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"``

        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.

        Note: This is a sync convenience wrapper. For async contexts,
        use the individual async methods directly.
        """
        if self._is_superuser:
            return PermissionSet(
                can_view=True,
                can_create=True,
                can_edit=True,
                can_delete=True,
            )
        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),
        )

    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")
        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"

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

Source code in fastapi_admin_kit/auth/permissions.py 
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
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"``

    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 
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
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.

Note: This is a sync convenience wrapper. For async contexts, use the individual async methods directly.

Source code in fastapi_admin_kit/auth/permissions.py 
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
def permission_set(self, table_name: str) -> PermissionSet:
    """Return a :class:`PermissionSet` for convenient template / UI use.

    Note: This is a sync convenience wrapper. For async contexts,
    use the individual async methods directly.
    """
    if self._is_superuser:
        return PermissionSet(
            can_view=True,
            can_create=True,
            can_edit=True,
            can_delete=True,
        )
    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),
    )

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 
197
198
199
200
201
202
203
204
205
206
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")
    return self.permission_set(table_name)

Models

User

fastapi_admin_kit.auth.models.User

Bases: AuthModelMixin, Base

Admin user model with authentication support via AuthModelMixin.

The mixin provides: hashed_password, is_active, is_superuser columns, role_ids property, verify_password() and hash_password() methods.

Source code in fastapi_admin_kit/auth/models.py 
 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
class User(AuthModelMixin, Base):
    """Admin user model with authentication support via AuthModelMixin.

    The mixin provides: hashed_password, is_active, is_superuser columns,
    role_ids property, verify_password() and hash_password() methods.
    """

    __tablename__ = "admin_users"

    id = Column(Integer, primary_key=True)
    email = Column(String(255), unique=True, nullable=False)
    full_name = Column(String(255))
    last_login = Column(DateTime(timezone=True))
    created_at = Column(DateTime(timezone=True), server_default=func.now())
    password_changed_at = Column(DateTime(timezone=True), nullable=True)

    # Many-to-many roles
    roles = relationship("Role", secondary=admin_user_roles, back_populates="users")
    # Direct permission overrides
    direct_permissions = relationship(
        "UserPermission",
        back_populates="user",
        cascade="all, delete-orphan",
    )
    refresh_tokens = relationship(
        "RefreshToken", back_populates="user", cascade="all, delete-orphan"
    )
    totp = relationship(
        "UserTOTP",
        back_populates="user",
        uselist=False,
        cascade="all, delete-orphan",
    )

    def __str__(self) -> str:
        return str(self.email)

    def __repr__(self) -> str:
        return f"<User {self.email!r}>"

Role

fastapi_admin_kit.auth.models.Role

Bases: Base

Source code in fastapi_admin_kit/auth/models.py 
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
class Role(Base):
    __tablename__ = "admin_roles"

    id = Column(Integer, primary_key=True)
    name = Column(String(100), unique=True, nullable=False)
    description = Column(Text)
    created_at = Column(DateTime(timezone=True), server_default=func.now())

    users = relationship("User", secondary=admin_user_roles, back_populates="roles")
    permissions = relationship(
        "Permission", secondary=admin_role_permissions, back_populates="roles"
    )

    def __str__(self) -> str:
        return str(self.name)

    def __repr__(self) -> str:
        return f"<Role {self.name!r}>"

Permission

fastapi_admin_kit.auth.models.Permission

Bases: Base

Permission matrix per model — shared across roles via M2M.

Source code in fastapi_admin_kit/auth/models.py 
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
class Permission(Base):
    """Permission matrix per model — shared across roles via M2M."""

    __tablename__ = "admin_permissions"
    __table_args__ = (UniqueConstraint("name", name="uq_admin_perm_name"),)

    id = Column(Integer, primary_key=True)
    name = Column(String(255), nullable=False)
    table_name = Column(String(255), nullable=False)
    can_view = Column(Boolean, default=False)
    can_create = Column(Boolean, default=False)
    can_edit = Column(Boolean, default=False)
    can_delete = Column(Boolean, default=False)

    roles = relationship("Role", secondary=admin_role_permissions, back_populates="permissions")

    def __str__(self) -> str:
        return str(self.name)

    def __repr__(self) -> str:
        return f"<Permission name={self.name!r} table={self.table_name!r}>"

UserPermission

fastapi_admin_kit.auth.models.UserPermission

Bases: Base

Direct per-user permission overrides — merged with role permissions.

Source code in fastapi_admin_kit/auth/models.py 
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
class UserPermission(Base):
    """Direct per-user permission overrides — merged with role permissions."""

    __tablename__ = "admin_user_permissions"
    __table_args__ = (
        UniqueConstraint("user_id", "table_name", name="uq_admin_user_perm_user_table"),
    )

    id = Column(Integer, primary_key=True)
    user_id = Column(
        Integer,
        ForeignKey("admin_users.id", ondelete="CASCADE"),
        nullable=False,
    )
    table_name = Column(String(255), nullable=False)
    can_view = Column(Boolean, default=False)
    can_create = Column(Boolean, default=False)
    can_edit = Column(Boolean, default=False)
    can_delete = Column(Boolean, default=False)

    user = relationship("User", back_populates="direct_permissions")

    def __str__(self) -> str:
        return f"{self.table_name} (user {self.user_id})"

    def __repr__(self) -> str:
        return f"<UserPermission user={self.user_id} table={self.table_name!r}>"

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
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.")

    # 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 
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
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 
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
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 
136
137
138
139
140
141
142
143
144
145
146
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

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 
16
17
18
19
20
21
22
23
24
25
26
27
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 
19
20
21
22
@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 
24
25
26
27
@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 
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
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
        self._serializer = URLSafeTimedSerializer(secret_key, salt="admin-session")
        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, or ``None``."""
        if not token:
            return None
        try:
            return self._serializer.loads(token, max_age=self._session_ttl)
        except (BadSignature, SignatureExpired, ValueError):
            return None

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 
64
65
66
67
68
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, or None.

Source code in fastapi_admin_kit/auth/session.py 
70
71
72
73
74
75
76
77
def decode(self, token: str | None) -> dict[str, Any] | None:
    """Verify *token* and return the decoded payload, or ``None``."""
    if not token:
        return None
    try:
        return self._serializer.loads(token, max_age=self._session_ttl)
    except (BadSignature, SignatureExpired, ValueError):
        return None