Docs / Introduction / Architecture

Architecture

A deep dive into every layer of AllSafe Fast: design principles, the layer overview, the fast path, core services, storage abstraction, the hook system, the plugin system, and domain models.

Design Principles

Six principles guide every decision in the framework. Understanding them makes the rest of the architecture self-explanatory.

⚡

Async-First

Every database, Redis, and HTTP operation is async. The framework never blocks the event loop. All storage protocol methods are async, all provider methods are async, and the in-memory backend uses asyncio locks.

📦

Protocol-Based Storage

Core logic depends on typing.Protocol interfaces, not concrete implementations. Swap in-memory for SQLAlchemy, or write your own adapter for MongoDB or DynamoDB — the core never changes.

🛡

Secure by Default

Every security decision defaults to the safest option. Cookies are HttpOnly + Secure + SameSite=Lax. Tokens are hashed before storage. Passwords use Argon2id. Production mode refuses insecure configs.

🧩

Provider-Driven

Authentication methods are pluggable providers implementing the AuthProvider protocol. Add password auth, OAuth, OIDC, or custom SSO without touching core logic. The engine orchestrates; providers do the work.

📝

Domain Model Separation

Domain models (User, Account, Session, Verification) are plain @dataclass objects — no ORM inheritance. The SQLAlchemy layer converts between ORM and domain objects via _to_domain() functions.

🚀

Fast Path

JWT verification happens entirely in memory on every request — no database or Redis query. The token's claims contain everything needed: user ID, email, verification status, roles, permissions.

Layer Overview

AllSafe Fast is organized into six layers, each with a single responsibility. Dependencies flow downward — the top layer depends on the bottom, never the reverse.

Your FastAPI App Your routes use auth.user() as a dependency. The framework handles everything else. You never interact with core services directly — only with the UserPrincipal returned by the dependency.

↓

API Layer (api/) FastAPI dependencies (dependencies.py) extract the JWT from the Authorization header or cookie, verify the CSRF header if using cookies, decode and validate the JWT in memory, build a UserPrincipal, and enforce role/permission checks. The router builder (router.py) combines core routes (session, refresh, sign-out) with routes from all registered providers and plugins.

↓

Core Services (core/) Six services contain all business logic: AuthenticationEngine, IdentityService, AccountService, SessionManager, AuthorizationService, VerificationService. Plus HookRegistry for lifecycle events. These never import SQLAlchemy — only storage Protocols and domain models.

↓

Providers (providers/) Five built-in providers implement the AuthProvider protocol: Password, OAuth 2.0, OIDC, Magic Link, Email OTP. Each has its own routes, schemas, and authentication logic. The ProviderRegistry manages registration and lookup by ID.

↓

Storage Layer (storage/) Four Protocol interfaces (UserStoreProtocol, AccountStoreProtocol, SessionStoreProtocol, VerificationStoreProtocol) bundled into StorageAdapter. Two implementations: in-memory (asyncio-lock-guarded) and async SQLAlchemy (PostgreSQL with ARRAY types).

↓

Security Layer (security/) Cryptographic primitives: Argon2id password hashing, HS256 JWT signing/verification, secure token generation (secrets.token_urlsafe), SHA-256 token hashing, constant-time comparison (hmac.compare_digest), cookie configuration, and secret key generation.

The Fast Path — In Detail

The most important architectural decision in AllSafe Fast is the fast path: JWT verification happens entirely in memory on every request. No database query, no Redis lookup, no network call. Here is exactly what happens when a request hits a protected route:

Token Extraction The user() dependency factory creates a FastAPI dependency. FastAPI calls it with the Request and the optional bearer token from OAuth2PasswordBearer(auto_error=False). The _extract_token() function checks: if a bearer token is present in the Authorization header, use it. Otherwise, check for the allsafe_access or allsafe_session cookie. If a cookie is found, the X-AllSafe-CSRF header must also be present — otherwise a 403 Forbidden is raised.

↓

JWT Verification The JWTService.verify_and_build_principal() method calls jwt.decode() from python-jose. This verifies: (a) the HMAC-SHA256 signature against the secret key, (b) the exp claim (token not expired), (c) the iss claim (matches configured issuer), (d) the aud claim (matches configured audience). If any check fails, a TokenExpiredError or TokenInvalidError is raised, which the dependency converts to a 401 Unauthorized response.

↓

Principal Construction A UserPrincipal (frozen dataclass) is built from the JWT claims: id (parsed from sub as UUID), email, email_verified, roles (tuple), permissions (tuple). The is_authenticated property always returns True. The has_role() and has_permission() methods do simple tuple membership checks.

