Skip to content

Authn, authz & tenancy

Three distinct concerns, three distinct seams:

  • Authentication — who is calling — happens at the HTTP boundary, in middleware.
  • Tenancy — whose data — is resolved at the boundary and bound into the context.
  • Authorization — may they do this — is enforced per-operation, inside the application.

Forze keeps them separate, and the conceptual model is in Identity & access — this recipe is the wiring. One shape to keep in mind: there is no Depends(...) for the current user. SecurityContextMiddleware binds the principal and tenant into the execution context (ctx.inv_ctx); your operation's hooks read that binding and enforce, raising exc.authentication / exc.authorization that the exception handlers map to 401 / 403.

Install the middleware

SecurityContextMiddleware authenticates; InvocationMetadataMiddleware binds request ids and the idempotency key. Both take a ctx_dep that returns the per-scope context:

from forze.application.contracts.authn import AuthnSpec
from forze_fastapi.exceptions import register_exception_handlers
from forze_fastapi.middlewares import InvocationMetadataMiddleware, SecurityContextMiddleware
from forze_fastapi.security import AuthnRequirement, HeaderTokenAuthn

API = AuthnSpec(name="api", enabled_methods=frozenset({"token"}))

app = FastAPI()
register_exception_handlers(app)  # exc.authentication → 401, exc.authorization → 403

app.add_middleware(
    SecurityContextMiddleware,
    ctx_dep=lambda: runtime.get_context(),
    authn=AuthnRequirement(
        ingress=(HeaderTokenAuthn(authn_spec=API, header_name="Authorization"),),
    ),
    when_multiple_credentials="first_in_order",
    anonymous_paths={  # the generated routes reachable without a credential
        "/auth/login",
        "/auth/refresh",
        "/auth/password-reset/request",
        "/auth/password-reset/confirm",
    },
)
app.add_middleware(InvocationMetadataMiddleware, ctx_dep=lambda: runtime.get_context())

A request no ingress authenticates gets a 401 before routing, except on anonymous_paths; list there every route that exists without an identity.

Ingress options are HeaderTokenAuthn, HeaderApiKeyAuthn (header_name), and CookieTokenAuthn (cookie_name) — each carries the AuthnSpec whose name selects the route's verifier and resolver.

Wire the planes

Three sibling deps modules. AuthnDepsModule registers the verify-then-resolve stack; AuthzDepsModule the decision + scope ports; TenancyDepsModule the tenant resolver. The "api" route name is the same string across all three:

from forze_identity.authn import AuthnDepsModule, AuthnKernelConfig
from forze_identity.authz import AuthzDepsModule, AuthzKernelConfig
from forze_identity.tenancy import TenancyDepsModule

deps = DepsRegistry.from_modules(
    AuthnDepsModule(
        kernel=AuthnKernelConfig(access_token_secret=secret),  # bytes, ≥ 32
        authn={"api": frozenset({"token"})},
    ),
    AuthzDepsModule(kernel=AuthzKernelConfig(), decision={"api"}, scope={"api"}),
    TenancyDepsModule(tenant_resolver={"api"}),
)

The authn stack is document-backed — the account/session stores are document specs you wire to a database (see the authn integration).

Enforce on operations

Attach hooks to the operations that need them, using the same registry-binding chain as idempotency. AuthnRequired and TenantRequired are before-hooks (.before); the authz scope filter is a wrap (.wrap):

from forze.application.contracts.authz import AuthzSpec
from forze.application.hooks.authn import AuthnRequired
from forze.application.hooks.authz import AuthzBeforeAuthorize, AuthzDocumentScopeWrap
from forze.application.hooks.tenancy import TenantRequired

AUTHZ = AuthzSpec(name="api")
CREATE = ORDER_SPEC.default_namespace.key(DocumentKernelOp.CREATE)
LIST = ORDER_SPEC.default_namespace.key(DocumentKernelOp.FIND_MANY)

registry = (
    build_document_registry(ORDER_SPEC, DocumentDTOs(read=ReadOrder, create=CreateOrder))
    .bind(CREATE)
        .bind_outer()
        .before(
            AuthnRequired().to_step(),       # provides "authn.principal", which authz requires
            TenantRequired().to_step(step_id="tenant.required"),
            AuthzBeforeAuthorize(spec=AUTHZ, action="orders:create").to_step(step_id="authz.create"),
        )
        .finish(deep=True)
    .bind(LIST)
        .bind_outer()
        .before(AuthnRequired().to_step(), TenantRequired().to_step(step_id="tenant.required"))
        .wrap(AuthzDocumentScopeWrap(spec=AUTHZ, document_name="orders", operation="find_many").to_step(step_id="authz.scope"))
        .finish(deep=True)
    .freeze()
)
  • AuthnRequired demands a principal; TenantRequired demands a bound tenant — both raise 401 when missing.
  • AuthzBeforeAuthorize asks the decision port whether the subject may perform action; a deny raises 403.
  • AuthzDocumentScopeWrap asks the scope port for a row filter and merges it into the query ($and) before the list runs — so a caller only ever sees rows they're entitled to. Authorization scoping is query-DSL filter injection.

