fix(security): create credential files atomically with 0o600 (#350)

Routes the two credential-writing call sites (xfetch session sync, no-Docker XHS cookie fallback) through the same atomic os.open(..., 0o600) pattern Config.save() already uses, closing a TOCTOU window where credential files were briefly world-readable (CWE-732). Also hardens _sync_bird_env() with shlex.quote against a shell-injection breakout in the sourceable env file — verified exploitable on the pre-fix code. Maintainer follow-up: scoped the injection-probe test markers to tmp_path so they can't poison reruns. 107 tests pass.

agent_reach/cli.py

@@ -1146,11 +1146,28 @@ def _configure_xhs_cookies(value):
     # Find the container
     docker = shutil.which("docker")
     if not docker:
-        # No Docker - write to a local file for manual import
+        # No Docker - write to a local file for manual import.
+        # Create with 0o600 atomically so the file is never world-readable
+        # between open() and a follow-up chmod() (same pattern Config.save()
+        # uses in config.py).
+        import stat
         cookie_path = os.path.expanduser("~/.agent-reach/xhs-cookies.json")
-        with open(cookie_path, "w") as f:
+        try:
-            f.write(cookies_json)
+            fd = os.open(
-        os.chmod(cookie_path, 0o600)
+                cookie_path,
+                os.O_WRONLY | os.O_CREAT | os.O_TRUNC,
+                stat.S_IRUSR | stat.S_IWUSR,  # 0o600
+            )
+            with os.fdopen(fd, "w", encoding="utf-8") as f:
+                f.write(cookies_json)
+        except OSError:
+            # Windows / unsupported flags — fall back to plain open + chmod.
+            with open(cookie_path, "w", encoding="utf-8") as f:
+                f.write(cookies_json)
+            try:
+                os.chmod(cookie_path, 0o600)
+            except OSError:
+                pass
         print(f"  Cookies saved to {cookie_path}")
         print("  Docker not found. Copy manually:")
         print(f"  docker cp {cookie_path} xiaohongshu-mcp:/app/data/cookies.json")

agent_reach/cookie_extract.py

@@ -148,6 +148,28 @@ def extract_all(browser: str = "chrome") -> Dict[str, dict]:
     return results
 
 
+def _open_owner_only(path: str):
+    """Open *path* for writing, atomically creating it with mode 0o600.
+
+    Mirrors the pattern used by Config.save() in config.py: O_WRONLY|O_CREAT|
+    O_TRUNC + an explicit mode argument so the file is never briefly
+    world-readable between open() and a later os.chmod(). On Windows (or any
+    OS that rejects the open flags) we fall back to a plain open().
+    """
+    import os
+    import stat
+
+    try:
+        fd = os.open(
+            path,
+            os.O_WRONLY | os.O_CREAT | os.O_TRUNC,
+            stat.S_IRUSR | stat.S_IWUSR,  # 0o600
+        )
+        return os.fdopen(fd, "w", encoding="utf-8")
+    except OSError:
+        return open(path, "w", encoding="utf-8")
+
+
 def _sync_xfetch_session(auth_token: str, ct0: str) -> None:
     """Sync Twitter credentials to ~/.config/xfetch/session.json (legacy xreach compat)."""
     import json
@@ -166,9 +188,8 @@ def _sync_xfetch_session(auth_token: str, ct0: str) -> None:
                 session_data = {}
         session_data["authToken"] = auth_token
         session_data["ct0"] = ct0
-        with open(session_path, "w", encoding="utf-8") as sf:
+        with _open_owner_only(session_path) as sf:
             json.dump(session_data, sf, indent=2)
-        os.chmod(session_path, 0o600)
     except Exception:
         # Non-fatal: agent-reach config is the source of truth, xfetch sync is best-effort
         pass
@@ -179,17 +200,19 @@ def _sync_bird_env(auth_token: str, ct0: str) -> None:
 
     bird reads AUTH_TOKEN and CT0 from environment variables. This writes a
     shell-sourceable file so users can `source ~/.config/bird/credentials.env`.
+    Values are passed through shlex.quote so a token containing a quote, $, or
+    backtick cannot break out into shell syntax when the file is sourced.
     """
     import os
+    import shlex
 
     try:
         bird_dir = os.path.join(os.path.expanduser("~"), ".config", "bird")
         os.makedirs(bird_dir, exist_ok=True)
         env_path = os.path.join(bird_dir, "credentials.env")
-        with open(env_path, "w", encoding="utf-8") as f:
+        with _open_owner_only(env_path) as f:
-            f.write(f'AUTH_TOKEN="{auth_token}"\n')
+            f.write(f"AUTH_TOKEN={shlex.quote(auth_token)}\n")
-            f.write(f'CT0="{ct0}"\n')
+            f.write(f"CT0={shlex.quote(ct0)}\n")
-        os.chmod(env_path, 0o600)
     except Exception:
         # Non-fatal: agent-reach config is the source of truth, bird env sync is best-effort
         pass

tests/test_cookie_extract_perms.py

@@ -0,0 +1,89 @@
+# -*- coding: utf-8 -*-
+"""Verify credential files written by cookie sync helpers and CLI helpers
+are owner-only (0o600) and that values containing shell metacharacters do
+not break the shell-sourceable env file produced by _sync_bird_env().
+
+Companion to tests/test_config.py::test_save_creates_file_with_restricted_permissions —
+the same threat-model claim ("Cookie/Token only stored locally, 600
+permissions") covers these paths.
+"""
+
+import json
+import os
+import stat
+import subprocess
+import sys
+import tempfile
+
+import pytest
+
+from agent_reach.cookie_extract import _sync_bird_env, _sync_xfetch_session
+
+
+def _owner_only(path: str) -> bool:
+    mode = os.stat(path).st_mode
+    return not (mode & (stat.S_IRGRP | stat.S_IWGRP | stat.S_IROTH | stat.S_IWOTH))
+
+
+@pytest.mark.skipif(sys.platform == "win32", reason="POSIX perm semantics only")
+def test_sync_xfetch_session_writes_0600(tmp_path, monkeypatch):
+    monkeypatch.setenv("HOME", str(tmp_path))
+    _sync_xfetch_session("auth_xxx", "ct0_yyy")
+    session_path = tmp_path / ".config" / "xfetch" / "session.json"
+    assert session_path.exists(), "expected ~/.config/xfetch/session.json"
+    assert _owner_only(str(session_path)), "session.json must be 0o600"
+    # Round-trip the content so we know we didn't accidentally corrupt JSON.
+    data = json.loads(session_path.read_text(encoding="utf-8"))
+    assert data["authToken"] == "auth_xxx"
+    assert data["ct0"] == "ct0_yyy"
+
+
+@pytest.mark.skipif(sys.platform == "win32", reason="POSIX perm semantics only")
+def test_sync_bird_env_writes_0600(tmp_path, monkeypatch):
+    monkeypatch.setenv("HOME", str(tmp_path))
+    _sync_bird_env("auth_xxx", "ct0_yyy")
+    env_path = tmp_path / ".config" / "bird" / "credentials.env"
+    assert env_path.exists(), "expected ~/.config/bird/credentials.env"
+    assert _owner_only(str(env_path)), "credentials.env must be 0o600"
+
+
+@pytest.mark.skipif(sys.platform == "win32", reason="POSIX sh needed for sourcing")
+def test_sync_bird_env_quotes_shell_metachars(tmp_path, monkeypatch):
+    """Tokens containing ", $, `, ; etc. must not break out of the assignment.
+
+    Prior implementation used `f'AUTH_TOKEN="{auth_token}"'` which an attacker-
+    controlled cookie containing a literal `"` could break out of, turning a
+    later `source ~/.config/bird/credentials.env` into arbitrary shell.
+    """
+    monkeypatch.setenv("HOME", str(tmp_path))
+    # Side-effect markers live under tmp_path (auto-cleaned by pytest) rather
+    # than a shared absolute /tmp path — otherwise one vulnerable run leaves a
+    # marker behind that fails every later run on the same machine/CI runner.
+    pwn_auth = tmp_path / "pwn-auth"
+    pwn_ct0 = tmp_path / "pwn-ct0"
+    hostile_auth = f'inj"; touch {pwn_auth}; #'
+    hostile_ct0 = f"ct0_$(touch {pwn_ct0})"
+    _sync_bird_env(hostile_auth, hostile_ct0)
+    env_path = tmp_path / ".config" / "bird" / "credentials.env"
+
+    # Sourcing the file must NOT execute the injected payload. Read back the
+    # exported values from a subshell instead — they should equal the originals.
+    probe = (
+        f". {env_path}; "
+        f'printf "AUTH=%s\\nCT0=%s\\n" "$AUTH_TOKEN" "$CT0"'
+    )
+    result = subprocess.run(
+        ["sh", "-c", probe],
+        capture_output=True,
+        text=True,
+        timeout=5,
+    )
+    assert result.returncode == 0, result.stderr
+    lines = dict(
+        line.split("=", 1) for line in result.stdout.strip().splitlines() if "=" in line
+    )
+    assert lines["AUTH"] == hostile_auth, "auth_token round-trip broke — injection possible"
+    assert lines["CT0"] == hostile_ct0, "ct0 round-trip broke — injection possible"
+    # And no side-effect files materialised.
+    assert not pwn_auth.exists()
+    assert not pwn_ct0.exists()