OAuth and Token-Based Authorization

Who May Do What

Authentication answers who someone is. Authorization answers what they may do. OAuth is the standard way to hand an application a limited, revocable permission to act on an API, without giving it a password. This lesson builds each piece small enough to see, with nothing but PyJWT and the cryptography package (pip install pyjwt cryptography). Every example runs offline.

• Core concepts: the three roles, scopes and audiences, short-lived access tokens with rotated refresh tokens, and PKCE.
• Hardening: why bearer tokens leak, and what DPoP changes.

To see these pieces in a real protocol, with a real identity provider, read MCP Authorization next.

Core Concepts

The three roles in one file

All the examples share this helper. issue plays the authorization server: it alone holds the signing key. check plays the resource server, the API: it only has the public key.

# authz.py - a toy authorization server and resource-server check, shared by the examples
import time
import jwt
from cryptography.hazmat.primitives.asymmetric import ec

ISSUER = "https://auth.example"
SIGNING_KEY = ec.generate_private_key(ec.SECP256R1())  # only the authorization server holds this
PUBLIC_KEY = SIGNING_KEY.public_key()                    # resource servers get this one


def issue(client, scope, audience, ttl=300, **extra):
    now = int(time.time())
    claims = {"iss": ISSUER, "sub": client, "aud": audience, "scope": scope, "iat": now, "exp": now + ttl}
    return jwt.encode(claims | extra, SIGNING_KEY, "ES256")


def check(token, audience, needed_scope):
    """What a resource server does with every request. Returns 'ok' or why it refused."""
    try:
        claims = jwt.decode(token, PUBLIC_KEY, algorithms=["ES256"], issuer=ISSUER, audience=audience)
    except jwt.PyJWTError as e:
        return f"401 {type(e).__name__}"
    return "ok" if needed_scope in claims["scope"].split() else f"403 missing scope {needed_scope!r}"

# roles.py - the three parties of OAuth, and what's inside the token they pass around
import jwt
from authz import check, issue

# 1. The client asks the authorization server for a token for one API, with one scope.
token = issue(client="reports-app", scope="invoices:read", audience="https://api.example/invoices")
print("client got a token:", token[:24] + "...")

# 2. Anyone can read the claims; only the signature makes them trustworthy.
print("claims:", {k: v for k, v in jwt.decode(token, options={"verify_signature": False}).items()
                  if k in ("iss", "sub", "aud", "scope")})

# 3. The resource server checks it on every request.
print("read invoices: ", check(token, "https://api.example/invoices", "invoices:read"))
print("pay an invoice:", check(token, "https://api.example/invoices", "invoices:pay"))
print("same token, payroll API:", check(token, "https://api.example/payroll", "invoices:read"))

$ python roles.py
client got a token: eyJhbGciOiJFUzI1NiIsInR5...
claims: {'iss': 'https://auth.example', 'sub': 'reports-app', 'aud': 'https://api.example/invoices', 'scope': 'invoices:read'}
read invoices:  ok
pay an invoice: 403 missing scope 'invoices:pay'
same token, payroll API: 401 InvalidAudienceError

The client (reports-app) got a token for one API and one scope, and three things happened at the API. Reading invoices worked. Paying one was refused with 403, because the token is genuine but its scope doesn’t allow it. The same token sent to the payroll API was refused with 401, because its audience names the invoices API. Anyone can read the claims, since a JWT is only base64; the signature is what makes them trustworthy. Never put secrets in a token’s claims.

Short access tokens, rotated refresh tokens

# refresh.py - short access tokens, rotated refresh tokens, and catching a stolen one
import secrets, time
from authz import check, issue

API = "https://api.example/invoices"
live, used = {}, set()  # the authorization server's refresh-token records


def grant(client):
    refresh = secrets.token_urlsafe(8)
    live[refresh] = client
    return issue(client, "invoices:read", API, ttl=1), refresh


def use_refresh(refresh):
    if refresh in used:  # an old token came back: someone has a copy
        live.clear()
        return "reuse detected: every refresh token for this client revoked"
    used.add(refresh)
    return grant(live.pop(refresh))


if __name__ == "__main__":
    access, refresh = grant("reports-app")
    time.sleep(2)
    print("after 2 s:", check(access, API, "invoices:read"))
    access, new_refresh = use_refresh(refresh)
    print("refreshed:", check(access, API, "invoices:read"))
    print("attacker replays the first refresh token:", use_refresh(refresh))
    print("legitimate client refreshes:", "revoked" if new_refresh not in live else "still valid")

$ python refresh.py
after 2 s: 401 ExpiredSignatureError
refreshed: ok
attacker replays the first refresh token: reuse detected: every refresh token for this client revoked
legitimate client refreshes: revoked

The access token lived one second, so after two seconds it was refused. The client didn’t ask the user again: it swapped its refresh token for a new pair. The authorization server rotates refresh tokens, issuing a new one on every use and remembering the old one. When the old one came back, it knew someone had a copy, but not whether it was the client or a thief, so it revoked every refresh token for that client. The real user logs in once more, and the attacker can’t refresh again. An access token already issued keeps working until it expires, which is the other reason to keep them short.

PKCE: protecting the authorization code

In the browser flow, the authorization server sends the client a short-lived code through a redirect, and the client exchanges it for tokens. Redirects can be intercepted.

# pkce.py - a stolen authorization code is useless without the code verifier
import base64, hashlib, secrets


def challenge(verifier):  # S256: base64url(sha256(verifier)), no padding
    return base64.urlsafe_b64encode(hashlib.sha256(verifier.encode()).digest()).rstrip(b"=").decode()


# The client makes a secret verifier and sends only its hash with the authorization request.
verifier = secrets.token_urlsafe(32)
codes = {"code-123": challenge(verifier)}  # the authorization server remembers the challenge


def redeem(code, presented_verifier):
    if code not in codes:
        return "invalid_grant: unknown or already used code"
    if challenge(presented_verifier) != codes[code]:
        return "invalid_grant: verifier does not match"
    del codes[code]  # a code works once
    return "token issued"


print("attacker with the stolen code:", redeem("code-123", secrets.token_urlsafe(32)))
print("real client with its verifier:", redeem("code-123", verifier))
print("the same code again:          ", redeem("code-123", verifier))

$ python pkce.py
attacker with the stolen code: invalid_grant: verifier does not match
real client with its verifier: token issued
the same code again:           invalid_grant: unknown or already used code

The client sent only the hash of its verifier at the start, and the verifier itself when redeeming. An attacker who intercepts the code doesn’t have the verifier, so it’s useless to them. Codes also work exactly once. OAuth 2.1 requires PKCE in the authorization code flow, including for clients that have a client secret.

Hardening: Beyond Bearer Tokens

Bearer means “whoever holds it”

# bearer.py - a bearer token works for whoever holds it
from authz import check, issue

API = "https://api.example/invoices"
token = issue("reports-app", "invoices:read", API)

log_line = f'GET /invoices Authorization="Bearer {token}"'  # a token that ended up in a log
stolen = log_line.split("Bearer ")[1].rstrip('"')

print("the real client:  ", check(token, API, "invoices:read"))
print("whoever read the log:", check(stolen, API, "invoices:read"))

$ python bearer.py
the real client:   ok
whoever read the log: ok

Nothing in a bearer token ties it to the client it was issued to. Tokens end up in logs, crash reports, browser history and proxies, and each copy works until it expires. Short lifetimes and audiences limit the damage. They don’t stop a thief inside those limits.

DPoP: binding the token to a key

With DPoP (RFC 9449) the client has its own key pair. The authorization server puts the public key’s thumbprint in the token (cnf.jkt), and every request carries a proof: a small JWT signed with the private key, naming the HTTP method and URL, with a unique jti. The resource server checks all of it:

# dpop_check.py - what a resource server checks for a DPoP-bound token
import base64, hashlib, json, time
import jwt
from authz import ISSUER, PUBLIC_KEY

AUDIENCE = "https://api.example/invoices"
seen_jti = set()


def thumbprint(public_key):  # RFC 7638 JWK thumbprint, base64url(sha256(canonical JWK))
    jwk = jwt.algorithms.ECAlgorithm.to_jwk(public_key, as_dict=True)
    canonical = json.dumps({k: jwk[k] for k in ("crv", "kty", "x", "y")}, separators=(",", ":"))
    return base64.urlsafe_b64encode(hashlib.sha256(canonical.encode()).digest()).rstrip(b"=").decode()


def check_dpop(token, proof, method, url):
    claims = jwt.decode(token, PUBLIC_KEY, algorithms=["ES256"], issuer=ISSUER, audience=AUDIENCE)
    header = jwt.get_unverified_header(proof)
    proof_key = jwt.algorithms.ECAlgorithm.from_jwk(header["jwk"])
    p = jwt.decode(proof, proof_key, algorithms=["ES256"])  # signed by the key in its own header
    if thumbprint(proof_key) != claims["cnf"]["jkt"]:
        return "401 proof key is not the key the token is bound to"
    if (p["htm"], p["htu"]) != (method, url):
        return "401 proof is for another request"
    if abs(time.time() - p["iat"]) > 60:
        return "401 proof is too old or from the future"
    if p["jti"] in seen_jti:
        return "401 proof already used"
    seen_jti.add(p["jti"])
    return "ok"

