fix(provision): keep the WiFi password owner-only and out of --state output

The per-port state file holds the WiFi password and seed token in clear, but
save_state() created the directory and file with the default umask, so they
landed 0755/0644 and any local user could read them (#1754). --state also
printed both values verbatim, and that output gets pasted into issues.

- The state dir is created 0700 and state files 0600. Staging uses mkstemp
  (unique name, O_EXCL, 0600) plus fchmod, so a stale temp file or a planted
  symlink can't receive the secret.
- harden_state_dir() tightens the dir and every .json/.tmp file in it,
  including ones left by older versions. main() runs it before any read or
  early exit, so --state repairs old files too. It skips symlinks and files
  owned by other users, and only warns if chmod fails.
- The --dry-run NVS binary and the generator-failure CSV fallback are written
  0600 even when the file already exists. The generator's CSV and binary now
  live in a private mkdtemp() dir instead of the shared temp dir.
- --state shows password and seed_token as (set)/(empty). --show-secrets
  prints them.

Windows is unchanged: chmod can't set owner-only ACLs there.

Tests: tests/test_provision_secrets.py adds 11 tests; all 11 fail on the
base and the suite goes from 14 to 25 passing.

Refs #1754

Co-Authored-By: claude-flow <ruv@ruv.net>

firmware/esp32-csi-node/provision.py

@@ -44,6 +44,7 @@ import csv
 import io
 import json
 import os
+import stat
 import struct
 import subprocess
 import sys
@@ -129,6 +130,84 @@ def _state_path_for(port: str, state_dir: str) -> str:
     return os.path.join(state_dir, f"{safe}.json")
 
 
+# State files hold the WiFi password and seed token in cleartext (#1754), so the
+# directory is owner-only and every file in it is 0600.
+STATE_DIR_MODE = 0o700
+STATE_FILE_MODE = 0o600
+
+# Values `--state` hides unless `--show-secrets` is passed (#1754).
+SECRET_ATTRS = ("password", "seed_token")
+
+
+def _restrict_mode(path: str, mode: int) -> None:
+    """chmod `path` to `mode` if it is a real file or dir owned by this user.
+
+    Never raises: a read-only or unusual filesystem must not block provisioning,
+    but a secret left readable by others is reported. Symlinks are not followed,
+    so a planted link can't redirect the chmod.
+    """
+    try:
+        st = os.lstat(path)
+    except FileNotFoundError:
+        return
+    except OSError as exc:
+        print(f"WARNING: could not stat {path}: {exc}", file=sys.stderr)
+        return
+    if stat.S_ISLNK(st.st_mode):
+        print(f"WARNING: not changing permissions through symlink {path}", file=sys.stderr)
+        return
+    if hasattr(os, "getuid") and st.st_uid != os.getuid():
+        print(f"WARNING: {path} is not owned by you; permissions left unchanged",
+              file=sys.stderr)
+        return
+    if stat.S_IMODE(st.st_mode) == mode:
+        return
+    try:
+        os.chmod(path, mode)
+    except OSError as exc:
+        print(f"WARNING: could not set {oct(mode)} on {path}: {exc}", file=sys.stderr)
+
+
+def harden_state_dir(state_dir: str) -> None:
+    """Make the state dir 0700 and its state/temp files 0600.
+
+    Covers files written by earlier versions of this script, which used the
+    default umask (0755 dir, 0644 files). POSIX only: chmod can't set
+    owner-only ACLs on Windows.
+    """
+    if sys.platform == "win32" or not os.path.isdir(state_dir):
+        return
+    _restrict_mode(state_dir, STATE_DIR_MODE)
+    try:
+        names = os.listdir(state_dir)
+    except OSError:
+        return
+    for name in names:
+        if name.endswith((".json", ".tmp")):
+            _restrict_mode(os.path.join(state_dir, name), STATE_FILE_MODE)
+
+
+def _write_private(path: str, data: bytes) -> None:
+    """Write a credential-bearing file (state, NVS CSV or binary) as 0600."""
+    flags = (os.O_WRONLY | os.O_CREAT | os.O_TRUNC
+             | getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_BINARY", 0))
+    fd = os.open(path, flags, STATE_FILE_MODE)
+    with os.fdopen(fd, "wb") as f:
+        # O_CREAT ignores the mode when the file already exists.
+        if hasattr(os, "fchmod"):
+            os.fchmod(f.fileno(), STATE_FILE_MODE)
+        f.write(data)
+
+
+def redact_secrets(state: dict) -> dict:
+    """Copy of `state` with secret values replaced by a fixed marker."""
+    shown = dict(state)
+    for name in SECRET_ATTRS:
+        if shown.get(name) is not None:
+            shown[name] = "(set)" if shown[name] else "(empty)"
+    return shown
+
+
 def load_state(port: str, state_dir: str) -> dict:
     """Return the merged-state dict for `port`, or `{}` if absent / unreadable."""
     path = _state_path_for(port, state_dir)
@@ -145,15 +224,32 @@ def load_state(port: str, state_dir: str) -> dict:
 
 
 def save_state(port: str, state_dir: str, state: dict) -> str:
-    """Write `state` to the per-port file, creating dirs as needed. Returns path."""
-    os.makedirs(state_dir, exist_ok=True)
+    """Write `state` to the per-port file, creating dirs as needed. Returns path.
+
+    The file holds secrets, so it is written 0600 inside a 0700 dir (#1754).
+    """
+    os.makedirs(state_dir, mode=STATE_DIR_MODE, exist_ok=True)
+    harden_state_dir(state_dir)
     path = _state_path_for(port, state_dir)
-    # Sort keys for deterministic on-disk content (easier to diff).
-    tmp = path + ".tmp"
-    with open(tmp, "w", encoding="utf-8") as f:
-        json.dump(state, f, indent=2, sort_keys=True)
-        f.write("\n")
-    os.replace(tmp, path)
+    # mkstemp picks a unique name and opens it O_EXCL with mode 0600, so a
+    # stale temp file or a planted symlink can't receive the secret.
+    fd, tmp = tempfile.mkstemp(
+        dir=state_dir, prefix=os.path.basename(path) + ".", suffix=".tmp"
+    )
+    try:
+        with os.fdopen(fd, "w", encoding="utf-8") as f:
+            if hasattr(os, "fchmod"):
+                os.fchmod(f.fileno(), STATE_FILE_MODE)
+            # Sort keys for deterministic on-disk content (easier to diff).
+            json.dump(state, f, indent=2, sort_keys=True)
+            f.write("\n")
+        os.replace(tmp, path)
+    except BaseException:
+        try:
+            os.unlink(tmp)
+        except OSError:
+            pass
+        raise
     return path
 
 
@@ -239,11 +335,12 @@ def build_nvs_csv(args):
 
 def generate_nvs_binary(csv_content, size):
     """Generate an NVS partition binary from CSV using nvs_partition_gen.py."""
-    with tempfile.NamedTemporaryFile(mode="w", suffix=".csv", delete=False) as f_csv:
-        f_csv.write(csv_content)
-        csv_path = f_csv.name
-
-    bin_path = csv_path.replace(".csv", ".bin")
+    # Both files carry the WiFi password. A private 0700 dir keeps the
+    # generator's output (written with the default umask) away from other users.
+    work_dir = tempfile.mkdtemp(prefix="provision-nvs-")
+    csv_path = os.path.join(work_dir, "nvs.csv")
+    bin_path = os.path.join(work_dir, "nvs.bin")
+    _write_private(csv_path, csv_content.encode("utf-8"))
 
     try:
         # Method 1: subprocess invocation (most reliable across package versions)
@@ -282,6 +379,10 @@ def generate_nvs_binary(csv_content, size):
         for p in (csv_path, bin_path):
             if os.path.isfile(p):
                 os.unlink(p)
+        try:
+            os.rmdir(work_dir)
+        except OSError:
+            pass
 
 
 def flash_nvs(port, baud, nvs_bin, chip):
@@ -367,10 +468,17 @@ def main():
                         help="Override the per-user state directory (default: per-OS user config dir).")
     parser.add_argument("--state", action="store_true",
                         help="Print the merged state that WOULD be flashed for this port and exit. "
-                        "Useful for debugging which keys are about to land on the device.")
+                        "Useful for debugging which keys are about to land on the device. "
+                        "The WiFi password and seed token are shown as (set)/(empty).")
+    parser.add_argument("--show-secrets", action="store_true",
+                        help="With --state, print the WiFi password and seed token in clear.")
 
     args = parser.parse_args()
 
