🛡️ Sentinel: [HIGH] OIDC 토큰 검증 시 crit 헤더 확인 누락 문제 해결

.jules/sentinel.md

@@ -2,3 +2,8 @@
 **Vulnerability:** User-provided string fields (like project and connection names) lacked strict validation against control characters, only relying on length constraints.
 **Learning:** This could potentially lead to Log Injection (CRLF injection), Null Byte Injection, or terminal escape injection if these strings are subsequently logged or rendered directly.
 **Prevention:** Use explicit regex validation `pattern=r'^[^\x00-\x1F\x7F]+$'` on Pydantic string fields to strictly reject control characters.
+
+## 2024-05-24 - JWT Header Verification Missing `crit` Processing
+**Vulnerability:** OIDC 토큰 검증 시, JWT 헤더의 `crit` (critical) 필드가 확인되지 않고 무시되고 있었습니다.
+**Learning:** RFC 7515에 따르면 `crit` 헤더는 배열 형태여야 하며, 명시된 확장 파라미터들을 서버가 이해하고 처리할 수 있어야만 토큰을 수용해야 합니다. 이를 확인하지 않으면 STRIX 보안 요건을 통과할 수 없고 알 수 없는 확장 기능이 포함된 토큰이 악용될 위험이 있습니다. 이 프로젝트의 경우 커스텀 확장을 지원하지 않으므로 `crit` 헤더가 포함된 토큰은 거부되어야 합니다.
+**Prevention:** `_validate_jwt_header` 함수에 `crit` 파라미터가 있는지 검사하는 로직을 추가하고, 만약 존재할 경우 문자열의 배열(길이 10 이하)인지 확인 후 지원되지 않는 확장 기능이므로 예외를 발생시키도록 수정했습니다.

backend/app/auth.py

@@ -10,7 +10,7 @@
 import httpx
 from fastapi import Depends, HTTPException, Request
 from jose import jwt
-from sqlalchemy import select, delete
+from sqlalchemy import delete, select
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.db import get_session
@@ -185,6 +185,18 @@ def _validate_jwt_header(header: dict[str, Any]) -> str:
     if content_type is not None:
         raise HTTPException(status_code=401, detail="unsupported token content type")
 
+    crit = header.get("crit")
+    if crit is not None:
+        if not isinstance(crit, list) or len(crit) > 10:
+            raise HTTPException(status_code=401, detail="invalid crit header")
+        for item in crit:
+            if not isinstance(item, str):
+                raise HTTPException(status_code=401, detail="invalid crit header")
+            # We do not support any critical extensions
+            raise HTTPException(
+                status_code=401, detail="unsupported critical extension"
+            )
+
     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")
@@ -194,8 +206,8 @@ def _validate_jwt_header(header: dict[str, Any]) -> str:
 async def revoke_token_jti(jwt_id: str, expires_at: dt.datetime) -> None:
     """Record a JWT ID as revoked until its natural expiry."""
 
-    from app.models import RevokedToken
     from app.db import SessionLocal
+    from app.models import RevokedToken
 
     if not jwt_id:
         return
@@ -213,8 +225,8 @@ async def revoke_token_jti(jwt_id: str, expires_at: dt.datetime) -> None:
 async def is_token_jti_revoked(jwt_id: str) -> bool:
     """Return whether the JWT ID is currently revoked."""
 
-    from app.models import RevokedToken
     from app.db import SessionLocal
+    from app.models import RevokedToken
 
     current = dt.datetime.now(dt.timezone.utc)
     async with SessionLocal() as session:
