Documentation/Démarrer/Exemple de bout en bout

Exemple de bout en bout

Un parcours complet avec tous les appels : de la redirection au webhook, et un programme Python prêt à adapter.

Cet exemple suit un cas concret : Acme Assurance veut inscrire un client avec son identité vérifiée, sa pièce d’identité et son e-mail. Le numéro de sécurité sociale n’est pas encore validé côté TrustMe : Acme utilise donc le partage différé.

Les valeurs (identifiants, jetons, dates) sont fictives et les réponses sont des extraits indicatifs ; reportez-vous à la référence pour le détail des champs.

Élément	Valeur
Scopes demandés	`openid profile email private:doc.id private:ssn.fr private:delayed_sharing`
`client_id`	`e251305f-5db8-11f1-8cb7-c8a3620c309f`
`redirect_uri`	`https://app.example.com/callback`
Webhook	`https://app.example.com/webhooks/a3bc`

# Variables utilisées dans les exemples (ne commitez jamais le secret)
export A3BC_CLIENT_ID=e251305f-5db8-11f1-8cb7-c8a3620c309f
export A3BC_CLIENT_SECRET=<votre secret>
export ISSUER=https://oidc-sandbox.a3bc.io

Étape 1 : lire le discovery

curl -s $ISSUER/.well-known/openid-configuration

{
  "issuer": "https://oidc-sandbox.a3bc.io/",
  "authorization_endpoint": "https://oidc-sandbox.a3bc.io/oauth2/auth",
  "token_endpoint": "https://oidc-sandbox.a3bc.io/oauth2/token",
  "userinfo_endpoint": "…",
  "jwks_uri": "…"
}

Extrait : le document contient d’autres champs. Mettez-le en cache plutôt que de le relire à chaque connexion.

Étape 2 : rediriger l’utilisateur

Quand le client clique sur « Se connecter avec TrustMe », votre serveur génère state et nonce, les garde en session, et répond par une redirection HTTP 302 vers :

https://oidc-sandbox.a3bc.io/oauth2/auth
  ?response_type=code
  &client_id=e251305f-5db8-11f1-8cb7-c8a3620c309f
  &redirect_uri=https%3A%2F%2Fapp.example.com%2Fcallback
  &scope=openid%20profile%20email%20private%3Adoc.id%20private%3Assn.fr%20private%3Adelayed_sharing
  &state=Zk3xQ0b9d1Yw
  &nonce=OEozzDDlfS0qb0uidFVZ

A3BC lui fait ouvrir l’app TrustMe, par saisie de son identifiant (téléphone) ou par scan d’un QR code. Il s’y authentifie et voit la liste de ce que demande Acme. Il choisit ce qu’il partage, y compris des documents encore en cours de validation (grâce à private:delayed_sharing). Rien n’est visible pour votre serveur pendant cette étape.

Étape 4 : le retour sur votre callback

https://app.example.com/callback?code=Qm1xX3...&state=Zk3xQ0b9d1Yw

Comparez state à celui de la session. S’il diffère, ou si l’URL contient error=…, arrêtez-vous. Voir les erreurs.

Étape 5 : échanger le code

curl -s -u "$A3BC_CLIENT_ID:$A3BC_CLIENT_SECRET" \
  -d grant_type=authorization_code \
  -d code="Qm1xX3..." \
  --data-urlencode redirect_uri="https://app.example.com/callback" \
  $ISSUER/oauth2/token

{
  "access_token": "rrQJcwEjCsUdTTxm4f0fzlufhdnsJcQK564mcp1dpM.inBXXtDnABPRHg…",
  "id_token": "eyJhbGciOiJSUzI1NiIs…",
  "token_type": "bearer",
  "expires_in": 599,
  "scope": "openid profile email private:doc.id private:ssn.fr private:delayed_sharing"
}

L’access_token est une chaîne opaque, l’id_token un JWT. Voir Jetons et durées de vie.

Étape 6 : vérifier et lire l’id_token

Après vérification de la signature, de iss, aud, exp et nonce, le contenu est :

