mirror of
https://github.com/pewdiepie-archdaemon/odysseus.git
synced 2026-10-06 06:52:20 +02:00
pewdiepie-archdaemon
2e8413a54a
Snapshot current maintainer-preview application changes and regression fixtures for integration into lab. Excludes local runtime data, evaluation outputs and source backups. Focused Python regression selection: 140 passed; full suite not certified.
219 lines
8.1 KiB
Python
219 lines
8.1 KiB
Python
"""Regression tests for password-change session revocation."""
|
|
|
|
import asyncio
|
|
import importlib
|
|
import sys
|
|
import types
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from unittest.mock import MagicMock
|
|
|
|
import pytest
|
|
from fastapi import HTTPException
|
|
|
|
from tests.helpers.import_state import clear_module
|
|
|
|
|
|
def _real_core_package():
|
|
root = Path(__file__).resolve().parent.parent
|
|
core_path = str(root / "core")
|
|
core = sys.modules.get("core")
|
|
if core is None:
|
|
core = types.ModuleType("core")
|
|
sys.modules["core"] = core
|
|
core.__path__ = [core_path]
|
|
clear_module("core.auth")
|
|
return core
|
|
|
|
|
|
def _auth_module():
|
|
_real_core_package()
|
|
return importlib.import_module("core.auth")
|
|
|
|
|
|
def _make_manager(tmp_path):
|
|
auth_mod = _auth_module()
|
|
auth_mod._hash_password = lambda password: f"hash:{password}"
|
|
auth_mod._verify_password = lambda password, hashed: hashed == f"hash:{password}"
|
|
auth_path = tmp_path / "auth.json"
|
|
mgr = auth_mod.AuthManager(str(auth_path))
|
|
assert mgr.create_user("alice", "old-password", is_admin=False)
|
|
assert mgr.create_user("bob", "bob-password", is_admin=False)
|
|
return mgr
|
|
|
|
|
|
async def _immediate_to_thread(fn, *args, **kwargs):
|
|
return fn(*args, **kwargs)
|
|
|
|
|
|
def test_admin_password_reset_revokes_only_target_sessions(tmp_path):
|
|
mgr = _make_manager(tmp_path)
|
|
mgr.create_user('admin', 'admin-password', is_admin=True)
|
|
alice = mgr.create_session('alice', 'old-password')
|
|
bob = mgr.create_session('bob', 'bob-password')
|
|
assert not mgr.reset_user_password('alice', 'new-password', 'bob')
|
|
assert not mgr.reset_user_password('admin', 'new-password', 'admin')
|
|
assert not mgr.reset_user_password('missing', 'new-password', 'admin')
|
|
assert mgr.validate_token(alice)
|
|
assert mgr.reset_user_password('alice', 'new-password', 'admin')
|
|
assert not mgr.validate_token(alice)
|
|
assert mgr.validate_token(bob)
|
|
assert not mgr.verify_password('alice', 'old-password')
|
|
assert mgr.verify_password('alice', 'new-password')
|
|
|
|
|
|
@pytest.mark.parametrize('admin,password,status', [
|
|
(False, 'valid-password', 403),
|
|
(True, 'x', 400),
|
|
(True, 'a' * 73, 400),
|
|
(True, '\u00e9' * 37, 400),
|
|
(True, 'valid-password', None),
|
|
])
|
|
def test_admin_password_reset_route(admin, password, status):
|
|
_real_core_package()
|
|
sys.modules.pop('routes.auth_routes', None)
|
|
from routes.auth_routes import ResetUserPasswordRequest, setup_auth_routes
|
|
auth = MagicMock()
|
|
auth.get_username_for_token.return_value = 'admin'
|
|
auth.is_admin.return_value = admin
|
|
auth.reset_user_password.return_value = True
|
|
endpoint = next(route.endpoint for route in setup_auth_routes(auth).routes
|
|
if route.path == '/api/auth/users/{username}/password')
|
|
request = SimpleNamespace(cookies={'odysseus_session': 'token'})
|
|
call = endpoint('alice', ResetUserPasswordRequest(new_password=password), request)
|
|
if status:
|
|
with pytest.raises(HTTPException) as exc:
|
|
asyncio.run(call)
|
|
assert exc.value.status_code == status
|
|
auth.reset_user_password.assert_not_called()
|
|
else:
|
|
assert asyncio.run(call) == {'ok': True}
|
|
auth.reset_user_password.assert_called_once_with('alice', password, 'admin')
|
|
|
|
|
|
def test_revoke_user_sessions_preserves_current_and_persists(tmp_path):
|
|
mgr = _make_manager(tmp_path)
|
|
current = mgr.create_session("alice", "old-password")
|
|
other = mgr.create_session("alice", "old-password")
|
|
bob = mgr.create_session("bob", "bob-password")
|
|
|
|
revoked = mgr.revoke_user_sessions("alice", except_token=current)
|
|
|
|
assert revoked == 1
|
|
assert mgr.validate_token(current) is True
|
|
assert mgr.validate_token(other) is False
|
|
assert mgr.validate_token(bob) is True
|
|
|
|
|
|
def test_wrong_current_password_does_not_revoke_sessions(tmp_path):
|
|
mgr = _make_manager(tmp_path)
|
|
current = mgr.create_session("alice", "old-password")
|
|
other = mgr.create_session("alice", "old-password")
|
|
|
|
assert mgr.change_password("alice", "wrong-password", "new-password") is False
|
|
|
|
assert mgr.validate_token(current) is True
|
|
assert mgr.validate_token(other) is True
|
|
|
|
|
|
def test_password_change_allows_new_password_and_blocks_old_password(tmp_path):
|
|
mgr = _make_manager(tmp_path)
|
|
|
|
assert mgr.change_password("alice", "old-password", "new-password") is True
|
|
|
|
assert mgr.create_session("alice", "old-password") is None
|
|
assert mgr.create_session("alice", "new-password") is not None
|
|
|
|
|
|
def test_create_session_trusted_rejects_username_renamed_after_verification(tmp_path):
|
|
mgr = _make_manager(tmp_path)
|
|
assert mgr.create_user("admin", "admin-password", is_admin=True)
|
|
|
|
assert mgr.verify_password("alice", "old-password") is True
|
|
assert mgr.rename_user("alice", "alice2", "admin") is True
|
|
|
|
assert mgr.create_session_trusted("alice") is None
|
|
|
|
|
|
def _change_password_endpoint(auth_manager):
|
|
sys.modules.pop("routes.auth_routes", None)
|
|
_real_core_package()
|
|
from routes.auth_routes import ChangePasswordRequest, setup_auth_routes
|
|
|
|
router = setup_auth_routes(auth_manager)
|
|
for route in router.routes:
|
|
if getattr(route, "path", None) == "/api/auth/change-password":
|
|
return route.endpoint, ChangePasswordRequest
|
|
raise AssertionError("change-password route not found")
|
|
|
|
|
|
def _login_endpoint(auth_manager):
|
|
sys.modules.pop("routes.auth_routes", None)
|
|
_real_core_package()
|
|
from routes.auth_routes import LoginRequest, setup_auth_routes
|
|
|
|
router = setup_auth_routes(auth_manager)
|
|
for route in router.routes:
|
|
if getattr(route, "path", None) == "/api/auth/login":
|
|
return route.endpoint, LoginRequest
|
|
raise AssertionError("login route not found")
|
|
|
|
|
|
def test_login_route_does_not_set_cookie_when_trusted_session_rejects_stale_user(monkeypatch):
|
|
auth = MagicMock()
|
|
auth.verify_password.return_value = True
|
|
auth.totp_enabled.return_value = False
|
|
auth.create_session_trusted.return_value = None
|
|
endpoint, LoginRequest = _login_endpoint(auth)
|
|
monkeypatch.setattr(
|
|
"routes.auth_routes.asyncio.to_thread",
|
|
lambda fn, *args, **kwargs: _immediate_to_thread(fn, *args, **kwargs),
|
|
)
|
|
request = SimpleNamespace(client=SimpleNamespace(host="127.0.0.1"))
|
|
response = MagicMock()
|
|
body = LoginRequest(username="alice", password="old-password")
|
|
|
|
with pytest.raises(HTTPException) as exc:
|
|
asyncio.run(endpoint(body=body, request=request, response=response))
|
|
|
|
assert exc.value.status_code == 401
|
|
response.set_cookie.assert_not_called()
|
|
|
|
|
|
def test_change_password_route_revokes_other_sessions_after_success(monkeypatch):
|
|
auth = MagicMock()
|
|
auth.get_username_for_token.return_value = "alice"
|
|
auth.change_password.return_value = True
|
|
endpoint, ChangePasswordRequest = _change_password_endpoint(auth)
|
|
monkeypatch.setattr(
|
|
"routes.auth_routes.asyncio.to_thread",
|
|
lambda fn, *args, **kwargs: _immediate_to_thread(fn, *args, **kwargs),
|
|
)
|
|
request = SimpleNamespace(cookies={"odysseus_session": "current-token"})
|
|
body = ChangePasswordRequest(current_password="old-password", new_password="new-password")
|
|
|
|
result = asyncio.run(endpoint(body=body, request=request))
|
|
|
|
assert result == {"ok": True}
|
|
auth.change_password.assert_called_once_with("alice", "old-password", "new-password")
|
|
auth.revoke_user_sessions.assert_called_once_with("alice", "current-token")
|
|
|
|
|
|
def test_change_password_route_wrong_password_does_not_revoke(monkeypatch):
|
|
auth = MagicMock()
|
|
auth.get_username_for_token.return_value = "alice"
|
|
auth.change_password.return_value = False
|
|
endpoint, ChangePasswordRequest = _change_password_endpoint(auth)
|
|
monkeypatch.setattr(
|
|
"routes.auth_routes.asyncio.to_thread",
|
|
lambda fn, *args, **kwargs: _immediate_to_thread(fn, *args, **kwargs),
|
|
)
|
|
request = SimpleNamespace(cookies={"odysseus_session": "current-token"})
|
|
body = ChangePasswordRequest(current_password="wrong-password", new_password="new-password")
|
|
|
|
with pytest.raises(HTTPException) as exc:
|
|
asyncio.run(endpoint(body=body, request=request))
|
|
|
|
assert exc.value.status_code == 400
|
|
auth.revoke_user_sessions.assert_not_called()
|