From 30dfafe59f089dad787455a8e01f21d91c383bf4 Mon Sep 17 00:00:00 2001
From: moshouhot <moshouhot@gmail.com>
Date: Sat, 19 Sep 2026 18:02:11 +0800
Subject: [PATCH 1/5] Persist admin sessions so a restart does not force
 another login

The management console kept its session table in memory only, so every
restart of the gateway invalidated the cb_admin_session cookie and the
WebUI asked for the API key again. This is easy to hit when updating or
restarting the service.

Sessions are now written to admin-sessions.json under the managed auth
directory and restored on startup.

Persisting a login must not weaken the existing revocation guarantees.
Revocation works by clearing the stored snapshot; the file additionally
carries a keyed fingerprint naming the key epoch it belongs to, and a
snapshot whose fingerprint does not match the current key is revoked
rather than adopted. That covers key rotation, a disabled key, logout and
expiry, in-process and across restarts.

Session expiry moves from time.monotonic to a wall clock, because a
monotonic clock restarts with the machine and cannot express an absolute
deadline across restarts. Login throttling still uses the monotonic clock,
which is the appropriate choice for a sliding window.

The snapshot is size bounded, read with a bounded read and a regular-file
check without following symlinks, parsed as strict JSON, and validated
field by field; a snapshot that fails validation is removed so it cannot
be adopted later. Writes go through a mode-0600 temporary file and an
atomic replace, with revocation of the previous snapshot as the fallback
when the write cannot complete. Note that chmod does not establish
owner-only ACLs on Windows.

When no path is configured the previous in-memory behaviour is unchanged.
---
 app/admin_auth.py              | 206 +++++++++++++++++++++++++++++++--
 app/runtime_management.py      |   3 +
 tests/test_admin_boundaries.py | 172 ++++++++++++++++++++++++++-
 3 files changed, 372 insertions(+), 9 deletions(-)

diff --git a/app/admin_auth.py b/app/admin_auth.py
index 701df8c..832751a 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
@@ -14,12 +22,54 @@
 COOKIE_NAME = "cb_admin_session"
 SESSION_TTL = 12 * 3600
 
+# 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 +114,161 @@ 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"))
+
+    @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
+        number = float(value)
+        return number if math.isfinite(number) and 0 < number < 1e11 else None
+
+    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:
+            return False
+        return True
+
+    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
+                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):
+            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)
+        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
+        except (OSError, ValueError):
+            # Fall back to removing the stale snapshot so revoked sessions cannot return.
+            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()):
-            self.sessions.clear()
+            restoring = self._configured_key is None
             self._configured_key = key
+            self.sessions.clear()
             # 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.
+                if not self._restore(key):
+                    self._revoke()
+            else:
+                # A rotated or cleared key revokes every session, on disk as well.
+                self._persist()
         return key
 
     def csrf_enabled(self):
@@ -114,9 +301,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 +328,17 @@ 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):
         with self.lock:
-            self.sessions.pop(request.cookies.get(COOKIE_NAME), None)
+            if self.sessions.pop(request.cookies.get(COOKIE_NAME), None) is not None:
+                self._persist()
 
 
 class AdminMiddleware:
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/tests/test_admin_boundaries.py b/tests/test_admin_boundaries.py
index e9c15c3..edec0bb 100644
--- a/tests/test_admin_boundaries.py
+++ b/tests/test_admin_boundaries.py
@@ -12,7 +12,7 @@
 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
 from app.gateway_management import install_pages
 
 
