diff --git a/app/admin_api.py b/app/admin_api.py index 65984a9..181a047 100644 --- a/app/admin_api.py +++ b/app/admin_api.py @@ -69,6 +69,9 @@ def install_admin(app, config, gateway): return app.state.admin_auth control, audit = config["control_store"], config["audit_store"] auth = AdminAuth(config) + # Resolve the key epoch now: a process that only serves inference traffic would + # otherwise never clear a snapshot belonging to a superseded key. + auth.reconcile() mutation_lock = threading.RLock() oauth_lock = threading.RLock() oauth_tasks = OrderedDict() @@ -102,7 +105,8 @@ def settings_result(): items.append({"key": "auto_accept_buddy", "value": config.get("auto_accept_buddy") is True, "stored": None, "source": config.get("auto_accept_buddy_source", "default"), "mode": "startup", "type": "boolean", "label": "全部国内账号首次领猫预授权", "locked": True}) - return {"revision": control.snapshot()["revision"], "items": items, "audit": audit.storage()} + return {"revision": control.snapshot()["revision"], "items": items, "audit": audit.storage(), + "session": auth.storage()} def audit_settings(values): mapping = {"audit_max_bytes": "max_bytes", "audit_retention_days": "retention_days", @@ -220,7 +224,10 @@ async def session_get(request): @route("DELETE", "/admin/session") async def session_delete(request): - auth.logout(request) + if not auth.logout(request): + # The session is still valid on disk; keep the cookie so the client can retry. + event("session.revoke_failed", {"code": "session_storage_unavailable"}) + return error_response(503, "会话未能持久撤销，请检查管理目录权限后重试") response = JSONResponse({"authenticated": False}) response.delete_cookie(COOKIE_NAME, path="/admin", httponly=True, samesite="strict", secure=request.url.scheme == "https") return response diff --git a/app/admin_auth.py b/app/admin_auth.py index 701df8c..79d3945 100644 --- a/app/admin_auth.py +++ b/app/admin_auth.py @@ -2,8 +2,16 @@ from __future__ import annotations from collections import OrderedDict +import hashlib import hmac +import json +import math +import os +from pathlib import Path +import re import secrets +import stat +import tempfile import threading import time from urllib.parse import urlsplit @@ -15,11 +23,62 @@ SESSION_TTL = 12 * 3600 +class SessionStoreError(RuntimeError): + """A superseded session snapshot could not be durably revoked. + + Startup must abort on this: the surviving snapshot could otherwise be adopted by a + later start under the superseded key, resurrecting sessions that were meant to die. + """ + + +# Optional persisted session table so a restart does not force another login. +# Revocation is enforced by clearing this file: the stored fingerprint only tells us +# which key epoch the snapshot belongs to, it is not an integrity MAC over the sessions. +SESSION_FILE_VERSION = 1 +MAX_PERSISTED_SESSIONS = 256 +_SESSION_FILE_BYTES = 256 * 1024 +_SESSION_KEY_LABEL = b"codebuddy2api-admin-session-key-v1" +# SIDs and CSRF tokens are token_urlsafe(32); bound them so a crafted file cannot +# install arbitrarily large values into memory. +_MAX_TOKEN_CHARS = 128 +_TOKEN = re.compile(r"\A[A-Za-z0-9_-]{16,%d}\Z" % _MAX_TOKEN_CHARS) +_FINGERPRINT = re.compile(r"\A[0-9a-f]{64}\Z") + + def error_response(status, message): return JSONResponse({"error": {"message": message, "type": "conflict_error" if status == 409 else "admin_error"}}, status_code=status, headers={"Cache-Control": "no-store", "Pragma": "no-cache"}) +def _session_path(value): + """Resolve the optional session file; a missing or unusable path keeps sessions in memory.""" + if not value: + return None + try: + path = Path(os.path.abspath(os.fspath(value))) + except (TypeError, ValueError): + return None + if not path.name or path.name in (".", ".."): + return None + return path + + +def _strict_json(content): + """Parse JSON rejecting duplicate keys and non-finite constants.""" + def pairs(items): + value = {} + for key, item in items: + if key in value: + raise ValueError("Duplicate JSON field") + value[key] = item + return value + + def invalid_constant(_): + raise ValueError("Invalid JSON constant") + + return json.loads(content, object_pairs_hook=pairs, parse_constant=invalid_constant) + + def origin_allowlist(value): """Parse a normalized origin list into comparable (scheme, host, port) triples.""" triples = set() @@ -64,24 +123,217 @@ def same_origin(request, allowed=()): class AdminAuth: - def __init__(self, config, *, clock=time.monotonic): + def __init__(self, config, *, clock=time.monotonic, wall_clock=time.time): self.config = config - self.clock = clock + self.clock = clock # Monotonic: login throttling only. + self.wall_clock = wall_clock # Wall clock: session expiry, so it survives a restart. self.lock = threading.RLock() self.sessions = OrderedDict() self.failures = OrderedDict() self._configured_key = None self._identity = None + self._path = _session_path(config.get("session_path")) + self._storage_error = None # Set when the snapshot could not be written or cleared. + + @staticmethod + def _fingerprint(key): + """Keyed fingerprint naming the key epoch a snapshot belongs to. + + It identifies the epoch; revocation itself is performed by clearing the file. + """ + return hmac.new(key.encode(), _SESSION_KEY_LABEL, hashlib.sha256).hexdigest() + + @staticmethod + def _token(value): + """Accept only bounded url-safe tokens, rejecting bools and non-strings.""" + return value if isinstance(value, str) and _TOKEN.match(value) else None + + @staticmethod + def _deadline(value): + """Accept only finite, in-range numeric deadlines; bool is not a deadline.""" + if type(value) not in (int, float): + return None + try: + number = float(value) + except OverflowError: + # An integer too large for float() is not a deadline. + return None + return number if math.isfinite(number) and 0 < number < 1e11 else None + + def _storage_failed(self, error): + """Record why the snapshot could not be made durable, so callers can report it.""" + self._storage_error = type(error).__name__ + + def _storage_ok(self): + self._storage_error = None + + def storage(self): + """Whether the snapshot's durable state is known-good, mirroring AuditStore.storage(). + + Reports the outcome of the last write or clear: a successful clear leaves no + snapshot, so the on-disk state is consistent again and the flag goes back to False. + """ + return {"path": str(self._path) if self._path is not None else None, + "degraded": self._storage_error is not None, + "last_error": self._storage_error} + + def _revoke(self): + """Revoke the persisted snapshot; returns False when it could not be cleared.""" + if self._path is None: + return True + try: + os.unlink(self._path) + except FileNotFoundError: + return True + except OSError as error: + self._storage_failed(error) + return False + self._storage_ok() + return True + + def _is_direct_file(self, metadata): + """True when the opened file is the path itself rather than a symlink target.""" + try: + link = os.lstat(self._path) + except OSError: + return False + if stat.S_ISLNK(link.st_mode): + return False + # Identity is the fallback where lstat cannot report a link; both values are + # zero on filesystems that do not expose inodes, which leaves the check above. + return (link.st_dev, link.st_ino) == (metadata.st_dev, metadata.st_ino) + + def _restore(self, key): + """Load persisted sessions for the current key epoch. + + Returns False when a snapshot exists that cannot be trusted or validated, + so the caller revokes it instead of leaving it available to a later start. + """ + if self._path is None: + return True + try: + fd = os.open(self._path, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_NONBLOCK", 0)) + except FileNotFoundError: + return True + except OSError: + # Unreadable or not a plain readable file: treat as an unusable snapshot. + return False + try: + with os.fdopen(fd, "rb") as stream: + metadata = os.fstat(stream.fileno()) + if not stat.S_ISREG(metadata.st_mode) or metadata.st_size > _SESSION_FILE_BYTES: + return False + # O_NOFOLLOW is absent on Windows, where os.open follows a symlink. Compare + # the opened file with the path itself, so a link is rejected, not followed. + if not self._is_direct_file(metadata): + return False + raw = stream.read(_SESSION_FILE_BYTES + 1) + except OSError: + return False + if len(raw) > _SESSION_FILE_BYTES: + return False + try: + document = _strict_json(raw) + except (ValueError, UnicodeDecodeError, RecursionError): + # RecursionError: size-bounded but deeply nested JSON still exhausts the parser. + return False + if (not isinstance(document, dict) or set(document) != {"version", "fingerprint", "sessions"} + or type(document["version"]) is not int or document["version"] != SESSION_FILE_VERSION): + return False + stored = document["fingerprint"] + if not isinstance(stored, str) or not _FINGERPRINT.match(stored): + return False + # A snapshot for another key epoch (including a disabled key) is never adopted. + if not key or not hmac.compare_digest(stored, self._fingerprint(key)): + return False + entries = document["sessions"] + if not isinstance(entries, dict) or len(entries) > MAX_PERSISTED_SESSIONS: + return False + now = self.wall_clock() + restored = OrderedDict() + for sid, item in entries.items(): + if not isinstance(item, dict) or set(item) != {"csrf_token", "expires"}: + return False + token = self._token(sid) + csrf_token = self._token(item["csrf_token"]) + expires = self._deadline(item["expires"]) + if token is None or csrf_token is None or expires is None: + return False + if expires > now: + restored[token] = {"csrf_token": csrf_token, "expires": expires} + self.sessions.update(restored) + if len(restored) != len(entries): + # Expired records were dropped: rewrite the snapshot, so a wall clock that + # later moves backward cannot restore them from the file we just read. + # A failure here is recorded as degraded state by _persist(); the entries we + # already adopted stay valid, so this is not a reason to reject the snapshot. + self._persist() + return True + + def _persist(self): + """Atomically rewrite the session table; returns False when it was not durable.""" + if self._path is None: + return True + if not self._configured_key: + # A disabled key revokes everywhere; drop the snapshot instead of writing one. + return self._revoke() + document = {"version": SESSION_FILE_VERSION, "fingerprint": self._fingerprint(self._configured_key), + "sessions": {sid: {"csrf_token": item["csrf_token"], "expires": item["expires"]} + for sid, item in self.sessions.items()}} + try: + content = json.dumps(document, ensure_ascii=False, separators=(",", ":"), + allow_nan=False).encode("utf-8") + except (TypeError, ValueError): + return False + temporary = None + try: + self._path.parent.mkdir(mode=0o700, parents=True, exist_ok=True) + fd, temporary = tempfile.mkstemp(prefix=".admin-sessions-", suffix=".tmp", dir=self._path.parent) + with os.fdopen(fd, "wb") as stream: + stream.write(content) + stream.flush() + os.fsync(stream.fileno()) + os.chmod(temporary, 0o600) + os.replace(temporary, self._path) + temporary = None + self._storage_ok() + except (OSError, ValueError) as error: + # Fall back to removing the stale snapshot so revoked sessions cannot return. + self._storage_failed(error) + return self._revoke() + finally: + if temporary is not None: + try: + os.unlink(temporary) + except OSError: + pass + return True def _key(self): key = self.config.get("api_key") or "" if not isinstance(key, str): key = "" if self._configured_key is None or not hmac.compare_digest(key.encode(), self._configured_key.encode()): + previous, self._configured_key = self._configured_key, key + restoring = previous is None self.sessions.clear() - self._configured_key = key # Restoring a previous key must not restore that epoch's OAuth owner. self._identity = secrets.token_urlsafe(32) + if restoring: + # Adopt only a snapshot that matches the current epoch; anything else is + # revoked, so switching back to an old key cannot resurrect its sessions. + durable = self._restore(key) or self._revoke() + else: + # A rotated or cleared key revokes every session, on disk as well. + durable = self._persist() + if not durable: + # Leave the epoch unactivated, so the failure is retried instead of being + # recorded as done. `_configured_key` is what _persist() fingerprints, so + # it has to be set for the attempt above and rolled back here. + self._configured_key = previous + raise SessionStoreError( + f"无法持久撤销上一 epoch 的会话快照（{self._storage_error or 'unknown'}）：" + f"{self._path}；请修复管理目录权限后重启，否则旧会话可能被复活") return key def csrf_enabled(self): @@ -92,6 +344,20 @@ def allowed_origins(self): """Extra trusted browser origins from hot configuration.""" return origin_allowlist(self.config.get("admin_allowed_origins")) + def reconcile(self): + """Resolve the key epoch at startup rather than on the first `/admin` request. + + `_key()` is otherwise lazy, so a process that only serves inference traffic never + reaches it and would leave the previous epoch's snapshot on disk. Restarting back + to the original key would then adopt that snapshot and resurrect a cookie which the + rotation in between was supposed to revoke. + + Raises SessionStoreError when that revocation cannot be made durable; the caller + must abort startup rather than activate the new epoch. + """ + with self.lock: + self._key() + def enabled(self): with self.lock: return bool(self._key()) @@ -114,9 +380,10 @@ def session(self, request): with self.lock: self._key() item = self.sessions.get(sid) - if item and item["expires"] > self.clock(): + if item and item["expires"] > self.wall_clock(): return sid, dict(item) - self.sessions.pop(sid, None) + if self.sessions.pop(sid, None) is not None: + self._persist() return None, None def login(self, request, key): @@ -140,15 +407,28 @@ def login(self, request, key): old = request.cookies.get(COOKIE_NAME) self.sessions.pop(old, None) sid = secrets.token_urlsafe(32) - item = {"csrf_token": secrets.token_urlsafe(32), "expires": now + SESSION_TTL} + item = {"csrf_token": secrets.token_urlsafe(32), "expires": self.wall_clock() + SESSION_TTL} self.sessions[sid] = item - while len(self.sessions) > 256: + while len(self.sessions) > MAX_PERSISTED_SESSIONS: self.sessions.popitem(last=False) + self._persist() return (sid, dict(item)), 200 def logout(self, request): + """Revoke the cookie's session; returns False when the snapshot could not be updated. + + The entry is kept when the snapshot cannot be rewritten, so a retry is still + authenticated and can finish the revocation instead of being acknowledged early. + """ + sid = request.cookies.get(COOKIE_NAME) with self.lock: - self.sessions.pop(request.cookies.get(COOKIE_NAME), None) + item = self.sessions.pop(sid, None) + if item is None: + return True + if self._persist(): + return True + self.sessions[sid] = item + return False class AdminMiddleware: @@ -166,7 +446,14 @@ async def no_cache(message): message = {**message, "headers": headers + [(b"cache-control", b"no-store"), (b"pragma", b"no-cache"), (b"expires", b"0")]} await send(message) - if not self.auth.enabled(): + try: + enabled = self.auth.enabled() + except SessionStoreError: + # The key epoch changed but its superseded snapshot survives. Deny rather than + # continue: startup refuses this case, so it is only reachable mid-process. + return await error_response( + 503, "会话快照无法持久撤销，管理接口已锁定；请检查管理目录权限后重启")(scope, receive, no_cache) + if not enabled: return await error_response(503, "未配置 API key，管理接口已锁定")(scope, receive, no_cache) path, method = scope["path"], scope["method"] public_session = path == "/admin/session" and method in ("POST", "GET") diff --git a/app/runtime_management.py b/app/runtime_management.py index 460de2a..ba0b1f3 100644 --- a/app/runtime_management.py +++ b/app/runtime_management.py @@ -66,6 +66,9 @@ def initialize(gateway, args, argv=None, *, parser=None): root = gateway.managed_auth_dir() control = ControlStore(root / "control.sqlite3") config["control_store"] = control + # Persist management sessions so a restart does not force another API-key login. + # The file stores an HMAC fingerprint of the key epoch, so key rotation still revokes them. + config["session_path"] = root / "admin-sessions.json" config.update(vars(args)) config["model_guard"] = not args.no_model_guard aliases = {"log": "log_path", "no_model_guard": "model_guard"} diff --git a/converter.py b/converter.py index 36198c5..f485007 100644 --- a/converter.py +++ b/converter.py @@ -68,6 +68,7 @@ def desensitize_body(body, roles=("system",), desensitize_harness_user=False, from app.message_normalization import merge_intl_user_images from app.model_catalog_view import INTERNATIONAL as SHARED_INTL_PROFILES, share_models from app.inference_auth import require_api_key +from app.admin_auth import SessionStoreError from app.content_filter import ContentFilterDetector, is_filter_error from app.request_limits import ImageLimitError, apply_image_policy from app.safe_logging import format_log_body, sanitize_log_text @@ -3919,7 +3920,13 @@ def main(): if CONFIG["usage_snapshots"].detail(): _publish_usage_daily(CONFIG["cred_pool"]) - runtime_management.install(sys.modules[__name__]) + try: + runtime_management.install(sys.modules[__name__]) + except SessionStoreError as error: + # An obsolete session snapshot survived, so the new key epoch must not activate: + # a later start under the superseded key could adopt it and revive admin cookies. + runtime_management.close(CONFIG) + ap.error(str(error)) threading.Thread(target=_refresher_loop, args=(CONFIG["cred_pool"],), daemon=True, name="cred-refresher").start() if credits_mod is not None: diff --git a/tests/test_admin_api.py b/tests/test_admin_api.py index a8addc8..76ef1b4 100644 --- a/tests/test_admin_api.py +++ b/tests/test_admin_api.py @@ -22,7 +22,8 @@ def setUp(self): self.root = Path(self.enterContext(tempfile.TemporaryDirectory())) self.store = ControlStore(self.root / "control.sqlite3") self.addCleanup(self.store.close) - self.config = {"api_key": "synthetic-key", "control_store": self.store, "max_images": 16} + self.config = {"api_key": "synthetic-key", "control_store": self.store, "max_images": 16, + "session_path": self.root / "admin-sessions.json"} self.audit = Mock() self.audit.storage.return_value = {"db_bytes": 0} self.audit.list_records.return_value = {"items": [], "next_cursor": None, "has_more": False} @@ -320,6 +321,39 @@ def test_session_cookie_flags_logout_and_rotation(self): self.assertEqual(self.client.get("/admin/settings", headers=self.headers).status_code, 401) self.assertEqual(self.client.get("/admin/settings", headers={"X-Api-Key": "rotated-key"}).status_code, 200) + def test_logout_reports_when_the_session_could_not_be_persistently_revoked(self): + """A logout that cannot reach durable storage must not answer `authenticated: false`.""" + from unittest.mock import patch + response = self.client.post("/admin/session", json={"api_key": "synthetic-key"}, headers={"Origin": "http://testserver"}) + csrf = {"Origin": "http://testserver", "X-CSRF-Token": response.json()["csrf_token"]} + old = self.client.cookies.get(COOKIE_NAME) + with patch("app.admin_auth.tempfile.mkstemp", side_effect=OSError("read-only")), \ + patch("app.admin_auth.os.unlink", side_effect=OSError("read-only")): + denied = self.client.delete("/admin/session", headers=csrf) + self.assertEqual(denied.status_code, 503) + self.assertTrue(self.auth.storage()["degraded"]) + # The cookie is still valid, so the client can retry rather than silently lose access. + self.assertEqual(self.client.get("/admin/settings", headers={"Cookie": f"{COOKIE_NAME}={old}"}).status_code, 200) + self.assertEqual(self.client.delete("/admin/session", headers=csrf).status_code, 200) + self.assertEqual(self.client.get("/admin/settings", headers={"Cookie": f"{COOKIE_NAME}={old}"}).status_code, 401) + + def test_session_storage_state_is_reported_in_settings(self): + state = self.client.get("/admin/settings", headers=self.headers).json()["session"] + self.assertFalse(state["degraded"]) + self.assertIsNone(state["last_error"]) + self.assertTrue(state["path"]) + + def test_a_superseded_snapshot_that_cannot_be_revoked_fails_closed(self): + """Mid-process epoch changes must deny with 503, never serve management traffic.""" + from unittest.mock import patch + self.assertEqual(self.client.get("/admin/settings", headers=self.headers).status_code, 200) + self.config["api_key"] = "rotated-synthetic-key" + with patch.object(type(self.auth), "_persist", return_value=False), \ + patch.object(type(self.auth), "_revoke", return_value=False): + denied = self.client.get("/admin/settings", headers=self.headers) + self.assertEqual(denied.status_code, 503) + self.assertIn("会话快照", denied.json()["error"]["message"]) + def test_https_cookie_and_bounded_sessions(self): from starlette.requests import Request secure = self.enterContext(TestClient(self.app, base_url="https://testserver")) diff --git a/tests/test_admin_boundaries.py b/tests/test_admin_boundaries.py index e9c15c3..ff12c9c 100644 --- a/tests/test_admin_boundaries.py +++ b/tests/test_admin_boundaries.py @@ -4,15 +4,19 @@ sys.path.insert(0, str(Path(__file__).resolve().parents[1])) +import os +import stat import tempfile import unittest +from contextlib import contextmanager from unittest.mock import patch from fastapi import FastAPI from fastapi.testclient import TestClient from starlette.requests import Request -from app.admin_auth import AdminAuth, COOKIE_NAME +from app.admin_auth import (AdminAuth, COOKIE_NAME, MAX_PERSISTED_SESSIONS, SESSION_FILE_VERSION, + SESSION_TTL, SessionStoreError) from app.gateway_management import install_pages @@ -79,6 +83,347 @@ def test_key_comparison_remains_constant_time(self): checked.assert_any_call(b"wrong-key", self.config["api_key"].encode()) +class PersistedSessionTests(unittest.TestCase): + """A restart must not force another login, while revocation must stay durable.""" + + def setUp(self): + self.directory = Path(self.enterContext(tempfile.TemporaryDirectory())) + self.path = self.directory / "admin-sessions.json" + self.key = "synthetic-management-key" + self.config = {"api_key": self.key, "session_path": self.path} + + def login(self, auth, key=None): + (sid, session), status = auth.login(request(), self.key if key is None else key) + self.assertEqual(status, 200) + return sid, session + + def revive(self, sid, config=None): + """Model a full process restart against the same session file.""" + return AdminAuth(dict(config or self.config)).session(request(cookie=sid)) + + def assert_revoked(self, sid): + """Assert that a restart cannot revive the session.""" + self.assertIsNone(self.revive(sid)[0]) + + def write_document(self, **document): + import json + self.path.write_text(json.dumps(document), encoding="utf-8") + + def write_document_at(self, path, **document): + import json + path.write_text(json.dumps(document), encoding="utf-8") + + def read_document(self): + import json + return json.loads(self.path.read_text(encoding="utf-8")) + + def fingerprint(self, key=None): + return AdminAuth._fingerprint(self.key if key is None else key) + + def test_session_survives_a_restart(self): + sid, session = self.login(AdminAuth(self.config)) + self.assertTrue(self.path.exists()) + self.assertEqual(self.revive(sid), (sid, session)) + + def test_restored_session_keeps_its_csrf_token_and_deadline(self): + first = AdminAuth(self.config) + sid, session = self.login(first) + restored = self.revive(sid)[1] + self.assertEqual(restored["csrf_token"], session["csrf_token"]) + self.assertEqual(restored["expires"], session["expires"]) + + def test_expiry_uses_the_injected_wall_clock_across_instances(self): + now = [1_000_000.0] + config = dict(self.config) + sid, _ = self.login(AdminAuth(config, wall_clock=lambda: now[0])) + alive = AdminAuth(dict(config), wall_clock=lambda: now[0] + SESSION_TTL - 1) + self.assertEqual(alive.session(request(cookie=sid))[0], sid) + # A restart past the deadline drops the session, on disk as well as in memory, so + # only the injected wall clock — not a monotonic one — decides expiry. + expired = AdminAuth(dict(config), wall_clock=lambda: now[0] + SESSION_TTL + 1) + self.assertIsNone(expired.session(request(cookie=sid))[0]) + self.assertNotIn(sid, self.read_document()["sessions"]) + + def test_rotated_key_discards_persisted_sessions(self): + sid, _ = self.login(AdminAuth(self.config)) + rotated = {"api_key": "rotated-synthetic-key", "session_path": self.path} + self.assertIsNone(AdminAuth(rotated).session(request(cookie=sid))[0]) + # Switching back to the original key must never resurrect the old epoch. + self.assert_revoked(sid) + + def test_starting_with_a_disabled_key_revokes_persisted_sessions(self): + sid, _ = self.login(AdminAuth(self.config)) + AdminAuth({"api_key": "", "session_path": self.path}).enabled() + self.assert_revoked(sid) + + def test_startup_reconciles_without_any_admin_request(self): + """An inference-only process must still retire the superseded key's snapshot. + + `_key()` is lazy and inference traffic bypasses `AdminMiddleware`, so without an + explicit startup reconcile the intermediate process never touches the file and + restarting back to the original key adopts it. + """ + for intermediate in ("rotated-synthetic-key", ""): + with self.subTest(intermediate=intermediate): + sid, _ = self.login(AdminAuth(self.config)) + # A restart that only serves inference: construct and reconcile, no request. + AdminAuth({"api_key": intermediate, "session_path": self.path}).reconcile() + self.assert_revoked(sid) + + def test_startup_reconcile_happens_for_the_installed_admin(self): + """`install_admin` must reconcile, not wait for the first `/admin` request.""" + from unittest.mock import Mock + from app.admin_api import install_admin + from fastapi import FastAPI + config = dict(self.config, control_store=Mock(), audit_store=Mock()) + with patch.object(AdminAuth, "reconcile", autospec=True) as reconciled: + install_admin(FastAPI(), config, Mock()) + reconciled.assert_called_once() + + def test_logout_revokes_the_persisted_session(self): + first = AdminAuth(self.config) + sid, _ = self.login(first) + self.assertTrue(first.logout(request(cookie=sid))) + self.assert_revoked(sid) + + def test_logout_reports_a_revocation_that_could_not_be_persisted(self): + """Both the rewrite and the unlink fallback can fail; that must not be acknowledged.""" + first = AdminAuth(self.config) + sid, _ = self.login(first) + with patch("app.admin_auth.tempfile.mkstemp", side_effect=OSError("read-only")), \ + patch("app.admin_auth.os.unlink", side_effect=OSError("read-only")): + self.assertFalse(first.logout(request(cookie=sid))) + self.assertTrue(first.storage()["degraded"]) + # The session is still live, so a retry can still revoke it once storage recovers. + self.assertEqual(first.session(request(cookie=sid))[0], sid) + self.assertTrue(first.logout(request(cookie=sid))) + self.assertFalse(first.storage()["degraded"]) + self.assert_revoked(sid) + + def test_in_process_revocation_clears_the_snapshot(self): + for disabled in ("", None, 1, []): + with self.subTest(disabled=disabled): + config = dict(self.config) + auth = AdminAuth(config) + sid, _ = self.login(auth) + config["api_key"] = disabled + self.assertFalse(auth.enabled()) + self.assertEqual(auth.session(request(cookie=sid)), (None, None)) + self.assertFalse(self.path.exists()) # Revoked on disk, not only in memory. + self.assert_revoked(sid) + + def test_expired_sessions_are_not_restored(self): + sid, _ = self.login(AdminAuth(self.config)) + self.write_document(version=SESSION_FILE_VERSION, fingerprint=self.fingerprint(), + sessions={sid: {"csrf_token": "a" * 32, "expires": 1}}) + self.assert_revoked(sid) + + def test_symlinked_snapshot_is_never_followed(self): + import json + # Both cases matter: a link to a plain file, and — the one that can actually tell + # "rejected" from "parsed and discarded" — a link to a valid session snapshot. + sentinel = self.directory / "sentinel.json" + sentinel.write_text("outside-sentinel", encoding="utf-8") + target = self.directory / "real.json" + self.write_document_at(target, version=SESSION_FILE_VERSION, fingerprint=self.fingerprint(), + sessions={"a" * 32: {"csrf_token": "b" * 32, "expires": 9_999_999_999}}) + for name, victim in (("plain file", sentinel), ("valid snapshot", target)): + with self.subTest(target=name): + link = self.directory / f"link-{name.replace(' ', '-')}.json" + try: + link.symlink_to(victim) + except (OSError, NotImplementedError): + self.skipTest("symlinks are unavailable on this platform") + auth = AdminAuth({"api_key": self.key, "session_path": link}) + self.assertTrue(auth.enabled()) # Rejected, not a startup crash. + self.assertEqual(dict(auth.sessions), {}) # The target was never adopted. + self.assertEqual(auth.session(request(cookie="a" * 32)), (None, None)) + self.assertTrue(victim.exists()) # And was left untouched. + self.assertIn("a" * 32, json.loads(target.read_text(encoding="utf-8"))["sessions"]) + + def test_expired_records_are_dropped_from_the_snapshot(self): + sid, _ = self.login(AdminAuth(self.config)) + self.write_document(version=SESSION_FILE_VERSION, fingerprint=self.fingerprint(), + sessions={sid: {"csrf_token": "a" * 32, "expires": 1_000.0}}) + later = AdminAuth(dict(self.config), wall_clock=lambda: 5_000.0) + self.assertTrue(later.enabled()) + self.assertEqual(dict(later.sessions), {}) + # The record is gone from disk, so a wall clock that moves backward cannot revive it. + self.assertNotIn(sid, self.read_document()["sessions"]) + rolled_back = AdminAuth(dict(self.config), wall_clock=lambda: 500.0) + self.assertTrue(rolled_back.enabled()) + self.assertIsNone(rolled_back.session(request(cookie=sid))[0]) + + def test_untrusted_snapshots_are_revoked_rather_than_adopted(self): + import json + sid, _ = self.login(AdminAuth(self.config)) + live = {"csrf_token": "a" * 32, "expires": 9_999_999_999} + huge = "9" * 400 + documents = { + "not-json": b"not json", + "empty object": b"{}", + "unsupported version": json.dumps({"version": 99, "fingerprint": self.fingerprint(), "sessions": {}}).encode(), + "boolean version": json.dumps({"version": True, "fingerprint": self.fingerprint(), "sessions": {}}).encode(), + "foreign fingerprint": json.dumps({"version": SESSION_FILE_VERSION, "fingerprint": "0" * 64, + "sessions": {sid: live}}).encode(), + "non-ascii fingerprint": json.dumps({"version": SESSION_FILE_VERSION, "fingerprint": "\u4e2d\u6587", + "sessions": {}}).encode(), + "unbounded expiry": json.dumps({"version": SESSION_FILE_VERSION, "fingerprint": self.fingerprint(), + "sessions": {sid: {"csrf_token": "a" * 32, "expires": 1e12}}}).encode(), + "non-finite expiry": b'{"version": 1, "fingerprint": "' + self.fingerprint().encode() + + b'", "sessions": {"' + sid.encode() + b'": {"csrf_token": "' + + b'a' * 32 + b'", "expires": Infinity}}}', + "oversized sid": json.dumps({"version": SESSION_FILE_VERSION, "fingerprint": self.fingerprint(), + "sessions": {"a" * 500: live}}).encode(), + "oversized csrf": json.dumps({"version": SESSION_FILE_VERSION, "fingerprint": self.fingerprint(), + "sessions": {sid: {"csrf_token": "a" * 500, "expires": 9e9}}}).encode(), + "duplicate field": ('{"version": 1, "fingerprint": "' + self.fingerprint() + + '", "fingerprint": "' + self.fingerprint() + '", "sessions": {}}').encode(), + "extra field": json.dumps({"version": SESSION_FILE_VERSION, "fingerprint": self.fingerprint(), + "sessions": {}, "extra": 1}).encode(), + "oversized file": b"x" * (300 * 1024), + # An integer too large for float() must not raise OverflowError out of enabled(). + "expiry beyond float": ('{"version": 1, "fingerprint": "' + self.fingerprint() + + '", "sessions": {"' + sid + '": {"csrf_token": "' + "a" * 32 + + '", "expires": ' + huge + '}}}').encode(), + "negative expiry beyond float": ('{"version": 1, "fingerprint": "' + self.fingerprint() + + '", "sessions": {"' + sid + '": {"csrf_token": "' + "a" * 32 + + '", "expires": -' + huge + '}}}').encode(), + # Deeply nested JSON inside the size bound must not raise RecursionError either. + "deeply nested": b"[" * 50_000 + b"]" * 50_000, + } + for name, content in documents.items(): + with self.subTest(name=name): + self.path.write_bytes(content) + auth = AdminAuth(dict(self.config)) + self.assertTrue(auth.enabled()) # Malformed input never breaks startup. + self.assertEqual(auth.session(request(cookie=sid)), (None, None)) + self.assertFalse(self.path.exists()) # It is revoked, not left for a later start. + self.assert_revoked(sid) + + def test_write_failure_revokes_instead_of_leaving_a_stale_snapshot(self): + first = AdminAuth(self.config) + sid, _ = self.login(first) + with patch("app.admin_auth.tempfile.mkstemp", side_effect=OSError("disk full")): + first.logout(request(cookie=sid)) + self.assert_revoked(sid) + + def test_key_rotation_survives_a_write_failure(self): + sid, _ = self.login(AdminAuth(self.config)) + with patch("app.admin_auth.tempfile.mkstemp", side_effect=OSError("disk full")): + AdminAuth({"api_key": "rotated-synthetic-key", "session_path": self.path}).enabled() + self.assert_revoked(sid) + + @contextmanager + def undeletable_snapshot(self): + """Lock the snapshot so the atomic replace and the unlink both fail. + + Windows protects a file by its read-only mode; POSIX protects removal by directory + permission, so the directory is locked too. Whether the lock actually took effect is + checked against a throwaway probe, never against the snapshot itself, so a platform + where removal cannot be blocked is skipped instead of silently passing. + """ + probe = self.directory / "probe.tmp" + originals = [(self.path, stat.S_IMODE(self.path.stat().st_mode))] + + def arm(): + """Recreate the probe, then apply the same protection the snapshot gets.""" + os.chmod(self.path, stat.S_IREAD) + try: + os.chmod(probe, 0o600) + os.unlink(probe) + except OSError: + pass + probe.write_text("x", encoding="utf-8") + os.chmod(probe, stat.S_IREAD) + + def removal_blocked(): + try: + os.unlink(probe) + except OSError: + return True + return False + + def restore(): + for target, mode in reversed(originals): + try: + os.chmod(target, mode) + except OSError: + pass + + arm() + if not removal_blocked(): + # The file mode did not protect it (POSIX), so lock the directory as well. The + # probe has to be recreated while the directory is still writable. + arm() + originals.append((self.directory, stat.S_IMODE(self.directory.stat().st_mode))) + os.chmod(self.directory, stat.S_IREAD | stat.S_IEXEC) + if not removal_blocked(): + restore() + self.skipTest("this platform cannot make the snapshot undeletable") + try: + yield + finally: + restore() + try: + os.chmod(probe, 0o600) + os.unlink(probe) + except OSError: + pass + + def test_startup_fails_when_a_superseded_snapshot_cannot_be_revoked(self): + """Activating a new epoch over a surviving snapshot would let a later start revive it. + + The rotation must abort instead, leaving the epoch unactivated so a retry can + finish the job once storage recovers. + """ + for label, intermediate in (("rotation", "rotated-synthetic-key"), ("disabled key", "")): + with self.subTest(case=label): + sid, _ = self.login(AdminAuth(self.config)) + with self.undeletable_snapshot(): + auth = AdminAuth({"api_key": intermediate, "session_path": self.path}) + with self.assertRaises(SessionStoreError): + auth.reconcile() + # Startup failed, so the epoch was not activated and is retried. + with self.assertRaises(SessionStoreError): + auth.reconcile() + self.assertTrue(auth.storage()["degraded"]) + # Once storage recovers, the retry revokes the snapshot for real. + auth.reconcile() + self.assertFalse(auth.storage()["degraded"]) + self.assert_revoked(sid) + + def test_startup_failure_reaches_the_installed_admin(self): + """`install_admin` must propagate the failure rather than completing startup.""" + from unittest.mock import Mock + from app.admin_api import install_admin + from fastapi import FastAPI + config = dict(self.config, control_store=Mock(), audit_store=Mock()) + with patch.object(AdminAuth, "reconcile", autospec=True, + side_effect=SessionStoreError("snapshot survives")): + with self.assertRaises(SessionStoreError): + install_admin(FastAPI(), config, Mock()) + + def test_missing_path_keeps_sessions_in_memory_only(self): + auth = AdminAuth({"api_key": self.key}) + sid, _ = self.login(auth) + self.assertEqual(list(self.directory.iterdir()), []) + self.assertIsNone(AdminAuth({"api_key": self.key}).session(request(cookie=sid))[0]) + + def test_snapshot_bounds_the_session_table(self): + auth = AdminAuth(self.config) + for _ in range(MAX_PERSISTED_SESSIONS + 20): + self.login(auth) + self.assertLessEqual(len(auth.sessions), MAX_PERSISTED_SESSIONS) + self.assertLessEqual(len(self.revive(next(iter(auth.sessions)))[1]), 2) + + def test_snapshot_is_owner_only_where_the_platform_supports_it(self): + import stat as stat_module + self.login(AdminAuth(self.config)) + if sys.platform != "win32": + self.assertEqual(stat_module.S_IMODE(self.path.stat().st_mode), 0o600) + + class StaticBoundaryTests(unittest.TestCase): def setUp(self): self.directory = Path(self.enterContext(tempfile.TemporaryDirectory())) diff --git a/web/e2e/admin.spec.ts b/web/e2e/admin.spec.ts index 6ebab48..f23be5b 100644 --- a/web/e2e/admin.spec.ts +++ b/web/e2e/admin.spec.ts @@ -68,6 +68,7 @@ const settings = { shm_bytes: 32768, degraded: false, }, + session: { degraded: false, last_error: null, path: "/tmp/admin-sessions.json" }, }; async function mockAPI(page: Page, authenticated = true) { const calls: { method: string; path: string; body: unknown }[] = []; diff --git a/web/src/copy.test.tsx b/web/src/copy.test.tsx index 4e06458..002d6f2 100644 --- a/web/src/copy.test.tsx +++ b/web/src/copy.test.tsx @@ -50,6 +50,7 @@ const settings = { }, ], audit: { degraded: false, logical_bytes: 0 }, + session: { degraded: false, last_error: null, path: "/tmp/admin-sessions.json" }, }; describe("concise user-facing copy", () => { @@ -105,6 +106,20 @@ describe("concise user-facing copy", () => { expect(screen.getByText("查看存储详情").closest("details")?.open).toBe(false); }); + it("surfaces a session store that could not persist a revocation", () => { + resource({ ...settings, session: { degraded: true, last_error: "OSError" } }); + render(); + expect(screen.getByText("会话存储")).toBeTruthy(); + expect(screen.getByText("撤销未生效")).toBeTruthy(); + expect(screen.getByText(/退出登录可能未真正生效/)).toBeTruthy(); + }); + + it("stays quiet while the session store is healthy", () => { + resource(settings); + render(); + expect(screen.queryByText("会话存储")).toBeNull(); + }); + it("still submits the revision and only edited settings", async () => { resource(settings); const patch = vi.spyOn(api, "patch").mockResolvedValue({ data: {} }); diff --git a/web/src/pages/Settings.tsx b/web/src/pages/Settings.tsx index a8c7a62..6c1b4b2 100644 --- a/web/src/pages/Settings.tsx +++ b/web/src/pages/Settings.tsx @@ -41,7 +41,12 @@ const modeLabels: Record = { function displayValue(value: unknown) { return typeof value === "boolean" ? (value ? "开启" : "关闭") : text(value); } -function normalize(value: unknown): { revision: number; items: Setting[]; audit: RecordValue } { +function normalize(value: unknown): { + revision: number; + items: Setting[]; + audit: RecordValue; + session: RecordValue; +} { const d = object(value); if (typeof d.revision !== "number") throw new Error("设置响应缺少 revision"); const items = list(d.items).map((item) => { @@ -69,7 +74,12 @@ function normalize(value: unknown): { revision: number; items: Setting[]; audit: max: typeof item.max === "number" ? item.max : undefined, }; }); - return { revision: d.revision, items, audit: object(d.audit, "审计状态") }; + return { + revision: d.revision, + items, + audit: object(d.audit, "审计状态"), + session: object(d.session ?? {}, "会话存储状态"), + }; } function SettingsForm({ data, @@ -241,6 +251,16 @@ function SettingsForm({ + {data.session.degraded === true && ( + + 撤销未生效 +

会话快照无法写入或清除，退出登录可能未真正生效。请检查管理目录权限后重试退出登录。

+
+ 查看存储详情 + +
+ + )} ); }