OAuth 2.1 Hardening: DPoP and PAR

docker exec -it ib-keycloak-1 /opt/keycloak/bin/kcadm.sh create clients \
  -r northgate \
  -s clientId=hr-portal-hardened \
  -s enabled=true \
  -s protocol=openid-connect \
  -s publicClient=true \
  -s standardFlowEnabled=true \
  -s 'redirectUris=["http://localhost:8081/callback"]' \
  -s 'attributes."pkce.code.challenge.method"=S256'

# Capture the client UUID for later steps
docker exec -it ib-keycloak-1 /opt/keycloak/bin/kcadm.sh get \
  clients?clientId=hr-portal-hardened -r northgate --fields id | jq -r '.[0].id'

# resource_server.py
# A deliberately small resource server. In BEARER_ONLY mode it accepts any
# validly signed, unexpired access token, which is exactly the replayable
# behaviour this lab sets out to fix. In Phase 3 you flip DPOP_REQUIRED to
# True and it additionally enforces sender-constrained binding.

import os
import hashlib
import base64
import json
import time
from flask import Flask, request, jsonify
import requests
from jwcrypto import jwt, jwk, jws

app = Flask(__name__)
ISSUER = "https://ib-lb:8443/realms/northgate"
JWKS_URL = f"{ISSUER}/protocol/openid-connect/certs"
DPOP_REQUIRED = os.environ.get("DPOP_REQUIRED", "false").lower() == "true"

# Lab only: ib-lb presents a self-signed certificate from Lab 06.
_jwks = jwk.JWKSet.from_json(requests.get(JWKS_URL, verify=False).text)


def _verify_access_token(token):
    verified = jwt.JWT(jwt=token, key=_jwks, algs=["RS256"])
    return json.loads(verified.claims)


def _b64url(data):
    return base64.urlsafe_b64encode(data).rstrip(b"=").decode()


def _verify_dpop_proof(proof, method, url, access_token, expected_jkt):
    # Parse without verification first to read the embedded JWK from the header.
    header_b64 = proof.split(".")[0]
    header_b64 += "=" * (-len(header_b64) % 4)
    header = json.loads(base64.urlsafe_b64decode(header_b64))

    if header.get("typ") != "dpop+jwt":
        raise ValueError("proof typ is not dpop+jwt")

    proof_key = jwk.JWK(**header["jwk"])

    # Verify the proof signature with the key it carries.
    verified = jwt.JWT(jwt=proof, key=proof_key, algs=["ES256"])
    claims = json.loads(verified.claims)

    # The proof's key thumbprint must equal the token's bound thumbprint.
    computed_jkt = _b64url(hashlib.sha256(proof_key.thumbprint().encode()).digest()) \
        if False else proof_key.thumbprint(hashalg=hashlib.sha256)
    if computed_jkt != expected_jkt:
        raise ValueError("proof key thumbprint does not match token cnf/jkt")

    if claims.get("htm") != method:
        raise ValueError("htm mismatch")
    if claims.get("htu") != url:
        raise ValueError("htu mismatch")

    # ath binds this proof to this exact access token.
    expected_ath = _b64url(hashlib.sha256(access_token.encode()).digest())
    if claims.get("ath") != expected_ath:
        raise ValueError("ath does not match presented access token")

    if abs(time.time() - claims.get("iat", 0)) > 60:
        raise ValueError("proof iat outside acceptable window")

    return claims


@app.get("/api/payslip")
def payslip():
    auth = request.headers.get("Authorization", "")
    scheme, _, token = auth.partition(" ")

    try:
        if DPOP_REQUIRED:
            if scheme != "DPoP":
                return jsonify({"error": "DPoP scheme required"}), 401
            claims = _verify_access_token(token)
            jkt = claims.get("cnf", {}).get("jkt")
            if not jkt:
                return jsonify({"error": "token is not sender-constrained"}), 401
            proof = request.headers.get("DPoP", "")
            _verify_dpop_proof(
                proof,
                method="GET",
                url="https://localhost:8095/api/payslip",
                access_token=token,
                expected_jkt=jkt,
            )
        else:
            if scheme.lower() != "bearer":
                return jsonify({"error": "Bearer scheme required"}), 401
            claims = _verify_access_token(token)
    except Exception as e:
        return jsonify({"error": str(e)}), 401

    return jsonify({
        "user": claims.get("preferred_username"),
        "payslip": "NET PAY 3,214.55 GBP (demo)",
        "sender_constrained": DPOP_REQUIRED,
    })


