diff --git a/backend/app/api/auth.py b/backend/app/api/auth.py index 17af9ce..0570d32 100644 --- a/backend/app/api/auth.py +++ b/backend/app/api/auth.py @@ -28,7 +28,7 @@ from argon2 import PasswordHasher from argon2.exceptions import VerifyMismatchError from fastapi import APIRouter, HTTPException, Request, WebSocket from pydantic import BaseModel, Field -from sqlalchemy import select +from sqlalchemy import select, update from sqlalchemy.exc import IntegrityError from starlette.websockets import WebSocketDisconnect @@ -56,6 +56,26 @@ _vanished: set[str] = set() AUTH_GENERATION_SYNC_SECONDS = 1.0 AUTH_GENERATION_MAX_AGE_SECONDS = 2.0 _generations_synced_at: float | None = None +# Один lock на ещё неизвестный узлу логин: параллельные HTTP и WS входы +# разделяют один запрос к БД до очередной синхронизации (lct-42). Запись +# удаляется, когда lock больше никто не ждёт: иначе словарь растёт с каждым +# логином, который узел когда-либо проверял. +_lookup_locks: dict[str, "_LoginLookup"] = {} +# Логины, которых разовая проверка не нашла в users. Держатся до следующей +# сверки: повтор старой cookie не должен давать SELECT на каждый запрос. +_missing_logins: set[str] = set() + + +class _LoginLookup: + __slots__ = ("lock", "users") + + def __init__(self) -> None: + self.lock = asyncio.Lock() + self.users = 0 + + +class _AuthStateUnavailable(Exception): + """Разовая проверка неизвестного узлу логина не смогла обратиться к БД.""" async def _close_revoked(ws: WebSocket) -> None: @@ -98,6 +118,7 @@ def prime_generations(values: dict[str, int]) -> None: global _generations_synced_at _generations.clear() _generations.update(values) + _missing_logins.clear() _generations_synced_at = time.monotonic() @@ -108,15 +129,38 @@ async def load_generations() -> None: async def sync_generations() -> None: - """Refresh shared account epochs and close sockets revoked on peer nodes.""" + """Сверить версии полномочий и закрыть отозванные на другом узле сокеты.""" + known_before_query = set(_generations) async with get_sessionmaker()() as db: rows = (await db.execute(select(User.login, User.auth_version))).all() - current = {login: version for login, version in rows} + current = {login: version for login, version in rows} + # Версия узла выше БД, если отзыв не удалось записать (неудачный + # logout) или учётку пересоздали после удаления. Без записи в БД + # узлы расходятся навсегда: соседний узел выдаёт cookie со старой + # версией, а этот её отвергает. Поднимаем БД до версии узла — отзыв + # сохраняется, остальные узлы догоняют за одну сверку. При гонке со + # снимком БД уже выше, и UPDATE ничего не меняет. + ahead = { + login: _generations[login] + for login, version in current.items() + if login in _generations and _generations[login] > version + } + if ahead: + try: + for login, local in ahead.items(): + await db.execute(_raise_auth_version(login, local)) + await db.commit() + except Exception as exc: # noqa: BLE001 — повторим на следующей сверке + log.error("не удалось записать версию полномочий узла (%s)", + type(exc).__name__) + await db.rollback() for login, version in current.items(): previous = _generations.get(login) + # Снимок БД мог устареть за время запроса; меньшая версия не должна + # отменять локальный отзыв, который уже поднял поколение. if previous is None: _generations[login] = version - elif previous != version: + elif version > previous: invalidate_login(login, version) # Account deletion is not exposed by the application. Still close active # sockets if an operator removes one directly from the shared directory DB. @@ -127,13 +171,88 @@ async def sync_generations() -> None: # lookup fall back to 0 and re-accept cookies issued before the revocation. synthetic = {"dev"} if get_settings().dev_auth_bypass else set() _vanished.intersection_update(_generations.keys() - current.keys()) - for login in _generations.keys() - current.keys() - synthetic - _vanished: + # Разовая проверка могла найти учётку после снимка этого запроса. + # Её отсутствие в старом снимке не означает отзыв. + for login in known_before_query - current.keys() - synthetic - _vanished: invalidate_login(login) _vanished.add(login) + # Снимок только что прочитан: учётка, созданная до него, уже в кэше. + _missing_logins.clear() global _generations_synced_at _generations_synced_at = time.monotonic() +def _raise_auth_version(login: str, version: int): + # Только вверх: параллельная запись в БД могла уже поднять версию выше. + return ( + update(User) + .where(User.login == login, User.auth_version < version) + .values(auth_version=version) + ) + + +async def _resolve_unknown_login(login: str) -> int | None: + """Проверить неизвестный узлу логин сразу, не дожидаясь опроса БД. + + Отозванный логин остаётся в `_generations` (см. `_vanished`), поэтому + отсутствие в кэше означает, что этот узел ещё не видел учётку. + Возвращает версию или None, если строки в `users` нет. При недоступной + или зависшей БД вызывает `_AuthStateUnavailable`, чтобы вход был закрыт + с 503/1013. + """ + if login in _missing_logins: + return None + entry = _lookup_locks.get(login) + if entry is None: + entry = _lookup_locks[login] = _LoginLookup() + entry.users += 1 + try: + # Тот же предел, что у сверки: при partition handshake не ждёт + # таймаута TCP, а очередь за lock не растягивает ожидание сверх него. + async with asyncio.timeout(AUTH_GENERATION_MAX_AGE_SECONDS): + async with entry.lock: + cached = _generations.get(login) + if cached is not None: + return cached # параллельный запрос уже получил версию + if login in _missing_logins: + return None + async with get_sessionmaker()() as db: + version = await db.scalar( + select(User.auth_version).where(User.login == login) + ) + # Пока шёл SELECT, локальный отзыв или синхронизация могли + # записать новую версию. Старый ответ не должен вернуть + # отозванную cookie. + cached = _generations.get(login) + if cached is not None: + return cached + if version is None: + _missing_logins.add(login) + return None + _generations[login] = version + log.warning( + "учётка %s найдена разовой проверкой до сверки узла", + login_log_marker(login), + ) + return version + except Exception as exc: # noqa: BLE001 — ошибка или таймаут БД закрывают вход + log.error("разовая проверка полномочий не удалась (%s)", type(exc).__name__) + raise _AuthStateUnavailable from exc + finally: + entry.users -= 1 + if entry.users == 0 and _lookup_locks.get(login) is entry: + del _lookup_locks[login] + + +def login_log_marker(login: str) -> str: + """Метка логина для журнала без самого логина. + + По ней кластерный смоук доказывает, что узел прошёл через разовую + проверку, а не увидел учётку обычной сверкой. + """ + return hashlib.sha256(f"lct-login:{login}".encode()).hexdigest()[:12] + + async def watch_generations() -> None: """Poll PostgreSQL once per node so remote logout/role changes close WS.""" while True: @@ -151,6 +270,21 @@ async def watch_generations() -> None: await asyncio.sleep(AUTH_GENERATION_SYNC_SECONDS) +async def _send_auth_state_unavailable(scope, send) -> None: + if scope["type"] == "websocket": + await send({"type": "websocket.close", "code": 1013}) + else: + await send({ + "type": "http.response.start", + "status": 503, + "headers": [(b"content-type", b"application/json")], + }) + await send({ + "type": "http.response.body", + "body": b'{"detail":"auth_state_unavailable"}', + }) + + class AuthVersionMiddleware: """Check signed-cookie epochs against the fresh, DB-synchronized node cache.""" @@ -186,22 +320,17 @@ class AuthVersionMiddleware: synced_at is None or time.monotonic() - synced_at > AUTH_GENERATION_MAX_AGE_SECONDS ): - if scope["type"] == "websocket": - await send({"type": "websocket.close", "code": 1013}) - else: - await send({ - "type": "http.response.start", - "status": 503, - "headers": [(b"content-type", b"application/json")], - }) - await send({ - "type": "http.response.body", - "body": b'{"detail":"auth_state_unavailable"}', - }) + await _send_auth_state_unavailable(scope, send) return cookie_version = session.get("auth_generation") version = _generations.get(login) + if version is None: + try: + version = await _resolve_unknown_login(login) + except _AuthStateUnavailable: + await _send_auth_state_unavailable(scope, send) + return if version is None or cookie_version != version: if session is not None: session.clear() @@ -536,7 +665,20 @@ async def login(payload: LoginIn, request: Request) -> dict: await audit_required(user.login, user.role, "login.blocked") raise HTTPException(status_code=403, detail="blocked") - if _generations.get(user.login) != user.auth_version: + local_version = _generations.get(user.login) + if local_version is not None and local_version > user.auth_version: + # Отзыв на этом узле не дошёл до БД (неудачный logout). Сброс к версии + # БД вернул бы силу cookie, выданной до выхода; выдача cookie с + # версией узла без записи в БД развела бы узлы. Поднимаем БД. + try: + async with get_sessionmaker()() as db: + await db.execute(_raise_auth_version(user.login, local_version)) + await db.commit() + except Exception as exc: # noqa: BLE001 — cookie без записанной версии не выдаём + log.error("вход: версию полномочий не удалось записать (%s)", + type(exc).__name__) + raise HTTPException(status_code=503, detail="auth_state_unavailable") from exc + elif local_version != user.auth_version: invalidate_login(user.login, user.auth_version) who = Principal( diff --git a/backend/tests/test_auth_hardening.py b/backend/tests/test_auth_hardening.py index eec963a..f116913 100644 --- a/backend/tests/test_auth_hardening.py +++ b/backend/tests/test_auth_hardening.py @@ -135,6 +135,229 @@ def test_generation_sync_preserves_synthetic_dev_account(monkeypatch): asyncio.run(run()) +@pytest.mark.asyncio +async def test_stale_generation_snapshot_does_not_revoke_newly_resolved_login(monkeypatch): + snapshot_read = asyncio.Event() + release_snapshot = asyncio.Event() + + class FakeResult: + def all(self): + return [] + + class FakeDb: + async def execute(self, _query): + snapshot_read.set() + await release_snapshot.wait() + return FakeResult() + + async def scalar(self, _query): + return 0 + + class FakeSession: + async def __aenter__(self): + return FakeDb() + + async def __aexit__(self, *_args): + return None + + auth.prime_generations({}) + monkeypatch.setattr(auth, "get_settings", lambda: SimpleNamespace(dev_auth_bypass=False)) + monkeypatch.setattr(auth, "get_sessionmaker", lambda: lambda: FakeSession()) + sync = asyncio.create_task(auth.sync_generations()) + await snapshot_read.wait() + assert await auth._resolve_unknown_login("just-created") == 0 + release_snapshot.set() + await sync + assert auth._generations["just-created"] == 0 + assert "just-created" not in auth._vanished + + +@pytest.mark.asyncio +async def test_stale_generation_snapshot_does_not_restore_revoked_cookie(monkeypatch): + snapshot_read = asyncio.Event() + release_snapshot = asyncio.Event() + + raised = [] + + class FakeResult: + def all(self): + return [("revoked", 0)] + + class FakeDb: + async def execute(self, query): + if query.is_update: + raised.append(query.compile().params) + return None + snapshot_read.set() + await release_snapshot.wait() + return FakeResult() + + async def commit(self): + return None + + class FakeSession: + async def __aenter__(self): + return FakeDb() + + async def __aexit__(self, *_args): + return None + + auth.prime_generations({"revoked": 0}) + monkeypatch.setattr(auth, "get_settings", lambda: SimpleNamespace(dev_auth_bypass=False)) + monkeypatch.setattr(auth, "get_sessionmaker", lambda: lambda: FakeSession()) + sync = asyncio.create_task(auth.sync_generations()) + await snapshot_read.wait() + auth.invalidate_login("revoked", 1) + release_snapshot.set() + await sync + assert auth._generations["revoked"] == 1 + # В БД версия уже 1 (отзыв записан до invalidate); UPDATE «только вверх» + # в реальной БД ничего не изменит, в снимке же он выглядит как отставание. + assert [params["auth_version"] for params in raised] == [1] + + +def _versioned_users_db(versions: dict[str, int]): + """Поддельная users: сверка читает версии, UPDATE поднимает их только вверх.""" + updates = [] + + class FakeResult: + def all(self): + return list(versions.items()) + + class FakeDb: + async def execute(self, query): + if query.is_update: + params = query.compile().params + login = next(v for k, v in params.items() if k.startswith("login")) + target = params["auth_version"] + updates.append((login, target)) + if versions.get(login, target) < target: + versions[login] = target + return None + return FakeResult() + + async def scalar(self, _query): + raise AssertionError("login is cached, no one-shot lookup expected") + + async def commit(self): + return None + + async def rollback(self): + return None + + class FakeSession: + async def __aenter__(self): + return FakeDb() + + async def __aexit__(self, *_args): + return None + + return updates, lambda: lambda: FakeSession() + + +@pytest.mark.asyncio +async def test_failed_logout_revocation_is_written_to_db_so_nodes_converge(monkeypatch): + """Неудачный logout поднял версию только на этом узле (B). Сверка не + откатывает отзыв, а записывает его в БД: соседний узел A догоняет, и вход + на A после этого выдаёт cookie, которую B принимает.""" + login = "failed-logout-user" + versions = {login: 4} + updates, sessionmaker = _versioned_users_db(versions) + monkeypatch.setattr(auth, "get_settings", lambda: SimpleNamespace(dev_auth_bypass=False)) + monkeypatch.setattr(auth, "get_sessionmaker", sessionmaker) + auth.prime_generations({login: 4}) + try: + auth.invalidate_login(login) # ветка except в logout: БД недоступна + assert auth._generations[login] == 5 + + await auth.sync_generations() + assert updates == [(login, 5)] + assert versions[login] == 5, "revocation must reach the shared DB" + assert auth._generations[login] == 5, "sync must not roll back a local revocation" + + # Вход на A берёт версию из БД, и теперь она совпадает с версией B. + assert versions[login] == auth._generations[login] + + await auth.sync_generations() + assert updates == [(login, 5)], "converged nodes must not write again" + finally: + auth._generations.pop(login, None) + + +@pytest.mark.asyncio +async def test_recreated_account_is_raised_to_the_node_revocation_version(monkeypatch): + """Учётку удалили (узел отозвал её, поколение +1) и создали заново с 0. + Без записи в БД узел остался бы впереди навсегда.""" + login = "recreated-user" + versions = {} + updates, sessionmaker = _versioned_users_db(versions) + monkeypatch.setattr(auth, "get_settings", lambda: SimpleNamespace(dev_auth_bypass=False)) + monkeypatch.setattr(auth, "get_sessionmaker", sessionmaker) + auth.prime_generations({login: 2}) + try: + await auth.sync_generations() + assert auth._generations[login] == 3 + assert login in auth._vanished + + versions[login] = 0 # оператор создал учётку заново + await auth.sync_generations() + assert updates == [(login, 3)] + assert versions[login] == 3 + assert login not in auth._vanished + finally: + auth._generations.pop(login, None) + auth._vanished.discard(login) + + +def test_login_raises_db_version_when_node_revocation_was_not_persisted(client, monkeypatch): + """Вход на узле, где отзыв не дошёл до БД: cookie получает версию узла, + а БД поднимается до неё. Сброс к версии БД вернул бы силу старой cookie.""" + from app.config import get_settings + + user = SimpleNamespace( + login="ahead-login", auth_provider="local", password_hash="hash", + blocked=False, role="instructor", full_name="Преподаватель", + service=None, trainee_id=None, auth_version=1, + ) + updates = [] + + class FakeDb: + async def scalar(self, _statement): + return user + + async def execute(self, query): + assert query.is_update + updates.append(query.compile().params["auth_version"]) + + async def commit(self): + return None + + class FakeSession: + async def __aenter__(self): + return FakeDb() + + async def __aexit__(self, *_args): + return None + + async def audit_ok(*_args, **_kwargs): + return True + + settings = get_settings().model_copy(update={"demo_no_db": False, "ldap_enabled": False}) + monkeypatch.setattr(auth, "get_settings", lambda: settings) + monkeypatch.setattr(auth, "get_sessionmaker", lambda: lambda: FakeSession()) + monkeypatch.setattr(auth, "verify_password", lambda *_args: True) + monkeypatch.setattr(auth, "audit", audit_ok) + auth._generations[user.login] = 2 # неудачный logout на этом узле + try: + response = client.post("/api/auth/login", json={"login": user.login, "password": "x"}) + assert response.status_code == 200 + assert updates == [2] + assert auth._generations[user.login] == 2, "local revocation must not be rolled back" + assert client.get("/api/auth/me").status_code == 200 + finally: + auth._generations.pop(user.login, None) + + def test_cross_origin_browser_websocket_is_rejected_before_handshake(client): assert client.post("/api/auth/dev-token").status_code == 200 with pytest.raises(WebSocketDisconnect) as exc: @@ -410,6 +633,401 @@ def test_auth_middleware_fails_closed_when_generation_cache_is_stale_but_allows_ asyncio.run(run()) +def test_middleware_resolves_and_caches_login_unknown_to_this_node(monkeypatch): + """Cluster handshake (lct-42): a login another node just authenticated is + fetched via a single SELECT rather than being treated as revoked, and the + result is cached so a second request for it does not query again.""" + login = "peer-node-fresh-login" + calls = [] + + class FakeDb: + async def scalar(self, _query): + calls.append(1) + return 5 + + class FakeSession: + async def __aenter__(self): + return FakeDb() + + async def __aexit__(self, *_args): + return None + + class InnerApp: + def __init__(self): + self.called = 0 + + async def __call__(self, _scope, _receive, _send): + self.called += 1 + + async def run(): + auth.prime_generations({}) + + def make_scope(): + return { + "type": "http", "path": "/api/admin/users", + "session": { + "principal": {"login": login}, + "auth_instance": auth._INSTANCE, + "auth_generation": 5, + }, + } + + async def receive(): + return {"type": "http.request", "body": b"", "more_body": False} + + async def send(_message): + return None + + first = InnerApp() + await auth.AuthVersionMiddleware(first)(make_scope(), receive, send) + assert first.called == 1, "a fresh, valid epoch must reach the route" + assert auth._generations[login] == 5 + assert calls == [1] + + second = InnerApp() + await auth.AuthVersionMiddleware(second)(make_scope(), receive, send) + assert second.called == 1 + assert calls == [1], "cached epoch must not trigger a second SELECT" + + settings = auth.get_settings().model_copy(update={"demo_no_db": False}) + monkeypatch.setattr(auth, "get_settings", lambda: settings) + monkeypatch.setattr(auth, "get_sessionmaker", lambda: lambda: FakeSession()) + try: + asyncio.run(run()) + finally: + auth._generations.pop(login, None) + auth._lookup_locks.pop(login, None) + + +def test_middleware_rejects_login_missing_from_users_via_one_shot_query(monkeypatch): + login = "peer-node-deleted-login" + + class FakeDb: + async def scalar(self, _query): + return None + + class FakeSession: + async def __aenter__(self): + return FakeDb() + + async def __aexit__(self, *_args): + return None + + class InnerApp: + def __init__(self): + self.called = False + + async def __call__(self, _scope, _receive, _send): + self.called = True + + async def run(): + auth.prime_generations({}) + cookie_session = { + "principal": {"login": login}, + "auth_instance": auth._INSTANCE, + "auth_generation": 3, + } + scope = {"type": "http", "path": "/api/admin/users", "session": cookie_session} + + async def receive(): + return {"type": "http.request", "body": b"", "more_body": False} + + async def send(_message): + return None + + protected = InnerApp() + await auth.AuthVersionMiddleware(protected)(scope, receive, send) + assert protected.called, "route still runs but the session was cleared below" + assert cookie_session == {} + assert login not in auth._generations + + settings = auth.get_settings().model_copy(update={"demo_no_db": False}) + monkeypatch.setattr(auth, "get_settings", lambda: settings) + monkeypatch.setattr(auth, "get_sessionmaker", lambda: lambda: FakeSession()) + try: + asyncio.run(run()) + finally: + auth._lookup_locks.pop(login, None) + + +def test_unknown_login_lookup_does_not_clobber_a_newer_local_revocation(monkeypatch): + """A stale SELECT reply landing after a concurrent local revoke must not + resurrect the revoked cookie's epoch (regression: lock only deduplicates + concurrent lookups, it does not order a lookup against a write).""" + login = "peer-node-race-login" + started = asyncio.Event() + resume = asyncio.Event() + + class FakeDb: + async def scalar(self, _query): + started.set() + await resume.wait() + return 0 # the epoch as it stood before the concurrent revoke below + + class FakeSession: + async def __aenter__(self): + return FakeDb() + + async def __aexit__(self, *_args): + return None + + async def run(): + auth.prime_generations({}) + task = asyncio.ensure_future(auth._resolve_unknown_login(login)) + await started.wait() # the SELECT is in flight, holding the login's lock + auth.invalidate_login(login, 1) # a local logout/edit races the reply + resume.set() + result = await task + assert result == 1, "the newer local revocation must win over the stale SELECT reply" + assert auth._generations[login] == 1 + + monkeypatch.setattr(auth, "get_sessionmaker", lambda: lambda: FakeSession()) + try: + asyncio.run(run()) + finally: + auth._generations.pop(login, None) + auth._lookup_locks.pop(login, None) + + +def test_unknown_login_lookup_deduplicates_concurrent_requests_into_one_select(monkeypatch): + login = "peer-node-concurrent-login" + calls = [] + + class FakeDb: + async def scalar(self, _query): + calls.append(1) + await asyncio.sleep(0.01) # widen the window for a racing second caller + return 9 + + class FakeSession: + async def __aenter__(self): + return FakeDb() + + async def __aexit__(self, *_args): + return None + + async def run(): + auth.prime_generations({}) + results = await asyncio.gather( + auth._resolve_unknown_login(login), + auth._resolve_unknown_login(login), + auth._resolve_unknown_login(login), + ) + assert results == [9, 9, 9] + assert calls == [1], "concurrent lookups for one login must issue a single SELECT" + assert login not in auth._lookup_locks, "lock entry must not outlive its waiters" + + monkeypatch.setattr(auth, "get_sessionmaker", lambda: lambda: FakeSession()) + try: + asyncio.run(run()) + finally: + auth._generations.pop(login, None) + auth._lookup_locks.pop(login, None) + + +def test_unknown_login_lookup_fails_closed_when_database_is_unreachable(monkeypatch): + login = "peer-node-db-outage-login" + + class BrokenSession: + async def __aenter__(self): + raise OSError("database unavailable") + + async def __aexit__(self, *_args): + return None + + class InnerApp: + def __init__(self): + self.called = False + + async def __call__(self, _scope, _receive, _send): + self.called = True + + async def run(): + auth.prime_generations({}) + cookie_session = { + "principal": {"login": login}, + "auth_instance": auth._INSTANCE, + "auth_generation": 0, + } + + async def receive(): + return {"type": "http.request", "body": b"", "more_body": False} + + http_messages = [] + + async def http_send(message): + http_messages.append(message) + + http_app = InnerApp() + await auth.AuthVersionMiddleware(http_app)( + {"type": "http", "path": "/api/admin/users", "session": dict(cookie_session)}, + receive, http_send, + ) + assert not http_app.called + assert http_messages[0]["status"] == 503 + + ws_messages = [] + + async def ws_send(message): + ws_messages.append(message) + + ws_app = InnerApp() + await auth.AuthVersionMiddleware(ws_app)( + {"type": "websocket", "path": "/ws/control/x", "session": dict(cookie_session)}, + receive, ws_send, + ) + assert not ws_app.called + assert ws_messages[0] == {"type": "websocket.close", "code": 1013} + + settings = auth.get_settings().model_copy(update={"demo_no_db": False}) + monkeypatch.setattr(auth, "get_settings", lambda: settings) + monkeypatch.setattr(auth, "get_sessionmaker", lambda: lambda: BrokenSession()) + try: + asyncio.run(run()) + finally: + auth._lookup_locks.pop(login, None) + + +def test_unknown_login_lookup_times_out_instead_of_hanging_on_partition(monkeypatch): + """При partition SELECT может висеть до таймаута TCP. Разовая проверка + ограничена тем же порогом, что сверка, и закрывает вход 503 / 1013 — + и для запроса, который ждёт lock за зависшим.""" + login = "peer-node-hanging-db-login" + monkeypatch.setattr(auth, "AUTH_GENERATION_MAX_AGE_SECONDS", 0.3) + + class HangingDb: + async def scalar(self, _query): + await asyncio.Event().wait() + + class FakeSession: + async def __aenter__(self): + return HangingDb() + + async def __aexit__(self, *_args): + return None + + class InnerApp: + called = False + + async def __call__(self, _scope, _receive, _send): + self.called = True + + async def run(): + auth.prime_generations({}) + cookie_session = { + "principal": {"login": login}, + "auth_instance": auth._INSTANCE, + "auth_generation": 0, + } + + async def receive(): + return {"type": "http.request", "body": b"", "more_body": False} + + http_messages, ws_messages = [], [] + + async def http_send(message): + http_messages.append(message) + + async def ws_send(message): + ws_messages.append(message) + + http_app, ws_app = InnerApp(), InnerApp() + started = asyncio.get_running_loop().time() + await asyncio.wait_for(asyncio.gather( + auth.AuthVersionMiddleware(http_app)( + {"type": "http", "path": "/api/admin/users", "session": dict(cookie_session)}, + receive, http_send, + ), + auth.AuthVersionMiddleware(ws_app)( + {"type": "websocket", "path": "/ws/control/x", "session": dict(cookie_session)}, + receive, ws_send, + ), + ), timeout=2) + elapsed = asyncio.get_running_loop().time() - started + assert elapsed < 1, "a queued handshake must not wait for a second timeout" + assert not http_app.called and not ws_app.called + assert http_messages[0]["status"] == 503 + assert ws_messages[0] == {"type": "websocket.close", "code": 1013} + assert login not in auth._lookup_locks, "timed-out lookup must release its lock entry" + + settings = auth.get_settings().model_copy(update={"demo_no_db": False}) + monkeypatch.setattr(auth, "get_settings", lambda: settings) + monkeypatch.setattr(auth, "get_sessionmaker", lambda: lambda: FakeSession()) + asyncio.run(run()) + + +def test_one_shot_lookup_logs_login_marker_not_login(caplog, monkeypatch): + """Кластерный смоук ищет эту метку в журнале узла B: она доказывает, что + вход прошёл через разовую проверку, а не через обычную сверку.""" + login = "peer-node-logged-login" + + class FakeDb: + async def scalar(self, _query): + return 0 + + class FakeSession: + async def __aenter__(self): + return FakeDb() + + async def __aexit__(self, *_args): + return None + + async def run(): + auth.prime_generations({}) + assert await auth._resolve_unknown_login(login) == 0 + + monkeypatch.setattr(auth, "get_sessionmaker", lambda: lambda: FakeSession()) + try: + with caplog.at_level("WARNING", logger="app.api.auth"): + asyncio.run(run()) + finally: + auth._generations.pop(login, None) + assert auth.login_log_marker(login) in caplog.text + assert login not in caplog.text + + +def test_missing_login_is_cached_until_next_sync_and_lock_entries_are_released(monkeypatch): + login = "peer-node-replayed-missing-login" + lookups = [] + + class FakeResult: + def all(self): + return [] + + class FakeDb: + async def scalar(self, _query): + lookups.append(1) + return None + + async def execute(self, _query): + return FakeResult() + + class FakeSession: + async def __aenter__(self): + return FakeDb() + + async def __aexit__(self, *_args): + return None + + async def run(): + auth.prime_generations({}) + assert await auth._resolve_unknown_login(login) is None + assert await auth._resolve_unknown_login(login) is None + assert lookups == [1], "a replayed cookie must not query users on every request" + assert auth._lookup_locks == {} + + await auth.sync_generations() + assert await auth._resolve_unknown_login(login) is None + assert lookups == [1, 1], "the negative result lives only until the next sync" + + monkeypatch.setattr(auth, "get_settings", lambda: SimpleNamespace(dev_auth_bypass=False)) + monkeypatch.setattr(auth, "get_sessionmaker", lambda: lambda: FakeSession()) + try: + asyncio.run(run()) + finally: + auth._missing_logins.discard(login) + + def test_auth_middleware_uses_fresh_generation_cache_without_per_request_database_query(monkeypatch): from app.config import get_settings diff --git a/scripts/test_cluster_failover.py b/scripts/test_cluster_failover.py index c4bcb5f..342d8ac 100644 --- a/scripts/test_cluster_failover.py +++ b/scripts/test_cluster_failover.py @@ -72,7 +72,7 @@ def login(base: str, login_name: str, password: str) -> str: return cookie -async def send_control(ws_base: str, session_id: UUID, cookie: str, payload: dict) -> None: +async def send_control(ws_base: str, session_id: UUID, cookie: str, payload: dict) -> float: async with websockets.connect( f"{ws_base}/ws/control/{session_id}", additional_headers={"Cookie": cookie}, @@ -81,7 +81,9 @@ async def send_control(ws_base: str, session_id: UUID, cookie: str, payload: dic ping_interval=None, ) as socket: await socket.send(json.dumps(payload, ensure_ascii=False)) + sent_at = time.monotonic() await asyncio.sleep(0.15) + return sent_at async def probe_ws(ws_base: str, path: str, cookie: str) -> dict: @@ -238,7 +240,7 @@ async def main(args: argparse.Namespace) -> int: from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine from sqlalchemy.pool import NullPool - from app.api.auth import hash_password + from app.api.auth import hash_password, login_log_marker from app.db.models import AuditLog, Session, Trainee, User database_url = ( @@ -251,7 +253,7 @@ async def main(args: argparse.Namespace) -> int: factory = async_sessionmaker(engine, expire_on_commit=False) observer_engine = create_async_engine(database_url, poolclass=NullPool) run_id = uuid4().hex[:12] - session_id = uuid4() + session_id = args.session_id or uuid4() trainee = Trainee(name=f"Failover smoke {run_id}") instructor_login = f"fo-i-{run_id}" trainee_login = f"fo-t-{run_id}" @@ -263,6 +265,7 @@ async def main(args: argparse.Namespace) -> int: fault_stopped = False result: dict = {"session_id": str(session_id), "checks": {}} backend = f"http://127.0.0.1:{args.backend_port}" + result["login_node"] = "backend-a" ws_base = args.frontend_url.replace("https://", "wss://").replace("http://", "ws://") instructor_cookie = trainee_cookie = None try: @@ -278,21 +281,32 @@ async def main(args: argparse.Namespace) -> int: trainee_id=trainee.id, blocked=False), ]) await db.commit() + users_committed_at = time.monotonic() instructor_cookie = login(backend, instructor_login, instructor_password) trainee_cookie = login(backend, trainee_login, trainee_password) # Учётки созданы прямо в БД, вход — на узле A, а сокет занятия Nginx - # может отдать узлу B. B узнаёт новый логин при очередной сверке - # поколений (раз в секунду); до неё middleware считает cookie отозванной. - for port in (args.backend_port, args.backend_b_port): - for cookie in (instructor_cookie, trainee_cookie): - if not await wait_auth_fresh(f"http://127.0.0.1:{port}", cookie): - raise RuntimeError(f"backend :{port} не узнал новый логин за 10 с") - await send_control(ws_base, session_id, instructor_cookie, { + # может отдать узлу B до его сверки поколений. Ждать сверку не нужно: + # B проверяет неизвестный логин разовым SELECT (lct-42). Ожидание здесь + # лишило бы доказательности проверку ниже: опрос B сам вызвал бы разовую + # проверку, а за секунду B узнал бы логин сверкой. + control_sent_at = await send_control(ws_base, session_id, instructor_cookie, { "type": "scenario.start", "scenario_id": args.scenario, "trainee": trainee.name, "trainee_id": str(trainee.id), "mode": "training", "exercise": "dds", }) + result["users_commit_to_control_seconds"] = round(control_sent_at - users_committed_at, 4) + if args.expect_initial_owner == "backend-b": + result["checks"]["control_sent_within_first_second"] = ( + result["users_commit_to_control_seconds"] < 1 + ) + # Сверка B идёт раз в секунду в случайной фазе, и время само по себе + # не доказывает, что B не знал логин. Доказательство — запись B о + # разовой проверке именно этого логина. + result["checks"]["backend_b_resolved_login_one_shot"] = ( + login_log_marker(instructor_login) + in compose(args, "logs", "--no-color", "backend-b") + ) async def session_row(): async with factory() as db: @@ -327,6 +341,12 @@ async def main(args: argparse.Namespace) -> int: old_epoch = row.backend_fencing_epoch result["initial_owner"] = row.backend_node_id result["initial_epoch"] = old_epoch + if args.expect_initial_owner and row.backend_node_id != args.expect_initial_owner: + raise RuntimeError( + f"session routed to {row.backend_node_id}, expected {args.expect_initial_owner}" + ) + if args.expect_initial_owner == "backend-b": + result["checks"]["login_a_initial_control_b"] = True # Commit real DDS work before the fault. These actions must survive the # checkpoint handoff and remain part of the final scored report. @@ -735,4 +755,6 @@ if __name__ == "__main__": parser.add_argument("--reconnect-attempts", type=int, default=5) parser.add_argument("--rest-timeout", type=float, default=60) parser.add_argument("--failure-mode", choices=("kill", "partition"), default="kill") + parser.add_argument("--expect-initial-owner", choices=("backend-a", "backend-b")) + parser.add_argument("--session-id", type=UUID) raise SystemExit(asyncio.run(main(parser.parse_args())))