Conversation
|
Warning Rate limit exceeded@touale has exceeded the limit for the number of commits that can be reviewed per hour. Please wait 13 minutes and 54 seconds before requesting another review. ⌛ How to resolve this issue?After the wait time has elapsed, a review can be triggered using the We recommend that you space out your commits to avoid hitting the rate limit. 🚦 How do rate limits work?CodeRabbit enforces hourly rate limits for each developer per organization. Our paid plans have higher rate limits than the trial, open-source and free plans. In all cases, we re-allow further reviews after a brief timeout. Please see our FAQ for further information. 📒 Files selected for processing (2)
WalkthroughAdds OAuth2/JWT-based authentication and configuration, replacing Basic auth for docs protection; introduces OauthConfig, JWT creation/validation, OAuth callback handling, and updates ingress and application auth to accept API key or JWT. Also adds PyJWT dependency and updates tests and proxy logging. Changes
Sequence DiagramssequenceDiagram
autonumber
participant User
participant App as Application
participant Auth as framex.driver.auth
participant Provider as OAuth Provider
participant UI as UserInfo Endpoint
User->>App: GET /docs (no JWT)
App->>Auth: authenticate(request)
Auth->>Auth: check JWT cookie
alt no valid JWT
Auth->>User: 301 Redirect to Provider auth URL
User->>Provider: Authorize (client_id, redirect_uri)
Provider-->>User: Redirect to /oauth/callback?code=...
User->>App: GET /oauth/callback?code=...
App->>Auth: oauth_callback(code)
Auth->>Provider: POST token exchange (httpx)
Provider-->>Auth: access_token
Auth->>UI: GET user info with access_token
UI-->>Auth: user info
Auth->>Auth: create_jwt(user_info), set HttpOnly cookie
Auth-->>User: Redirect to docs with JWT cookie
end
User->>App: GET /docs (with JWT cookie)
App->>Auth: authenticate(request)
Auth->>Auth: validate JWT cookie
Auth-->>App: authentication success
App-->>User: Serve docs
Estimated code review effort🎯 4 (Complex) | ⏱️ ~45 minutes Possibly related PRs
Poem
Pre-merge checks and finishing touches❌ Failed checks (1 warning, 1 inconclusive)
✅ Passed checks (1 passed)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
Codecov Report❌ Patch coverage is
📢 Thoughts on this report? Let us know! |
|
@coderabbitai help |
ChatThere are 3 ways to chat with CodeRabbit:
CodeRabbit commands
Other keywords and placeholders
CodeRabbit configuration file (
|
|
@coderabbitai generate tests |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 2
🧹 Nitpick comments (5)
src/framex/config.py (1)
55-83: Ephemeral JWT secret invalidates sessions on restart.The auto-generated
jwt_secretviasecrets.token_urlsafe(32)is cryptographically sound, but it's regenerated on every application restart. This means all existing JWT tokens (user sessions) become invalid after a restart or deployment.Consider:
- Documenting this behavior explicitly so operators know to configure a persistent secret in production.
- Adding a warning log when the secret is auto-generated to alert operators.
Additionally, the default empty strings for OAuth fields (
client_id,client_secret, etc.) could lead to confusing runtime errors if OAuth is partially configured. Consider adding validation inmodel_post_initto ensure required fields are set when OAuth is intended to be used.🔎 Suggested validation enhancement
def model_post_init(self, context: Any) -> None: super().model_post_init(context) + # Warn if critical OAuth fields are empty but OAuth is being used + if self.base_url and (not self.client_id or not self.client_secret): + import warnings + warnings.warn("OAuth base_url is set but client_id or client_secret is empty") if not self.authorization_url: self.authorization_url = f"{self.base_url}/oauth/authorize" if not self.token_url: self.token_url = f"{self.base_url}/oauth/token" if not self.user_info_url: self.user_info_url = f"{self.base_url}/api/v4/user" if not self.jwt_secret: + import logging + logging.warning("JWT secret auto-generated; sessions will not survive restarts") self.jwt_secret = secrets.token_urlsafe(32)src/framex/driver/auth.py (4)
16-28: Avoid mutating the input payload dictionary.The
create_jwtfunction directly mutates thepayloadparameter by addingiatandexpkeys. This can cause unexpected side effects for callers who might reuse the dict.🔎 Proposed fix
def create_jwt(payload: dict) -> str: if not settings.auth.oauth: raise RuntimeError("OAuth not configured") now_utc = datetime.now(UTC) - payload.update( - { - "iat": int(now_utc.timestamp()), - "exp": int((now_utc + timedelta(hours=24)).timestamp()), - } - ) - return jwt.encode(payload, settings.auth.oauth.jwt_secret, algorithm=settings.auth.oauth.jwt_algorithm) + token_payload = { + **payload, + "iat": int(now_utc.timestamp()), + "exp": int((now_utc + timedelta(hours=24)).timestamp()), + } + return jwt.encode(token_payload, settings.auth.oauth.jwt_secret, algorithm=settings.auth.oauth.jwt_algorithm)
50-80: Consider using 302 instead of 301 for OAuth redirect.HTTP 301 (Moved Permanently) is typically cached by browsers, which may cause issues if the OAuth configuration changes. HTTP 302 (Found) or 303 (See Other) are more appropriate for temporary redirects like OAuth authorization flows.
Also, as flagged by static analysis, catching bare
Exceptionon line 61 is overly broad—consider catchingjwt.PyJWTErrororjwt.InvalidTokenErrorspecifically.🔎 Proposed fixes
- except Exception as e: + except jwt.PyJWTError as e: from framex.log import logger logger.warning(f"JWT decode failed: {e}")raise HTTPException( - status_code=status.HTTP_301_MOVED_PERMANENTLY, + status_code=status.HTTP_302_FOUND, headers={ "Location": (
87-98: Add timeout to the token exchange HTTP call.The user info request on line 108 has a timeout of 10 seconds, but the token exchange POST request has no timeout. This could cause the callback to hang indefinitely if the OAuth provider is unresponsive.
🔎 Proposed fix
async with httpx.AsyncClient() as client: resp = await client.post( settings.auth.oauth.token_url, data={ "client_id": settings.auth.oauth.client_id, "client_secret": settings.auth.oauth.client_secret, "grant_type": "authorization_code", "code": code, "redirect_uri": settings.auth.oauth.call_back_url, }, + timeout=10, )
125-131: Consider addingsecure=Truefor the JWT cookie in production.The cookie is set with
httponly=Trueandsamesite="lax", which are good security practices. However, for production deployments over HTTPS, addingsecure=Trueensures the cookie is only sent over encrypted connections.🔎 Proposed enhancement
res.set_cookie( "token", create_jwt(user_info), httponly=True, samesite="lax", + secure=True, # Consider making this configurable for local dev )Alternatively, make this configurable based on environment to support local development without HTTPS.
📜 Review details
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
⛔ Files ignored due to path filters (1)
uv.lockis excluded by!**/*.lock
📒 Files selected for processing (9)
pyproject.tomlpytest.inisrc/framex/config.pysrc/framex/driver/application.pysrc/framex/driver/auth.pysrc/framex/driver/ingress.pysrc/framex/plugins/proxy/__init__.pytests/api/test_echo.pytests/driver/test_application.py
🧰 Additional context used
🧬 Code graph analysis (3)
src/framex/driver/application.py (1)
src/framex/driver/auth.py (2)
authenticate(50-80)oauth_callback(83-132)
src/framex/driver/ingress.py (1)
src/framex/driver/auth.py (1)
auth_jwt(31-47)
src/framex/driver/auth.py (1)
src/framex/config.py (2)
get_auth_keys(97-116)call_back_url(70-71)
🪛 Ruff (0.14.10)
src/framex/driver/auth.py
18-18: Avoid specifying long messages outside the exception class
(TRY003)
45-45: Consider moving this statement to an else block
(TRY300)
59-59: Consider moving this statement to an else block
(TRY300)
61-61: Do not catch blind exception: Exception
(BLE001)
85-85: Avoid specifying long messages outside the exception class
(TRY003)
🔇 Additional comments (12)
src/framex/plugins/proxy/__init__.py (1)
297-300: Logging simplification looks reasonable, but verify OAuth changes are captured elsewhere.The logging change on line 298 appropriately simplifies output to essential details (tag and function name) rather than verbose parameter dumps. However, the PR objective mentions introducing OAuth2/JWT authentication, but this file only shows a logging change. The OAuth-related changes (OauthConfig class, JWT token generation/validation, route authentication updates) should be reviewed in their respective files.
Since only the proxy plugin file is provided for review, I cannot assess the completeness of the OAuth2/JWT authentication implementation described in the PR objective.
Please confirm: Are the OAuth2/JWT authentication implementation changes present in separate files that should be included in this review? The PR objective describes:
- PyJWT dependency addition
- OauthConfig class for OAuth settings
- OAuth callback handling and JWT token generation/validation
- Route authentication supporting both API key and JWT methods
These changes are not visible in the provided file.
pytest.ini (1)
32-32: LGTM!The addition of the OpenAPI endpoint authentication rule aligns with the OAuth/JWT-based authentication refactor. The test configuration correctly mirrors the new auth requirements.
tests/api/test_echo.py (1)
19-20: LGTM!The change from 403 to 401 is semantically correct—401 Unauthorized is appropriate when credentials are missing or invalid, whereas 403 would indicate the credentials were valid but access is denied. The error message format is now consistent with
test_echo_with_error_api_key.src/framex/driver/application.py (2)
25-25: LGTM!Clean import of the new authentication utilities from the centralized auth module.
97-102: LGTM!The OAuth callback route is conditionally registered only when OAuth is enabled, following the configuration-driven approach. The route correctly uses
settings.auth.oauth.redirect_urias the path and wires upoauth_callbackas the handler.src/framex/driver/ingress.py (3)
5-5: LGTM!Added
Requestimport to support the updated authentication signature.
15-15: LGTM!Centralizing
api_key_headerandauth_jwtin the auth module promotes consistency and avoids duplication across ingress and application modules.
116-122: Closure capture is safe and logic is correct.
_verify_api_keysafely capturesauth_keysbecause it's a local parameter toregister_route—each invocation creates a distinct binding. Any reassignment ofauth_keys(line 76) occurs before the nested function is defined (line 116), and no further mutations happen after. The authentication logic correctly denies access only when both the API key is invalid AND JWT validation fails.src/framex/config.py (2)
1-1: LGTM!The
secretsmodule is the correct choice for cryptographically secure random generation.
86-86: LGTM!Making OAuth an optional field with
Nonedefault is appropriate—it allows the existing API key-based auth to continue working when OAuth is not configured.src/framex/driver/auth.py (1)
31-47: LGTM!The
auth_jwtfunction correctly handles the case when OAuth is disabled and catches specific JWT exceptions. This is a clean, defensive implementation.tests/driver/test_application.py (1)
11-11: Add integration tests for OAuth-protected docs endpoints or document that this is intentionally deferred.The import cleanup is correct—API_STR is the only constant needed in this test file. The OAuth/JWT-based docs protection is properly implemented in
application.py(lines 104–119 showDepends(authenticate)on all docs endpoints). However, no integration tests exist for this flow: there are no tests verifying that unauthenticated requests to docs endpoints redirect to OAuth, that JWT tokens are validated, or that the authenticate dependency actually protects these endpoints. Current tests only verify thatdocs_urlandopenapi_urlareNoneby default, but not the runtime authentication behavior.
| "loguru>=0.7.3", | ||
| "pydantic>=2.11.7", | ||
| "pydantic-settings>=2.10.1", | ||
| "pyjwt>=2.10.1", |
There was a problem hiding this comment.
🧩 Analysis chain
🌐 Web query:
PyJWT latest version PyPI 2.10.1
💡 Result:
The latest PyJWT release on PyPI is 2.10.1 (uploaded Nov 28, 2024). [1][2]
Sources:
[1] PyPI project page for PyJWT.
[2] PyJWT changelog.
🌐 Web query:
PyJWT security vulnerabilities CVE 2024 2025
💡 Result:
Summary — known CVEs for PyJWT (2024–2025)
-
CVE-2024-53861 — Issuer (iss) partial-comparison bug introduced in PyJWT 2.10.0 that allowed partial matches (e.g., "abc" matched "abc"). Patched in 2.10.1; upgrade recommended. [1][2]
-
CVE-2025-45768 — Reported July 31, 2025: weak/insufficient key length handling in JWS (CWE-311) affecting pyjwt 2.10.1 (disputed by the supplier). Scoring classified as High (CVSS ~7.0); vendors/trackers advise remediation or upgrading per vendor guidance. [3][4]
Recommended actions
- Upgrade to the vendor-fixed release(s) where available (CVE-2024-53861 → 2.10.1). For CVE-2025-45768 follow your vendor/security bulletin for the fixed version or mitigation guidance (enforce minimum key lengths, validate JWS algorithm/key choices). [1][3]
Sources
- NVD entry for CVE-2024-53861 (pyjwt issuer partial-comparison; fixed in 2.10.1). [NVD]. [1]
- Debian / security trackers summarizing CVE-2024-53861 and fix. [2]
- CVE-2025-45768 advisory / vulnerability entries (INCIBE / Wiz summary of weak key-length issue; published July 31, 2025). [3][4]
If you want, I can: (a) fetch the PyJWT project advisory/GitHub commits for each CVE, (b) check whether a fixed PyPI release for CVE-2025-45768 exists, or (c) produce a remediation checklist for your codebase.
Address CVE-2025-45768 security vulnerability in PyJWT 2.10.1 before merging.
PyJWT 2.10.1 is the latest stable release, but it contains an active HIGH-severity vulnerability (CVE-2025-45768) related to weak/insufficient key length handling in JWS (CVSS ~7.0). Until a patched version is available, enforce strict minimum key length requirements and validate all JWS algorithm/key choices in the authentication implementation.
🤖 Prompt for AI Agents
In pyproject.toml around line 16, the PyJWT dependency ("pyjwt>=2.10.1") is
flagged as vulnerable (CVE-2025-45768); instead of leaving the vulnerable spec,
update the project to either pin to a patched release when available (e.g.,
change the constraint to the patched minimum version once released) and/or
temporarily tighten requirements by replacing the loose spec with a constrained
range that excludes the vulnerable release, and implement runtime safeguards in
the auth code that validate JWS algorithm and key choices: explicitly reject
"none" alg, require HS keys to meet a minimum byte length for HS256/384/512,
require RSA keys >= 2048 bits and EC keys using secure curves only, add unit
tests exercising algorithm/key validation and update CI dependency scans to fail
the build if a vulnerable PyJWT is installed until a patched version is adopted.
| if not (auth_token := user_resp.get("username")): | ||
| raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Failed to get GitLab user") | ||
|
|
||
| user = user_resp.get("user") | ||
|
|
||
| user_info = { | ||
| "message": f"Welcome {user}", | ||
| "username": user, | ||
| "name": user_resp.get("name"), | ||
| "email": user_resp.get("email"), | ||
| } |
There was a problem hiding this comment.
Bug: Variable shadowing causes incorrect user extraction.
On line 113, auth_token is reassigned to user_resp.get("username"), shadowing the access token. Then on line 116, user is extracted from user_resp.get("user") which is likely incorrect—the username was already retrieved.
Based on the GitLab API, the response likely has username at the top level, not nested under user. This causes user_info["message"] and user_info["username"] to be None.
🔎 Proposed fix
- if not (auth_token := user_resp.get("username")):
+ username = user_resp.get("username")
+ if not username:
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Failed to get GitLab user")
- user = user_resp.get("user")
-
user_info = {
- "message": f"Welcome {user}",
- "username": user,
+ "message": f"Welcome {username}",
+ "username": username,
"name": user_resp.get("name"),
"email": user_resp.get("email"),
}🤖 Prompt for AI Agents
In src/framex/driver/auth.py around lines 113 to 123, the code mistakenly
reassigns auth_token from user_resp.get("username") and then reads user from
user_resp.get("user"), causing username to be None; restore the access token
retrieval (do not shadow auth_token), read the username directly from
user_resp.get("username") into a new variable (e.g., username), validate it and
raise the HTTPException if missing, and populate user_info using that username
(and other top-level fields like name and email) instead of
user_resp.get("user").
|
Tip For best results, initiate chat on the files or code changes. 🧩 Analysis chain🏁 Script executed: Length of output: 19022 @touale, I'll generate comprehensive tests for the new authentication functionality in PR #42. Based on the changes, I'll create tests for the new # tests/driver/test_auth.py
"""Tests for authentication module."""
from datetime import UTC, datetime, timedelta
from unittest.mock import AsyncMock, Mock, patch
import jwt
import pytest
from fastapi import HTTPException, status
from fastapi.testclient import TestClient
from starlette.requests import Request
from framex.config import settings
from framex.driver.application import create_fastapi_application
from framex.driver.auth import authenticate, auth_jwt, create_jwt, oauth_callback
class TestCreateJWT:
"""Tests for JWT creation."""
def test_create_jwt_success(self):
"""Test successful JWT creation with valid payload."""
payload = {"username": "testuser", "email": "test@example.com"}
with patch("framex.config.settings.auth.oauth") as mock_oauth:
mock_oauth.jwt_secret = "test_secret"
mock_oauth.jwt_algorithm = "HS256"
token = create_jwt(payload)
assert token is not None
decoded = jwt.decode(token, "test_secret", algorithms=["HS256"])
assert decoded["username"] == "testuser"
assert decoded["email"] == "test@example.com"
assert "iat" in decoded
assert "exp" in decoded
def test_create_jwt_expiration(self):
"""Test JWT has correct expiration time (24 hours)."""
payload = {"username": "testuser"}
with patch("framex.config.settings.auth.oauth") as mock_oauth:
mock_oauth.jwt_secret = "test_secret"
mock_oauth.jwt_algorithm = "HS256"
token = create_jwt(payload)
decoded = jwt.decode(token, "test_secret", algorithms=["HS256"])
exp_time = datetime.fromtimestamp(decoded["exp"], tz=UTC)
iat_time = datetime.fromtimestamp(decoded["iat"], tz=UTC)
time_diff = exp_time - iat_time
# Should be 24 hours
assert time_diff.total_seconds() == pytest.approx(24 * 60 * 60, rel=1)
def test_create_jwt_no_oauth_configured(self):
"""Test JWT creation fails when OAuth is not configured."""
with patch("framex.config.settings.auth.oauth", None):
with pytest.raises(RuntimeError, match="OAuth not configured"):
create_jwt({"username": "test"})
class TestAuthJWT:
"""Tests for JWT authentication validation."""
def test_auth_jwt_valid_token(self):
"""Test authentication succeeds with valid JWT token."""
with patch("framex.config.settings.auth.oauth") as mock_oauth:
mock_oauth.jwt_secret = "test_secret"
mock_oauth.jwt_algorithm = "HS256"
# Create a valid token
now = datetime.now(UTC)
payload = {
"username": "testuser",
"iat": int(now.timestamp()),
"exp": int((now + timedelta(hours=1)).timestamp()),
}
token = jwt.encode(payload, "test_secret", algorithm="HS256")
# Mock request with token cookie
mock_request = Mock(spec=Request)
mock_request.cookies.get.return_value = token
result = auth_jwt(mock_request)
assert result is True
def test_auth_jwt_expired_token(self):
"""Test authentication fails with expired JWT token."""
with patch("framex.config.settings.auth.oauth") as mock_oauth:
mock_oauth.jwt_secret = "test_secret"
mock_oauth.jwt_algorithm = "HS256"
# Create an expired token
now = datetime.now(UTC)
payload = {
"username": "testuser",
"iat": int((now - timedelta(hours=25)).timestamp()),
"exp": int((now - timedelta(hours=1)).timestamp()),
}
token = jwt.encode(payload, "test_secret", algorithm="HS256")
mock_request = Mock(spec=Request)
mock_request.cookies.get.return_value = token
result = auth_jwt(mock_request)
assert result is False
def test_auth_jwt_invalid_token(self):
"""Test authentication fails with invalid JWT token."""
with patch("framex.config.settings.auth.oauth") as mock_oauth:
mock_oauth.jwt_secret = "test_secret"
mock_oauth.jwt_algorithm = "HS256"
mock_request = Mock(spec=Request)
mock_request.cookies.get.return_value = "invalid.token.here"
result = auth_jwt(mock_request)
assert result is False
def test_auth_jwt_no_token(self):
"""Test authentication fails when no token present."""
with patch("framex.config.settings.auth.oauth") as mock_oauth:
mock_oauth.jwt_secret = "test_secret"
mock_request = Mock(spec=Request)
mock_request.cookies.get.return_value = None
result = auth_jwt(mock_request)
assert result is False
def test_auth_jwt_no_oauth_configured(self):
"""Test authentication returns False when OAuth not configured."""
with patch("framex.config.settings.auth.oauth", None):
mock_request = Mock(spec=Request)
result = auth_jwt(mock_request)
assert result is False
class TestAuthenticate:
"""Tests for authenticate dependency function."""
def test_authenticate_with_valid_jwt(self):
"""Test authentication succeeds with valid JWT token."""
with patch("framex.config.settings.auth.oauth") as mock_oauth:
mock_oauth.jwt_secret = "test_secret"
mock_oauth.jwt_algorithm = "HS256"
now = datetime.now(UTC)
payload = {
"username": "testuser",
"iat": int(now.timestamp()),
"exp": int((now + timedelta(hours=1)).timestamp()),
}
token = jwt.encode(payload, "test_secret", algorithm="HS256")
mock_request = Mock(spec=Request)
mock_request.cookies.get.return_value = token
# Should not raise exception
result = authenticate(mock_request, api_key=None)
assert result is None
def test_authenticate_with_valid_api_key(self):
"""Test authentication succeeds with valid API key."""
with patch("framex.config.settings.auth.oauth") as mock_oauth, \
patch("framex.config.settings.auth.get_auth_keys") as mock_get_keys:
mock_oauth.jwt_secret = "test_secret"
mock_oauth.jwt_algorithm = "HS256"
mock_get_keys.return_value = ["valid_api_key"]
mock_request = Mock(spec=Request)
mock_request.cookies.get.return_value = None
mock_request.url.path = "/api/v1/test"
# Should not raise exception
result = authenticate(mock_request, api_key="valid_api_key")
assert result is None
def test_authenticate_redirects_to_oauth_when_no_credentials(self):
"""Test authentication redirects to OAuth when no valid credentials."""
with patch("framex.config.settings.auth.oauth") as mock_oauth, \
patch("framex.config.settings.auth.get_auth_keys") as mock_get_keys:
mock_oauth.authorization_url = "https://oauth.example.com/authorize"
mock_oauth.client_id = "test_client_id"
mock_oauth.call_back_url = "http://localhost/callback"
mock_get_keys.return_value = ["valid_key"]
mock_request = Mock(spec=Request)
mock_request.cookies.get.return_value = None
mock_request.url.path = "/api/v1/docs"
with pytest.raises(HTTPException) as exc_info:
authenticate(mock_request, api_key=None)
assert exc_info.value.status_code == status.HTTP_301_MOVED_PERMANENTLY
location = exc_info.value.headers["Location"]
assert "https://oauth.example.com/authorize" in location
assert "client_id=test_client_id" in location
assert "response_type=code" in location
def test_authenticate_with_invalid_jwt(self):
"""Test authentication handles invalid JWT gracefully."""
with patch("framex.config.settings.auth.oauth") as mock_oauth, \
patch("framex.config.settings.auth.get_auth_keys") as mock_get_keys:
mock_oauth.jwt_secret = "test_secret"
mock_oauth.jwt_algorithm = "HS256"
mock_oauth.authorization_url = "https://oauth.example.com/authorize"
mock_oauth.client_id = "test_client_id"
mock_oauth.call_back_url = "http://localhost/callback"
mock_get_keys.return_value = []
mock_request = Mock(spec=Request)
mock_request.cookies.get.return_value = "invalid.jwt.token"
mock_request.url.path = "/api/v1/docs"
with pytest.raises(HTTPException) as exc_info:
authenticate(mock_request, api_key=None)
assert exc_info.value.status_code == status.HTTP_301_MOVED_PERMANENTLY
def test_authenticate_with_wrong_api_key(self):
"""Test authentication fails with wrong API key."""
with patch("framex.config.settings.auth.oauth") as mock_oauth, \
patch("framex.config.settings.auth.get_auth_keys") as mock_get_keys:
mock_oauth.authorization_url = "https://oauth.example.com/authorize"
mock_oauth.client_id = "test_client_id"
mock_oauth.call_back_url = "http://localhost/callback"
mock_get_keys.return_value = ["valid_key"]
mock_request = Mock(spec=Request)
mock_request.cookies.get.return_value = None
mock_request.url.path = "/api/v1/test"
with pytest.raises(HTTPException) as exc_info:
authenticate(mock_request, api_key="wrong_key")
assert exc_info.value.status_code == status.HTTP_301_MOVED_PERMANENTLY
class TestOAuthCallback:
"""Tests for OAuth callback handler."""
@pytest.mark.asyncio
async def test_oauth_callback_success(self):
"""Test successful OAuth callback flow."""
with patch("framex.config.settings.auth.oauth") as mock_oauth, \
patch("httpx.AsyncClient") as mock_client_class:
# Setup OAuth config
mock_oauth.token_url = "https://oauth.example.com/token"
mock_oauth.user_info_url = "https://oauth.example.com/user"
mock_oauth.client_id = "test_client"
mock_oauth.client_secret = "test_secret"
mock_oauth.call_back_url = "http://localhost/callback"
mock_oauth.jwt_secret = "jwt_secret"
mock_oauth.jwt_algorithm = "HS256"
# Mock httpx responses
mock_client = AsyncMock()
mock_client_class.return_value.__aenter__.return_value = mock_client
# Mock token exchange response
token_response = Mock()
token_response.json.return_value = {"access_token": "oauth_token_123"}
token_response.raise_for_status = Mock()
# Mock user info response
user_response = Mock()
user_response.json.return_value = {
"username": "testuser",
"user": "testuser",
"name": "Test User",
"email": "test@example.com",
}
user_response.raise_for_status = Mock()
mock_client.post.return_value = token_response
mock_client.get.return_value = user_response
# Call the callback
response = await oauth_callback(code="auth_code_123")
# Verify response
assert response.status_code == status.HTTP_302_FOUND
assert response.headers["location"] == "/api/v1/docs"
# Verify token cookie is set
assert "token" in response.headers.get("set-cookie", "")
@pytest.mark.asyncio
async def test_oauth_callback_token_exchange_fails(self):
"""Test OAuth callback handles token exchange failure."""
with patch("framex.config.settings.auth.oauth") as mock_oauth, \
patch("httpx.AsyncClient") as mock_client_class:
mock_oauth.token_url = "https://oauth.example.com/token"
mock_oauth.client_id = "test_client"
mock_oauth.client_secret = "test_secret"
mock_oauth.call_back_url = "http://localhost/callback"
mock_client = AsyncMock()
mock_client_class.return_value.__aenter__.return_value = mock_client
# Mock failed token response (no access_token)
token_response = Mock()
token_response.json.return_value = {"error": "invalid_grant"}
token_response.raise_for_status = Mock()
mock_client.post.return_value = token_response
with pytest.raises(HTTPException) as exc_info:
await oauth_callback(code="bad_code")
assert exc_info.value.status_code == status.HTTP_400_BAD_REQUEST
assert "token exchange failed" in exc_info.value.detail.lower()
@pytest.mark.asyncio
async def test_oauth_callback_user_info_fails(self):
"""Test OAuth callback handles user info retrieval failure."""
with patch("framex.config.settings.auth.oauth") as mock_oauth, \
patch("httpx.AsyncClient") as mock_client_class:
mock_oauth.token_url = "https://oauth.example.com/token"
mock_oauth.user_info_url = "https://oauth.example.com/user"
mock_oauth.client_id = "test_client"
mock_oauth.client_secret = "test_secret"
mock_oauth.call_back_url = "http://localhost/callback"
mock_client = AsyncMock()
mock_client_class.return_value.__aenter__.return_value = mock_client
# Mock successful token response
token_response = Mock()
token_response.json.return_value = {"access_token": "oauth_token_123"}
token_response.raise_for_status = Mock()
# Mock failed user info response (no username)
user_response = Mock()
user_response.json.return_value = {"error": "unauthorized"}
user_response.raise_for_status = Mock()
mock_client.post.return_value = token_response
mock_client.get.return_value = user_response
with pytest.raises(HTTPException) as exc_info:
await oauth_callback(code="auth_code")
assert exc_info.value.status_code == status.HTTP_400_BAD_REQUEST
assert "failed to get" in exc_info.value.detail.lower()
@pytest.mark.asyncio
async def test_oauth_callback_no_oauth_configured(self):
"""Test OAuth callback fails when OAuth not configured."""
with patch("framex.config.settings.auth.oauth", None):
with pytest.raises(RuntimeError, match="OAuth not configured"):
await oauth_callback(code="some_code")
class TestOAuthConfig:
"""Tests for OAuth configuration."""
def test_oauth_config_callback_url_property(self):
"""Test OAuth config generates correct callback URL."""
from framex.config import OauthConfig
config = OauthConfig(
app_url="https://example.com",
redirect_uri="/auth/callback",
)
assert config.call_back_url == "https://example.com/auth/callback"
def test_oauth_config_default_urls(self):
"""Test OAuth config sets default URLs based on base_url."""
from framex.config import OauthConfig
config = OauthConfig(base_url="https://gitlab.example.com")
assert config.authorization_url == "https://gitlab.example.com/oauth/authorize"
assert config.token_url == "https://gitlab.example.com/oauth/token"
assert config.user_info_url == "https://gitlab.example.com/api/v4/user"
def test_oauth_config_jwt_secret_generation(self):
"""Test OAuth config generates JWT secret if not provided."""
from framex.config import OauthConfig
config = OauthConfig()
assert config.jwt_secret != ""
assert len(config.jwt_secret) > 20 # Should be a reasonable length
def test_oauth_config_custom_urls(self):
"""Test OAuth config uses custom URLs when provided."""
from framex.config import OauthConfig
config = OauthConfig(
base_url="https://gitlab.example.com",
authorization_url="https://custom.auth.url",
token_url="https://custom.token.url",
user_info_url="https://custom.user.url",
)
assert config.authorization_url == "https://custom.auth.url"
assert config.token_url == "https://custom.token.url"
assert config.user_info_url == "https://custom.user.url"
class TestAuthenticationIntegration:
"""Integration tests for authentication with FastAPI application."""
@pytest.fixture
def app(self):
return create_fastapi_application()
@pytest.fixture
def client(self, app):
return TestClient(app)
def test_docs_redirects_to_oauth_when_not_authenticated(self, client):
"""Test that docs endpoint redirects to OAuth when not authenticated."""
with patch("framex.config.settings.auth.oauth") as mock_oauth:
mock_oauth.authorization_url = "https://oauth.example.com/authorize"
mock_oauth.client_id = "test_client"
mock_oauth.call_back_url = "http://testserver/callback"
response = client.get("/api/v1/docs", follow_redirects=False)
# Should redirect (301)
assert response.status_code == status.HTTP_301_MOVED_PERMANENTLY
location = response.headers.get("location", "")
assert "oauth.example.com" in location
def test_docs_accessible_with_valid_jwt(self, client):
"""Test that docs endpoint is accessible with valid JWT token."""
with patch("framex.config.settings.auth.oauth") as mock_oauth:
mock_oauth.jwt_secret = "test_secret"
mock_oauth.jwt_algorithm = "HS256"
# Create valid token
now = datetime.now(UTC)
payload = {
"username": "testuser",
"iat": int(now.timestamp()),
"exp": int((now + timedelta(hours=1)).timestamp()),
}
token = jwt.encode(payload, "test_secret", algorithm="HS256")
response = client.get(
"/api/v1/docs",
cookies={"token": token},
)
assert response.status_code == status.HTTP_200_OK
assert "text/html" in response.headers["content-type"]This test file provides comprehensive coverage for:
The tests follow the existing patterns in your codebase and use proper mocking to avoid external dependencies. They cover both positive and negative scenarios to ensure robust authentication handling. |
There was a problem hiding this comment.
Actionable comments posted: 5
🧹 Nitpick comments (3)
tests/driver/test_auth.py (3)
21-34: Remove unusednoqadirective.The
noqa: C408directive on line 22 is unnecessary, as confirmed by static analysis.🔎 Proposed fix
- data = dict( # noqa: C408 + data = dict( authorization_url="https://oauth.example.com/authorize",
61-81: Consider adding edge case tests forauth_jwt.The current tests cover valid JWT and missing OAuth config. Consider adding tests for:
- Expired JWT (should return False)
- Invalid JWT signature (should return False)
- Missing token in cookies (should return False)
🔎 Example test for expired JWT
def test_expired_jwt(self): with patch("framex.config.settings.auth.oauth", fake_oauth()): # Create JWT that expired 1 hour ago past = datetime.now(UTC) - timedelta(hours=2) token = jwt.encode( { "username": "test", "iat": int((past - timedelta(hours=1)).timestamp()), "exp": int(past.timestamp()), }, "secret", algorithm="HS256", ) req = Mock(spec=Request) req.cookies.get.return_value = token assert auth_jwt(req) is False
115-137: Consider adding failure scenario tests foroauth_callback.The success path is well covered. Consider adding tests for:
- Token exchange failure (when
access_tokenis missing from response)- User info retrieval failure (when
usernameis missing from response)- OAuth not configured (should raise RuntimeError)
These would ensure robust error handling throughout the OAuth flow.
📜 Review details
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (3)
ruff.tomlsrc/framex/driver/auth.pytests/driver/test_auth.py
🧰 Additional context used
🧬 Code graph analysis (2)
tests/driver/test_auth.py (2)
src/framex/driver/application.py (1)
create_fastapi_application(51-185)src/framex/driver/auth.py (4)
auth_jwt(31-47)authenticate(50-80)create_jwt(16-28)oauth_callback(83-128)
src/framex/driver/auth.py (1)
src/framex/config.py (2)
get_auth_keys(97-116)call_back_url(70-71)
🪛 ast-grep (0.40.3)
tests/driver/test_auth.py
[warning] 64-72: Hardcoded JWT secret or private key is used. This is a Insufficiently Protected Credentials weakness: https://cwe.mitre.org/data/definitions/522.html Consider using an appropriate security mechanism to protect the credentials (e.g. keeping secrets in environment variables).
Context: token = jwt.encode(
{
"username": "test",
"iat": int(now.timestamp()),
"exp": int((now + timedelta(hours=1)).timestamp()),
},
"secret",
algorithm="HS256",
)
Note: [CWE-522] Insufficiently Protected Credentials. [REFERENCES]
- https://semgrep.dev/blog/2020/hardcoded-secrets-unverified-tokens-and-other-common-jwt-mistakes/
(jwt-python-hardcoded-secret-python)
[warning] 160-168: Hardcoded JWT secret or private key is used. This is a Insufficiently Protected Credentials weakness: https://cwe.mitre.org/data/definitions/522.html Consider using an appropriate security mechanism to protect the credentials (e.g. keeping secrets in environment variables).
Context: token = jwt.encode(
{
"username": "test",
"iat": int(now.timestamp()),
"exp": int((now + timedelta(hours=1)).timestamp()),
},
"secret",
algorithm="HS256",
)
Note: [CWE-522] Insufficiently Protected Credentials. [REFERENCES]
- https://semgrep.dev/blog/2020/hardcoded-secrets-unverified-tokens-and-other-common-jwt-mistakes/
(jwt-python-hardcoded-secret-python)
[warning] 45-45: A secret is hard-coded in the application. Secrets stored in source code, such as credentials, identifiers, and other types of sensitive data, can be leaked and used by internal or external malicious actors. Use environment variables to securely provide credentials and other secrets or retrieve them from a secure vault or Hardware Security Module (HSM).
Context: jwt.decode(token, "secret", algorithms=["HS256"])
Note: [CWE-798]: Use of Hard-coded Credentials [OWASP A01:2021]: Identification and Authentication Failures [REFERENCES]
https://cheatsheetseries.owasp.org/cheatsheets/Secrets_Management_Cheat_Sheet.html
(python-pyjwt-hardcoded-secret-python)
[warning] 64-72: A secret is hard-coded in the application. Secrets stored in source code, such as credentials, identifiers, and other types of sensitive data, can be leaked and used by internal or external malicious actors. Use environment variables to securely provide credentials and other secrets or retrieve them from a secure vault or Hardware Security Module (HSM).
Context: jwt.encode(
{
"username": "test",
"iat": int(now.timestamp()),
"exp": int((now + timedelta(hours=1)).timestamp()),
},
"secret",
algorithm="HS256",
)
Note: [CWE-798]: Use of Hard-coded Credentials [OWASP A01:2021]: Identification and Authentication Failures [REFERENCES]
https://cheatsheetseries.owasp.org/cheatsheets/Secrets_Management_Cheat_Sheet.html
(python-pyjwt-hardcoded-secret-python)
[warning] 160-168: A secret is hard-coded in the application. Secrets stored in source code, such as credentials, identifiers, and other types of sensitive data, can be leaked and used by internal or external malicious actors. Use environment variables to securely provide credentials and other secrets or retrieve them from a secure vault or Hardware Security Module (HSM).
Context: jwt.encode(
{
"username": "test",
"iat": int(now.timestamp()),
"exp": int((now + timedelta(hours=1)).timestamp()),
},
"secret",
algorithm="HS256",
)
Note: [CWE-798]: Use of Hard-coded Credentials [OWASP A01:2021]: Identification and Authentication Failures [REFERENCES]
https://cheatsheetseries.owasp.org/cheatsheets/Secrets_Management_Cheat_Sheet.html
(python-pyjwt-hardcoded-secret-python)
🪛 Ruff (0.14.10)
tests/driver/test_auth.py
22-22: Unused noqa directive (non-enabled: C408)
Remove unused noqa directive
(RUF100)
src/framex/driver/auth.py
18-18: Avoid specifying long messages outside the exception class
(TRY003)
45-45: Consider moving this statement to an else block
(TRY300)
59-59: Consider moving this statement to an else block
(TRY300)
61-61: Do not catch blind exception: Exception
(BLE001)
85-85: Avoid specifying long messages outside the exception class
(TRY003)
🔇 Additional comments (7)
ruff.toml (1)
82-82: LGTM!Adding ANN003 to the test file ignores is appropriate, as strict type annotations for
**kwargsin test fixtures can be overly verbose without significant benefit.tests/driver/test_auth.py (3)
42-53: LGTM!The tests for
create_jwtcover both the success path and the error case when OAuth is not configured. The validation of JWT payload fields (username, iat, exp) is thorough.
89-108: LGTM!The tests cover the key authentication scenarios: redirect when unauthenticated and successful authentication with a valid API key.
144-172: LGTM!The integration tests effectively validate the end-to-end authentication flow, including OAuth redirects and JWT-based access to protected endpoints.
Note: Static analysis warnings about hardcoded secrets in this file are false positives—these are test fixtures, not production credentials.
src/framex/driver/auth.py (3)
16-28: LGTM!The JWT creation logic correctly adds issued-at and expiration timestamps, and appropriately raises an error when OAuth is not configured.
31-47: LGTM!The JWT validation logic correctly handles all edge cases: missing OAuth config, missing token, and invalid/expired tokens.
112-119: LGTM!The user info extraction and validation logic correctly handles missing username and builds the user_info payload. The previous variable shadowing issue has been resolved.
| def authenticate(request: Request, api_key: str | None = Depends(api_key_header)) -> None: | ||
| if settings.auth.oauth: | ||
| if token := request.cookies.get("token"): | ||
| try: | ||
| jwt.decode( | ||
| token, | ||
| settings.auth.oauth.jwt_secret, | ||
| algorithms=[settings.auth.oauth.jwt_algorithm], | ||
| ) | ||
| return | ||
|
|
||
| except Exception as e: | ||
| from framex.log import logger | ||
|
|
||
| logger.warning(f"JWT decode failed: {e}") | ||
|
|
||
| if api_key and api_key in (settings.auth.get_auth_keys(request.url.path) or []): | ||
| return | ||
|
|
||
| raise HTTPException( | ||
| status_code=status.HTTP_301_MOVED_PERMANENTLY, | ||
| headers={ | ||
| "Location": ( | ||
| f"{settings.auth.oauth.authorization_url}" | ||
| f"?client_id={settings.auth.oauth.client_id}" | ||
| "&response_type=code" | ||
| f"&redirect_uri={settings.auth.oauth.call_back_url}" | ||
| "&scope=read_user" | ||
| ) | ||
| }, | ||
| ) |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Search for settings.auth.oauth configuration and usage patterns
rg -n "settings\.auth\.oauth" --type py -A 3 -B 3Repository: touale/FrameX-kit
Length of output: 9430
🏁 Script executed:
#!/bin/bash
# Search for API key authentication and get_auth_keys usage
rg -n "get_auth_keys" --type py -A 3 -B 3Repository: touale/FrameX-kit
Length of output: 4485
🏁 Script executed:
#!/bin/bash
# Look for the settings definition and OAuth configuration
fd -e py | xargs grep -l "class.*Settings\|auth.*oauth" | head -20Repository: touale/FrameX-kit
Length of output: 166
🏁 Script executed:
#!/bin/bash
# Search for authenticate function calls and other auth functions
rg -n "def authenticate" --type py -A 10Repository: touale/FrameX-kit
Length of output: 782
🏁 Script executed:
#!/bin/bash
# Check the full authenticate function and surrounding context
sed -n '50,82p' src/framex/driver/auth.pyRepository: touale/FrameX-kit
Length of output: 1181
🏁 Script executed:
#!/bin/bash
# Look for how api_key_header dependency works
rg -n "api_key_header" --type py -B 5 -A 5Repository: touale/FrameX-kit
Length of output: 3206
🏁 Script executed:
#!/bin/bash
# Search for any tests or logic handling authenticate when oauth is None
rg -n "def test.*authenticate\|authenticate.*oauth.*None\|settings\.auth\s*=" --type py -A 3 -B 1Repository: touale/FrameX-kit
Length of output: 43
🏁 Script executed:
#!/bin/bash
# Look at how authenticate is used as a dependency in the application
rg -n "Depends(authenticate)" --type py -B 3 -A 3Repository: touale/FrameX-kit
Length of output: 43
🏁 Script executed:
#!/bin/bash
# Check the settings.auth configuration to understand defaults
cat src/framex/config.py | head -150Repository: touale/FrameX-kit
Length of output: 4600
🏁 Script executed:
#!/bin/bash
# Get the full settings.auth configuration definition
rg -n "class AuthConfig\|oauth\s*:" src/framex/config.py -A 20 | head -80Repository: touale/FrameX-kit
Length of output: 43
API key validation is completely skipped when OAuth is not configured.
The authenticate() function gates all authentication logic—including API key validation—behind if settings.auth.oauth:. Since OAuth is optional by design (oauth: OauthConfig | None), disabling it breaks API-key-only authentication entirely. The API key parameter is extracted but never validated in this scenario. The codebase demonstrates that API key validation should work independently of OAuth (see ingress.py line 117). Either move API key validation outside the OAuth block or add fallback handling for when OAuth is disabled.
🧰 Tools
🪛 Ruff (0.14.10)
59-59: Consider moving this statement to an else block
(TRY300)
61-61: Do not catch blind exception: Exception
(BLE001)
| except Exception as e: | ||
| from framex.log import logger | ||
|
|
||
| logger.warning(f"JWT decode failed: {e}") |
There was a problem hiding this comment.
Replace broad exception catch with specific JWT exceptions.
Catching Exception on line 61 masks potential bugs and security issues. Specifically catch JWT-related exceptions instead.
🔎 Proposed fix
- except Exception as e:
+ except (jwt.InvalidTokenError, jwt.ExpiredSignatureError) as e:
from framex.log import logger
logger.warning(f"JWT decode failed: {e}")📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| except Exception as e: | |
| from framex.log import logger | |
| logger.warning(f"JWT decode failed: {e}") | |
| except (jwt.InvalidTokenError, jwt.ExpiredSignatureError) as e: | |
| from framex.log import logger | |
| logger.warning(f"JWT decode failed: {e}") |
🧰 Tools
🪛 Ruff (0.14.10)
61-61: Do not catch blind exception: Exception
(BLE001)
🤖 Prompt for AI Agents
In src/framex/driver/auth.py around lines 61 to 64, the code currently catches a
broad Exception when JWT decoding fails; replace this with specific PyJWT
exceptions (for example jwt.ExpiredSignatureError, jwt.DecodeError and/or
jwt.InvalidTokenError) by importing them from the jwt library and handling them
explicitly (log the appropriate warning for token expiration or invalid/failed
decode), and allow any other unexpected exceptions to propagate so they aren't
silently swallowed.
| async with httpx.AsyncClient() as client: | ||
| resp = await client.post( | ||
| settings.auth.oauth.token_url, | ||
| data={ | ||
| "client_id": settings.auth.oauth.client_id, | ||
| "client_secret": settings.auth.oauth.client_secret, | ||
| "grant_type": "authorization_code", | ||
| "code": code, | ||
| "redirect_uri": settings.auth.oauth.call_back_url, | ||
| }, | ||
| ) |
There was a problem hiding this comment.
Add timeout to token exchange POST request.
The POST request to the token URL (line 88) does not have a timeout, while the subsequent GET request to user_info_url does (line 107). Without a timeout, the request can hang indefinitely, degrading reliability and potentially causing resource exhaustion.
🔎 Proposed fix
resp = await client.post(
settings.auth.oauth.token_url,
data={
"client_id": settings.auth.oauth.client_id,
"client_secret": settings.auth.oauth.client_secret,
"grant_type": "authorization_code",
"code": code,
"redirect_uri": settings.auth.oauth.call_back_url,
},
+ timeout=10,
)📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| async with httpx.AsyncClient() as client: | |
| resp = await client.post( | |
| settings.auth.oauth.token_url, | |
| data={ | |
| "client_id": settings.auth.oauth.client_id, | |
| "client_secret": settings.auth.oauth.client_secret, | |
| "grant_type": "authorization_code", | |
| "code": code, | |
| "redirect_uri": settings.auth.oauth.call_back_url, | |
| }, | |
| ) | |
| async with httpx.AsyncClient() as client: | |
| resp = await client.post( | |
| settings.auth.oauth.token_url, | |
| data={ | |
| "client_id": settings.auth.oauth.client_id, | |
| "client_secret": settings.auth.oauth.client_secret, | |
| "grant_type": "authorization_code", | |
| "code": code, | |
| "redirect_uri": settings.auth.oauth.call_back_url, | |
| }, | |
| timeout=10, | |
| ) |
🤖 Prompt for AI Agents
In src/framex/driver/auth.py around lines 87 to 97, the token exchange POST
request is missing a timeout which can cause the call to hang; update the
client.post invocation to include an explicit timeout (match the same timeout
used for the subsequent user_info GET request) by passing the timeout argument
to client.post so the request fails fast and prevents resource exhaustion.
| res.set_cookie( | ||
| "token", | ||
| create_jwt(user_info), | ||
| httponly=True, | ||
| samesite="lax", | ||
| ) |
There was a problem hiding this comment.
Add secure=True flag to JWT cookie for production.
The JWT cookie is set without the secure flag, which means it can be transmitted over unencrypted HTTP connections. This exposes the authentication token to potential interception.
🔎 Proposed fix
res.set_cookie(
"token",
create_jwt(user_info),
httponly=True,
samesite="lax",
+ secure=True,
)📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| res.set_cookie( | |
| "token", | |
| create_jwt(user_info), | |
| httponly=True, | |
| samesite="lax", | |
| ) | |
| res.set_cookie( | |
| "token", | |
| create_jwt(user_info), | |
| httponly=True, | |
| samesite="lax", | |
| secure=True, | |
| ) |
🤖 Prompt for AI Agents
In src/framex/driver/auth.py around lines 122 to 127, the JWT cookie is being
set without the secure flag; update the res.set_cookie call to include
secure=True when running in production (e.g., check an environment/config flag
like settings.DEBUG or os.getenv("ENV") != "development", or detect request
scheme == "https") so the cookie is only sent over HTTPS; keep httponly and
samesite as-is and ensure the conditional sets secure=False only for local/dev
to allow testing.
There was a problem hiding this comment.
Actionable comments posted: 0
♻️ Duplicate comments (5)
src/framex/driver/auth.py (5)
50-80: API key validation is still completely skipped when OAuth is not configured.This critical issue was flagged in a previous review but remains unresolved. When
settings.auth.oauthisNone, the entire authentication function returns without validating the API key, even though the API key parameter is extracted. This breaks API-key-only authentication scenarios.
61-64: Replace broad exception catch with specific JWT exceptions.This issue was flagged in a previous review but remains unresolved. Line 61 still catches
Exception, which masks potential bugs. Theauth_jwtfunction (line 46) correctly demonstrates catching specific exceptions:jwt.InvalidTokenErrorandjwt.ExpiredSignatureError.
69-70: Use HTTP 302 or 303 for authentication redirects, not 301.This critical issue was flagged in a previous review but remains unresolved. Using
301 MOVED_PERMANENTLYcan cause browsers and proxies to cache the redirect, bypassing authentication when users later provide valid credentials.
87-97: Add timeout to token exchange POST request.This issue was flagged in a previous review but remains unresolved. The POST request lacks a timeout while the subsequent GET request (line 107) has
timeout=10. Without a timeout, the request can hang indefinitely.
122-127: Addsecure=Trueflag to JWT cookie for production.This issue was flagged in a previous review but remains unresolved. The JWT cookie is set without the
secureflag, allowing transmission over unencrypted HTTP and exposing the token to potential interception.
🧹 Nitpick comments (3)
tests/driver/test_auth.py (3)
21-34: Remove unusednoqadirective on line 22.The
noqa: C408directive is flagged as unused by Ruff. The suppression can be safely removed.🔎 Proposed fix
- data = dict( # noqa: C408 + data = dict( authorization_url="https://oauth.example.com/authorize",
135-153: Consider adding more test cases forauthenticatefunction.The current tests cover redirect and valid API key scenarios. Consider adding tests for:
- Invalid API key (should redirect to OAuth)
- Valid JWT cookie (should allow access)
- Expired JWT with valid API key fallback
These additional tests would provide more comprehensive coverage of the authentication logic.
161-182: Consider adding error path tests foroauth_callback.The current test covers the success path. Consider adding tests for:
- Token exchange failure (POST returns 4xx/5xx)
- User info fetch failure (GET returns 4xx/5xx)
- Missing
access_tokenin token response- Missing
usernamein user info responseThese tests would verify proper error handling in the OAuth callback flow.
📜 Review details
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (2)
src/framex/driver/auth.pytests/driver/test_auth.py
🧰 Additional context used
🧬 Code graph analysis (2)
src/framex/driver/auth.py (1)
src/framex/config.py (2)
get_auth_keys(97-116)call_back_url(70-71)
tests/driver/test_auth.py (3)
src/framex/config.py (2)
AuthConfig(85-116)call_back_url(70-71)src/framex/driver/application.py (1)
create_fastapi_application(51-185)src/framex/driver/auth.py (4)
auth_jwt(31-47)authenticate(50-80)create_jwt(16-28)oauth_callback(83-128)
🪛 ast-grep (0.40.3)
tests/driver/test_auth.py
[warning] 45-45: A secret is hard-coded in the application. Secrets stored in source code, such as credentials, identifiers, and other types of sensitive data, can be leaked and used by internal or external malicious actors. Use environment variables to securely provide credentials and other secrets or retrieve them from a secure vault or Hardware Security Module (HSM).
Context: jwt.decode(token, "secret", algorithms=["HS256"])
Note: [CWE-798]: Use of Hard-coded Credentials [OWASP A01:2021]: Identification and Authentication Failures [REFERENCES]
https://cheatsheetseries.owasp.org/cheatsheets/Secrets_Management_Cheat_Sheet.html
(python-pyjwt-hardcoded-secret-python)
[warning] 82-90: A secret is hard-coded in the application. Secrets stored in source code, such as credentials, identifiers, and other types of sensitive data, can be leaked and used by internal or external malicious actors. Use environment variables to securely provide credentials and other secrets or retrieve them from a secure vault or Hardware Security Module (HSM).
Context: jwt.encode(
{
"username": "test",
"iat": int(now.timestamp()),
"exp": int((now + timedelta(hours=1)).timestamp()),
},
"secret",
algorithm="HS256",
)
Note: [CWE-798]: Use of Hard-coded Credentials [OWASP A01:2021]: Identification and Authentication Failures [REFERENCES]
https://cheatsheetseries.owasp.org/cheatsheets/Secrets_Management_Cheat_Sheet.html
(python-pyjwt-hardcoded-secret-python)
[warning] 113-121: A secret is hard-coded in the application. Secrets stored in source code, such as credentials, identifiers, and other types of sensitive data, can be leaked and used by internal or external malicious actors. Use environment variables to securely provide credentials and other secrets or retrieve them from a secure vault or Hardware Security Module (HSM).
Context: jwt.encode(
{
"username": "test",
"iat": int((now - timedelta(days=2)).timestamp()),
"exp": int((now - timedelta(days=1)).timestamp()),
},
"secret",
algorithm="HS256",
)
Note: [CWE-798]: Use of Hard-coded Credentials [OWASP A01:2021]: Identification and Authentication Failures [REFERENCES]
https://cheatsheetseries.owasp.org/cheatsheets/Secrets_Management_Cheat_Sheet.html
(python-pyjwt-hardcoded-secret-python)
[warning] 206-214: A secret is hard-coded in the application. Secrets stored in source code, such as credentials, identifiers, and other types of sensitive data, can be leaked and used by internal or external malicious actors. Use environment variables to securely provide credentials and other secrets or retrieve them from a secure vault or Hardware Security Module (HSM).
Context: jwt.encode(
{
"username": "test",
"iat": int(now.timestamp()),
"exp": int((now + timedelta(hours=1)).timestamp()),
},
"secret",
algorithm="HS256",
)
Note: [CWE-798]: Use of Hard-coded Credentials [OWASP A01:2021]: Identification and Authentication Failures [REFERENCES]
https://cheatsheetseries.owasp.org/cheatsheets/Secrets_Management_Cheat_Sheet.html
(python-pyjwt-hardcoded-secret-python)
[warning] 82-90: Hardcoded JWT secret or private key is used. This is a Insufficiently Protected Credentials weakness: https://cwe.mitre.org/data/definitions/522.html Consider using an appropriate security mechanism to protect the credentials (e.g. keeping secrets in environment variables).
Context: token = jwt.encode(
{
"username": "test",
"iat": int(now.timestamp()),
"exp": int((now + timedelta(hours=1)).timestamp()),
},
"secret",
algorithm="HS256",
)
Note: [CWE-522] Insufficiently Protected Credentials. [REFERENCES]
- https://semgrep.dev/blog/2020/hardcoded-secrets-unverified-tokens-and-other-common-jwt-mistakes/
(jwt-python-hardcoded-secret-python)
[warning] 113-121: Hardcoded JWT secret or private key is used. This is a Insufficiently Protected Credentials weakness: https://cwe.mitre.org/data/definitions/522.html Consider using an appropriate security mechanism to protect the credentials (e.g. keeping secrets in environment variables).
Context: expired_token = jwt.encode(
{
"username": "test",
"iat": int((now - timedelta(days=2)).timestamp()),
"exp": int((now - timedelta(days=1)).timestamp()),
},
"secret",
algorithm="HS256",
)
Note: [CWE-522] Insufficiently Protected Credentials. [REFERENCES]
- https://semgrep.dev/blog/2020/hardcoded-secrets-unverified-tokens-and-other-common-jwt-mistakes/
(jwt-python-hardcoded-secret-python)
[warning] 206-214: Hardcoded JWT secret or private key is used. This is a Insufficiently Protected Credentials weakness: https://cwe.mitre.org/data/definitions/522.html Consider using an appropriate security mechanism to protect the credentials (e.g. keeping secrets in environment variables).
Context: token = jwt.encode(
{
"username": "test",
"iat": int(now.timestamp()),
"exp": int((now + timedelta(hours=1)).timestamp()),
},
"secret",
algorithm="HS256",
)
Note: [CWE-522] Insufficiently Protected Credentials. [REFERENCES]
- https://semgrep.dev/blog/2020/hardcoded-secrets-unverified-tokens-and-other-common-jwt-mistakes/
(jwt-python-hardcoded-secret-python)
🪛 Ruff (0.14.10)
src/framex/driver/auth.py
18-18: Avoid specifying long messages outside the exception class
(TRY003)
45-45: Consider moving this statement to an else block
(TRY300)
59-59: Consider moving this statement to an else block
(TRY300)
61-61: Do not catch blind exception: Exception
(BLE001)
85-85: Avoid specifying long messages outside the exception class
(TRY003)
tests/driver/test_auth.py
22-22: Unused noqa directive (non-enabled: C408)
Remove unused noqa directive
(RUF100)
There was a problem hiding this comment.
Actionable comments posted: 0
🧹 Nitpick comments (1)
tests/test_config.py (1)
31-35: LGTM! JWT secret generation test validates auto-generation.The test correctly validates that a JWT secret is automatically generated when not provided. The length check (>= 32) is appropriate for
secrets.token_urlsafe(32), which produces a base64-encoded string of approximately 43 characters.Optional: Consider testing uniqueness
You could optionally add an assertion to verify that multiple instances generate different secrets:
def test_oauth_config_generates_unique_jwt_secrets(): cfg1 = OauthConfig() cfg2 = OauthConfig() assert cfg1.jwt_secret != cfg2.jwt_secret
📜 Review details
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (1)
tests/test_config.py
🧰 Additional context used
🧬 Code graph analysis (1)
tests/test_config.py (1)
src/framex/config.py (2)
OauthConfig(55-82)call_back_url(70-71)
🔇 Additional comments (4)
tests/test_config.py (4)
1-1: LGTM! Import updated to reflect OAuth refactoring.The import correctly reflects the transition from AuthConfig to OauthConfig as part of the authentication refactoring.
13-18: LGTM! Callback URL property test is correct.The test properly validates the
call_back_urlproperty concatenation logic.
21-28: LGTM! Default URL generation test is comprehensive.The test correctly validates that default OAuth URLs are generated from
base_urlwhen specific URLs are not provided.
38-48: LGTM! Custom URL preservation test is thorough.The test correctly validates that explicitly provided OAuth URLs are not overridden by the default generation logic, even when
base_urlis present.
1 participant
Summary by CodeRabbit
New Features
Bug Fixes
Chores
Tests
✏️ Tip: You can customize this high-level summary in your review settings.