phase: 24_dev_mtls_rollout

**Phase 24 final verification — all green**

- **Verified (read-only re-run, 62/62 PASS)** via `data/logs/verify-24-mtls.py`: nine sources `ok`; `/api/certs` all three `active` with `deployed_bundle_rev == bundle_rev == 2`; `agent.tls: true` + `agent.version 1.1.0` on every host; driveripper 16/16 drives live; gateway absent from `/disks` but `disks:gateway ok`; all served target URLs `https://`, no userinfo; pre-flip backup present; no transport/cert alert.
- **Criterion 1 (export):** `data/certs/{driveripper,deskwork,gateway}.bundle` re-checked byte-identical (sha256) to `GET /api/certs/<name>/bundle`, mode `0600`, under gitignored `data/`; live arms re-probed (issued→file, unknown/`issued:false`→one-line note + no file, wrong password→exit 2, nothing written).
- **Criterion 2 (`--fetch-bundle`):** 54 sandbox pins in `tests/test_deploy_scripts.py` pass — byte-identical `0640` install + TLS unit lines, refusals (unissued / 401 / non-bundle body / dead dashboard) before `need_root` with no write, token absent from token file, unit, stdout and sudoers hint.
- **Criterion 3–4:** `./scripts/dev-server.sh status` nine `ok` + bundles block; screenshots `data/logs/24-task04-{certs,disks,system}.png` exist.
- **Criterion 7:** no behavior change — `agent/` untouched, `src/` diff is a `certs.ensure_certs` docstring only; token-mode, revocation-before-dial and bundle-route tests green.
- **Tests/lint:** `bash .agents/validate.sh` → **EXIT=0**; `uv run pytest --cov --cov-fail-under=94` → **1542 passed, 15 skipped, coverage 98.94%**; `ruff check src tests agent` → All checks passed; `pyright` → 0 errors; `node --check` clean; `uv run pytest -m live --live` → **11 passed**.
- **Docs:** README.md (all 1144 lines) + AGENTS.md reviewed against the live project — every config.py env var is in README and both templates, route set / severity thresholds / test counts / coverage figure match, all repo paths, cited test names and `Dash.*`/`DashCharts.*` helpers resolve, phase-24 doc pins green. **Verified consistent — no changes needed.**
- **Notable:** left `fastapi 0.153.0 → 0.158.0` (validate reports it as advisory; no dependency was touched, rule 8 conditions the check on deps moving). Killed two abandoned `/tmp/fbtest` loopback stub processes left by task 02's manual probe; dev server unaffected.
- **Next pending phase:** `.agents/phases/todo/25_masthead_control_placement/`.

tests/test_certs.py

@@ -25,13 +25,15 @@ import socket
 import ssl
 import tempfile
 import threading
+from datetime import UTC, datetime, timedelta
 from pathlib import Path
 from typing import Any
 
 import pytest
 from cryptography import x509
 from cryptography.exceptions import InvalidSignature
-from cryptography.hazmat.primitives.asymmetric import rsa
+from cryptography.hazmat.primitives import hashes
+from cryptography.hazmat.primitives.asymmetric import ec, rsa
 from cryptography.hazmat.primitives.asymmetric.ec import EllipticCurvePrivateKey
 from cryptography.hazmat.primitives.asymmetric.types import PublicKeyTypes
 from cryptography.hazmat.primitives.serialization import (
@@ -371,6 +373,146 @@ def test_every_certificate_carries_the_identifiers_a_strict_verifier_requires()
     assert context.verify_flags & ssl.VERIFY_X509_STRICT, "this is the mode being satisfied"
 
 
+def _legacy_authority() -> FakeMeta:
+    """The meta rows a pre-identifier `certs.py` left in a database: a root **with no SKI**.
+
+    This is the shape phase 24 task 03 found on a live store. The CA row was minted before
+    `SubjectKeyIdentifier` landed, and "a stored row always wins" means no later start re-mints
+    it, so every leaf signed afterwards carries an AKI naming an identifier its issuer does not
+    have. Written by hand here rather than by an old revision of the module, because the point is
+    to read what a database can hold, not what this module would choose to write.
+
+    The real clock, for the same reason the handshake test below uses it: `ssl` judges the
+    validity window itself, and the fixed `NOW` would read as an expired certificate.
+    """
+    def pem_key(key: Any) -> str:
+        return key.private_bytes(Encoding.PEM, PrivateFormat.PKCS8,
+                                 NoEncryption()).decode("ascii")
+
+    def pem_cert(cert: x509.Certificate) -> str:
+        return cert.public_bytes(Encoding.PEM).decode("ascii")
+
+    def subject(name: str) -> x509.Name:
+        return x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, name)])
+
+    moment = datetime.now(tz=UTC)
+    root_key = ec.generate_private_key(ec.SECP256R1())
+    root = (x509.CertificateBuilder().subject_name(subject(certs.ROOT_CN))
+            .issuer_name(subject(certs.ROOT_CN))
+            .public_key(root_key.public_key())
+            .serial_number(x509.random_serial_number())
+            .not_valid_before(moment - timedelta(days=1))
+            .not_valid_after(moment + certs.ROOT_LIFETIME)
+            .add_extension(x509.BasicConstraints(ca=True, path_length=None), critical=True)
+            .sign(root_key, hashes.SHA256()))
+    client_key = ec.generate_private_key(ec.SECP256R1())
+    client = (x509.CertificateBuilder().subject_name(subject(certs.CLIENT_CN))
+              .issuer_name(subject(certs.ROOT_CN))
+              .public_key(client_key.public_key())
+              .serial_number(x509.random_serial_number())
+              .not_valid_before(moment - timedelta(days=1))
+              .not_valid_after(moment + certs.LEAF_LIFETIME)
+              .add_extension(x509.BasicConstraints(ca=False, path_length=None), critical=True)
+              .sign(root_key, hashes.SHA256()))
+    return FakeMeta({certs.META_ROOT_KEY: pem_key(root_key),
+                     certs.META_ROOT_CERT: pem_cert(root),
+                     certs.META_CLIENT_KEY: pem_key(client_key),
+                     certs.META_CLIENT_CERT: pem_cert(client),
+                     certs.META_AGENTS: "{}"})
+
+
+def _strict_dial_error(root_pem: str, leaf_pem: str, leaf_key_pem: str,
+                       tmp_path: Path) -> str | None:
+    """Serve `leaf_pem` and connect with the context the collector builds.
+
+    Returns `None` when the handshake completes, or the verification error's own sentence.
+    `create_default_context` is the whole point: it is what `collectors.AgentTls.ssl_context`
+    calls, so this is the dial the fleet actually makes and not a laxer stand-in.
+    """
+    here = tmp_path / "strict"
+    here.mkdir(parents=True, exist_ok=True)
+    (here / "leaf.pem").write_text(leaf_pem)
+    (here / "leaf.key").write_text(leaf_key_pem)
+    (here / "ca.pem").write_text(root_pem)
+
+    server = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
+    server.load_cert_chain(str(here / "leaf.pem"), str(here / "leaf.key"))
+    client = ssl.create_default_context(cafile=str(here / "ca.pem"))
+
+    listener = socket.socket()
+    listener.bind(("127.0.0.1", 0))
+    listener.listen(1)
+    listener.settimeout(15)
+    port = listener.getsockname()[1]
+
+    def serve() -> None:
+        try:
+            raw, _peer = listener.accept()
+        except OSError:
+            return
+        try:
+            with server.wrap_socket(raw, server_side=True) as tls:
+                tls.recv(1)
+        except OSError:
+            pass  # the server's half of a handshake the client refused
+
+    thread = threading.Thread(target=serve, daemon=True)
+    thread.start()
+    try:
+        with socket.create_connection(("127.0.0.1", port), timeout=15) as raw, \
+                client.wrap_socket(raw, server_hostname="localhost") as tls:
+            tls.sendall(b"x")
+        return None
+    except ssl.SSLCertVerificationError as exc:
+        return str(exc)
+    finally:
+        listener.close()
+        thread.join(timeout=15)
+
+
+def test_a_stored_root_without_an_identifier_still_boots_and_root_rotate_repairs_it(
+        tmp_path: Path) -> None:
+    """A CA older than the identifier fix is a rotatable CA, not a bricked dashboard.
+
+    Three claims, all of them learned on 2026-10-21 while moving the fleet onto installed bundles:
+
+    * `ssl.create_default_context()` — strict mode — refuses a chain whose **CA** carries no
+      Subject Key Identifier, so every leaf such a root signs is rejected by the dashboard's own
+      dial. The live store hit exactly this: its `certs.root.cert` row predated the extension, and
+      a leaf minted afterwards (with its own SKI and a correct AKI) still failed the handshake
+      with `CERTIFICATE_VERIFY_FAILED: Missing Subject Key Identifier`.
+    * Nothing may fail closed over it. `read_state` reads those rows because `/certs` is the page
+      that fixes the problem — a startup that refused this table would leave the operator no
+      **Rotate root** button, and `DELETE FROM meta` is the last resort rather than the first.
+    * `rotate_root` is therefore the repair, and it has to actually repair the chain rather than
+      only bump a revision: a new root with its identifiers, a new client identity, every agent
+      re-issued under it, and the strict dial accepting the result.
+    """
+    store = _legacy_authority()
+    state = certs.read_state(store)
+    with pytest.raises(x509.ExtensionNotFound):
+        root_of(state).extensions.get_extension_for_class(x509.SubjectKeyIdentifier)
+
+    # A leaf issued *now*, with every identifier current `certs.py` adds, is still unverifiable:
+    # the gap is in the issuer, and no care at signing time reaches it.
+    state = issue(store, state, "localhost", ["localhost", "127.0.0.1"])
+    before = bundle_pems(entry_of(state, "localhost").bundle)
+    refused = _strict_dial_error(state.root_cert, before["cert"], before["key"], tmp_path)
+    assert refused is not None, (
+        "a strict client accepted a chain from a CA with no Subject Key Identifier, so this "
+        "OpenSSL is not the strict mode the fleet's dial runs in and the pin below means nothing")
+    assert "certificate verify failed" in refused.lower()
+
+    rotated = certs.rotate_root(store, state)
+    entry = entry_of(rotated, "localhost")
+    assert entry.bundle_rev == 2, "every host is re-issued under the new root, one bump each"
+    assert cert_of(rotated.root_cert).extensions.get_extension_for_class(
+        x509.SubjectKeyIdentifier), "the rotated root carries the identifier its leaves name"
+    after = bundle_pems(entry.bundle)
+    accepted = _strict_dial_error(rotated.root_cert, after["cert"], after["key"], tmp_path)
+    assert accepted is None, f"rotating the root must make the chain dialable: {accepted}"
+
+
 def test_the_client_certificate_chains_to_the_root_and_is_a_client_identity() -> None:
     _store, state = boot()
     client = cert_of(state.client_cert)

tests/test_deploy_scripts.py

@@ -1,4 +1,4 @@
-"""Contract tests for the deployment scripts in `deploy/`.
+"""Contract tests for the scripts an operator runs: `deploy/` and `scripts/dev-*.sh`/`.py`.
 
 There is no shell test framework here — and adding one would buy nothing a subprocess does
 not — but phase 10 rests on a sentence that lives in **two** files: the sudoers grant in
@@ -8,25 +8,41 @@ silently, and the failure is an install that cannot happen on a host nobody is s
 So these tests read the file, run the script the way the operator does (unprivileged, which
 exits before the script can touch anything), and require the two lines to be the same line
 apart from the path the script was invoked through.
+
+The same reasoning covers `scripts/dev-certs-export.py` and the `scripts/dev-server.sh` call that
+runs it (phase 24): the exporter is what turns the durable CA into a file an operator can `scp`,
+and the composition — *after* the readiness probe, *as* a warning line — is as much of the
+contract as the bytes it writes. So both files are run here and read here.
 """
 
 from __future__ import annotations
 
 import difflib
+import importlib.util
 import json
 import os
 import re
 import shutil
 import stat
 import subprocess
+import sys
 import threading
-from http.server import ThreadingHTTPServer
+import time
+from collections.abc import Iterator
+from contextlib import contextmanager
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 from pathlib import Path
 from typing import Any
 
+import httpx
 import pytest
+import uvicorn
 
-from dashboards import certs
+from auth_contract import ADMIN_PASSWORD, FAST_PBKDF2
+from dashboards import auth, certs
+from dashboards.app import create_app
+from dashboards.config import Settings, Target
+from dashboards.store import SampleStore
 
 ROOT = Path(__file__).resolve().parent.parent
 DEPLOY = ROOT / "deploy"
@@ -227,6 +243,56 @@ def test_no_operational_file_mentions_the_tunnel_machinery() -> None:
     assert not offenders, "tunnel-era names/ports are back:\n" + "\n".join(offenders)
 
 
+# --- the mTLS registration is the documented default (phase 24) ---------------------------
+
+#: The three flags that make a host's **install** its registration moment: the run that installs
+#: the agent also pulls that host's bundle from the dashboard that owns the authority. The
+#: token-mode install is still supported (a host nobody has bundled keeps its bearer token), but
+#: it is the compatibility path now, and a setup file that leads with it is a recipe for a fleet
+#: of `http://` agents — which is the exact state `TODO.md:3` asked to be rid of.
+REGISTRATION_FLAGS = ("--fetch-bundle", "--cert-name", "--session")
+
+#: The files that tell an operator what to type, per host.
+REGISTRATION_DOCS = ("README.md", "AGENTS.md", ".env.example", ".env.dev.example")
+
+
+def _host_target_lines(path: Path) -> list[str]:
+    """Every `HOST_TARGETS` assignment line in a file, comments included.
+
+    The commented example in `.env.example` counts, because it is the line a reader copies.
+    """
+    return [line for line in path.read_text().splitlines()
+            if re.search(r"^\s*#?\s*(export\s+)?HOST_TARGETS\s*=", line)]
+
+
+@pytest.mark.parametrize("name", REGISTRATION_DOCS)
+def test_the_setup_docs_teach_registration_with_a_bundle(name: str) -> None:
+    """The per-host recipe a reader copies is the bundle fetch, not the bearer token."""
+    text = (ROOT / name).read_text()
+    for flag in REGISTRATION_FLAGS:
+        assert flag in text, f"{name} stopped naming {flag} — the fetch is no longer documented"
+
+
+@pytest.mark.parametrize("name", (".env.example", ".env.dev.example"))
+def test_the_template_target_urls_are_https_and_credential_free(name: str) -> None:
+    """A `HOST_TARGETS` line an operator copies is `https://` with no userinfo.
+
+    Pinning the *example* is what keeps this true: prose can and should still document the
+    token-mode shape (it is a supported dial, `collectors.fetch_snapshot` keeps its path byte-for-
+    byte), but the line that gets pasted into a real config must not seed a non-mTLS fleet. The
+    empty `HOST_TARGETS=` is allowed — unconfigured is a different fact from misconfigured.
+    """
+    lines = _host_target_lines(ROOT / name)
+    assert lines, f"{name} lost its HOST_TARGETS documentation entirely"
+    for line in lines:
+        value = line.split("=", 1)[1]
+        urls = re.findall(r"\w+://[^\s\"']+", value)
+        if not urls:
+            continue  # the empty assignment
+        bad = [u for u in urls if not u.startswith("https://") or "@" in u]
+        assert not bad, f"{name} seeds a non-mTLS or userinfo-bearing target: {bad}"
+
+
 # --- shell syntax (the only gate these scripts get) -------------------------------------
 
 
@@ -1544,3 +1610,727 @@ def test_a_token_mode_status_probe_stays_plain_http(tmp_path: Path) -> None:
     assert _installer_status(result.stdout).startswith("(no answer"), (
         "token mode must probe plain http — nothing listens here, and the refusal is the point"
     )
+
+
+# --- scripts/dev-certs-export.py (phase 24 task 01) ---------------------------------------
+
+#: The dev exporter, and the dev server that calls it. Both are **run** below and **read**
+#: below: the bytes a host installs are the contract of the first, and the composition of the
+#: second (export *after* the readiness probe, failure *as* a warning) is the contract of the
+#: dev loop that has to stop producing `http://` agents.
+EXPORTER = ROOT / "scripts" / "dev-certs-export.py"
+DEV_SERVER = ROOT / "scripts" / "dev-server.sh"
+
+#: The invariant-3 pair, as two configured hosts: one with a bundle in `meta`, one without.
+ISSUED_HOST = "issuedhost"
+UNISSUED_HOST = "unissuedhost"
+
+
+def _load_exporter() -> Any:
+    """The exporter as a module — it is a script, not a package member, so import it by path."""
+    spec = importlib.util.spec_from_file_location("dev_certs_export", EXPORTER)
+    assert spec is not None and spec.loader is not None
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+@contextmanager
+def served_dashboard(tmp_path: Path) -> Iterator[tuple[str, SampleStore]]:
+    """A real `create_app(...)` on a real socket, with one host's bundle already issued.
+
+    uvicorn in a thread rather than `TestClient`, because the unit under test is an **HTTP
+    client**: the exporter signs in over the public route and reads the session token back out
+    of a `Set-Cookie` header, and a transport-level test client can pass that without ever
+    serialising a header. Port 0, so nothing here can collide with a port another test bound.
+
+    The store is injected and closed here, and the two `HOST_TARGETS` entries point at a port
+    nothing listens on: the app under test must never dial, and a dead target is what keeps
+    this hermetic. Every other source is switched off explicitly, so an operator running the
+    suite with `.env.dev` sourced still gets a test of two hosts and no fleet.
+    """
+    db = tmp_path / "export.sqlite3"
+    store = SampleStore(db)
+    state = certs.ensure_certs(store)
+    certs.issue_agent_cert(store, state, ISSUED_HOST, san=[ISSUED_HOST])
+    settings = Settings(
+        db_path=str(db),
+        admin_password=ADMIN_PASSWORD,
+        pbkdf2_iterations=FAST_PBKDF2,
+        host_targets=(Target(name=ISSUED_HOST, url="http://127.0.0.1:1"),
+                      Target(name=UNISSUED_HOST, url="http://127.0.0.1:1")),
+        host_interval=3600,
+        host_timeout=0.2,
+        gufo_url="",
+        vllm_url="",
+        nut_host="",
+    )
+    server = uvicorn.Server(uvicorn.Config(create_app(settings, store), host="127.0.0.1",
+                                           port=0, log_level="error", lifespan="on"))
+    thread = threading.Thread(target=server.run, daemon=True, name="dev-certs-export-test")
+    thread.start()
+    try:
+        for _ in range(2_000):
+            if server.started:
+                break
+            time.sleep(0.01)
+        assert server.started, "the test dashboard never came up (see the captured stderr)"
+        port = int(server.servers[0].sockets[0].getsockname()[1])
+        yield f"http://127.0.0.1:{port}", store
+    finally:
+        server.should_exit = True
+        thread.join(timeout=60)
+        store.close()
+
+
+def _export(base: str, out_dir: Path, names: list[str], *,
+            password: str | None = ADMIN_PASSWORD) -> subprocess.CompletedProcess[str]:
+    """Run the exporter exactly the way `dev-server.sh` does, as a subprocess."""
+    args = [sys.executable, str(EXPORTER), "--url", base, "--out-dir", str(out_dir)]
+    if password is not None:
+        args += ["--password", password]
+    if names:
+        args += ["--names", *names]
+    return _run(args)
+
+
+def _served_bundle(base: str, name: str) -> bytes:
+    """What `GET /api/certs/<name>/bundle` answers, fetched the way the exporter fetches it."""
+    with httpx.Client(base_url=base, timeout=10.0) as client:
+        login = client.post("/api/auth/login", json={"password": ADMIN_PASSWORD})
+        login.raise_for_status()
+        token = login.cookies[auth.SESSION_COOKIE]
+        answer = client.get(f"/api/certs/{name}/bundle",
+                            headers={"authorization": f"Bearer {token}"})
+        answer.raise_for_status()
+        return answer.content
+
+
+def test_the_export_writes_the_issued_bundle_byte_identical_at_0600(tmp_path: Path) -> None:
+    """The task's headline: the file is the stored bundle, at a mode that keeps it private.
+
+    Byte-identity is compared against the **route** rather than the `meta` row because the
+    question this answers is "is the file I am about to `scp` the file the dashboard issued?",
+    and a bundle that re-signed between the two would make that question unanswerable.
+    """
+    with served_dashboard(tmp_path) as (base, _store):
+        out = tmp_path / "certs"
+        result = _export(base, out, [ISSUED_HOST])
+        assert result.returncode == 0, result.stderr
+        path = out / f"{ISSUED_HOST}.bundle"
+        assert path.is_file(), f"no bundle exported; stdout was:\n{result.stdout}"
+        assert stat.S_IMODE(path.stat().st_mode) == 0o600, (
+            "a bundle carries a private key; 0600 is its birth mode and stays its mode"
+        )
+        assert path.read_bytes() == _served_bundle(base, ISSUED_HOST)
+        assert ISSUED_HOST in result.stdout and "0600" in result.stdout
+
+
+def test_an_unissued_host_writes_no_file_and_names_the_certs_route(tmp_path: Path) -> None:
+    """`issued: false` is a fact, not a failure: one line, no file, exit 0 (invariant 3)."""
+    with served_dashboard(tmp_path) as (base, _store):
+        out = tmp_path / "certs"
+        result = _export(base, out, [ISSUED_HOST, UNISSUED_HOST])
+        assert result.returncode == 0, result.stderr
+        assert not (out / f"{UNISSUED_HOST}.bundle").exists()
+        note = next(line for line in result.stdout.splitlines() if UNISSUED_HOST in line)
+        assert "no bundle issued" in note and "/certs" in note, note
+        assert "!" not in note, f"an unissued host must not read as a failure: {note}"
+
+
+def test_a_name_outside_host_targets_is_a_note_and_not_a_crash(tmp_path: Path) -> None:
+    """The `{"configured": false}` arm: impossible from `HOST_TARGETS`, still never a crash."""
+    with served_dashboard(tmp_path) as (base, _store):
+        out = tmp_path / "certs"
+        result = _export(base, out, ["nosuchhost"])
+        assert result.returncode == 0, result.stderr
+        assert not out.exists(), "nothing was issued, so nothing — not even a directory"
+        assert "HOST_TARGETS" in result.stdout
+
+
+def test_a_wrong_password_exits_non_zero_and_writes_nothing(tmp_path: Path) -> None:
+    """A dev server that cannot be signed in to says so loudly (and the CA stays untouched)."""
+    with served_dashboard(tmp_path) as (base, _store):
+        out = tmp_path / "certs"
+        result = _export(base, out, [ISSUED_HOST], password="definitely-not-the-password")
+        assert result.returncode == 2, f"a sign-in failure must not look like success: {result}"
+        assert not out.exists()
+        assert "sign-in" in result.stderr.lower()
+
+
+def test_no_password_at_all_is_a_refusal_and_not_a_hang(tmp_path: Path) -> None:
+    """`--password` is optional only because `$DASH_ADMIN_PASSWORD` is the dev source of it."""
+    env = {key: value for key, value in os.environ.items() if key != "DASH_ADMIN_PASSWORD"}
+    result = subprocess.run(
+        [sys.executable, str(EXPORTER), "--url", "http://127.0.0.1:1",
+         "--out-dir", str(tmp_path / "certs"), "--names", ISSUED_HOST],
+        capture_output=True, text=True, check=False, env=env, timeout=60,
+    )
+    assert result.returncode == 2
+    assert "DASH_ADMIN_PASSWORD" in result.stderr
+    assert not (tmp_path / "certs").exists()
+
+
+def test_no_names_at_all_is_a_quiet_zero(tmp_path: Path) -> None:
+    """A dashboard with no `HOST_TARGETS` has nothing to export and is not broken."""
+    result = _export("http://127.0.0.1:1", tmp_path / "certs", [])
+    assert result.returncode == 0, result.stderr
+    assert "nothing to export" in result.stdout
+    assert not (tmp_path / "certs").exists()
+
+
+def test_a_second_export_is_idempotent_and_says_unchanged(tmp_path: Path) -> None:
+    """`restart` runs this every time; a dev loop that rewrote keys on every boot would be noise."""
+    with served_dashboard(tmp_path) as (base, _store):
+        out = tmp_path / "certs"
+        first = _export(base, out, [ISSUED_HOST])
+        assert first.returncode == 0, first.stderr
+        path = out / f"{ISSUED_HOST}.bundle"
+        before = path.read_bytes()
+        path.chmod(0o644)  # somebody loosened it; the export is what puts the mode back
+        second = _export(base, out, [ISSUED_HOST])
+        assert second.returncode == 0, second.stderr
+        assert path.read_bytes() == before, "the stored bundle must not move between exports"
+        assert stat.S_IMODE(path.stat().st_mode) == 0o600, "an existing file is re-chmodded"
+        assert "unchanged" in second.stdout
+
+
+def test_the_export_reads_the_authority_and_never_mints_anything(tmp_path: Path) -> None:
+    """The difference between `read_state` and `ensure_certs`, pinned as an observable.
+
+    "The dev server should generate certs" is the TODO's wording; the safe reading is *export
+    what the first boot minted*. If this script ever issued, a stray `restart` would rotate a
+    fleet's identities — so the serials, the revisions and the root all have to be the same
+    bytes after the run as before it.
+    """
+    with served_dashboard(tmp_path) as (base, store):
+        def fingerprint() -> tuple[str, ...]:
+            listed = certs.read_state(store)
+            rows = tuple(f"{e.name}:{e.serial}:{e.bundle_rev}:{e.status}"
+                         for e in certs.read_agents(store))
+            return (f"root:{listed.root_serial}", *rows)
+
+        before = fingerprint()
+        assert _export(base, tmp_path / "certs",
+                       [ISSUED_HOST, UNISSUED_HOST, "nosuchhost"]).returncode == 0
+        assert fingerprint() == before, "the exporter changed the certificate authority"
+
+
+def test_the_export_prints_no_key_material_and_no_password(tmp_path: Path) -> None:
+    """Invariant 6, on the one path that holds both a session token and a private key."""
+    with served_dashboard(tmp_path) as (base, _store):
+        result = _export(base, tmp_path / "certs", [ISSUED_HOST, UNISSUED_HOST])
+        printed = result.stdout + result.stderr
+        assert "BEGIN" not in printed, "a PEM block reached the terminal (and the dev log)"
+        assert ADMIN_PASSWORD not in printed
+        assert auth.SESSION_COOKIE not in printed, "the token itself must never be printed"
+
+
+def test_the_exporter_splits_names_the_way_host_targets_looks() -> None:
+    """One argument with spaces in it, three arguments, commas: all the same list."""
+    module = _load_exporter()
+    assert module.split_names(["alpha beta", "gamma,delta"]) == ["alpha", "beta", "gamma", "delta"]
+    assert module.split_names(["  alpha,,beta  "]) == ["alpha", "beta"]
+    assert module.split_names([]) == []
+    assert module.split_names(None) == []
+
+
+def test_the_exporter_reads_the_cookie_the_app_sets() -> None:
+    """`auth.SESSION_COOKIE` is imported, not restated — a rename must not silently break dev."""
+    module = _load_exporter()
+    assert module.SESSION_COOKIE == auth.SESSION_COOKIE
+
+
+def test_write_bundle_refuses_a_name_that_would_leave_the_directory(tmp_path: Path) -> None:
+    """The route gates names long before this, but this is the function that picks a path."""
+    module = _load_exporter()
+    for bad in ("../escape", "nested/x", "/absolute", "", "."):
+        with pytest.raises(module.ExportError):
+            module.write_bundle(tmp_path, bad, b"secret")
+    assert list(tmp_path.iterdir()) == []
+
+
+def test_the_bundles_only_ever_land_under_the_gitignored_data_tree() -> None:
+    """No key, bundle or PEM outside `data/`: the exporter's default and the caller's flag."""
+    module = _load_exporter()
+    assert module.DEFAULT_OUT_DIR == ROOT / "data" / "certs"
+    gitignore = (ROOT / ".gitignore").read_text()
+    assert re.search(r"^data/$", gitignore, re.MULTILINE), "data/ is the gitignored tree"
+    dev_server = DEV_SERVER.read_text()
+    assignments = [line for line in dev_server.splitlines() if line.startswith("CERTDIR=")]
+    assert assignments == ['CERTDIR="$ROOT/data/certs"'], (
+        f"the dev server must name exactly one export dir, under data/: {assignments}"
+    )
+    assert '--out-dir "$CERTDIR"' in dev_server, "the export must be told where it writes"
+
+
+def _shell_function(text: str, name: str) -> str:
+    """One function's body, so a pin can talk about *that* block and not the whole file."""
+    match = re.search(rf"^{re.escape(name)}\(\) \{{$", text, re.MULTILINE)
+    assert match is not None, f"{name}() is no longer a plain function in dev-server.sh"
+    body = text[match.end():]
+    end = body.index("\n}")
+    return body[:end]
+
+
+def test_dev_server_exports_the_bundles_after_the_readiness_probe() -> None:
+    """The order is the whole safety of the dev loop: no bundle export against a dead server."""
+    text = DEV_SERVER.read_text()
+    probe = text.index("curl -sf -o /dev/null")
+    call = text.index('export_bundles "$PORT"')
+    assert probe < call, "the export must run only once /api/health has answered"
+    assert "dev-certs-export.py" in _shell_function(text, "export_bundles")
+    body = _shell_function(text, "export_bundles")
+    assert "http://127.0.0.1:${port}" in body, "the exporter dials the dev server and nothing else"
+    # The names are HOST_TARGETS' names, split by the caller: the exporter re-parses no config.
+    assert "${entry%%=*}" in body
+
+
+def test_a_failed_bundle_export_never_fails_the_dev_server_start() -> None:
+    """A dev-convenience failure is a warning line: the readiness probe already proved the app."""
+    body = _shell_function(DEV_SERVER.read_text(), "export_bundles")
+    assert "|| echo" in body, "a failed export must print a warning rather than propagate"
+    for escape in ("return 1", "exit 1", "set -e"):
+        assert escape not in body, f"the export can now abort `start` ({escape})"
+
+
+def test_dev_server_status_prints_the_exported_bundles_block() -> None:
+    """`status` answers "is this host's bundle there yet?" without a signed-in curl."""
+    text = DEV_SERVER.read_text()
+    body = _shell_function(text, "bundles")
+    assert "bundles exported" in body and ".bundle" in body
+    assert "mode" in body, "the block shows the 0600, which is the thing worth checking"
+    assert re.search(r"^    bundles ;;$", text, re.MULTILINE), "`status` must print the block"
+    assert "bundles)" in text, "`bundles` is a subcommand of its own"
+
+
+def test_a_dashboard_that_never_answers_is_named_rather_than_guessed(tmp_path: Path) -> None:
+    """Nothing listened: the line says which URL, the exit is non-zero, nothing is written."""
+    result = _export("http://127.0.0.1:1", tmp_path / "certs", [ISSUED_HOST])
+    assert result.returncode == 2
+    assert "no dashboard answered" in result.stderr and "127.0.0.1:1" in result.stderr
+    assert not (tmp_path / "certs").exists()
+
+
+def test_an_unexpected_answer_on_the_bundle_route_is_not_read_as_unissued(tmp_path: Path) -> None:
+    """Only the two 404 shapes are notes. A 401 means the caller is not authorised, which is news."""
+    with served_dashboard(tmp_path) as (base, _store):
+        module = _load_exporter()
+        with pytest.raises(module.ExportError) as raised:
+            module.fetch_bundle(base, ISSUED_HOST, "not-a-session-token")
+        assert "401" in str(raised.value)
+
+
+def test_one_host_the_route_refuses_never_stops_the_others(tmp_path: Any,
+                                                           monkeypatch: pytest.MonkeyPatch) -> None:
+    """The aggregation `main` owns: every name is attempted, and a failure moves the exit code."""
+    module = _load_exporter()
+    attempted: list[str] = []
+
+    def fake_fetch(base: str, name: str, token: str) -> bytes | str:
+        attempted.append(name)
+        if name == "bad":
+            raise module.ExportError("bad: GET /api/certs/bad/bundle answered 500: boom")
+        return b"bundle-bytes"
+
+    monkeypatch.setattr(module, "sign_in", lambda base, password: "a-token")
+    monkeypatch.setattr(module, "fetch_bundle", fake_fetch)
+    code = module.main(["--url", "http://127.0.0.1:1", "--password", "p",
+                        "--names", "good bad later", "--out-dir", str(tmp_path / "certs")])
+    assert attempted == ["good", "bad", "later"], "one refused host must not end the run"
+    assert code == 1, "a bundle nobody exported is a host that stays on http:// — say so"
+    assert (tmp_path / "certs" / "good.bundle").is_file()
+    assert not (tmp_path / "certs" / "bad.bundle").exists()
+
+
+# --- fetching the bundle during registration (phase 24 task 02) ---------------------------
+
+#: `install-agent.sh --fetch-bundle <url> --cert-name <name> --session <token>`: the install
+#: run pulls its own bundle from the dashboard, so a host is never registered onto a token URL
+#: because an operator forgot a `curl`. The contract has three parts — the bytes that land are
+#: the bytes the route serves, a refusal writes *nothing*, and the session token is a fetch
+#: credential that appears in no artefact the run leaves behind.
+FETCH_UNISSUED = "no bundle issued for"
+FETCH_ALL_OR_NOTHING = "--fetch-bundle, --cert-name and --session go together"
+FETCH_TWO_SOURCES = "two bundle sources"
+FETCH_MODE_REFUSAL = "--fetch-bundle needs an install"
+
+
+def _sign_in(base: str) -> str:
+    """A real session token for the served dashboard — the credential `--session` carries."""
+    with httpx.Client(base_url=base, timeout=10.0) as client:
+        login = client.post("/api/auth/login", json={"password": ADMIN_PASSWORD})
+        login.raise_for_status()
+        return login.cookies[auth.SESSION_COOKIE]
+
+
+def _fetch_run(tmp_path: Path, *flags: str,
+               ) -> tuple[subprocess.CompletedProcess[str], Path, Path]:
+    """One sandboxed install with a sandbox `TMPDIR`, so the fetched temp file is observable.
+
+    The fetch writes to `${TMPDIR:-/tmp}`, so pointing TMPDIR into the sandbox is what turns
+    "the fetched private key never stays behind" into an assertion instead of a hope — and it
+    catches the temp file on the refusal paths too, where nothing else was written to look at.
+    """
+    script, call_log, env = _sandbox_installer(tmp_path)
+    tmpdir = tmp_path / "tmp"
+    tmpdir.mkdir(parents=True, exist_ok=True)
+    result = _install(script, {**env, "TMPDIR": str(tmpdir)}, *flags)
+    return result, tmpdir, call_log
+
+
+def _fetch(base: str, name: str, token: str) -> list[str]:
+    """The three fetch flags, spelled once."""
+    return ["--fetch-bundle", base, "--cert-name", name, "--session", token]
+
+
+@contextmanager
+def served_body(*, status: int = 200, body: bytes = b"wrong",
+                media_type: str = "text/html") -> Iterator[str]:
+    """A loopback port that answers every GET with one canned body.
+
+    Only for the shapes the real route cannot produce: a proxy's 200 error page, a 200 with no
+    bytes at all. The issued host, the unissued host and a refused session are all real answers
+    from `served_dashboard`, and those are the legs that matter most.
+    """
+
+    class Handler(BaseHTTPRequestHandler):
+        def do_GET(self) -> None:
+            self.send_response(status)
+            self.send_header("content-type", media_type)
+            self.send_header("content-length", str(len(body)))
+            self.end_headers()
+            if body:
+                self.wfile.write(body)
+
+        def log_message(self, *_args: Any, **_kwargs: Any) -> None:
+            pass  # the installer's answer is the assertion, not this server's access log
+
+    httpd = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
+    threading.Thread(target=httpd.serve_forever, daemon=True).start()
+    try:
+        yield f"http://127.0.0.1:{httpd.server_address[1]}"
+    finally:
+        httpd.shutdown()
+        httpd.server_close()
+
+
+@pytest.mark.skipif(hasattr(os, "geteuid") and os.geteuid() == 0, reason="running as root")
+def test_a_fetched_bundle_installs_byte_identical_at_0640(tmp_path: Path) -> None:
+    """The registration leg: fetch, validate, install, TLS unit — one run, no hand-carried file.
+
+    Byte-identity is against the same route's answer as task 01's export, because the claim the
+    installer makes is the same one: the file on the host *is* the row in `meta`. Everything
+    after the fetch is the existing `--cert-bundle` path, so the mode, the unit's TLS lines and
+    the https hint are asserted here as unchanged rather than as new.
+    """
+    with served_dashboard(tmp_path) as (base, _store):
+        token = _sign_in(base)
+        result, tmpdir, call_log = _fetch_run(
+            tmp_path, "--bind", "10.0.0.5", "--port", "9120", *_fetch(base, ISSUED_HOST, token))
+        if result.returncode != 0:
+            pytest.fail(f"the fetch install failed:\n{result.stdout}\n{result.stderr}")
+
+        cert_path = tmp_path / "target" / "certs"
+        assert cert_path.read_bytes() == _served_bundle(base, ISSUED_HOST)
+        mode = stat.S_IMODE(cert_path.stat().st_mode)
+        assert mode == 0o640, f"bundle mode is {oct(mode)}, expected 0o640 (group = agent user)"
+
+        unit = (tmp_path / "target" / "dashboards-agent.service").read_text()
+        assert f"--cert-bundle {cert_path}" in unit
+        assert "RuntimeDirectory=dashboards-agent" in unit
+        assert 'HOST_TARGETS="hostname=https://10.0.0.5:9120"' in result.stdout
+        assert f"fetched the certificate bundle for {ISSUED_HOST} from {base}" in result.stdout
+        assert "chown root:dashboards-agent" in call_log.read_text()
+        assert list(tmpdir.iterdir()) == [], "the fetched private key outlived the install"
+
+
+@pytest.mark.skipif(hasattr(os, "geteuid") and os.geteuid() == 0, reason="running as root")
+def test_an_unissued_bundle_is_refused_before_anything_is_written(tmp_path: Path) -> None:
+    """`issued: false` is one sentence naming the fix, and the host stays exactly as it was.
+
+    This is the invariant-3 answer on the install path: the route says "configured, nothing
+    issued", and the installer must not read that as a bundle, as a sudoers problem, or as a
+    reason to create a service user. The stub log is the proof nothing privileged ran: `id -u`
+    is the first privileged call a real install makes, and it is absent.
+    """
+    with served_dashboard(tmp_path) as (base, _store):
+        token = _sign_in(base)
+        result, tmpdir, call_log = _fetch_run(
+            tmp_path, "--bind", "10.0.0.5", *_fetch(base, UNISSUED_HOST, token))
+        assert result.returncode == 2, f"a refused fetch keeps exit 2: {result.returncode}"
+        assert result.stdout == "", "a refusal is a failure: nothing on stdout"
+        assert FETCH_UNISSUED in result.stderr and UNISSUED_HOST in result.stderr
+        assert "/certs" in result.stderr, "the sentence names where the certificate is issued"
+        assert GRANT_PREFIX not in result.stderr, "a refusal is not a sudoers problem"
+        assert not call_log.exists(), "refused before the root boundary: no privileged command ran"
+        assert not (tmp_path / "target").exists(), "a refused fetch writes no files at all"
+        assert list(tmpdir.iterdir()) == [], "the temp file is gone even though it was written"
+
+
+@pytest.mark.skipif(hasattr(os, "geteuid") and os.geteuid() == 0, reason="running as root")
+def test_a_name_the_dashboard_does_not_dial_is_refused_with_its_own_404(tmp_path: Path) -> None:
+    """`{"configured": false}` is a third fact, and it is not read as "nothing issued".
+
+    The route refuses a name outside `HOST_TARGETS` before any signing, so a typo in
+    `--cert-name` must say *that* — an operator who is told "no bundle issued" would go looking
+    for a certificate that was never supposed to exist for a host this dashboard does not dial.
+    """
+    with served_dashboard(tmp_path) as (base, _store):
+        token = _sign_in(base)
+        result, tmpdir, call_log = _fetch_run(
+            tmp_path, "--bind", "10.0.0.5", *_fetch(base, "nosuchhost", token))
+        assert result.returncode == 2
+        assert "404" in result.stderr, result.stderr
+        assert FETCH_UNISSUED not in result.stderr, "not configured is not the same as unissued"
+        assert "HOST_TARGETS" in result.stderr, "the dashboard's own sentence is echoed"
+        assert not call_log.exists()
+        assert list(tmpdir.iterdir()) == []
+
+
+@pytest.mark.skipif(hasattr(os, "geteuid") and os.geteuid() == 0, reason="running as root")
+def test_a_refused_session_names_the_http_status(tmp_path: Path) -> None:
+    """A stale or wrong session is a 401 on the wire, and the run says 401.
+
+    Not "no bundle" and not a sudoers hint: the operator's move is a fresh sign-in, and the
+    only way they learn that is the status the dashboard actually answered.
+    """
+    with served_dashboard(tmp_path) as (base, _store):
+        result, tmpdir, call_log = _fetch_run(
+            tmp_path, "--bind", "10.0.0.5", *_fetch(base, ISSUED_HOST, "not-a-session-token"))
+        assert result.returncode == 2
+        assert "401" in result.stderr, result.stderr
+        assert "sign in required" in result.stderr, "the dashboard's own sentence is echoed"
+        assert not call_log.exists()
+        assert not (tmp_path / "target").exists()
+        assert list(tmpdir.iterdir()) == []
+
+
+@pytest.mark.skipif(hasattr(os, "geteuid") and os.geteuid() == 0, reason="running as root")
+@pytest.mark.parametrize(
+    ("status", "body", "reason"),
+    [
+        (200, b"<html><body>502 from the proxy</body></html>", "# BUNDLE-REV"),
+        (200, b"", "empty body"),
+        (500, b"upstream exploded", "500"),
+    ],
+    ids=["proxy-error-page", "empty-200", "server-error"],
+)
+def test_a_body_that_is_not_a_bundle_never_becomes_an_install(
+    tmp_path: Path, status: int, body: bytes, reason: str
+) -> None:
+    """The shapes the dashboard cannot answer with — and the check that catches each one.
+
+    A 200 with a proxy's HTML page is the dangerous one: it is a *success* to any client that
+    only looks at the status, and installing it produces a unit that exits at startup. It falls
+    through to the existing shape check, which is the reason one validator serves both bundle
+    sources. The refusal names `--fetch-bundle` (the flag that was used), not the temp file.
+    """
+    with served_body(status=status, body=body) as base:
+        result, tmpdir, call_log = _fetch_run(
+            tmp_path, "--bind", "10.0.0.5", *_fetch(base, ISSUED_HOST, "a-token"))
+        assert result.returncode == 2, f"{status}/{len(body)} bytes must be refused"
+        assert reason in result.stderr, result.stderr
+        assert "--fetch-bundle" in result.stderr
+        assert not call_log.exists()
+        assert not (tmp_path / "target").exists()
+        assert list(tmpdir.iterdir()) == []
+
+
+@pytest.mark.skipif(hasattr(os, "geteuid") and os.geteuid() == 0, reason="running as root")
+def test_a_dashboard_that_never_answers_stops_the_fetch(tmp_path: Path) -> None:
+    """Nothing listening on the dashboard URL: the line names it, and no install happens."""
+    result, tmpdir, call_log = _fetch_run(
+        tmp_path, "--bind", "10.0.0.5",
+        *_fetch("http://127.0.0.1:1", ISSUED_HOST, "a-token"))
+    assert result.returncode == 2
+    assert "no dashboard answered" in result.stderr and "127.0.0.1:1" in result.stderr
+    assert not call_log.exists()
+    assert list(tmpdir.iterdir()) == []
+
+
+@pytest.mark.skipif(hasattr(os, "geteuid") and os.geteuid() == 0, reason="running as root")
+def test_the_session_token_appears_in_no_written_artefact(tmp_path: Path) -> None:
+    """Completion criterion: the token is a fetch credential, not a stored one.
+
+    The test signs in for a real token and then searches **every** file the run could have
+    touched plus both output streams. The agent's own bearer token is expected in the token
+    file and on stdout — that is a different secret, deliberately printed once — so the pin is
+    specifically that this session value is nowhere.
+    """
+    with served_dashboard(tmp_path) as (base, _store):
+        token = _sign_in(base)
+        assert token and token != ADMIN_PASSWORD
+        result, _tmpdir, call_log = _fetch_run(
+            tmp_path, "--bind", "10.0.0.5", "--port", "9120",
+            *_fetch(base, ISSUED_HOST, token))
+        assert result.returncode == 0, f"install failed:\n{result.stderr}"
+
+        where = tmp_path / "target"
+        assert token not in (where / "token").read_text(), "the token file is the agent's, not ours"
+        assert token not in (where / "certs").read_text(), "a bundle is not a place for a session"
+        assert token not in (where / "dashboards-agent.service").read_text()
+        assert token not in call_log.read_text(), "a privileged call must not carry it either"
+        assert token not in result.stdout + result.stderr
+
+        hits = [str(path.relative_to(tmp_path)) for path in tmp_path.rglob("*")
+                if path.is_file() and token in path.read_text(encoding="utf-8", errors="replace")]
+        assert hits == [], f"the session token was written to {hits}"
+
+
+@pytest.mark.parametrize(
+    "flags",
+    [
+        ["--fetch-bundle", "http://127.0.0.1:8090"],
+        ["--cert-name", "deskwork"],
+        ["--session", "a-token"],
+        ["--fetch-bundle", "http://127.0.0.1:8090", "--cert-name", "deskwork"],
+        ["--cert-name", "deskwork", "--session", "a-token"],
+    ],
+    ids=["url", "name", "token", "url+name", "name+token"],
+)
+@pytest.mark.skipif(hasattr(os, "geteuid") and os.geteuid() == 0, reason="running as root")
+def test_the_fetch_flags_are_all_or_nothing(tmp_path: Path, flags: list[str]) -> None:
+    """Half a fetch is a refusal, not a guess: no URL, no name, or no token means nothing dials.
+
+    A URL without a name has no route to fetch, and a name without a token is a 401 the
+    operator would have to diagnose from a stack of unrelated output. Both are answered with
+    one sentence before the root boundary and before any write — the `--caps` reasoning again.
+    """
+    script, call_log, env = _sandbox_installer(tmp_path)
+    result = _install(script, env, "--bind", "127.0.0.1", "--port", "9199", *flags)
+    assert result.returncode == 1, "a flag mistake keeps exit 1, not the fetch's 2"
+    assert result.stdout == ""
+    assert FETCH_ALL_OR_NOTHING in result.stderr
+    assert GRANT_PREFIX not in result.stderr, "a flag mistake is not a sudoers problem"
+    assert not call_log.exists(), "refused before a single privileged command"
+    assert not (tmp_path / "target").exists()
+
+
+@pytest.mark.skipif(hasattr(os, "geteuid") and os.geteuid() == 0, reason="running as root")
+def test_two_bundle_sources_are_refused_rather_than_ordered(tmp_path: Path) -> None:
+    """`--fetch-bundle` beside `--cert-bundle` is a contradiction: one install, one answer.
+
+    "Which one wins" has no good answer here — a stale file and a fresh bundle are two
+    different identities, and silently preferring either would install a certificate the
+    operator did not mean to install. So the run stops.
+    """
+    bundle = tmp_path / "agent.bundle"
+    bundle.write_text(_fake_bundle_text(), encoding="ascii")
+    script, call_log, env = _sandbox_installer(tmp_path)
+    result = _install(script, env, "--bind", "127.0.0.1", "--port", "9199",
+                      *_fetch("http://127.0.0.1:8090", "deskwork", "a-token"),
+                      "--cert-bundle", str(bundle))
+    assert result.returncode == 1
+    assert FETCH_TWO_SOURCES in result.stderr
+    assert GRANT_PREFIX not in result.stderr
+    assert not call_log.exists()
+    assert not (tmp_path / "target").exists()
+
+
+@pytest.mark.parametrize("mode", ["--status", "--uninstall"], ids=["status", "uninstall"])
+@pytest.mark.skipif(hasattr(os, "geteuid") and os.geteuid() == 0, reason="running as root")
+def test_fetching_a_bundle_is_refused_outside_an_install(tmp_path: Path, mode: str) -> None:
+    """The fetch belongs to the run that installs, or a key is fetched to be thrown away.
+
+    `--status` answers about a port and `--uninstall` removes files; neither has a use for the
+    bytes, and a silent "flags ignored" on the uninstall path would leave an operator believing
+    a bundle had been refreshed.
+    """
+    script, call_log, env = _sandbox_installer(tmp_path)
+    result = _install(script, env, mode,
+                      *_fetch("http://127.0.0.1:8090", "deskwork", "a-token"))
+    assert result.returncode == 1
+    assert FETCH_MODE_REFUSAL in result.stderr
+    assert not call_log.exists(), "and refused before the mode's own privileged work"
+
+
+def test_help_teaches_the_registration_fetch() -> None:
+    """The three flags and the recipe are learnable from `--help`, like every other flag here."""
+    stdout = _run([BASH, str(INSTALLER), "--help"]).stdout
+    assert "--fetch-bundle URL" in stdout
+    assert "--cert-name NAME" in stdout
+    assert "--session TOKEN" in stdout
+    assert "GET <url>/api/certs/<cert-name>/bundle" in stdout
+    assert "fetch credential only" in stdout, "the token's scope has to be learnable, not folklore"
+    recipe = "--fetch-bundle https://dashboards.example.com --cert-name <HOST_TARGETS name>"
+    assert recipe in stdout.replace("\\\n", " ").replace("  ", " "), (
+        "the recipe line must be copy-pasteable from --help"
+    )
+
+
+@pytest.mark.skipif(hasattr(os, "geteuid") and os.geteuid() == 0, reason="running as root")
+def test_the_sudoers_hint_never_repeats_the_session_token(tmp_path: Path) -> None:
+    """The hint quotes a command, so it quotes only the flags that are safe to re-type.
+
+    `need_root` ends with `sudo <this script> …` so the operator can copy it. Handing that
+    string the real argv would print a bearer credential into the terminal, the ssh log and
+    whoever's scrollback — and the fetch runs *before* the root boundary, so this is the one
+    path where a good fetch and a refusal share a run. The fetch succeeding here is also the
+    proof of the order: an unprivileged run can only reach the hint after fetching.
+    """
+    with served_dashboard(tmp_path) as (base, _store):
+        token = _sign_in(base)
+        result = _installer_without_root("--bind", "10.0.0.5", "--port", "9120",
+                                         *_fetch(base, ISSUED_HOST, token))
+        assert result.returncode == 1
+        assert _grant_lines(result.stderr), "the fetch succeeded, so what remains is the grant"
+        assert f"fetched the certificate bundle for {ISSUED_HOST}" in result.stdout
+        assert token not in result.stdout + result.stderr
+        assert "--session" not in result.stderr, "the echoed command must not name the flag either"
+
+
+def test_the_session_token_is_confined_to_the_fetch() -> None:
+    """Structural pin: the install body — everything that writes — cannot see the token.
+
+    `unset SESSION_TOKEN` is the promise; this is the proof that the promise is enforceable
+    rather than aspirational. Every file the installer writes is written inside `do_install`
+    (the token file, the bundle, the unit), so if the name does not appear anywhere in that
+    function or in the unit template, no artefact can carry it — the same style of pin
+    `tests/test_auth.py` and `tests/test_certs.py` hold over the two modules that hold secrets.
+    """
+    text = INSTALLER.read_text()
+    assert 'unset SESSION_TOKEN' in text, "the fetch must forget the credential when it is done"
+    assert 'FETCH_SESSION="$SESSION_TOKEN" python3' in text, (
+        "the token reaches python in the environment: argv is world-readable through `ps`, so a"
+        " second copy on a command line would widen the exposure instead of limiting it"
+    )
+    body = text.split("do_install() {", 1)[1]
+    assert "SESSION_TOKEN" not in body, "nothing that writes may read the fetch credential"
+    assert text.count("SESSION_TOKEN") <= 8, (
+        "the name belongs to the parser, the all-or-nothing check and the fetch; every further"
+        " mention is another place a bearer token could be printed or written"
+    )
+
+
+def test_the_fetch_and_its_shape_check_both_run_before_the_root_boundary() -> None:
+    """Text pin for the order the no-partial-install promise depends on.
+
+    Behaviourally, a refusal as `ops` proves the refusal comes first; this proves the *shape*
+    check is not quietly moved after `need_root` (where a bad fetched bundle would produce a
+    sudoers hint instead of a reason, on the host where the operator can least afford it).
+    """
+    body = INSTALLER.read_text().split("do_install() {", 1)[1]
+    fetch = body.index("fetch_bundle\n")
+    shape = body.index('validate_bundle "$BUNDLE_SRC"')
+    root = body.index("need_root --bind")
+    assert fetch < shape < root, "fetch, then shape, then root — in that order"
+    assert 'trap cleanup_fetch EXIT' in INSTALLER.read_text(), (
+        "the fetched key is cleaned up on every exit path, not only the happy one"
+    )
+
+
+@pytest.mark.skipif(hasattr(os, "geteuid") and os.geteuid() == 0, reason="running as root")
+def test_a_local_bundle_still_validates_with_its_own_flag_name(tmp_path: Path) -> None:
+    """The validator grew a label, and the pre-existing path must read exactly as before.
+
+    `--fetch-bundle` needed the refusal to name the flag that was used; the `--cert-bundle`
+    path must not start naming a flag nobody passed. That is the whole change, so this is the
+    whole pin — the file-local behaviour around it is covered by the phase-15 tests.
+    """
+    path = tmp_path / "revless.bundle"
+    path.write_text(_fake_bundle_text().replace("# BUNDLE-REV 3\n", ""), encoding="ascii")
+    result = _installer_without_root("--cert-bundle", str(path))
+    assert result.returncode == 2
+    assert result.stderr.startswith("--cert-bundle"), result.stderr
+    assert "--fetch-bundle" not in result.stderr

tests/test_live_fleet.py

@@ -22,12 +22,14 @@ from __future__ import annotations
 
 import asyncio
 import os
+import sqlite3
+from collections.abc import Iterator
 
 import httpx
 import pytest
 
 from card_contract import assert_card_contract
-from dashboards import collectors
+from dashboards import certs, collectors
 from dashboards.config import Settings, load
 from dashboards.rates import RateEstimator
 from dashboards.severity import classify
@@ -46,25 +48,88 @@ def settings() -> Settings:
     return load()
 
 
+class _StoredAuthority:
+    """`certs.MetaStore` over an existing database, opened **read-only**.
+
+    The live suite has to dial the fleet the way the app does, and since the phase-24 rollout that
+    means presenting this dashboard's own client certificate. The authority lives in the `meta`
+    table of the store the dev server owns, so this is the smallest honest reader of it: `mode=ro`
+    on the URI, `get_meta` the only method that does anything, and the two write methods a
+    *refusal* rather than a no-op — this file's docstring says nothing in here may write, and
+    minting or repairing an authority from a test would be the worst write it could make.
+    """
+
+    def __init__(self, path: str) -> None:
+        self._conn = sqlite3.connect(f"file:{os.path.abspath(path)}?mode=ro", uri=True)
+
+    def close(self) -> None:
+        self._conn.close()
+
+    def get_meta(self, key: str) -> str | None:
+        row = self._conn.execute("SELECT value FROM meta WHERE key = ?", (key,)).fetchone()
+        return None if row is None else str(row[0])
+
+    def set_meta(self, key: str, value: str) -> None:
+        raise AssertionError(f"live tests are read-only: refused to write {key!r}")
+
+    def delete_meta(self, key: str) -> int:
+        raise AssertionError(f"live tests are read-only: refused to delete {key!r}")
+
+
+@pytest.fixture(scope="module")
+def agent_tls(settings: Settings) -> Iterator[dict[str, collectors.AgentTls]]:
+    """One :class:`collectors.AgentTls` per `https://` host, from the authority in the store.
+
+    Empty when no `HOST_TARGETS` entry is https — every host is then dialled on the pre-phase-15
+    token path, which is what an `http://` fleet still is — and skipped when the store or its
+    `certs.*` rows are absent, because a box running the live suite without the dev database is a
+    normal partial environment rather than a failure (the same rule that lets half a fleet be
+    reachable).
+
+    `certs.read_state` is the request-time read and never `ensure_certs`: a live test may not mint
+    a root CA, and the material this writes is removed on teardown.
+    """
+    targets = settings.host_targets
+    if not any(target.url.startswith("https://") for target in targets):
+        yield {}
+        return
+    if not os.path.exists(settings.db_path):
+        pytest.skip(f"no store at {settings.db_path} to read the CA from")
+    authority = _StoredAuthority(settings.db_path)
+    try:
+        state = certs.read_state(authority)
+    except (RuntimeError, sqlite3.Error) as exc:
+        authority.close()
+        pytest.skip(f"no readable authority in {settings.db_path}: {type(exc).__name__}")
+    material = certs.write_material(state)
+    try:
+        yield collectors.agent_tls_map(material, authority, targets)
+    finally:
+        certs.remove_material(material)
+        authority.close()
+
+
 @pytest.fixture
 def host_collector() -> HostCollector:
     return HostCollector(rates=RateEstimator(span=10.0))
 
 
 async def test_every_host_agent_answers_and_parses(settings: Settings,
-                                                   host_collector: HostCollector) -> None:
+                                                   host_collector: HostCollector,
+                                                   agent_tls: dict[str, collectors.AgentTls]) -> None:
     """Each HOST_TARGETS entry must answer /snapshot and yield the metrics the page renders."""
     if not settings.host_targets:
         pytest.skip("HOST_TARGETS is unset")
     required = {"cpu.usage_pct", "mem.used_pct", "disk.used_pct"}
     async with httpx.AsyncClient(timeout=settings.host_timeout) as client:
         for target in settings.host_targets:
