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
30 changes: 19 additions & 11 deletions auth/utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,7 @@
import jwt
import os
import secrets
from datetime import datetime, timedelta
from datetime import datetime, timedelta, timezone
from functools import wraps
from flask import request, jsonify, g
from flask_jwt_extended import get_jwt_identity, verify_jwt_in_request
Expand Down Expand Up @@ -34,28 +34,36 @@ def verify_password(password: str, hash_: str) -> bool:
class JWTUtils:
"""JWT token utilities"""

@staticmethod
def _get_secret_key() -> str:
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_key = JWTUtils._get_secret_key()
access_token = jwt.encode(
{
'user_id': user_id,
'username': username,
'exp': datetime.utcnow() + timedelta(hours=1),
'exp': datetime.now(timezone.utc) + timedelta(hours=1),
'type': 'access'
},
os.getenv('JWT_SECRET_KEY', 'jwt-secret'),
secret_key,
algorithm='HS256'
)

refresh_token = jwt.encode(
{
'user_id': user_id,
'username': username,
'exp': datetime.utcnow() + timedelta(days=30),
'exp': datetime.now(timezone.utc) + timedelta(days=30),
'type': 'refresh'
},
os.getenv('JWT_SECRET_KEY', 'jwt-secret'),
secret_key,
algorithm='HS256'
)

Expand All @@ -65,15 +73,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_key = JWTUtils._get_secret_key()
payload = jwt.decode(
token,
os.getenv('JWT_SECRET_KEY', 'jwt-secret'),
secret_key,
algorithms=['HS256']
)
return payload
except jwt.ExpiredSignatureError:
return None
except jwt.InvalidTokenError:
except (ValueError, jwt.ExpiredSignatureError, jwt.InvalidTokenError):
return None


Expand Down Expand Up @@ -128,8 +135,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
81 changes: 81 additions & 0 deletions tests/test_auth_utils.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,81 @@
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_models = MagicMock()
sys.modules["cache_db"] = mock_cache_db
sys.modules["cache_db.redis_client"] = mock_cache_db
mock_cache_db.redis_client = mock_redis_client
sys.modules["cache_db.models"] = mock_models
mock_models.User = MagicMock()
mock_models.RefreshToken = MagicMock()

import os
import pytest
from flask import Flask
from auth.utils import (
PasswordUtils,
JWTUtils,
SessionUtils,
login_required,
admin_required,
verify_required,
)


def test_password_utils():
hashed = PasswordUtils.hash_password("securepassword123")
assert PasswordUtils.verify_password("securepassword123", hashed) is True
assert PasswordUtils.verify_password("wrongpassword", hashed) is False

with pytest.raises(ValueError):
PasswordUtils.hash_password("short")


def test_jwt_utils_missing_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.jwt.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

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


def test_login_required_does_not_leak_error_details(monkeypatch):
app = Flask(__name__)
app.config["TESTING"] = True

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

client = app.test_client()

# Mock verify_jwt_in_request to raise an exception with internal sensitive info
def raise_sensitive_error():
raise RuntimeError("Sensitive DB / JWT Exception Internal Stack Trace Details")

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

response = client.get("/protected")
assert response.status_code == 401
json_data = response.get_json()
assert json_data == {"error": "Unauthorized"}
# Ensure details field is not present and no stack trace is leaked
assert "details" not in json_data
assert "Sensitive DB" not in response.get_data(as_text=True)