Codex Release Bot
61 lines
2.1 KiB
Python
61 lines
2.1 KiB
Python
"""Refresh token reuse detection (rotated token replay)."""
|
|
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
|
|
import pytest
|
|
|
|
from server.application.services.auth_service import AuthService
|
|
from server.domain.models import Admin
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_refresh_reuse_invalidates_all_sessions(monkeypatch):
|
|
admin = Admin(id=1, username="admin", password_hash="x", is_active=True, token_version=2)
|
|
service = AuthService(
|
|
admin_repo=AsyncMock(),
|
|
attempt_repo=AsyncMock(),
|
|
audit_repo=AsyncMock(),
|
|
)
|
|
service.admin_repo.get_by_id = AsyncMock(return_value=admin)
|
|
service.admin_repo.update = AsyncMock(return_value=admin)
|
|
service._audit = AsyncMock()
|
|
service._delete_all_refresh_tokens = AsyncMock()
|
|
service._remove_refresh_token = AsyncMock()
|
|
service._store_refresh_token = AsyncMock()
|
|
service._create_access_token = MagicMock(return_value="access")
|
|
service._create_refresh_token = MagicMock(return_value="new:1:2")
|
|
|
|
token = "oldsecret:1:2"
|
|
|
|
async def membership(_admin_id, _token):
|
|
return False
|
|
|
|
monkeypatch.setattr(service, "_refresh_token_membership", membership)
|
|
|
|
result = await service.refresh_token(token, ip_address="127.0.0.1")
|
|
|
|
assert result["success"] is False
|
|
assert admin.token_version == 3
|
|
service.admin_repo.update.assert_awaited_once()
|
|
service._delete_all_refresh_tokens.assert_awaited_once_with(1)
|
|
service._audit.assert_awaited()
|
|
assert service._audit.await_args[0][0] == "refresh_token_reuse"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_refresh_redis_down_returns_unavailable(monkeypatch):
|
|
admin = Admin(id=1, username="admin", password_hash="x", is_active=True, token_version=1)
|
|
service = AuthService(AsyncMock(), AsyncMock(), AsyncMock())
|
|
service.admin_repo.get_by_id = AsyncMock(return_value=admin)
|
|
service._audit = AsyncMock()
|
|
|
|
async def membership(_admin_id, _token):
|
|
return None
|
|
|
|
monkeypatch.setattr(service, "_refresh_token_membership", membership)
|
|
|
|
result = await service.refresh_token("t:1:1", ip_address="")
|
|
|
|
assert result["success"] is False
|
|
assert result["reason"] == "service_unavailable"
|