security(photon): create auth.json temp file with 0o600 atomically

plugins/platforms/photon/auth.py

@@ -40,7 +40,9 @@
 import logging
 import os
 import re
+import stat
 import time
+import uuid
 from base64 import b64encode
 from dataclasses import dataclass
 from pathlib import Path
@@ -109,14 +111,46 @@ def _load_auth() -> Dict[str, Any]:
 def _save_auth(data: Dict[str, Any]) -> None:
     path = _auth_json_path()
     path.parent.mkdir(parents=True, exist_ok=True)
-    tmp = path.with_suffix(".json.tmp")
-    with tmp.open("w", encoding="utf-8") as fh:
-        json.dump(data, fh, indent=2, sort_keys=True)
+    # Per-process random temp suffix avoids collisions between concurrent
+    # writers and stale leftovers from a crashed prior write.
+    tmp = path.with_name(f"{path.name}.tmp.{os.getpid()}.{uuid.uuid4().hex}")
+    # Create with 0o600 atomically via os.open(O_EXCL) + fdopen: the old
+    # open() → write → chmod() sequence left a window where the bearer
+    # token sat world-readable at process umask (typically 0o644), and the
+    # predictable temp name could be pre-planted (symlink attack). Mirrors
+    # hermes_cli/auth.py:_save_auth_store (#19673, #21148).
+    fd = os.open(
+        str(tmp),
+        os.O_WRONLY | os.O_CREAT | os.O_EXCL,
+        stat.S_IRUSR | stat.S_IWUSR,
+    )
     try:
-        os.chmod(tmp, 0o600)
-    except OSError:
-        pass
-    tmp.replace(path)
+        fh = os.fdopen(fd, "w", encoding="utf-8")
+    except BaseException:
+        # os.fdopen() failed before taking ownership of the raw descriptor,
+        # so nothing else will ever close it — do it here, then drop the
+        # just-created temp file.
+        try:
+            os.close(fd)
+        except OSError:
+            pass
+        try:
+            tmp.unlink()
+        except OSError:
+            pass
+        raise
+    try:
+        with fh:
+            json.dump(data, fh, indent=2, sort_keys=True)
+            fh.flush()
+            os.fsync(fh.fileno())
+        tmp.replace(path)
+    except BaseException:
+        try:
+            tmp.unlink()
+        except OSError:
+            pass
+        raise
 
 
 def load_photon_token() -> Optional[str]:

tests/plugins/platforms/photon/test_auth.py

@@ -3,9 +3,11 @@
 
 import json
 import os
+import stat
 from base64 import b64encode
 from pathlib import Path
 from typing import Any, Dict
+from unittest import mock
 
 import pytest
 
@@ -71,6 +73,92 @@ def test_store_and_load_photon_token(tmp_hermes_home: Path) -> None:
     assert auth_json["credential_pool"]["photon"][0]["access_token"] == "abc123def456"
 
 
+@pytest.mark.skipif(os.name != "posix", reason="POSIX mode bits only")
+def test_save_auth_never_world_readable(tmp_hermes_home: Path) -> None:
+    """auth.json must be created 0o600 — no window at process umask."""
+    photon_auth.store_photon_token("secret-token")
+    mode = (tmp_hermes_home / "auth.json").stat().st_mode & 0o777
+    assert mode == 0o600
+
+
+def test_save_auth_leaves_no_temp_files(tmp_hermes_home: Path) -> None:
+    photon_auth.store_photon_token("secret-token")
+    leftovers = [p.name for p in tmp_hermes_home.iterdir() if p.name != "auth.json"]
+    assert leftovers == []
+
+
+def test_save_auth_uses_os_open_with_0o600_mode(tmp_hermes_home: Path) -> None:
+    """Regression: the writer must call ``os.open`` with O_CREAT | O_EXCL and
+    an explicit 0o600 mode so the temp file is created restricted atomically.
+    The final-mode check alone would also pass under the old open() → write →
+    chmod() writer, so this spy protects the atomic-create guarantee itself."""
+    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 mock.patch.object(os, "open", spying_os_open):
+        photon_auth.store_photon_token("secret-token")
+
+    tmp_opens = [
+        (p, fl, m) for (p, fl, m) in observed_opens if "auth.json.tmp" in p
+    ]
+    assert tmp_opens, (
+        f"os.open was never called for the auth.json temp file; "
+        f"observed={observed_opens!r}"
+    )
+    for path, flags, mode in tmp_opens:
+        assert flags & os.O_CREAT, f"temp open missing O_CREAT: path={path}"
+        assert flags & os.O_EXCL, (
+            f"temp open missing O_EXCL — TOCTOU-safe pattern regressed: "
+            f"path={path}, flags={flags}"
+        )
+        expected = stat.S_IRUSR | stat.S_IWUSR
+        assert mode == expected, (
+            f"temp open mode 0o{mode:o} != 0o{expected:o} — "
+            f"umask would apply and potentially expose tokens"
+        )
+
+
+def test_save_auth_closes_raw_fd_when_fdopen_fails(tmp_hermes_home: Path) -> None:
+    """If ``os.fdopen`` raises before taking ownership of the raw descriptor,
+    the writer must close the fd itself (and still remove the temp file)."""
+    opened_fds: list[int] = []
+    closed_fds: list[int] = []
+    real_os_open = os.open
+    real_os_close = os.close
+
+    def spying_os_open(path, flags, mode=0o777, *args, **kwargs):
+        fd = real_os_open(path, flags, mode, *args, **kwargs)
+        if "auth.json.tmp" in str(path):
+            opened_fds.append(fd)
+        return fd
+
+    def spying_os_close(fd):
+        closed_fds.append(fd)
+        return real_os_close(fd)
+
+    def failing_fdopen(*args, **kwargs):
+        raise MemoryError("forced fdopen failure")
+
+    with mock.patch.object(os, "open", spying_os_open), \
+            mock.patch.object(os, "close", spying_os_close), \
+            mock.patch.object(os, "fdopen", failing_fdopen):
+        with pytest.raises(MemoryError):
+            photon_auth.store_photon_token("secret-token")
+
+    assert opened_fds, "os.open was never called for the auth.json temp file"
+    for fd in opened_fds:
+        assert fd in closed_fds, (
+            f"raw fd {fd} leaked after forced os.fdopen failure; "
+            f"closed={closed_fds!r}"
+        )
+    leftovers = [p.name for p in tmp_hermes_home.iterdir() if p.name != "auth.json"]
+    assert leftovers == [], f"temp file leaked after fdopen failure: {leftovers}"
+
+
 def test_store_project_credentials_round_trip(
     tmp_hermes_home: Path, monkeypatch: pytest.MonkeyPatch,
 ) -> None: