Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
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
4 changes: 4 additions & 0 deletions .jules/sentinel.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,4 @@
## 2026-03-31 - JWT Secret Enforcement and Auth Exception Sanitization
**Vulnerability:** JWT authentication fell back to a default hardcoded secret key (`'jwt-secret'`), and `login_required` leaked exception details (`str(e)`) in 401 response payloads.
**Learning:** Downstream services expecting `auth.utils` must provide `JWT_SECRET_KEY` in environment variables. Unit tests for `auth.utils` require mocking the external `cache_db` package before import.
**Prevention:** Always raise errors on missing security configuration in production helpers rather than relying on weak default secrets, and never return unhandled exception details to unauthenticated API clients.
25 changes: 17 additions & 8 deletions auth/utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -34,17 +34,26 @@ def verify_password(password: str, hash_: str) -> bool:
class JWTUtils:
"""JWT token utilities"""

@staticmethod
def _get_secret() -> str:
"""Get secret key for JWT signing/verification or raise ValueError if missing"""
secret = os.getenv('JWT_SECRET_KEY')
if not secret:
raise ValueError("JWT_SECRET_KEY environment variable is not set")
return secret

@staticmethod
def create_tokens(user_id: str, username: str) -> Tuple[str, str]:
"""Create access and refresh tokens"""
secret = JWTUtils._get_secret()
access_token = jwt.encode(
{
'user_id': user_id,
'username': username,
'exp': datetime.utcnow() + timedelta(hours=1),
'type': 'access'
},
os.getenv('JWT_SECRET_KEY', 'jwt-secret'),
secret,
algorithm='HS256'
)

Expand All @@ -55,7 +64,7 @@ def create_tokens(user_id: str, username: str) -> Tuple[str, str]:
'exp': datetime.utcnow() + timedelta(days=30),
'type': 'refresh'
},
os.getenv('JWT_SECRET_KEY', 'jwt-secret'),
secret,
algorithm='HS256'
)

Expand All @@ -65,15 +74,14 @@ def create_tokens(user_id: str, username: str) -> Tuple[str, str]:
def decode_token(token: str) -> Optional[dict]:
"""Decode and verify token"""
try:
secret = JWTUtils._get_secret()
payload = jwt.decode(
token,
os.getenv('JWT_SECRET_KEY', 'jwt-secret'),
secret,
algorithms=['HS256']
)
return payload
except jwt.ExpiredSignatureError:
return None
except jwt.InvalidTokenError:
except (jwt.ExpiredSignatureError, jwt.InvalidTokenError, ValueError):
return None


Expand Down Expand Up @@ -128,8 +136,9 @@ def decorated_function(*args, **kwargs):
g.user_id = user_id
g.user = user_data
return f(*args, **kwargs)
except Exception as e:
return jsonify({'error': 'Unauthorized', 'details': str(e)}), 401
except Exception:
# Do not leak internal exception details to unauthenticated callers
return jsonify({'error': 'Unauthorized'}), 401
return decorated_function


Expand Down
67 changes: 67 additions & 0 deletions tests/test_auth.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,67 @@
import os
import sys
from unittest.mock import MagicMock

# Mock external cache_db module before importing auth.utils
mock_cache_db = MagicMock()
mock_redis_client = MagicMock()
mock_user = MagicMock()
mock_refresh_token = MagicMock()

mock_cache_db.redis_client = mock_redis_client
mock_cache_db.models = MagicMock()
mock_cache_db.models.User = mock_user
mock_cache_db.models.RefreshToken = mock_refresh_token

sys.modules['cache_db'] = mock_cache_db
sys.modules['cache_db.redis_client'] = mock_cache_db.redis_client
sys.modules['cache_db.models'] = mock_cache_db.models

import pytest
from flask import Flask
from auth.utils import JWTUtils, PasswordUtils, login_required


def test_jwt_utils_requires_secret(monkeypatch):
monkeypatch.delenv('JWT_SECRET_KEY', raising=False)

with pytest.raises(ValueError, match="JWT_SECRET_KEY environment variable is not set"):
JWTUtils.create_tokens("user123", "alice")

assert JWTUtils.decode_token("some.invalid.token") is None


def test_jwt_utils_with_secret(monkeypatch):
monkeypatch.setenv('JWT_SECRET_KEY', 'super-secret-key-12345')

access_token, refresh_token = JWTUtils.create_tokens("user123", "alice")
assert access_token is not None
assert refresh_token is not None

payload = JWTUtils.decode_token(access_token)
assert payload is not None
assert payload["user_id"] == "user123"
assert payload["username"] == "alice"


def test_login_required_does_not_leak_error_details(monkeypatch):
app = Flask(__name__)

@app.route("/protected")
@login_required
def protected_route():
return "ok"

# Mock verify_jwt_in_request to raise an exception with internal details
def mock_verify():
raise Exception("Internal database connection failed: secret_db_uri")

monkeypatch.setattr("auth.utils.verify_jwt_in_request", mock_verify)

client = app.test_client()
response = client.get("/protected")

assert response.status_code == 401
json_data = response.get_json()
assert json_data == {"error": "Unauthorized"}
assert "details" not in json_data