fix(p0-5): atomic write_secret with flock + fsync + atomic replace (#431)

* test(secrets-store): add failing scaffold for concurrent write atomicity (F007 S1)

* feat(secrets-store): add write_secret with flock + atomic replace + fsync (F007 S2)

* fix(cli): configure --replace routes through atomic write_secret (F007 S3)

* fix(cli): _write_cookie_to_secrets routes through atomic write_secret (F007 S4)

* test(secrets-store): cover comment + unknown line preservation (F007 S5)

* fix(secrets-store): make fcntl import optional for Windows compatibility

* fix(secrets-store): replace empty except with explanatory comment + debug log

* fix(secrets-store): validate key/value to reject newlines and bad keys

* refactor(cli): drop PLC0415 suppression by moving import to module scope

Author: 0xmariowu

autosearch/cli/main.py

@@ -9,6 +9,7 @@
 
 from autosearch import __version__
 from autosearch.cli.mcp_config_writers import MCPConfigWriteError
+from autosearch.core import secrets_store
 from autosearch.core.environment_probe import probe_environment
 from autosearch.core.models import SearchMode
 from autosearch.core.scope_clarifier import ScopeClarifier
@@ -82,9 +83,7 @@ def main(
     # Push ~/.config/ai-secrets.env keys into process env so subcommands and
     # any provider/channel that does `os.getenv("FOO_API_KEY")` actually sees
     # what the user configured via `autosearch configure`.
-    from autosearch.core.secrets_store import inject_into_env
-
-    inject_into_env()
+    secrets_store.inject_into_env()
 
     if ctx.invoked_subcommand is None and not version:
         raise typer.Exit(code=0)
@@ -483,9 +482,6 @@ def configure(
     Default flow: prompts for the value with hidden input so the secret never
     appears on the command line, in shell history, or in `ps`.
     """
-    import shlex
-    import sys
-
     if from_stdin:
         value = sys.stdin.read().rstrip("\n")
     elif value is None:
@@ -502,19 +498,8 @@ def configure(
         typer.echo("error: value must not be empty.", err=True)
         raise typer.Exit(code=2)
 
-    # Bug 3 (fix-plan): write target must follow AUTOSEARCH_SECRETS_FILE so
-    # containers / CI / multi-user installs don't end up writing to A while
-    # the runtime reads B.
-    from autosearch.core.secrets_store import secrets_path as _secrets_path  # noqa: PLC0415
-
-    secrets_path = _secrets_path()
-    existing: dict[str, str] = {}
-    if secrets_path.exists():
-        for line in secrets_path.read_text(encoding="utf-8").splitlines():
-            stripped = line.strip()
-            if stripped and not stripped.startswith("#") and "=" in stripped:
-                k, _, v = stripped.partition("=")
-                existing[k.strip()] = v
+    secrets_path = secrets_store.secrets_path()
+    existing = secrets_store.load_secrets(secrets_path)
 
     if key in existing and not replace:
         typer.echo(
@@ -523,24 +508,7 @@ def configure(
         )
         raise typer.Exit(code=0)
 
-    secrets_path.parent.mkdir(parents=True, exist_ok=True)
-    if key in existing and replace:
-        # Rewrite the file in place with the new value, preserve all others.
-        existing[key] = shlex.quote(value)
-        secrets_path.write_text(
-            "\n".join(f"{k}={v}" for k, v in existing.items()) + "\n",
-            encoding="utf-8",
-        )
-    else:
-        with secrets_path.open("a", encoding="utf-8") as fh:
-            fh.write(f"\n{key}={shlex.quote(value)}\n")
-
-    # Restrict file permissions so other users on the box can't read the secrets.
-    try:
-        secrets_path.chmod(0o600)
-    except OSError:
-        pass
-
+    secrets_store.write_secret(key, value, path=secrets_path)
     typer.echo(f"Written: {key} -> {secrets_path}")
 
 
@@ -770,41 +738,17 @@ def _write_cookie_to_secrets(
     the same file we just wrote. Bug 4: chmods the file 0o600 after write
     so cookies aren't world-readable on shared boxes.
     """
-    import shlex
-
-    from autosearch.core.secrets_store import secrets_path as _secrets_path
-
-    secrets_path = _secrets_path()
-    secrets_path.parent.mkdir(parents=True, exist_ok=True)
-
-    existing_keys: set[str] = set()
-    if secrets_path.exists():
-        for line in secrets_path.read_text(encoding="utf-8").splitlines():
-            stripped = line.strip()
-            if stripped and not stripped.startswith("#") and "=" in stripped:
-                existing_keys.add(stripped.split("=", 1)[0].strip())
-
+    secrets_path = secrets_store.secrets_path()
+    existing = secrets_store.load_secrets(secrets_path)
     label = f"{n_cookies} cookies" if n_cookies else "cookies"
-    if env_key in existing_keys:
-        lines = secrets_path.read_text(encoding="utf-8").splitlines()
-        new_lines = [
-            f"{env_key}={shlex.quote(cookie_str)}"
-            if line.strip().startswith(f"{env_key}=")
-            else line
-            for line in lines
-        ]
-        secrets_path.write_text("\n".join(new_lines) + "\n", encoding="utf-8")
+    existed = env_key in existing
+
+    secrets_store.write_secret(env_key, cookie_str, path=secrets_path)
+    if existed:
         typer.echo(f"Updated {env_key} ({label}) → {secrets_path}")
     else:
-        with secrets_path.open("a", encoding="utf-8") as fh:
-            fh.write(f"\n{env_key}={shlex.quote(cookie_str)}\n")
         typer.echo(f"Written {env_key} ({label}) → {secrets_path}")
 
-    try:
-        secrets_path.chmod(0o600)
-    except OSError:
-        pass
-
 
 def _stderr_event_writer(event: dict[str, object]) -> None:
     sys.stderr.write(json.dumps(event, ensure_ascii=False) + "\n")

autosearch/core/secrets_store.py

@@ -17,11 +17,21 @@
 
 from __future__ import annotations
 
+import logging
 import os
+import re
 import shlex
+import tempfile
 from pathlib import Path
 
+try:
+    import fcntl as _fcntl
+except ImportError:  # pragma: no cover - exercised by import-time compatibility test
+    _fcntl = None
+
 _FILE_INJECTED_VALUES: dict[str, str] = {}
+_ENV_KEY_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$")
+_log = logging.getLogger(__name__)
 
 
 def secrets_path() -> Path:
@@ -63,6 +73,100 @@ def load_secrets(path: Path | None = None) -> dict[str, str]:
     return result
 
 
+def write_secret(key: str, value: str, *, path: Path | None = None) -> None:
+    """Atomically write or replace one KEY=value entry in the secrets file."""
+    _validate_secret_entry(key, value)
+    target = path or secrets_path()
+    target.parent.mkdir(parents=True, exist_ok=True)
+    lock_path = target.with_name(f"{target.name}.lock")
+    temp_path: str | None = None
+
+    with lock_path.open("a+b") as lock_fh:
+        if _fcntl is not None:
+            _fcntl.flock(lock_fh.fileno(), _fcntl.LOCK_EX)
+        try:
+            try:
+                existing_text = target.read_text(encoding="utf-8")
+            except FileNotFoundError:
+                existing_text = ""
+
+            next_text = _replace_or_append_secret(existing_text, key, value)
+            with tempfile.NamedTemporaryFile(
+                "w",
+                encoding="utf-8",
+                dir=target.parent,
+                prefix=f"{target.name}.tmp.{os.getpid()}.",
+                delete=False,
+            ) as temp_fh:
+                temp_path = temp_fh.name
+                temp_fh.write(next_text)
+                temp_fh.flush()
+                os.fsync(temp_fh.fileno())
+
+            os.replace(temp_path, target)
+            temp_path = None
+            try:
+                target.chmod(0o600)
+            except OSError:
+                # Some filesystems reject chmod; the secret write itself still succeeded.
+                _log.debug("Unable to chmod secrets file to 0600: %s", target, exc_info=True)
+        finally:
+            try:
+                if temp_path is not None:
+                    os.unlink(temp_path)
+            except FileNotFoundError:
+                # Preserve the original write error if another cleanup path removed the temp file.
+                _log.debug("Temporary secrets file already removed: %s", temp_path, exc_info=True)
+            finally:
+                if _fcntl is not None:
+                    _fcntl.flock(lock_fh.fileno(), _fcntl.LOCK_UN)
+
+
+def _replace_or_append_secret(text: str, key: str, value: str) -> str:
+    replacement = _format_secret_line(key, value)
+    lines = text.splitlines()
+    replaced = False
+    next_lines: list[str] = []
+
+    for line in lines:
+        if _secret_line_key(line) == key:
+            next_lines.append(replacement)
+            replaced = True
+        else:
+            next_lines.append(line)
+
+    if not replaced:
+        next_lines.append(replacement)
+    return "\n".join(next_lines) + "\n"
+
+
+def _format_secret_line(key: str, value: str) -> str:
+    _validate_secret_entry(key, value)
+    return f"{key}={shlex.quote(value)}"
+
+
+def _validate_secret_entry(key: str, value: str) -> None:
+    if not _ENV_KEY_RE.fullmatch(key):
+        raise ValueError("secret key must match [A-Za-z_][A-Za-z0-9_]*")
+    if any(char in value for char in ("\n", "\r", "\0")):
+        raise ValueError("secret value must not contain newline, carriage return, or NUL")
+
+
+def _secret_line_key(raw: str) -> str | None:
+    line = raw.strip()
+    if not line or line.startswith("#") or "=" not in line:
+        return None
+    key, _, value = line.partition("=")
+    key = key.strip()
+    if not key or not key.replace("_", "").isalnum():
+        return None
+    try:
+        shlex.split(value)
+    except ValueError:
+        return None
+    return key
+
+
 def available_env_keys(path: Path | None = None) -> set[str]:
     """Return the set of keys with non-empty values in env OR the secrets file."""
     keys = {key for key, val in os.environ.items() if val}