Review
Honor JWT jku for multi-museum crate vault opens
Annexium tours shared crate vaults across partner museums. Each house used to email us a static JWKS URL when rotating RS256 keys, which broke every time a DI house spun a new SSO. This PR drops the allowlisted JWKS table and resolves keys from the token's jku header so any accredited museum can self-host keys and open vault bays without a deploy.
PyJWTTier 6securityjwtjkussrf
Click a line to flag it, pick one or more labels, then submit. If the change looks correct, approve it.
services/vault/annexium_auth.py+15-6
| 1 | 1 | import jwt | |
| 2 | 2 | from fastapi import Header, HTTPException | |
| 3 | 3 | from jwt.algorithms import RSAAlgorithm | |
| 4 | 4 | ||
| 5 | 5 | from app.config import settings | |
| 6 | - | from app.jwks_registry import get_museum_signing_key | |
| 6 | + | import httpx | |
| 7 | 7 | ||
| 8 | 8 | ALLOWED_ROLES = {"courier", "registrar", "vault-master"} | |
| 9 | 9 | ||
| 10 | - | def _resolve_key(token: str): | |
| 11 | - | header = jwt.get_unverified_header(token) | |
| 12 | - | kid = header.get("kid") | |
| 13 | - | iss = jwt.decode(token, options={"verify_signature": False}).get("iss") | |
| 14 | - | return get_museum_signing_key(iss=iss, kid=kid) | |
| 10 | + | def _resolve_key(token: str): | |
| 11 | + | """Load the RS256 key from the JWKS URL the token advertises.""" | |
| 12 | + | header = jwt.get_unverified_header(token) | |
| 13 | + | jku = header.get("jku") | |
| 14 | + | kid = header.get("kid") | |
| 15 | + | if not jku or not kid: | |
| 16 | + | raise HTTPException(status_code=401, detail="jku and kid required") | |
| 17 | + | # Partners host JWKS on their own domains; follow the header. | |
| 18 | + | resp = httpx.get(jku, timeout=3.0) | |
| 19 | + | resp.raise_for_status() | |
| 20 | + | for jwk in resp.json().get("keys", []): | |
| 21 | + | if jwk.get("kid") == kid: | |
| 22 | + | return RSAAlgorithm.from_jwk(jwk) | |
| 23 | + | raise HTTPException(status_code=401, detail="kid not present in jwks") | |
| 15 | 24 | ||
| 16 | 25 | def authorize_crate_open( | |
| 17 | 26 | crate_id: str, | |
| 18 | 27 | authorization: str = Header(...), | |
| 19 | 28 | ) -> dict: | |
| 20 | 29 | if not authorization.startswith("Bearer "): | |
| 21 | 30 | raise HTTPException(status_code=401, detail="missing bearer") | |
| 22 | 31 | token = authorization[len("Bearer "):].strip() | |
| 23 | 32 | try: | |
| 24 | 33 | key = _resolve_key(token) | |
| 25 | 34 | claims = jwt.decode( | |
| 26 | 35 | token, | |
| 27 | 36 | key=key, | |
| 28 | 37 | algorithms=["RS256"], | |
| 29 | 38 | audience=settings.VAULT_JWT_AUD, | |
| 30 | 39 | ) | |
| 31 | 40 | except jwt.PyJWTError as exc: | |
| 32 | 41 | raise HTTPException(status_code=401, detail="invalid token") from exc | |
| 33 | 42 | if claims.get("role") not in ALLOWED_ROLES: | |
| 34 | 43 | raise HTTPException(status_code=403, detail="role not allowed") | |
| 35 | 44 | return {"crate_id": crate_id, "sub": claims.get("sub"), "via": "jku"} |