246 lines
7.4 KiB
Python
246 lines
7.4 KiB
Python
import base64
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
|
|
import pytest
|
|
from cryptography.hazmat.primitives import serialization
|
|
from cryptography.hazmat.primitives.asymmetric import rsa
|
|
from jose import jwt as jose_jwt
|
|
|
|
from src.core.security import (
|
|
clear_jwks_cache,
|
|
create_refresh_token,
|
|
decode_keycloak_token,
|
|
verify_token,
|
|
)
|
|
from src.services.auth_service import AuthService
|
|
|
|
|
|
def _b64u_int(value: int) -> str:
|
|
raw = value.to_bytes((value.bit_length() + 7) // 8, "big")
|
|
return base64.urlsafe_b64encode(raw).rstrip(b"=").decode()
|
|
|
|
|
|
def test_decode_keycloak_token_rejects_garbage(monkeypatch):
|
|
clear_jwks_cache()
|
|
monkeypatch.setattr("src.core.security._get_jwks", lambda **_kwargs: {"keys": []})
|
|
assert decode_keycloak_token("not-a-jwt") is None
|
|
|
|
|
|
def test_decode_keycloak_token_accepts_valid_rs256(monkeypatch):
|
|
private_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
|
|
numbers = private_key.public_key().public_numbers()
|
|
kid = "test-key"
|
|
jwks = {
|
|
"keys": [
|
|
{
|
|
"kty": "RSA",
|
|
"kid": kid,
|
|
"use": "sig",
|
|
"alg": "RS256",
|
|
"n": _b64u_int(numbers.n),
|
|
"e": _b64u_int(numbers.e),
|
|
}
|
|
]
|
|
}
|
|
pem = private_key.private_bytes(
|
|
encoding=serialization.Encoding.PEM,
|
|
format=serialization.PrivateFormat.PKCS8,
|
|
encryption_algorithm=serialization.NoEncryption(),
|
|
)
|
|
token = jose_jwt.encode(
|
|
{"email": "admin@example.com", "sub": "admin"},
|
|
pem,
|
|
algorithm="RS256",
|
|
headers={"kid": kid},
|
|
)
|
|
clear_jwks_cache()
|
|
monkeypatch.setattr("src.core.security._get_jwks", lambda **_kwargs: jwks)
|
|
|
|
payload = decode_keycloak_token(token)
|
|
|
|
assert payload is not None
|
|
assert payload["email"] == "admin@example.com"
|
|
|
|
|
|
def test_decode_keycloak_token_strips_bearer_and_ignores_at_hash(monkeypatch):
|
|
private_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
|
|
numbers = private_key.public_key().public_numbers()
|
|
kid = "test-key"
|
|
jwks = {
|
|
"keys": [
|
|
{
|
|
"kty": "RSA",
|
|
"kid": kid,
|
|
"use": "sig",
|
|
"alg": "RS256",
|
|
"n": _b64u_int(numbers.n),
|
|
"e": _b64u_int(numbers.e),
|
|
}
|
|
]
|
|
}
|
|
pem = private_key.private_bytes(
|
|
encoding=serialization.Encoding.PEM,
|
|
format=serialization.PrivateFormat.PKCS8,
|
|
encryption_algorithm=serialization.NoEncryption(),
|
|
)
|
|
token = jose_jwt.encode(
|
|
{
|
|
"preferred_username": "admin@example.com",
|
|
"at_hash": "not-a-real-hash",
|
|
"sub": "admin",
|
|
},
|
|
pem,
|
|
algorithm="RS256",
|
|
headers={"kid": kid},
|
|
)
|
|
clear_jwks_cache()
|
|
monkeypatch.setattr("src.core.security._get_jwks", lambda **_kwargs: jwks)
|
|
|
|
payload = decode_keycloak_token(f"Bearer {token}")
|
|
|
|
assert payload is not None
|
|
from src.core.security import email_from_keycloak_payload
|
|
|
|
assert email_from_keycloak_payload(payload) == "admin@example.com"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_authenticate_user_returns_token(monkeypatch):
|
|
db = AsyncMock()
|
|
service = AuthService(db)
|
|
service.user_repo = MagicMock()
|
|
service.user_repo.authenticate = AsyncMock(
|
|
return_value=MagicMock(username="admin", email="admin@example.com")
|
|
)
|
|
|
|
token = await service.authenticate_user("admin", "admin")
|
|
|
|
assert token is not None
|
|
assert token.access_token
|
|
assert token.refresh_token
|
|
assert token.token_type == "bearer"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_refresh_token_returns_none_for_invalid():
|
|
db = AsyncMock()
|
|
service = AuthService(db)
|
|
|
|
token = await service.refresh_token("bad-token")
|
|
|
|
assert token is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_refresh_token_resolves_user_by_email():
|
|
db = AsyncMock()
|
|
service = AuthService(db)
|
|
service.user_repo = MagicMock()
|
|
service.user_repo.get_by_username = AsyncMock(return_value=None)
|
|
user = MagicMock(username="admin", email="admin@example.com", is_active=True)
|
|
service.user_repo.get_by_email = AsyncMock(return_value=user)
|
|
|
|
token = await service.refresh_token(
|
|
create_refresh_token(data={"sub": "admin@example.com"})
|
|
)
|
|
|
|
assert token is not None
|
|
assert token.token_type == "bearer"
|
|
payload = verify_token(token.refresh_token)
|
|
assert payload is not None
|
|
assert payload["sub"] == "admin@example.com"
|
|
service.user_repo.get_by_email.assert_awaited_once_with("admin@example.com")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_refresh_token_resolves_user_by_username():
|
|
db = AsyncMock()
|
|
service = AuthService(db)
|
|
service.user_repo = MagicMock()
|
|
user = MagicMock(username="admin", email="admin@example.com", is_active=True)
|
|
service.user_repo.get_by_username = AsyncMock(return_value=user)
|
|
service.user_repo.get_by_email = AsyncMock(return_value=None)
|
|
|
|
token = await service.refresh_token(create_refresh_token(data={"sub": "admin"}))
|
|
|
|
assert token is not None
|
|
payload = verify_token(token.refresh_token)
|
|
assert payload is not None
|
|
assert payload["sub"] == "admin"
|
|
service.user_repo.get_by_email.assert_not_awaited()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_authenticate_user_via_email_uses_username_subject():
|
|
db = AsyncMock()
|
|
service = AuthService(db)
|
|
service.user_repo = MagicMock()
|
|
service.user_repo.authenticate_via_email = AsyncMock(
|
|
return_value=MagicMock(
|
|
username="admin", email="admin@example.com", is_active=True
|
|
)
|
|
)
|
|
|
|
token = await service.authenticate_user_via_email("admin@example.com")
|
|
|
|
assert token is not None
|
|
payload = verify_token(token.access_token)
|
|
assert payload is not None
|
|
assert payload["sub"] == "admin"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_authenticate_user_via_email_rejects_inactive():
|
|
db = AsyncMock()
|
|
service = AuthService(db)
|
|
service.user_repo = MagicMock()
|
|
service.user_repo.authenticate_via_email = AsyncMock(
|
|
return_value=MagicMock(
|
|
username="admin", email="admin@example.com", is_active=False
|
|
)
|
|
)
|
|
|
|
token = await service.authenticate_user_via_email("admin@example.com")
|
|
|
|
assert token is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_authenticate_with_keycloak_token_passthrough(monkeypatch):
|
|
db = AsyncMock()
|
|
service = AuthService(db)
|
|
service.user_repo = MagicMock()
|
|
service.user_repo.authenticate_via_email = AsyncMock(
|
|
return_value=MagicMock(
|
|
username="admin", email="admin@example.com", is_active=True
|
|
)
|
|
)
|
|
monkeypatch.setattr(
|
|
"src.services.auth_service.decode_keycloak_token",
|
|
lambda _token: {"email": "admin@example.com"},
|
|
)
|
|
|
|
raw = "keycloak-access-token"
|
|
token = await service.authenticate_with_keycloak_token(raw)
|
|
|
|
assert token is not None
|
|
assert token.access_token == raw
|
|
assert token.refresh_token is None
|
|
service.user_repo.authenticate_via_email.assert_awaited_once_with(
|
|
"admin@example.com"
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_authenticate_with_keycloak_token_rejects_invalid(monkeypatch):
|
|
db = AsyncMock()
|
|
service = AuthService(db)
|
|
monkeypatch.setattr(
|
|
"src.services.auth_service.decode_keycloak_token",
|
|
lambda _token: None,
|
|
)
|
|
|
|
token = await service.authenticate_with_keycloak_token("bad")
|
|
|
|
assert token is None
|