↓

Authorization Checks If the route specified verified=True, AuthorizationService.check_verified() is called — it checks principal.email_verified. If a role was specified, check_role() checks principal.has_role(role). Similarly for permission, roles (any-of), and permissions (all-of). All checks are in-memory tuple lookups. On failure, an AuthorizationError is raised and converted to a 403 Forbidden response.

↓

Route Handler Execution The UserPrincipal is injected into the route handler as the user parameter. The handler can access user.id, user.email, user.email_verified, user.roles, and user.permissions directly. No database query was needed for any of this.

ℹ

Why no database query?

JWT tokens are self-contained: the user's identity, roles, and permissions are embedded in the token's claims. Since the signature is verified with the secret key, the claims can be trusted without looking up the user in the database on every request. The database is only consulted during sign-in, sign-up, token refresh, and password reset — not on every authenticated request.

Core Services — Deep Dive

Six services contain all business logic. They never import SQLAlchemy — only storage Protocols and domain models.

AuthenticationEngine async

The orchestrator. It coordinates the full sign-in flow by calling providers, identity, accounts, and sessions in sequence. It also contains the static verify_token() method used by the fast path.

authenticate()

async Looks up the provider by ID in the registry, calls provider.authenticate(context), raises AuthenticationError on failure, calls provider.get_identity(result), resolves/creates the user via IdentityService, ensures an account exists via AccountService, and creates a session via SessionManager. Returns a TokenPair.

verify_token() static

Decodes and validates a JWT in memory using JWTService.verify_access_token(), extracts claims (sub, email, email_verified, roles, permissions), and returns a UserPrincipal. No database or Redis query.

IdentityService async

Resolves a provider identity to a local User. If the email is new, a user record is created. If the email exists, the profile is updated with any new verified information.

resolve_or_create_user()

Looks up user by email. If found, calls _maybe_update(). If not found, calls _create_user(). Raises ValidationError if the provider identity has no email.

_create_user()

Creates a User with UUID, is_active=True, empty roles/permissions. Fires before_user_create and after_user_create hooks. Saves to storage.

_maybe_update()

Updates existing user if: provider says email is verified but user's isn't, provider has a name but user doesn't, provider has an image but user doesn't. Only saves if something changed.

get_user_by_id()

Looks up user by UUID. Raises NotFoundError if not found.

update_user()

Updates a user with a dict of field changes. Uses user.with_fields() (immutable copy) and saves.

AccountService async

Manages the link between a user and an authentication provider. Each user can have multiple accounts — one for password, one for Google, one for GitHub, etc.

🔒

Security Policy: No Auto-Merge

Accounts are never auto-merged based on email match alone. If someone signs in with Google and their email matches an existing user, a new account is linked to that user — but only during the authentication flow. Explicit account linking (link_account()) requires authenticated_user_id to equal user_id. This prevents account takeover via provider manipulation.

find_account()

Looks up an account by provider name + provider_account_id. Returns None if not found.

create_account()

Creates an Account record. Raises ConflictError if an account already exists for this provider + provider_account_id. Stores OAuth tokens, scope, and expiry.

link_account()

Links a new provider to an existing user. Requires authenticated_user_id == user_id. If the account is already linked to this user, returns it. If linked to another user, raises AccountLinkError.

unlink_account()

Removes a provider account. Verifies the account belongs to the user before deleting.

list_accounts()

Returns all accounts linked to a user.

update_password_hash()

Updates the password hash on the account for a given provider. Used by password reset and change-password flows.

SessionManager async

Creates JWT access tokens and refresh tokens. Refresh tokens are stored as SHA-256 hashes — the raw token is returned to the caller once and never persisted.

create_session()

Generates JWT access token (with all claims: sub, iss, aud, iat, exp, jti, email, email_verified, roles, permissions). Generates raw refresh token (32 bytes, secrets.token_urlsafe). Hashes both with SHA-256. Creates a Session record with user_id, token hashes, expiry, IP, user agent. Fires before_session_create and after_session_create hooks. Returns TokenPair.

refresh_session()

Hashes the refresh token, looks up the session by hash. If session is None → TokenInvalidError. If session is revoked → theft detection: revokes ALL user sessions, raises TokenReuseError. If expired → revokes session, raises AuthenticationError. Otherwise: revokes old session, loads user, creates a new session with the user's current roles/permissions.

revoke_session()

Marks a single session as revoked by setting revoked_at timestamp.

