Skip to content

Authentication in Castlecraft Engineer

Authentication in castlecraft-engineer is handled by the AuthenticationService located in castlecraft_engineer.application.auth. This service is responsible for verifying user identity, primarily through token-based mechanisms. It integrates caching for performance and supports standard OpenID Connect (OIDC) flows like JWKS fetching for token signature verification, token introspection, and fetching user information.

Core AuthenticationService Functionality

The AuthenticationService is designed to: * Verify tokens against a JSON Web Key Set (JWKS). * Introspect tokens via a dedicated endpoint. * Fetch user information from a UserInfo endpoint. * Utilize caching (synchronous and asynchronous Redis) to store JWKS responses, verified token payloads, and introspection results to reduce external calls.

Initialization

The service can be initialized with optional synchronous and asynchronous Redis cache clients: 

# from redis import Redis
# from redis.asyncio import Redis as AsyncRedis
from castlecraft_engineer.application.auth import AuthenticationService

# sync_cache = Redis(host='localhost', port=6666, db=0)
# async_cache = AsyncRedis(host='localhost', port=6666, db=0)

auth_service = AuthenticationService(
    # cache_client=sync_cache, # Optional synchronous Redis client
    # async_cache_client=async_cache # Optional asynchronous Redis client or an awaitable that resolves to one
)

Caching Strategy

The service employs caching for: * JWKS Response: Fetched from the JWKS_URL environment variable and cached with a TTL (configured by AUTH_JWKS_TTL_SEC, default 10800s). The cache key is JWKS_RESPONSE_KEY. * Verified Token Payloads: After successful ID token verification or token introspection, the resulting user/token data is cached. The cache key is prefixed with "bearer_token|" followed by the token itself. The TTL is derived from the token's exp claim or defaults to AUTH_TOKEN_TTL_SEC (environment variable, default 3600s). Pickle is used for serialization/deserialization of cached objects. Ensure that your Redis instance is secured and data sources are trusted. If Backchannel Logout is enabled, additional cache structures are used: * SID-to-Token Mapping: Maps a Session ID (sid) to a set of token cache keys. Prefixed with BACKCHANNEL_LOGOUT_SID_MAP_PREFIX (default "bcl_sid_map|"). TTL is configured by BACKCHANNEL_SID_MAP_TTL_SEC. * SUB-to-SID Mapping: Maps a Subject ID (sub) to a set of SIDs. Prefixed with BACKCHANNEL_LOGOUT_SUB_MAP_PREFIX (default "bcl_sub_map|"). TTL is configured by BACKCHANNEL_SUB_MAP_TTL_SEC. This is only used if ENABLE_LOGOUT_BY_SUB is true.

Key Operations

1. Fetching JWKS: get_active_jwks_response()

This asynchronous method retrieves the JSON Web Key Set. 1. Checks the cache (async or sync) for JWKS_RESPONSE_KEY. 2. If cached and valid (checked by _is_jwks_valid), returns the cached JWKS. 3. If not cached or invalid, fetches from the URL specified by the JWKS_URL environment variable. 4. Validates the fetched JWKS structure. 5. Caches the valid JWKS response. Returns None if the JWKS_URL environment variable is not set or if fetching/validation fails.

2. Verifying ID Tokens: verify_id_token(token: str)

This asynchronous method verifies an ID token (typically a JWT) signed by an external Identity Provider. 1. Fetches the active JWKS using get_active_jwks_response(). 2. Extracts the kid (Key ID) from the unverified token header. 3. Finds the corresponding public key in the JWKS. 4. Decodes and verifies the token using the public key, checking signature, expiration (exp), audience (from ALLOWED_AUD environment variable), issued at (iat), and not before (nbf) claims. A leeway of 60 seconds is applied for clock skew. 5. If verification is successful, the decoded user payload is cached (prefixed with "bearer_token|") with a TTL based on the token's exp claim. * If Backchannel Logout is enabled (see below), the token's sid (Session ID) and sub (Subject ID) are used to create mappings in the cache to facilitate session invalidation. Returns the decoded user payload (a dictionary) upon success, or None upon failure. The cached token is deleted on verification failure.

3. Introspecting Tokens: introspect_token(token: str)

