Skip to content

fix: set size cap on reads #180

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
AjayThorve merged 2 commits into from
Apr 14, 2026
Merged

fix: set size cap on reads #180

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
8056fd3
set size cap on reads
cdgamarose-nv Apr 14, 2026
8c83208
add failure log
cdgamarose-nv Apr 14, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 19 additions & 2 deletions frontends/aiq_api/src/aiq_api/auth/jwt_validator.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,7 @@
logger = logging.getLogger(__name__)

_MISSING_PYJWT = "PyJWT[cryptography] is required for JWT validation. Install with: pip install 'PyJWT[cryptography]'"
_MAX_FETCH_BYTES = 64 << 10 # 64 KB cap on OIDC/JWKS responses


class JWTValidator(TokenValidator):
Expand Down Expand Up @@ -91,7 +92,15 @@ def _fetch_oidc_config(self) -> dict:
url = f"{self.issuer_url}/.well-known/openid-configuration"
logger.debug("Fetching OIDC discovery from %s", url)
with urllib.request.urlopen(url, timeout=10) as resp: # noqa: S310
return json.loads(resp.read())
raw = resp.read(_MAX_FETCH_BYTES + 1)
if len(raw) > _MAX_FETCH_BYTES:
logger.warning(
"OIDC discovery response from %s exceeded %d-byte cap; truncating",
url,
_MAX_FETCH_BYTES,
)
raw = raw[:_MAX_FETCH_BYTES]
return json.loads(raw)

def _fetch_jwks_keys(self) -> list[tuple[str | None, Any]]:
"""Fetch JWKS and return (kid, PyJWK) pairs.
Expand All @@ -107,7 +116,15 @@ def _fetch_jwks_keys(self) -> list[tuple[str | None, Any]]:

assert self._jwks_uri is not None # caller ensures this
with urllib.request.urlopen(self._jwks_uri, timeout=10) as resp: # noqa: S310
data = json.loads(resp.read())
raw = resp.read(_MAX_FETCH_BYTES + 1)
if len(raw) > _MAX_FETCH_BYTES:
logger.warning(
"JWKS response from %s exceeded %d-byte cap; truncating",
self._jwks_uri,
_MAX_FETCH_BYTES,
)
raw = raw[:_MAX_FETCH_BYTES]
data = json.loads(raw)

keys: list[tuple[str | None, Any]] = []
for key_data in data.get("keys", []):
Expand Down
Loading