refactor(auth): positive store-identity check for the clone strip; drop the save kwarg

The alias check re-implemented most of _is_same_auth_store (#101356) inline, but
that helper answers False on a samefile() OSError — on the strip path that would
fail OPEN. Resolve identity positively (same resolved path, or samefile says so)
and treat any error as "refuse"; three lines instead of the inline stat dance.
The `preserve_symlinks=False` save path (raw os.replace) bypassed atomic_replace's
EXDEV/Windows fallback to defend against a path swap no concurrent writer performs
on a directory this process just created — removed with its mock-injected test.
Tests kept: shared-store invariant [symlink|hardlink] and two fail-closed variants.
This commit is contained in:
kshitijk4poor
2026-09-07 00:17:45 +05:30
committed by kshitij
parent fee38b03cd
commit 4f6ef8806f
3 changed files with 10 additions and 47 deletions

View File

@@ -702,9 +702,7 @@ def _write_private_file_atomic(
pass
def _save_auth_store(
auth_store: Dict[str, Any], target_path: Optional[Path] = None, *,
preserve_symlinks: bool = True) -> Path:
def _save_auth_store(auth_store: Dict[str, Any], target_path: Optional[Path] = None) -> 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()
@@ -713,9 +711,7 @@ def _save_auth_store(
# 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",
replace=None if preserve_symlinks else os.replace, fsync_dir=True)
_write_private_file_atomic(auth_file, json.dumps(auth_store, indent=2) + "\n", fsync_dir=True)
try:
auth_file.chmod(stat.S_IRUSR | stat.S_IWUSR)
except OSError:

View File

@@ -70,21 +70,18 @@ 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
# A profile auth.json that IS the shared root store (symlink / hardlink) is not a clone;
# stripping it would delete every profile's single-use grants. _is_same_auth_store swallows
# a samefile() OSError as "two stores", which here would fail OPEN — so resolve identity
# positively: same path, or samefile() says so; any error refuses.
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):
if _same_path(auth_path, root_auth_path) or (
root_auth_path.exists() and auth_path.samefile(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"))
@@ -122,9 +119,7 @@ def strip_cloned_single_use_oauth_grants(profile_dir: Path) -> Dict[str, Any]:
if not changed:
return stripped
try:
# 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)
_save_auth_store(store, target_path=auth_path)
except Exception:
logger.debug(
"Failed to strip cloned single-use OAuth grants from %s", auth_path, exc_info=True)

View File

@@ -222,34 +222,6 @@ def test_strip_helper_fails_closed_when_store_identity_check_errors(fleet, monke
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):
from agent.credential_pool import load_pool