if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8095, ssl_context="adhoc")

# In its own terminal, from the lab virtual environment
export DPOP_REQUIRED=false
python3.11 resource_server.py

# Obtain a token (direct grant used purely for baseline demonstration).
TOKEN=$(curl -sk -X POST \
  "https://localhost:8443/realms/northgate/protocol/openid-connect/token" \
  -d "client_id=hr-portal-hardened" \
  -d "grant_type=password" \
  -d "username=jpatel" \
  -d "password=REPLACE_WITH_JPATEL_PASSWORD" \
  -d "scope=openid profile" | jq -r '.access_token')

# "Legitimate" call
curl -sk -H "Authorization: Bearer $TOKEN" \
  https://localhost:8095/api/payslip | jq .

# The SAME token string, as if copied to an attacker's machine, replays fine.
echo "$TOKEN" > /tmp/stolen_token.txt
curl -sk -H "Authorization: Bearer $(cat /tmp/stolen_token.txt)" \
  https://localhost:8095/api/payslip | jq .

# Replace CLIENT_UUID with the value captured in Step 1.1.
docker exec -it ib-keycloak-1 /opt/keycloak/bin/kcadm.sh update \
  clients/CLIENT_UUID -r northgate \
  -s 'attributes."require.pushed.authorization.requests"=true'

# DPoP is a preview feature in some Keycloak 24.x builds. Confirm it is on.
# If your cluster was started without it, add --features=dpop to the start
# command and rebuild, following the same rolling restart pattern as Lab 08.
docker exec -it ib-keycloak-1 /opt/keycloak/bin/kcadm.sh get \
  serverinfo -r master | jq '.features[] | select(.name=="dpop")'

# Require DPoP binding on the client.
docker exec -it ib-keycloak-1 /opt/keycloak/bin/kcadm.sh update \
  clients/CLIENT_UUID -r northgate \
  -s 'attributes."dpop.bound.access.tokens"=true'

# dpop.py
# Purpose: hold the client DPoP keypair and build a DPoP proof JWT for a
# given HTTP method and URL, optionally binding to an access token (ath)
# and a server nonce.

import time
import uuid
import hashlib
import base64
import json
from jwcrypto import jwk, jwt

# Generate an EC P-256 key for ES256 proofs. In a real client this is
# created once and kept for the client's lifetime (or per session).
CLIENT_KEY = jwk.JWK.generate(kty="EC", crv="P-256")


def _b64url(data):
    return base64.urlsafe_b64encode(data).rstrip(b"=").decode()


def public_jwk():
    # Public portion only, as embedded in the proof header.
    return json.loads(CLIENT_KEY.export_public())


def key_thumbprint():
    return CLIENT_KEY.thumbprint(hashalg=hashlib.sha256)


def build_proof(method, url, access_token=None, nonce=None):
    header = {
        "typ": "dpop+jwt",
        "alg": "ES256",
        "jwk": public_jwk(),
    }
    claims = {
        "htm": method,
        "htu": url,
        "jti": str(uuid.uuid4()),
        "iat": int(time.time()),
    }
    if access_token is not None:
        claims["ath"] = _b64url(hashlib.sha256(access_token.encode()).digest())
    if nonce is not None:
        claims["nonce"] = nonce

    proof = jwt.JWT(header=header, claims=claims)
    proof.make_signed_token(CLIENT_KEY)
    return proof.serialize()

# hardened_client.py
# Purpose: run the OAuth 2.1 hardened flow end to end. For lab clarity this
# uses the authorization code flow with a manual browser step for the login,
# then focuses on the machine-to-machine hardening: PAR, DPoP and the nonce
# retry loop on the token endpoint.

import base64
import hashlib
import os
import secrets
import sys
import requests
import dpop

BASE = "https://localhost:8443/realms/northgate/protocol/openid-connect"
PAR_URL = f"{BASE}/ext/par/request"
AUTH_URL = f"{BASE}/auth"
TOKEN_URL = f"{BASE}/token"
CLIENT_ID = "hr-portal-hardened"
REDIRECT_URI = "http://localhost:8081/callback"
requests.packages.urllib3.disable_warnings()  # lab: self-signed ib-lb cert


def pkce_pair():
    verifier = base64.urlsafe_b64encode(secrets.token_bytes(32)).rstrip(b"=").decode()
    challenge = base64.urlsafe_b64encode(
        hashlib.sha256(verifier.encode()).digest()
    ).rstrip(b"=").decode()
    return verifier, challenge


