Zero Trust Architecture With Http Dpop Proof Binding For Ci Ephemeral Tokens

I fell into this rabbit hole because I kept seeing the same annoying pattern in my homelab: my CI pipeline would mint short-lived API tokens, those tokens would get used successfully, and then—without any “real” breach—I’d still be able to replay requests by copying the bearer token into a different machine. That’s the moment it clicked: Zero Trust isn’t just “short-lived tokens.” It’s also about binding proof of the caller to where and how the request was made.

So I built a small, working prototype: a Zero Trust-ish “request authentication layer” where the client sends an DPoP proof—DPoP stands for Demonstration of Proof of Possession, meaning the client proves it controls a private key associated with the token. The server verifies that proof, and also checks that the proof is bound to the specific request details (method + path). That prevents a stolen bearer token from being trivially replayed elsewhere.

This is intentionally niche: DPoP proof binding for CI ephemeral tokens, enforced at the edge using a minimal Flask middleware.

---

The problem: short-lived bearer tokens still replay

In the simplest setup, a CI job gets a bearer token:

• Authorization header: Authorization: Bearer <token>
• Server checks token signature/expiry
• Server accepts the request

Even if the token expires quickly, an attacker who captures it can replay the HTTP request during the token’s lifetime. In my tests, replaying the exact request from a different host worked.

What I wanted instead: proof that the caller is the holder of a private key, not just someone who has a token.

---

What I built: DPoP-bound tokens enforced on the server

Key idea

Each request includes:

1. Authorization: DPoP <access_token>
2. DPoP: <jwt_proof>

The DPoP proof is signed by a private key held by the CI job. The server verifies:

• the DPoP JWT signature using the public key embedded in the proof (jwk)
• that the proof htu (HTTP URL hash) and htm (HTTP method) match the actual request
• that the proof includes a unique jti (JWT ID) which the server rejects after first use (simple replay defense)

This turns “token replay” into “token replay plus proof-of-private-key”, which is much harder.

---

Step 1: Generate a keypair for the CI job

In a real CI environment, you’d generate this key per job (ephemeral). Here’s a tiny script that generates an EC keypair and produces a JWK for the DPoP proof.

# gen_keypair.py
import base64
import json
from cryptography.hazmat.primitives.asymmetric import ec
from cryptography.hazmat.primitives import serialization

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

private_key = ec.generate_private_key(ec.SECP256R1())
public_key = private_key.public_key()

numbers = public_key.public_numbers()

jwk = {
    "kty": "EC",
    "crv": "P-256",
    "x": b64url(numbers.x.to_bytes(32, "big")),
    "y": b64url(numbers.y.to_bytes(32, "big")),
}

pem_private = private_key.private_bytes(
    encoding=serialization.Encoding.PEM,
    format=serialization.PrivateFormat.PKCS8,
    encryption_algorithm=serialization.NoEncryption(),
)

pem_public = public_key.public_bytes(
    encoding=serialization.Encoding.PEM,
    format=serialization.PublicFormat.SubjectPublicKeyInfo,
)

print("JWK:", json.dumps(jwk))
print("PRIVATE_PEM:", pem_private.decode())
print("PUBLIC_PEM:", pem_public.decode())

Why this matters: DPoP needs the server to learn the public key from the proof itself (or from a registry). For a PoC, embedding the public key in the JWT proof is the simplest path.

---

Step 2: Mint an access token bound to DPoP (server-side expectation)

I used a short-lived HMAC-signed “access token” to keep the demo focused (in production you’d use OAuth 2.0 / OIDC and proper key management). The only important part is that the server expects the Authorization scheme to be DPoP (not Bearer) and treats the token as “requires DPoP proof.”

Here’s a simple token minting endpoint.

# server.py (token endpoint)
from flask import Flask, request, jsonify
import time
import jwt

app = Flask(__name__)

ACCESS_TOKEN_SECRET = "change-me"
DPoP_PROOF_AUD = "urn:demo-dpop"  # demo audience

@app.post("/mint")
def mint():
    # CI job would authenticate using its own mechanism here.
    # For the demo, we skip it.
    payload = {
        "iss": "demo-ci",
        "aud": "demo-api",
        "sub": "ci-job-123",
        "exp": int(time.time()) + 60,  # 60s lifetime
        "iat": int(time.time()),
        "dpop": True,  # marker: server should expect DPoP proof
    }
    token = jwt.encode(payload, ACCESS_TOKEN_SECRET, algorithm="HS256")
    return jsonify({"access_token": token, "token_type": "DPoP", "aud": "demo-api"})