@@ -464,7 +476,7 @@ async def get_current_user(
     """
     auth_header = request.headers.get("Authorization", "")
     if auth_header.startswith("Bearer " + API_KEY_PREFIX):
-        return await _user_from_api_key(session, auth_header[len("Bearer "):])
+        return await _user_from_api_key(session, auth_header[len("Bearer ") :])
     subject, display_name = await _get_subject_from_request(request)
     async with session.begin():
         return await _ensure_user(session, subject, display_name)

backend/tests/test_auth_security.py

@@ -71,7 +71,7 @@ class FakeAsyncClient:
         def __init__(self, **kwargs: object) -> None:
             observed.update(kwargs)
 
-        async def __aenter__(self) -> "FakeAsyncClient":
+        async def __aenter__(self) -> FakeAsyncClient:
             return self
 
         async def __aexit__(self, *_args: object) -> None:
@@ -109,7 +109,7 @@ class FakeAsyncClient:
         def __init__(self, **_kwargs: object) -> None:
             return None
 
-        async def __aenter__(self) -> "FakeAsyncClient":
+        async def __aenter__(self) -> FakeAsyncClient:
             return self
 
         async def __aexit__(self, *_args: object) -> None:
@@ -147,7 +147,7 @@ class FakeAsyncClient:
         def __init__(self, **kwargs: object) -> None:
             observed.update(kwargs)
 
-        async def __aenter__(self) -> "FakeAsyncClient":
+        async def __aenter__(self) -> FakeAsyncClient:
             return self
 
         async def __aexit__(self, *_args: object) -> None:
@@ -467,6 +467,33 @@ async def mock_revoke(jti, ext):
     assert exc_info.value.detail == "token revoked"
 
 
+@pytest.mark.asyncio
+async def test_oidc_rejects_invalid_crit_header() -> None:
+    # crit is not a list
+    with pytest.raises(HTTPException) as excinfo:
+        auth._validate_jwt_header({"alg": "RS256", "crit": "b64"})
+    assert excinfo.value.status_code == 401
+    assert excinfo.value.detail == "invalid crit header"
+
+    # crit list is too long
+    with pytest.raises(HTTPException) as excinfo:
+        auth._validate_jwt_header({"alg": "RS256", "crit": ["ext"] * 11})
+    assert excinfo.value.status_code == 401
+    assert excinfo.value.detail == "invalid crit header"
+
+    # crit item is not a string
+    with pytest.raises(HTTPException) as excinfo:
+        auth._validate_jwt_header({"alg": "RS256", "crit": [123]})
+    assert excinfo.value.status_code == 401
+    assert excinfo.value.detail == "invalid crit header"
+
+    # crit contains unsupported extension
+    with pytest.raises(HTTPException) as excinfo:
+        auth._validate_jwt_header({"alg": "RS256", "crit": ["b64"]})
+    assert excinfo.value.status_code == 401
+    assert excinfo.value.detail == "unsupported critical extension"
+
+
 @pytest.mark.asyncio
 async def test_auth_fails_closed_without_oidc(
     monkeypatch: pytest.MonkeyPatch,
@@ -594,6 +621,7 @@ async def mock_is_token_revoked2(jti):
     assert exc_info.value.status_code == 401
     assert exc_info.value.detail == "token verification failed"
 
+
 @pytest.mark.asyncio
 async def test_oidc_rejects_algorithm_key_type_mismatch(
     monkeypatch: pytest.MonkeyPatch,
@@ -617,7 +645,9 @@ async def mock_is_token_revoked2(jti):
     monkeypatch.setattr(auth, "is_token_jti_revoked", mock_is_token_revoked2)
 
     def fail_decode(*_: object, **__: object) -> dict:
-        raise AssertionError("jwt.decode must not run for mismatched algorithm/key type")
+        raise AssertionError(
+            "jwt.decode must not run for mismatched algorithm/key type"
+        )
 
     monkeypatch.setattr(auth.jwt, "decode", fail_decode)
 
@@ -626,6 +656,8 @@ def fail_decode(*_: object, **__: object) -> dict:
 
     assert exc_info.value.status_code == 401
     assert exc_info.value.detail == "algorithm/key type mismatch"
+
+
 @pytest.mark.asyncio
 async def test_oidc_jwks_refresh_rate_limiting(
     monkeypatch: pytest.MonkeyPatch,
@@ -636,7 +668,7 @@ class FakeAsyncClient:
         def __init__(self, **kwargs: object) -> None:
             pass
 
-        async def __aenter__(self) -> "FakeAsyncClient":
+        async def __aenter__(self) -> FakeAsyncClient:
             return self
 
         async def __aexit__(self, *_args: object) -> None:
@@ -684,7 +716,7 @@ class FakeAsyncClient:
         def __init__(self, **kwargs: object) -> None:
             pass
 
-        async def __aenter__(self) -> "FakeAsyncClient":
+        async def __aenter__(self) -> FakeAsyncClient:
             return self
 
         async def __aexit__(self, *_args: object) -> None: