From 5e108489a01bbef64cce00d8a76a71dca38e26a8 Mon Sep 17 00:00:00 2001 From: GuardianZ71 Date: Sun, 6 Sep 2026 00:37:13 -0600 Subject: [PATCH] fix(auth): preserve shared store during clone cleanup --- hermes_cli/auth.py | 8 +- hermes_cli/auth_oauth_grants.py | 22 ++++- ...test_credential_pool_profile_oauth_fork.py | 88 +++++++++++++++++++ 3 files changed, 114 insertions(+), 4 deletions(-) diff --git a/hermes_cli/auth.py b/hermes_cli/auth.py index a3f483cdf0..91db1243ca 100644 --- a/hermes_cli/auth.py +++ b/hermes_cli/auth.py @@ -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: diff --git a/hermes_cli/auth_oauth_grants.py b/hermes_cli/auth_oauth_grants.py index 2e4dc8da66..58dabeb5c1 100644 --- a/hermes_cli/auth_oauth_grants.py +++ b/hermes_cli/auth_oauth_grants.py @@ -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) diff --git a/tests/agent/test_credential_pool_profile_oauth_fork.py b/tests/agent/test_credential_pool_profile_oauth_fork.py index eec0c6c52e..3024f04a4f 100644 --- a/tests/agent/test_credential_pool_profile_oauth_fork.py +++ b/tests/agent/test_credential_pool_profile_oauth_fork.py @@ -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):