+            tls = agent_tls.get(target.name)
             # Poll twice: usage_pct is a Δ/jiffies rate, and a rate has no second point yet
             # on the first scrape. That is the design (the agent never remembers who polled
             # it), so a live test that scraped once would be asserting a false failure.
-            await collectors.fetch_host_snapshot(client, target, host_collector)
+            await collectors.fetch_host_snapshot(client, target, host_collector, tls=tls)
             await asyncio.sleep(1.2)
-            result = await collectors.fetch_host_snapshot(client, target, host_collector)
+            result = await collectors.fetch_host_snapshot(client, target, host_collector, tls=tls)
             metrics = {sample.metric for sample in result.samples}
             assert result.payload["info"]["hostname"], f"{target.name}: no hostname"
             missing = required - metrics
@@ -76,7 +141,8 @@ async def test_every_host_agent_answers_and_parses(settings: Settings,
 
 
 async def test_every_host_serves_the_hero_card_contract(settings: Settings,
-                                                        host_collector: HostCollector) -> None:
+                                                        host_collector: HostCollector,
+                                                        agent_tls: dict[str, collectors.AgentTls]) -> None:
     """The phase-03 `/system` card paths exist on the **real** agents, not just the captures.
 
     Values may honestly be `n/a` on real hardware (gateway reports no CPU sensor and no GPU
@@ -87,16 +153,18 @@ async def test_every_host_serves_the_hero_card_contract(settings: Settings,
         pytest.skip("HOST_TARGETS is unset")
     async with httpx.AsyncClient(timeout=settings.host_timeout) as client:
         for target in settings.host_targets:
-            await collectors.fetch_host_snapshot(client, target, host_collector)
+            tls = agent_tls.get(target.name)
+            await collectors.fetch_host_snapshot(client, target, host_collector, tls=tls)
             await asyncio.sleep(1.2)
-            result = await collectors.fetch_host_snapshot(client, target, host_collector)
+            result = await collectors.fetch_host_snapshot(client, target, host_collector, tls=tls)
             assert_card_contract(result.payload)
             assert result.payload["cpu"]["temp_c"] == result.payload["temperatures"]["cpu_c"], (
                 f"{target.name}: the card and /api/summary would report different CPU temps"
             )
 
 
-async def test_agent_disk_rows_carry_the_smart_shape(settings: Settings) -> None:
+async def test_agent_disk_rows_carry_the_smart_shape(
+        settings: Settings, agent_tls: dict[str, collectors.AgentTls]) -> None:
     """Phase-09: a redeployed agent answers `smart` per physical disk of `/snapshot`.
 
     Agents are redeployed by hand, so an agent whose rows carry no `smart` key at all