def push_par(challenge):
    resp = requests.post(PAR_URL, data={
        "client_id": CLIENT_ID,
        "response_type": "code",
        "redirect_uri": REDIRECT_URI,
        "scope": "openid profile",
        "code_challenge": challenge,
        "code_challenge_method": "S256",
    }, verify=False)
    resp.raise_for_status()
    return resp.json()["request_uri"]


def token_with_dpop(code, verifier):
    """Exchanges the code, attaching a DPoP proof. Handles the server nonce
    challenge: if Keycloak responds use_dpop_nonce, retry once with the
    supplied nonce embedded in a fresh proof."""
    def attempt(nonce=None):
        proof = dpop.build_proof("POST", TOKEN_URL, nonce=nonce)
        return requests.post(TOKEN_URL, headers={"DPoP": proof}, data={
            "client_id": CLIENT_ID,
            "grant_type": "authorization_code",
            "code": code,
            "redirect_uri": REDIRECT_URI,
            "code_verifier": verifier,
        }, verify=False)

    resp = attempt()
    if resp.status_code == 400 and resp.json().get("error") == "use_dpop_nonce":
        nonce = resp.headers.get("DPoP-Nonce")
        resp = attempt(nonce=nonce)
    resp.raise_for_status()
    return resp.json()


if __name__ == "__main__":
    verifier, challenge = pkce_pair()
    request_uri = push_par(challenge)
    print("PAR pushed. Opaque request_uri returned:")
    print("  ", request_uri)
    print("\nOpen this URL in a browser, log in as jpatel, and copy the")
    print("'code' parameter from the redirect back here:")
    print(f"  {AUTH_URL}?client_id={CLIENT_ID}&request_uri={request_uri}")

    code = input("\nPaste the authorization code: ").strip()
    tokens = token_with_dpop(code, verifier)
    with open("dpop_access_token.txt", "w") as f:
        f.write(tokens["access_token"])
    print("\nToken type returned:", tokens.get("token_type"))
    print("Saved sender-constrained access token to dpop_access_token.txt")

python3.11 hardened_client.py

# In the resource server terminal, stop it (Ctrl+C) and restart enforcing DPoP
export DPOP_REQUIRED=true
python3.11 resource_server.py

# call_api.py: legitimate holder builds a fresh proof bound to the token
python3.11 - <<'PYEOF'
import requests, dpop
requests.packages.urllib3.disable_warnings()
API = "https://localhost:8095/api/payslip"
token = open("dpop_access_token.txt").read().strip()
proof = dpop.build_proof("GET", API, access_token=token)
r = requests.get(API, headers={
    "Authorization": f"DPoP {token}",
    "DPoP": proof,
}, verify=False)
print("Legitimate DPoP call:", r.status_code, r.json())
PYEOF

# Attacker replay: has the token string but NOT dpop.CLIENT_KEY.
# Simulate by trying a bearer replay and a forged proof from a different key.
python3.11 - <<'PYEOF'
import requests, json
from jwcrypto import jwk, jwt
import base64, hashlib, time, uuid
requests.packages.urllib3.disable_warnings()
API = "https://localhost:8095/api/payslip"
token = open("dpop_access_token.txt").read().strip()

# Attempt 1: plain bearer replay of the stolen token.
r1 = requests.get(API, headers={"Authorization": f"Bearer {token}"}, verify=False)
print("Stolen token, bearer replay:", r1.status_code, r1.json())

# Attempt 2: attacker forges a proof with THEIR OWN key (they lack the real one).
attacker_key = jwk.JWK.generate(kty="EC", crv="P-256")
def b64url(d): return base64.urlsafe_b64encode(d).rstrip(b"=").decode()
header = {"typ":"dpop+jwt","alg":"ES256","jwk":json.loads(attacker_key.export_public())}
claims = {"htm":"GET","htu":API,"jti":str(uuid.uuid4()),"iat":int(time.time()),
          "ath": b64url(hashlib.sha256(token.encode()).digest())}
p = jwt.JWT(header=header, claims=claims); p.make_signed_token(attacker_key)
r2 = requests.get(API, headers={"Authorization": f"DPoP {token}", "DPoP": p.serialize()}, verify=False)
print("Stolen token, forged proof:", r2.status_code, r2.json())
PYEOF