This asynchronous method validates a token by calling an OAuth 2.0 Token Introspection endpoint. 1. Retrieves the introspection endpoint URL from the INTROSPECT_URL environment variable. 2. Constructs a POST request with the token (using the key from the INTROSPECT_TOKEN_KEY environment variable or its default value). 3. Optionally includes client credentials (from CLIENT_ID and CLIENT_SECRET environment variables) if the INTROSPECT_REQUIRES_AUTH environment variable is true. 4. If the introspection response indicates the token is active (active: true): a. The introspection response (a dictionary) is cached (prefixed with "bearer_token|") with a TTL based on its exp claim. b. If ENABLE_FETCH_USERINFO is true and USERINFO_URL is set (via environment variables), it then calls fetch_userinfo(). Returns the introspection response (possibly merged with userinfo) or None if introspection fails or the token is inactive. The cache is cleared for the token if introspection indicates inactivity.

4. Fetching UserInfo: fetch_userinfo(userinfo_url: str, token: str)

This synchronous method fetches user information from the OIDC UserInfo endpoint. 1. Makes a GET request to the provided userinfo_url with the access token in the Authorization: Bearer <token> header. Returns the userinfo response (a dictionary) or None on failure. Note: This method is called internally by introspect_token if configured, and its result is merged with the introspection data before caching.

5. Unified Token Verification: verify_user(token: str)

This is the primary asynchronous method applications should use to verify a token and get user data. 1. Checks the cache (async or sync) for the token (key: "bearer_token|" + token). 2. If a valid (non-expired) cached entry exists, it's returned. 3. If cached data is expired, it's deleted from the cache. If Backchannel Logout is enabled, an attempt is also made to clean up related SID-to-token mappings. 4. If not cached or expired: a. If ENABLE_VERIFY_ID_TOKEN (environment variable) is true, it attempts verify_id_token(token). If successful, the result is returned. b. If ID token verification is disabled or fails, and if ENABLE_INTROSPECT_TOKEN (environment variable) is true, it attempts introspect_token(token). If successful, the result is returned. 5. Returns None if all verification methods fail.

6. Validating Logout Tokens: validate_backchannel_logout_token(logout_token_jwt: str)

This asynchronous method validates a JWT specifically designed for backchannel logout, as per OpenID Connect Back-Channel Logout 1.0. This feature must be enabled via the ENABLE_BACKCHANNEL_LOGOUT environment variable. 1. Ensures backchannel logout is enabled and essential configurations like BACKCHANNEL_LOGOUT_TOKEN_ISS and audience (derived from ALLOWED_AUD or CLIENT_ID) are set. 2. Fetches the active JWKS. 3. Verifies the logout token's signature using the appropriate public key from JWKS. 4. Validates standard JWT claims: issuer (iss), audience (aud), expiration (exp), issued at (iat), JWT ID (jti). 5. Specifically validates backchannel logout claims: * Must contain either a sid (Session ID) or a sub (Subject ID) claim. * Must contain an events claim, which in turn must have a http://schemas.openid.net/event/backchannel-logout member that is an empty JSON object ({}). * Must NOT contain a nonce claim. Returns the decoded claims dictionary upon successful validation, or None otherwise.

7. Invalidating Sessions by SID: invalidate_sessions_by_sid(sid: str, sub: Optional[str] = None)

This asynchronous method invalidates all cached tokens associated with a given Session ID (sid). This is typically called after a backchannel logout token targeting a sid has been validated. 1. Retrieves all token cache keys mapped to the sid (from BACKCHANNEL_LOGOUT_SID_MAP_PREFIX + sid). 2. Deletes each of these token cache entries from the primary token cache. 3. Deletes the sid-to-tokens mapping itself. 4. If sub is provided (or can be derived from one of the token payloads before deletion) and ENABLE_LOGOUT_BY_SUB is true, it also removes the sid from the sub's set of SIDs (from BACKCHANNEL_LOGOUT_SUB_MAP_PREFIX + sub) to keep the SUB-to-SID mapping consistent. Returns True if successful (or if no sessions were found for the SID), False on error.

8. Invalidating Sessions by SUB: invalidate_sessions_by_sub(sub: str)

