fix(auth): preserve shared store during clone cleanup

This commit is contained in:
GuardianZ71
2026-09-06 00:37:13 -06:00
committed by kshitij
parent 3ac671dbba
commit 5e108489a0
3 changed files with 114 additions and 4 deletions
+6 -2
View File
@@ -702,7 +702,9 @@ def _write_private_file_atomic(
pass
def _save_auth_store(auth_store: Dict[str, Any], target_path: Optional[Path] = None) -> Path:
def _save_auth_store(
auth_store: Dict[str, Any], target_path: Optional[Path] = None, *,
preserve_symlinks: bool = True) -> Path:
"""Atomically persist *auth_store* (0o600, parent tightened to 0o700) to the active store, or to
an explicit *target_path* (e.g. the global-root write-through for rotating xAI OAuth grants)."""
auth_file = target_path if target_path is not None else _auth_file_path()
@@ -711,7 +713,9 @@ def _save_auth_store(auth_store: Dict[str, Any], target_path: Optional[Path] = N
# install tree (#25821, #93050).
auth_store["version"] = AUTH_STORE_VERSION
auth_store["updated_at"] = datetime.now(timezone.utc).isoformat()
_write_private_file_atomic(auth_file, json.dumps(auth_store, indent=2) + "\n", fsync_dir=True)
_write_private_file_atomic(
auth_file, json.dumps(auth_store, indent=2) + "\n",
replace=None if preserve_symlinks else os.replace, fsync_dir=True)
try:
auth_file.chmod(stat.S_IRUSR | stat.S_IWUSR)
except OSError:
+20 -2
View File
@@ -56,7 +56,7 @@ def strip_cloned_single_use_oauth_grants(profile_dir: Path) -> Dict[str, Any]:
"files": [...]}`` of what was stripped. Never raises: a clone must not fail because hygiene
could not run — the caller logs the summary.
"""
from hermes_cli.auth import _save_auth_store
from hermes_cli.auth import _same_path, _save_auth_store
stripped: Dict[str, Any] = {"pool": [], "providers": [], "files": []}
profile_dir = Path(profile_dir)
for name in SINGLE_USE_OAUTH_SINGLETON_FILES:
@@ -70,6 +70,22 @@ def strip_cloned_single_use_oauth_grants(profile_dir: Path) -> Dict[str, Any]:
auth_path = profile_dir / "auth.json"
if not auth_path.is_file():
return stripped
try:
from hermes_constants import get_default_hermes_root
root_auth_path = get_default_hermes_root() / "auth.json"
if _same_path(auth_path, root_auth_path):
return stripped
try:
root_auth_path.stat()
except FileNotFoundError:
pass # A missing root store cannot be the existing profile file.
else:
if auth_path.samefile(root_auth_path):
return stripped
except Exception:
# Fail closed: an unresolved root or transient stat failure must not turn an aliased
# shared store into a credential-deletion target.
return stripped
try:
store = json.loads(auth_path.read_text(encoding="utf-8-sig"))
except (OSError, json.JSONDecodeError):
@@ -106,7 +122,9 @@ def strip_cloned_single_use_oauth_grants(profile_dir: Path) -> Dict[str, Any]:
if not changed:
return stripped
try:
_save_auth_store(store, target_path=auth_path)
# Replace the profile entry itself: following a symlink introduced after the identity
# check could erase the shared root store.
_save_auth_store(store, target_path=auth_path, preserve_symlinks=False)
except Exception:
logger.debug(
"Failed to strip cloned single-use OAuth grants from %s", auth_path, exc_info=True)
@@ -160,6 +160,94 @@ def test_strip_helper_is_a_noop_without_credentials(tmp_path):
assert strip_cloned_single_use_oauth_grants(tmp_path) == {"pool": [], "providers": [], "files": []}
@pytest.mark.parametrize(
"link",
[
lambda target, alias: alias.symlink_to(target),
lambda target, alias: os.link(target, alias),
],
ids=["symlink", "hardlink"],
)
def test_strip_helper_leaves_shared_root_auth_store_unchanged(fleet, link):
"""A shared auth store is one grant, not a cloned credential copy."""
from hermes_cli.auth import strip_cloned_single_use_oauth_grants
root = fleet["root"]
_seed_codex_grant(root)
before = (root / "auth.json").read_text()
shared = _shared_profile(fleet, "shared", link=link)
assert strip_cloned_single_use_oauth_grants(shared) == {
"pool": [], "providers": [], "files": [],
}
assert (root / "auth.json").read_text() == before
def test_strip_helper_fails_closed_when_root_store_cannot_be_resolved(fleet, monkeypatch):
"""Credential hygiene must not mutate auth when store identity is unknown."""
from hermes_cli.auth import strip_cloned_single_use_oauth_grants
import hermes_constants
root = fleet["root"]
_seed_codex_grant(root)
before = (root / "auth.json").read_text()
shared = _shared_profile(
fleet, "shared", link=lambda target, alias: alias.symlink_to(target))
monkeypatch.setattr(
hermes_constants, "get_default_hermes_root",
lambda: (_ for _ in ()).throw(OSError("root unavailable")))
assert strip_cloned_single_use_oauth_grants(shared) == {
"pool": [], "providers": [], "files": [],
}
assert (root / "auth.json").read_text() == before
def test_strip_helper_fails_closed_when_store_identity_check_errors(fleet, monkeypatch):
"""A transient stat failure must not be interpreted as two stores."""
from hermes_cli.auth import strip_cloned_single_use_oauth_grants
root = fleet["root"]
_seed_codex_grant(root)
copied = _profile(fleet, "copied")
(copied / "auth.json").write_text((root / "auth.json").read_text())
before = (copied / "auth.json").read_text()
monkeypatch.setattr(
type(copied), "samefile",
lambda *args, **kwargs: (_ for _ in ()).throw(OSError("stat unavailable")))
assert strip_cloned_single_use_oauth_grants(copied) == {
"pool": [], "providers": [], "files": [],
}
assert (copied / "auth.json").read_text() == before
def test_strip_helper_does_not_follow_auth_symlink_created_during_save(fleet, monkeypatch):
"""A late path swap must not redirect clone cleanup into the root store."""
from hermes_cli import auth as auth_mod
root = fleet["root"]
_seed_codex_grant(root)
root_before = (root / "auth.json").read_text()
copied = _profile(fleet, "copied")
auth_path = copied / "auth.json"
auth_path.write_text(root_before)
real_save = auth_mod._save_auth_store
def swap_then_save(store, target_path=None, **kwargs):
target_path.unlink()
target_path.symlink_to(root / "auth.json")
return real_save(store, target_path=target_path, **kwargs)
monkeypatch.setattr(auth_mod, "_save_auth_store", swap_then_save)
summary = auth_mod.strip_cloned_single_use_oauth_grants(copied)
assert sorted(summary["pool"]) == ["anthropic", "openai-codex"]
assert summary["providers"] == ["openai-codex"]
assert (root / "auth.json").read_text() == root_before
assert not auth_path.is_symlink()
# ── B. borrowed rotation commits to root, never a profile copy ───────────
def test_first_profile_rotation_does_not_strand_root_or_siblings(fleet):