{
  "iss": "https://oidc-sandbox.a3bc.io/",
  "aud": ["e251305f-5db8-11f1-8cb7-c8a3620c309f"],
  "sub": "f997cfff165741f98038628f103fedb3",
  "nonce": "OEozzDDlfS0qb0uidFVZ",
  "iat": 1781853814,
  "exp": 1781854414,
  "given_name": "JOHN",
  "family_name": "DOE",
  "gender": "male",
  "birthdate": "1978-03-12",
  "email": "john.doe@example.com",
  "email_verified": true,
  "private:doc.id": "https://oidc-sandbox.a3bc.io/doc/id",
  "private:doc.id.hash": ["sha256:f20a7da2c8e3ebba9afc91fcb8592b9090c87427f5f332bb2bfeb74cb6ece73e"],
  "private:ssn": "",
  "private:ssn_key": "",
  "private:ssn_verified": "",
  "private:ssn.credential.issuance": "",
  "private:ssn.credential.expiration": "",
  "private:delayed_sharing": "https://…/delayed-sharing/dddb6b7c-2730-4f52-bc5b-76c5e7faf549"
}

À lire ainsi :

les claims d’identité et d’e-mail sont présents et validés ;
private:doc.id est une URL à appeler (étape 8) ;
les claims private:ssn* sont présents mais vides : le numéro de sécurité sociale est en attente de validation, et private:delayed_sharing donne l’URL pour le récupérer plus tard (étape 9).

Stockez maintenant tout ce qui est non vide. Voir Lire les claims.

Étape 7 : userinfo (facultatif)

Les mêmes informations sont disponibles avec l’access_token :

curl -s -H "Authorization: Bearer $ACCESS_TOKEN" "$USERINFO_ENDPOINT"

C’est utile pour comparer avec l’id_token ; ce n’est pas nécessaire si vous avez déjà vérifié celui-ci.

Étape 8 : récupérer le document

curl -s -H "Authorization: Bearer $ACCESS_TOKEN" https://oidc-sandbox.a3bc.io/doc/id

{
  "metadata": { "name": "identity card", "identifier": "SPECI2014", "issued": "2020-04-05",
                "expired": "2030-04-04", "country": "fr", "issuedBy": "…" },
  "document": [
    { "name": "front.png", "content": "iVBORw0KGgoAAAANSUhEU…" },
    { "name": "back.png",  "content": "iVBORw0KGgoAAAANSUhEU…" }
  ]
}

Décodez chaque content (base64) et comparez l’empreinte à private:doc.id.hash :

echo "$CONTENT" | base64 -d > front.png
sha256sum front.png     # doit correspondre à l'entrée du claim, sans le préfixe « sha256: »

Étape 9 : partage différé

Quelques heures plus tard, TrustMe valide le numéro de sécurité sociale. A3BC vous envoie un webhook :

POST https://app.example.com/webhooks/a3bc
X-Webhook-Signature: v1:006974CFD99911E1C67870A0FD57F4BAF51B4C9B02A339590D97EDB0D7D94FA7

{
  "id": "84a459b6-d297-11ec-8ff0-c3c3ea77040f",
  "name": "delayed_sharing.update",
  "timestamp_ms": 1781857790000,
  "attempt": 1,
  "data": {
    "delayed_sharing": "https://…/delayed-sharing/dddb6b7c-2730-4f52-bc5b-76c5e7faf549",
    "new_claims": ["private:ssn", "private:ssn_key", "private:ssn_verified"]
  }
}

Après avoir vérifié la signature (voir Webhooks) et répondu 204, votre serveur demande un jeton client_credentials, puis appelle l’URL data.delayed_sharing :

TOKEN=$(curl -s -u "$A3BC_CLIENT_ID:$A3BC_CLIENT_SECRET" \
  -d grant_type=client_credentials -d scope=private:delayed_sharing \
  $ISSUER/oauth2/token | jq -r .access_token)
