fix(session): split safe_key and _storage_key to fix WebUI coupling (#4533)

This commit is contained in:
axelray-dev
2026-06-27 16:52:54 +08:00
committed by Xubin Ren
parent cf2f589615
commit 00a907c493
4 changed files with 63 additions and 15 deletions
+6 -6
View File
@@ -401,17 +401,17 @@ class SessionManager:
@staticmethod
def safe_key(key: str) -> str:
"""Collision-resistant encoding of a session key for use as a filename stem.
"""Public helper used by HTTP handlers to map an arbitrary key to a stable filename stem."""
return safe_filename(key.replace(":", "_"))
Uses base64url (no padding) so distinct keys always map to distinct
filenames, unlike the previous replace(":", "_") approach which
could collide (e.g. telegram:a_b vs telegram:a:b).
"""
@staticmethod
def _storage_key(key: str) -> str:
"""Collision-resistant encoding for internal session storage filenames."""
return base64.urlsafe_b64encode(key.encode()).decode().rstrip("=")
def _get_session_path(self, key: str) -> Path:
"""Get the collision-resistant workspace path for a session."""
return self.sessions_dir / f"{self.safe_key(key)}.jsonl"
return self.sessions_dir / f"{self._storage_key(key)}.jsonl"
def _get_legacy_lossy_path(self, key: str) -> Path:
"""Previous workspace session path using lossy ':' to '_' replacement."""
+15 -1
View File
@@ -231,8 +231,22 @@ def _indexed_row_for_session(session: Session, path: Path) -> dict[str, Any]:
}
def _try_decode_storage_stem(stem: str) -> str | None:
"""Try to decode a base64url (no-padding) session storage stem back to the original key."""
import base64
try:
padding = 4 - len(stem) % 4
if padding != 4:
stem += "=" * padding
return base64.urlsafe_b64decode(stem).decode("utf-8")
except Exception:
return None
def _scan_session_row(session_manager: SessionManager, path: Path) -> dict[str, Any] | None:
fallback_key = path.stem.replace("_", ":", 1)
storage_key = _try_decode_storage_stem(path.stem)
fallback_key = storage_key or path.stem.replace("_", ":", 1)
try:
with open(path, encoding="utf-8") as f:
first_line = f.readline().strip()