jinghui/hermes-webui
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(auth): harden OIDC issuer and EC token checks (#3825)

Browse Source
This commit is contained in:
Rod Boev committed 2026-06-29 22:31:45 -04:00
1 parent 3b26cab859
commit 39fbcb493b
2 files changed
+197 -10
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

api/auth_oidc.py
+45 -10
View File
@@ -14,7 +14,7 @@ from typing import Any
from cryptography.exceptions import InvalidSignature
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.asymmetric import ec, padding, rsa
from cryptography.hazmat.primitives.asymmetric import ec, padding, rsa, utils
from api.config import get_config
@@ -22,6 +22,7 @@ logger = logging.getLogger(__name__)
_DEFAULT_SCOPES = ("openid", "profile", "email")
_PENDING_TTL_SECONDS = 600
_MAX_PENDING_FLOWS = 128
_CLOCK_SKEW_SECONDS = 60
_CACHE_TTL_SECONDS = 300
@@ -101,6 +102,9 @@ def complete_authorization_code_flow(
if pending is None:
raise OIDCAuthError("Invalid OIDC state", status_code=401)
discovery = _get_discovery_document(cfg["issuer"])
discovery_issuer = str(discovery.get("issuer") or "").strip()
if discovery_issuer and discovery_issuer != cfg["issuer"]:
raise OIDCAuthError("OIDC discovery issuer did not match the configured issuer", status_code=502)
token_endpoint = str(discovery.get("token_endpoint") or "").strip()
if not token_endpoint:
raise OIDCConfigError("OIDC discovery document is missing token_endpoint")
@@ -122,7 +126,7 @@ def complete_authorization_code_flow(
claims = _validate_id_token(
id_token,
client_id=cfg["client_id"],
issuer=str(discovery.get("issuer") or cfg["issuer"]).strip(),
issuer=cfg["issuer"],
nonce=pending["nonce"],
jwks_uri=str(discovery.get("jwks_uri") or "").strip(),
)
@@ -235,6 +239,7 @@ def _store_pending_flow(state: str, payload: dict[str, Any]) -> None:
now = time.time()
with _pending_lock:
_prune_pending_flows(now)
_trim_pending_flows()
_pending_flows[state] = payload
@@ -256,6 +261,18 @@ def _prune_pending_flows(now: float) -> None:
_pending_flows.pop(state, None)
def _trim_pending_flows() -> None:
overflow = len(_pending_flows) - _MAX_PENDING_FLOWS + 1
if overflow <= 0:
return
oldest = sorted(
_pending_flows,
key=lambda state: float(_pending_flows[state].get("created_at") or 0),
)
for state in oldest[:overflow]:
_pending_flows.pop(state, None)
def _get_discovery_document(issuer: str) -> dict[str, Any]:
discovery_url = _discovery_url_for_issuer(issuer)
cached = _cache_get(_discovery_lock, _discovery_cache, discovery_url)
@@ -274,12 +291,16 @@ def _discovery_url_for_issuer(issuer: str) -> str:
return issuer.rstrip("/") + "/.well-known/openid-configuration"
def _get_jwks_document(jwks_uri: str) -> dict[str, Any]:
def _get_jwks_document(jwks_uri: str, *, force_refresh: bool = False) -> dict[str, Any]:
if not jwks_uri:
raise OIDCConfigError("OIDC discovery document is missing jwks_uri")
cached = _cache_get(_jwks_lock, _jwks_cache, jwks_uri)
if cached is not None:
return cached
if force_refresh:
with _jwks_lock:
_jwks_cache.pop(jwks_uri, None)
else:
cached = _cache_get(_jwks_lock, _jwks_cache, jwks_uri)
if cached is not None:
return cached
data = _fetch_json(jwks_uri)
if not isinstance(data, dict):
raise OIDCAuthError("OIDC JWKS response was not a JSON object", status_code=502)
@@ -363,7 +384,13 @@ def _validate_id_token(
if not alg or alg == "none":
raise OIDCAuthError("OIDC id_token uses an unsupported signing algorithm")
jwks = _get_jwks_document(jwks_uri)
public_key = _select_public_key(jwks, header)
try:
public_key = _select_public_key(jwks, header)
except OIDCAuthError as exc:
if "did not contain the signing key" not in str(exc):
raise
jwks = _get_jwks_document(jwks_uri, force_refresh=True)
public_key = _select_public_key(jwks, header)
_verify_jwt_signature(public_key, alg, signed, signature)
_validate_registered_claims(claims, client_id=client_id, issuer=issuer, nonce=nonce)
if not str(claims.get("sub") or "").strip():
@@ -453,19 +480,27 @@ def _verify_jwt_signature(public_key, alg: str, signed: bytes, signature: bytes)
public_key.verify(signature, signed, padding.PKCS1v15(), hashes.SHA512())
return
if alg == "ES256":
public_key.verify(signature, signed, ec.ECDSA(hashes.SHA256()))
public_key.verify(_jose_ecdsa_signature_to_der(signature, 32), signed, ec.ECDSA(hashes.SHA256()))
return
if alg == "ES384":
public_key.verify(signature, signed, ec.ECDSA(hashes.SHA384()))
public_key.verify(_jose_ecdsa_signature_to_der(signature, 48), signed, ec.ECDSA(hashes.SHA384()))
return
if alg == "ES512":
public_key.verify(signature, signed, ec.ECDSA(hashes.SHA512()))
public_key.verify(_jose_ecdsa_signature_to_der(signature, 66), signed, ec.ECDSA(hashes.SHA512()))
return
except InvalidSignature as exc:
raise OIDCAuthError("OIDC id_token signature verification failed") from exc
raise OIDCAuthError(f"Unsupported OIDC signing algorithm: {alg}", status_code=502)
def _jose_ecdsa_signature_to_der(signature: bytes, part_size: int) -> bytes:
if len(signature) != part_size * 2:
raise OIDCAuthError("OIDC id_token ECDSA signature was malformed")
r = int.from_bytes(signature[:part_size], "big")
s = int.from_bytes(signature[part_size:], "big")
return utils.encode_dss_signature(r, s)
def _validate_registered_claims(
claims: dict[str, Any],
*,
tests/test_issue3825_oidc_auth.py
+152
View File
@@ -1,9 +1,12 @@
import io
import json
import time
from types import SimpleNamespace
from urllib.parse import parse_qs, urlparse
import pytest
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.asymmetric import ec, utils
class FakeHeaders(dict):
@@ -35,6 +38,31 @@ class RouteFakeHandler:
needle = name.lower()
return [value for key, value in self.sent_headers if key.lower() == needle]
def _ec_jwk(private_key, *, kid="key-1", alg="ES256"):
numbers = private_key.public_key().public_numbers()
size = (numbers.curve.key_size + 7) // 8
import api.auth_oidc as auth_oidc
return {
"kid": kid,
"kty": "EC",
"alg": alg,
"crv": "P-256",
"x": auth_oidc._b64u(numbers.x.to_bytes(size, "big")),
"y": auth_oidc._b64u(numbers.y.to_bytes(size, "big")),
}
def _signed_es256_jwt(private_key, header, claims):
import api.auth_oidc as auth_oidc
header_b64 = auth_oidc._b64u(json.dumps(header, separators=(",", ":")).encode("utf-8"))
claims_b64 = auth_oidc._b64u(json.dumps(claims, separators=(",", ":")).encode("utf-8"))
signed = f"{header_b64}.{claims_b64}".encode("ascii")
der_signature = private_key.sign(signed, ec.ECDSA(hashes.SHA256()))
r, s = utils.decode_dss_signature(der_signature)
raw_signature = r.to_bytes(32, "big") + s.to_bytes(32, "big")
return f"{header_b64}.{claims_b64}.{auth_oidc._b64u(raw_signature)}"
def test_oidc_start_redirects_with_pkce_state_and_nonce(monkeypatch):
import api.routes as routes
@@ -326,3 +354,127 @@ def test_validate_id_token_rejects_mismatched_jwk_key_family(monkeypatch):
nonce="nonce-token",
jwks_uri="https://issuer.example/jwks",
)
def test_validate_id_token_accepts_real_es256_jose_signature(monkeypatch):
import api.auth_oidc as auth_oidc
private_key = ec.generate_private_key(ec.SECP256R1())
token = _signed_es256_jwt(
private_key,
{"alg": "ES256", "kid": "key-1"},
{
"iss": "https://issuer.example",
"aud": "webui-client",
"exp": 32503680000,
"nonce": "nonce-token",
"sub": "user-123",
},
)
monkeypatch.setattr(
auth_oidc,
"_get_jwks_document",
lambda _jwks_uri, **_kwargs: {"keys": [_ec_jwk(private_key)]},
)
claims = auth_oidc._validate_id_token(
token,
client_id="webui-client",
issuer="https://issuer.example",
nonce="nonce-token",
jwks_uri="https://issuer.example/jwks",
)
assert claims["sub"] == "user-123"
def test_complete_authorization_pins_discovery_to_configured_issuer(monkeypatch):
import api.auth_oidc as auth_oidc
from api.auth_oidc import OIDCAuthError
monkeypatch.setattr(
auth_oidc,
"_resolve_oidc_config",
lambda: {
"issuer": "https://issuer.example",
"client_id": "webui-client",
"client_secret": "",
"redirect_uri": "",
"scopes": ["openid"],
"allow_claim": "email",
"allow_values": ["user@example.com"],
},
)
monkeypatch.setattr(
auth_oidc,
"_get_discovery_document",
lambda _issuer: {
"issuer": "https://evil.example",
"token_endpoint": "https://issuer.example/token",
"jwks_uri": "https://issuer.example/jwks",
},
)
auth_oidc._pending_flows.clear()
auth_oidc._pending_flows["state-token"] = {
"created_at": time.time(),
"nonce": "nonce-token",
"code_verifier": "verifier",
"next_path": "/",
}
with pytest.raises(OIDCAuthError, match="discovery issuer"):
auth_oidc.complete_authorization_code_flow(
"http://localhost:8787",
"state-token",
"code-token",
)
def test_validate_id_token_refetches_jwks_once_on_key_miss(monkeypatch):
import api.auth_oidc as auth_oidc
old_key = ec.generate_private_key(ec.SECP256R1())
new_key = ec.generate_private_key(ec.SECP256R1())
token = _signed_es256_jwt(
new_key,
{"alg": "ES256", "kid": "new-key"},
{
"iss": "https://issuer.example",
"aud": "webui-client",
"exp": 32503680000,
"nonce": "nonce-token",
"sub": "user-123",
},
)
jwks_uri = "https://issuer.example/jwks"
auth_oidc._jwks_cache.clear()
auth_oidc._jwks_cache[jwks_uri] = (
time.time() + 300,
{"keys": [_ec_jwk(old_key, kid="old-key")]},
)
fetches = []
def fake_fetch_json(url):
fetches.append(url)
return {"keys": [_ec_jwk(new_key, kid="new-key")]}
monkeypatch.setattr(auth_oidc, "_fetch_json", fake_fetch_json)
claims = auth_oidc._validate_id_token(
token,
client_id="webui-client",
issuer="https://issuer.example",
nonce="nonce-token",
jwks_uri=jwks_uri,
)
assert claims["sub"] == "user-123"
assert fetches == [jwks_uri]
def test_pending_oidc_flows_are_bounded(monkeypatch):
import api.auth_oidc as auth_oidc
monkeypatch.setattr(auth_oidc, "_MAX_PENDING_FLOWS", 2)
auth_oidc._pending_flows.clear()
auth_oidc._store_pending_flow("old", {"created_at": 1, "nonce": "old"})
auth_oidc._store_pending_flow("middle", {"created_at": 2, "nonce": "middle"})
auth_oidc._store_pending_flow("new", {"created_at": 3, "nonce": "new"})
assert set(auth_oidc._pending_flows) == {"middle", "new"}