Why the marker: it’s a cheap way to force “DPoP-required” semantics without implementing the full OAuth stack.

---

Step 3: Verify DPoP proofs bound to method + path + anti-replay

The heart of the Zero Trust enforcement is the middleware that runs before your handler.

DPoP claims I verify

A DPoP proof JWT commonly includes:

• htm: HTTP method (e.g., POST)
• htu: HTTP URL (or hash representation; here I used the full URL string for simplicity)
• jti: unique proof id (used for replay detection)
• iat: issued-at time
• aud: audience (server)
• typ: token type (not strictly required in this demo)
• jwk: the public key parameters for EC verification
• signature: verified with that jwk

Anti-replay

I keep an in-memory set of used jti values for a short window.

Note: for a real system, you’d store jti in Redis or another shared datastore across instances.

# server.py (full example)
from flask import Flask, request, jsonify
import time
import jwt
from cryptography.hazmat.primitives.asymmetric import ec
from cryptography.hazmat.primitives import serialization

app = Flask(__name__)

ACCESS_TOKEN_SECRET = "change-me"
DPoP_PROOF_AUD = "urn:demo-dpop"
used_jti = {}  # jti -> exp time

def b64url_to_int(s: str) -> int:
    import base64
    data = base64.urlsafe_b64decode(s + "===")  # tolerant padding
    return int.from_bytes(data, "big")

def jwk_to_public_key(jwk: dict):
    if jwk["kty"] != "EC" or jwk["crv"] != "P-256":
        raise ValueError("Only P-256 supported in this demo")

    x = b64url_to_int(jwk["x"])
    y = b64url_to_int(jwk["y"])
    public_numbers = ec.EllipticCurvePublicNumbers(x, y, ec.SECP256R1())
    return public_numbers.public_key()

def cleanup_used_jti(now: int):
    expired = [jti for jti, exp in used_jti.items() if exp <= now]
    for jti in expired:
        used_jti.pop(jti, None)

@app.before_request
def dpop_protect():
    # Protect only the API endpoint(s) for demo.
    if request.path != "/api/data":
        return

    authz = request.headers.get("Authorization", "")
    if not authz.startswith("DPoP "):
        return jsonify({"error": "missing_or_invalid_authorization_scheme"}), 401

    access_token = authz.split(" ", 1)[1].strip()
    dpop_jwt = request.headers.get("DPoP")

    if not dpop_jwt:
        return jsonify({"error": "missing_dpop_proof"}), 401

    # 1) Validate access token (signature + expiry + aud/sub)
    try:
        token = jwt.decode(
            access_token,
            ACCESS_TOKEN_SECRET,
            algorithms=["HS256"],
            audience="demo-api",
            issuer="demo-ci",
        )
    except Exception as e:
        return jsonify({"error": "invalid_access_token", "detail": str(e)}), 401

    if not token.get("dpop"):
        return jsonify({"error": "dpop_required"}), 401

    # 2) Validate DPoP proof JWT
    try:
        proof = jwt.decode(dpop_jwt, options={"verify_signature": False})
    except Exception as e:
        return jsonify({"error": "invalid_dpop_jwt", "detail": str(e)}), 401

    jwk = proof.get("jwk")
    if not jwk:
        return jsonify({"error": "missing_jwk_in_dpop_proof"}), 401

    # method + url binding
    htm = proof.get("htm")
    htu = proof.get("htu")
    aud = proof.get("aud")
    jti = proof.get("jti")
    iat = proof.get("iat")

    if htm != request.method:
        return jsonify({"error": "htm_mismatch"}), 401

    # Build the URL string the client is expected to hash/bind to.
    # In this demo, I compare full URL including scheme/host as received.
    expected_htu = request.url
    if htu != expected_htu:
        return jsonify({"error": "htu_mismatch", "expected": expected_htu, "got": htu}), 401

    if aud != DPoP_PROOF_AUD:
        return jsonify({"error": "aud_mismatch"}), 401

    if not jti or not isinstance(jti, str):
        return jsonify({"error": "missing_jti"}), 401

    now = int(time.time())
    cleanup_used_jti(now)

    # basic iat freshness window
    if not iat or abs(now - int(iat)) > 60:
        return jsonify({"error": "iat_out_of_range"}), 401

    # replay detection
    if jti in used_jti:
        return jsonify({"error": "replay_detected"}), 401

    # verify signature with public key from jwk
    try:
        public_key = jwk_to_public_key(jwk)
        pem_public = public_key.public_bytes(
            encoding=serialization.Encoding.PEM,
            format=serialization.PublicFormat.SubjectPublicKeyInfo,
        )
        # PyJWT wants a key object or PEM
        jwt.decode(
            dpop_jwt,
            pem_public,
            algorithms=["ES256"],
            options={"verify_aud": False},  # we do aud manually above
        )
    except Exception as e:
        return jsonify({"error": "dpop_signature_invalid", "detail": str(e)}), 401

    # mark jti as used for a short TTL
    used_jti[jti] = now + 120

