Documentation/Getting started/End-to-end example

End-to-end example

A complete journey with every call: from the redirect to the webhook, and a Python program ready to adapt.

This example follows a concrete case: Acme Insurance wants to onboard a customer with their verified identity, identity document and email. The social security number is not yet validated on the TrustMe side: Acme therefore uses delayed sharing.

Values (identifiers, tokens, dates) are fictitious and responses are indicative excerpts; see the reference for field details.

Item	Value
Requested scopes	`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 used in the examples (never commit the secret)
export A3BC_CLIENT_ID=e251305f-5db8-11f1-8cb7-c8a3620c309f
export A3BC_CLIENT_SECRET=<your secret>
export ISSUER=https://oidc-sandbox.a3bc.io

Step 1: read the 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": "…"
}

Excerpt: the document has other fields. Cache it rather than reading it on every sign-in.

Step 2: redirect the user

When the customer clicks “Sign in with TrustMe”, your server generates state and nonce, keeps them in session, and answers with an HTTP 302 redirect to:

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

Step 3: the user consents in TrustMe

A3BC has them open the TrustMe app, by entering their identifier (phone) or scanning a QR code. They authenticate there and see the list of what Acme requests. They choose what to share, including documents still being validated (thanks to private:delayed_sharing). Nothing is visible to your server during this step.

Step 4: the callback on your site

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

Compare state with the session one. If it differs, or if the URL contains error=…, stop. See the errors.

Step 5: exchange the 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"
}

The access_token is an opaque string, the id_token a JWT. See Tokens and lifetimes.

Step 6: verify and read the id_token

After verifying the signature, iss, aud, exp and nonce, the content is:

{
  "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"
}

Read it as follows:

identity and email claims are present and validated;
private:doc.id is a URL to call (step 8);
the private:ssn* claims are present but empty: the social security number is waiting for validation, and private:delayed_sharing gives the URL to retrieve it later (step 9).

Store now everything that is non-empty. See Reading claims.

Step 7: userinfo (optional)

The same information is available with the access_token:

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

It is useful to compare with the id_token; it is not needed if you already verified the latter.

Step 8: fetch the 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…" }
  ]
}

Decode each content (base64) and compare the digest with private:doc.id.hash:

echo "$CONTENT" | base64 -d > front.png
sha256sum front.png     # must match the claim entry, without the "sha256:" prefix

A few hours later, TrustMe validates the social security number. A3BC sends you a 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"]
  }
}

After verifying the signature (see Webhooks) and answering 204, your server requests a client_credentials token, then calls the data.delayed_sharing URL:

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 is empty: no more information will be added. Store the non-empty values: the user could delete the credential afterwards, or revoke their consent (delayed_sharing.revoke event, then HTTP 403 code 00020).

The complete program

This Flask program performs all the steps above: /login, /callback (exchange, id_token verification, document with digest check, delayed sharing) and /webhooks/a3bc. It is tested against a fake A3BC server (state, nonce, signature, replayed webhook, wrong signature).

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

"""Minimal Relying Party: "Sign in with TrustMe" login, callback and 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"]          # e.g. 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"]          # signs the session cookie

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()                                       # replace with Redis/DB (with TTL)


# --- 1. Redirect the user ------------------------------------------------------
@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. Receive the callback, exchange the code, verify the id_token -----------
@app.get("/callback")
def callback():
    if "error" in request.args:                           # consent refused, ACR not satisfied...
        return f"Sign-in refused: {request.args.get('error_description') or request.args['error']}", 400
    if not secrets.compare_digest(request.args.get("state", ""), session.pop("state", "!")):
        abort(400, "invalid state")
    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"code exchange refused: {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, "invalid nonce")

    # Store what you need RIGHT NOW: access is only guaranteed for 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, "document digest differs from the claim")
            save_document(claims["sub"], doc["name"], raw)

    if claims.get("private:delayed_sharing"):
        refresh_delayed(claims["private:delayed_sharing"])   # some claims may still be missing
    return f"Welcome {claims.get('given_name', '')}"


# --- 3. Delayed sharing ----------------------------------------------------------
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: request a new one each time


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: retry later; 00020/00021/00022: business rule, no 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())                # the key is base64: decode it first
    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()                              # raw bytes, before any JSON parsing
    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:                        # retry of the same event
        return "", 204
    SEEN_EVENTS.add(event["id"])
    url = event["data"]["delayed_sharing"]
    if event["name"] == "delayed_sharing.update":
        refresh_delayed(url)                              # ideally: put on a queue
    elif event["name"] == "delayed_sharing.revoke":
        forget_delayed(url)
    return "", 204                                        # answer fast


# --- Persistence: replace with your database -------------------------------------
def save_user(sub, claims): print("user", sub, sorted(claims))
def save_document(sub, name, raw): print("document", sub, name, len(raw), "bytes")
def save_claims_by_url(url, claims): print("claims", url, sorted(claims))
def forget_delayed(url): print("revoked", url)


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

To replace for production: the four persistence functions, the SEEN_EVENTS set (database or Redis with expiry), the webhook processing queue, and logging. See the production checklist.