"""Reference solution — the agent and workload identity fabric. The hardest unsolved problem in the JD, and the one where a bank's risk concentrates. Human identity is solved. Service identity is solved. AGENT identity is neither, because an agent acts on behalf of a user, over multiple hops, with dynamically discovered tools, sometimes delegating to other agents. Four properties, and every one of them is a control an examiner will ask you to demonstrate: credentials are DERIVED (from a verified assertion, not asserted), NARROWED (at every hop), CHAINED (the actor list is append-only), and SHORT-LIVED (seconds to minutes, never rotated-in-place). Deterministic: HMAC signing, an injected clock, counter-derived identifiers. No wall clock, no randomness. ``python solution.py`` runs the worked example. """ from __future__ import annotations import base64 import hashlib import hmac import json from dataclasses import dataclass, field, replace from enum import Enum from typing import Callable, Dict, Iterable, List, Mapping, Optional, Sequence, Set, Tuple # ====================================================================================== # 1. JOSE: base64url, JWS, and the claim set # ====================================================================================== def b64url_encode(data: bytes) -> str: """Base64url WITHOUT padding, as JOSE requires.""" return base64.urlsafe_b64encode(data).decode("ascii").rstrip("=") def b64url_decode(text: str) -> bytes: """Restore the padding the encoder stripped. Getting this wrong produces a binascii error on exactly 2/3 of otherwise-valid tokens, which is a memorable afternoon.""" padding = "=" * (-len(text) % 4) return base64.urlsafe_b64decode(text + padding) class TokenError(Exception): """Verification failed. The message is for the LOG; the caller gets a generic 401. A verifier that tells an attacker *which* check failed is an oracle: try an expired token, try a wrong audience, and you have mapped the trust boundaries. """ def __init__(self, reason: str, *, code: str = "invalid_token") -> None: super().__init__(reason) self.reason = reason self.code = code @dataclass(frozen=True) class Actor: """One link in the delegation chain — RFC 8693's ``act`` claim. Nested: the innermost actor is the most recent. That ordering is fixed by the RFC and it is the opposite of what most people assume, which is worth checking before you read a chain in an incident. """ sub: str kind: str = "agent" # "agent" | "service" | "workload" @dataclass(frozen=True) class Claims: """The claim set. Every field here is checked by somebody, and the fields nobody checks are the vulnerabilities.""" iss: str # who minted it sub: str # the principal it is about aud: str # who it is FOR — the confused-deputy defence exp: float # expiry iat: float # issued at nbf: float = 0.0 # not before jti: str = "" # unique id, for replay detection scope: Tuple[str, ...] = () # what may be done tenant: str = "" act: Tuple[Actor, ...] = () # the delegation chain, innermost = latest cnf: Optional[str] = None # confirmation: a key thumbprint (RFC 7800) client_id: str = "" may_delegate: bool = False # may this credential be exchanged onward? def to_payload(self) -> Dict[str, object]: payload: Dict[str, object] = { "iss": self.iss, "sub": self.sub, "aud": self.aud, "exp": self.exp, "iat": self.iat, } if self.nbf: payload["nbf"] = self.nbf if self.jti: payload["jti"] = self.jti if self.scope: payload["scope"] = " ".join(self.scope) # RFC 6749: space-delimited if self.tenant: payload["tenant"] = self.tenant if self.client_id: payload["client_id"] = self.client_id if self.may_delegate: payload["may_delegate"] = True if self.cnf: payload["cnf"] = {"jkt": self.cnf} # RFC 7800 / RFC 9449 shape if self.act: payload["act"] = _nest_actors(self.act) return payload @classmethod def from_payload(cls, payload: Mapping[str, object]) -> "Claims": cnf = payload.get("cnf") return cls( iss=str(payload["iss"]), sub=str(payload["sub"]), aud=str(payload["aud"]), exp=float(payload["exp"]), iat=float(payload["iat"]), nbf=float(payload.get("nbf", 0.0)), jti=str(payload.get("jti", "")), scope=tuple(str(payload.get("scope", "")).split()) if payload.get("scope") else (), tenant=str(payload.get("tenant", "")), client_id=str(payload.get("client_id", "")), may_delegate=bool(payload.get("may_delegate", False)), cnf=str(cnf["jkt"]) if isinstance(cnf, dict) and "jkt" in cnf else None, act=_flatten_actors(payload.get("act")), ) def _nest_actors(actors: Sequence[Actor]) -> Dict[str, object]: """RFC 8693 nests ``act`` claims: the outermost object is the MOST RECENT actor, and each ``act`` inside it is the one before.""" node: Dict[str, object] = {} for actor in actors: # actors[0] is the earliest node = {"sub": actor.sub, "kind": actor.kind, **({"act": node} if node else {})} return node def _flatten_actors(value: object) -> Tuple[Actor, ...]: """Inverse of :func:`_nest_actors`: earliest first.""" chain: List[Actor] = [] node = value while isinstance(node, dict) and "sub" in node: chain.append(Actor(str(node["sub"]), str(node.get("kind", "agent")))) node = node.get("act") chain.reverse() return tuple(chain) class Signer: """HMAC-SHA256 JWS. Real deployments use asymmetric signing (RS256/ES256) so a resource server can verify without holding a minting key. HMAC keeps the lab dependency-free; every verification rule below is identical either way. """ def __init__(self, key_id: str, secret: bytes, *, algorithm: str = "HS256") -> None: self.key_id = key_id self.secret = secret self.algorithm = algorithm def sign(self, claims: Claims) -> str: header = {"alg": self.algorithm, "typ": "at+jwt", "kid": self.key_id} signing_input = (_json_b64(header) + "." + _json_b64(claims.to_payload())) signature = hmac.new(self.secret, signing_input.encode("ascii"), hashlib.sha256).digest() return signing_input + "." + b64url_encode(signature) def verify_signature(self, token: str) -> Tuple[Dict[str, object], Dict[str, object]]: parts = token.split(".") if len(parts) != 3: raise TokenError("malformed token") header_b64, payload_b64, signature_b64 = parts try: header = json.loads(b64url_decode(header_b64)) payload = json.loads(b64url_decode(payload_b64)) except Exception: raise TokenError("undecodable token") from None # Never take the algorithm from the token without checking it against what we # expect. "alg": "none" and algorithm-confusion are the two classic JWT breaks. if header.get("alg") != self.algorithm: raise TokenError(f"unexpected alg {header.get('alg')!r}", code="invalid_alg") expected = hmac.new(self.secret, (header_b64 + "." + payload_b64).encode("ascii"), hashlib.sha256).digest() # Constant-time compare: a short-circuiting == leaks the signature byte by byte. if not hmac.compare_digest(expected, b64url_decode(signature_b64)): raise TokenError("bad signature", code="invalid_signature") return header, payload def _json_b64(value: Mapping[str, object]) -> str: """Canonical JSON — sorted keys, no spaces — so the same claims always produce the same token. Not required by JOSE; required by a deterministic test.""" return b64url_encode(json.dumps(value, sort_keys=True, separators=(",", ":")).encode()) # ====================================================================================== # 2. Verification — the checks, and the attack each one stops # ====================================================================================== MAX_CHAIN_DEPTH = 4 @dataclass(frozen=True) class VerificationPolicy: """What THIS resource server requires.""" audience: str trusted_issuers: Tuple[str, ...] clock_skew_seconds: float = 30.0 required_scopes: Tuple[str, ...] = () require_proof_of_possession: bool = False max_chain_depth: int = MAX_CHAIN_DEPTH max_lifetime_seconds: Optional[float] = None class ReplayCache: """Seen ``jti`` values, with expiry. One-time tokens are only one-time if somebody remembers.""" def __init__(self, *, now: Callable[[], float]) -> None: self.now = now self._seen: Dict[str, float] = {} def check_and_record(self, jti: str, expires_at: float) -> None: current = self.now() for key, expiry in list(self._seen.items()): if expiry <= current: del self._seen[key] if jti in self._seen: raise TokenError(f"token {jti} has already been used", code="replay") self._seen[jti] = expires_at class Verifier: """Validates a token against a policy. The order is deliberate: signature first (nothing else is trustworthy until the token is authentic), then issuer, then audience, then time, then the platform checks. """ def __init__(self, *, signers: Mapping[str, Signer], policy: VerificationPolicy, now: Callable[[], float], replay_cache: Optional[ReplayCache] = None) -> None: self.signers = dict(signers) self.policy = policy self.now = now self.replay_cache = replay_cache def verify(self, token: str, *, presented_key: Optional[str] = None) -> Claims: parts = token.split(".") if len(parts) != 3: raise TokenError("malformed token") try: header = json.loads(b64url_decode(parts[0])) except Exception: raise TokenError("undecodable header") from None # Key selection by `kid`, but the SIGNER decides the algorithm — never the token. signer = self.signers.get(str(header.get("kid", ""))) if signer is None: raise TokenError("unknown key id", code="invalid_key") _, payload = signer.verify_signature(token) try: claims = Claims.from_payload(payload) except Exception: raise TokenError("missing required claims") from None if claims.iss not in self.policy.trusted_issuers: raise TokenError(f"untrusted issuer {claims.iss!r}", code="invalid_issuer") # THE confused-deputy defence. A token minted for another service must not be # accepted here, or that service can replay it against us. if claims.aud != self.policy.audience: raise TokenError( f"audience {claims.aud!r} is not {self.policy.audience!r}", code="invalid_audience") current = self.now() skew = self.policy.clock_skew_seconds if claims.exp <= current - skew: raise TokenError("token expired", code="expired") if claims.nbf and claims.nbf > current + skew: raise TokenError("token not yet valid", code="not_yet_valid") if claims.iat > current + skew: raise TokenError("token issued in the future", code="bad_iat") # A short-lived credential that is minted with a long lifetime is not short-lived. if (self.policy.max_lifetime_seconds is not None and claims.exp - claims.iat > self.policy.max_lifetime_seconds): raise TokenError("token lifetime exceeds policy", code="lifetime_too_long") missing = [s for s in self.policy.required_scopes if s not in claims.scope] if missing: raise TokenError(f"missing scopes {missing}", code="insufficient_scope") if len(claims.act) > self.policy.max_chain_depth: raise TokenError( f"delegation chain is {len(claims.act)} deep, max {self.policy.max_chain_depth}", code="chain_too_deep") # Sender constraint: a stolen bearer string is useless without the bound key. if self.policy.require_proof_of_possession or claims.cnf is not None: if claims.cnf is None: raise TokenError("proof of possession required", code="pop_required") if presented_key is None: raise TokenError("no key presented", code="pop_missing") if key_thumbprint(presented_key) != claims.cnf: raise TokenError("presented key does not match cnf", code="pop_mismatch") if claims.jti and self.replay_cache is not None: self.replay_cache.check_and_record(claims.jti, claims.exp) return claims def key_thumbprint(public_key: str) -> str: """A stand-in for RFC 7638 JWK thumbprinting. Deterministic, which is what matters.""" return b64url_encode(hashlib.sha256(public_key.encode("utf-8")).digest())[:32] # ====================================================================================== # 3. Scopes — the narrowing algebra # ====================================================================================== def normalize_scopes(scopes: Iterable[str]) -> Tuple[str, ...]: return tuple(sorted(set(s for s in scopes if s))) def scope_covers(held: str, requested: str) -> bool: """Does a held scope cover a requested one? Supports one wildcard form: ``payments.*`` covers ``payments.read``. Deliberately limited — a scope language with real pattern matching becomes a policy engine, and then nobody can tell what a credential permits by reading it. """ if held == requested: return True if held.endswith(".*"): return requested.startswith(held[:-1]) return False def narrow_scopes(held: Sequence[str], requested: Sequence[str]) -> Tuple[str, ...]: """The intersection, and the ONLY legal direction of travel. An exchange that returns a scope not covered by the input has escalated privilege. This function makes that structurally impossible: it can only ever return a subset. """ out = [r for r in requested if any(scope_covers(h, r) for h in held)] return normalize_scopes(out) class ScopeEscalation(Exception): pass def require_no_escalation(held: Sequence[str], requested: Sequence[str]) -> Tuple[str, ...]: """Narrow, and REFUSE rather than silently dropping. Silently returning a smaller scope than asked for is worse than an error: the caller proceeds believing it has authority it does not, and fails later at a point far from the cause. """ granted = narrow_scopes(held, requested) rejected = [r for r in normalize_scopes(requested) if r not in granted] if rejected: raise ScopeEscalation(f"cannot grant {rejected}; held scopes are {list(held)}") return granted # ====================================================================================== # 4. OAuth 2.1 — authorization code with PKCE, and client credentials # ====================================================================================== def code_challenge(verifier: str, *, method: str = "S256") -> str: """PKCE (RFC 7636). ``S256`` is the only method OAuth 2.1 allows. ``plain`` exists in RFC 7636 and is useless: an attacker who intercepts the challenge has the verifier. The lab refuses it, which is what OAuth 2.1 does. """ if method != "S256": raise ValueError("OAuth 2.1 requires S256; 'plain' offers no protection") if len(verifier) < 43: raise ValueError("a code_verifier must be at least 43 characters (RFC 7636)") return b64url_encode(hashlib.sha256(verifier.encode("ascii")).digest()) @dataclass(frozen=True) class ClientRegistration: client_id: str redirect_uris: Tuple[str, ...] allowed_scopes: Tuple[str, ...] confidential: bool = False secret: str = "" @dataclass class AuthorizationCode: code: str client_id: str user_id: str tenant: str scopes: Tuple[str, ...] challenge: str redirect_uri: str expires_at: float used: bool = False class AuthorizationServer: """OAuth 2.1 + OIDC, reduced to what an agent platform actually uses. What OAuth 2.1 removed and why: the implicit grant (tokens in a URL fragment leak through history and referrers), the resource-owner password grant (the client sees the password, which defeats the point of delegated authorization), and bearer tokens in query strings. PKCE became mandatory for ALL clients, not just public ones. """ def __init__(self, *, issuer: str, signer: Signer, now: Callable[[], float], clients: Mapping[str, ClientRegistration], code_ttl: float = 60.0, access_ttl: float = 300.0) -> None: self.issuer = issuer self.signer = signer self.now = now self.clients = dict(clients) self.code_ttl = code_ttl self.access_ttl = access_ttl self._codes: Dict[str, AuthorizationCode] = {} self._counter = 0 def _next_id(self, prefix: str) -> str: self._counter += 1 return f"{prefix}-{self._counter}" # -- the authorization endpoint ------------------------------------------------ def authorize(self, *, client_id: str, redirect_uri: str, scopes: Sequence[str], challenge: str, user_id: str, tenant: str, challenge_method: str = "S256") -> str: client = self.clients.get(client_id) if client is None: raise TokenError("unknown client", code="invalid_client") # EXACT redirect-URI matching. Prefix matching is how an open redirect becomes a # code-interception attack; OAuth 2.1 requires exact. if redirect_uri not in client.redirect_uris: raise TokenError("redirect_uri does not match exactly", code="invalid_redirect") if challenge_method != "S256": raise TokenError("S256 required", code="invalid_challenge_method") if not challenge: raise TokenError("PKCE challenge required", code="pkce_required") granted = require_no_escalation(client.allowed_scopes, scopes) code = self._next_id("code") self._codes[code] = AuthorizationCode( code=code, client_id=client_id, user_id=user_id, tenant=tenant, scopes=granted, challenge=challenge, redirect_uri=redirect_uri, expires_at=self.now() + self.code_ttl) return code # -- the token endpoint -------------------------------------------------------- def exchange_code(self, *, code: str, client_id: str, verifier: str, redirect_uri: str, audience: str) -> str: record = self._codes.get(code) if record is None: raise TokenError("unknown code", code="invalid_grant") # Single use, enforced even for an expired code, so a replay is never ambiguous. if record.used: raise TokenError("authorization code already used", code="invalid_grant") record.used = True if record.expires_at <= self.now(): raise TokenError("authorization code expired", code="invalid_grant") if record.client_id != client_id: raise TokenError("code was issued to another client", code="invalid_grant") if record.redirect_uri != redirect_uri: raise TokenError("redirect_uri mismatch", code="invalid_grant") # PKCE: the code is bound to whoever requested it. if code_challenge(verifier) != record.challenge: raise TokenError("PKCE verification failed", code="invalid_grant") now = self.now() return self.signer.sign(Claims( iss=self.issuer, sub=record.user_id, aud=audience, iat=now, exp=now + self.access_ttl, nbf=now, jti=self._next_id("jti"), scope=record.scopes, tenant=record.tenant, client_id=client_id, may_delegate=True)) def client_credentials(self, *, client_id: str, secret: str, scopes: Sequence[str], audience: str) -> str: """Machine-to-machine, no user. Simple, and the WRONG default for an agent acting for a person: it erases the user from the chain, so every downstream audit record says the platform did it. """ client = self.clients.get(client_id) if client is None or not client.confidential: raise TokenError("unknown or public client", code="invalid_client") if not hmac.compare_digest(client.secret, secret): raise TokenError("bad client secret", code="invalid_client") granted = require_no_escalation(client.allowed_scopes, scopes) now = self.now() return self.signer.sign(Claims( iss=self.issuer, sub=client_id, aud=audience, iat=now, exp=now + self.access_ttl, nbf=now, jti=self._next_id("jti"), scope=granted, client_id=client_id)) def id_token(self, *, user_id: str, tenant: str, client_id: str) -> str: """OIDC's ID token: about the USER, for the CLIENT. Distinct from an access token, which is for the API. Sending an ID token to an API — or an access token to a client expecting identity — is the confusion that makes people think OIDC and OAuth are the same thing. """ now = self.now() return self.signer.sign(Claims( iss=self.issuer, sub=user_id, aud=client_id, iat=now, exp=now + self.access_ttl, nbf=now, jti=self._next_id("jti"), tenant=tenant)) # ====================================================================================== # 5. RFC 8693 token exchange — the backbone of multi-hop agent identity # ====================================================================================== class ExchangeError(Exception): def __init__(self, reason: str, *, code: str = "invalid_request") -> None: super().__init__(reason) self.reason = reason self.code = code @dataclass(frozen=True) class ExchangeRequest: subject_token: str actor_id: str # who is doing the exchanging actor_kind: str = "agent" audience: str = "" # the NEW audience — narrower scopes: Tuple[str, ...] = () # the NEW scopes — a subset lifetime_seconds: Optional[float] = None bind_to_key: Optional[str] = None delegation: bool = True # False = impersonation (chain erased) class TokenExchange: """RFC 8693, with the four properties that make agent identity work. Every exchange must NARROW (audience and scope), APPEND to the chain, SHORTEN the lifetime, and be refused if the subject token is not delegable. """ def __init__(self, *, issuer: str, signer: Signer, verifier: Verifier, now: Callable[[], float], max_chain_depth: int = MAX_CHAIN_DEPTH, max_lifetime_seconds: float = 300.0, allow_impersonation: bool = False) -> None: self.issuer = issuer self.signer = signer self.verifier = verifier self.now = now self.max_chain_depth = max_chain_depth self.max_lifetime_seconds = max_lifetime_seconds self.allow_impersonation = allow_impersonation self._counter = 0 def exchange(self, request: ExchangeRequest) -> str: # The subject token is VERIFIED, not parsed. The chain in the new token is # derived from a verified assertion — never from anything the caller supplies. subject = self.verifier.verify(request.subject_token) if not subject.may_delegate: raise ExchangeError("subject token is not delegable", code="not_delegable") if not request.audience: raise ExchangeError("an exchange must name a new audience") if request.audience == subject.aud: raise ExchangeError("exchange must narrow the audience", code="no_narrowing") if not request.delegation and not self.allow_impersonation: raise ExchangeError( "impersonation erases the delegation chain and is disabled", code="impersonation_disabled") try: granted = require_no_escalation(subject.scope, request.scopes) except ScopeEscalation as exc: raise ExchangeError(str(exc), code="scope_escalation") from None chain = subject.act if request.delegation: actor = Actor(request.actor_id, request.actor_kind) if any(a.sub == actor.sub for a in chain) or actor.sub == subject.sub: raise ExchangeError( f"{actor.sub} is already in the chain", code="chain_cycle") chain = chain + (actor,) if len(chain) > self.max_chain_depth: raise ExchangeError( f"chain would be {len(chain)} deep, max {self.max_chain_depth}", code="chain_too_deep") else: chain = () # impersonation: the chain is ERASED now = self.now() remaining = subject.exp - now requested = request.lifetime_seconds or self.max_lifetime_seconds # A derived credential can never outlive its parent, and never exceed policy. lifetime = min(requested, self.max_lifetime_seconds, remaining) if lifetime <= 0: raise ExchangeError("subject token has no remaining lifetime", code="expired") self._counter += 1 return self.signer.sign(Claims( iss=self.issuer, sub=subject.sub if request.delegation else request.actor_id, aud=request.audience, iat=now, exp=now + lifetime, nbf=now, jti=f"jti-x{self._counter}", scope=granted, tenant=subject.tenant, act=chain, cnf=key_thumbprint(request.bind_to_key) if request.bind_to_key else None, may_delegate=len(chain) < self.max_chain_depth)) # ====================================================================================== # 6. SPIFFE / SPIRE — workload identity without a secret # ====================================================================================== @dataclass(frozen=True) class Selector: """An attested property of a running workload. The whole point of SPIFFE: identity is derived from what the platform can VERIFY, not from a secret the workload holds.""" kind: str # "k8s:ns" | "k8s:sa" | "docker:image" | "unix:uid" value: str @dataclass(frozen=True) class SpiffeID: trust_domain: str path: str def __str__(self) -> str: return f"spiffe://{self.trust_domain}{self.path}" @classmethod def parse(cls, value: str) -> "SpiffeID": if not value.startswith("spiffe://"): raise ValueError("a SPIFFE ID must start with spiffe://") rest = value[len("spiffe://"):] domain, _, path = rest.partition("/") if not domain or not path: raise ValueError("a SPIFFE ID needs a trust domain and a path") return cls(domain, "/" + path) @dataclass(frozen=True) class SVID: """A short-lived credential carrying a SPIFFE ID. Rotated automatically; never stored; never a secret to manage.""" spiffe_id: SpiffeID expires_at: float serial: int public_key: str def is_valid(self, now: float) -> bool: return now < self.expires_at @dataclass(frozen=True) class RegistrationEntry: spiffe_id: str selectors: Tuple[Selector, ...] def matches(self, attested: Sequence[Selector]) -> bool: """ALL registered selectors must be present. A subset match would let a workload with one matching property claim an identity meant for a narrower set.""" return all(s in attested for s in self.selectors) class SpireServer: """Attest a workload, issue an SVID. The lesson is the direction of trust: the workload does not present a credential to prove who it is. The PLATFORM observes verifiable properties — this pod, this namespace, this service account — and issues an identity based on them. That is what "secret-less" means, and it is why an exfiltrated SVID is useless to an attacker who cannot reproduce the selectors. """ def __init__(self, *, trust_domain: str, now: Callable[[], float], svid_ttl: float = 60.0) -> None: self.trust_domain = trust_domain self.now = now self.svid_ttl = svid_ttl self._entries: List[RegistrationEntry] = [] self._serial = 0 def register(self, entry: RegistrationEntry) -> None: parsed = SpiffeID.parse(entry.spiffe_id) if parsed.trust_domain != self.trust_domain: raise ValueError( f"{entry.spiffe_id} is not in trust domain {self.trust_domain!r}") if not entry.selectors: raise ValueError("a registration entry with no selectors matches everything") self._entries.append(entry) def attest(self, attested: Sequence[Selector], *, public_key: str) -> SVID: matches = [e for e in self._entries if e.matches(attested)] if not matches: raise TokenError("no registration entry matches the attested selectors", code="attestation_failed") if len(matches) > 1: # Ambiguity is a configuration error, and guessing would assign an identity # nondeterministically. Refuse. raise TokenError( f"{len(matches)} entries match; registration is ambiguous", code="ambiguous_attestation") self._serial += 1 return SVID(SpiffeID.parse(matches[0].spiffe_id), self.now() + self.svid_ttl, self._serial, public_key) @dataclass(frozen=True) class MtlsPolicy: """Who may talk to whom, by SPIFFE ID. Federation across trust domains is EXPLICIT — a workload from another domain is refused unless the domain is federated.""" allowed_callers: Tuple[str, ...] federated_domains: Tuple[str, ...] = () def mtls_authorize(client: SVID, server_policy: MtlsPolicy, *, now: float) -> None: if not client.is_valid(now): raise TokenError("client SVID has expired", code="svid_expired") caller = str(client.spiffe_id) if client.spiffe_id.trust_domain not in server_policy.federated_domains: if caller not in server_policy.allowed_callers: raise TokenError(f"{caller} is not an allowed caller", code="mtls_denied") elif caller not in server_policy.allowed_callers: raise TokenError(f"federated caller {caller} is not allowed", code="mtls_denied") # ====================================================================================== # 7. Non-human identity lifecycle # ====================================================================================== class NHIState(str, Enum): REGISTERED = "registered" APPROVED = "approved" ACTIVE = "active" SUSPENDED = "suspended" RETIRED = "retired" NHI_TRANSITIONS: Mapping[NHIState, frozenset] = { NHIState.REGISTERED: frozenset({NHIState.APPROVED, NHIState.RETIRED}), NHIState.APPROVED: frozenset({NHIState.ACTIVE, NHIState.RETIRED}), NHIState.ACTIVE: frozenset({NHIState.SUSPENDED, NHIState.RETIRED}), NHIState.SUSPENDED: frozenset({NHIState.ACTIVE, NHIState.RETIRED}), } @dataclass(frozen=True) class NonHumanIdentity: """An agent, as an identity. ``owner`` is not decoration: an NHI without a named human owner is the finding every identity audit produces, because nobody can answer "should this still exist?" """ identity_id: str kind: str # "agent" | "workload" | "service" owner: str # a HUMAN tenant: str allowed_scopes: Tuple[str, ...] state: NHIState = NHIState.REGISTERED max_credential_lifetime: float = 300.0 class IdentityRegistry: """The NHI inventory: lifecycle, ownership, and the only place a credential is authorized to be minted from.""" def __init__(self, *, now: Callable[[], float]) -> None: self.now = now self._identities: Dict[str, NonHumanIdentity] = {} def register(self, identity: NonHumanIdentity) -> NonHumanIdentity: if identity.identity_id in self._identities: raise ValueError(f"{identity.identity_id} is already registered") if not identity.owner: raise ValueError("every non-human identity needs a human owner") self._identities[identity.identity_id] = identity return identity def get(self, identity_id: str) -> NonHumanIdentity: try: return self._identities[identity_id] except KeyError: raise KeyError(f"unknown identity: {identity_id}") from None def transition(self, identity_id: str, target: NHIState) -> NonHumanIdentity: identity = self.get(identity_id) allowed = NHI_TRANSITIONS.get(identity.state, frozenset()) if target not in allowed: raise ValueError( f"{identity.state.value} -> {target.value} is not a legal transition") updated = replace(identity, state=target) self._identities[identity_id] = updated return updated def may_receive_credentials(self, identity_id: str) -> bool: return self.get(identity_id).state is NHIState.ACTIVE # ====================================================================================== # 8. Just-in-time credentials # ====================================================================================== @dataclass(frozen=True) class CredentialRequest: identity_id: str audience: str scopes: Tuple[str, ...] subject_token: Optional[str] = None # the user context, when acting for someone bind_to_key: Optional[str] = None lifetime_seconds: float = 60.0 class JitCredentialBroker: """Mint a credential at the moment of use, scoped to the action, expiring in seconds. This replaces the long-lived secret. Nothing is stored, so nothing can leak from storage; the lifetime is short, so a leaked credential is worth little; the audience is narrow, so it is worth little *somewhere else*. """ def __init__(self, *, registry: IdentityRegistry, exchange: TokenExchange, issuer: str, signer: Signer, now: Callable[[], float], max_lifetime_seconds: float = 120.0) -> None: self.registry = registry self.exchange = exchange self.issuer = issuer self.signer = signer self.now = now self.max_lifetime_seconds = max_lifetime_seconds self._counter = 0 def issue(self, request: CredentialRequest) -> str: identity = self.registry.get(request.identity_id) if identity.state is not NHIState.ACTIVE: raise TokenError( f"identity {identity.identity_id} is {identity.state.value}", code="identity_not_active") # The agent's own registered ceiling bounds what it can ever ask for. granted = require_no_escalation(identity.allowed_scopes, request.scopes) lifetime = min(request.lifetime_seconds, self.max_lifetime_seconds, identity.max_credential_lifetime) if request.subject_token is not None: # Acting for a user: EXCHANGE, so the user stays in the chain. return self.exchange.exchange(ExchangeRequest( subject_token=request.subject_token, actor_id=identity.identity_id, actor_kind=identity.kind, audience=request.audience, scopes=granted, lifetime_seconds=lifetime, bind_to_key=request.bind_to_key)) # No user context: a workload credential. It can do only what the workload may do # on its own behalf — which is why user-scoped actions must supply a subject token. self._counter += 1 now = self.now() return self.signer.sign(Claims( iss=self.issuer, sub=identity.identity_id, aud=request.audience, iat=now, exp=now + lifetime, nbf=now, jti=f"jti-w{self._counter}", scope=granted, tenant=identity.tenant, act=(Actor(identity.identity_id, identity.kind),), cnf=key_thumbprint(request.bind_to_key) if request.bind_to_key else None, may_delegate=False)) def describe_chain(claims: Claims) -> str: """The audit rendering: user, then every actor, earliest first.""" hops = [claims.sub] + [a.sub for a in claims.act] return " -> ".join(hops) # ====================================================================================== # Worked example # ====================================================================================== def _clock(start: float = 1_000.0, step: float = 1.0) -> Callable[[], float]: state = {"t": start - step} def now() -> float: state["t"] += step return state["t"] return now def main() -> None: # pragma: no cover - narrative output now = _clock() entra = Signer("entra-1", b"enterprise-idp-key") platform = Signer("platform-1", b"platform-sts-key") clients = { "teams-channel": ClientRegistration( "teams-channel", ("https://bank.ae/cb",), ("payments.read", "payments.release", "crm.read"), confidential=False), "batch-runner": ClientRegistration( "batch-runner", (), ("reports.generate",), confidential=True, secret="s3cret"), } auth = AuthorizationServer(issuer="https://login.bank.ae", signer=entra, now=now, clients=clients, access_ttl=600.0) print("=" * 78) print("1. OAUTH 2.1: AUTHORIZATION CODE + PKCE") print("=" * 78) verifier_secret = "a" * 43 challenge = code_challenge(verifier_secret) code = auth.authorize(client_id="teams-channel", redirect_uri="https://bank.ae/cb", scopes=["payments.read", "payments.release"], challenge=challenge, user_id="u-42", tenant="wholesale") user_token = auth.exchange_code(code=code, client_id="teams-channel", verifier=verifier_secret, redirect_uri="https://bank.ae/cb", audience="agent-platform") print(f" code -> access token for aud='agent-platform'") print(f" what OAuth 2.1 removed: implicit grant, password grant, tokens in query") print(f" strings; PKCE is now mandatory for ALL clients, and S256 only.") try: auth.exchange_code(code=code, client_id="teams-channel", verifier=verifier_secret, redirect_uri="https://bank.ae/cb", audience="agent-platform") except TokenError as exc: print(f" code replay -> [{exc.code}] {exc.reason}") try: auth.authorize(client_id="teams-channel", redirect_uri="https://bank.ae/cb-evil", scopes=["payments.read"], challenge=challenge, user_id="u-42", tenant="wholesale") except TokenError as exc: print(f" redirect mismatch-> [{exc.code}] {exc.reason}") try: auth.authorize(client_id="teams-channel", redirect_uri="https://bank.ae/cb", scopes=["treasury.trade"], challenge=challenge, user_id="u-42", tenant="wholesale") except ScopeEscalation as exc: print(f" scope escalation -> {exc}") print() print("=" * 78) print("2. VERIFICATION — AND THE ATTACK EACH CHECK STOPS") print("=" * 78) policy = VerificationPolicy(audience="agent-platform", trusted_issuers=("https://login.bank.ae",)) verifier = Verifier(signers={"entra-1": entra}, policy=policy, now=now, replay_cache=ReplayCache(now=now)) claims = verifier.verify(user_token) print(f" verified: sub={claims.sub} aud={claims.aud} scope={list(claims.scope)}") wrong_audience = Verifier( signers={"entra-1": entra}, policy=VerificationPolicy(audience="core-banking", trusted_issuers=("https://login.bank.ae",)), now=now) try: wrong_audience.verify(user_token) except TokenError as exc: print(f" wrong audience -> [{exc.code}] {exc.reason}") print(" ^ THE confused-deputy defence: a token for us must not work elsewhere.") forged = Signer("entra-1", b"attacker-key") try: verifier.verify(forged.sign(claims)) except TokenError as exc: print(f" forged signature -> [{exc.code}] {exc.reason}") try: verifier.verify(user_token) except TokenError as exc: print(f" jti replay -> [{exc.code}] {exc.reason}") print() print("=" * 78) print("3. THE DELEGATION CHAIN — DERIVED, NARROWED, CHAINED, SHORT-LIVED") print("=" * 78) platform_policy = VerificationPolicy(audience="agent-platform", trusted_issuers=("https://login.bank.ae", "https://sts.bank.ae")) platform_verifier = Verifier(signers={"entra-1": entra, "platform-1": platform}, policy=platform_policy, now=now) exchange = TokenExchange(issuer="https://sts.bank.ae", signer=platform, verifier=platform_verifier, now=now, max_lifetime_seconds=120.0) print(" user token: aud=agent-platform, scope=[payments.read, payments.release]") hop1 = exchange.exchange(ExchangeRequest( subject_token=user_token, actor_id="orchestrator", audience="payments-investigator", scopes=("payments.read", "payments.release"))) hop1_claims = Claims.from_payload(json.loads(b64url_decode(hop1.split(".")[1]))) print(f" hop 1 -> aud={hop1_claims.aud:<24} scope={list(hop1_claims.scope)}") print(f" chain: {describe_chain(hop1_claims)}") inv_verifier = Verifier( signers={"platform-1": platform}, policy=VerificationPolicy(audience="payments-investigator", trusted_issuers=("https://sts.bank.ae",)), now=now) inv_exchange = TokenExchange(issuer="https://sts.bank.ae", signer=platform, verifier=inv_verifier, now=now, max_lifetime_seconds=120.0) hop2 = inv_exchange.exchange(ExchangeRequest( subject_token=hop1, actor_id="payments-investigator", audience="core-banking", scopes=("payments.read",))) hop2_claims = Claims.from_payload(json.loads(b64url_decode(hop2.split(".")[1]))) print(f" hop 2 -> aud={hop2_claims.aud:<24} scope={list(hop2_claims.scope)}") print(f" chain: {describe_chain(hop2_claims)}") print(f" lifetime {hop2_claims.exp - hop2_claims.iat:.0f}s " f"(never exceeds the parent's remaining life)") print(" -> narrowed at every hop; the chain is APPEND-ONLY and derived from a") print(" VERIFIED assertion, never from anything the caller supplied.") try: inv_exchange.exchange(ExchangeRequest( subject_token=hop1, actor_id="payments-investigator", audience="core-banking", scopes=("treasury.trade",))) except ExchangeError as exc: print(f" widening scope -> [{exc.code}] {exc.reason}") try: inv_exchange.exchange(ExchangeRequest( subject_token=hop1, actor_id="orchestrator", audience="core-banking", scopes=("payments.read",))) except ExchangeError as exc: print(f" chain cycle -> [{exc.code}] {exc.reason}") try: inv_exchange.exchange(ExchangeRequest( subject_token=hop1, actor_id="x", audience="core-banking", scopes=("payments.read",), delegation=False)) except ExchangeError as exc: print(f" impersonation -> [{exc.code}] {exc.reason}") print() print("=" * 78) print("4. SPIFFE / SPIRE — IDENTITY WITHOUT A SECRET") print("=" * 78) spire = SpireServer(trust_domain="bank.ae", now=now, svid_ttl=60.0) spire.register(RegistrationEntry( "spiffe://bank.ae/ns/agents/sa/investigator", (Selector("k8s:ns", "agents"), Selector("k8s:sa", "investigator")))) svid = spire.attest([Selector("k8s:ns", "agents"), Selector("k8s:sa", "investigator"), Selector("k8s:pod", "investigator-7f9")], public_key="pk-investigator") print(f" attested -> {svid.spiffe_id} (ttl {spire.svid_ttl:.0f}s, serial {svid.serial})") print(" the workload presented NO secret: the platform observed verifiable") print(" properties and issued an identity from them. That is 'secret-less'.") try: spire.attest([Selector("k8s:ns", "agents")], public_key="pk-x") except TokenError as exc: print(f" partial selectors-> [{exc.code}] {exc.reason}") mtls = MtlsPolicy(allowed_callers=("spiffe://bank.ae/ns/agents/sa/investigator",)) mtls_authorize(svid, mtls, now=now()) print(f" mTLS: allowed") foreign = SVID(SpiffeID.parse("spiffe://partner.example/ns/x/sa/y"), expires_at=1e12, serial=1, public_key="pk") try: mtls_authorize(foreign, mtls, now=now()) except TokenError as exc: print(f" foreign domain -> [{exc.code}] {exc.reason} (federation is explicit)") print() print("=" * 78) print("5. NHI LIFECYCLE AND JIT CREDENTIALS") print("=" * 78) registry = IdentityRegistry(now=now) registry.register(NonHumanIdentity( "payments-investigator", "agent", owner="layla.almansouri", tenant="wholesale", allowed_scopes=("payments.read", "crm.read"))) print(f" registered with a HUMAN owner (an NHI without one is the standard finding)") registry.transition("payments-investigator", NHIState.APPROVED) registry.transition("payments-investigator", NHIState.ACTIVE) broker = JitCredentialBroker(registry=registry, exchange=exchange, issuer="https://sts.bank.ae", signer=platform, now=now) workload_token = broker.issue(CredentialRequest( identity_id="payments-investigator", audience="crm", scopes=("crm.read",), bind_to_key="pk-investigator", lifetime_seconds=60.0)) wl = Claims.from_payload(json.loads(b64url_decode(workload_token.split(".")[1]))) print(f" JIT workload cred: aud={wl.aud} scope={list(wl.scope)} " f"ttl={wl.exp - wl.iat:.0f}s cnf={wl.cnf[:12]}...") print(" nothing stored, 60s lifetime, one audience, bound to a key.") crm_verifier = Verifier( signers={"platform-1": platform}, policy=VerificationPolicy(audience="crm", trusted_issuers=("https://sts.bank.ae",), require_proof_of_possession=True), now=now) crm_verifier.verify(workload_token, presented_key="pk-investigator") print(" proof of possession: correct key accepted") try: crm_verifier.verify(workload_token, presented_key="pk-stolen") except TokenError as exc: print(f" stolen token -> [{exc.code}] {exc.reason}") print(" ^ a bearer string alone is useless. Theft is no longer sufficient.") registry.transition("payments-investigator", NHIState.SUSPENDED) try: broker.issue(CredentialRequest("payments-investigator", "crm", ("crm.read",))) except TokenError as exc: print(f" suspended agent -> [{exc.code}] {exc.reason}") print(" revocation latency = the credential TTL. That is why 60 seconds matters.") print() print("=" * 78) print("6. WHY NOT JUST A SERVICE ACCOUNT?") print("=" * 78) m2m = auth.client_credentials(client_id="batch-runner", secret="s3cret", scopes=("reports.generate",), audience="reporting") m2m_claims = Claims.from_payload(json.loads(b64url_decode(m2m.split(".")[1]))) print(f" client-credentials token: sub={m2m_claims.sub} act={list(m2m_claims.act)}") print(f" audit chain: {describe_chain(m2m_claims)}") print() print(f" delegated token chain: {describe_chain(hop2_claims)}") print() print(" The first says 'batch-runner did it'. The second says 'u-42 asked, the") print(" orchestrator delegated, the investigator acted'. Only the second answers") print(" an examiner's question, and only the second can be bounded by what u-42") print(" is personally entitled to do.") if __name__ == "__main__": # pragma: no cover main()