@@ -79,6 +79,176 @@ 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 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)
+        first = AdminAuth(config, wall_clock=lambda: now[0])
+        sid, _ = self.login(first)
+        expired = AdminAuth(dict(config), wall_clock=lambda: now[0] + SESSION_TTL + 1)
+        self.assertIsNone(expired.session(request(cookie=sid))[0])
+        alive = AdminAuth(dict(config), wall_clock=lambda: now[0] + SESSION_TTL - 1)
+        self.assertEqual(alive.session(request(cookie=sid))[0], sid)
+
+    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_logout_revokes_the_persisted_session(self):
+        first = AdminAuth(self.config)
+        sid, _ = self.login(first)
+        first.logout(request(cookie=sid))
+        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_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}
+        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),
+        }
+        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)
+
+    def test_symlinked_snapshot_is_never_followed(self):
+        target = self.directory / "real.json"
+        target.write_text("outside-sentinel", encoding="utf-8")
+        link = self.directory / "link.json"
+        try:
+            link.symlink_to(target)
+        except (OSError, NotImplementedError):
+            self.skipTest("symlinks are unavailable on this platform")
+        auth = AdminAuth({"api_key": self.key, "session_path": link})
+        self.assertTrue(auth.enabled())
+        self.assertEqual(auth.session(request(cookie="sid")), (None, None))
+        self.assertTrue(target.exists())                     # The target is untouched.
+
+    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()))

From a4da831c9b895e7aec2fd52c57bcf5c8343ba532 Mon Sep 17 00:00:00 2001
From: moshouhot <moshouhot@gmail.com>
Date: Sat, 19 Sep 2026 22:09:35 +0800
Subject: [PATCH 2/5] Reject unusable snapshots instead of trusting or crashing
 on them

Four defects found in review of the persisted session snapshot:

- `_deadline` called `float()` on a JSON integer without guarding the
  conversion, so a value beyond the float range raised `OverflowError` out of
  `enabled()`. The exception escaped `_restore`, which also skipped the
  revoke-on-invalid path, leaving the file in place for the next start.
- `_strict_json` can raise `RecursionError` on deeply nested input well inside
  the 256 KiB size bound, with the same effect.
- `os.O_NOFOLLOW` does not exist on Windows, where `getattr(..., 0)` silently
  degraded the no-symlink guarantee to nothing and the link target was adopted.
- Expired records were filtered out of memory but left on disk, so a wall clock
  that moved backward before the next restart could revive them.

`_deadline` now rejects non-convertible numbers, `_restore` treats
`RecursionError` as malformed input, the opened file is compared against the
path itself so a link is rejected where `O_NOFOLLOW` is unavailable, and a
snapshot that dropped expired records is rewritten.

Tests: regression cases for each, including a symlink pointing at a valid
snapshot so following it would be observable. `tests/test_admin_boundaries.py`
and `tests/test_admin_api.py` pass (63); the full suite shows the same failures
as before the change.
---
 app/admin_auth.py              | 29 +++++++++++++++--
 tests/test_admin_boundaries.py | 59 +++++++++++++++++++++++++++++++---
 2 files changed, 82 insertions(+), 6 deletions(-)

diff --git a/app/admin_auth.py b/app/admin_auth.py
index 832751a..84f4205 100644
--- a/app/admin_auth.py
+++ b/app/admin_auth.py
@@ -143,7 +143,11 @@ def _deadline(value):
         """Accept only finite, in-range numeric deadlines; bool is not a deadline."""
         if type(value) not in (int, float):
             return None
-        number = float(value)
+        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 _revoke(self):
@@ -158,6 +162,18 @@ def _revoke(self):
             return False
         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.
 
@@ -178,6 +194,10 @@ def _restore(self, key):
                 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
@@ -185,7 +205,8 @@ def _restore(self, key):
             return False
         try:
             document = _strict_json(raw)
-        except (ValueError, UnicodeDecodeError):
+        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):
@@ -212,6 +233,10 @@ def _restore(self, key):
             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.
+            self._persist()
         return True
 
     def _persist(self):
diff --git a/tests/test_admin_boundaries.py b/tests/test_admin_boundaries.py
index edec0bb..b72e25d 100644
--- a/tests/test_admin_boundaries.py
+++ b/tests/test_admin_boundaries.py
@@ -105,6 +105,14 @@ 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)
 
@@ -123,12 +131,14 @@ def test_restored_session_keeps_its_csrf_token_and_deadline(self):
     def test_expiry_uses_the_injected_wall_clock_across_instances(self):
         now = [1_000_000.0]
         config = dict(self.config)
