|
| 1 | +"""DPoP proof generation (RFC 9449) using ES256 (P-256).""" |
| 2 | + |
| 3 | +import base64 |
| 4 | +import hashlib |
| 5 | +import json |
| 6 | +import os |
| 7 | +import time |
| 8 | + |
| 9 | +from cryptography.hazmat.primitives.asymmetric import ec, utils |
| 10 | +from cryptography.hazmat.primitives.hashes import SHA256 |
| 11 | +from cryptography.hazmat.primitives.serialization import load_pem_private_key |
| 12 | + |
| 13 | + |
| 14 | +def load_private_key(pem_data: str) -> ec.EllipticCurvePrivateKey: |
| 15 | + """Load an EC P-256 private key from PEM string.""" |
| 16 | + key = load_pem_private_key(pem_data.encode(), password=None) |
| 17 | + if not isinstance(key, ec.EllipticCurvePrivateKey): |
| 18 | + raise ValueError("expected EC private key") |
| 19 | + if not isinstance(key.curve, ec.SECP256R1): |
| 20 | + raise ValueError("expected P-256 curve") |
| 21 | + return key |
| 22 | + |
| 23 | + |
| 24 | +def public_key_jwk(key: ec.EllipticCurvePrivateKey) -> dict[str, str]: |
| 25 | + """Return the JWK representation of the public key.""" |
| 26 | + pub = key.public_key() |
| 27 | + numbers = pub.public_numbers() |
| 28 | + return { |
| 29 | + "kty": "EC", |
| 30 | + "crv": "P-256", |
| 31 | + "x": _b64url_int(numbers.x, 32), |
| 32 | + "y": _b64url_int(numbers.y, 32), |
| 33 | + } |
| 34 | + |
| 35 | + |
| 36 | +def jwk_thumbprint(key: ec.EllipticCurvePrivateKey) -> str: |
| 37 | + """Compute RFC 7638 JWK thumbprint (base64url SHA-256).""" |
| 38 | + jwk = public_key_jwk(key) |
| 39 | + # RFC 7638: members in lexicographic order for EC: crv, kty, x, y |
| 40 | + canonical = f'{{"crv":"{jwk["crv"]}","kty":"{jwk["kty"]}","x":"{jwk["x"]}","y":"{jwk["y"]}"}}' |
| 41 | + digest = hashlib.sha256(canonical.encode()).digest() |
| 42 | + return base64.urlsafe_b64encode(digest).rstrip(b"=").decode() |
| 43 | + |
| 44 | + |
| 45 | +def create_proof(key: ec.EllipticCurvePrivateKey, method: str, url: str) -> str: |
| 46 | + """Create a DPoP proof JWT for the given HTTP method and URL.""" |
| 47 | + jwk = public_key_jwk(key) |
| 48 | + |
| 49 | + header = {"typ": "dpop+jwt", "alg": "ES256", "jwk": jwk} |
| 50 | + claims = { |
| 51 | + "jti": base64.urlsafe_b64encode(os.urandom(16)).rstrip(b"=").decode(), |
| 52 | + "htm": method, |
| 53 | + "htu": url, |
| 54 | + "iat": int(time.time()), |
| 55 | + } |
| 56 | + |
| 57 | + header_b64 = _b64url_json(header) |
| 58 | + claims_b64 = _b64url_json(claims) |
| 59 | + signed_content = f"{header_b64}.{claims_b64}" |
| 60 | + |
| 61 | + # Sign with ES256 — cryptography hashes internally |
| 62 | + der_sig = key.sign(signed_content.encode(), ec.ECDSA(SHA256())) |
| 63 | + # Convert DER to IEEE P1363 (fixed 64 bytes) |
| 64 | + r, s = utils.decode_dss_signature(der_sig) |
| 65 | + sig_bytes = r.to_bytes(32, "big") + s.to_bytes(32, "big") |
| 66 | + sig_b64 = base64.urlsafe_b64encode(sig_bytes).rstrip(b"=").decode() |
| 67 | + |
| 68 | + return f"{signed_content}.{sig_b64}" |
| 69 | + |
| 70 | + |
| 71 | +def _b64url_json(obj: dict) -> str: |
| 72 | + data = json.dumps(obj, separators=(",", ":")).encode() |
| 73 | + return base64.urlsafe_b64encode(data).rstrip(b"=").decode() |
| 74 | + |
| 75 | + |
| 76 | +def _b64url_int(n: int, size: int) -> str: |
| 77 | + b = n.to_bytes(size, "big") |
| 78 | + return base64.urlsafe_b64encode(b).rstrip(b"=").decode() |
0 commit comments