This asynchronous method invalidates all SIDs (and their associated tokens) for a given subject identifier (sub). This is typically called after a backchannel logout token targeting a sub has been validated, and ENABLE_LOGOUT_BY_SUB is true. 1. Retrieves all SIDs mapped to the sub (from BACKCHANNEL_LOGOUT_SUB_MAP_PREFIX + sub). 2. For each sid found, it effectively calls invalidate_sessions_by_sid(sid, sub=sub) to ensure all tokens for that SID are cleared and the SID is unlinked from the SUB. 3. Deletes the sub-to-SIDs mapping itself. Returns True if successful (or if no SIDs were found for the SUB), False on error.

Backchannel Logout Support Details

The AuthenticationService can be configured to support OpenID Connect Back-Channel Logout 1.0. This allows your application (as a Relying Party) to be notified by the OpenID Provider (OP) when a user's session has been terminated elsewhere (e.g., user logs out directly from OP, or another RP initiates logout). This enables the application to log out the user locally by invalidating their cached sessions.

Enabling Backchannel Logout

Set the ENABLE_BACKCHANNEL_LOGOUT environment variable to true. You can further enable invalidation by subject identifier (sub) by setting ENABLE_LOGOUT_BY_SUB to true (this requires ENABLE_BACKCHANNEL_LOGOUT to also be true).

How it Works

  1. Token Linking: When an ID token is successfully verified (via verify_id_token) and backchannel logout is enabled: If the token contains a Session ID (sid), a mapping is created in the cache: BACKCHANNEL_LOGOUT_SID_MAP_PREFIX + sid -> set of ("bearer_token|" + token). If the token also contains a subject identifier (sub) and ENABLE_LOGOUT_BY_SUB is true, another mapping is created: BACKCHANNEL_LOGOUT_SUB_MAP_PREFIX + sub -> set of sid. These mappings allow the service to efficiently find all active tokens for a given sid or all sids for a given sub.
  2. Logout Token Reception: Your application must expose an endpoint (URL) that the OP can call to deliver logout tokens. This endpoint is specific to your application and needs to be registered with the OP.
  3. Processing Logout Tokens: When your endpoint receives a logout token (a JWT) from the OP: a. Call claims = await auth_service.validate_backchannel_logout_token(logout_token_jwt) to validate it. b. If claims are valid: i. If claims.get('sid') is present, call await auth_service.invalidate_sessions_by_sid(claims['sid'], sub=claims.get('sub')). ii. Else if claims.get('sub') is present and ENABLE_LOGOUT_BY_SUB is true, call await auth_service.invalidate_sessions_by_sub(claims['sub']). c. Your endpoint should then respond to the OP, typically with a 200 OK HTTP status.

Configuration (Environment Variables)

The service heavily relies on environment variables for its configuration: You can also configure the AuthenticationService programmatically by passing an auth_config dictionary to the ContainerBuilder.with_authentication() method, which can override these environment variables.

  • JWKS_URL: URL to fetch the JSON Web Key Set.
  • ALLOWED_AUD: Comma-separated list of allowed audiences for ID token verification.
  • INTROSPECT_URL: URL of the token introspection endpoint.
  • INTROSPECT_TOKEN_KEY: The key name for the token in the introspection request form data (default: token).
  • INTROSPECT_REQUIRES_AUTH: Set to true if the introspection endpoint requires client authentication (default: false).
  • CLIENT_ID: Client ID for introspection endpoint authentication.
  • CLIENT_SECRET: Client secret for introspection endpoint authentication.
  • USERINFO_URL: URL of the UserInfo endpoint.
  • ENABLE_VERIFY_ID_TOKEN: Set to true to enable ID token verification flow (default: false).
  • ENABLE_INTROSPECT_TOKEN: Set to true to enable token introspection flow (default: false).
  • ENABLE_FETCH_USERINFO: Set to true to enable fetching userinfo after successful introspection (default: false).
  • AUTHENTICATION_REQUEST_VERIFY_SSL: Controls SSL verification for outgoing HTTP requests made by the service (e.g., to JWKS, introspection, or userinfo endpoints). Set to true or false (default: true).
  • AUTH_JWKS_TTL_SEC: Time-to-live in seconds for the cached JWKS response (default: 10800).
  • AUTH_TOKEN_TTL_SEC: Default time-to-live in seconds for cached token payloads when a token's own expiry cannot be used (e.g., introspection response without exp, or already expired ID tokens) (default: 3600).
  • ENABLE_BACKCHANNEL_LOGOUT: Set to true to enable backchannel logout functionality (default: false).
  • ENABLE_LOGOUT_BY_SUB: Set to true to enable invalidating all sessions for a subject ID (sub) when a logout token targeting that sub is received. Requires ENABLE_BACKCHANNEL_LOGOUT to be true (default: false).
  • BACKCHANNEL_LOGOUT_TOKEN_ISS: The expected issuer URL for backchannel logout tokens. This must be configured if backchannel logout is enabled.
  • BACKCHANNEL_SID_MAP_TTL_SEC: TTL in seconds for the SID-to-token_cache_key mapping in Redis (default: twice AUTH_TOKEN_TTL_SEC).
  • BACKCHANNEL_SUB_MAP_TTL_SEC: TTL in seconds for the SUB-to-SID mapping in Redis (default: thrice AUTH_TOKEN_TTL_SEC).
    • Note: The audience (aud) for backchannel logout tokens (BACKCHANNEL_LOGOUT_TOKEN_AUD) is derived from the ALLOWED_AUD environment variable. If ALLOWED_AUD is not set, it falls back to CLIENT_ID.