-        first = AdminAuth(config, wall_clock=lambda: now[0])
-        sid, _ = self.login(first)
-        expired = AdminAuth(dict(config), wall_clock=lambda: now[0] + SESSION_TTL + 1)
-        self.assertIsNone(expired.session(request(cookie=sid))[0])
+        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))
@@ -166,10 +176,42 @@ def test_expired_sessions_are_not_restored(self):
                             sessions={sid: {"csrf_token": "a" * 32, "expires": 1}})
         self.assert_revoked(sid)
 
+    def test_symlinked_snapshot_is_never_followed(self):
+        import json
+        # A valid snapshot sits at the target, so following the link would be observable
+        # even on platforms that do not provide O_NOFOLLOW.
+        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}})
+        link = self.directory / "link.json"
+        try:
+            link.symlink_to(target)
+        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(auth.session(request(cookie="a" * 32)), (None, None))
+        self.assertTrue(target.exists())                      # The target is 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"{}",
@@ -193,6 +235,15 @@ def test_untrusted_snapshots_are_revoked_rather_than_adopted(self):
             "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):

From 79386ba6f6979a94627b6a644d4e6fdd63dcd2a1 Mon Sep 17 00:00:00 2001
From: moshouhot <moshouhot@gmail.com>
Date: Sun, 20 Sep 2026 15:23:21 +0800
Subject: [PATCH 3/5] Revoke superseded sessions at startup and report failed
 revocations

Three regressions found while reviewing the persisted session snapshot.

Startup never resolved the key epoch. `_key()` is lazy and only `/admin`
requests reach it, so a restart that served inference traffic alone left the
previous key's snapshot on disk; restarting back to that key adopted it and
resurrected a cookie the rotation was meant to revoke. `install_admin` now
calls `AdminAuth.reconcile()`, so the epoch is resolved during startup,
including when management is disabled.

`logout` discarded the result of `_persist()`. When the session directory is
not writable both the replacement write and the `_revoke()` fallback fail, so
the session stayed valid on disk while `DELETE /admin/session` answered
`authenticated: false` and the client dropped its cookie. The entry is now
kept when the snapshot cannot be rewritten, so a retry can still revoke it, and
`session_delete` answers 503 instead of acknowledging a revocation that did not
happen. The failure is recorded as session-store state, surfaced in
`/admin/settings` as `session` alongside the existing `audit` state, and shown
in the settings page.

The strengthened symlink test was shadowed: `PersistedSessionTests` defined
`test_symlinked_snapshot_is_never_followed` twice, and the later definition
replaced the earlier one, so the case pointing at a valid snapshot never ran.
Both cases are now covered by one test that asserts the target is neither
adopted nor modified.

Tests: restart cases for A -> B -> A and A -> empty -> A that deliberately make
no admin request during the intermediate run, an installed-admin reconcile
check, logout with both writing and unlinking failing plus its retry, the 503
logout path over HTTP, the settings session state, and the WebUI warning panel.
`tests/test_admin_boundaries.py` and `tests/test_admin_api.py` pass (68);
`pnpm test` passes (132). The full Python suite shows the same failures as
before the change apart from the pre-existing flaky trial-ledger concurrency
test, which fails on `main` too.
---
 app/admin_api.py               | 11 ++++-
 app/admin_auth.py              | 54 +++++++++++++++++++++--
 tests/test_admin_api.py        | 25 ++++++++++-
 tests/test_admin_boundaries.py | 80 +++++++++++++++++++++++-----------
 web/e2e/admin.spec.ts          |  1 +
 web/src/copy.test.tsx          | 15 +++++++
 web/src/pages/Settings.tsx     | 24 +++++++++-
 7 files changed, 176 insertions(+), 34 deletions(-)

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 84f4205..871408a 100644
--- a/app/admin_auth.py
+++ b/app/admin_auth.py
@@ -124,6 +124,7 @@ def __init__(self, config, *, clock=time.monotonic, wall_clock=time.time):
         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):
@@ -150,6 +151,23 @@ def _deadline(value):
             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:
