Caller identity and authorization

Loom answers “who is running this?” with one type, Identity, produced by whatever mechanism authenticated the request and consumed by everything downstream. Business rules never see a token, a header or an ASGI scope, so swapping JWT for mutual TLS changes the composition root and nothing else.

The identity

from loom.core.identity import Identity, current_identity

identity = current_identity()

identity.is_authenticated          # False for the anonymous caller
identity.subject                   # stable caller id, "" when anonymous
identity.has_role("role_admin")    # exact match, never a prefix
identity.attribute("email")        # verified string attribute, or None
identity.mechanism                 # "jwt", "api-key", ... — audit only

Two accessors fail closed instead of returning None:

Call

Anonymous caller

Authenticated caller missing the value

require_subject()

raises Unauthenticated (401)

—

require_attribute("email")

raises Unauthenticated (401)

raises Forbidden (403)

The distinction is deliberate: a 401 tells the caller that credentials would help, a 403 tells them they would not.

Warning

repr(identity) prints attribute names but never their values — they are personal data and must not reach a log through a stray f-string.

current_identity() never returns None: with no authenticated caller it yields ANONYMOUS, so no consumer can mistake “unknown” for “authorized”.

Reading the caller from a use case

Declare it in the signature, like every other Loom binding:

from loom.core.identity import Identity
from loom.core.use_case import Caller, UseCase


class SendReceiptUseCase(UseCase[Receipt, None]):
    def __init__(self, mailer: Mailer) -> None:
        self._mailer = mailer

    async def execute(self, receipt_id: int, caller: Identity = Caller()) -> None:
        await self._mailer.send(to=caller.require_attribute("email"), receipt=receipt_id)

Caller() is a declaration, not an ambient read. The transport hands the identity to the executor for that one execution, so the same use case behaves identically behind HTTP and behind a Celery worker, where no context variable would have survived.

Important

Binding is fail-closed. If a transport executes a use case declaring Caller() without handing an identity over, the execution raises Unauthenticated naming the use case and the parameter. To run without a caller, a transport must pass ANONYMOUS explicitly — the framework never substitutes it.

A policy that narrows a query by owner

The most common use of the caller is not a role check but a data boundary:

from loom.core.repository.abc.query import FilterGroup, FilterOp, FilterSpec, QuerySpec


def owned_by(query: QuerySpec, subject: str) -> QuerySpec:
    """Restrict *query* to the rows the caller owns, whatever else it asked for."""
    owner = FilterSpec(field="owner_id", op=FilterOp.EQ, value=subject)
    existing = query.filters.filters if query.filters else ()
    return replace(query, filters=FilterGroup(filters=(*existing, owner)))


class ListMyOrdersUseCase(UseCase[Order, PageResult[Order]]):
    read_only = True

    async def execute(
        self,
        query: QuerySpec,
        caller: Identity = Caller(),
    ) -> PageResult[Order]:
        return await self.main_repo.find(owned_by(query, caller.require_subject()))

The caller controls filters, sorting and pagination; the policy is applied after them, so a crafted query cannot widen the result set.

Route-level roles

For coarse-grained access, declare the roles on the route instead of writing the check in every use case:

class ReportsInterface(RestInterface[Report]):
    prefix = "/reports"
    requires_roles = ("report_reader",)          # default for every route
    routes = (
        RestRoute(use_case=ListReportsUseCase, method="GET", path="/"),
        RestRoute(
            use_case=PurgeReportsUseCase,
            method="DELETE",
            path="/",
            requires_roles=("report_admin",),    # route wins over the interface
        ),
    )

  • Holding any declared role is enough; the tuple is a set of alternatives.

  • The check runs before the use case is constructed, so a denied caller never causes a repository or a session to be resolved.

  • A caller without the role — or without an identity at all — gets a 403 with the standard error body. The message does not name the required roles: the response must not become an oracle for the route’s policy.

  • Declaring nothing leaves the route open, so this is opt-in and additive.

Use requires_roles for “who may reach this endpoint” and Caller() for “what this caller may see”. They compose: a route can require a role and the use case can still narrow the data to the caller’s own rows.

Configuring authentication

The built-in JWT mechanism

app:
  rest:
    auth:
      jwt:
        secret_path: ${oc.env:LOOM_JWT_SECRET_PATH}
        algorithms: [HS256]
        audience: loom-api
        roles_claim: loom_roles

Claims are projected onto the identity: sub becomes the subject, roles_claim becomes the roles, and every other string claim becomes an attribute. Registered claims (iss, aud, exp, nbf, iat, jti) describe the token rather than the caller and never cross. A malformed roles claim — a number, an object, a list holding anything but non-empty strings — grants no role at all rather than a filtered subset.

See SQL API (ClickHouse) for the full JWT reference and the SQL endpoint startup gates.

Signing keys from a managed store

JwtIssuerConfig accepts the signing key as a filesystem path or as a managed-store reference — exactly one of the two:

config = JwtIssuerConfig(
    private_key_ref="secrets:/myapp/prod/jwt-signing-key",  # or "ssm:/..."
    algorithm="EdDSA",
    audience="my-api",
    issuer="my-gateway",
    roles_claim="loom_sql_roles",
    kid="2026-08",
)

The reference is resolved once, when the issuer loads the key, so the material never touches disk or config dumps. Rotating the stored value therefore requires a restart. key_ref_region pins the AWS region when it is not in boto3’s own chain. Requires the loom-kernel[config-ssm] extra.

Deriving the verifier from the signing key

The service that both issues and verifies should not configure the public key by hand: two values that must match are two values that can disagree, and a stale public key does not fail loudly. Derive it instead:

auth = JwtAuthConfig.from_signing_key(
    "/run/secrets/jwt-signing-key.pem",   # or private_key_ref="secrets:/..."
    kid="2026-09",
    algorithms=("EdDSA",),
    additional_public_keys={"2026-08": previous_public_pem},
)

additional_public_keys keeps the previous kid published for one rotation window — public material, so unlike the signing key it is safe to carry as a value.

A mechanism of your own

Implement Authenticator and hand it to create_app:

from loom.core.identity import Identity
from loom.rest.auth import Authenticator, RequestCredentials


class ApiKeyAuthenticator:
    name = "api-key"
    provides_roles = True

    def __init__(self, keys: KeyStore) -> None:
        self._keys = keys

    async def authenticate(self, credentials: RequestCredentials) -> Identity | None:
        key = credentials.header("x-api-key")
        owner = await self._keys.owner_of(key) if key else None
        if owner is None:
            return None                      # a refusal, with no reason attached
        return Identity(
            subject=owner.id,
            roles=owner.roles,
            attributes={"email": owner.email},
            mechanism=self.name,
        )


app = create_app("config/app.yaml", authenticator=ApiKeyAuthenticator(store))

  • RequestCredentials exposes headers (case-insensitive), path and peer address — and deliberately not the body, which would have to be buffered before deciding whether the caller exists at all.

  • Returning None is a refusal. It carries no reason on purpose: the 401 must not say which part of the credentials failed.

  • provides_roles is read by startup gates — a role-based SQL endpoint refuses to mount behind a mechanism that issues no role.

  • The authenticator argument is mutually exclusive with app.rest.auth.jwt: two mechanisms would mean two sources of truth for the caller.

Everything else — requires_roles, Caller(), SQL role resolution — works unchanged, because none of it knows what a token is.

Paths served without authentication

Left unset, exclude_paths follows the paths the application actually publishes — docs_url, redoc_url, openapi_url, the metrics endpoint when enabled and /health — instead of a hardcoded list that goes stale the moment an operator moves Swagger. An explicit list replaces that detection entirely, so it must name /health itself for the readiness probe to stay anonymous.

Warning

Exclusions are matched by the router, not by string comparison. A route declared as /{tenant} answers /openapi.json too, so excluding the schema would serve a business route with no credentials at all. create_app walks the registered routes at startup and refuses to boot when an exclusion is captured that way.

An authenticated application that still publishes its OpenAPI document anonymously gets a startup warning: the schema lists every route, parameter and field. Set app.rest.openapi_url: null (with docs_url and redoc_url) in production.

Request limits and CORS

app:
  rest:
    max_body_bytes: 1048576        # 1 MiB, applied to every route
    cors:
      allow_origins: ["https://app.example.com"]
      allow_credentials: true
      allow_methods: [GET, POST]

  • Body size. Neither uvicorn nor Starlette caps a request body, so a chunked upload with no end takes the worker down. The cap is enforced by a middleware — it covers routes the application mounted by hand too — and endpoints with a stricter budget, such as the SQL one, apply theirs on top.

  • Pagination. ?limit= is clamped to RestApiDefaults.max_limit (1000 by default) and ?page= must be a positive integer; anything else answers 400 instead of reaching the database as a full scan.

  • CORS is only mounted when the section exists. allow_origins: ["*"] together with allow_credentials: true fails at config parse: Starlette does not reject that pair, it starts reflecting the caller’s Origin and allowing credentials, which turns the wildcard into “any site, with cookies”. Preflight OPTIONS requests are answered before authentication — they carry no credentials by definition — while the request that follows is authenticated normally.

  • Trace ids supplied by the caller are accepted only when they match [A-Za-z0-9._-]{1,128}; anything else is replaced by a generated one, because the value is echoed back and reaches every log line.

Identity in jobs

A context variable does not cross a broker, so the identity travels inside the job envelope:

class ExportUseCase(UseCase[Export, None]):
    async def execute(self, jobs: JobService = ..., caller: Identity = Caller()) -> None:
        jobs.dispatch(BuildExportJob, payload={"format": "csv"})

dispatch captures the caller at registration time, the worker republishes it for the whole task and hands it to the executor, and a job dispatching another job propagates the same caller onward.

An envelope minted before this contract carries no identity: the job runs without a caller, and a job declaring Caller() fails closed with a message naming what is missing — rather than running as an unknown caller.

Testing

from loom.core.identity import ANONYMOUS, Identity
from loom.testing.runner import UseCaseTest

caller = Identity(subject="user-1", roles=("report_reader",), mechanism="test")

result = await UseCaseTest(ListMyOrdersUseCase(repo)).with_caller(caller).run()

# Pin the unauthenticated path explicitly:
with pytest.raises(Unauthenticated):
    await UseCaseTest(ListMyOrdersUseCase(repo)).with_caller(ANONYMOUS).run()

GoldenHarness.run(..., identity=...) takes the same argument. Omitting it on a use case that declares Caller() raises Unauthenticated: an authorization test that forgot to state whose request it is would otherwise be vacuous.

Migrating from jwt_claims

scope["state"]["jwt_claims"] has been removed. It was a second, transport-shaped source of truth for a security decision, sitting next to the identity context — and two sources of truth for who the caller is were the defect this design closes.

Before

Now

request.scope["state"]["jwt_claims"]["sub"]

current_identity().subject

jwt_claims["email"]

current_identity().attribute("email")

jwt_claims.get("loom_roles", [])

current_identity().roles

reading claims inside execute()

declaring caller: Identity = Caller()

Other renames in the same change:

  • sql_endpoint.auth: jwt → auth: identity (jwt still works, with a DeprecationWarning).

  • The 401 error code is now unauthenticated, matching ErrorCode, and the response carries a WWW-Authenticate: Bearer challenge.