diff --git a/backend/app/api/auth.py b/backend/app/api/auth.py index 17af9ce..bb25c65 100644 --- a/backend/app/api/auth.py +++ b/backend/app/api/auth.py @@ -56,6 +56,14 @@ _vanished: set[str] = set() AUTH_GENERATION_SYNC_SECONDS = 1.0 AUTH_GENERATION_MAX_AGE_SECONDS = 2.0 _generations_synced_at: float | None = None +# One lock per login this node has not synced yet, so a concurrent HTTP +# request and WS handshake for the same just-migrated login collapse into a +# single SELECT instead of one each (lct-42). +_lookup_locks: dict[str, asyncio.Lock] = {} + + +class _AuthStateUnavailable(Exception): + """The one-shot lookup for a login unknown to this node could not reach PostgreSQL.""" async def _close_revoked(ws: WebSocket) -> None: @@ -134,6 +142,36 @@ async def sync_generations() -> None: _generations_synced_at = time.monotonic() +async def _resolve_unknown_login(login: str) -> int | None: + """Look up a login absent from this node's cache without waiting a tick. + + A revoked login stays cached in `_generations` (see `_vanished` above), + so absence from `_generations` unambiguously means "this node has not + synced this login yet" — an unrecognized login is checked directly + rather than treated as revoked. Returns the current epoch, or `None` if + the login is not (or no longer) in `users`. Raises + `_AuthStateUnavailable` if PostgreSQL cannot be reached; the caller + fails closed. + """ + lock = _lookup_locks.setdefault(login, asyncio.Lock()) + async with lock: + cached = _generations.get(login) + if cached is not None: + return cached # a concurrent request already resolved it + try: + async with get_sessionmaker()() as db: + version = await db.scalar( + select(User.auth_version).where(User.login == login) + ) + except Exception as exc: # noqa: BLE001 — fail closed, not FORBIDDEN + log.error("разовая проверка полномочий не удалась (%s)", type(exc).__name__) + raise _AuthStateUnavailable from exc + if version is None: + return None + _generations[login] = version + return version + + async def watch_generations() -> None: """Poll PostgreSQL once per node so remote logout/role changes close WS.""" while True: @@ -151,6 +189,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 +239,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() diff --git a/backend/tests/test_auth_hardening.py b/backend/tests/test_auth_hardening.py index eec963a..00cd8bf 100644 --- a/backend/tests/test_auth_hardening.py +++ b/backend/tests/test_auth_hardening.py @@ -410,6 +410,221 @@ 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_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" + + 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_auth_middleware_uses_fresh_generation_cache_without_per_request_database_query(monkeypatch): from app.config import get_settings