refactor(secrets): every private-credential file is written by utils.atomic_json_write(mode=0o600)

Ten hand-rolled "write a token file safely" routines each carried a
different subset of {0600-on-create, fsync, atomic_replace, parent-0700
guard, BaseException cleanup}. Two of them (iron_proxy state files,
the exchanged-JWT store) still opened the temp file at process umask
and chmod'ed afterwards - the exact TOCTOU window the others document
as fixed. None of the bare-os.replace copies got atomic_replace's
Windows-contention retry or EXDEV fallback.

utils gains fsync_dir= (absorbs auth.py's dir fsync), atomic_write_bytes
(vault blob) and mode= on atomic_write_text; the ten sites become 1-3
line callers. mkstemp creates the temp file O_EXCL at 0600 regardless of
umask, so the payload is never umask-readable.

Behavior change: iron_proxy proxy.yaml/mappings.json and the exchanged-JWT
store are now 0600 from creation and fsync'd; every credential write goes
through atomic_replace (symlink-preserving, Windows retry, EXDEV copy).
auth_nous shared store now uses atomic_replace too (it forced os.replace
with no recorded reason). secret_sources cache parent-0700 goes through
the guarded secure_parent_dir instead of an unguarded chmod.

agent/anthropic_credentials.py

@@ -18,7 +18,6 @@ import logging
 import os
 import platform
 import secrets
-import stat
 import subprocess
 import threading
 import time
@@ -27,6 +26,7 @@ from pathlib import Path
 from typing import Any, Dict, Optional
 
 from hermes_constants import get_hermes_home
+from utils import atomic_json_write
 from agent.secret_scope import get_secret as _get_secret
 
 logger = logging.getLogger(__name__)
@@ -82,22 +82,9 @@ def _load_json_if_exists(path: Path, what: str) -> Optional[Any]:
 
 
 def _atomic_write_private_json(path: Path, payload: Any) -> None:
-    """Write *payload* via a 0o600 O_EXCL temp file + fsync + os.replace: the token is never briefly umask-readable
-    (write_text + chmod had a TOCTOU window); the random suffix avoids collisions with concurrent writers and
-    crashed leftovers. The parent dir's mode is left alone (~/.claude/ is owned by Claude Code)."""
-    path.parent.mkdir(parents=True, exist_ok=True)
-    tmp = path.with_suffix(f".tmp.{os.getpid()}.{secrets.token_hex(4)}")
-    try:
-        fd = os.open(str(tmp), os.O_WRONLY | os.O_CREAT | os.O_EXCL, stat.S_IRUSR | stat.S_IWUSR)
-        with os.fdopen(fd, "w", encoding="utf-8") as fh:
-            json.dump(payload, fh, indent=2)
-            fh.flush()
-            os.fsync(fh.fileno())
-        os.replace(tmp, path)
-    except OSError:
-        with contextlib.suppress(OSError):
-            tmp.unlink(missing_ok=True)
-        raise
+    """0600-from-creation temp file + fsync + atomic replace (the token is never briefly umask-readable).
+    The parent dir's mode is left alone (~/.claude/ is owned by Claude Code)."""
+    atomic_json_write(path, payload, mode=0o600)
 
 
 def _commit_private_json(path: Path, payload: Any, what: str) -> None:

agent/proxy_sources/iron_proxy.py

@@ -29,6 +29,8 @@ from dataclasses import dataclass, field, replace
 from pathlib import Path
 from typing import Dict, List, Optional, Tuple
 
+from utils import atomic_json_write, atomic_write_text
+
 logger = logging.getLogger(__name__)
 
 # Pinned: never auto-resolve "latest" — the YAML schema may change between releases.
@@ -577,21 +579,15 @@ def ensure_audit_log(audit_path: Path) -> None:
         ) from exc
 
 
-def _write_state_file_atomic(state: Path, name: str, dump) -> Path:
-    """0600 temp file + atomic replace: the file holds proxy tokens; chmod-after-replace would be a world-readable TOCTOU window."""
-    tmp_path = state / f".{name}.tmp"
-    with open(tmp_path, "w", encoding="utf-8") as f:
-        dump(f)
-    os.chmod(tmp_path, 0o600)
-    os.replace(tmp_path, state / name)
-    return state / name
-
-
 def write_proxy_config(config: Dict) -> Path:
-    """Serialize the config dict to ``<hermes_home>/proxy/proxy.yaml`` (safe_dump, no Python tags)."""
+    """Serialize the config dict to ``<hermes_home>/proxy/proxy.yaml`` (safe_dump, no Python tags).
+
+    The file holds proxy tokens: written 0600 from creation, never at process umask."""
     if (yaml := _yaml()) is None:
         raise RuntimeError("PyYAML is required to write the iron-proxy config but is not installed.")
-    return _write_state_file_atomic(_proxy_state_dir(), "proxy.yaml", lambda f: yaml.safe_dump(config, f, default_flow_style=False, sort_keys=False))
+    path = _proxy_state_dir() / "proxy.yaml"
+    atomic_write_text(path, yaml.safe_dump(config, default_flow_style=False, sort_keys=False), mode=0o600)
+    return path
 
 
 def write_mappings(mappings: List[TokenMapping]) -> Path:
@@ -600,7 +596,9 @@ def write_mappings(mappings: List[TokenMapping]) -> Path:
         "proxy_token": m.proxy_token, "env_name": m.real_env_name, "upstream_hosts": list(m.upstream_hosts),
         "match_headers": list(m.match_headers), "alias_env_names": list(m.alias_env_names),
     } for m in mappings]}
-    return _write_state_file_atomic(_proxy_state_dir(), "mappings.json", lambda f: json.dump(payload, f, indent=2))
+    path = _proxy_state_dir() / "mappings.json"
+    atomic_json_write(path, payload, mode=0o600)
+    return path
 
 
 def load_mappings() -> List[TokenMapping]:

agent/secret_sources/_cache.py

@@ -12,12 +12,14 @@ from __future__ import annotations
 import hashlib
 import json
 import os
-import tempfile
 import time
 from dataclasses import dataclass
 from pathlib import Path
 from typing import Callable, Dict, Generic, Optional, TypeVar
 
+from hermes_constants import secure_parent_dir
+from utils import atomic_json_write
+
 __all__ = [
     "CachedFetch",
     "DiskCache",
@@ -68,32 +70,13 @@ def entry_from_payload(payload: object) -> Optional[CachedFetch]:
     return CachedFetch(secrets=typed, fetched_at=float(fetched_at))
 
 
-def atomic_write_json(path: Path, payload: dict, *, tmp_prefix: str) -> None:
-    """Write ``payload`` to ``path`` via mkstemp → chmod 0600 → os.replace.
-
-    The containing dir is forced to ``0700`` (``mkdir``'s mode is umask-subject,
-    so the chmod is the reliable form). Raises ``OSError`` on failure; callers
-    decide whether that is best-effort.
-    """
-    cache_dir = path.parent
-    cache_dir.mkdir(parents=True, exist_ok=True)
-    try:
-        os.chmod(cache_dir, 0o700)
-    except OSError:
-        pass
-    # tempfile honours os.umask, so chmod 0600 explicitly before the rename.
-    fd, tmp = tempfile.mkstemp(prefix=tmp_prefix, suffix=".tmp", dir=str(cache_dir))
-    try:
-        with os.fdopen(fd, "w", encoding="utf-8") as f:
-            json.dump(payload, f)
-        os.chmod(tmp, 0o600)
-        os.replace(tmp, path)
-    except BaseException:
-        try:
-            os.unlink(tmp)
-        except OSError:
-            pass
-        raise
+def atomic_write_json(path: Path, payload: dict) -> None:
+    """Secret cache entry at 0600 from creation; the containing dir is tightened to 0700
+    (``secure_parent_dir`` refuses ``/``, top-level dirs and the install tree). Raises ``OSError``
+    on failure; callers decide whether that is best-effort."""
+    path.parent.mkdir(parents=True, exist_ok=True)
+    secure_parent_dir(path)
+    atomic_json_write(path, payload, indent=None, mode=0o600)
 
 
 K = TypeVar("K")
@@ -116,8 +99,6 @@ class DiskCache(Generic[K]):
     def __init__(self, basename: str, *, key_serializer: Callable[[K], str]) -> None:
         self._basename = basename
         self._key_serializer = key_serializer
-        # Per-backend temp prefix so concurrent writers in one dir never collide.
-        self._tmp_prefix = f".{basename.split('.', 1)[0]}_"
 
     def path(self, home_path: Optional[Path] = None) -> Path:
         return resolve_cache_home(home_path) / "cache" / self._basename
@@ -142,7 +123,7 @@ class DiskCache(Generic[K]):
             return
         payload = {"key": self._key_serializer(key), "secrets": entry.secrets, "fetched_at": entry.fetched_at}
         try:
-            atomic_write_json(self.path(home_path), payload, tmp_prefix=self._tmp_prefix)
+            atomic_write_json(self.path(home_path), payload)
         except OSError:
             pass  # best-effort — a disk-cache miss next invocation is fine
 

agent/secret_sources/bitwarden.py

@@ -268,7 +268,7 @@ def _write_encrypted_disk_cache(*, cache_key: _CacheKey, access_token: str, entr
         ciphertext = AESGCM(key).encrypt(nonce, plaintext, serialized_key.encode("utf-8"))
         payload = {"version": _ENCRYPTED_CACHE_VERSION, "key": serialized_key,
                    "salt": _b64e(salt), "nonce": _b64e(nonce), "ciphertext": _b64e(ciphertext)}
-        atomic_write_json(_encrypted_disk_cache_path(home_path), payload, tmp_prefix=".bws_cache_enc_")
+        atomic_write_json(_encrypted_disk_cache_path(home_path), payload)
         _STORE.disk.clear(home_path)
     except Exception:  # noqa: BLE001 — best-effort cache only
         return

agent/vault_store.py

@@ -30,6 +30,7 @@ from typing import Any, Dict, List, Optional
 from urllib.parse import urlsplit
 
 from hermes_constants import get_hermes_home
+from utils import atomic_write_bytes
 
 VAULT_KINDS = ("login", "payment", "address")
 
@@ -282,24 +283,8 @@ class VaultStore:
         self._ensure_dir()
         payload = json.dumps({"version": 1, "items": items}).encode("utf-8")
         blob = self._fernet().encrypt(payload)
-        tmp = self._vault_path.with_suffix(".enc.tmp")
-        fd = os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
-        try:
-            os.write(fd, blob)
-            os.fsync(fd)  # the blob must be on disk before the rename makes it THE vault
-        finally:
-            os.close(fd)
-        os.replace(tmp, self._vault_path)
-        with suppress(OSError):  # directory entry durable too (power loss between rename and next sync)
-            dfd = os.open(self._base, os.O_RDONLY)
-            try:
-                os.fsync(dfd)
-            finally:
-                os.close(dfd)
-        try:
-            os.chmod(self._vault_path, 0o600)
-        except OSError:
-            pass
+        # fsync_dir: the directory entry must be durable too (power loss between rename and next sync).
+        atomic_write_bytes(self._vault_path, blob, mode=0o600, fsync_dir=True)
 
     # -- public API ----------------------------------------------------------
 

gateway/pairing.py

@@ -13,7 +13,6 @@ import json
 import logging
 import os
 import secrets
-import tempfile
 import threading
 import time
 from pathlib import Path
@@ -21,7 +20,7 @@ from typing import Optional
 
 from gateway.whatsapp_identity import expand_whatsapp_aliases, normalize_whatsapp_identifier
 from hermes_constants import get_default_hermes_root, get_hermes_dir, get_hermes_home
-from utils import atomic_replace
+from utils import atomic_json_write
 
 logger = logging.getLogger(__name__)
 
@@ -279,7 +278,7 @@ def _load_json_file(path: Path) -> dict:
 
 
 def _save_json_file(path: Path, data: dict) -> None:
-    _secure_write(path, json.dumps(data, indent=2, ensure_ascii=False))
+    atomic_json_write(path, data, mode=0o600)
 
 
 def _migrate_split_pairing_dirs(*, home: Optional[Path] = None, active: Optional[Path] = None) -> None:
@@ -305,24 +304,6 @@ def _migrate_split_pairing_dirs(*, home: Optional[Path] = None, active: Optional
             _save_json_file(active / src.name, merged)
 
 
-def _secure_write(path: Path, data: str) -> None:
-    """Write 0600 via temp file + atomic rename so readers never see a partial file."""
-    path.parent.mkdir(parents=True, exist_ok=True)
-    fd, tmp_path = tempfile.mkstemp(dir=str(path.parent), suffix=".tmp")
-    try:
-        with os.fdopen(fd, "w", encoding="utf-8") as f:
-            f.write(data)
-            f.flush()
-            os.fsync(f.fileno())
-        atomic_replace(tmp_path, path)
-        with contextlib.suppress(OSError):  # Windows doesn't support chmod the same way
-            os.chmod(path, 0o600)
-    except BaseException:
-        with contextlib.suppress(OSError):
-            os.unlink(tmp_path)
-        raise
-
-
 def _is_hashed_entry(entry) -> bool:
     return isinstance(entry, dict) and "salt" in entry and "hash" in entry
 

hermes_cli/auth.py

@@ -16,10 +16,8 @@ import logging
 import os
 import shutil
 import shlex
-import stat
 import threading
 import time
-import uuid
 import webbrowser  # noqa: F401  (tests patch auth_mod.webbrowser.open; same module object)
 
 from contextlib import ExitStack, contextmanager
@@ -34,7 +32,7 @@ from hermes_cli.config import (
     get_hermes_home, get_config_path, read_raw_config, require_readable_config_before_write)
 from hermes_constants import OPENROUTER_BASE_URL, hermes_home_key, secure_parent_dir
 from agent.credential_persistence import sanitize_borrowed_credential_payload
-from utils import atomic_replace, atomic_yaml_write, env_float, is_truthy_value  # noqa: F401  (env_float: agent.credential_pool reads auth_mod.env_float)
+from utils import atomic_json_write, atomic_yaml_write, env_float, is_truthy_value  # noqa: F401  (env_float: agent.credential_pool reads auth_mod.env_float)
 from hermes_cli.auth_zai_kimi import (  # noqa: F401  re-exported
     KIMI_CODE_BASE_URL, ZAI_ENDPOINTS, _normalize_lmstudio_runtime_base_url, _resolve_kimi_base_url,
     _resolve_zai_base_url, detect_zai_endpoint)
@@ -697,61 +695,26 @@ def _load_auth_store(auth_file: Optional[Path] = None) -> Dict[str, Any]:
     return _empty_auth_store()
 
 
-def _write_private_file_atomic(
-    target: Path, payload: str, *, replace: Optional[Callable[[Any, Any], Any]] = None,
-    fsync_dir: bool = False) -> None:
-    """Write *payload* to *target* via a 0o600 temp file + atomic rename.
-
-    ``os.open(O_EXCL, 0o600)`` closes the TOCTOU window where ``write_text()`` + post-write
-    ``chmod`` briefly exposed tokens at process umask. The per-process random temp suffix avoids
-    collisions between concurrent writers and stale leftovers from a crashed prior write."""
+def _save_private_json(target: Path, data: Any, *, fsync_dir: bool = False, **dump_kwargs: Any) -> None:
+    """0600 credential JSON under a 0700 parent (``secure_parent_dir`` refuses ``/``, top-level dirs
+    and the install tree). ``atomic_json_write`` creates the temp file 0600 before any byte lands."""
     target.parent.mkdir(parents=True, exist_ok=True)
-    secure_parent_dir(target)  # refuses to chmod /, top-level dirs, or the install tree
-    tmp_path = target.with_name(f"{target.name}.tmp.{os.getpid()}.{uuid.uuid4().hex}")
-    try:
-        fd = os.open(str(tmp_path), os.O_WRONLY | os.O_CREAT | os.O_EXCL, stat.S_IRUSR | stat.S_IWUSR)
-        with os.fdopen(fd, "w", encoding="utf-8") as handle:
-            handle.write(payload)
-            handle.flush()
-            os.fsync(handle.fileno())
-        (replace or atomic_replace)(tmp_path, target)
-        if fsync_dir:
-            try:
-                dir_fd = os.open(str(target.parent), os.O_RDONLY)
-            except OSError:
-                pass
-            else:
-                try:
-                    os.fsync(dir_fd)
-                finally:
-                    os.close(dir_fd)
-    finally:
-        try:
-            if tmp_path.exists():
-                tmp_path.unlink()
-        except OSError:
-            pass
+    secure_parent_dir(target)
+    atomic_json_write(target, data, mode=0o600, fsync_dir=fsync_dir, **dump_kwargs)
 
 
 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()
-    # Tighten parent dir to 0o700 so siblings can't traverse to creds. No-op on Windows (POSIX mode bits not
-    # enforced); ignore failures. secure_parent_dir refuses to chmod /, top-level dirs, or the hermes-agent
-    # 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)
+    _save_private_json(auth_file, auth_store, fsync_dir=True)
     if target_path is not None:
         # A write-through to the global root must not be masked by the mtime memo: on coarse-mtime
         # filesystems a read-after-write in the same tick would keep serving the pre-write store.
         global _global_auth_store_cache
         _global_auth_store_cache = None
-    try:
-        auth_file.chmod(stat.S_IRUSR | stat.S_IWUSR)
-    except OSError:
-        pass
     return auth_file
 
 

hermes_cli/auth_nous.py

@@ -384,7 +384,7 @@ def _write_shared_nous_state(state: Dict[str, Any]) -> None:
 
     Best-effort: failures are logged and swallowed; per-profile auth.json stays the source of truth.
     """
-    from hermes_cli.auth import _nonempty_str, _write_private_file_atomic
+    from hermes_cli.auth import _nonempty_str, _save_private_json
     refresh_token = state.get("refresh_token")
     # Nothing worth sharing without refresh material: an OAuth refresh_token (with its access token),
     # or a guest's anon_ credential, which is the whole identity and may not have been exchanged yet.
@@ -397,8 +397,7 @@ def _write_shared_nous_state(state: Dict[str, Any]) -> None:
     try:
         with _nous_shared_store_lock():
             path = _nous_shared_store_path()
-            _write_private_file_atomic(
-                path, json.dumps(shared, indent=2, sort_keys=True), replace=os.replace)
+            _save_private_json(path, shared, sort_keys=True)
         _oauth_trace(
             "nous_shared_store_written", path=str(path),
             refresh_token_fp=_token_fingerprint(refresh_token))

hermes_cli/auth_qwen.py

@@ -43,9 +43,9 @@ def _read_qwen_cli_tokens() -> Dict[str, Any]:
 
 
 def _save_qwen_cli_tokens(tokens: Dict[str, Any]) -> Path:
-    from hermes_cli.auth import _qwen_cli_auth_path, _write_private_file_atomic
+    from hermes_cli.auth import _qwen_cli_auth_path, _save_private_json
     auth_path = _qwen_cli_auth_path()
-    _write_private_file_atomic(auth_path, json.dumps(tokens, indent=2, sort_keys=True) + "\n")
+    _save_private_json(auth_path, tokens, sort_keys=True)
     return auth_path
 
 

hermes_cli/copilot_auth.py

@@ -19,6 +19,7 @@ from pathlib import Path
 from typing import Optional
 
 from hermes_cli._subprocess_compat import IS_WINDOWS, windows_hide_flags
+from utils import atomic_json_write
 
 logger = logging.getLogger(__name__)
 
@@ -270,15 +271,6 @@ def _read_jwt_store(path: Path) -> Optional[dict]:
         return None
 
 
-def _write_jwt_store(path: Path, store: dict) -> None:
-    """Atomically write the JWT store (tmp + os.replace), best-effort 0o600."""
-    tmp = path.with_suffix(path.suffix + ".tmp")
-    tmp.write_text(json.dumps(store), encoding="utf-8")
-    with contextlib.suppress(Exception):
-        os.chmod(tmp, 0o600)
-    os.replace(tmp, path)
-
-
 def _jwt_disk_path() -> Optional[Path]:
     """Path to the on-disk exchanged-JWT cache (profile-aware), or None."""
     try:
@@ -313,7 +305,7 @@ def evict_cached_exchanged_token(raw_token: str) -> None:
     def _evict(path, store):
         if store is not None and fp in store:
             del store[fp]
-            _write_jwt_store(path, store)
+            atomic_json_write(path, store, indent=None, mode=0o600)
 
     _with_jwt_store("evict cached", _evict)
 
@@ -339,7 +331,7 @@ def _save_jwt_to_disk(fp: str, api_token: str, expires_at: float, base_url: Opti
             k: v for k, v in (store or {}).items()
             if isinstance(v, dict) and float(v.get("expires_at", 0) or 0) > now}
         kept[fp] = {"api_token": api_token, "expires_at": expires_at, "base_url": base_url}
-        _write_jwt_store(path, kept)
+        atomic_json_write(path, kept, indent=None, mode=0o600)
 
     _with_jwt_store("persist", _save)
 

plugins/platforms/photon/adapter.py

@@ -41,6 +41,7 @@ from gateway.platforms._shared import get_scoped_secret as _get_scoped_secret
 from gateway.platforms.base import BasePlatformAdapter, SendResult
 from gateway.platforms.event import MessageEvent, MessageType
 from gateway.platforms.helpers import compile_mention_patterns, strip_markdown
+from utils import atomic_json_write
 
 from .auth import load_project_credentials
 # Sidecar dir resolution is lazy (never at import): it probes the filesystem and may
@@ -89,22 +90,9 @@ def _runtime_record_path() -> Path:
 
 
 def _write_runtime_record(port: int, token: str, pid: int) -> None:
-    """Atomically persist ``{port, token, pid}`` with owner-only perms (best-effort)."""
-    import tempfile
+    """Atomically persist ``{port, token, pid}`` 0600 from creation (best-effort)."""
     try:
-        path = _runtime_record_path()
-        path.parent.mkdir(parents=True, exist_ok=True)
-        fd, tmp = tempfile.mkstemp(dir=str(path.parent), prefix=".photon-sidecar.", suffix=".tmp")
-        try:
-            with contextlib.suppress(OSError):  # perms BEFORE the token hits disk (Windows / odd fs)
-                os.chmod(tmp, 0o600)
-            with os.fdopen(fd, "w", encoding="utf-8") as fh:
-                json.dump({"port": port, "token": token, "pid": pid}, fh)
-            os.replace(tmp, path)
-        except BaseException:
-            with contextlib.suppress(OSError):
-                os.unlink(tmp)
-            raise
+        atomic_json_write(_runtime_record_path(), {"port": port, "token": token, "pid": pid}, indent=None, mode=0o600)
     except Exception as e:
         logger.warning("[photon] failed to write sidecar runtime record: %s", e)
 

scripts/docker_rebootstrap_nous_session.py

@@ -187,14 +187,22 @@ def reseed_if_terminal(auth_path: str, seed_raw: str) -> str:
     # Surgical replacement: swap ONLY providers.nous, preserve everything else.
     providers["nous"] = seed_nous
 
-    tmp_path = f"{auth_path}.rebootstrap.tmp"
-    with open(tmp_path, "w", encoding="utf-8") as fh:
-        json.dump(store, fh)
-    os.replace(tmp_path, auth_path)
+    # 0600 from creation: the seed holds a refresh token and must never sit at umask, even briefly.
+    # (stdlib only by design — see module docstring — so this mirrors utils.atomic_json_write by hand.)
+    tmp_path = f"{auth_path}.rebootstrap.{os.getpid()}.tmp"
+    fd = os.open(tmp_path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
     try:
-        os.chmod(auth_path, 0o600)
-    except OSError:
-        pass
+        with os.fdopen(fd, "w", encoding="utf-8") as fh:
+            json.dump(store, fh)
+            fh.flush()
+            os.fsync(fh.fileno())
+        os.replace(tmp_path, auth_path)
+    except BaseException:
+        try:
+            os.unlink(tmp_path)
+        except OSError:
+            pass
+        raise
     return "reseeded" if terminal else "reseeded_newer"
 
 

tests/gateway/test_pairing.py

@@ -17,7 +17,7 @@ from gateway.pairing import (
     RATE_LIMIT_SECONDS,
     MAX_PENDING_PER_PLATFORM,
     MAX_FAILED_ATTEMPTS,
-    _secure_write,
+    _save_json_file,
 )
 
 
@@ -82,11 +82,11 @@ class TestProfileScopedDiscovery:
 
 
 # ---------------------------------------------------------------------------
-# _secure_write
+# _save_json_file
 # ---------------------------------------------------------------------------
 
 
-class TestSecureWrite:
+class TestSaveJsonFile:
 
     @pytest.mark.skipif(
         sys.platform.startswith("win"),
@@ -94,7 +94,7 @@ class TestSecureWrite:
     )
     def test_sets_file_permissions(self, tmp_path):
         target = tmp_path / "secret.json"
-        _secure_write(target, "data")
+        _save_json_file(target, {"data": 1})
         mode = oct(target.stat().st_mode & 0o777)
         assert mode == "0o600"
 

tests/hermes_cli/test_auth_toctou_file_modes.py

@@ -6,10 +6,9 @@ The three writers below used to create a temp file via ``Path.write_text`` /
 ``Path.open('w')`` and only ``chmod``'d it to ``0o600`` afterward. Between
 create and chmod the file existed at the process umask (typically ``0o644``),
 briefly exposing OAuth tokens to other local users on multi-user hosts. The
-fix switches them to ``os.open(O_EXCL, mode=0o600)`` + ``os.fdopen`` +
-``fsync`` so the file is atomic at ``0o600`` on creation. Mirrors the fixes
-shipped for ``agent/google_oauth.py`` (#19673) and ``tools/mcp_oauth.py``
-(#21148).
+writers now go through ``utils.atomic_json_write(mode=0o600)`` whose mkstemp temp
+file is ``O_EXCL`` at 0600 on creation (the cross-writer invariant lives in
+``tests/test_private_credential_writers.py``).
 
 These tests stay green only while the token file and its parent directory
 end up at ``0o600`` / ``0o700`` after every write. POSIX-only — the mode-bit
@@ -22,7 +21,6 @@ import json
 import os
 import stat
 import sys
-from unittest.mock import patch
 
 import pytest
 
@@ -153,50 +151,3 @@ def test_shared_nous_store_writes_0o600_with_0o700_parent(tmp_path, monkeypatch)
 
     data = json.loads(path.read_text())
     assert data["refresh_token"] == "nous-refresh-xxx"
-
-
-# ---------------------------------------------------------------------------
-# Atomicity: verify ``os.open`` is called with an explicit 0o600 mode.
-# ---------------------------------------------------------------------------
-
-
-def test_save_auth_store_uses_os_open_with_0o600_mode(tmp_path, monkeypatch):
-    """Regression: the writer must call ``os.open`` with an explicit restricted
-    mode so the file is created at 0o600 atomically — closing the TOCTOU
-    window the previous ``Path.open('w')`` left open (fd inherited process
-    umask and was briefly 0o644 before post-write chmod)."""
-    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
-
-    observed_opens: list[tuple[str, int, int]] = []
-    real_os_open = os.open
-
-    def spying_os_open(path, flags, mode=0o777, *args, **kwargs):
-        observed_opens.append((str(path), flags, mode))
-        return real_os_open(path, flags, mode, *args, **kwargs)
-
-    with patch.object(os, "open", spying_os_open):
-        from hermes_cli import auth as auth_mod
-
-        auth_mod._save_auth_store(
-            {"version": auth_mod.AUTH_STORE_VERSION, "providers": {}}
-        )
-
-    auth_tmp_opens = [
-        (p, fl, m) for (p, fl, m) in observed_opens if "auth.json.tmp" in p
-    ]
-    assert auth_tmp_opens, (
-        f"os.open was never called for the auth.json temp file; "
-        f"observed={observed_opens!r}"
-    )
-    for path, flags, mode in auth_tmp_opens:
-        assert flags & os.O_CREAT, f"auth.json temp open missing O_CREAT: path={path}"
-        assert flags & os.O_EXCL, (
-            f"auth.json temp open missing O_EXCL — TOCTOU-safe pattern regressed: "
-            f"path={path}, flags={flags}"
-        )
-        # Must be exactly S_IRUSR | S_IWUSR (0o600) — no group/other bits.
-        expected = stat.S_IRUSR | stat.S_IWUSR
-        assert mode == expected, (
-            f"auth.json temp open mode 0o{mode:o} != 0o{expected:o} — "
-            f"umask would apply and potentially expose tokens"
-        )

tests/test_private_credential_writers.py

@@ -0,0 +1,113 @@
+"""Cross-writer invariant: every private-credential file is 0600 from the moment its temp file exists.
+
+The credential writers (auth.json, MCP OAuth tokens, secret-source cache, iron-proxy state, the
+exchanged-JWT store, the Photon sidecar record, pairing data, the vault blob, the third-party
+credential file) all funnel through ``utils.atomic_json_write`` / ``atomic_write_text`` /
+``atomic_write_bytes`` with ``mode=0o600``. The contract under test: the *temp* file is created
+with mode 0600 (``O_EXCL``) BEFORE any byte lands and the final file carries 0600 — never
+"open at umask, then chmod" (the #19673 window). POSIX-only: mode bits are not enforced on Windows.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import stat
+import sys
+from pathlib import Path
+
+import pytest
+
+pytestmark = pytest.mark.skipif(sys.platform.startswith("win"), reason="POSIX mode bits not enforced on Windows")
+
+_PRIVATE = stat.S_IRUSR | stat.S_IWUSR
+
+
+@pytest.fixture
+def opens_spy(monkeypatch):
+    """Record every ``os.open`` create (path, flags, mode) issued through ``utils``."""
+    import utils
+
+    observed: list[tuple[str, int, int]] = []
+    real_open = os.open
+
+    def spying(path, flags, mode=0o777, *args, **kwargs):
+        if flags & os.O_CREAT:
+            observed.append((os.fspath(path), flags, mode))
+        return real_open(path, flags, mode, *args, **kwargs)
+
+    # mkstemp resolves ``os.open`` at call time from the ``tempfile`` module namespace.
+    monkeypatch.setattr(utils.tempfile._os, "open", spying)
+    return observed
+
+
+def _writers(home: Path, monkeypatch):
+    """``(label, callable, target_path)`` for every private-credential writer."""
+    from agent import anthropic_credentials
+    from agent.secret_sources._cache import CachedFetch, DiskCache
+    from agent.proxy_sources import iron_proxy
+    from agent.vault_store import VaultStore
+    from gateway import pairing
+    from hermes_cli import auth as auth_mod, copilot_auth
+    from tools import mcp_oauth
+    from plugins.platforms.photon import adapter as photon_adapter
+
+    photon_record = home / "runtime" / "photon.json"
+    monkeypatch.setattr(photon_adapter, "_runtime_record_path", lambda: photon_record)
+    cache = DiskCache("probe.json", key_serializer=str)
+    vault = VaultStore(home / "vault")
+    return [
+        ("auth.json", lambda: auth_mod._save_auth_store({"version": auth_mod.AUTH_STORE_VERSION, "providers": {}}),
+         auth_mod._auth_file_path()),
+        ("third-party credentials", lambda: anthropic_credentials._atomic_write_private_json(
+            home / "cc" / ".credentials.json", {"tok": 1}), home / "cc" / ".credentials.json"),
+        ("mcp oauth tokens", lambda: mcp_oauth._write_json(home / "mcp" / "probe.tokens.json", {"access_token": "x"}),
+         home / "mcp" / "probe.tokens.json"),
+        ("secret-source cache", lambda: cache.write("k", CachedFetch(secrets={"A": "b"}, fetched_at=1.0), 60, home),
+         cache.path(home)),
+        ("iron-proxy mappings", lambda: iron_proxy.write_mappings([]), home / "proxy" / "mappings.json"),
+        ("exchanged JWT store", lambda: copilot_auth._save_jwt_to_disk("fp", "jwt", 9e12, None),
+         copilot_auth._jwt_disk_path()),
+        ("photon sidecar record", lambda: photon_adapter._write_runtime_record(1, "tok", 2), photon_record),
+        ("pairing", lambda: pairing._save_json_file(home / "pairing" / "p.json", {"a": 1}), home / "pairing" / "p.json"),
+        ("vault blob", lambda: vault._write_all([]), vault._vault_path),
+    ]
+
+
+def test_every_credential_writer_creates_its_temp_file_at_0600(tmp_path, monkeypatch, opens_spy):
+    pytest.importorskip("cryptography")
+    home = tmp_path / "home"
+    home.mkdir()
+    monkeypatch.setenv("HERMES_HOME", str(home))
+    old_umask = os.umask(0o022)  # a "write then chmod" regression would surface as 0o644
+    try:
+        for label, write, target in _writers(home, monkeypatch):
+            del opens_spy[:]
+            write()
+            assert target.exists(), f"{label}: nothing written at {target}"
+            assert stat.S_IMODE(target.stat().st_mode) == _PRIVATE, f"{label}: final file not 0600"
+            creates = [(p, m) for p, fl, m in opens_spy if Path(p).parent == target.parent]
+            assert creates, f"{label}: no temp file created in {target.parent}; opens={opens_spy!r}"
+            for path, mode in creates:
+                assert mode == _PRIVATE, f"{label}: temp file {path} created 0o{mode:o}, not 0600"
+    finally:
+        os.umask(old_umask)
+
+
+def test_canonical_private_writers_round_trip_json_text_and_bytes(tmp_path):
+    from utils import atomic_json_write, atomic_write_bytes, atomic_write_text
+
+    target = tmp_path / "nested" / "creds.json"
+    payload = {"token": "sk-\u00e9\u2603", "n": [1, 2]}
+    atomic_json_write(target, payload, mode=0o600, fsync_dir=True)
+    assert json.loads(target.read_text(encoding="utf-8")) == payload
+    assert stat.S_IMODE(target.stat().st_mode) == 0o600
+
+    atomic_write_text(target, "plain\n", mode=0o600)
+    assert target.read_text(encoding="utf-8") == "plain\n"
+
+    blob = bytes(range(256)) * 3
+    atomic_write_bytes(target, blob, mode=0o600, fsync_dir=True)
+    assert target.read_bytes() == blob
+    assert stat.S_IMODE(target.stat().st_mode) == 0o600
+    assert [p.name for p in target.parent.iterdir()] == ["creds.json"], "temp files must not survive"

tools/mcp_oauth.py

@@ -16,7 +16,6 @@ import json
 import logging
 import os
 import re
-import secrets
 import socket
 import stat
 import sys
@@ -30,6 +29,7 @@ from typing import TYPE_CHECKING, Any
 from urllib.parse import parse_qs, urlparse
 
 from hermes_constants import secure_parent_dir
+from utils import atomic_json_write
 from tools.mcp_dashboard_oauth import contextvar_set as _contextvar_set, get_dashboard_oauth_flow
 
 if TYPE_CHECKING:  # annotations only; the SDK is imported lazily at runtime
@@ -236,33 +236,11 @@ def _read_json(path: Path) -> dict | None:
 
 
 def _write_json(path: Path, data: dict) -> None:
-    """Atomically write *data* as JSON created at 0o600 (``O_EXCL`` + mode avoids the write-then-chmod
-    window where the file inherits a world-readable umask); parent dir tightened to 0o700. The random
-    per-process tmp suffix avoids clashes with concurrent writers/crash leftovers.
-
-    The previous ``write_text`` + post-write ``chmod`` opened a TOCTOU window where the temp file briefly
-    inherited the process umask (commonly 0o644 = world-readable), exposing OAuth tokens to other local
-    users between create and chmod. Mirrors the fix in ``agent/google_oauth.py`` (#19673).
-    """
+    """OAuth tokens/client info at 0600 from creation, parent tightened to 0700 (``secure_parent_dir``
+    refuses ``/``, top-level dirs and the install tree — #25821, #93050)."""
     path.parent.mkdir(parents=True, exist_ok=True)
-    # secure_parent_dir refuses to chmod /, top-level dirs, or the hermes-agent install tree (#25821,
-    # #93050).
-    # Tighten parent dir to 0o700 so siblings can't traverse to the creds. No-op on Windows (POSIX mode bits
-    # aren't enforced); ignore failures. secure_parent_dir refuses to chmod /, top-level dirs, or the
-    # hermes-agent install tree (#25821, #93050).
     secure_parent_dir(path)
-    tmp = path.with_suffix(f".tmp.{os.getpid()}.{secrets.token_hex(4)}")
-    try:
-        fd = os.open(str(tmp), os.O_WRONLY | os.O_CREAT | os.O_EXCL, stat.S_IRUSR | stat.S_IWUSR)
-        with os.fdopen(fd, "w", encoding="utf-8") as fh:
-            json.dump(data, fh, indent=2, default=str)
-            fh.flush()
-            os.fsync(fh.fileno())
-        os.replace(tmp, path)
-    except OSError:
-        with contextlib.suppress(OSError):
-            tmp.unlink(missing_ok=True)
-        raise
+    atomic_json_write(path, data, mode=0o600, default=str)
 
 
 def _model_json(model: Any) -> dict:

utils.py

@@ -174,25 +174,51 @@ def atomic_replace(tmp_path: Union[str, Path], target: Union[str, Path]) -> str:
     return real_path
 
 
-def _atomic_write(path: Path, write, *, prefix: str, encoding: str = "utf-8", mode: "int | None" = None, preserve_owner: bool = True) -> None:
+def fsync_directory(path: Union[str, Path]) -> None:
+    """Best-effort fsync of a directory entry so a just-renamed file survives power loss.
+
+    No-op on Windows (directories can't be opened with ``os.open``; the file fsync still applies)
+    and on any OSError — durability of the directory entry is never worth failing a write that
+    has already been replaced into place.
+    """
+    if os.name == "nt":
+        return
+    try:
+        fd = os.open(path, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0))
+    except OSError:
+        return
+    try:
+        with suppress(OSError):
+            os.fsync(fd)
+    finally:
+        os.close(fd)
+
+
+def _atomic_write(path: Path, write, *, prefix: str, encoding: str = "utf-8", mode: "int | None" = None,
+                  preserve_owner: bool = True, binary: bool = False, fsync_dir: bool = False) -> None:
     """Temp file + fsync + :func:`atomic_replace`, then re-apply owner/mode.
 
-    *write(f)* emits the payload into the open text handle. *mode* is fchmod'd onto the temp fd
-    BEFORE the replace so the target never transits through mkstemp's 0600 (fchmod is Unix-only;
-    the post-replace chmod is the sole path on Windows). The temp file is removed on any failure —
+    *write(f)* emits the payload into the open handle (text, or bytes when *binary*). The temp file
+    is created by ``mkstemp`` — ``O_CREAT|O_EXCL`` at 0600 regardless of umask — so a secret is
+    never readable at process umask, not even between create and chmod. *mode* is fchmod'd onto
+    the temp fd BEFORE the replace so the target never transits through mkstemp's 0600 (fchmod is
+    Unix-only; the post-replace chmod is the sole path on Windows). *fsync_dir* also fsyncs the
+    parent so the rename itself is durable. The temp file is removed on any failure —
     ``BaseException`` on purpose, so KeyboardInterrupt / SystemExit still clean up.
     """
     path.parent.mkdir(parents=True, exist_ok=True)
     original_owner = _preserve_file_owner(path) if preserve_owner else None
     fd, tmp_path = tempfile.mkstemp(dir=str(path.parent), prefix=prefix, suffix=".tmp")
     try:
-        with os.fdopen(fd, "w", encoding=encoding) as f:
+        with os.fdopen(fd, "wb" if binary else "w", encoding=None if binary else encoding) as f:
             if mode is not None and hasattr(os, "fchmod"):
                 os.fchmod(f.fileno(), mode)
             write(f)
             f.flush()
             os.fsync(f.fileno())
         _restore_file_metadata(Path(atomic_replace(tmp_path, path)), original_owner, mode)  # symlink-preserving
+        if fsync_dir:
+            fsync_directory(path.parent)
     except BaseException:
         with suppress(OSError):
             os.unlink(tmp_path)
@@ -206,29 +232,44 @@ def _mode_for_write(path: Path, create_mode: "int | None", preserve: bool = True
 
 
 def atomic_write_text(path: Union[str, Path], content: str, *, encoding: str = "utf-8", tmp_prefix: str = ".tmp_",
-                      preserve_mode: bool = False, create_mode: "int | None" = None) -> None:
+                      preserve_mode: bool = False, create_mode: "int | None" = None, mode: "int | None" = None,
+                      fsync_dir: bool = False) -> None:
     """Write *content* to *path* via temp file + fsync + atomic rename.
 
     The target is never left partially written on crash/interrupt. Shared by every destructive
-    file rewrite (memory store, skill manager, agent importer, ...).
+    file rewrite (memory store, skill manager, agent importer, ...). *mode* forces the final
+    permission bits (secret files: ``0o600``) regardless of what exists; *create_mode* applies only
+    when the target is new and *preserve_mode* carries an existing file's bits and owner across.
     """
     path = Path(path)
     _atomic_write(path, lambda f: f.write(content), prefix=tmp_prefix, encoding=encoding,
-                  mode=_mode_for_write(path, create_mode, preserve=preserve_mode), preserve_owner=preserve_mode)
+                  mode=mode if mode is not None else _mode_for_write(path, create_mode, preserve=preserve_mode),
+                  preserve_owner=preserve_mode, fsync_dir=fsync_dir)
+
+
+def atomic_write_bytes(path: Union[str, Path], content: bytes, *, tmp_prefix: str = ".tmp_",
+                       mode: "int | None" = None, fsync_dir: bool = False) -> None:
+    """Bytes variant of :func:`atomic_write_text` (encrypted blobs, key material)."""
+    path = Path(path)
+    _atomic_write(path, lambda f: f.write(content), prefix=tmp_prefix, binary=True, preserve_owner=False,
+                  mode=mode if mode is not None else _preserve_file_mode(path), fsync_dir=fsync_dir)
 
 
 def atomic_json_write(
     path: Union[str, Path], data: Any, *, indent: int = 2, mode: int | None = None,
-    ensure_ascii: bool = False, **dump_kwargs: Any,
+    ensure_ascii: bool = False, fsync_dir: bool = False, **dump_kwargs: Any,
 ) -> None:
     """Write JSON to *path* atomically (temp file + fsync + replace).
 
     ``ensure_ascii=True`` lets callers persist surrogate-escaped strings (non-UTF-8 argv/paths)
-    that a utf-8 text handle would otherwise reject with ``UnicodeEncodeError``.
+    that a utf-8 text handle would otherwise reject with ``UnicodeEncodeError``. ``mode=0o600``
+    is the private-credential form: the temp file is 0600 from creation (mkstemp), so the payload
+    is never umask-readable.
     """
     path = Path(path)
     _atomic_write(path, lambda f: json.dump(data, f, indent=indent, ensure_ascii=ensure_ascii, **dump_kwargs),
-                  prefix=f".{path.stem}_", mode=mode if mode is not None else _preserve_file_mode(path))
+                  prefix=f".{path.stem}_", mode=mode if mode is not None else _preserve_file_mode(path),
+                  fsync_dir=fsync_dir)
 
 
 def warn_if_credential_file_broadly_readable(path: Union[str, Path], *, label: str = "", log: logging.Logger | None = None) -> bool: