76 lines
2.9 KiB
Python
76 lines
2.9 KiB
Python
"""Regression coverage for the prototype's python-jose to PyJWT migration."""
|
|
|
|
from datetime import UTC, datetime, timedelta
|
|
import importlib.util
|
|
from pathlib import Path
|
|
import sys
|
|
from types import ModuleType, SimpleNamespace
|
|
|
|
from fastapi import HTTPException
|
|
from fastapi.security import HTTPAuthorizationCredentials
|
|
import jwt
|
|
import pytest
|
|
|
|
|
|
@pytest.fixture
|
|
def auth(monkeypatch):
|
|
# Load only the authentication module, without Docker, GCP or real settings.
|
|
config = ModuleType("app.config")
|
|
config.settings = SimpleNamespace(
|
|
JWT_SECRET_KEY="test-only-signing-key-with-at-least-32-bytes",
|
|
JWT_ALGORITHM="HS256",
|
|
JWT_ACCESS_TOKEN_EXPIRE_MINUTES=60,
|
|
)
|
|
monkeypatch.setitem(sys.modules, "app.config", config)
|
|
path = (
|
|
Path(__file__).resolve().parents[2] / "playground/backend_manager/app/auth/jwt_handler.py"
|
|
)
|
|
spec = importlib.util.spec_from_file_location("backend_manager_jwt_test", path)
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_signed_token_preserves_claims_and_authenticated_user(auth):
|
|
token = auth.create_access_token("user@example.com", {"role": "tester"})
|
|
claims = auth.decode_access_token(token)
|
|
assert claims["sub"] == "user@example.com"
|
|
assert claims["role"] == "tester"
|
|
assert claims["iss"] == "artemis-backend-manager"
|
|
assert claims["exp"] > claims["iat"]
|
|
credentials = HTTPAuthorizationCredentials(scheme="Bearer", credentials=token)
|
|
assert await auth.get_current_user(credentials) == "user@example.com"
|
|
|
|
|
|
@pytest.mark.parametrize("invalid", ["expired", "signature", "algorithm", "unsigned", "malformed"])
|
|
def test_invalid_tokens_remain_unauthorized(auth, invalid):
|
|
claims = {"sub": "user", "exp": datetime.now(UTC) + timedelta(minutes=5)}
|
|
key = auth.settings.JWT_SECRET_KEY
|
|
algorithm = "HS256"
|
|
if invalid == "expired":
|
|
claims["exp"] = datetime.now(UTC) - timedelta(minutes=5)
|
|
elif invalid == "signature":
|
|
key = "different-test-only-signing-key-with-32-bytes"
|
|
elif invalid == "algorithm":
|
|
algorithm = "HS384"
|
|
key = "another-test-only-key-long-enough-for-hs384-signatures"
|
|
elif invalid == "unsigned":
|
|
algorithm, key = "none", None
|
|
token = jwt.encode(claims, key, algorithm=algorithm)
|
|
if invalid == "malformed":
|
|
token = "not-a-jwt"
|
|
with pytest.raises(HTTPException) as error:
|
|
auth.decode_access_token(token)
|
|
assert error.value.status_code == 401
|
|
assert error.value.headers == {"WWW-Authenticate": "Bearer"}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_missing_subject_is_rejected(auth):
|
|
token = jwt.encode({}, auth.settings.JWT_SECRET_KEY, algorithm="HS256")
|
|
credentials = HTTPAuthorizationCredentials(scheme="Bearer", credentials=token)
|
|
with pytest.raises(HTTPException) as error:
|
|
await auth.get_current_user(credentials)
|
|
assert error.value.status_code == 401
|