fix: неизвестный узлу логин проверяется разово, не отзывается сразу
lct-42: узел кластера принимал вход другого узла первую секунду за отзыв и рвал сокет FORBIDDEN. Неизвестный логин теперь проверяется разовым SELECT auth_version, результат кэшируется, параллельные запросы одного логина дедуплицируются; логина нет в users или БД недоступна — как раньше.
This commit is contained in:
kaifarikman
parent
aa6860fc6c
commit
b6617d7df8
2 changed files with 275 additions and 12 deletions
|
|
@ -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()
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
||||
|
|
|
|||
Loading…
Reference in a new issue