revoke_all_user_sessions()

Revokes all non-revoked sessions for a user. Used by password reset and theft detection.

get_session() / list_sessions()

Look up sessions by ID or by user. Returns Session domain objects.

AuthorizationService static

Stateless checks against a UserPrincipal. All methods are static and raise AuthorizationError on failure, return None on success.

check_authenticated()

Verifies principal is not None and principal.is_authenticated is True.

check_verified()

Calls check_authenticated(), then verifies principal.email_verified is True.

check_role(role)

Calls check_authenticated(), then verifies principal.has_role(role).

check_permission(perm)

Calls check_authenticated(), then verifies principal.has_permission(perm).

check_any_role(roles)

Calls check_authenticated(), then verifies any(principal.has_role(r) for r in roles).

check_all_permissions(perms)

Calls check_authenticated(), then verifies all permissions are present. Reports which are missing.

VerificationService async

Creates and consumes short-lived verification tokens for email verification, password reset, magic links, and email OTP.

Token Type	Expiry	Format	Max Attempts
`email_verify`	1 hour	32-byte URL-safe random	3
`password_reset`	1 hour	32-byte URL-safe random	3
`email_otp`	15 minutes	6-digit numeric (`secrets.randbelow`)	3
`magic_link`	15 minutes	32-byte URL-safe random	3

create_verification()

Generates a raw token (random or 6-digit OTP), hashes it with SHA-256, creates a Verification record with user_id, token_hash, type, email, expiry, attempts=0. Returns the raw token once.

verify_token()

Hashes the raw token, looks up the record. Checks: type matches, not already used, not expired, attempts < 3, hash matches (constant-time comparison via hmac.compare_digest). On success: marks as used, returns user_id. On failure: increments attempts or deletes the token.

cleanup_expired()

Deletes all expired verification records. Returns count of deleted records.

Storage Abstraction

The core services never import SQLAlchemy. They depend on four Protocol interfaces, each with async methods. The StorageAdapter bundles all four into one object and proxies methods for convenience.

The Four Protocols

UserStoreProtocol

get_by_id, get_by_email, create, update, delete — all async, all take/return User domain objects.

AccountStoreProtocol

get_by_id, get_by_user, get_by_provider, create, update, delete — all async, all take/return Account domain objects.

SessionStoreProtocol

get_by_id, get_by_user, get_by_token_hash, create, update, revoke, revoke_all_user_sessions, get_user — all async.

VerificationStoreProtocol

get_by_token_hash, create, mark_used, increment_attempts, delete, delete_expired — all async.

Implementations

Implementation	Use Case	Thread Safety	Persistence
InMemoryUserStore, InMemoryAccountStore, InMemorySessionStore, InMemoryVerificationStore	Development, testing	`asyncio.Lock` per store	Lost on restart
SQLAlchemyUserStore, SQLAlchemyAccountStore, SQLAlchemySessionStore, SQLAlchemyVerificationStore	Production	Async session per request	PostgreSQL

StorageAdapter

The StorageAdapter bundles all four stores and proxies their methods. Core services can call storage.get_by_email(...) or storage.get_by_provider(...) without knowing about sub-stores. The sub-stores are also accessible directly as storage.users, storage.accounts, storage.sessions, storage.verifications.

💡

Write your own adapter

Implement the four Protocols for any backend — MongoDB, DynamoDB, a remote API — and pass your StorageAdapter to Auth(). Core services won't know the difference. The Protocols are structural (runtime_checkable), so you don't need to inherit from them.

Hook System

AllSafe Fast emits 14 lifecycle events at key points. You can register handlers to run custom logic. Handlers are synchronous, fire in registration order, and exceptions are logged but never break the flow.

Available Events

Event	When It Fires	Handler Receives
`before_user_create`	Before a new user is saved to the database	`user`, `identity`
`after_user_create`	After a new user is successfully created	`user` (created), `identity`
`before_sign_in`	Before authentication begins	Context-dependent
`after_sign_in`	After successful authentication	`user_id` (TokenPair)
`before_session_create`	Before a session is saved	`session`
`after_session_create`	After a session is created and tokens returned	`session`
`before_session_revoke`	Before a session is revoked	Context-dependent
`after_session_revoke`	After a session is revoked	Context-dependent
`before_password_reset`	Before a password is reset	Context-dependent
`after_password_reset`	After a password is successfully reset	Context-dependent
`before_email_verify`	Before email verification is processed	Context-dependent
`after_email_verify`	After a user's email is verified	`user` (updated)
`before_account_link`	Before a new provider account is linked	Context-dependent
`after_account_link`	After a provider account is linked	Context-dependent

Registration

# Method 1: Decorator
@auth.hooks.on("after_user_create")
def handler(user, identity):
    send_welcome_email(user.email)

# Method 2: Dict at construction
auth = Auth(hooks={
    "after_user_create": [handler1, handler2],
    "after_sign_in": [handler3],
})

# Method 3: register_many
auth.hooks.register_many({
    "after_user_create": [handler1],
})

ℹ

Hook Safety

Hook handlers run synchronously in registration order. If a handler raises an exception, the error is logged via logging.getLogger("allsafe_fast.hooks") but does not break the authentication flow. This ensures a misbehaving hook never prevents users from signing in.

Plugin System

Plugins extend AllSafe Fast with new features. A plugin implements the AllSafePlugin protocol and can contribute routes, database tables, hooks, and schemas.

PluginContext

When Auth is constructed, it calls plugin.setup(context) for each plugin. The PluginContext provides access to:

config

The AuthConfig instance

storage

The StorageAdapter for database access

jwt_service

The JWTService for token operations

session_manager

The SessionManager for session operations

hooks

The HookRegistry for registering event handlers

extra

A dict for additional context (currently contains "engine")

Plugin Protocol Methods

Method	Required	Returns
`setup(context)`	Yes	`None` — initialize the plugin
`get_router()`	Yes	`APIRouter \| None` — plugin routes
`get_schemas()`	Yes	`list` — Pydantic models for request/response
`get_hooks()`	Yes	`dict[str, list[Callable]]` — event handlers
`get_table_metadata()`	Yes	`list` — SQLAlchemy table definitions

How Plugins Are Wired

Auth.__init__ creates a PluginRegistry and registers all plugins
For each plugin, plugin.setup(PluginContext(...)) is called
Plugin hooks are collected via plugin.get_hooks() and merged into the global HookRegistry
When auth.router is accessed, build_auth_router() calls plugins.collect_routers() and includes each plugin's router

Domain Models

Four plain @dataclass objects form the domain model. They have no ORM inheritance — the SQLAlchemy layer converts between ORM objects and these domain objects.

User

id: UUID

Unique identifier (generated with uuid4())

email: str

Email address (always lowercased before storage)

email_verified: bool

Whether the email has been verified (default: False)

name: str | None

Display name (from OAuth provider or sign-up form)

image: str | None

Avatar URL (from OAuth provider)

is_active: bool

Whether the user can sign in (default: True)

roles: list[str]

Role assignments (embedded in JWT claims)

permissions: list[str]

Permission assignments (embedded in JWT claims)

created_at / updated_at

UTC timestamps

Methods: with_fields(**changes) (immutable copy via dataclasses.replace), to_principal_dict() (for JWT claims), is_disabled().

Account

id: UUID

Unique identifier

user_id: UUID

The linked user

provider: str

Provider ID: "password", "google", "github", etc.

provider_account_id: str | None

The provider's unique ID for this user

password_hash: str | None

Argon2id hash (only for password provider)

access_token: str | None

OAuth access token (for provider API calls)

refresh_token: str | None

OAuth refresh token

access_token_expires_at

When the OAuth access token expires

scope: str | None

OAuth scopes granted

Methods: with_fields(), is_password_account(), public_dict() (excludes sensitive fields).

Session

id: UUID

Session identifier

user_id: UUID

The authenticated user

token_hash: str

SHA-256 hash of the JWT access token (64 hex chars)

refresh_token_hash: str

SHA-256 hash of the refresh token (unique, 64 hex chars)

expires_at: datetime

When the refresh token expires (30 days default)

ip_address: str | None

Client IP at sign-in time

user_agent: str | None

Client user agent at sign-in time

revoked_at: datetime | None

When the session was revoked (None = active)

Methods: is_revoked(), is_expired(), public_dict().

Verification

id: UUID

Verification record identifier

user_id: UUID

The user requesting verification

token_hash: str

SHA-256 hash of the verification token

type: str

One of: "email_verify", "password_reset", "email_otp", "magic_link"

email: str

The email being verified or reset

expires_at: datetime

When the token expires (15 min or 1 hour depending on type)

used_at: datetime | None

When the token was consumed (None = unused)

attempts: int

Number of failed verification attempts (max 3)

Methods: is_expired(), is_used(), with_fields().

← PreviousFeatures Next →Roadmap

Edit this page on GitHub →