WARNING: Security Consideration Disabling SSL/TLS verification ("false") can expose your application to security risks, such as man-in-the-middle attacks. This option should only be used in controlled development or testing environments where you understand and accept the risks. For production environments, SSL verification should always be enabled ("true").

Usage in an Application

  1. Setup: Ensure Redis (sync and/or async) is available if caching is desired. Set the necessary environment variables.
  2. Instantiation: Create an instance of AuthenticationService, providing cache clients if used.
  3. Dependency Injection: Register the AuthenticationService instance with your DI container.
  4. Protecting Endpoints: In your web framework (e.g., FastAPI), create a dependency that retrieves the token from the request (e.g., Authorization: Bearer <token>) and calls await auth_service.verify_user(token).
    • If verify_user returns a user dictionary, authentication is successful. The user data can be passed to the route handler.
    • If verify_user returns None, authentication fails. An appropriate HTTP error (e.g., 401 Unauthorized) should be returned.
  5. (Optional) Backchannel Logout Endpoint: If using Backchannel Logout, create an endpoint (e.g., a POST request handler) that receives the logout token from your Identity Provider. This endpoint should: a. Call claims = await auth_service.validate_backchannel_logout_token(logout_token_jwt). b. If claims are valid: i. If claims.get('sid') is present, call await auth_service.invalidate_sessions_by_sid(claims['sid'], claims.get('sub')). ii. Else if claims.get('sub') is present and ENABLE_LOGOUT_BY_SUB (environment variable) is true, call await auth_service.invalidate_sessions_by_sub(claims['sub']). c. Respond appropriately to the Identity Provider (typically a 200 OK).

# Conceptual FastAPI dependency example
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from castlecraft_engineer.application.auth import AuthenticationService
from typing import Optional, Dict, Any

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/v1/auth/token") # Example token URL

async def get_current_verified_user(
    token: str = Depends(oauth2_scheme),
    auth_service: AuthenticationService = Depends() # Assuming DI setup
) -> Optional[Dict[str, Any]]: # Returns the user payload dict or None
    user_payload = await auth_service.verify_user(token)
    if not user_payload:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid authentication credentials",
            headers={"WWW-Authenticate": "Bearer"},
        )
    # You might want to transform user_payload into an AuthenticatedSubject-like object here
    # or perform additional checks (e.g., user active status).
    return user_payload

# Example FastAPI route
# @app.get("/users/me")
# async def read_users_me(current_user: Dict[str, Any] = Depends(get_current_verified_user)):
#     return current_user
The user dictionary returned by verify_user contains the claims from the token (if ID token verification) or the introspection response (possibly merged with userinfo). This data can then be used for authorization decisions.

Error Handling

The service logs errors during its operations (e.g., HTTP errors, JSON decoding errors, cache errors). Applications should handle the None return from verify_user to indicate authentication failure to the client.