curl -s -H "Authorization: Bearer $TOKEN" \
  "https://…/delayed-sharing/dddb6b7c-2730-4f52-bc5b-76c5e7faf549"

{
  "claims": {
    "given_name": "JOHN",
    "private:ssn": "1780375115124",
    "private:ssn_key": "83",
    "private:ssn_verified": "document"
  },
  "shared_until": "2026-06-19T08:21:58Z",
  "waiting_validation_claims": []
}

waiting_validation_claims est vide : plus aucune information ne sera ajoutée. Stockez les valeurs non vides : l’utilisateur pourrait supprimer le credential ensuite, ou révoquer son consentement (événement delayed_sharing.revoke, puis HTTP 403 code 00020).

Le programme complet

Ce programme Flask réalise toutes les étapes ci-dessus : /login, /callback (échange, vérification de l’id_token, document avec contrôle d’empreinte, partage différé) et /webhooks/a3bc. Il est testé contre un faux serveur A3BC (états, nonce, signature, webhook rejoué, signature fausse).

Dépendances : pip install flask requests "PyJWT[crypto]". Variables : A3BC_CLIENT_ID, A3BC_CLIENT_SECRET, A3BC_REDIRECT_URI, A3BC_WEBHOOK_SECRET, FLASK_SECRET_KEY.

"""Relying Party minimal : connexion « Se connecter avec TrustMe », callback et webhook."""
import base64
import hashlib
import hmac
import json
import os
import secrets
import urllib.parse

import jwt  # PyJWT
import requests
from flask import Flask, abort, redirect, request, session

ISSUER = os.environ.get("A3BC_ISSUER", "https://oidc-sandbox.a3bc.io")
CLIENT_ID = os.environ["A3BC_CLIENT_ID"]
CLIENT_SECRET = os.environ["A3BC_CLIENT_SECRET"]
REDIRECT_URI = os.environ["A3BC_REDIRECT_URI"]          # ex. https://app.example.com/callback
WEBHOOK_SECRETS = {"v1": os.environ.get("A3BC_WEBHOOK_SECRET", "")}
SCOPES = "openid profile email private:doc.id private:delayed_sharing"

app = Flask(__name__)
app.secret_key = os.environ["FLASK_SECRET_KEY"]          # signe le cookie de session

DISCO = requests.get(f"{ISSUER}/.well-known/openid-configuration", timeout=10).json()
JWKS = jwt.PyJWKClient(DISCO["jwks_uri"])
BASIC = (CLIENT_ID, CLIENT_SECRET)                        # client_secret_basic
SEEN_EVENTS = set()                                       # à remplacer par Redis/BDD (avec TTL)


# --- 1. Rediriger l'utilisateur ------------------------------------------------
@app.get("/login")
def login():
    session["state"] = secrets.token_urlsafe(24)
    session["nonce"] = secrets.token_urlsafe(24)
    params = {
        "response_type": "code",
        "client_id": CLIENT_ID,
        "redirect_uri": REDIRECT_URI,
        "scope": SCOPES,
        "state": session["state"],
        "nonce": session["nonce"],
    }
    return redirect(DISCO["authorization_endpoint"] + "?" + urllib.parse.urlencode(params))


