Article 5

Deploy API Auth Access Tokens with ML-DSA-65 on Cloudflare Workers (Python)

Published: 2026-07-11

This Python version mirrors the Rust flow: issue access tokens with POST /token and verify them on GET /protected using ML-DSA-65 signatures.

1) Create a Python Worker

npm create cloudflare@latest pqc-auth-worker-py -- --type=python
cd pqc-auth-worker-py

2) requirements.txt

mldsa==0.2.1

3) src/worker.py (full API auth implementation)

import base64
import json
import secrets
import time

from workers import Response
from mldsa import MLDSA65PrivateKey, MLDSA65PublicKey, MLDSA65Signature


def b64url_encode(data: bytes) -> str:
    return base64.urlsafe_b64encode(data).rstrip(b"=").decode("ascii")


def b64url_decode(value: str) -> bytes:
    pad = "=" * (-len(value) % 4)
    return base64.urlsafe_b64decode((value + pad).encode("ascii"))


def json_response(payload: dict, status: int = 200) -> Response:
    return Response(
        json.dumps(payload),
        status=status,
        headers={"content-type": "application/json; charset=utf-8"},
    )


def load_signing_key(env) -> MLDSA65PrivateKey:
    raw = b64url_decode(env.get("ML_DSA65_SIGNING_KEY_B64"))
    return MLDSA65PrivateKey.from_bytes(raw)


def load_verifying_key(env) -> MLDSA65PublicKey:
    raw = b64url_decode(env.get("ML_DSA65_VERIFYING_KEY_B64"))
    return MLDSA65PublicKey.from_bytes(raw)


def issue_access_token(signing_key: MLDSA65PrivateKey, claims: dict) -> str:
    payload = json.dumps(claims, separators=(",", ":")).encode("utf-8")
    signature = signing_key.sign(payload)
    return f"{b64url_encode(payload)}.{b64url_encode(signature.to_bytes())}"


def verify_access_token(verifying_key: MLDSA65PublicKey, token: str) -> dict:
    parts = token.split(".")
    if len(parts) != 2:
        raise ValueError("token must contain payload and signature")

    payload = b64url_decode(parts[0])
    signature = MLDSA65Signature.from_bytes(b64url_decode(parts[1]))

    if not verifying_key.verify(payload, signature):
        raise ValueError("signature verification failed")

    claims = json.loads(payload.decode("utf-8"))
    if int(time.time()) > int(claims["exp"]):
        raise ValueError("token expired")

    return claims


def bearer_token(request) -> str:
    auth = request.headers.get("Authorization")
    if not auth or not auth.startswith("Bearer "):
        raise ValueError("missing/invalid Authorization header")
    return auth[7:].strip()


async def on_fetch(request, env):
    url = request.url
    path = url.pathname

    if request.method == "POST" and path == "/token":
        try:
            body = await request.json()
            ttl = int(body.get("ttl_seconds", 900))
            ttl = max(60, min(ttl, 3600))
            now = int(time.time())

            claims = {
                "sub": body["sub"],
                "scope": body["scope"],
                "iat": now,
                "exp": now + ttl,
                "nonce": secrets.token_hex(16),
            }

            signing_key = load_signing_key(env)
            token = issue_access_token(signing_key, claims)

            return json_response(
                {
                    "access_token": token,
                    "token_type": "Bearer",
                    "expires_in": ttl,
                }
            )
        except Exception as err:
            return json_response({"ok": False, "error": str(err)}, status=400)

    if request.method == "GET" and path == "/protected":
        try:
            token = bearer_token(request)
            verifying_key = load_verifying_key(env)
            claims = verify_access_token(verifying_key, token)
            return json_response(
                {
                    "ok": True,
                    "message": "Access granted",
                    "subject": claims["sub"],
                    "scope": claims["scope"],
                    "exp": claims["exp"],
                }
            )
        except Exception as err:
            return json_response({"ok": False, "error": str(err)}, status=401)

    return json_response({"ok": False, "error": "not found"}, status=404)

4) wrangler.toml

name = "pqc-auth-worker-py"
main = "src/worker.py"
compatibility_date = "2026-07-01"
compatibility_flags = ["python_workers"]

5) Set ML-DSA-65 secrets

npx wrangler secret put ML_DSA65_SIGNING_KEY_B64
npx wrangler secret put ML_DSA65_VERIFYING_KEY_B64

6) Deploy

npx wrangler deploy

7) Test token issue + protected API

# Issue token
curl -s -X POST "https://<your-worker>.workers.dev/token" \
  -H "content-type: application/json" \
  -d '{"sub":"user-123","scope":"read:reports","ttl_seconds":900}'

# Call protected endpoint
curl -s "https://<your-worker>.workers.dev/protected" \
  -H "Authorization: Bearer <access_token>"

Production notes

Keep token TTL short, enforce issuer/audience checks, rotate keys with overlap windows, and use KV/Durable Objects for nonce replay tracking on high-risk endpoints.

Back to all articles