@@ -158,8 +176,10 @@ def _revoke(self):
             os.unlink(self._path)
         except FileNotFoundError:
             return True
-        except OSError:
+        except OSError as error:
+            self._storage_failed(error)
             return False
+        self._storage_ok()
         return True
 
     def _is_direct_file(self, metadata):
@@ -236,6 +256,8 @@ def _restore(self, key):
         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
 
@@ -265,8 +287,10 @@ def _persist(self):
             os.chmod(temporary, 0o600)
             os.replace(temporary, self._path)
             temporary = None
-        except (OSError, ValueError):
+            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:
@@ -304,6 +328,17 @@ 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.
+        """
+        with self.lock:
+            self._key()
+
     def enabled(self):
         with self.lock:
             return bool(self._key())
@@ -361,9 +396,20 @@ def login(self, request, key):
             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:
-            if self.sessions.pop(request.cookies.get(COOKIE_NAME), None) is not None:
-                self._persist()
+            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:
diff --git a/tests/test_admin_api.py b/tests/test_admin_api.py
index a8addc8..caa3511 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,28 @@ 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_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 b72e25d..d8de330 100644
--- a/tests/test_admin_boundaries.py
+++ b/tests/test_admin_boundaries.py
@@ -152,10 +152,48 @@ def test_starting_with_a_disabled_key_revokes_persisted_sessions(self):
         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)
-        first.logout(request(cookie=sid))
+        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):
@@ -178,20 +216,25 @@ def test_expired_sessions_are_not_restored(self):
 
     def test_symlinked_snapshot_is_never_followed(self):
         import json
-        # A valid snapshot sits at the target, so following the link would be observable
-        # even on platforms that do not provide O_NOFOLLOW.
+        # 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}})
-        link = self.directory / "link.json"
-        try:
-            link.symlink_to(target)
-        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(auth.session(request(cookie="a" * 32)), (None, None))
-        self.assertTrue(target.exists())                      # The target is untouched.
+        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):
@@ -267,19 +310,6 @@ def test_key_rotation_survives_a_write_failure(self):
             AdminAuth({"api_key": "rotated-synthetic-key", "session_path": self.path}).enabled()
         self.assert_revoked(sid)
 
-    def test_symlinked_snapshot_is_never_followed(self):
-        target = self.directory / "real.json"
-        target.write_text("outside-sentinel", encoding="utf-8")
-        link = self.directory / "link.json"
-        try:
-            link.symlink_to(target)
-        except (OSError, NotImplementedError):
-            self.skipTest("symlinks are unavailable on this platform")
-        auth = AdminAuth({"api_key": self.key, "session_path": link})
-        self.assertTrue(auth.enabled())
-        self.assertEqual(auth.session(request(cookie="sid")), (None, None))
-        self.assertTrue(target.exists())                     # The target is untouched.
-
     def test_missing_path_keeps_sessions_in_memory_only(self):
         auth = AdminAuth({"api_key": self.key})
         sid, _ = self.login(auth)
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(<Settings />);
+    expect(screen.getByText("会话存储")).toBeTruthy();
+    expect(screen.getByText("撤销未生效")).toBeTruthy();
+    expect(screen.getByText(/退出登录可能未真正生效/)).toBeTruthy();
+  });
+
+  it("stays quiet while the session store is healthy", () => {
+    resource(settings);
+    render(<Settings />);
+    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<string, string> = {
 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({
           <Fields data={data.audit} />
         </details>
       </Panel>
+      {data.session.degraded === true && (
+        <Panel title="会话存储">
+          <Badge tone="bad">撤销未生效</Badge>
+          <p>会话快照无法写入或清除，退出登录可能未真正生效。请检查管理目录权限后重试退出登录。</p>
+          <details className={s.chartData}>
+            <summary>查看存储详情</summary>
+            <Fields data={data.session} />
+          </details>
+        </Panel>
+      )}
     </form>
   );
 }

From e0bd1543c88427fe1248374859020f8327313e03 Mon Sep 17 00:00:00 2001
From: moshouhot <moshouhot@gmail.com>
Date: Sun, 20 Sep 2026 18:27:54 +0800
Subject: [PATCH 4/5] Refuse startup when a superseded session snapshot cannot
 be revoked

`_key()` discarded the result of the revocation it performs when the key epoch
changes, so a read-only `admin-sessions.json` made the rotation a silent no-op:
the file survived, the new epoch was recorded as active, and a later start under
the superseded key adopted it. Restarting A -> B -> A with the file left
read-only therefore accepted the original cookie again.

`_key()` now raises `SessionStoreError` when neither the rewrite nor the unlink
fallback made the revocation durable, and rolls `_configured_key` back so the
epoch stays unactivated and a later attempt retries rather than reporting the
work as done. `converter.main` aborts startup with a readable message instead of
serving traffic, and `AdminMiddleware` answers 503 if an epoch change degrades
after startup, so management traffic fails closed rather than open.

Tests use real permissions rather than mocked I/O: a read-only snapshot file
(plus a read-only directory where POSIX allows the unlink) with the sibling
SQLite stores left writable, covering rotation and disabled-key cases, the
failed retry, recovery once permissions are restored, propagation through
`install_admin`, and the 503 fail-closed path. The new startup test fails
against the previous `_key()` behaviour.

Verified end-to-end: `python converter.py` exits 2 with the reason while the
snapshot is undeletable, and still starts and serves `/health` with a valid
snapshot under a matching key. `tests/test_admin_boundaries.py`,
`tests/test_admin_api.py`, `tests/test_runtime_endpoints.py` and
`tests/test_inference_admission.py` pass (127); `pnpm test` passes (132); the
full Python suite matches its baseline apart from the pre-existing flaky
trial-ledger concurrency test.
---
 app/admin_auth.py              | 38 +++++++++++++++---
 converter.py                   |  9 ++++-
 tests/test_admin_api.py        | 11 ++++++
 tests/test_admin_boundaries.py | 71 +++++++++++++++++++++++++++++++++-
 4 files changed, 121 insertions(+), 8 deletions(-)

diff --git a/app/admin_auth.py b/app/admin_auth.py
index 871408a..79d3945 100644
--- a/app/admin_auth.py
+++ b/app/admin_auth.py
@@ -22,6 +22,15 @@
 COOKIE_NAME = "cb_admin_session"
 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.
@@ -305,19 +314,26 @@ def _key(self):
         if not isinstance(key, str):
             key = ""
         if self._configured_key is None or not hmac.compare_digest(key.encode(), self._configured_key.encode()):
-            restoring = self._configured_key is None
-            self._configured_key = key
+            previous, self._configured_key = self._configured_key, key
+            restoring = previous is None
             self.sessions.clear()
             # 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.
-                if not self._restore(key):
-                    self._revoke()
+                durable = self._restore(key) or self._revoke()
             else:
                 # A rotated or cleared key revokes every session, on disk as well.
-                self._persist()
+                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):
@@ -335,6 +351,9 @@ def reconcile(self):
         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()
@@ -427,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/converter.py b/converter.py
index 7581585..cb9a134 100644
--- a/converter.py
+++ b/converter.py
@@ -66,6 +66,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
@@ -3647,7 +3648,13 @@ def main():
             managed_auth_dir() / "model-catalog.json", ttl=args.model_catalog_ttl)
         CONFIG["cred_pool"].set_ledger(ledger)  # Verify balance ownership before publishing catalogs.
     _publish_model_cache()
-    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 caa3511..76ef1b4 100644
--- a/tests/test_admin_api.py
+++ b/tests/test_admin_api.py
@@ -343,6 +343,17 @@ def test_session_storage_state_is_reported_in_settings(self):
         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 d8de330..c7df726 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, MAX_PERSISTED_SESSIONS, SESSION_FILE_VERSION, SESSION_TTL
+from app.admin_auth import (AdminAuth, COOKIE_NAME, MAX_PERSISTED_SESSIONS, SESSION_FILE_VERSION,
+                            SESSION_TTL, SessionStoreError)
 from app.gateway_management import install_pages
 
 
@@ -310,6 +314,71 @@ def test_key_rotation_survives_a_write_failure(self):
             AdminAuth({"api_key": "rotated-synthetic-key", "session_path": self.path}).enabled()
         self.assert_revoked(sid)
 
+    @contextmanager
+    def undeletable_snapshot(self):
+        """Make the snapshot impossible to replace or remove, with real permissions.
+
+        Windows blocks unlink and the atomic replace on a read-only file; POSIX needs a
+        read-only directory for the same effect. Sibling files stay writable either way.
+        """
+        def locked():
+            try:
+                os.unlink(self.path)
+            except OSError:
+                return True
+            return False
+
+        originals = [(self.path, stat.S_IMODE(self.path.stat().st_mode))]
+        os.chmod(self.path, stat.S_IREAD)
+        if not locked():
+            originals.append((self.directory, stat.S_IMODE(self.directory.stat().st_mode)))
+            os.chmod(self.directory, stat.S_IREAD | stat.S_IEXEC)
+        if not locked():
+            for target, mode in reversed(originals):
+                os.chmod(target, mode)
+            self.skipTest("this platform cannot make the snapshot undeletable")
+        try:
+            yield
+        finally:
+            for target, mode in reversed(originals):
+                try:
+                    os.chmod(target, mode)
+                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)

From 3a8a3e5327779ad6e8fbaa2c0622bd4c5f5fd0f7 Mon Sep 17 00:00:00 2001
From: moshouhot <moshouhot@gmail.com>
Date: Sun, 20 Sep 2026 19:18:13 +0800
Subject: [PATCH 5/5] Make the undeletable-snapshot fixture portable

The fixture verified its own lock by unlinking the snapshot, which removes the
file it is supposed to protect. On Windows the read-only file mode blocks that
unlink and the probe doubled as the check, but on POSIX the file mode does not
affect removal, so the snapshot was deleted and the test asserted nothing while
still passing. It also re-created the probe after chmod'ing it read-only, which
raises on POSIX.

The check now runs against a throwaway probe file, the directory is locked only
after that check shows the file mode did not protect removal, and the probe is
re-created before being protected. Platforms where removal cannot be blocked
skip instead of silently passing.

Verified against simulated POSIX semantics (unlink gated on directory write
permission) and against a platform where every unlink succeeds: the POSIX path
locks the directory, keeps the snapshot undeletable, refuses the reconcile, and
restores the directory mode; the unblockable path skips. Windows behaviour is
unchanged and the test still fails against the previous `_key()`.
---
 tests/test_admin_boundaries.py | 59 ++++++++++++++++++++++++----------
 1 file changed, 42 insertions(+), 17 deletions(-)

diff --git a/tests/test_admin_boundaries.py b/tests/test_admin_boundaries.py
index c7df726..ff12c9c 100644
--- a/tests/test_admin_boundaries.py
+++ b/tests/test_admin_boundaries.py
@@ -316,36 +316,61 @@ def test_key_rotation_survives_a_write_failure(self):
 
     @contextmanager
     def undeletable_snapshot(self):
-        """Make the snapshot impossible to replace or remove, with real permissions.
+        """Lock the snapshot so the atomic replace and the unlink both fail.
 
-        Windows blocks unlink and the atomic replace on a read-only file; POSIX needs a
-        read-only directory for the same effect. Sibling files stay writable either way.
+        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.
         """
-        def locked():
+        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(self.path)
+                os.unlink(probe)
             except OSError:
                 return True
             return False
 
-        originals = [(self.path, stat.S_IMODE(self.path.stat().st_mode))]
-        os.chmod(self.path, stat.S_IREAD)
-        if not locked():
-            originals.append((self.directory, stat.S_IMODE(self.directory.stat().st_mode)))
-            os.chmod(self.directory, stat.S_IREAD | stat.S_IEXEC)
-        if not locked():
-            for target, mode in reversed(originals):
-                os.chmod(target, mode)
-            self.skipTest("this platform cannot make the snapshot undeletable")
-        try:
-            yield
-        finally:
+        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.