-
Notifications
You must be signed in to change notification settings - Fork 0
π‘οΈ Sentinel: [CRITICAL] Fix JWT κ²μ¦ μ 보 λ ΈμΆ λ° crit ν€λ κ²μ¦ λλ½ #993
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.
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
base: main
Are you sure you want to change the base?
Changes from all commits
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change | ||||
|---|---|---|---|---|---|---|
|
|
@@ -166,7 +166,7 @@ def _jwt_expiry(claims: dict[str, Any]) -> dt.datetime: | |||||
|
|
||||||
| exp = claims.get("exp") | ||||||
| if not isinstance(exp, int | float): | ||||||
| raise HTTPException(status_code=401, detail="token missing exp") | ||||||
| raise HTTPException(status_code=401, detail="invalid token") | ||||||
| return dt.datetime.fromtimestamp(float(exp), tz=dt.timezone.utc) | ||||||
|
|
||||||
|
|
||||||
|
|
@@ -179,15 +179,26 @@ def _validate_jwt_header(header: dict[str, Any]) -> str: | |||||
| not isinstance(token_type, str) | ||||||
| or token_type.strip().lower() not in OIDC_ALLOWED_TOKEN_TYPES | ||||||
| ): | ||||||
| raise HTTPException(status_code=401, detail="unsupported token type") | ||||||
| raise HTTPException(status_code=401, detail="invalid token") | ||||||
|
|
||||||
| content_type = header.get("cty") | ||||||
| if content_type is not None: | ||||||
| raise HTTPException(status_code=401, detail="unsupported token content type") | ||||||
| raise HTTPException(status_code=401, detail="invalid token") | ||||||
|
|
||||||
| crit = header.get("crit") | ||||||
| if crit is not None: | ||||||
| if not isinstance(crit, list) or len(crit) == 0 or len(crit) > 10: | ||||||
| raise HTTPException(status_code=401, detail="invalid token") | ||||||
| for item in crit: | ||||||
| if not isinstance(item, str): | ||||||
| raise HTTPException(status_code=401, detail="invalid token") | ||||||
| # For this app, we don't recognize ANY critical extensions. | ||||||
| # If `crit` is present and specifies any extensions, we MUST reject the token. | ||||||
| raise HTTPException(status_code=401, detail="invalid token") | ||||||
|
Comment on lines
+188
to
+197
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. π Info: Redundant checks in crit validation loop The loop unconditionally raises on the first Was this helpful? React with π or π to provide feedback. |
||||||
|
|
||||||
| header_alg_raw = header.get("alg") | ||||||
| if not isinstance(header_alg_raw, str) or not header_alg_raw: | ||||||
| raise HTTPException(status_code=401, detail="token missing alg") | ||||||
| raise HTTPException(status_code=401, detail="invalid token") | ||||||
| return header_alg_raw.upper() | ||||||
|
|
||||||
|
|
||||||
|
|
@@ -240,13 +251,13 @@ async def _decode_verified_oidc_token(token: str) -> dict[str, Any]: | |||||
| try: | ||||||
| header = cast(dict[str, Any], jwt.get_unverified_header(token)) | ||||||
| except Exception: # noqa: BLE001 | ||||||
| raise HTTPException(status_code=401, detail="invalid token header") | ||||||
| raise HTTPException(status_code=401, detail="invalid token") | ||||||
|
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. π Maintainability & Code Quality | π‘ Minor | β‘ Quick win μμΈ μμΈ μ°κ²° λ°©μμ λͺ μνμΈμ.
μμ μμ- raise HTTPException(status_code=401, detail="invalid token")
+ raise HTTPException(status_code=401, detail="invalid token") from Noneπ Committable suggestion
Suggested change
π§° Toolsπͺ Ruff (0.16.2)[warning] 254-254: Within an (B904) π€ Prompt for AI AgentsSource: Linters/SAST tools |
||||||
|
|
||||||
| header_alg = _validate_jwt_header(header) | ||||||
| if header_alg not in OIDC_ALLOWED_ALGORITHMS: | ||||||
| raise HTTPException( | ||||||
| status_code=401, | ||||||
| detail="unsupported token algorithm", | ||||||
| detail="invalid token", | ||||||
| ) | ||||||
|
|
||||||
| jwks = await _get_jwks() | ||||||
|
|
@@ -255,20 +266,20 @@ async def _decode_verified_oidc_token(token: str) -> dict[str, Any]: | |||||
| jwks = await _get_jwks(force_refresh=True) | ||||||
| jwk = _pick_jwk(jwks, header.get("kid")) | ||||||
| if jwk is None: | ||||||
| raise HTTPException(status_code=401, detail="unknown signing key") | ||||||
| raise HTTPException(status_code=401, detail="invalid token") | ||||||
|
|
||||||
| kty = jwk.get("kty") | ||||||
| if not isinstance(kty, str): | ||||||
| raise HTTPException(status_code=401, detail="algorithm/key type mismatch") | ||||||
| raise HTTPException(status_code=401, detail="invalid token") | ||||||
| jwk_kty = kty.upper() | ||||||
| if jwk_kty == "RSA": | ||||||
| if not (header_alg.startswith("RS") or header_alg.startswith("PS")): | ||||||
| raise HTTPException(status_code=401, detail="algorithm/key type mismatch") | ||||||
| raise HTTPException(status_code=401, detail="invalid token") | ||||||
| elif jwk_kty == "EC": | ||||||
| if not header_alg.startswith("ES"): | ||||||
| raise HTTPException(status_code=401, detail="algorithm/key type mismatch") | ||||||
| raise HTTPException(status_code=401, detail="invalid token") | ||||||
| else: | ||||||
| raise HTTPException(status_code=401, detail="algorithm/key type mismatch") | ||||||
| raise HTTPException(status_code=401, detail="invalid token") | ||||||
|
|
||||||
| try: | ||||||
| claims = jwt.decode( | ||||||
|
|
@@ -288,7 +299,7 @@ async def _decode_verified_oidc_token(token: str) -> dict[str, Any]: | |||||
| ) | ||||||
| except Exception as err: | ||||||
| raise HTTPException( | ||||||
| status_code=401, detail="token verification failed" | ||||||
| status_code=401, detail="invalid token" | ||||||
| ) from err | ||||||
|
|
||||||
| return cast(dict[str, Any], claims) | ||||||
|
|
@@ -303,13 +314,13 @@ async def _verified_token_from_claims( | |||||
| jwt_id = claims.get("jti") | ||||||
| name = claims.get("name") or claims.get("preferred_username") | ||||||
| if not isinstance(sub, str): | ||||||
| raise HTTPException(status_code=401, detail="token missing sub") | ||||||
| raise HTTPException(status_code=401, detail="invalid token") | ||||||
| if not isinstance(jwt_id, str) or not jwt_id.strip(): | ||||||
| raise HTTPException(status_code=401, detail="token missing jti") | ||||||
| raise HTTPException(status_code=401, detail="invalid token") | ||||||
|
|
||||||
| expires_at = _jwt_expiry(claims) | ||||||
| if verify_revocation and await is_token_jti_revoked(jwt_id): | ||||||
| raise HTTPException(status_code=401, detail="token revoked") | ||||||
| raise HTTPException(status_code=401, detail="invalid token") | ||||||
|
|
||||||
| return VerifiedToken( | ||||||
| subject=sub, | ||||||
|
|
||||||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,33 @@ | ||
| import pytest | ||
| from fastapi import HTTPException | ||
| from app.auth import _validate_jwt_header | ||
|
|
||
| def test_crit_validates_as_list(): | ||
| with pytest.raises(HTTPException) as exc_info: | ||
| _validate_jwt_header({"alg": "RS256", "crit": "not-a-list"}) | ||
| assert exc_info.value.status_code == 401 | ||
| assert exc_info.value.detail == "invalid token" | ||
|
|
||
| def test_crit_rejects_empty_list(): | ||
| with pytest.raises(HTTPException) as exc_info: | ||
| _validate_jwt_header({"alg": "RS256", "crit": []}) | ||
| assert exc_info.value.status_code == 401 | ||
| assert exc_info.value.detail == "invalid token" | ||
|
|
||
| def test_crit_rejects_long_list(): | ||
| with pytest.raises(HTTPException) as exc_info: | ||
| _validate_jwt_header({"alg": "RS256", "crit": ["item"] * 11}) | ||
| assert exc_info.value.status_code == 401 | ||
| assert exc_info.value.detail == "invalid token" | ||
|
|
||
| def test_crit_rejects_non_string_items(): | ||
| with pytest.raises(HTTPException) as exc_info: | ||
| _validate_jwt_header({"alg": "RS256", "crit": [123]}) | ||
| assert exc_info.value.status_code == 401 | ||
| assert exc_info.value.detail == "invalid token" | ||
|
|
||
| def test_crit_rejects_unrecognized_items(): | ||
| with pytest.raises(HTTPException) as exc_info: | ||
| _validate_jwt_header({"alg": "RS256", "crit": ["b64"]}) | ||
|
Comment on lines
+5
to
+31
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. π Maintainability & Code Quality | π Major | β‘ Quick win μ ν μ€νΈ ν¨μμ λ°ν νμκ³Ό docstringμ μΆκ°νμΈμ. μλ‘ μΆκ°ν As per coding guidelines, μμ μμ-def test_crit_validates_as_list():
+def test_crit_validates_as_list() -> None:
+ """Reject a non-list `crit` header."""π€ Prompt for AI AgentsSource: Coding guidelines |
||
| assert exc_info.value.status_code == 401 | ||
| assert exc_info.value.detail == "invalid token" | ||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
π Maintainability & Code Quality | π Major | β‘ Quick win
보μ κ·Όκ±°μ μΈμ©μ μΆκ°νμΈμ.
μ΄ λ³κ²½μ JWT 보μ λμμ λ¬Έμννμ§λ§, κ΄λ ¨ νμ λ¬Ένμ μΈμ© λλ λ§ν¬μ μμ½μ ν¬ν¨νμ§ μμ΅λλ€. PR μ€λͺ μ΄λ μ΄ λ¬Έμμ κ·Όκ±°λ₯Ό μΆκ°νμΈμ.
As per coding guidelines, substantive feature pull requests must provide relevant academic literature with full citations, or citations, links, and summaries.
π€ Prompt for AI Agents
Source: Coding guidelines