Authorization

A policy per route (FastAPIAuthz) or for every request (AuthzMiddleware), held by Casbin, Cedar (CedarEnforcer) or Open Policy Agent (OPAEnforcer). In fastapi_alcazaba.authz.

In the guide: Authorization. In an app: the examples.

fastapi_alcazaba.authz.FastAPIAuthz

FastAPIAuthz(enforcer: Enforcer, *, subject: Subject | None = None, anonymous: str | None = None, challenge: str = 'Bearer', metrics: Metrics | None = None, audit: FastAPIAudit | None = None, actor: Actor | None = None)

FastAPI dependency that lets a request through only when the policy allows it.

authz = FastAPIAuthz(enforcer, subject=current_username)

# (subject, path, method): the path the app routes, and the HTTP method.
router = APIRouter(dependencies=[Depends(authz)])

# (subject, *values): your own objects and actions; {id} takes the path parameter.
archive = authz.require("projects/{id}", "archive")
@app.post("/projects/{id}/archive", dependencies=[Depends(archive)])

A request without a subject gets a 401 (with WWW-Authenticate: <challenge>), unless anonymous names the subject the policy knows them by; one that the policy refuses, a 403; one the engine couldn't answer (PolicyUnavailable), a 503. Paths with control characters (%0A...) get a 400 without asking the policy, as the app could route them elsewhere than the policy reads them. On WebSocket routes, the connection is closed with 1008 (policy violation) instead, 1013 (try again later) for the 503. Both dependencies return the subject (None for anonymous requests let through).

Parameters:

Name	Type	Description	Default
enforcer	Enforcer	a Casbin enforcer, a CedarEnforcer, or anything with enforce (a plain function or a coroutine function), with its policy loaded (await enforcer.load_policy() in the lifespan for an AsyncEnforcer). PolicyUnavailable refuses the request with a 503; any other exception it raises, with a 500: never the route.	required
subject	Subject | None	a dependency returning who makes the request (a user name, an id, an object for ABAC matchers), or None when nobody is authenticated. By default the user of Starlette's AuthenticationMiddleware (starlette_user).	None
anonymous	str | None	the subject of requests without one, for policies that open some routes to everybody (p, anonymous, /catalog, GET); None refuses them without asking the policy. A name no user can have (*anonymous), or a user who takes it gets its rules, and every anonymous request that user's.	None
challenge	str	the WWW-Authenticate of the 401s: the scheme your users log in with.	'Bearer'
metrics	Metrics | None	where to count the decisions (fastapi_alcazaba.authz); FastAPICache's by default.	None
audit	FastAPIAudit | None	the audit trail that records the requests the policy refuses (authz.denied: the 403s, and the 400s of paths with control characters) and those the engine couldn't decide (authz.unavailable), with what was asked, the route and the client. Not the 401s of requests without a subject, which authentication tells, nor enforce()'s checks in a route, whose answer the route decides what to do with.	None
actor	Actor | None	who the subject is, in those events: actor(subject), an ID; subject_id by default.	None

require

require(*values: str) -> Callable[..., Awaitable[Any]]

A dependency that checks (subject, *values) instead of the path and method: Depends(authz.require("articles", "publish")). {name} in a value takes the path parameter name ("projects/{project_id}"); {{ and }} are braces.

enforce async

enforce(subject: Any, *values: Any) -> bool

Whether the policy lets subject do values, counted in the metrics: for checks outside a dependency (a WebSocket message, an object loaded in the route). When the engine doesn't answer, an HTTPException 503: the route doesn't go on as if it had.

refusal async

refusal(subject: Any, *values: Any, connection: HTTPConnection | None = None) -> int | None

None when the policy lets subject (None for nobody) do values; else the status to refuse it with: 401 without a subject, 403 with one, 503 when the engine didn't answer. With the connection it's for, the refusals go to the audit trail.

path_refusal async

path_refusal(subject: Any, scope: Scope) -> int | None

