akif/arclet-boards

Arclet Boards — private Kanban workspaces with AppID, DataK3, shared web and MCP actions.

• dodil-challenge
• dodil-challenge-boards

A remix template · 1 remixes · Remix akif/arclet-boards

app/infrastructure/identity.py

"""Trust only the verified person identity injected by Ignite's gateway.

The gateway strips inbound identity headers. This module must never be exposed
directly as a publicly reachable uvicorn service without that gateway.
See the current AppID Ignite recipe and AppID's documented amr claims.
"""
import base64
import json
import os
import time

from app.domain.model import Invalid, Person, email


def person_from(headers) -> Person | None:
    if headers.get("x-dodil-auth-source") != "pool":
        return None
    pool = os.environ.get("DODIL_USER_POOL", "").strip()
    if not pool or "/" not in pool:
        return None
    try:
        user = json.loads(headers.get("x-dodil-user", ""))
        encoded = headers.get("x-dodil-user-jwt", "").split(".")[1]
        claims = json.loads(base64.urlsafe_b64decode(encoded + "=" * (-len(encoded) % 4)))
        if not isinstance(user, dict) or not isinstance(claims, dict):
            return None
        expected = f"https://appid.dodil.io/{pool}"
        audience = claims.get("aud")
        audiences = audience if isinstance(audience, list) else [audience]
        amr = claims.get("amr", [])
        if not isinstance(amr, list) or not ({"pwd", "ext"} & set(amr)):
            return None
        if "client_credentials" in amr or claims.get("iss") != expected:
            return None
        if f"pool:{pool.split('/', 1)[1]}" not in audiences:
            return None
        if float(claims.get("exp", 0)) <= time.time():
            return None
        sub = claims.get("sub")
        address = email(claims.get("email", ""))
        if not isinstance(sub, str) or not sub or user.get("sub") != sub:
            return None
        if email(user.get("email", "")) != address:
            return None
        connection = claims.get("connection", "")
        if connection not in ("local", "oidc") and not connection.startswith("oauth:"):
            return None
        return Person(sub=sub, email=address)
    except (ValueError, IndexError, TypeError, KeyError, Invalid, AttributeError):
        return None