@app.post("/api/data")
def api_data():
    return jsonify({"ok": True, "message": "Zero Trust DPoP proof accepted."})

if __name__ == "__main__":
    # For local demo only.
    app.run(host="127.0.0.1", port=5000, debug=False)

Why this works (in practice):

• If someone copies the access token, they still need the private key to sign a fresh DPoP proof.
• If they capture a DPoP proof and replay it, jti gets rejected.
• If they alter the method/path, htm/htu mismatches and the server denies it.

---

Step 4: Client code that sends a DPoP proof for the request

Now I created a client that:

1. Requests an access token from /mint
2. Builds a DPoP proof JWT for POST /api/data including htm and htu
3. Signs it with the CI job private key
4. Sends the request with both headers

# client.py
import time
import uuid
import jwt
import requests
from cryptography.hazmat.primitives import serialization

ACCESS_TOKEN_SECRET = "change-me"
DPoP_PROOF_AUD = "urn:demo-dpop"

# Paste your PRIVATE_PEM from gen_keypair.py here
PRIVATE_PEM = """-----BEGIN PRIVATE KEY-----
REPLACE_ME
-----END PRIVATE KEY-----"""

def make_dpop_proof(private_pem: str, url: str, method: str, jwk: dict) -> str:
    now = int(time.time())
    payload = {
        "htu": url,
        "htm": method,
        "aud": DPoP_PROOF_AUD,
        "jti": str(uuid.uuid4()),
        "iat": now,
        "typ": "dpop+jwt",
        "jwk": jwk,  # public parameters
    }
    # Sign using ES256 with the private key.
    return jwt.encode(payload, private_pem, algorithm="ES256")

def main():
    base = "http://127.0.0.1:5000"

    # 1) Mint token (demo)
    r = requests.post(f"{base}/mint")
    r.raise_for_status()
    token = r.json()["access_token"]

    # 2) Create DPoP proof
    # The server compares htu to the full request.url, so we must match exactly.
    api_url = f"{base}/api/data"

    # The jwk must match the public key corresponding to PRIVATE_PEM.
    # Paste your JWK from gen_keypair.py here:
    jwk = {
        "kty": "EC",
        "crv": "P-256",
        "x": "REPLACE_ME",
        "y": "REPLACE_ME",
    }

    dpop = make_dpop_proof(PRIVATE_PEM, api_url, "POST", jwk)

    # 3) Call protected API
    headers = {
        "Authorization": f"DPoP {token}",
        "DPoP": dpop,
    }
    rr = requests.post(api_url, headers=headers, json={"hello": "world"})
    print("status:", rr.status_code)
    print(rr.text)

if __name__ == "__main__":
    main()

---

Step 5: Run it and watch replay fail

Terminal A: start the server

python server.py

Terminal B: run the client once

python client.py

You should see something like:

{"ok": true, "message": "Zero Trust DPoP proof accepted."}

Terminal C: attempt replay by reusing the same proof

To test replay protection, modify the client to print the generated dpop value, copy it, and send the same request again within the 120s window. The second attempt should get:

• {"error":"replay_detected"} with HTTP 401

Even worse for an attacker: copying only the access token and generating no DPoP proof will also be denied with:

• {"error":"missing_dpop_proof"}

---

Practical takeaways I learned building this

• Zero Trust is about binding, not just limiting time. Short-lived tokens reduce blast radius, but replayable bearer semantics still leave a practical gap.
• Proof-of-possession changes the attacker model. A token alone is data; a DPoP proof turns it into “data + private key possession.”
• Request binding matters. Verifying htm and htu stops “valid proof for one endpoint” being reused for another.
• Anti-replay requires tracking jti. Stateless verification is not enough if proofs can be reused verbatim.

I started this thinking “token lifetime is the security lever,” but what I ended up with is a more Zero Trust-aligned pattern: CI issues ephemeral credentials, and every request must include cryptographic proof that’s bound to the exact request and rejected after reuse.