Merge PR #2153 (aepod) into release integration

firmware/esp32-csi-node/README.md

@@ -161,10 +161,16 @@ board default and emits no antenna-path claim.
 ### 3. Provision WiFi credentials (no reflash needed)
 
 ```bash
+# Keep the WiFi password in an owner-only file, not on the command line.
+printf '%s\n' 'YourPass' > ~/.wifi-pass && chmod 600 ~/.wifi-pass
 python firmware/esp32-csi-node/provision.py --port COM7 \
-  --ssid "YourSSID" --password "YourPass" --target-ip 192.168.1.20
+  --ssid "YourSSID" --password-file ~/.wifi-pass --target-ip 192.168.1.20
 ```
 
+Leave out both `--password` and `--password-file` and the script asks for the
+password when run from a terminal. `--password` still works, but it shows up in
+`ps` and shell history.
+
 ### 4. Start the sensing server
 
 ```bash
@@ -440,10 +446,15 @@ The easiest way to write NVS settings:
 ```bash
 python firmware/esp32-csi-node/provision.py --port COM7 \
   --ssid "MyWiFi" \
-  --password "MyPassword" \
+  --password-file ~/.wifi-pass \
   --target-ip 192.168.1.20
 ```
 
+The password file must be owner-only (`chmod 600`); one trailing newline is
+dropped. Without a password flag the script prompts on a terminal; in scripts
+and CI it doesn't prompt and fails if no password was given now or saved
+earlier for that port.
+
 ### NVS Key Reference
 
 #### Network Settings

firmware/esp32-csi-node/provision.py

@@ -41,9 +41,11 @@ ADDITIVE-BY-DEFAULT (issue #391, #574 phase 1):
 
 import argparse
 import csv
+import getpass
 import io
 import json
 import os
+import stat
 import struct
 import subprocess
 import sys
@@ -129,6 +131,121 @@ 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 read_password_file(path: str, allow_insecure: bool = False) -> str:
+    """Read the WiFi password from `path`, dropping one trailing newline.
+
+    Keeps the password out of argv and shell history (#1754). Refuses a file
+    that group or others can read unless `allow_insecure` is set. Raises
+    ValueError with a message meant for the user.
+    """
+    try:
+        with open(path, "rb") as f:
+            mode = stat.S_IMODE(os.fstat(f.fileno()).st_mode)
+            data = f.read()
+    except FileNotFoundError:
+        raise ValueError(f"--password-file {path} does not exist") from None
+    except OSError as exc:
+        raise ValueError(f"could not read --password-file {path}: {exc}") from None
+    if sys.platform != "win32" and mode & 0o044:
+        if not allow_insecure:
+            raise ValueError(
+                f"--password-file {path} is readable by group or others "
+                f"(mode {oct(mode)}). Run 'chmod 600 {path}', or pass "
+                f"--allow-insecure-password-file."
+            )
+        print(f"WARNING: --password-file {path} is readable by group or others "
+              f"(mode {oct(mode)}).", file=sys.stderr)
+    try:
+        text = data.decode("utf-8")
+    except UnicodeDecodeError:
+        raise ValueError(f"--password-file {path} is not UTF-8 text") from None
+    if text.endswith("\r\n"):
+        text = text[:-2]
+    elif text.endswith("\n"):
+        text = text[:-1]
+    if not text:
+        raise ValueError(f"--password-file {path} is empty")
+    return text
+
+
 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 +262,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 +373,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 +417,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):
@@ -324,7 +463,16 @@ def main():
     )
     parser.add_argument("--baud", type=int, default=460800, help="Flash baud rate (default: 460800)")
     parser.add_argument("--ssid", help="WiFi SSID")
-    parser.add_argument("--password", help="WiFi password")
+    password_source = parser.add_mutually_exclusive_group()
+    password_source.add_argument("--password",
+                                 help="WiFi password. Visible in ps and shell history; "
+                                 "prefer --password-file.")
+    password_source.add_argument("--password-file", metavar="PATH",
+                                 help="Read the WiFi password from PATH (one trailing newline "
+                                 "is dropped). The file must not be readable by group or others.")
+    parser.add_argument("--allow-insecure-password-file", action="store_true",
+                        help="Accept a --password-file that group or others can read, with a "
+                        "warning (e.g. a read-only 0444 secrets mount).")
     parser.add_argument("--target-ip", help="Aggregator host IP (e.g. 192.168.1.20)")
     parser.add_argument("--target-port", type=int, help="Aggregator UDP port (default: 5005)")
     parser.add_argument("--node-id", type=int, help="Node ID 0-255 (default: 1)")
@@ -368,10 +516,28 @@ 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)
+
+    if args.password_file is not None:
+        try:
+            args.password = read_password_file(
+                args.password_file, args.allow_insecure_password_file)
+        except ValueError as exc:
+            parser.error(str(exc))
+    elif args.password is not None:
+        print("WARNING: --password is visible in ps and shell history; "
+              "use --password-file instead.", file=sys.stderr)
+    cli_ssid, cli_password = args.ssid, args.password
+
     # --- Per-port state load + merge (additive-by-default, #391 / #574) ---
     if args.reset:
         # Don't delete the state file here: validation below can still fail,
@@ -385,8 +551,21 @@ def main():
         prior = load_state(args.port, args.state_dir)
     merged = merge_state_into_args(args, prior)
 
+    # A new --ssid with no password given: ask for it on a terminal instead of
+    # requiring it on the command line. The saved password is reused only when
+    # it belongs to the same SSID. Without a terminal (scripts, CI) nothing is
+    # asked and the WiFi-credential check below applies as before.
+    if (not args.state and cli_ssid is not None and cli_password is None
+            and (prior.get("password") is None or prior.get("ssid") != cli_ssid)
+            and sys.stdin.isatty()):
+        args.password = getpass.getpass(f"WiFi password for {cli_ssid}: ")
+        merged["password"] = args.password
+
     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):
@@ -490,8 +669,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 "
@@ -500,8 +678,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,345 @@
+"""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)
+
+
+class TestPasswordInput(_TempDirCase):
+    """Ways to give the password without putting it on the command line."""
+
+    BASE = ("--port", "COM7", "--target-ip", "192.0.2.10")
+
+    def setUp(self):
+        super().setUp()
+        self.flashed_csv = []
+
+    def password_file(self, content, mode=0o600):
+        path = os.path.join(self.root, "wifi-pass.txt")
+        with open(path, "w", encoding="utf-8", newline="") as f:
+            f.write(content)
+        os.chmod(path, mode)
+        return path
+
+    def provision(self, *argv, tty=False, typed=None):
+        """Run main() with flashing stubbed; return (code, stderr, getpass mock)."""
+        def fake_generate(csv_content, _size):
+            self.flashed_csv.append(csv_content)
+            return b"nvs"
+
+        stdin = mock.Mock()
+        stdin.isatty.return_value = tty
+        # read_chip_mac only exists once state is keyed by board MAC (#1755);
+        # stub it so these tests never reach esptool or a serial port.
+        with mock.patch.object(provision, "generate_nvs_binary", side_effect=fake_generate), \
+                mock.patch.object(provision, "flash_nvs"), \
+                mock.patch.object(provision, "read_chip_mac", return_value=None,
+                                  create=True), \
+                mock.patch.object(sys, "stdin", stdin), \
+                mock.patch("getpass.getpass", return_value=typed) as prompt:
+            code, _, err = self.run_main(*self.BASE, *argv, "--state-dir", self.state_dir)
+        return code, err, prompt
+
+    def flashed_password(self):
+        rows = provision.csv.DictReader(io.StringIO(self.flashed_csv[-1]))
+        return {row["key"]: row["value"] for row in rows}.get("password")
+
+    def test_password_file_is_read_and_one_newline_dropped(self):
+        path = self.password_file(FAKE_PASSWORD + "  \n\n")
+
+        code, err, _ = self.provision("--ssid", "test-ssid", "--password-file", path)
+
+        self.assertEqual(code, 0)
+        # Only one newline goes; trailing spaces and the second newline stay.
+        self.assertEqual(self.flashed_password(), FAKE_PASSWORD + "  \n")
+        self.assertNotIn("visible in ps", err)
+
+    def test_password_file_crlf_counts_as_one_newline(self):
+        path = self.password_file(FAKE_PASSWORD + "\r\n")
+        self.provision("--ssid", "test-ssid", "--password-file", path)
+        self.assertEqual(self.flashed_password(), FAKE_PASSWORD)
+
+    @POSIX_ONLY
+    def test_group_or_world_readable_password_file_is_refused(self):
+        for mode in (0o644, 0o640, 0o604):
+            with self.subTest(mode=oct(mode)):
+                path = self.password_file(FAKE_PASSWORD + "\n", mode)
+                code, err, _ = self.provision("--ssid", "test-ssid", "--password-file", path)
+                self.assertEqual(code, 2)
+                self.assertIn("chmod 600", err)
+
+    @POSIX_ONLY
+    def test_insecure_password_file_allowed_with_flag_and_warning(self):
+        path = self.password_file(FAKE_PASSWORD + "\n", 0o644)
+
+        code, err, _ = self.provision("--ssid", "test-ssid", "--password-file", path,
+                                      "--allow-insecure-password-file")
+
+        self.assertEqual(code, 0)
+        self.assertIn("readable by group or others", err)
+        self.assertEqual(self.flashed_password(), FAKE_PASSWORD)
+
+    def test_missing_or_empty_password_file_is_an_error(self):
+        missing = os.path.join(self.root, "nope.txt")
+        code, err, _ = self.provision("--ssid", "test-ssid", "--password-file", missing)
+        self.assertEqual(code, 2)
+        self.assertIn("does not exist", err)
+
+        empty = self.password_file("\n")
+        code, err, _ = self.provision("--ssid", "test-ssid", "--password-file", empty)
+        self.assertEqual(code, 2)
+        self.assertIn("is empty", err)
+
+    def test_password_and_password_file_are_mutually_exclusive(self):
+        path = self.password_file(FAKE_PASSWORD)
+
+        code, err, _ = self.provision("--ssid", "test-ssid", "--password", FAKE_PASSWORD,
+                                      "--password-file", path)
+
+        self.assertEqual(code, 2)
+        self.assertIn("not allowed with argument", err)
+
+    def test_password_on_command_line_warns(self):
+        code, err, _ = self.provision("--ssid", "test-ssid", "--password", FAKE_PASSWORD)
+
+        self.assertEqual(code, 0)
+        self.assertIn("--password is visible in ps and shell history", err)
+        self.assertIn("--password-file", err)
+
+    def test_terminal_prompts_for_password_of_a_new_ssid(self):
+        code, _, prompt = self.provision("--ssid", "test-ssid", tty=True, typed=FAKE_PASSWORD)
+
+        self.assertEqual(code, 0)
+        prompt.assert_called_once()
+        self.assertEqual(self.flashed_password(), FAKE_PASSWORD)
+        self.assertEqual(provision.load_state("COM7", self.state_dir)["password"], FAKE_PASSWORD)
+
+    def test_no_prompt_when_saved_password_matches_the_ssid(self):
+        provision.save_state("COM7", self.state_dir,
+                             {"ssid": "test-ssid", "password": FAKE_PASSWORD})
+
+        code, _, prompt = self.provision("--ssid", "test-ssid", tty=True, typed="unused")
+
+        self.assertEqual(code, 0)
+        prompt.assert_not_called()
+        self.assertEqual(self.flashed_password(), FAKE_PASSWORD)
+
+    def test_no_prompt_without_a_terminal(self):
+        code, err, prompt = self.provision("--ssid", "test-ssid", tty=False, typed="unused")
+
+        prompt.assert_not_called()
+        self.assertEqual(code, 2)
+        self.assertIn("--password", err)
+
+
+if __name__ == "__main__":
+    unittest.main()