altaforge.auth

Shared FastAPI authentication for AltaForge services: middleware, pluggable backends, the trusted-subsystem (mint) model, Casbin RBAC, and the organization registry. For the design and adoption guide, see Trusted-subsystem auth.

altaforge.auth

Shared FastAPI authentication middleware for AltaForge services.

ClientCredentialsAuthBackend

ClientCredentialsAuthBackend(org_resolver=None, *, expected_tenant_id=None)

Validates an Entra service-account (client-credentials / M2M) token.

The machine-to-machine counterpart of :class:JwtAuthBackend, for headless callers with no interactive user (a backend service authenticating as itself). Differences from the user backend:

• Audience is the target service's API app-registration id (AZURE_API_APP_CLIENT_ID), distinct from the user-facing AZURE_APP_CLIENT_ID a delegated-user token carries.
• Identity is the calling service principal, taken from the azp (or appid) claim — there is no user-identifier claim. The returned user_id is service:<app-id>.
• Roles come from the roles claim (Entra application roles).

Organizations are resolved via the injected OrgResolver keyed on that service identity — exactly as for a user — so tenancy enforcement downstream is identical regardless of caller kind.

Initialise the backend.

Parameters:

Name	Type	Description	Default
org_resolver	OrgResolver | None	Resolver invoked once per request to hydrate the service principal's organizations. When omitted, organizations is always an empty list.	None
expected_tenant_id	str | None	When set, the token's tid must match it.	None

Raises:

Type	Description
RuntimeError	If AZURE_API_APP_CLIENT_ID (or the issuer/JWKS env) is not set.

authenticate async

authenticate(request)

Validate the request's service-account Bearer token and return UserData.

CompositeAuthBackend

CompositeAuthBackend(routes, default, *, claim='iss')

Routes each request to a backend by one unverified claim on the token.

The routing claim (iss by default) is read WITHOUT verifying the signature — only to choose which backend validates the token. The selected backend then performs the real cryptographic checks, so a forged claim merely routes to a backend that rejects it. This is the seam that lets one endpoint accept, e.g., delegated user tokens and service-account tokens side by side without either backend knowing about the other.

Choose claim for what actually distinguishes the token kinds you serve:

• iss — when the kinds come from different issuers (e.g. an Entra tenant vs a local IP-allowlist issuer).
• aud — when a delegated user token and a service-account (client-credentials) token come from the same Entra tenant (same iss) but target different app registrations. This is the DocGen case: route the API app-registration id to a ClientCredentialsAuthBackend and fall through to a user JwtAuthBackend.

Initialise with a claim-value→backend map and a fallback backend.

Parameters:

Name	Type	Description	Default
routes	dict[str, AuthBackend]	Maps an exact value of the routing claim to the backend that should validate tokens carrying it.	required
default	AuthBackend	Backend used when the routing claim is absent or not in the map (so unknown tokens still get validated, and rejected, by a real backend rather than silently passing).	required
claim	str	The top-level token claim to route on. Defaults to iss.	'iss'

authenticate async

authenticate(request)

Pick a backend by the unverified routing claim, then delegate real validation to it.

FakeAuthBackend

FakeAuthBackend(is_admin=False, org_resolver=None)

Returns a hardcoded local-dev user. Never use in production.

When an OrgResolver is supplied and returns a non-empty list, that list is used; otherwise the value of FAKE_AUTH_ORGANIZATIONS is used (or the single-element list ['local'] when the env var is unset). This keeps the local-dev path consistent with whatever Casbin/RBAC policy the app has wired up, while still providing a sensible default for the cold-start case.

Build a fake user and log a startup warning.

authenticate async

authenticate(request)

Return the hardcoded local-dev user.

InternalTokenAuthBackend

InternalTokenAuthBackend(minter, org_resolver=None)

Validates the backend's own internal token from the Authorization header.