# dpop.py - DPoP (RFC 9449): a token bound to a key, plus a fresh proof on every request
import time, uuid
import jwt
from cryptography.hazmat.primitives.asymmetric import ec
from authz import issue
from dpop_check import check_dpop, thumbprint

API = "https://api.example/invoices"
client_key = ec.generate_private_key(ec.SECP256R1())  # never leaves the client
token = issue("reports-app", "invoices:read", API, cnf={"jkt": thumbprint(client_key.public_key())})


def proof(key, method, url):  # a short-lived JWT signed with the client's key, naming this request
    jwk = jwt.algorithms.ECAlgorithm.to_jwk(key.public_key(), as_dict=True)
    claims = {"htm": method, "htu": url, "iat": int(time.time()), "jti": str(uuid.uuid4())}
    return jwt.encode(claims, key, "ES256", headers={"typ": "dpop+jwt", "jwk": jwk})


good = proof(client_key, "GET", API)
print("client with its key:        ", check_dpop(token, good, "GET", API))
print("same proof sent again:      ", check_dpop(token, good, "GET", API))
attacker_key = ec.generate_private_key(ec.SECP256R1())
print("stolen token, attacker's key:", check_dpop(token, proof(attacker_key, "GET", API), "GET", API))
read_proof = proof(client_key, "GET", API)
print("proof replayed for a payment:", check_dpop(token, read_proof, "POST", API + "/7/pay"))

$ python dpop.py
client with its key:         ok
same proof sent again:       401 proof already used
stolen token, attacker's key: 401 proof key is not the key the token is bound to
proof replayed for a payment: 401 proof is for another request

Each refusal stops a different attack. A proof can’t be sent twice (jti). A stolen token is useless without the client’s private key, because the attacker’s proof is signed by the wrong key. A captured proof can’t authorize a different request, because it names one method and one URL. Real implementations also keep the jti cache for the proof’s lifetime only, and bind the proof to the token with an ath claim, a hash of the access token. Support varies: Keycloak advertises DPoP, for example, while the MCP Python SDK doesn’t support it yet, as MCP Authorization shows.

Practice

1. Forget the audience. Write a payroll API check that verifies the signature and issuer but switches off the audience check, and send it the invoice token.

# op_1.py - a payroll API that forgot to check the audience
import jwt
from authz import ISSUER, PUBLIC_KEY, issue

invoice_token = issue("reports-app", "invoices:read", "https://api.example/invoices")


def payroll_check(token):  # the bug: verify_aud switched off
    claims = jwt.decode(token, PUBLIC_KEY, algorithms=["ES256"], issuer=ISSUER, options={"verify_aud": False})
    return f"ok, serving payroll to {claims['sub']}"


print("payroll API, invoice token:", payroll_check(invoice_token))

$ python op_1.py
payroll API, invoice token: ok, serving payroll to reports-app

2. The thief goes first. In refresh.py the real client refreshed first. Swap the order: the attacker uses the stolen refresh token, then the real client tries.

# op_2.py - the attacker refreshes first; the real client's refresh is now the "reuse"
from refresh import grant, live, use_refresh

access, refresh = grant("reports-app")
stolen = use_refresh(refresh)
print("attacker uses the stolen refresh token:", "got new tokens" if isinstance(stolen, tuple) else stolen)
print("real client uses the same one later:   ", use_refresh(refresh))
print("refresh tokens still valid:", len(live))

$ python op_2.py
attacker uses the stolen refresh token: got new tokens
real client uses the same one later:    reuse detected: every refresh token for this client revoked
refresh tokens still valid: 0

3. An old proof. Make a DPoP proof that is correct in every way, but with an iat two minutes in the past.

# op_3.py - a DPoP proof made two minutes ago is refused, even for the right request
import time, uuid
import jwt
from cryptography.hazmat.primitives.asymmetric import ec
from authz import issue
from dpop_check import check_dpop, thumbprint

API = "https://api.example/invoices"
key = ec.generate_private_key(ec.SECP256R1())
token = issue("reports-app", "invoices:read", API, cnf={"jkt": thumbprint(key.public_key())})
jwk = jwt.algorithms.ECAlgorithm.to_jwk(key.public_key(), as_dict=True)
old = {"htm": "GET", "htu": API, "iat": int(time.time()) - 120, "jti": str(uuid.uuid4())}
proof = jwt.encode(old, key, "ES256", headers={"typ": "dpop+jwt", "jwk": jwk})
print("two-minute-old proof:", check_dpop(token, proof, "GET", API))

$ python op_3.py
two-minute-old proof: 401 proof is too old or from the future