zanii-id

Python SDK for Zanii ID — OAuth 2.1 / OpenID Connect sign-in for products in the Zanii ecosystem.

Authorization Code flow with mandatory PKCE, offline ID-token verification against the issuer's JWKS, and a FastAPI integration that mounts the whole login round-trip for you.

pip install zanii-id

Use it

Configuration comes from ZANII_ISSUER, ZANII_CLIENT_ID, ZANII_CLIENT_SECRET and ZANII_REDIRECT_URI, and is validated eagerly so a misconfigured deployment fails at startup rather than on a user's first login.

from zanii_id import ZaniiClient

zanii = ZaniiClient()
req = zanii.get_authorization_url()       # keep req.state / req.nonce / req.verifier in session
# ... redirect the user to req.url, then on your callback route:
tokens = await zanii.exchange_code(code, req, received_state)
claims = zanii.verify_id_token(tokens.id_token, nonce=req.nonce)
user = await zanii.get_user(tokens.access_token)

FastAPI

from zanii_id import ZaniiClient
from zanii_id.integrations.fastapi import build_auth_router, install_zanii, require_zanii_auth

zanii = ZaniiClient()
install_zanii(app, zanii, session_secret=SECRET)
app.include_router(build_auth_router(zanii, session_secret=SECRET))

@app.get("/dashboard")
async def dashboard(user = Depends(require_zanii_auth)):
    return {"zanii_user_id": user.zanii_user_id}

That mounts /auth/login, /auth/callback and /auth/logout. require_zanii_auth refreshes a stale access token once and rotates the session cookie before giving up.

Provisioning users locally

sub is an immutable zanii_user_id. Upsert on it — never on email, which users change:

INSERT INTO users (zanii_user_id, ...) VALUES ($1, ...)
ON CONFLICT (zanii_user_id) DO NOTHING;

Agent activity

If your product records agent receipts on the Zanii ledger, stamp them with the user's subject tag so they can audit their own slice:

from zanii_id.activity import subject_tag, fetch_activity   # pip install 'zanii-id[subject]'

tag = subject_tag(user.did, client_id)     # pass to record(..., subject_tag=tag)
entries = await fetch_activity(tag)        # every receipt verified offline

Invalid receipts come back with verified=False and a flag_reason rather than being dropped — a truncated slice appends its own flagged entry, so a cut page never reads as complete.

Lifecycle webhooks

Register a lifecycle_webhook_url on your client (console or /orgs/clients) and Zanii ID POSTs account events to it, signed over the raw body with the same webhook_secret you already hold: X-Zanii-Signature: sha256=<hmac> plus X-Zanii-Event. verify_lifecycle_event checks it in constant time and returns the parsed event. Give it the raw request bytes.

from zanii_id import LifecycleSignatureError, verify_lifecycle_event

@app.post("/webhooks/zanii-id")
async def lifecycle(request: Request):
    try:
        ev = verify_lifecycle_event(await request.body(), request.headers.get("X-Zanii-Signature"), WEBHOOK_SECRET)
    except LifecycleSignatureError:
        return Response(status_code=401)
    if ev.event == "user.deleted":
        await users.erase(ev.sub)
    return Response(status_code=204)

Deliveries retry with backoff for about an hour (8 attempts). Answer 2xx quickly and do the work afterwards. sub is the identifier you receive at login, so key on it as usual.

Two-step verification and amr

Every id_token carries acr (urn:zanii:mfa when the session has a second factor, urn:zanii:pwd otherwise) and amr (["pwd"], ["pwd","otp"] or ["webauthn"]). claims = zanii.verify_id_token(tokens.id_token) gives you claims.acr. To require a second factor on a route, start the flow with acr_values=ACR_MFA:

from zanii_id import ACR_MFA
req = zanii.get_authorization_url(acr_values=ACR_MFA)   # or prompt="login", max_age=0

Enrolled users are challenged in place; users with no second factor come back with error=unmet_authentication_requirements - send them to {issuer}/ui/account to enrol. With the FastAPI router, /auth/login?acr=mfa does the same.

Public clients and revocation

Leave client_secret unset for a public client and the SDK sends client_id in the token request body, as RFC 6749 requires. revoke() authenticates the same way /token does. Add age to the scopes to receive age_over + age_method + age_provider (a verified check) or age_declared (the sign-up gate only); decide per app which one is enough. The default scope includes offline_access, which is what a refresh token requires; drop it to get session-bound tokens only.

Machines, agents and logout (0.3.0)

# A service acting as itself, or an agent that will act for users.
cc = await client.client_credentials(scope="agent", audience="cli_other")

# RFC 8693: act for a user towards another Zanii service; name the agent in `act`.
ex = await client.exchange_token(user_access_token, actor_token=cc.access_token, audience="cli_other")
# invalid_grant here means the audience's organization refuses this actor (receipted either way).

# RFC 8628 for CLIs, TVs and headless agents: show the code, then poll at the server's pace.
d = await client.device_authorize(scope="openid")
print(d.user_code, d.verification_uri_complete)
tokens = await client.poll_device_token(d)  # honours interval, slow_down, expires_in, Retry-After

# DPoP (RFC 9449): tokens bound to a key; a stolen token is useless.
from zanii_id import generate_dpop_key
client = ZaniiClient(dpop_key=generate_dpop_key())  # every token, userinfo and revoke call carries a proof

# private_key_jwt: no shared secret; register the public JWK Set on the client.
client = ZaniiClient(client_assertion_key=pem_private_key, client_assertion_kid="k1")

# PAR (RFC 9126): nothing but client_id + request_uri reaches the browser.
req = await client.get_authorization_url_par(acr_values=ACR_MFA)

# Logout: RP-initiated URL, and the two signals the IdP sends you.
client.logout_url(id_token, "https://app.test/bye", state="s1")
token = client.verify_logout_token(logout_token)      # back-channel POST -> end sessions for token.sid
sid = client.parse_frontchannel_logout(request.query)  # front-channel iframe ?iss=&sid=

The FastAPI router mounts POST /auth/backchannel-logout and GET /auth/frontchannel-logout when you pass on_backchannel_logout=async def (sid): ...; the hook ends every local session tied to that IdP session id. Register the two URIs on the client.

Server-to-server: post_receipt_event(issuer, platform_client_id=..., webhook_secret=..., event=...) relays a ledger receipt event to the user's activity page, and OrgClient(issuer, api_key) wraps /orgs/* (settings and the agent deny list, clients, lifecycle URLs, domain and workload binding, screen_agent).

Claims: IdTokenClaims and UserInfo type act, cnf and the age scope (age_over + age_method + age_provider, or age_declared; .age_verified).

Cheap for the issuer: one connection pool per event loop (10 connections), a timeout on every call, JWKS cached, reads retried once only when the server answered 429/502/503/504 (honouring Retry-After), token grants never retried, device polling never faster than told.

Notes

• alg is pinned from the discovery document, never trusted from the token header.
• Token POSTs are never retried; grants are single-use.
• The JWKS cache refetches exactly once on an unseen kid, then fails hard.

Licence

Apache-2.0. See LICENSE.