refusal of (subject, path, method): the path the app routes and the HTTP method (WEBSOCKET for WebSockets; GET for HEAD, which runs the GET route). A 400 for paths with control characters, without asking the policy: the app could route them elsewhere than the policy reads them.

fastapi_alcazaba.authz.AuthzMiddleware

AuthzMiddleware(app: ASGIApp, enforcer: Enforcer, *, anonymous: str | None = None, challenge: str = 'Bearer', metrics: Metrics | None = None, audit: FastAPIAudit | None = None, actor: Actor | None = None)

Checks every request against the policy, with the user of Starlette's AuthenticationMiddleware. Add it before AuthenticationMiddleware, so that it runs after it:

app.add_middleware(AuthzMiddleware, enforcer=enforcer)
app.add_middleware(AuthenticationMiddleware, backend=BasicAuth())

Each request is checked as (user, path, method): the user's identity (anonymous when it isn't authenticated), the path the app routes (without the root_path a proxy or a mount adds in front) and the method, GET for HEAD and WEBSOCKET for WebSockets. Refused requests get a 403, or a 401 (with WWW-Authenticate: <challenge>) when they have no user, and a 503 when the engine didn't answer (PolicyUnavailable); WebSockets are closed with 1008 (policy violation), or 1013 (try again later). Paths with control characters (%0A...) get a 400 without asking the policy, as the app could route them elsewhere than the policy reads them. Every request is checked, OPTIONS too: add CORSMiddleware after this one (outside it), and it answers the preflights before they get here.

Parameters:

Name	Type	Description	Default
enforcer	Enforcer	a Casbin enforcer, a CedarEnforcer, or anything with enforce (a plain function or a coroutine function), with its policy loaded.	required
anonymous	str | None	the subject of requests without a user, a name no user can have (*anonymous); None refuses them without asking the policy.	None
challenge	str	the WWW-Authenticate of the 401s.	'Bearer'
metrics	Metrics | None	where to count the decisions (fastapi_alcazaba.authz); FastAPICache's by default.	None
audit	FastAPIAudit | None	the audit trail that records the refusals, as FastAPIAuthz's audit says.	None
actor	Actor | None	who the subject is, in those events; subject_id by default.	None

fastapi_alcazaba.authz.cedar.CedarEnforcer

CedarEnforcer(policies: str, *, entities: Sequence[Entity] = (), entities_of: Callable[[Any], Sequence[Entity] | Awaitable[Sequence[Entity]]] | None = None, schema: str | Mapping[str, Any] | None = None, principal_type: str = 'User', action_type: str = 'Action', resource_type: str = 'Resource')

An enforcer of Cedar policies, for FastAPIAuthz(enforcer) and AuthzMiddleware.

await enforce(subject, resource, action, context=None) asks Cedar whether subject may do action on resource. Cedar denies what no policy permits, and a forbid beats every permit. A policy that fails to evaluate (an attribute that isn't there, an entity that doesn't exist) refuses the request and logs why: Cedar would skip it, and a forbid that errors would let the request through.

Parameters:

Name	Type	Description	Default
policies	str	the policies, in Cedar's language. Invalid ones raise ValueError here.	required
entities	Sequence[Entity]	the entities every request sees: roles and their hierarchy, resources with their attributes.	()
entities_of	Callable[[Any], Sequence[Entity] | Awaitable[Sequence[Entity]]] | None	entities_of(subject): the entities of one request, such as the user and its roles from your database. A plain function, for what's at hand (the user the authentication already loaded), or a coroutine function, awaited, for what it reads through the network.	None
schema	str | Mapping[str, Any] | None	a Cedar schema (its language, or JSON): the policies are validated against it, and the entities checked.	None
principal_type	str	the entity type of subjects given as strings.	'User'
action_type	str	the entity type of actions.	'Action'
resource_type	str	the entity type of resources given as strings, paths too.	'Resource'

enforce async

enforce(subject: Any, resource: Any, action: Any, context: Mapping[str, Any] | None = None) -> bool

