loom.core.identity¶

Caller identity: the verified answer to “who is running this?”.

Exposes the immutable Identity value object, the explicit ANONYMOUS absence of one, and the context guard that propagates it across the async stack. This package belongs to the domain layer: it imports no transport and no infrastructure.

class loom.core.identity.Identity(subject, roles=(), attributes=<factory>, mechanism='')[source]¶

Bases: object

Verified caller of a single execution.

Instances are immutable and safe to share across an async context. Only the transport layer builds them, from credentials it has already verified; every other layer consumes them read-only.

Parameters:

subject (str)
roles (tuple[str, ...])
attributes (Mapping[str, str])
mechanism (str)

subject¶

Stable identifier of the caller. Empty means anonymous.

Type:: str

roles¶

Roles the caller holds, in the order the mechanism reported them and without duplicates.

Type:: tuple[str, …]

attributes¶

Verified string-valued facts about the caller (e-mail, tenant, department, …). Copied on construction, so a later mutation of the source mapping cannot rewrite the identity.

Type:: collections.abc.Mapping[str, str]

mechanism¶

Label of the mechanism that authenticated the caller (e.g. "jwt"). Used for the audit trail, never for authorization.

Type:: str

Example:

identity = Identity(
    subject="user-1",
    roles=("role_viz_reader",),
    attributes={"email": "ada@example.com"},
    mechanism="jwt",
)
if identity.has_role("role_viz_reader"):
    ...

property is_authenticated: bool¶

Whether the caller was identified by an authentication mechanism.

Returns:: True when a non-empty subject is present.

has_role(role)[source]¶

Report whether the caller holds role.

Matching is exact: no case folding and no prefix matching, so a role name can never be widened by accident.

Parameters:: role (str) – Role name to look for.
Returns:: True when the caller holds exactly that role.
Return type:: bool

attribute(name)[source]¶

Return the verified attribute name, or None when absent.

Parameters:: name (str) – Attribute key as published by the authenticator.
Returns:: The attribute value, or None when the caller does not carry it.
Return type:: str | None

require_subject()[source]¶

Return the subject, refusing anonymous callers.

Returns:: The caller subject.
Raises:: Unauthenticated – When the identity is anonymous.
Return type:: str

require_attribute(name)[source]¶

Return a mandatory verified attribute.

The two failure modes are distinct on purpose: an anonymous caller can fix the request by authenticating (401), while an authenticated caller missing the attribute cannot (403).

Parameters:

name (str) – Attribute key the caller must carry.

Returns:

The attribute value.

Raises:

Unauthenticated – When the identity is anonymous.
Forbidden – When the authenticated caller does not carry name.

Return type:

str

class loom.core.identity.IssuedToken(token, expires_at, jti)[source]¶

Bases: object

A minted credential together with what the issuer already knew about it.

Returned instead of a bare string so a login endpoint never has to decode the token it just signed: the expiry the HTTP response advertises and the identifier the audit trail records are both produced by the signing step.

Parameters:

token (str)
expires_at (datetime)
jti (str)

token¶

The encoded credential, ready to travel as a bearer token.

Type:: str

expires_at¶

Instant the credential stops being valid, timezone-aware.

Type:: datetime.datetime

jti¶

Unique identifier of this minting, for correlation and audit.

Type:: str

class loom.core.identity.TokenIssuer(*args, **kwargs)[source]¶

Bases: Protocol

Mints a credential that an Authenticator can later verify.

Implementations guarantee the round trip: whatever the identity carries — subject, roles and attributes — a matching authenticator recovers intact. Everything the credential says comes from the identity, so no caller can smuggle a claim past it.

Example:

issued = issuer.issue(identity)
response = {"access_token": issued.token, "expires_at": issued.expires_at}

issue(identity, *, ttl=None)[source]¶

Mint a credential for identity.

Parameters:

identity (Identity) – Verified caller the credential speaks for.
ttl (timedelta | None) – Lifetime override. None uses the configured one, which is also the ceiling: a longer lifetime is refused.

Returns:

The minted credential and its metadata.

Raises:

ValueError – If the identity cannot be represented — anonymous, or carrying an attribute that would be unreadable once encoded — or if ttl exceeds the configured lifetime.
RuntimeError – If the credential cannot be produced. Implementations keep the cause out of the traceback: it can carry key material.

Return type:

IssuedToken

loom.core.identity.current_identity()[source]¶

Return the identity active in the current async context.

Never returns None: contexts with no authenticated caller yield ANONYMOUS, so consumers cannot accidentally treat “unknown” as “authorized”.

Returns:: The active identity, or the anonymous one.
Return type:: Identity

loom.core.identity.reset_identity(token)[source]¶

Restore the identity active before the matching set_identity().

Parameters:: token (Token) – Token returned by the corresponding set_identity() call.
Return type:: None

loom.core.identity.set_identity(identity)[source]¶

Install identity for the current async context.

Parameters:: identity (Identity) – Verified identity produced by an authentication mechanism.
Returns:: A Token that must be passed to reset_identity() in a finally block.
Return type:: Token

Example:

token = set_identity(identity)
try:
    await handle_request()
finally:
    reset_identity(token)