+    # State written by older versions may be 0644. Tighten it before any read
+    # or early exit (#1754).
+    harden_state_dir(args.state_dir)
+
     # --- Per-port state load + merge (additive-by-default, #391 / #574) ---
     if args.reset:
         path = _state_path_for(args.port, args.state_dir)
@@ -383,7 +491,10 @@ def main():
     merged = merge_state_into_args(args, prior)
 
     if args.state:
-        print(json.dumps(merged, indent=2, sort_keys=True))
+        shown = merged if args.show_secrets else redact_secrets(merged)
+        print(json.dumps(shown, indent=2, sort_keys=True))
+        if shown != merged:
+            print("Secrets hidden; pass --show-secrets to print them.", file=sys.stderr)
         return
 
     if not has_config_value(args):
@@ -487,8 +598,7 @@ def main():
         print(f"\nError generating NVS binary: {e}", file=sys.stderr)
         print("\nFallback: save CSV and flash manually with ESP-IDF tools.", file=sys.stderr)
         fallback_path = "nvs_config.csv"
-        with open(fallback_path, "w") as f:
-            f.write(csv_content)
+        _write_private(fallback_path, csv_content.encode("utf-8"))
         print(f"Saved NVS CSV to {fallback_path}", file=sys.stderr)
         print(f"Flash with: python $IDF_PATH/components/nvs_flash/"
               f"nvs_partition_generator/nvs_partition_gen.py generate "
@@ -497,8 +607,7 @@ def main():
 
     if args.dry_run:
         out = "nvs_provision.bin"
-        with open(out, "wb") as f:
-            f.write(nvs_bin)
+        _write_private(out, nvs_bin)
         print(f"NVS binary saved to {out} ({len(nvs_bin)} bytes)")
         print(f"Flash manually: python -m esptool --chip {args.chip} --port {args.port} "
               f"write_flash 0x9000 {out}")

firmware/esp32-csi-node/tests/test_provision_secrets.py

@@ -0,0 +1,218 @@
+"""Tests for how provision.py protects the WiFi password and seed token (#1754).
+
+The state file, the --dry-run NVS binary and the fallback CSV all hold the
+secrets in clear, so they must be owner-only, and --state must not print them
+unless asked. main() runs in-process with the NVS generator stubbed, so no
+serial port is opened and no ESP-IDF tooling is needed. All credentials here
+are placeholders.
+"""
+
+import contextlib
+import importlib.util
+import io
+import json
+import os
+import shutil
+import stat
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+from unittest import mock
+
+PROVISION_PATH = Path(__file__).resolve().parents[1] / "provision.py"
+SPEC = importlib.util.spec_from_file_location("provision", PROVISION_PATH)
+provision = importlib.util.module_from_spec(SPEC)
+SPEC.loader.exec_module(provision)
+
+FAKE_PASSWORD = "fake-pass-not-real"
+FAKE_TOKEN = "fake-token-not-real"
+POSIX_ONLY = unittest.skipIf(sys.platform == "win32", "POSIX permission model only")
+
+
+def mode_of(path):
+    return stat.S_IMODE(os.lstat(path).st_mode)
+
+
+class _TempDirCase(unittest.TestCase):
+    def setUp(self):
+        self.root = tempfile.mkdtemp(prefix="provision-secrets-")
+        self.state_dir = os.path.join(self.root, "state")
+        # The default umask is what made files 0644 in the first place.
+        self.old_umask = os.umask(0o022)
+
+    def tearDown(self):
+        os.umask(self.old_umask)
+        shutil.rmtree(self.root, ignore_errors=True)
+
+    def run_main(self, *argv, cwd=None):
+        """Run provision.main() with argv; return (exit code, stdout, stderr)."""
+        out, err = io.StringIO(), io.StringIO()
+        code = 0
+        old_cwd = os.getcwd()
+        try:
+            os.chdir(cwd or self.root)
+            with mock.patch.object(sys, "argv", ["provision.py", *argv]), \
+                    contextlib.redirect_stdout(out), contextlib.redirect_stderr(err):
+                try:
+                    provision.main()
+                except SystemExit as exc:
+                    code = exc.code
+        finally:
+            os.chdir(old_cwd)
+        return code, out.getvalue(), err.getvalue()
+
+
+@POSIX_ONLY
+class TestStateFilePermissions(_TempDirCase):
+    def test_new_state_dir_and_file_are_owner_only(self):
+        path = provision.save_state("COM7", self.state_dir, {"password": FAKE_PASSWORD})
+        self.assertEqual(mode_of(self.state_dir), 0o700)
+        self.assertEqual(mode_of(path), 0o600)
+
+    def test_save_tightens_files_left_by_an_older_version(self):
+        os.makedirs(self.state_dir)
+        os.chmod(self.state_dir, 0o755)
+        legacy = provision._state_path_for("/dev/ttyUSB0", self.state_dir)
+        stale_tmp = legacy + ".tmp"
+        for path in (legacy, stale_tmp):
+            with open(path, "w", encoding="utf-8") as f:
+                json.dump({"password": FAKE_PASSWORD}, f)
+            os.chmod(path, 0o644)
+
+        provision.save_state("COM7", self.state_dir, {"ssid": "test-ssid"})
+
+        self.assertEqual(mode_of(self.state_dir), 0o700)
+        self.assertEqual(mode_of(legacy), 0o600)
+        self.assertEqual(mode_of(stale_tmp), 0o600)
+
+    def test_planted_symlink_at_old_temp_path_does_not_receive_secret(self):
+        os.makedirs(self.state_dir, mode=0o700)
+        victim = os.path.join(self.root, "victim.txt")
+        with open(victim, "w", encoding="utf-8") as f:
+            f.write("original")
+        os.symlink(victim, provision._state_path_for("COM7", self.state_dir) + ".tmp")
+
+        provision.save_state("COM7", self.state_dir, {"password": FAKE_PASSWORD})
+
+        with open(victim, encoding="utf-8") as f:
+            self.assertEqual(f.read(), "original")
+
+    def test_hardening_does_not_chmod_through_a_symlink(self):
+        os.makedirs(self.state_dir, mode=0o700)
+        victim = os.path.join(self.root, "victim.txt")
+        with open(victim, "w", encoding="utf-8") as f:
+            f.write("not a state file")
+        os.chmod(victim, 0o644)
+        os.symlink(victim, os.path.join(self.state_dir, "link.json"))
+
+        with contextlib.redirect_stderr(io.StringIO()):
+            provision.harden_state_dir(self.state_dir)
+
+        self.assertEqual(mode_of(victim), 0o644)
+
+    def test_state_inspection_repairs_legacy_permissions(self):
+        # --state never writes, so the repair has to happen before the read.
+        os.makedirs(self.state_dir)
+        os.chmod(self.state_dir, 0o755)
+        legacy = provision._state_path_for("COM7", self.state_dir)
+        with open(legacy, "w", encoding="utf-8") as f:
+            json.dump({"password": FAKE_PASSWORD}, f)
+        os.chmod(legacy, 0o644)
+
+        code, _, _ = self.run_main("--port", "COM7", "--state-dir", self.state_dir, "--state")
+
+        self.assertEqual(code, 0)
+        self.assertEqual(mode_of(self.state_dir), 0o700)
+        self.assertEqual(mode_of(legacy), 0o600)
+
+
+class TestStateOutputMasking(_TempDirCase):
+    def setUp(self):
+        super().setUp()
+        provision.save_state("COM7", self.state_dir, {
+            "ssid": "test-ssid",
+            "password": FAKE_PASSWORD,
+            "seed_token": FAKE_TOKEN,
+            "target_ip": "192.0.2.10",
+        })
+
+    def test_state_hides_password_and_seed_token(self):
+        code, out, err = self.run_main(
+            "--port", "COM7", "--state-dir", self.state_dir, "--state")
+
+        self.assertEqual(code, 0)
+        self.assertNotIn(FAKE_PASSWORD, out)
+        self.assertNotIn(FAKE_TOKEN, out)
+        shown = json.loads(out)
+        self.assertEqual(shown["password"], "(set)")
+        self.assertEqual(shown["seed_token"], "(set)")
+        self.assertEqual(shown["ssid"], "test-ssid")
+        self.assertIn("--show-secrets", err)
+
+    def test_show_secrets_prints_them(self):
+        code, out, _ = self.run_main(
+            "--port", "COM7", "--state-dir", self.state_dir, "--state", "--show-secrets")
+
+        self.assertEqual(code, 0)
+        shown = json.loads(out)
+        self.assertEqual(shown["password"], FAKE_PASSWORD)
+        self.assertEqual(shown["seed_token"], FAKE_TOKEN)
+
+    def test_redact_marks_empty_and_leaves_absent_alone(self):
+        shown = provision.redact_secrets({"password": "", "ssid": "test-ssid"})
+        self.assertEqual(shown, {"password": "(empty)", "ssid": "test-ssid"})
+
+
+@POSIX_ONLY
+class TestIntermediateArtifacts(_TempDirCase):
+    ARGS = ("--port", "COM7", "--ssid", "test-ssid", "--password", FAKE_PASSWORD,
+            "--target-ip", "192.0.2.10", "--dry-run")
+
+    def test_dry_run_binary_is_owner_only_even_if_it_already_existed(self):
+        out_path = os.path.join(self.root, "nvs_provision.bin")
+        with open(out_path, "wb") as f:
+            f.write(b"old")
+        os.chmod(out_path, 0o644)
+
+        with mock.patch.object(provision, "generate_nvs_binary",
+                               return_value=b"nvs-with-" + FAKE_PASSWORD.encode()):
+            code, _, _ = self.run_main(*self.ARGS, "--state-dir", self.state_dir)
+
+        self.assertIn(code, (0, None))
+        self.assertEqual(mode_of(out_path), 0o600)
+
+    def test_fallback_csv_is_owner_only(self):
+        with mock.patch.object(provision, "generate_nvs_binary",
+                               side_effect=RuntimeError("generator missing")):
+            code, _, _ = self.run_main(*self.ARGS, "--state-dir", self.state_dir)
+
+        self.assertEqual(code, 1)
+        csv_path = os.path.join(self.root, "nvs_config.csv")
+        self.assertEqual(mode_of(csv_path), 0o600)
+
+    def test_generator_files_live_in_a_private_dir(self):
+        seen = {}
+
+        def fake_generator(cmd, **_kwargs):
+            csv_path, bin_path = cmd[-3], cmd[-2]
+            seen["dir_mode"] = mode_of(os.path.dirname(bin_path))
+            seen["csv_mode"] = mode_of(csv_path)
+            with open(bin_path, "wb") as f:
+                f.write(b"nvs")
+
+        # A shared temp dir like Linux /tmp, rather than macOS's per-user one.
+        shared_tmp = os.path.join(self.root, "shared-tmp")
+        os.makedirs(shared_tmp)
+        os.chmod(shared_tmp, 0o755)
+        with mock.patch.object(provision.tempfile, "tempdir", shared_tmp), \
+                mock.patch.object(provision.subprocess, "check_call",
+                                  side_effect=fake_generator):
+            self.assertEqual(provision.generate_nvs_binary("key,type\n", 0x6000), b"nvs")
+
+        self.assertEqual(seen["dir_mode"], 0o700)
+        self.assertEqual(seen["csv_mode"], 0o600)
+
+
+if __name__ == "__main__":
+    unittest.main()