Identity verification (optional)
Install doublegate-sdk[identity]. Base SDK/identity imports remain stdlib-only;
verification lazily loads the already-declared PyJWT/cryptography dependencies.
Access tokens
from doublegate_sdk.identity import (
AccessTokenError, IdentityMappingRequest, IssuerProfile,
ResolvedIdentity, TokenClaims, verify_access_token,
)
# These values come from owner-approved enrollment, NEVER the incoming token.
profile = IssuerProfile(
issuer="https://issuer.example",
tenant_id="tenant",
audience="organization",
tenant_claim="tid", # illustrative enrolled claim, not an OIDC convention
mapping_revision="map-1",
token_kind="direct", # or explicitly enrolled "app_only"
)
# verified = verify_access_token(
# incoming_token, issuer_profile=profile,
# jwks=trusted_current_jwks, resolver=trusted_identity_resolver,
# now=receiver_unix_seconds,
# )
verify_access_token(token, *, issuer_profile, jwks, resolver, now) -> TokenClaims
raises AccessTokenError on verification or mapping failure. All profile, mapping
request, resolved identity and result objects are frozen dataclasses.
The trusted resolver receives IdentityMappingRequest(issuer, tenant_id, subject,
client_id, mapping_revision, token_kind) only after signature and claim checks.
It must resolve that exact tuple against enrolled, current authority, including
whether that subject/client pair is permitted for the selected direct/app-only
interpretation, and return ResolvedIdentity(requester_id, requester_kind,
actor_id, actor_kind, client_principal_id). Unknown mappings must fail, not create
accounts or merge identities by email. The SDK cannot establish that a supplied
resolver or profile was enrolled. That trust boundary belongs to the receiving gate.
First bounded profile
Implements design 10 §§2–3's selected RFC 9068 RS256 fixture, not arbitrary customer
OIDC interoperability:
- Protected header is exactly
{alg, typ, kid}, with alg="RS256",
typ="at+jwt" and nonempty kid. All other headers reject, including jku,
x5u, jwk, crit and unencoded/detached payload options. Three nonempty,
canonical unpadded base64url compact segments are required.
- JSON objects reject duplicate keys, including nested duplicates, and non-JSON
constants. Required identity claims must be nonempty strings. Unknown access
claims may parse, but are neither passed to the resolver nor returned as authority.
- Required claims:
iss, sub, aud, exp, iat, jti, client_id and the
explicitly enrolled tenant claim. Issuer/tenant matching is exact, case-sensitive
and unnormalized. This bounded profile accepts only one exact string audience;
audience arrays, even a singleton, reject. Use a separately qualified adapter
rather than silently broadening the destination contract.
exp, iat, optional nbf, and receiver now are nonnegative integer Unix
seconds (not booleans, floats, strings or null). Require 0 < exp-iat <= 300,
iat <= now+30, nbf <= now+30, exp > now-30, and nbf < exp when present.
Tolerance does not extend the maximum issued lifetime.
- Keys come exclusively from the caller-supplied JWKS snapshot. Exactly one key
must match
kid; missing or duplicate matches reject with no fallback. Selected
keys must be public RSA keys, with alg=RS256, use=sig, key_ops=["verify"]
if those optional metadata fields are present. Private parameters reject.
PyJWT supplies key parsing, signature verification and issuer/audience checks;
there is no bespoke cryptography.
- Direct mapping requires requester=actor with equal
human or workload kind.
Presenting client_id resolves separately; it is not assumed to equal sub.
App-only additionally requires a workload requester/actor equal to the resolved
presenting client principal. Mapping an app-only token to a human rejects.
- Delegated
act claims reject (including null). The design does not qualify an
actual exchange/OBO issuer or prescribe a customer token-kind discriminator.
No OBO issuer, actor-claim guessing, exchange, or universal tid mapping is invented.
A deployment must only select a profile through trusted enrollment, never from
token hints; the resolver must validate the subject/client/token-kind binding.
JWKS acquisition, bounded refresh on unknown kid, size/time/redirect limits,
rotation overlap deadlines and freshness checks remain caller-owned. The SDK never
fetches a URL, starts a service, caches keys or trusts token-directed networking.
Callers can supply an updated JWKS and retry; no automatic refresh occurs.
Verification is not authorization
TokenClaims contains verified external identifiers, token time/id/destination,
mapping revision and resolved requester/actor/client identities. It has no
authorized flag or raw claim bag. Token groups, roles, scope, home-team or
grant claims confer no team membership or permission here. No enrolled
membership authority adapter is implemented in this slice.
Gate integration remains pending: current membership, applicable grant binding
(including jti/subject/actor/client), every parent grant and revocation, action and
team permission, freshness, human-action proof, replay handling and transactional
rechecks before queue execution/admission/publication must occur outside this SDK.
Token validity is not an admission receipt, grant, durable permission, or evidence
of human presence. There are no gate or admission-status side effects and no
changes to the single submission format.
Attribution
The separate verify_attribution API continues to verify enrolled Ed25519
attribution signatures and request bindings. Its signed historical evidence is
not a replacement for a fresh access token or current gate authority.
Signed reference payload (attribution v1, submission event)
The only accepted signed payload has identity_refs with exactly membership,
mapping, relationship, policy, and attester_enrollment. Each is a closed
{revision, digest} object: an independent nonempty string revision and lowercase
SHA-256 digest of the exact retained record. authority_ref is a closed
{revision, digest} object whose revision is a positive safe integer.
grant_refs is a closed object with required requester and actor arrays;
each contains 1–8 root-to-leaf {grant_id, grant_digest} records, with distinct
IDs within that chain. Both chains remain required when the identities coincide.
The old scalar revisions and flat grant list are rejected, not upgraded or
retried under a second format. Access-token mapping_revision above is a
separate enrolled resolver input, not a signed attribution payload field.
The two phases may name different actors, record revisions and attesters.
Gate-owned exact retained lookup and historical/current authorization remain
necessary; signature success alone is not admission or a durable receipt.
doublegate_sdk.identity
ADR-0064 attribution parsing and optional Ed25519 JWS verification.
validate_attribution and decode_attribution check wire data only; parsing
NEVER authenticates a signer. verify_attribution(jws, *, signer_key: PublicKey,
expected_binding: AttributionBinding) -> AttributionV1 requires the optional
doublegate-sdk[identity] extra, a caller-supplied enrolled public key and
trusted request bindings. It checks the signature, bindings and a 30-second
future-issuance tolerance against the receiver's supplied time.
A valid signature is NOT current authority. No enrollment, membership/grant
lookup, revocation, replay prevention, network access or gate effects occur here.
Gate callers must check attestation scope and current authority before effects.
AttributionError
Bases: ValueError
Malformed attribution wire data (not an authentication decision).
Source code in src/doublegate_sdk/identity.py
| class AttributionError(ValueError):
"""Malformed attribution wire data (not an authentication decision)."""
|
AttributionV1
dataclass
Immutable canonical payload bytes; untrusted even after validation.
Source code in src/doublegate_sdk/identity.py
| @dataclass(frozen=True, slots=True)
class AttributionV1:
"""Immutable canonical payload bytes; untrusted even after validation."""
_canonical: bytes
def to_dict(self) -> dict[str, Any]:
return json.loads(self._canonical)
def canonical_bytes(self) -> bytes:
return self._canonical
|
to_dict
to_dict() -> dict[str, Any]
Source code in src/doublegate_sdk/identity.py
| def to_dict(self) -> dict[str, Any]:
return json.loads(self._canonical)
|
canonical_bytes
canonical_bytes() -> bytes
Source code in src/doublegate_sdk/identity.py
| def canonical_bytes(self) -> bytes:
return self._canonical
|
PublicKey
dataclass
Caller-supplied enrolled attribution key, NOT evidence of enrollment.
Gate-owned trusted configuration must bind this kid and deployment to the raw
32-byte Ed25519 public key. Never construct it from untrusted statement claims.
No private keys, JWKS lookup, TOFU or enrollment occurs in the SDK.
Source code in src/doublegate_sdk/identity.py
| @dataclass(frozen=True, slots=True)
class PublicKey:
"""Caller-supplied enrolled attribution key, NOT evidence of enrollment.
Gate-owned trusted configuration must bind this kid and deployment to the raw
32-byte Ed25519 public key. Never construct it from untrusted statement claims.
No private keys, JWKS lookup, TOFU or enrollment occurs in the SDK.
"""
kid: str
issuer_deployment_id: str
ed25519_bytes: bytes
|
issuer_deployment_id
instance-attribute
issuer_deployment_id: str
ed25519_bytes
instance-attribute
AttributionBinding
dataclass
Expected gate-owned request binding, not authority or token claims.
Supply digests from the actual envelope/retained evidence manifest. For a
submission, supply the digest of the exact production JWS as parent. now
is receiver time in Unix seconds, not the signer's claimed observation time.
Source code in src/doublegate_sdk/identity.py
| @dataclass(frozen=True, slots=True)
class AttributionBinding:
"""Expected gate-owned request binding, not authority or token claims.
Supply digests from the actual envelope/retained evidence manifest. For a
submission, supply the digest of the exact production JWS as parent. ``now``
is receiver time in Unix seconds, not the signer's claimed observation time.
"""
tenant_id: str
operation_id: str
envelope_digest: str
evidence_digest: str
requester_id: str
actor_id: str
contributor_id: str
home_team_id: str
audience: str
phase: str
now: int
parent_statement_digest: str | None = None
|
tenant_id
instance-attribute
operation_id
instance-attribute
envelope_digest
instance-attribute
evidence_digest
instance-attribute
requester_id
instance-attribute
actor_id
instance-attribute
contributor_id
instance-attribute
home_team_id
instance-attribute
audience
instance-attribute
parent_statement_digest
class-attribute
instance-attribute
parent_statement_digest: str | None = None
AccessTokenError
Bases: ValueError
Access-token verification or trusted identity mapping failed closed.
Source code in src/doublegate_sdk/identity.py
| class AccessTokenError(ValueError):
"""Access-token verification or trusted identity mapping failed closed."""
|
IssuerProfile
dataclass
Owner-enrolled RFC 9068 profile, never constructed from a token.
Exact tenant claim name and direct/app_only interpretation are enrollment
inputs, not universal OIDC conventions. 300s TTL and 30s tolerance are fixed
for this bounded profile. Two algorithms, chosen at enrollment and never by
the token: RS256 for an external issuer, EdDSA (Ed25519, RFC 8037)
for a local issuer — an operator or agent key that mints its own per-gate
assertions (ADR-0080 d1, d4, d5). Delegation/OBO is deliberately
unsupported. No membership authority is implemented here; token groups
confer nothing.
Source code in src/doublegate_sdk/identity.py
| @dataclass(frozen=True, slots=True)
class IssuerProfile:
"""Owner-enrolled RFC 9068 profile, never constructed from a token.
Exact tenant claim name and direct/app_only interpretation are enrollment
inputs, not universal OIDC conventions. 300s TTL and 30s tolerance are fixed
for this bounded profile. Two algorithms, chosen at enrollment and never by
the token: ``RS256`` for an external issuer, ``EdDSA`` (Ed25519, RFC 8037)
for a *local* issuer — an operator or agent key that mints its own per-gate
assertions (ADR-0080 d1, d4, d5). Delegation/OBO is deliberately
unsupported. No membership authority is implemented here; token groups
confer nothing.
"""
issuer: str
tenant_id: str
audience: str
tenant_claim: str
mapping_revision: str
token_kind: str
algorithm: str = 'RS256'
def __post_init__(self) -> None:
for value in (self.issuer, self.tenant_id, self.audience, self.tenant_claim,
self.mapping_revision, self.token_kind, self.algorithm):
_access_identifier(value)
if self.token_kind not in ('direct', 'app_only'):
raise AccessTokenError('only direct and app_only profiles are supported')
if self.algorithm not in ('RS256', 'EdDSA'):
raise AccessTokenError('only RS256 (external) and EdDSA (local key) profiles are supported')
if self.tenant_claim in {'iss', 'sub', 'aud', 'exp', 'iat', 'nbf', 'jti',
'client_id', 'act', 'scope', 'groups', 'roles'}:
raise AccessTokenError('tenant mapping must use a distinct enrolled claim')
|
issuer
instance-attribute
tenant_id
instance-attribute
audience
instance-attribute
tenant_claim
instance-attribute
mapping_revision
instance-attribute
token_kind
instance-attribute
algorithm
class-attribute
instance-attribute
IdentityMappingRequest
dataclass
Verified external identifiers only, for an enrolled trusted resolver.
Source code in src/doublegate_sdk/identity.py
| @dataclass(frozen=True, slots=True)
class IdentityMappingRequest:
"""Verified external identifiers only, for an enrolled trusted resolver."""
issuer: str
tenant_id: str
subject: str
client_id: str
mapping_revision: str
token_kind: str
|
issuer
instance-attribute
tenant_id
instance-attribute
subject
instance-attribute
client_id
instance-attribute
mapping_revision
instance-attribute
token_kind
instance-attribute
ResolvedIdentity
dataclass
Trusted mapping output, NOT membership, a grant or proof of human presence.
Source code in src/doublegate_sdk/identity.py
| @dataclass(frozen=True, slots=True)
class ResolvedIdentity:
"""Trusted mapping output, NOT membership, a grant or proof of human presence."""
requester_id: str
requester_kind: str
actor_id: str
actor_kind: str
client_principal_id: str
|
requester_id
instance-attribute
requester_kind
instance-attribute
actor_id
instance-attribute
actor_kind
instance-attribute
client_principal_id
instance-attribute
TokenClaims
dataclass
Cryptographically valid hop identity, NEVER an authorization capability.
No raw groups, roles, scope, grant or team claims are promoted to authority.
Gates still own current grants, membership, revocation and human-action proof.
Source code in src/doublegate_sdk/identity.py
| @dataclass(frozen=True, slots=True)
class TokenClaims:
"""Cryptographically valid hop identity, NEVER an authorization capability.
No raw groups, roles, scope, grant or team claims are promoted to authority.
Gates still own current grants, membership, revocation and human-action proof.
"""
issuer: str
tenant_id: str
subject: str
client_id: str
audience: str
token_id: str
issued_at: int
expires_at: int
not_before: int | None
mapping_revision: str
token_kind: str
requester_id: str
requester_kind: str
actor_id: str
actor_kind: str
client_principal_id: str
|
issuer
instance-attribute
tenant_id
instance-attribute
subject
instance-attribute
client_id
instance-attribute
audience
instance-attribute
token_id
instance-attribute
issued_at
instance-attribute
expires_at
instance-attribute
not_before
instance-attribute
mapping_revision
instance-attribute
token_kind
instance-attribute
requester_id
instance-attribute
requester_kind
instance-attribute
actor_id
instance-attribute
actor_kind
instance-attribute
client_principal_id
instance-attribute
validate_attribution
validate_attribution(payload: Mapping[str, Any]) -> AttributionV1
Validate closed shape and pure invariants, NEVER identity or authority.
Source code in src/doublegate_sdk/identity.py
| def validate_attribution(payload: Mapping[str, Any]) -> AttributionV1:
"""Validate closed shape and pure invariants, NEVER identity or authority."""
_closed(payload, _REQUIRED, _OPTIONAL_IDS | {'parent_statement_digest'})
if type(payload['attribution_version']) is not int or payload['attribution_version'] != 1:
raise AttributionError('attribution_version must be integer 1')
if payload['phase'] not in ('production', 'submission'):
raise AttributionError('phase must be production or submission')
for name in _IDS | (_OPTIONAL_IDS & payload.keys()):
_identifier(payload[name])
for name in ('envelope_digest', 'evidence_digest'):
_digest(payload[name])
if payload['phase'] == 'submission':
_digest(payload.get('parent_statement_digest'))
elif 'parent_statement_digest' in payload:
raise AttributionError('production forbids a parent')
for name in ('issued_at', 'membership_observed_at'):
value = payload[name]
if type(value) is not int or not 0 <= value <= 9007199254740991:
raise AttributionError(f'{name} must be a safe nonnegative integer')
if payload['membership_observed_at'] > payload['issued_at']:
raise AttributionError('membership observation follows issuance')
_closed(payload['identity_refs'], _IDENTITY_KINDS)
for ref in payload['identity_refs'].values():
_closed(ref, {'revision', 'digest'})
_identifier(ref['revision'])
_digest(ref['digest'])
authority = payload['authority_ref']
_closed(authority, {'revision', 'digest'})
if type(authority['revision']) is not int or not 1 <= authority['revision'] <= 9007199254740991:
raise AttributionError('authority revision must be a positive safe integer')
_digest(authority['digest'])
_closed(payload['grant_refs'], {'requester', 'actor'})
for refs in payload['grant_refs'].values():
if not isinstance(refs, list) or not 1 <= len(refs) <= 8:
raise AttributionError('each role grant_refs must contain 1–8 records')
seen = set()
for ref in refs:
_closed(ref, {'grant_id', 'grant_digest'})
_identifier(ref['grant_id'])
_digest(ref['grant_digest'])
if ref['grant_id'] in seen:
raise AttributionError('duplicate grant_id')
seen.add(ref['grant_id'])
return AttributionV1(canonical_json(dict(payload)))
|
verify_attribution
verify_attribution(jws: str, *, signer_key: PublicKey, expected_binding: AttributionBinding) -> AttributionV1
Verify Ed25519 JWS cryptography; the returned payload is NOT authorization.
Requires doublegate-sdk[identity]. No network, membership/grant lookup,
revocation check, replay journal, evidence retrieval or mutation takes place.
Gates must verify attestation scope and current authority before any effects.
Source code in src/doublegate_sdk/identity.py
| def verify_attribution(jws: str, *, signer_key: PublicKey,
expected_binding: AttributionBinding) -> AttributionV1:
"""Verify Ed25519 JWS cryptography; the returned payload is NOT authorization.
Requires ``doublegate-sdk[identity]``. No network, membership/grant lookup,
revocation check, replay journal, evidence retrieval or mutation takes place.
Gates must verify attestation scope and current authority before any effects.
"""
from doublegate_sdk.identity_wire import decode_attribution_jws
try:
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey
from jwt import api_jws
from jwt.exceptions import PyJWTError
except ImportError as exc:
raise AttributionError('verification requires doublegate-sdk[identity]') from exc
parsed = decode_attribution_jws(jws)
claims = parsed.payload.to_dict()
now = expected_binding.now
if type(now) is not int or not 0 <= now <= 9007199254740991:
raise AttributionError('receiver time must be a safe nonnegative integer')
if claims['issued_at'] > now + 30:
raise AttributionError('attribution observation is in the future')
if parsed.header['kid'] != signer_key.kid:
raise AttributionError('kid does not match enrolled key')
if claims['issuer_deployment_id'] != signer_key.issuer_deployment_id:
raise AttributionError('issuer_deployment_id does not match enrolled key')
for field in ('tenant_id', 'operation_id', 'envelope_digest', 'evidence_digest',
'requester_id', 'actor_id', 'contributor_id', 'home_team_id',
'audience', 'phase', 'parent_statement_digest'):
if claims.get(field) != getattr(expected_binding, field):
raise AttributionError(f'{field} does not match expected binding')
try:
key = Ed25519PublicKey.from_public_bytes(signer_key.ed25519_bytes)
api_jws.decode_complete(jws, key=key, algorithms=['EdDSA'])
except (PyJWTError, ValueError, TypeError) as exc:
raise AttributionError('invalid attribution signature or public key') from exc
return parsed.payload
|
decode_attribution
decode_attribution(raw: bytes) -> AttributionV1
Decode an exact canonical payload, without authenticating it.
Source code in src/doublegate_sdk/identity.py
| def decode_attribution(raw: bytes) -> AttributionV1:
"""Decode an exact canonical payload, without authenticating it."""
result = validate_attribution(_strict_json(raw))
if result.canonical_bytes() != raw:
raise AttributionError('payload is not canonical JSON')
return result
|
verify_access_token
verify_access_token(token: str, *, issuer_profile: IssuerProfile, jwks: Mapping[str, Any], resolver: Callable[[IdentityMappingRequest], ResolvedIdentity], now: int) -> TokenClaims
Verify with caller-supplied enrolled JWKS and a trusted identity resolver.
Requires the optional identity extra. Does not discover issuers, fetch keys,
refresh unknown kids, enroll identities or perform gate/admission effects.
The caller owns bounded JWKS refresh/rotation and mapping freshness. Resolver
input is signature-verified identifiers, not token-directed lookup locations.
Source code in src/doublegate_sdk/identity.py
| def verify_access_token(
token: str, *, issuer_profile: IssuerProfile, jwks: Mapping[str, Any],
resolver: Callable[[IdentityMappingRequest], ResolvedIdentity], now: int,
) -> TokenClaims:
"""Verify with caller-supplied enrolled JWKS and a trusted identity resolver.
Requires the optional identity extra. Does not discover issuers, fetch keys,
refresh unknown kids, enroll identities or perform gate/admission effects.
The caller owns bounded JWKS refresh/rotation and mapping freshness. Resolver
input is signature-verified identifiers, not token-directed lookup locations.
"""
try:
import jwt
from jwt.exceptions import PyJWTError
except ImportError as exc:
raise AccessTokenError('verification requires doublegate-sdk[identity]') from exc
p = issuer_profile
if not isinstance(p, IssuerProfile):
raise AccessTokenError('an enrolled IssuerProfile is required')
if type(now) is not int or now < 0:
raise AccessTokenError('receiver time must be nonnegative integer seconds')
try:
from jwt.utils import base64url_decode, base64url_encode
if not isinstance(token, str) or len(token.split('.')) != 3:
raise AccessTokenError('compact JWT must contain three segments')
parts = []
for segment in token.split('.'):
if not segment or any(c not in
'ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789-_'
for c in segment):
raise AccessTokenError('JWT segments must be unpadded base64url')
raw = base64url_decode(segment)
if base64url_encode(raw).decode('ascii') != segment:
raise AccessTokenError('invalid base64url encoding')
parts.append(raw)
header = _access_json(parts[0])
claims = _access_json(parts[1])
if set(header) != {'alg', 'typ', 'kid'} or (
header['alg'] != p.algorithm or header['typ'] != 'at+jwt'):
raise AccessTokenError('unsupported access-token JOSE profile')
_access_identifier(header['kid'])
if not isinstance(jwks, Mapping) or not isinstance(jwks.get('keys'), list):
raise AccessTokenError('JWKS must contain a keys list')
if any(not isinstance(key, Mapping) for key in jwks['keys']):
raise AccessTokenError('invalid JWKS key record')
matches = [key for key in jwks['keys'] if key.get('kid') == header['kid']]
if len(matches) != 1:
raise AccessTokenError('unknown or ambiguous kid')
enrolled = dict(matches[0])
public_shape = ((enrolled.get('kty') == 'RSA') if p.algorithm == 'RS256'
else (enrolled.get('kty') == 'OKP' and enrolled.get('crv') == 'Ed25519'))
if (not public_shape or enrolled.get('alg', p.algorithm) != p.algorithm
or enrolled.get('use', 'sig') != 'sig'
or enrolled.get('key_ops', ['verify']) != ['verify']
or {'d', 'p', 'q', 'dp', 'dq', 'qi', 'oth'} & enrolled.keys()):
raise AccessTokenError(f'enrolled key must be a public {p.algorithm} signing key')
key = jwt.PyJWK.from_dict(enrolled, algorithm=p.algorithm)
# PyJWT performs crypto and exact issuer/audience/required-claim checks.
# Time is checked below against caller-supplied receiver time, not a
# hidden wall clock. Strict JSON above precedes PyJWT's permissive parser.
jwt.decode(token, key=key.key, algorithms=[p.algorithm], issuer=p.issuer,
audience=p.audience, options={
'require': ['iss', 'sub', 'aud', 'exp', 'iat', 'jti', 'client_id'],
'verify_exp': False, 'verify_iat': False, 'verify_nbf': False,
'strict_aud': True})
except (PyJWTError, ValueError, TypeError, KeyError, OverflowError) as exc:
raise AccessTokenError('invalid access token or enrolled key') from exc
for name in ('sub', 'client_id', 'jti', p.tenant_claim):
_access_identifier(claims.get(name))
if claims[p.tenant_claim] != p.tenant_id:
raise AccessTokenError('token tenant conflicts with enrolled tenant')
if 'act' in claims:
raise AccessTokenError('delegation requires a separately qualified issuer profile')
for name in ('exp', 'iat', 'nbf'):
if name in claims and (type(claims[name]) is not int or claims[name] < 0):
raise AccessTokenError('token times must be nonnegative integer seconds')
issued, expires = claims['iat'], claims['exp']
if not 0 < expires - issued <= 300:
raise AccessTokenError('access-token lifetime must be 1–300 seconds')
if issued > now + 30 or claims.get('nbf', 0) > now + 30:
raise AccessTokenError('access token is not yet valid')
if expires <= now - 30:
raise AccessTokenError('access token expired')
if 'nbf' in claims and claims['nbf'] >= expires:
raise AccessTokenError('not-before must precede expiry')
try:
mapped = resolver(IdentityMappingRequest(p.issuer, p.tenant_id, claims['sub'],
claims['client_id'], p.mapping_revision,
p.token_kind))
except Exception as exc:
raise AccessTokenError('trusted identity mapping unavailable') from exc
if not isinstance(mapped, ResolvedIdentity):
raise AccessTokenError('identity mapping unresolved')
for value in (mapped.requester_id, mapped.actor_id, mapped.client_principal_id):
_access_identifier(value)
if (mapped.requester_kind not in ('human', 'workload')
or mapped.actor_kind != mapped.requester_kind
or mapped.actor_id != mapped.requester_id):
raise AccessTokenError('direct/app-only mapping must have requester=actor')
if p.token_kind == 'app_only' and (mapped.requester_kind != 'workload'
or mapped.client_principal_id != mapped.actor_id):
raise AccessTokenError('app-only mapping must identify the presenting workload')
return TokenClaims(p.issuer, p.tenant_id, claims['sub'], claims['client_id'],
p.audience, claims['jti'], claims['iat'], claims['exp'],
claims.get('nbf'), p.mapping_revision, p.token_kind,
mapped.requester_id, mapped.requester_kind, mapped.actor_id,
mapped.actor_kind, mapped.client_principal_id)
|