This is the adapter-agnostic seam: whatever upstream provider the BFF used (Entra, Google, dev), it forwards a single internal token whose issuer is this backend. Roles ride in the token (Casbin-sourced at mint); organizations are resolved per request via the unchanged OrgResolver path, exactly as :class:JwtAuthBackend does — only the token validation differs (own key/issuer/audience instead of an IdP's JWKS). A token minted by a foreign issuer therefore fails validation and is rejected.

Initialise the backend.

Parameters:

Name	Type	Description	Default
minter	InternalTokenMinter	Validates the internal token (shares the backend's key, issuer, and audience with the mint endpoint).	required
org_resolver	OrgResolver | None	Resolver invoked once per authenticated request to hydrate the user's organizations. When omitted, organizations is always an empty list.	None

authenticate async

authenticate(request)

Validate the request's internal Bearer token and return UserData.

JwtAuthBackend

JwtAuthBackend(org_resolver=None, *, expected_tenant_id=None)

Validates Azure AD JWT tokens from the Authorization header.

Configuration is resolved once at construction time so misconfiguration fails at app startup rather than on the first protected request. JWKS and issuer URLs are derived from AZURE_TENANT_ID when AZURE_JWKS_URL and AZURE_ISSUER_URL are not set explicitly.

A single :class:jwt.PyJWKClient is constructed and reused so the JWKS response is cached across requests (defeating the cache by reconstructing per-request would mean a JWKS HTTP fetch on every authenticated call).

Initialise the backend with an optional OrgResolver.

Parameters:

Name	Type	Description	Default
org_resolver	OrgResolver | None	Resolver invoked once per authenticated request to hydrate the user's organizations. When omitted, organizations is always an empty list.	None
expected_tenant_id	str | None	When set, the token's tid claim must match it (case-insensitive) or the token is rejected. Defaults to None (no tenant cross-check) so existing single-tenant consumers are unaffected.	None

Raises:

Type	Description
RuntimeError	If required Azure environment variables are not set.

authenticate async

authenticate(request)

Validate the request's Bearer token and return UserData.

CasbinConfig

Bases: BaseModel

Validated configuration for :class:~altaforge.auth.casbin.enforcer.ManagedCasbinEnforcer.

At least one of casbin_source_url or af_policy_file must be supplied. If both are provided, casbin_source_url takes priority and af_policy_file is ignored — the database is the source of truth in all deployed environments. af_policy_file exists solely as a local-development fallback when no database is available.

Pydantic raises a ValidationError at construction time if neither is provided, so misconfiguration surfaces immediately on startup rather than on the first authenticated request.

Example — database source (production)::

config = CasbinConfig(
    casbin_source_url=settings.CASBIN_SOURCE_URL,
)

Example — file source (local dev)::

config = CasbinConfig(
    af_policy_file='/app/policy.csv',
)

CasbinOrgResolver

CasbinOrgResolver(registry_loader, enforcer)

Resolves the organizations a user may access by combining registry and Casbin.

Fetches all known organizations from a :class:~altaforge.auth.registry.protocols.RegistryLoader, then filters them to those where Casbin grants the user access. The registry fetch and enforcer refresh are issued concurrently via :func:asyncio.gather to minimise latency.

Implements the :class:~altaforge.auth.protocols.OrgResolver protocol so it can be passed directly to :class:~altaforge.auth.backends.JwtAuthBackend and :class:~altaforge.auth.backends.FakeAuthBackend.

Example::

from altaforge.auth import AuthMiddleware, JwtAuthBackend, OrgResolver
from altaforge.auth.casbin import ManagedCasbinEnforcer, CasbinConfig
from altaforge.auth.registry import ManagedRegistryLoader, RegistryConfig

casbin_config = CasbinConfig(
    casbin_source_url=settings.CASBIN_SOURCE_URL,  # postgresql://...
    casbin_require_ssl=True,
)
enforcer = ManagedCasbinEnforcer.from_config(casbin_config)

registry_config = RegistryConfig(
    registry_source_url=settings.REGISTRY_SOURCE_URL,
    registry_require_ssl=True,
)
registry_loader = ManagedRegistryLoader.from_config(registry_config)
resolver = CasbinOrgResolver(registry_loader=registry_loader, enforcer=enforcer)
app.add_middleware(AuthMiddleware, auth_backend=JwtAuthBackend(org_resolver=resolver))

Create the resolver.

Parameters:

Name	Type	Description	Default
registry_loader	ManagedRegistryLoader	Source of all known organizations.	required
enforcer	ManagedEnforcer	Managed Casbin enforcer used for authorization checks.	required

organizations_for async

organizations_for(user_id)

Return the organization IDs user_id is permitted to access.

Performs a Casbin enforce(user_id, org_id, 'access') check for each organization returned by the registry loader.

CasbinRoleResolver

CasbinRoleResolver(enforcer, *, base_role=DEFAULT_BASE_ROLE, admin_role=DEFAULT_ADMIN_ROLE, admin_group=DEFAULT_ADMIN_GROUP)

Resolves a user's application roles from Casbin group membership.

Google (and other non-Entra providers) carry no app-role claim, so roles are sourced here instead of from an IdP claim (T-decision-5). Every user gets base_role; a user who belongs to the Casbin global-admin group additionally gets admin_role.

Implements :class:~altaforge.auth.protocols.RoleResolver so it can be passed directly to the internal-auth router.

Create the resolver.

Parameters:

Name	Type	Description	Default
enforcer	ManagedEnforcer	Managed Casbin enforcer used to read group membership.	required
base_role	str	Role granted to every authenticated user.	DEFAULT_BASE_ROLE
admin_role	str	Role additionally granted to global admins.	DEFAULT_ADMIN_ROLE
admin_group	str	Casbin grouping subject that denotes a global admin.	DEFAULT_ADMIN_GROUP

roles_for async

roles_for(user_id)

Return the application roles user_id holds, per Casbin membership.

DatabasePolicyLoader

DatabasePolicyLoader(url, require_ssl=False)

Loads Casbin policy rules from a PostgreSQL casbin_rule table.

Owns its async SQLAlchemy engine for the duration of the process. Call :meth:aclose (or use async with) on shutdown to dispose it.

The DSN is passed directly — the caller is responsible for sourcing it (e.g. from their own environment or secrets manager).

require_ssl=True emits psycopg-only SSL connect args — see :mod:altaforge.auth.pg_ssl for the driver/SSL contract.

load async

load()

Query casbin_rule and return all rows as a newline-delimited policy string.

aclose async

aclose()

Dispose the async engine, releasing all pooled connections.

__aenter__ async

__aenter__()

Enter the async context manager.

__aexit__ async

__aexit__(*_)

Exit the async context manager.

FilePolicyLoader

FilePolicyLoader(policy_file)

Loads Casbin policy rules from a CSV policy file on disk.

The file path is passed directly — the caller is responsible for sourcing it. The read is offloaded to a thread so the event loop is not blocked.

load async

load()

Read the policy file and return its contents.

aclose async

aclose()

No-op — file loaders hold no persistent resources.

__aenter__ async

__aenter__()

Enter the async context manager.

__aexit__ async

__aexit__(*_)

Exit the async context manager.

ManagedCasbinEnforcer

ManagedCasbinEnforcer(policy_loader, model_path=DEFAULT_MODEL_PATH)

Casbin enforcer that loads fresh policy on every :meth:get_enforcer call.

Wraps a :class:~altaforge.auth.casbin.protocols.PolicyLoader and builds a new casbin.Enforcer on each call, ensuring policy is always current with no staleness window.

Wildcard role matching (glob_match) is registered automatically so patterns like *@altaml.com work as expected in policy assignments.

The preferred way to construct this in production is via :meth:from_config, which accepts a validated :class:~altaforge.auth.casbin.config.CasbinConfig and selects the correct loader automatically. The direct constructor is available for testing and advanced use cases where the loader is injected.

Usage (production)::

from altaforge.auth.casbin import CasbinConfig, ManagedCasbinEnforcer

config = CasbinConfig(
    casbin_source_url=settings.CASBIN_SOURCE_URL,
)
enforcer = ManagedCasbinEnforcer.from_config(config)

# At app startup:
await enforcer.initialize()

# In a request handler:
e = await enforcer.get_enforcer()
allowed = e.enforce(user_id, resource, 'access')

# At app shutdown:
await enforcer.aclose()

Create the enforcer with an explicit loader.

Parameters:

Name	Type	Description	Default
policy_loader	PolicyLoader	Strategy that supplies raw Casbin policy text.	required
model_path	str	Path to a Casbin model .conf file. Defaults to the bundled AltaForge RBAC model.	DEFAULT_MODEL_PATH

from_config classmethod

from_config(config, model_path=DEFAULT_MODEL_PATH)

Construct from a validated :class:~altaforge.auth.casbin.config.CasbinConfig.

Selects :class:~altaforge.auth.casbin.loaders.DatabasePolicyLoader when config.casbin_source_url is set, otherwise :class:~altaforge.auth.casbin.loaders.FilePolicyLoader.

Parameters:

Name	Type	Description	Default
config	CasbinConfig	Validated Casbin configuration provided by the client app.	required
model_path	str	Override the Casbin model path (optional).	DEFAULT_MODEL_PATH

get_enforcer async

get_enforcer()

Load and return a fresh enforcer from the policy source.

initialize async

initialize()

Eagerly verify the policy source at application startup.

Call this from your ASGI lifespan startup handler so that policy load failures surface immediately rather than on the first authenticated request.

aclose async

aclose()

Close the underlying policy loader.

__aenter__ async

__aenter__()

Enter the async context manager.

__aexit__ async

__aexit__(*_)

Exit the async context manager.

ManagedEnforcer

Bases: Protocol

Manages the lifecycle of a Casbin enforcer with TTL-based policy refresh.

get_enforcer async

get_enforcer()

Return the active enforcer, refreshing policy if the TTL has elapsed.

initialize async

initialize()

Eagerly load the enforcer at application startup (fail-fast on errors).

aclose async

aclose()

Tear down the enforcer and release any underlying loader resources.

PolicyLoader

Bases: Protocol

Loads raw Casbin policy rules from a backing store.

load async

load()

Return all policy rules as a newline-delimited string.

Each line must be a valid Casbin CSV rule, e.g.::

p, alice, data1, read
g, bob, admin

Returns an empty string when no rules are defined.

aclose async

aclose()

Release any resources held by this loader (connections, engines, etc.).

DevKeyPair

DevKeyPair(private_key, kid=DEFAULT_DEV_KID)

An RSA keypair used to sign (client) and verify (server) dev tokens.

Wraps a single RSA private key plus a stable kid (key id). The kid is written into every token header and into the exported JWK so a :class:jwt.PyJWKClient can match the token to its verifying key.

Wrap an existing RSA private key.

Parameters:

Name	Type	Description	Default
private_key	RSAPrivateKey	The RSA private key used for signing.	required
kid	str	Key id stamped into token headers and the exported JWK.	DEFAULT_DEV_KID

kid property

kid

The key id stamped into token headers and the exported JWK.

generate classmethod

generate(kid=DEFAULT_DEV_KID)

Generate a fresh ephemeral RSA keypair.

The key lives only in memory for the life of the process — fine for a single client that also configures the verifying server from :meth:jwks, but two independently-started processes will NOT share it. Use :meth:from_pem with a committed dev key when both sides must agree.

from_pem classmethod

from_pem(private_key_pem, kid=DEFAULT_DEV_KID)

Load a keypair from a PKCS#8/PKCS#1 PEM string (no passphrase).

Parameters:

Name	Type	Description	Default
private_key_pem	str	PEM-encoded RSA private key.	required
kid	str	Key id stamped into token headers and the exported JWK.	DEFAULT_DEV_KID

Raises:

Type	Description
ValueError	If the PEM does not decode to an RSA private key.

private_pem

private_pem()

Return the private key as an unencrypted PKCS#8 PEM string.

Useful to capture a generated ephemeral key so another process (a dev server) can load the same key via :meth:from_pem.

public_jwk

public_jwk()

Return the public key as a single JWK dict (RS256, use=sig).

jwks

jwks()

Return the public key wrapped as a JWKS document ({"keys": [...]}).

Point a JwtAuthBackend at this — serve it at the URL named by AZURE_JWKS_URL for a running dev server, or feed it to the backend's :class:jwt.PyJWKClient directly in a test.

sign

sign(claims)

RS256-sign claims, stamping the keypair's kid into the header.

DevTokenConfig

Bases: BaseModel

Parameters for minting a locally-signed RS256 dev token.

issuer / audience MUST match what the target JwtAuthBackend is configured to accept (AZURE_ISSUER_URL / AZURE_APP_CLIENT_ID), and the backend must resolve this keypair's public JWK (serve :meth:DevKeyPair.jwks at AZURE_JWKS_URL). user_identifier_claim must match the backend's USER_IDENTIFIER_PROPERTY (default unique_name) or the backend rejects the token for a missing identifier.

M2MTokenConfig

Bases: BaseModel

Azure AD client-credentials parameters for the production m2m path.

All four fields are required when mode='m2m'; the provider raises a clear error naming any that are missing before making the network call.

The acquisition is synchronous: a blocking client-credentials request made inside ServiceTokenProvider.token() (which keeps its str return type — no async variant), so consumers never migrate their call sites to await. See :meth:ServiceTokenProvider.token.

ServiceTokenConfig

Bases: BaseModel

Top-level provider config: a mode plus the matching mode's parameters.

from_env classmethod

from_env()

Build a config from SERVICE_TOKEN_* environment variables.

Mode is selected by SERVICE_TOKEN_MODE (default dev). In dev mode the private key comes from SERVICE_TOKEN_DEV_PRIVATE_KEY (inline PEM), else SERVICE_TOKEN_DEV_PRIVATE_KEY_FILE (path to a PEM), else an ephemeral key is generated. Missing optional values fall back to the model defaults.

For m2m mode, SERVICE_TOKEN_M2M_AUTHORITY overrides the Azure authority host (default: the commercial cloud) for sovereign/national clouds; the other SERVICE_TOKEN_M2M_* vars supply tenant/client/ secret/scope.

ServiceTokenProvider

ServiceTokenProvider(config, *, key_pair=None, m2m_transport=None)

Returns a Bearer token string for calling an altaforge-auth service.

One interface, mode-selected by :class:ServiceTokenConfig. Call :meth:token to get a currently-valid token; the provider handles caching and refresh-before-expiry internally.

Build a provider from validated config.

Parameters:

Name	Type	Description	Default
config	ServiceTokenConfig	The provider configuration (mode + mode parameters).	required
key_pair	DevKeyPair | None	Dev keypair override (tests / a shared committed key). When omitted in dev mode, one is built from config.dev — loaded from private_key_pem if set, otherwise generated ephemerally.	None
m2m_transport	BaseTransport | None	httpx transport used for the m2m client-credentials request. Injected in tests (httpx.MockTransport); None in production so httpx uses its default network transport.	None

mode property

mode

The active token mode (dev or m2m).

from_env classmethod

from_env()

Build a provider from SERVICE_TOKEN_* environment variables.

jwks

jwks()

Return the dev JWKS a JwtAuthBackend must verify against (dev mode).

Serve this at the URL named by AZURE_JWKS_URL on the target service, or feed it to a backend's :class:jwt.PyJWKClient in a test.

Raises:

Type	Description
RuntimeError	If the provider is not in dev mode.

token

token()

Return a currently-valid Bearer token, minting/refreshing as needed.

Returns the cached token while it is still comfortably valid (more than refresh_leeway_seconds from expiry); otherwise acquires a fresh one.

Synchronous and blocking in every mode, now and after #352. In dev mode a refresh signs a JWT locally; in m2m mode (once #352 lands) a refresh performs a blocking Azure AD client-credentials call inside this method. Either way the return type stays str — this never becomes a coroutine, so async callers keep the same call site. Because a refresh happens at most once per TTL (not per request), async consumers should offload just that occasional blocking call off the event loop, e.g. await asyncio.to_thread(provider.token).

AdmissionGate

AdmissionGate(org_resolver, require_org_access=False)

Decides whether an authenticated email may be minted a session.

When require_org_access is False (the default) every authenticated email is admitted — preserving today's "logged in, empty data" behaviour, so enabling the gate is a deliberate, non-breaking opt-in. When True, the email must resolve to at least one accessible organization in Casbin.

The email is lowercased before it reaches Casbin so a mixed-case address cannot silently resolve to zero orgs (a silent lock-out — conv:T-q-3).

Create the gate.

Parameters:

Name	Type	Description	Default
org_resolver	OrgResolver	Resolver used to count a user's accessible orgs.	required
require_org_access	bool	Whether to enforce the gate (default False).	False

admit async

admit(email)

Return whether email may be admitted (minted a session).

InternalAuthConfig

Bases: BaseModel

Validated configuration for the internal-token minter and backend.

The backend is the sole minter and validator of the internal token, so a symmetric HS256 secret suffices in v1 (T-q-1); move to RS256 only if a second, independent verifier appears.

Example::

config = InternalAuthConfig(
    signing_key=settings.INTERNAL_AUTH_SIGNING_KEY,
    service_credential=settings.INTERNAL_AUTH_SERVICE_CREDENTIAL,
    require_org_access=True,
)

InternalTokenMinter

InternalTokenMinter(config)

Signs and validates internal identity tokens with the backend-held key.

Email is lowercased before it becomes the sub claim, preserving the load-bearing invariant that Casbin is always queried with a lowercased identifier (conv:T-q-3).

Create the minter from validated internal-auth configuration.

mint

mint(email, roles)

Mint an access + refresh token pair for email carrying roles.

Parameters:

Name	Type	Description	Default
email	str	The BFF-asserted, already-verified email. Lowercased here.	required
roles	list[str]	Application roles to embed (sourced from Casbin, not the IdP).	required

decode_access

decode_access(token)

Validate an access token (own key/iss/aud) and return its identity claims.

Raises:

Type	Description
ExpiredTokenError	If the token has expired.
InvalidTokenError	If the signature, issuer, audience, or token type is wrong — including any token minted by a foreign issuer.

decode_refresh

decode_refresh(token)

Validate a refresh token and return the email (sub) it was minted for.

Raises:

Type	Description
ExpiredTokenError	If the refresh token has expired.
InvalidTokenError	If it is not a valid refresh token.

MintedTokens dataclass

MintedTokens(token, refresh_token, expires_at, roles=list())

The result of a mint: an access token, its refresh token, and metadata.

MintRequest

Bases: ApiModel

Body for POST /internal/auth/mint.

MintResponse

Bases: ApiModel

Response body for both mint and refresh.

RefreshRequest

Bases: ApiModel

Body for POST /internal/auth/refresh.

AuthMiddleware

AuthMiddleware(app, auth_backend, excluded_paths=None)

Bases: BaseHTTPMiddleware

Authenticates each request via a pluggable AuthBackend.

Excluded paths pass through without authentication. All other paths return 401 if authentication fails. On success, request.state.user is set to the UserData returned by the backend.

When an authenticated request carries ?organization_id=<id> as a query parameter, the middleware additionally checks that <id> is in the user's organizations. If it is not, the request is rejected with a generic 404 so callers cannot use the endpoint to enumerate organizations they don't belong to. Routes whose organization id comes from a path parameter or request body must call :func:check_organization_access explicitly — the middleware can only inspect the query string.

Failure detail is logged server-side; the 401 response carries a generic message so the API does not leak token-validation internals to a caller probing for valid token shapes.

Initialise the middleware with the given backend and excluded paths.

Parameters:

Name	Type	Description	Default
app	ASGIApp	The ASGI application to wrap.	required
auth_backend	AuthBackend	The authentication strategy to delegate to.	required
excluded_paths	Iterable[str] | None	Paths that bypass authentication. When None, uses DEFAULT_EXCLUDED_PATHS. Apps that want to extend the defaults should pass DEFAULT_EXCLUDED_PATHS | {...}.	None

dispatch async

dispatch(request, call_next)

Authenticate the request and call the next handler.

MintedTokenAuthBackend

MintedTokenAuthBackend(minters, *, audience)

Validates a minted product token against a small, fixed trusted-minter registry.

Constructed once with the full registry (issuer -> TrustedMinter); the registry never changes per-request. See the module docstring for the trust model. audience is this service's own identity (the aud claim every minted token must carry) — fixed for the whole backend, not per-issuer.

Initialise the backend with a trusted-minter registry.

Parameters:

Name	Type	Description	Default
minters	dict[str, TrustedMinter]	Maps an issuer (iss claim value) to the :class:TrustedMinter that verifies tokens carrying it.	required
audience	str	This service's own identity — every minted token must carry it as aud. Required and has no default: the generic layer has no single service identity, so each service names itself here (e.g. a DocGen deployment passes "docgen").	required

Raises:

Type	Description
ValueError	If audience is blank.

authenticate async

authenticate(request)

Validate the request's minted Bearer token and return UserData.

Raises:

Type	Description
MissingTokenError	No Authorization: Bearer header.
InvalidTokenError	Unknown issuer, signature/audience/issuer mismatch, or an org/sub/roles claim that fails shape or entitlement validation.
ExpiredTokenError	A structurally valid but expired token.

TrustedMinter dataclass

TrustedMinter(issuer, allowed_orgs, public_key_pem=None, jwks_url=None)

One trusted product's token-verification config.

Exactly one of public_key_pem/jwks_url must be set: public_key_pem verifies offline against a fixed key (the common case — a small, static set of products); jwks_url supports a product that rotates its signing key behind a hosted JWKS endpoint instead.

Raises:

Type	Description
ValueError	If zero or both of public_key_pem/jwks_url are set.

__post_init__

__post_init__()

Validate that exactly one key source is set.

AuthBackend

Bases: Protocol

Pluggable authentication strategy used by AuthMiddleware.

authenticate async

authenticate(request)

Authenticate the request and return user data.

Raises:

Type	Description
MissingTokenError	If no token is present.
ExpiredTokenError	If the token has expired.
InvalidTokenError	If the token is invalid.

AuthError

Bases: Exception

Base class for authentication failures raised by AuthBackend implementations.

ExpiredTokenError

Bases: AuthError

Raised when the token has expired.

InvalidTokenError

Bases: AuthError

Raised when the token is structurally or semantically invalid.

MissingTokenError

Bases: AuthError

Raised when no Bearer token is present in the request.

OrgResolver

Bases: Protocol

Resolves the list of organization IDs a user is permitted to access.

organizations_for async

organizations_for(user_id)

Return the organization IDs user_id is permitted to access.

ResourceAuthorizer

Bases: Protocol

Resolves a user's role on a single resource — the extension seam for resource-scoped RBAC (see :func:~altaforge.auth.resource_auth.require_permission).

Unlike :class:RoleResolver (global app roles), this answers the per-resource question "what role, if any, does this user hold on this resource". Each app implements it over its own membership store (e.g. a user_sites join table); the library never stores membership itself.

role_for async

role_for(user_id, resource_id)

Return user_id's role on resource_id, or None if they hold none.

RoleResolver

Bases: Protocol

Resolves the application roles a user holds.

The internal-token minter uses this to source roles from the authorization store (Casbin) rather than from an upstream IdP claim — see :class:~altaforge.auth.casbin.role_resolver.CasbinRoleResolver.

roles_for async

roles_for(user_id)

Return the application role strings user_id holds.

UserData

Bases: TypedDict

Authenticated user identity returned by any AuthBackend.

DatabaseRegistryLoader

DatabaseRegistryLoader(url, require_ssl=False)

Loads organizations from a PostgreSQL organizations table.

Owns its async SQLAlchemy engine. Call :meth:aclose (or use async with) on application shutdown to dispose the engine.

The DSN and SSL flag are passed directly — the caller (client app) is responsible for sourcing them.

require_ssl=True emits psycopg-only SSL connect args — see :mod:altaforge.auth.pg_ssl for the driver/SSL contract.

load_organizations async

load_organizations(app=None)

Query the organizations table, optionally filtering by app.

aclose async

aclose()

Dispose the async engine, releasing all pooled connections.

__aenter__ async

__aenter__()

Enter the async context manager.

__aexit__ async

__aexit__(*_)

Exit the async context manager.

FileRegistryLoader

FileRegistryLoader(registry_file)

Loads organizations from a JSON or YAML file on disk.

The file is re-read on each :meth:load_organizations call. Wrap with :class:~altaforge.auth.registry.managed.ManagedRegistryLoader to add TTL-based caching so the file is not re-read on every request.

The file path is passed directly — the caller is responsible for sourcing it. The read is offloaded to a thread so the event loop is not blocked.

load_organizations async

load_organizations(app=None)

Read and parse the registry file, optionally filtering by app.

aclose async

aclose()

No-op — file loaders hold no persistent resources.

__aenter__ async

__aenter__()

Enter the async context manager.

__aexit__ async

__aexit__(*_)

Exit the async context manager.

ManagedRegistryLoader

ManagedRegistryLoader(registry_loader, app=None)

Wrapper around any :class:~altaforge.auth.registry.protocols.RegistryLoader that fetches fresh organizations on every :meth:load_organizations call.

Accepts a concrete loader (Database, File, or Text) that satisfies the :class:~altaforge.auth.registry.protocols.RegistryLoader protocol, and exposes a no-arg :meth:load_organizations interface for use by :class:~altaforge.auth.casbin.org_resolver.CasbinOrgResolver.

The preferred way to construct this in production is via :meth:from_config. The direct constructor is available for testing and DI.

Usage::

from altaforge.auth.registry import ManagedRegistryLoader, RegistryConfig

config = RegistryConfig(
    registry_source_url=settings.REGISTRY_SOURCE_URL,  # postgresql://...
    registry_require_ssl=True,
)
registry = ManagedRegistryLoader.from_config(config)

# At app startup:
await registry.initialize()

# Pass to CasbinOrgResolver (or any other consumer):
resolver = CasbinOrgResolver(registry_loader=registry, enforcer=enforcer)

# At app shutdown:
await registry.aclose()

Create the managed loader.

Parameters:

Name	Type	Description	Default
registry_loader	RegistryLoader	Underlying loader that fetches organizations.	required
app	str | None	The application to which the organization belongs.	None

from_config classmethod

from_config(config)

Construct from a validated :class:~altaforge.auth.registry.config.RegistryConfig.

Selects the correct loader based on which source field is set.

Parameters:

Name	Type	Description	Default
config	RegistryConfig	Validated registry configuration provided by the client app.	required

Raises:

Type	Description
ValidationError	Propagated from RegistryConfig if no source is set.

load_organizations async

load_organizations()

Fetch and return organizations fresh from the underlying loader.

initialize async

initialize()

Eagerly verify the registry source at application startup.

Call this from your ASGI lifespan startup handler so that registry load failures surface immediately rather than on the first authenticated request.

aclose async

aclose()

Close the underlying registry loader.

__aenter__ async

__aenter__()

Enter the async context manager.

__aexit__ async

__aexit__(*_)

Exit the async context manager.

MinimalOrganizationManifest

Bases: BaseModel

Lightweight organization record used for authentication and authorization.

Intentionally minimal — it carries only the fields needed to make auth decisions. Consuming services that need richer metadata (solutions, dashboards, agents, etc.) should maintain their own full registry models.

RegistryConfig

Bases: BaseModel

Validated configuration for :class:~altaforge.auth.registry.managed.ManagedRegistryLoader.

At least one of registry_source_url, af_registry_file, or af_registry_text must be supplied. If multiple are provided, the first match in the resolution order below wins and the rest are ignored — the database is the source of truth in all deployed environments. af_registry_file and af_registry_text exist as local-development and test fallbacks only. Pydantic raises a ValidationError at construction time if none is provided.

Resolution order (first match wins) in :func:~altaforge.auth.registry.loaders.create_registry_loader:

1. registry_source_url → :class:~altaforge.auth.registry.loaders.DatabaseRegistryLoader (production)
2. af_registry_file → :class:~altaforge.auth.registry.loaders.FileRegistryLoader (local dev)
3. af_registry_text → :class:~altaforge.auth.registry.loaders.TextRegistryLoader (tests)

Example — database source::

config = RegistryConfig(
    registry_source_url=settings.REGISTRY_SOURCE_URL,
    registry_require_ssl=True,
)

Example — file source (local dev)::

config = RegistryConfig(af_registry_file='/app/registry.json')

Example — inline text (tests)::

config = RegistryConfig(
    af_registry_text='{"organizations": [{"id": "org-1", "display_name": "Org One"}]}'
)

RegistryLoader

Bases: Protocol

Loads organization records from a backing store.

Concrete implementations are selected by :func:~altaforge.auth.registry.loaders.create_registry_loader based on which source field is set in :class:~altaforge.auth.registry.config.RegistryConfig.

:class:~altaforge.auth.registry.managed.ManagedRegistryLoader wraps a concrete implementation and applies app from its construction-time config so callers do not pass it per-call.

load_organizations async

load_organizations(app)

Return organizations, optionally filtered by app.

aclose async

aclose()

Release any resources held by this loader (connections, engines, etc.).

TextRegistryLoader

TextRegistryLoader(text)

Loads organizations from an inline JSON or YAML string.

The text is parsed once on the first call and the result is cached for the lifetime of the loader, since the text is immutable after construction.

Useful for tests and embedded configuration.

load_organizations async

load_organizations(app=None)

Parse the registry text (cached after the first call), optionally filtering by app.

aclose async

aclose()

No-op — text loaders hold no persistent resources.

__aenter__ async

__aenter__()

Enter the async context manager.

__aexit__ async

__aexit__(*_)

Exit the async context manager.

create_policy_loader

create_policy_loader(config)

Instantiate the appropriate PolicyLoader from a validated :class:~altaforge.auth.casbin.config.CasbinConfig.

Resolution order matches the config validator:

1. casbin_source_url is set → :class:DatabasePolicyLoader
2. af_policy_file is set → :class:FilePolicyLoader

Because :class:~altaforge.auth.casbin.config.CasbinConfig already enforces that at least one source is present, no further validation is needed here.

mint_trusted_subsystem_token

mint_trusted_subsystem_token(user_id, org, roles, *, issuer, private_key_pem, kid, audience, ttl_seconds=DEFAULT_MINT_TTL_SECONDS)

Mint a short-lived RS256 JWT for calling a trusted-subsystem service.

Parameters:

Name	Type	Description	Default
user_id	str	The already-authenticated end user's id (becomes sub).	required
org	str	The org/tenant slug this call is scoped to (becomes org). The downstream service enforces this against the issuer's OWN allowlist (MintedTokenAuthBackend/TrustedMinter.allowed_orgs) — asserting an org this product isn't entitled to is rejected downstream; it is not prevented here.	required
roles	list[str]	Role strings to carry (may be empty).	required
issuer	str	This product's slug (becomes iss), e.g. "site-rfp". Must match the issuer key the downstream service's trusted-minter registry has configured for this product's public key.	required
private_key_pem	str	This product's RSA private signing key (PEM, PKCS#1 or PKCS#8, no passphrase).	required
kid	str	This product's key id, stamped into the token header so the validator (when configured via a JWKS url rather than a fixed key) can select the matching public key.	required
audience	str	The downstream service's identity (the aud claim every minted token must carry). Required — the caller names the service it is calling (e.g. "docgen"); the generic layer presumes no single target, and this must match that service's configured MintedTokenAuthBackend(audience=...).	required
ttl_seconds	int	Token lifetime in seconds. Keep short — see module docstring.	DEFAULT_MINT_TTL_SECONDS

Raises:

Type	Description
ValueError	If audience is blank, or private_key_pem does not decode to an RSA private key.

scoped_org_context

scoped_org_context(organization_id)

Bind organization_id_var for the duration of the block, then reset it.

Uses the token returned by :meth:ContextVar.set so the previous value is restored exactly on exit — safe under nesting and concurrent tasks.

create_internal_auth_router

create_internal_auth_router(*, minter, admission_gate, role_resolver, config)

Build the internal-auth router from injected collaborators.

Parameters:

Name	Type	Description	Default
minter	InternalTokenMinter	Signs/validates the internal token.	required
admission_gate	AdmissionGate	The opt-in Casbin org-access login gate.	required
role_resolver	RoleResolver	Sources the user's roles from Casbin.	required
config	InternalAuthConfig	Supplies the service credential the endpoints require.	required

make_service_credential_dependency

make_service_credential_dependency(config)

Build a FastAPI dependency that rejects callers without the service credential.

Returns a dependency that raises 401 unauthorized when the X-AltaForge-Service-Credential header is missing or does not match, using a constant-time comparison so the endpoint does not leak the secret via response timing.

check_organization_access

check_organization_access(request, organization_id)

Raise HTTP 404 if the authenticated user may not access organization_id.

Routes that receive organization_id as a ?organization_id=<id> query parameter do not need to call this — :class:AuthMiddleware performs the same check on the query string. Call this helper for routes whose organization id comes from a path parameter or request body.

The denial is reported as 404 rather than 403 so callers cannot use this endpoint to enumerate organizations they don't belong to — "exists but forbidden" and "does not exist" are indistinguishable from the client.

Parameters:

Name	Type	Description	Default
request	Request	The current FastAPI request (must have request.state.user set).	required
organization_id	str	The organization ID being requested.	required

Raises:

Type	Description
HTTPException	404 if the user is not permitted.

custom_openapi

custom_openapi(app, original_openapi, excluded_paths=None)

Tag non-excluded routes with BearerAuth in the OpenAPI schema.

Caches the modified schema on app.openapi_schema so subsequent calls return it directly.

Parameters:

Name	Type	Description	Default
app	FastAPI	The FastAPI app whose schema is being customized.	required
original_openapi	Callable[[], dict[str, Any]]	The original app.openapi callable.	required
excluded_paths	Iterable[str] | None	Paths to leave unauthenticated. Defaults to DEFAULT_EXCLUDED_PATHS.	None

Returns:

Type	Description
dict[str, Any]	The modified OpenAPI schema.

create_registry_loader

create_registry_loader(config)

Instantiate the appropriate loader from a validated :class:~altaforge.auth.registry.config.RegistryConfig.

Resolution order:

1. registry_source_url → :class:DatabaseRegistryLoader
2. af_registry_file → :class:FileRegistryLoader
3. af_registry_text → :class:TextRegistryLoader

Because :class:~altaforge.auth.registry.config.RegistryConfig already enforces that at least one source is present, no further validation is needed here.

require_permission

require_permission(authorizer, enforcer, *, domain, action, resource_param='resource_id', obj='*')

Dependency factory: allow only if the caller's role on the route's resource permits action in domain (per the app's Casbin policy). Returns the caller's role on success.

Parameters:

Name	Type	Description	Default
authorizer	AuthorizerDependency	dependency yielding the app's ResourceAuthorizer.	required
enforcer	ManagedCasbinEnforcer	the app's enforcer loaded with domain_model.conf.	required
domain	str	the permission group being acted on (e.g. "voice").	required
action	str	the action (e.g. "edit").	required
resource_param	str	the path param naming the resource (default resource_id).	'resource_id'
obj	str	the object within the domain; "*" unless the policy is per-object.	'*'

require_resource_role

require_resource_role(authorizer, *roles, resource_param='resource_id')

Dependency factory: allow only if the caller's role on the route's resource is one of roles. A lightweight, Casbin-free gate for routes that need a role rather than a specific action (e.g. "any editorial_lead"). Returns the role.