@@ -111,10 +179,16 @@ async def test_agent_disk_rows_carry_the_smart_shape(settings: Settings) -> None
     seen_smart = 0
     async with httpx.AsyncClient(timeout=settings.host_timeout) as client:
         for target in settings.host_targets:
-            response = await client.get(f"{target.url}/snapshot", headers=target.headers)
-            if response.status_code >= 400:
+            try:
+                # through `fetch_snapshot`, not a hand-rolled `get`, because that is the function
+                # that owns the certificate dial: an `https://` agent needs this app's own CA and
+                # client leaf, and a test dialling it with the platform trust store would fail on
+                # a healthy fleet (`CERTIFICATE_VERIFY_FAILED: self-signed certificate in chain`).
+                response = await collectors.fetch_snapshot(client, target,
+                                                           tls=agent_tls.get(target.name))
+            except collectors.UpstreamError:
                 continue  # reachability has its own live test
-            devices = (response.json().get("disk_io") or {}).get("devices") or []
+            devices = (response.get("disk_io") or {}).get("devices") or []
             if devices and not all("smart" in device for device in devices):
                 continue  # agent predates the smart pass — redeploy it to test this
             for device in devices:
@@ -138,7 +212,7 @@ async def test_agent_disk_rows_carry_the_smart_shape(settings: Settings) -> None
 
 
 async def test_disks_source_charts_the_physical_drives_the_agent_reports(
-        settings: Settings) -> None:
+        settings: Settings, agent_tls: dict[str, collectors.AgentTls]) -> None:
     """Phase 09 task 02: the `disks` source over a real agent snapshot.
 
     This is the production code path — the fetch the poller runs, not a hand-built dict —
@@ -153,8 +227,9 @@ async def test_disks_source_charts_the_physical_drives_the_agent_reports(
     async with httpx.AsyncClient(timeout=settings.disks_timeout) as client:
         for target in settings.host_targets:
             try:
-                result = await collectors.fetch_disk_snapshot(client, target, collector,
-                                                              timeout=settings.disks_timeout)
+                result = await collectors.fetch_disk_snapshot(
+                    client, target, collector, timeout=settings.disks_timeout,
+                    tls=agent_tls.get(target.name))
             except collectors.UpstreamError:
                 continue  # reachability has its own live test
             checked += 1