# `sandloom.sdk.authz`

Authorization: policies and atoms, access facts, the billing-account subject (`Account`).

Public API 1.0 (spec §2.2, Appendix C).

`from sandloom.sdk.authz import <name>`. Signatures in a terminal: `sandloom docs <name>`. Beginner names are the ones generated code and the examples use; any other name is advanced (mark its import line `# sandloom: advanced <reason>` in generated code and examples).

## Start with (beginner)

| Name | Kind | Summary |
| --- | --- | --- |
| [`authorize_resource`](<#authorize_resource>) | function | Load a resource by path id, authorize, and return it to the endpoint. |
| [`Credits`](<#credits>) | class | Subject's live credit balance is at least `minimum`. |
| [`OrgRole`](<#orgrole>) | class | Subject holds at least `minimum` role in the request's organization. |
| [`Owner`](<#owner>) | class | Subject owns the request resource. |
| [`require`](<#require>) | function | Return a dependency enforcing `policy` (no request resource). |
| [`Subscribed`](<#subscribed>) | class | Subject's subscription grants access (coarse, claim-based). |
| [`SubscriptionActive`](<#subscriptionactive>) | class | Subject's subscription is active right now (live, authoritative). |

## Advanced

| Name | Kind | Summary |
| --- | --- | --- |
| [`AccessDecision`](<#accessdecision>) | class | The outcome of evaluating a policy against an `AuthzContext`. |
| [`AccessDenied`](<#accessdenied>) | exception | Raised by `enforce` (and the HTTP guards) when a policy is not satisfied. |
| [`AccessFact`](<#accessfact>) | class | Declarative binding of a fact name to its three resolution paths. |
| [`Account`](<#account>) | class | The billing account an atom checks (`of=`); build it with `user`, `org` or `bound_tenant`. |
| [`ACTIVE_SUBSCRIPTION_STATUSES`](<#active_subscription_statuses>) | constant |  |
| [`And`](<#and>) | class | Conjunction: all operands must hold. Short-circuits on first non-allow. |
| [`authorize`](<#authorize>) | async function | Inline check over an already-loaded `resource`; raises on failure. |
| [`AuthzContext`](<#authzcontext>) | class | Inputs available to a `Requirement` during evaluation. |
| [`Category`](<#category>) | class | Mandatory denial categories (TZ §10.2). |
| [`ClaimRule`](<#claimrule>) | class | Declarative claim -\> bool interpretation: wire-safe, no callables. |
| [`current_context`](<#current_context>) | async function | Authenticated subject from the Bearer access token (cached per request). |
| [`enforce`](<#enforce>) | async function | Evaluate `policy` for `subject`; raise `AccessDenied` on deny. |
| [`Entitled`](<#entitled>) | class | Subject satisfies the registered access fact `fact_name`. |
| [`ENTITLEMENTS`](<#entitlements>) | constant |  |
| [`EntitlementsAdapter`](<#entitlementsadapter>) | class | Answers subscription / credit questions (`SubscriptionActive`/`Credits`). |
| [`FactDenied`](<#factdenied>) | exception | A resolver's *definite* denial for a reason other than "the fact is False". |
| [`FactOwnerUnreachable`](<#factownerunreachable>) | exception | The fact's owner is bound but could not describe its facts (503). |
| [`FactUnavailable`](<#factunavailable>) | exception | A fact could not be resolved in any acceptable mode (a misconfiguration). |
| [`Feature`](<#feature>) | class | The account's plan enables the feature flag `key` (`of=`, §7.2). |
| [`lookup_access_fact`](<#lookup_access_fact>) | async function | The fact `name`: the local registration, else the owner's public one. |
| [`make_rebuild_gate`](<#make_rebuild_gate>) | function | A startup task that rebuilds an authz projection from its sources. |
| [`MEMBERSHIP`](<#membership>) | constant |  |
| [`MembershipAdapter`](<#membershipadapter>) | class | Answers organization-role questions for `OrgRole`. |
| [`MODE_AUTO`](<#mode_auto>) | constant |  |
| [`MODE_CLAIM`](<#mode_claim>) | constant |  |
| [`MODE_LIVE`](<#mode_live>) | constant |  |
| [`MODE_PROJECTION`](<#mode_projection>) | constant |  |
| [`Not`](<#not>) | class | Negation of a single requirement. |
| [`Or`](<#or>) | class | Disjunction: any operand may hold. Short-circuits on first allow. |
| [`Perm`](<#perm>) | class | Subject holds the fine-grained permission `perm`. |
| [`Permissions`](<#permissions>) | class | Predefined fine-grained permission codes. |
| [`PolicyIndeterminate`](<#policyindeterminate>) | exception | Raised by an atom's `evaluate` when it cannot answer at all. |
| [`register_authz_projection`](<#register_authz_projection>) | function | Register an authorization projection, its consumers and rebuilder. |
| [`register_live_adapter`](<#register_live_adapter>) | function | Bind the adapter the live atoms use for `kind` — CONSUMER-side config. |
| [`Requirement`](<#requirement>) | class | An authorization predicate, composable with `& \\| ~`. |
| [`resolve_fact`](<#resolve_fact>) | async function | Resolve `fact` to a boolean, caching the result on `ctx`. |
| [`Role`](<#role>) | class | Subject holds the global role `role`. |
| [`Roles`](<#roles>) | class | Predefined global role codes. |
| [`service_subject`](<#service_subject>) | function | A subject representing another service calling across a module boundary. |
| [`SOURCE_HTTP`](<#source_http>) | constant |  |
| [`SOURCE_INTERNAL`](<#source_internal>) | constant |  |
| [`SOURCE_SYSTEM`](<#source_system>) | constant |  |
| [`SOURCE_TASK`](<#source_task>) | constant |  |
| [`system_subject`](<#system_subject>) | function | A subject representing the platform itself (trusted background work). |
| [`TIER_CLAIM`](<#tier_claim>) | constant |  |
| [`TIER_LIVE`](<#tier_live>) | constant |  |
| [`TIER_RESOURCE`](<#tier_resource>) | constant |  |
| [`WithinLimit`](<#withinlimit>) | class | `used` is strictly below the account's plan limit for `key`. |

## Beginner names

### `authorize_resource`

```
def authorize_resource(
    *,
    loader: Callable[[Any, AsyncSession], Awaitable[Any]],
    by: str,
    policy: Requirement,
    cast: Callable[[str], Any] | None = None,
    not_found_detail: str = 'Not found',
    status_code: int | None = None,
) -> Callable[..., Awaitable[Any]]
```

Load a resource by path id, authorize, and return it to the endpoint.

**Args:**

- `loader`: `async (id_value, db) -> entity | None` — the module supplies its own loader (keeps `shared` free of ORM imports).
- `by`: name of the path parameter holding the resource id.
- `policy`: requirement to enforce; resource atoms (`Owner`) read the loaded entity.
- `cast`: optional converter applied to the raw path value before `loader` (e.g. `UUID`). Defaults to passing the raw string.
- `not_found_detail`: 404 message when the loader returns `None`.
- `status_code`: explicit status override for a definite denial (default: the decision's own status).

The resource is fetched **once**: claim atoms evaluate first, so a failing cheap check raises without ever loading; otherwise the entity loaded for the ownership check is the same one returned to the endpoint.

### `Credits`

```
class Credits(_TenantScopedLiveAtom)(
    minimum: int,
    *,
    of: Account = <object object>,
    adapter: Any = None,
) -> None
```

Subject's live credit balance is at least `minimum`.

Reads the real-time balance via the entitlements contract (never a stale token claim — this guards money). `of=` names the billing account.

#### `Credits.evaluate`

```
async def evaluate(self, ctx: AuthzContext) -> bool
```

Return True iff the subject satisfies this requirement.

Raise `PolicyIndeterminate` (never return `False`) when the requirement cannot be evaluated at all.

#### `Credits.describe`

```
def describe(self) -> str
```

Human-readable label used in 403 detail messages.

### `OrgRole`

```
class OrgRole(Requirement)(
    minimum: str,
    *,
    org_param: str = 'org_id',
    adapter: Any = None,
) -> None
```

Subject holds at least `minimum` role in the request's organization.

Relationship (ReBAC) authorization: org membership/role is per-organization and not carried in the global JWT, so it is resolved live through the membership contract (local call in a monolith, an RPC when tenancy is a separate service; 501 when neither). The organization id is read from the request path parameter `org_param`, else `X-Tenant-Id`, else the bound tenant; a missing organization is indeterminate (denies even under `Not`).

For a hot path, a module may instead maintain an authorization projection of memberships and gate on a projection-backed `Entitled` fact.

#### `OrgRole.org_params`

```
def org_params(self) -> frozenset[str]
```

Path parameters this requirement may read an organisation from.

Non-empty for an atom that resolves an organisation (path parameter, `X-Tenant-Id`, else the bound tenant): `create_app` then refuses a route that solves the gate before its `TenantScope()` (SANDLOOM-B007).

#### `OrgRole.risks`

```
def risks(self) -> frozenset[str]
```

The framework risks a gate enforcing this requirement carries (ADR-0009 item 6): `{"admin"}` for an admin role, else none. An `Or` carries its branches' risks; a `Not` grants no authority.

#### `OrgRole.contracts`

```
def contracts(self) -> frozenset[str]
```

Sync contracts evaluating this requirement may call (wiring time).

A module's derived `consumes` includes them (optional), so the startup preflight reports an unbound one and `sandloom add deploy` binds it in a split deployment. Default: none (a claims-only atom).

#### `OrgRole.adapter`

```
def adapter(self) -> Any
```

#### `OrgRole.evaluate`

```
async def evaluate(self, ctx: AuthzContext) -> bool
```

Return True iff the subject satisfies this requirement.

Raise `PolicyIndeterminate` (never return `False`) when the requirement cannot be evaluated at all.

#### `OrgRole.describe`

```
def describe(self) -> str
```

Human-readable label used in 403 detail messages.

### `Owner`

```
class Owner(Requirement)(get_owner: Callable[[Any], UUID | None]) -> None
```

Subject owns the request resource.

`get_owner` maps the loaded resource to its owner id; it is supplied by the module declaring the policy (e.g. `lambda post: post.author_id`), so the atom never needs to know the resource's type.

#### `Owner.evaluate`

```
async def evaluate(self, ctx: AuthzContext) -> bool
```

Return True iff the subject satisfies this requirement.

Raise `PolicyIndeterminate` (never return `False`) when the requirement cannot be evaluated at all.

#### `Owner.describe`

```
def describe(self) -> str
```

Human-readable label used in 403 detail messages.

### `require`

```
def require(
    policy: Requirement,
    *,
    status_code: int | None = None,
) -> Callable[..., Awaitable[UserContext]]
```

Return a dependency enforcing `policy` (no request resource).

Use in `dependencies=[Depends(require(...))]` or as a parameter when the endpoint wants the resulting `UserContext`. A denial answers with the decision's own status (e.g. `Subscribed(of=Account.user())` → 402, a missing contract → 501); `status_code` is an explicit override for definite denials.

A claims-only policy (every atom tier 1) never opens a database session / Unit of Work; only policies that may reach a live atom depend on `get_db`.

Raises `ValueError` at wiring time if `policy` contains a resource-scoped atom (e.g. `Owner`): `require` has no request resource, so such a policy belongs in `authorize_resource` / `authorize`.

### `Subscribed`

```
class Subscribed(Requirement)(
    statuses: frozenset[str] | None = None,
    *,
    of: Account = <object object>,
    claim_key: str = 'subscription_status',
) -> None
```

Subject's subscription grants access (coarse, claim-based).

`of=Account.user()`: reads the token's `subscription_status` claim with ZERO I/O, interpreted by the `subscription.active` access fact — billing's declarative rule, the same predicate as the live check (`active`/`trialing` pass; `past_due` passes during the grace period unless `BILLING_DENY_ON_PAST_DUE`). The fact is the local registration or, in a process without billing, the one billing publishes over `billing.facts.v1` (described once, cached).

Provenance is honoured: when the token's authoritative provenance shows billing did NOT contribute (it failed at mint, or the issuer could not reach it) the claim proves nothing, so the atom escalates to billing's LIVE answer instead of denying 402 for the token's whole lifetime (503 indeterminate when no live path exists). With no subscription owner in the deployment at all, only `active`/`trialing` pass. May be up to the token TTL stale; for real-time / money-sensitive checks use `sandloom.shared.authz.atoms_live.SubscriptionActive` / `Credits`.

`of=Account.org(...)`: the token carries NO per-organization subscription claim (claims are personal-only), so this atom resolves LIVE — it delegates to `SubscriptionActive(of=...)` and is a tier-3 atom. `of=Account.bound_tenant()` resolves the bound tenant's kind first (personal → the claim, organization → live).

Passing explicit `statuses` pins the accepted set verbatim (no grace interpretation).

#### `Subscribed.tenant_type`

```
tenant_type  (property)
```

#### `Subscribed.org_params`

```
def org_params(self) -> frozenset[str]
```

Path parameters this requirement may read an organisation from.

Non-empty for an atom that resolves an organisation (path parameter, `X-Tenant-Id`, else the bound tenant): `create_app` then refuses a route that solves the gate before its `TenantScope()` (SANDLOOM-B007).

#### `Subscribed.contracts`

```
def contracts(self) -> frozenset[str]
```

Sync contracts evaluating this requirement may call (wiring time).

A module's derived `consumes` includes them (optional), so the startup preflight reports an unbound one and `sandloom add deploy` binds it in a split deployment. Default: none (a claims-only atom).

#### `Subscribed.evaluate`

```
async def evaluate(self, ctx: AuthzContext) -> bool
```

Return True iff the subject satisfies this requirement.

Raise `PolicyIndeterminate` (never return `False`) when the requirement cannot be evaluated at all.

#### `Subscribed.describe`

```
def describe(self) -> str
```

Human-readable label used in 403 detail messages.

### `SubscriptionActive`

```
class SubscriptionActive(_TenantScopedLiveAtom)(
    *,
    of: Account = <object object>,
    adapter: Any = None,
) -> None
```

Subject's subscription is active right now (live, authoritative).

Live counterpart to the claim-based `sandloom.shared.authz.atoms_claims.Subscribed`. Use this when staleness is unacceptable (e.g. gating immediately after a purchase); use `Subscribed` for the cheap, zero-I/O common path. `of=` names the billing account (spec §7).

#### `SubscriptionActive.evaluate`

```
async def evaluate(self, ctx: AuthzContext) -> bool
```

Return True iff the subject satisfies this requirement.

Raise `PolicyIndeterminate` (never return `False`) when the requirement cannot be evaluated at all.

#### `SubscriptionActive.describe`

```
def describe(self) -> str
```

Human-readable label used in 403 detail messages.

## Advanced names

### `AccessDecision`

```
class AccessDecision(
    allowed: bool,
    code: str = 'ok',
    category: str | None = None,
    reason: str = '',
    http_status: int = 200,
    failed_requirement: str | None = None,
    policy_code: str | None = None,
    metadata: dict[str, Any] = <factory>,
    indeterminate: bool = False,
) -> None
```

The outcome of evaluating a policy against an `AuthzContext`.

`allowed` is the only field that matters for the boolean question; the rest exist so a denial is reportable, auditable and observable.

#### `AccessDecision.allow`

```
@classmethod
def allow(cls, *, policy_code: str | None = None) -> AccessDecision
```

#### `AccessDecision.deny`

```
@classmethod
def deny(
    cls,
    *,
    code: str,
    category: str = 'forbidden',
    reason: str = '',
    http_status: int | None = None,
    failed_requirement: str | None = None,
    policy_code: str | None = None,
    metadata: dict[str, Any] | None = None,
) -> AccessDecision
```

#### `AccessDecision.error`

```
@classmethod
def error(
    cls,
    *,
    code: str,
    category: str = 'policy_misconfigured',
    reason: str = '',
    http_status: int | None = None,
    failed_requirement: str | None = None,
    metadata: dict[str, Any] | None = None,
) -> AccessDecision
```

An *indeterminate* denial: the requirement could not be evaluated.

Distinct from `deny` (the subject definitively lacks the requirement): combinators propagate it unchanged, so `~Entitled(x)` with `x` unregistered / its service down DENIES with this category and status instead of inverting the failure into an allow.

#### `AccessDecision.with_policy_code`

```
def with_policy_code(self, policy_code: str | None) -> AccessDecision
```

Copy of this decision tagged with `policy_code`.

#### `AccessDecision.audit_fields`

```
def audit_fields(self) -> dict[str, Any]
```

The subset safe and useful to record on a deny (no PII).

### `AccessDenied`

```
class AccessDenied(HTTPException)(
    decision: AccessDecision,
    *,
    status_code: int | None = None,
) -> None
```

Raised by `enforce` (and the HTTP guards) when a policy is not satisfied.

Carries the full `AccessDecision` so the caller (HTTP adapter, task runner) can map it to the right transport response. Subclasses FastAPI's `HTTPException` so a module calling `enforce()` inside an endpoint gets the decision's status (402/403/501/…) and `{"detail", "code"}` body from the framework HTTP handler instead of an unhandled 500. Non-HTTP callers (workers, services) still catch the typed `AccessDenied` and read `.decision`. It also exposes the framework error-protocol attributes `http_status` / `code` / `detail`.

`status_code` explicitly overrides the decision's status (transport override, e.g. `require(..., status_code=402)`).

### `AccessFact`

```
class AccessFact(
    name: str,
    owner_module: str,
    claim_key: str | None = None,
    projection_reader: ProjectionReader | None = None,
    live_service: str | None = None,
    live_resolver: LiveResolver | None = None,
    min_freshness: str = 'claim',
    default_mode: str = 'auto',
    deny_category: str | None = None,
    claim_rule: ClaimRule | Callable[[], ClaimRule] | None = None,
) -> None
```

Declarative binding of a fact name to its three resolution paths.

A module registers one of these in `register(core)` via `core.register_access_fact(...)`, wiring together mechanisms that already exist: its claims contributor (`claim_key`), an optional local projection (`projection_reader`) and its service contract (`live_service` + `live_resolver`).

#### `AccessFact.current_claim_rule`

```
def current_claim_rule(self) -> ClaimRule | None
```

The declarative claim rule in effect now (`None`: not declarative).

#### `AccessFact.interpret_claim`

```
def interpret_claim(self, value: Any) -> bool
```

Claim value -\> bool, exactly as the owner interprets it.

#### `AccessFact.to_wire`

```
def to_wire(self) -> dict[str, Any]
```

Wire-safe descriptor published by the owner's `describe()`.

#### `AccessFact.build_facts_contract`

```
@classmethod
def build_facts_contract(cls, core: Any, owner: str) -> Any
```

The `<owner>.facts.v1` provider (called by the kernel on register).

A duck-typed hook, so the kernel exposes the contract without importing the authorization subsystem.

### `Account`

```
class Account(
    kind: AccountKind,
    param: str | None = None,
    reason: str | None = None,
) -> None
```

The billing account an atom checks (`of=`); build it with `user`, `org` or `bound_tenant`.

#### `Account.user`

```
@staticmethod
def user(*, reason: str | None = None) -> Account
```

The caller's personal account. `reason=` waives SANDLOOM-C027 on a route that carries an organization parameter.

#### `Account.org`

```
@staticmethod
def org(*, param: str | None = 'org_id') -> Account
```

An organization account: `param` (path, then query), else the `X-Tenant-Id` header, else the bound tenant (`param=None`: header or bound tenant only).

#### `Account.bound_tenant`

```
@staticmethod
def bound_tenant() -> Account
```

The account of the bound tenant, by its kind (personal or organization; canonical / unknown are indeterminate).

#### `Account.describe`

```
def describe(self) -> str
```

### `ACTIVE_SUBSCRIPTION_STATUSES`

```
ACTIVE_SUBSCRIPTION_STATUSES: frozenset = frozenset({'active', 'trialing'})
```

*(no docstring)*

### `And`

```
class And(Requirement)(*parts: Requirement) -> None
```

Conjunction: all operands must hold. Short-circuits on first non-allow.

A definite deny or an indeterminate operand both stop evaluation and are returned as-is (both deny; the indeterminate keeps its 5xx category).

#### `And.evaluate`

```
async def evaluate(self, ctx: AuthzContext) -> bool
```

Return True iff the subject satisfies this requirement.

Raise `PolicyIndeterminate` (never return `False`) when the requirement cannot be evaluated at all.

#### `And.decide`

```
async def decide(self, ctx: AuthzContext) -> AccessDecision
```

Structured result. Default: allow iff `evaluate` is True.

#### `And.mentions_resource`

```
def mentions_resource(self) -> bool
```

True if evaluating this requirement needs the request resource.

Used by `require()` to reject resource-scoped policies at wiring time (they belong in `authorize_resource`/`authorize`), turning a runtime 500 into a clear startup error.

#### `And.org_params`

```
def org_params(self) -> frozenset[str]
```

Path parameters this requirement may read an organisation from.

Non-empty for an atom that resolves an organisation (path parameter, `X-Tenant-Id`, else the bound tenant): `create_app` then refuses a route that solves the gate before its `TenantScope()` (SANDLOOM-B007).

#### `And.contracts`

```
def contracts(self) -> frozenset[str]
```

Sync contracts evaluating this requirement may call (wiring time).

A module's derived `consumes` includes them (optional), so the startup preflight reports an unbound one and `sandloom add deploy` binds it in a split deployment. Default: none (a claims-only atom).

#### `And.risks`

```
def risks(self) -> frozenset[str]
```

The framework risks a gate enforcing this requirement carries (ADR-0009 item 6): `{"admin"}` for an admin role, else none. An `Or` carries its branches' risks; a `Not` grants no authority.

#### `And.describe`

```
def describe(self) -> str
```

Human-readable label used in 403 detail messages.

### `authorize`

```
async def authorize(
    resource: Any,
    policy: Requirement,
    *,
    ctx: UserContext,
    request: Request | None = None,
    db: AsyncSession | None = None,
    status_code: int | None = None,
) -> None
```

Inline check over an already-loaded `resource`; raises on failure.

The escape hatch for cases that don't fit `authorize_resource` (multiple resources, ownership known only after a non-trivial query, authz interleaved with business logic). The endpoint owns the single fetch. Raises `AccessDenied` with the decision's status (or the explicit `status_code` override for a definite denial).

### `AuthzContext`

```
class AuthzContext(
    subject: UserContext,
    *,
    request: Request | None = None,
    db: AsyncSession | None = None,
    resource: Any = <object object>,
    loader: Callable[[AuthzContext], Awaitable[Any]] | None = None,
    source: str = 'http',
    core: Any = None,
) -> None
```

Inputs available to a `Requirement` during evaluation.

#### `AuthzContext.get_resource`

```
async def get_resource(self) -> Any
```

Return the request resource, loading it once on first access.

Loading is lazy so that a policy like `Perm("x") & Owner(...)` never fetches the resource when the cheap claim atom already fails.

#### `AuthzContext.has_resource`

```
has_resource  (property)
```

#### `AuthzContext.core`

```
def core(self) -> Any
```

Resolve the bound `CoreFacade`, or None.

Order: an explicitly-passed core (non-HTTP paths pass `core=` to `enforce`) → the app the request was served by. There is deliberately **no** process-global fallback — the subsystem never reaches a global singleton (that would break multi-app / test isolation). When no core is bound, atoms that need one (live / fact / `OrgRole`) degrade explicitly (indeterminate 501 / `FactUnavailable`).

### `Category`

```
class Category
```

Mandatory denial categories (TZ §10.2).

### `ClaimRule`

```
class ClaimRule(kind: str = 'truthy', values: frozenset[Any] = frozenset()) -> None
```

Declarative claim -\> bool interpretation: wire-safe, no callables.

`ClaimRule.one_of({"active", "trialing"})` is True iff the claim value is one of the listed scalars; `ClaimRule.truthy()` is `bool(value)`. The owner describes its facts with these, so a consumer in another process interprets the token claim EXACTLY like the owner does locally (one source of truth). A rule that depends on the owner's settings (billing's `past_due` grace) is a zero-argument factory on the fact (`AccessFact(claim_rule=factory)`), serialized with the owner's current settings at `describe()` time.

#### `ClaimRule.one_of`

```
@classmethod
def one_of(cls, values: Iterable[Any]) -> ClaimRule
```

#### `ClaimRule.truthy`

```
@classmethod
def truthy(cls) -> ClaimRule
```

#### `ClaimRule.interpret`

```
def interpret(self, value: Any) -> bool
```

#### `ClaimRule.to_wire`

```
def to_wire(self) -> dict[str, Any]
```

#### `ClaimRule.from_wire`

```
@classmethod
def from_wire(cls, raw: Any) -> ClaimRule | None
```

Parse a wire rule; `None` for anything not understood.

Fail closed: a consumer that cannot parse the rule never reads the claim and escalates to the owner's live answer instead.

### `current_context`

```
async def current_context(
    request: Request,
    credentials: HTTPAuthorizationCredentials | None = Depends(dependency=<fastapi.security.http.HTTPBearer object>, use_cache=True, scope=None),
) -> UserContext
```

Authenticated subject from the Bearer access token (cached per request).

### `enforce`

```
async def enforce(
    policy: Requirement,
    *,
    subject: UserContext,
    db: AsyncSession | None = None,
    request: Request | None = None,
    resource: Any = <object object>,
    loader: Callable[[AuthzContext], Awaitable[Any]] | None = None,
    source: str = 'internal',
    core: Any = None,
    policy_code: str | None = None,
) -> AccessDecision
```

Evaluate `policy` for `subject`; raise `AccessDenied` on deny.

The one chokepoint. HTTP/use-case/task adapters all call this. Returns the allow `AccessDecision` on success so callers may inspect it.

### `Entitled`

```
class Entitled(Requirement)(
    fact_name: str,
    *,
    category: str | None = None,
    code: str | None = None,
    mode: str | None = None,
) -> None
```

Subject satisfies the registered access fact `fact_name`.

**Args:**

- `fact_name`: name of an `AccessFact` registered via `core.register_access_fact(...)`.
- `category`: denial category when the fact is False. Defaults to the fact's own `deny_category` (billing's subscription facts declare `Category.PAYMENT_REQUIRED` → 402), else `forbidden` (403). A resolver raising `FactDenied` picks its own category.
- `mode`: force a resolution mode, overriding the `ACCESS` setting (for tests / special endpoints). Normally left `None`.

#### `Entitled.contracts`

```
def contracts(self) -> frozenset[str]
```

Sync contracts evaluating this requirement may call (wiring time).

A module's derived `consumes` includes them (optional), so the startup preflight reports an unbound one and `sandloom add deploy` binds it in a split deployment. Default: none (a claims-only atom).

#### `Entitled.evaluate`

```
async def evaluate(self, ctx: AuthzContext) -> bool
```

Return True iff the subject satisfies this requirement.

Raise `PolicyIndeterminate` (never return `False`) when the requirement cannot be evaluated at all.

#### `Entitled.decide`

```
async def decide(self, ctx: AuthzContext) -> AccessDecision
```

Structured result. Default: allow iff `evaluate` is True.

#### `Entitled.describe`

```
def describe(self) -> str
```

Human-readable label used in 403 detail messages.

### `ENTITLEMENTS`

```
ENTITLEMENTS: str = 'entitlements'
```

*(no docstring)*

### `EntitlementsAdapter`

```
class EntitlementsAdapter(Protocol)
```

Answers subscription / credit questions (`SubscriptionActive`/`Credits`).

#### `EntitlementsAdapter.subscription_active`

```
async def subscription_active(
    self,
    service: Any,
    *,
    tenant_type: str,
    tenant_id: UUID,
    ctx: AuthzContext,
) -> bool
```

#### `EntitlementsAdapter.credit_balance`

```
async def credit_balance(
    self,
    service: Any,
    *,
    tenant_type: str,
    tenant_id: UUID,
    ctx: AuthzContext,
) -> int
```

### `FactDenied`

```
class FactDenied(Exception)(
    reason: str,
    *,
    category: str = 'forbidden',
    code: str | None = None,
) -> None
```

A resolver's *definite* denial for a reason other than "the fact is False".

Raised by a projection reader / live resolver when the fact is inapplicable to the subject — e.g. `organization.subscription.active` with no organization in context, or for an org the subject is not a member of. `sandloom.shared.authz.atoms_fact.Entitled` maps it to a definite deny with `category` (default `forbidden`, 403) instead of the fact's `deny_category` — "you may not ask" is not "pay to unlock".

### `FactOwnerUnreachable`

```
class FactOwnerUnreachable(FactUnavailable)
```

The fact's owner is bound but could not describe its facts (503).

Distinct from "not configured" (501): the deployment is correct, the owner is down. Never a silent allow or a definite deny.

### `FactUnavailable`

```
class FactUnavailable(Exception)
```

A fact could not be resolved in any acceptable mode (a misconfiguration).

Raised when an explicitly-requested mode cannot be honored (live service not bound, no projection reader) or `auto` exhausts every option. The atom maps this to a `policy_misconfigured` decision — never a silent allow.

### `Feature`

```
class Feature(_AccountClaimAtom)(
    key: str,
    *,
    of: Account = <object object>,
    claim_key: str = 'features',
) -> None
```

The account's plan enables the feature flag `key` (`of=`, §7.2).

#### `Feature.describe`

```
def describe(self) -> str
```

Human-readable label used in 403 detail messages.

### `lookup_access_fact`

```
async def lookup_access_fact(ctx: AuthzContext, name: str) -> AccessFact | None
```

The fact `name`: the local registration, else the owner's public one.

Order: `core.get_access_fact(name)` (owner co-deployed) → every bound `<owner>.facts.v1` with no local provider (the owner named by the fact's first dotted segment first), whose `describe()` is cached on the core for `REMOTE_FACT_TTL_SECONDS`. `None` when no owner knows the fact (the caller reports 501); `FactOwnerUnreachable` (503) when a bound owner could not be asked and none of the others knows it.

### `make_rebuild_gate`

```
def make_rebuild_gate(rebuilder: Any, *, name: str = '') -> StartupTask
```

A startup task that rebuilds an authz projection from its sources.

Captures the `rebuilder` supplied at registration in its closure (no process-global / registry lookup at run time). Install as `critical=True` so the process refuses traffic until the projection is populated — otherwise an empty table makes every fact resolve False and the consumer fail-closed-denies all callers. The rebuilder must be idempotent (the protocol guarantees it), so re-running on boot is safe; for large projections prefer `sandloom projection rebuild <name>` in the one-shot migrate step with `rebuild_on_startup=False`.

### `MEMBERSHIP`

```
MEMBERSHIP: str = 'membership'
```

*(no docstring)*

### `MembershipAdapter`

```
class MembershipAdapter(Protocol)
```

Answers organization-role questions for `OrgRole`.

#### `MembershipAdapter.has_role_at_least`

```
async def has_role_at_least(
    self,
    service: Any,
    *,
    org_id: UUID,
    user_id: UUID,
    minimum: str,
    ctx: AuthzContext,
) -> bool
```

### `MODE_AUTO`

```
MODE_AUTO: str = 'auto'
```

*(no docstring)*

### `MODE_CLAIM`

```
MODE_CLAIM: str = 'claim'
```

*(no docstring)*

### `MODE_LIVE`

```
MODE_LIVE: str = 'live'
```

*(no docstring)*

### `MODE_PROJECTION`

```
MODE_PROJECTION: str = 'projection'
```

*(no docstring)*

### `Not`

```
class Not(Requirement)(inner: Requirement) -> None
```

Negation of a single requirement.

Negates only a *definite* answer: an indeterminate inner decision is returned unchanged (deny), so `~Entitled("account.suspended")` never allows because the fact is unregistered or its service is down.

#### `Not.evaluate`

```
async def evaluate(self, ctx: AuthzContext) -> bool
```

Return True iff the subject satisfies this requirement.

Raise `PolicyIndeterminate` (never return `False`) when the requirement cannot be evaluated at all.

#### `Not.decide`

```
async def decide(self, ctx: AuthzContext) -> AccessDecision
```

Structured result. Default: allow iff `evaluate` is True.

#### `Not.mentions_resource`

```
def mentions_resource(self) -> bool
```

True if evaluating this requirement needs the request resource.

Used by `require()` to reject resource-scoped policies at wiring time (they belong in `authorize_resource`/`authorize`), turning a runtime 500 into a clear startup error.

#### `Not.org_params`

```
def org_params(self) -> frozenset[str]
```

Path parameters this requirement may read an organisation from.

Non-empty for an atom that resolves an organisation (path parameter, `X-Tenant-Id`, else the bound tenant): `create_app` then refuses a route that solves the gate before its `TenantScope()` (SANDLOOM-B007).

#### `Not.contracts`

```
def contracts(self) -> frozenset[str]
```

Sync contracts evaluating this requirement may call (wiring time).

A module's derived `consumes` includes them (optional), so the startup preflight reports an unbound one and `sandloom add deploy` binds it in a split deployment. Default: none (a claims-only atom).

#### `Not.describe`

```
def describe(self) -> str
```

Human-readable label used in 403 detail messages.

### `Or`

```
class Or(Requirement)(*parts: Requirement) -> None
```

Disjunction: any operand may hold. Short-circuits on first allow.

Only a genuinely satisfied operand allows — and one that does allows even if a sibling is indeterminate (`Role("admin") | Entitled("x")` admits an admin while the `x` fact is unregistered): the grant comes from the satisfied operand, never from inverting or guessing the broken one. When none allows and at least one operand was indeterminate, the disjunction is indeterminate (the first such decision is returned, preserving its category/status) — a broken branch is reported, not silently folded into a plain 403.

#### `Or.evaluate`

```
async def evaluate(self, ctx: AuthzContext) -> bool
```

Return True iff the subject satisfies this requirement.

Raise `PolicyIndeterminate` (never return `False`) when the requirement cannot be evaluated at all.

#### `Or.decide`

```
async def decide(self, ctx: AuthzContext) -> AccessDecision
```

Structured result. Default: allow iff `evaluate` is True.

#### `Or.mentions_resource`

```
def mentions_resource(self) -> bool
```

True if evaluating this requirement needs the request resource.

Used by `require()` to reject resource-scoped policies at wiring time (they belong in `authorize_resource`/`authorize`), turning a runtime 500 into a clear startup error.

#### `Or.org_params`

```
def org_params(self) -> frozenset[str]
```

Path parameters this requirement may read an organisation from.

Non-empty for an atom that resolves an organisation (path parameter, `X-Tenant-Id`, else the bound tenant): `create_app` then refuses a route that solves the gate before its `TenantScope()` (SANDLOOM-B007).

#### `Or.contracts`

```
def contracts(self) -> frozenset[str]
```

Sync contracts evaluating this requirement may call (wiring time).

A module's derived `consumes` includes them (optional), so the startup preflight reports an unbound one and `sandloom add deploy` binds it in a split deployment. Default: none (a claims-only atom).

#### `Or.risks`

```
def risks(self) -> frozenset[str]
```

The framework risks a gate enforcing this requirement carries (ADR-0009 item 6): `{"admin"}` for an admin role, else none. An `Or` carries its branches' risks; a `Not` grants no authority.

#### `Or.describe`

```
def describe(self) -> str
```

Human-readable label used in 403 detail messages.

### `Perm`

```
class Perm(Requirement)(perm: str) -> None
```

Subject holds the fine-grained permission `perm`.

The permission is stored as `perm` — `code` stays the machine-readable denial code of the atom (`"Perm"`), not the permission name.

#### `Perm.evaluate`

```
async def evaluate(self, ctx: AuthzContext) -> bool
```

Return True iff the subject satisfies this requirement.

Raise `PolicyIndeterminate` (never return `False`) when the requirement cannot be evaluated at all.

#### `Perm.describe`

```
def describe(self) -> str
```

Human-readable label used in 403 detail messages.

### `Permissions`

```
class Permissions
```

Predefined fine-grained permission codes.

### `PolicyIndeterminate`

```
class PolicyIndeterminate(Exception)(decision: AccessDecision) -> None
```

Raised by an atom's `evaluate` when it cannot answer at all.

`evaluate` is boolean, so an atom that cannot evaluate (service absent, fact unregistered, no tenant context) must not answer `False` — a caller composing `not await atom.evaluate(ctx)` would turn the failure into an allow. It raises this instead; `Requirement.decide` (and `enforce`) convert it into the carried indeterminate `AccessDecision`. Never an `HTTPException` — atoms run in workers too.

### `register_authz_projection`

```
def register_authz_projection(
    core: Any,
    *,
    descriptor: Any,
    consumers: list[Any],
    rebuilder: Any,
    rebuild_on_startup: bool = False,
    critical: bool = True,
) -> None
```

Register an authorization projection, its consumers and rebuilder.

`descriptor` is a `QueryProjectionDescriptor`; `consumers` a list of `ProjectionConsumerDescriptor` (one per source channel); `rebuilder` a `ProjectionRebuilder`. When `rebuild_on_startup` is True a rebuild gate is installed (`critical` controls whether its failure aborts startup).

### `register_live_adapter`

```
def register_live_adapter(kind: str, adapter: Any) -> None
```

Bind the adapter the live atoms use for `kind` — CONSUMER-side config.

`kind` is `ENTITLEMENTS` or `MEMBERSHIP`; `adapter` must satisfy `EntitlementsAdapter` / `MembershipAdapter`.

The table is per PROCESS and is read where the atom is EVALUATED — the consumer. Register an adapter from the consumer's composition root or the consuming module's `register(core)`, in every process that evaluates `Credits` / `SubscriptionActive` / `OrgRole`. A provider module that registers an adapter for its OWN contract only affects processes that load it: in a split deployment the consumer process silently keeps the default — so that is logged as a warning.

### `Requirement`

```
class Requirement(ABC)
```

An authorization predicate, composable with `& | ~`.

`evaluate` is the primitive boolean check (it raises `PolicyIndeterminate` when it cannot answer). `decide` wraps it into a structured `AccessDecision` (the engine/enforcement path uses `decide` so a denial carries a machine-readable code/category and the failing requirement); the two never disagree.

#### `Requirement.evaluate`

```
async def evaluate(self, ctx: AuthzContext) -> bool
```

Return True iff the subject satisfies this requirement.

Raise `PolicyIndeterminate` (never return `False`) when the requirement cannot be evaluated at all.

#### `Requirement.decide`

```
async def decide(self, ctx: AuthzContext) -> AccessDecision
```

Structured result. Default: allow iff `evaluate` is True.

#### `Requirement.describe`

```
def describe(self) -> str
```

Human-readable label used in 403 detail messages.

#### `Requirement.mentions_resource`

```
def mentions_resource(self) -> bool
```

True if evaluating this requirement needs the request resource.

Used by `require()` to reject resource-scoped policies at wiring time (they belong in `authorize_resource`/`authorize`), turning a runtime 500 into a clear startup error.

#### `Requirement.org_params`

```
def org_params(self) -> frozenset[str]
```

Path parameters this requirement may read an organisation from.

Non-empty for an atom that resolves an organisation (path parameter, `X-Tenant-Id`, else the bound tenant): `create_app` then refuses a route that solves the gate before its `TenantScope()` (SANDLOOM-B007).

#### `Requirement.contracts`

```
def contracts(self) -> frozenset[str]
```

Sync contracts evaluating this requirement may call (wiring time).

A module's derived `consumes` includes them (optional), so the startup preflight reports an unbound one and `sandloom add deploy` binds it in a split deployment. Default: none (a claims-only atom).

#### `Requirement.risks`

```
def risks(self) -> frozenset[str]
```

The framework risks a gate enforcing this requirement carries (ADR-0009 item 6): `{"admin"}` for an admin role, else none. An `Or` carries its branches' risks; a `Not` grants no authority.

### `resolve_fact`

```
async def resolve_fact(
    fact: AccessFact,
    ctx: AuthzContext,
    *,
    mode: str | None = None,
) -> bool
```

Resolve `fact` to a boolean, caching the result on `ctx`.

`mode` overrides the configured mode (used by tests and to force a path); otherwise the mode comes from settings, clamped to the fact's floor.

### `Role`

```
class Role(Requirement)(role: str) -> None
```

Subject holds the global role `role`.

The role is stored as `role` — `code` stays the machine-readable denial code of the atom (`"Role"`), not the role name.

#### `Role.evaluate`

```
async def evaluate(self, ctx: AuthzContext) -> bool
```

Return True iff the subject satisfies this requirement.

Raise `PolicyIndeterminate` (never return `False`) when the requirement cannot be evaluated at all.

#### `Role.risks`

```
def risks(self) -> frozenset[str]
```

The framework risks a gate enforcing this requirement carries (ADR-0009 item 6): `{"admin"}` for an admin role, else none. An `Or` carries its branches' risks; a `Not` grants no authority.

#### `Role.describe`

```
def describe(self) -> str
```

Human-readable label used in 403 detail messages.

### `Roles`

```
class Roles
```

Predefined global role codes.

### `service_subject`

```
def service_subject(*, roles: frozenset[str] = frozenset({'service'})) -> UserContext
```

A subject representing another service calling across a module boundary.

### `SOURCE_HTTP`

```
SOURCE_HTTP: str = 'http'
```

*(no docstring)*

### `SOURCE_INTERNAL`

```
SOURCE_INTERNAL: str = 'internal'
```

*(no docstring)*

### `SOURCE_SYSTEM`

```
SOURCE_SYSTEM: str = 'system'
```

*(no docstring)*

### `SOURCE_TASK`

```
SOURCE_TASK: str = 'task'
```

*(no docstring)*

### `system_subject`

```
def system_subject(*, roles: frozenset[str] = frozenset({'system'})) -> UserContext
```

A subject representing the platform itself (trusted background work).

### `TIER_CLAIM`

```
TIER_CLAIM: int = 1
```

*(no docstring)*

### `TIER_LIVE`

```
TIER_LIVE: int = 3
```

*(no docstring)*

### `TIER_RESOURCE`

```
TIER_RESOURCE: int = 2
```

*(no docstring)*

### `WithinLimit`

```
class WithinLimit(_AccountClaimAtom)(
    key: str,
    used: int,
    *,
    of: Account = <object object>,
    claim_key: str = 'limits',
) -> None
```

`used` is strictly below the account's plan limit for `key`.

For `Account.user()` the limit comes from the token claim; `used` is the current count the caller already knows (e.g. number of seats). Absent / non-integer limit ⇒ treated as 0 (i.e. denied), so a missing entitlement fails closed.

#### `WithinLimit.describe`

```
def describe(self) -> str
```

Human-readable label used in 403 detail messages.

## Examples and related guides

- [Authorization policies](<https://sandloom.dev/docs/modules/http/authorization/index.md>)
- [Python API overview](<https://sandloom.dev/docs/reference/sdk/index.md>)