With explain_empty=True, the wrap also says why a page came back empty: when the policy restricted nothing, an empty page is no_match outright; when it did, one probe re-runs the read with the caller's own filters, clamped to a single first-page row — rows there mean not_permitted, none mean no_match. The probe's rows never reach the caller, only whether any exist. Off by default; note that not_permitted discloses that hidden rows exist, so surfacing it to end users is your API's disclosure decision.

A hub or federated search reads several members under one operation, so an action on the operation would let a caller read a member it may not. Attach AuthzSearchMembers, which checks each member the request reaches against that member's own read action:

from forze.application.hooks.authz import AuthzSearchMembers

SEARCH = EVERYTHING.default_namespace.key(SearchKernelOp.TYPED)

registry = (
    build_federated_search_registry(EVERYTHING)  # members: orders, invoices
    .bind(SEARCH)
        .bind_outer()
        .before(AuthnRequired().to_step())
        .wrap(
            AuthzSearchMembers(
                spec=AUTHZ,
                search=EVERYTHING,
                actions={"orders": "orders:read", "invoices": "invoices:read"},
                scope_operation="search",
            ).to_step()
        )
        .finish(deep=True)
    .freeze()
)
  • A request that names its members (members, or member_weights above zero) is refused with 403 when it names one the caller may not read.
  • A request that names none searches the members the search runs by default (on a hub, those with a default weight above zero). It is narrowed to the members the caller may read, and refused only when none remains.
  • scope_operation (federated only) also asks the scope port for each member's row filter and applies it to that member's leg alone. The request names the member as its document_name, unless scope_documents maps the member to another document.

The hook decides before the handler runs and binds the result as a SearchMemberGrant. The adapter applies that grant to the request it actually receives, so a request mapper in SearchMappers cannot widen it: a member the mapper adds is refused or skipped by the same two rules, and the policy filters stay on their legs.

actions must name every member. The registry refuses to freeze a hub or federated search operation that has an authz hook but no AuthzSearchMembers, or that also carries an IdempotencyWrap, whose replayed result would skip the check. To let one action read every member, map each member to it. Two AuthzSearchMembers on one operation both apply: a member must pass each.

HTTP login endpoints

The login flows themselves come for free: build_authn_registry registers the password-login, refresh, logout, change-password and password-reset operations, and attach_authn_routes projects them onto a router — POST /auth/login, /auth/refresh, /auth/logout, /auth/change-password, /auth/password-reset/request, /auth/password-reset/confirm:

from fastapi import APIRouter
from forze_fastapi.routes import attach_authn_routes
from forze_kits.aggregates.authn import build_authn_registry

AUTH = AuthnSpec(name="api", enabled_methods=frozenset({"password", "token"}))

auth_router = APIRouter(prefix="/auth", tags=["auth"])
attach_authn_routes(
    auth_router,
    registry=build_authn_registry(AUTH).freeze(),
    ns=AUTH.default_namespace,
    ctx_dep=lambda: runtime.get_context(),
)
app.include_router(auth_router)

curl -X POST /auth/login -d '{"login": "alice", "password": "…"}'
# → {"access_token": "…", "refresh_token": "…", "access_token_type": "Bearer", …}
curl -X POST /auth/logout -H "Authorization: Bearer <access_token>"   # → 204

Each of these operations runs its writes in a transaction on "default", so the deps need a transaction manager on that route, or pass tx_route= to name another. The authn integration says what the transaction guarantees.

/login and /refresh are deliberately reachable without a bearer token — the operations authenticate via their request bodies. /logout and /change-password answer 401 on their own when the middleware bound no identity. The operations that act on another principal exist only behind guards you pass as build_authn_registry(AUTH, admin_guards=(...)) — AuthnRequired plus an AuthzBeforeAuthorize, the same chain as above: /auth/deactivate (deactivate_principal), GET /auth/admin/principals/{id}/api-keys, which lists any principal's API keys, and DELETE /auth/admin/api-keys/{id}, which revokes one. Without admin_guards none is registered. All are global: credential accounts are not tenant-scoped, so with tenant-scoped administrators expose none, or guard them with a policy only platform administrators pass. At least one guard must declare its permission keys (permission_keys()), as AuthzBeforeAuthorize does, or the registry is refused when built; trust_admin_guards=True skips the check for a guard that authorizes without declaring it.

Self-service password reset

Request → deliver → confirm is a separate route set on the same registry: add the reset pepper and password_reset={"api"}, then deliver the token out of band (it is never returned in the response). The full flow — wiring, the outbox delivery seam, and the token-handling caveats — is the password-reset recipe.

Multiple organizations

When a principal belongs to several tenants, the self-service org-switcher (list, activate, leave) and the privileged tenant-admin plane (create orgs, manage members) are projector-driven route sets that build on this wiring — see Tenant selector & administration.

Notes

  • Tenant binding. The tenant comes from the verified credential's issuer hint or, with trust_tenant_header=True, an X-Tenant-Id header. A bound tenant flows into every authz check via the operation's scope.
  • Ordering is enforced. AuthzBeforeAuthorize declares requires=("authn.principal",), so the principal is always resolved before the authorization decision.
  • This is the foundation the other identity recipes build on — see Local identity, External bootstrap → Forze JWT, and Social sign-in.