Whether the policies let subject do action on resource.

subject and resource are each a name (of principal_type or resource_type), an entity uid ({"type": "Team", "id": "core"}), or a whole entity with its attributes ({"uid": ..., "attrs": {"owner": ...}, "parents": [...]}), which this request sees: the user your authentication loaded, the document the route just read. action is a name of action_type or a uid. context is Cedar's context.

An entity may come more than once (from entities, entities_of and the arguments) only the same each time: one that differs raises ValueError rather than pick either.

fastapi_alcazaba.authz.opa.OPAEnforcer

OPAEnforcer(url: str, *, rule: str = 'authz.allow', token: str | SecretStr | None = None, timeout: float = 1.0, client: AsyncClient | None = None)

An enforcer that asks Open Policy Agent, for FastAPIAuthz(enforcer) and AuthzMiddleware.

await enforce(subject, resource, action, context=None) queries rule with the input {"subject": subject, "resource": resource, "action": action, "context": context}: the subject your dependency gives (a name, or a dict with the user's attributes and roles), the path or the value of authz.require, and the method or the action. Only a result of true allows; false refuses, and so does a rule that is undefined for the input (with a warning: give it a default, default allow := false).

Fails closed: a service that doesn't answer within timeout, or answers with a 5xx, raises PolicyUnavailable, and the request gets a 503. Its own client gives each call a deadline and the service a circuit breaker (ResilientTransport): when it's down, requests are refused at once instead of each waiting for it. A query OPA refuses (a 4xx: a wrong token, an input that isn't JSON) and a result that isn't a boolean (a rule that names a package, or a rule that gives an object) raise RuntimeError, a 500: they're mistakes to fix, not refusals.

Parameters:

Name	Type	Description	Default
url	str	where OPA listens, http://localhost:8181 for a sidecar.	required
rule	str	the rule to query, as Rego references it after data.: authz.allow is the rule allow of package authz.	'authz.allow'
token	str | SecretStr | None	the bearer token OPA asks for, when it runs with --authentication=token.	None
timeout	float	seconds for each query, the answer included.	1.0
client	AsyncClient | None	the httpx2 client to ask OPA with (for its TLS, its proxy, a transport of your own); one of its own by default, with a ResilientTransport. aclose() closes only its own.	None

enforce async

enforce(subject: Any, resource: Any, action: Any, context: Mapping[str, Any] | None = None) -> bool

Whether rule is true for the input (subject, resource, action, context). Every value goes to OPA as JSON: names, numbers, lists and dicts of them.

aclose async

aclose() -> None

Closes the client it made (not one it was given): in the lifespan, at shutdown.

fastapi_alcazaba.authz.PolicyUnavailable

Bases: Exception

The policy engine didn't answer: a service that is down, slow or overloaded. The request is refused with a 503 (1013, try again later, for a WebSocket), neither allowed nor taken for a refusal of the policy. Raise it from an enforcer of your own.

fastapi_alcazaba.authz.starlette_user

starlette_user(connection: HTTPConnection) -> str | None

The subject of Starlette's AuthenticationMiddleware: its user's identity (not the display_name, which users may choose), None when it isn't authenticated.

fastapi_alcazaba.authz.subject_id

subject_id(subject: Any) -> str | int | uuid.UUID | None

The ID of a subject, for the audit trail: itself when it's an ID; a Cedar entity's id; a mapping's sub, id or name; an object's id or identity.

fastapi_alcazaba.authz.Enforcer

Bases: Protocol

What FastAPIAuthz needs of a policy engine: Casbin's enforcers have it, CedarEnforcer gives it to Cedar.

enforce

enforce(*values: Any) -> bool | Awaitable[bool]

Whether the policy allows the request values (sub, obj, act...): a bool, or an awaitable of one for an engine that answers through the network.

fastapi_alcazaba.authz.WEBSOCKET module-attribute

WEBSOCKET = 'WEBSOCKET'