# --- 2. Recevoir le retour, échanger le code, vérifier l'id_token --------------
@app.get("/callback")
def callback():
    if "error" in request.args:                           # consentement refusé, ACR non satisfaite...
        return f"Connexion refusée : {request.args.get('error_description') or request.args['error']}", 400
    if not secrets.compare_digest(request.args.get("state", ""), session.pop("state", "!")):
        abort(400, "state invalide")
    nonce = session.pop("nonce", None)

    r = requests.post(
        DISCO["token_endpoint"], auth=BASIC, timeout=15,
        data={"grant_type": "authorization_code", "code": request.args["code"],
              "redirect_uri": REDIRECT_URI},
    )
    if r.status_code != 200:
        abort(502, f"échange du code refusé : {r.text}")
    tokens = r.json()

    key = JWKS.get_signing_key_from_jwt(tokens["id_token"])
    claims = jwt.decode(tokens["id_token"], key.key, algorithms=["RS256", "ES256"],
                        audience=CLIENT_ID, issuer=DISCO["issuer"])
    if claims.get("nonce") != nonce:
        abort(400, "nonce invalide")

    # Stockez TOUT de suite ce dont vous avez besoin : l'accès n'est garanti que 30 minutes.
    save_user(claims["sub"], claims)

    doc_url = claims.get("private:doc.id")
    if doc_url:
        d = requests.get(doc_url, headers={"Authorization": f"Bearer {tokens['access_token']}"}, timeout=20)
        d.raise_for_status()
        for doc, expected in zip(d.json()["document"], claims.get("private:doc.id.hash", [])):
            raw = base64.b64decode(doc["content"])
            if "sha256:" + hashlib.sha256(raw).hexdigest() != expected:
                abort(502, "empreinte du document différente du claim")
            save_document(claims["sub"], doc["name"], raw)

    if claims.get("private:delayed_sharing"):
        refresh_delayed(claims["private:delayed_sharing"])   # certains claims peuvent encore manquer
    return f"Bienvenue {claims.get('given_name', '')}"


# --- 3. Partage différé ---------------------------------------------------------
def delayed_token():
    r = requests.post(DISCO["token_endpoint"], auth=BASIC, timeout=15,
                      data={"grant_type": "client_credentials", "scope": "private:delayed_sharing"})
    r.raise_for_status()
    return r.json()["access_token"]                       # ~10 min : redemandez-en un à chaque fois


def refresh_delayed(url):
    r = requests.get(url, headers={"Authorization": f"Bearer {delayed_token()}"}, timeout=15)
    if r.status_code != 200:
        code = r.json().get("code") if r.headers.get("content-type", "").startswith("application/json") else None
        app.logger.warning("delayed sharing HTTP %s code=%s", r.status_code, code)
        return None   # 00001/00010 : réessayer plus tard ; 00020/00021/00022 : règle métier, pas de retry
    body = r.json()
    save_claims_by_url(url, {k: v for k, v in body["claims"].items() if v not in ("", None)})
    return body


# --- 4. Webhook -----------------------------------------------------------------
def signature_ok(raw: bytes, header: str) -> bool:
    version, _, digest = header.partition(":")
    secret = WEBHOOK_SECRETS.get(version.strip().lower())
    if not secret or not digest:
        return False
    key = base64.b64decode(secret.strip())                # la clé est en base64 : la décoder d'abord
    expected = hmac.new(key, raw, hashlib.sha256).hexdigest().upper()
    return hmac.compare_digest(expected, digest.strip().upper())


@app.post("/webhooks/a3bc")
def webhook():
    raw = request.get_data()                              # octets bruts, avant tout parsing JSON
    if not signature_ok(raw, request.headers.get("X-Webhook-Signature", "")):
        return {"error": "invalid_signature"}, 401
    event = json.loads(raw)
    if event["id"] in SEEN_EVENTS:                        # nouvelle tentative du même événement
        return "", 204
    SEEN_EVENTS.add(event["id"])
    url = event["data"]["delayed_sharing"]
    if event["name"] == "delayed_sharing.update":
        refresh_delayed(url)                              # idéalement : mise en file d'attente
    elif event["name"] == "delayed_sharing.revoke":
        forget_delayed(url)
    return "", 204                                        # répondre vite


# --- Persistance : à remplacer par votre base -----------------------------------
def save_user(sub, claims): print("user", sub, sorted(claims))
def save_document(sub, name, raw): print("document", sub, name, len(raw), "octets")
def save_claims_by_url(url, claims): print("claims", url, sorted(claims))
def forget_delayed(url): print("révoqué", url)


if __name__ == "__main__":
    app.run(port=5000)

À remplacer pour la production : les quatre fonctions de persistance, l’ensemble SEEN_EVENTS (base ou Redis avec expiration), la file de traitement du webhook, et les logs. Voir la checklist de mise en production.