dashboards/scripts/dev-certs-export.py

#!/usr/bin/env python3
"""Materialise the dev dashboard's stored agent bundles in gitignored `data/certs/`.

    .venv/bin/python scripts/dev-certs-export.py \
        --url http://127.0.0.1:8090 --password "$DASH_ADMIN_PASSWORD" \
        --names deskwork gateway --out-dir data/certs

Phase 24 task 01: the dev loop must stop producing `http://` agents, and the first step of
that is making the bundle a host needs *be there* after `./scripts/dev-server.sh restart`
instead of being a `curl` somebody remembers to run.

Two things this deliberately is not.

* **Not a second authority.** The CA is durable in the SQLite `meta` table (`certs.py`), and
  this script only reads the route that serves it (`GET /api/certs/<name>/bundle`), which is
  the request-time read (`certs.read_state`) and never the startup one (`ensure_certs`). So
  "generate certs during dev" means *export what the first boot minted*: running this against
  a dashboard changes no certificate, and a bundle it writes is byte-identical to the row.
* **Not a place a key is committed.** Everything it writes lands under the repo's gitignored
  `data/` tree (the default), and a bundle carries that host's **private key**, so the file is
  created by `os.open` with `0600` as its birth mode and re-chmodded in case it already
  existed looser. Keep `--out-dir` inside `data/`.

Stdlib only (`urllib`), the way `install-agent.sh` reaches for `python3`: this runs on the dev
box against a loopback URL, and a helper that needs the app's own dependency set to sign in
would be a helper that cannot run before `uv sync`. The session token is read out of the
`Set-Cookie` of the public `POST /api/auth/login` and sent back as `Authorization: Bearer` —
the bearer carrier is the one the gate exempts from CSRF, and it never needs a cookie jar.

Exit codes: `0` every name resolved to a file or to the "nothing issued" note, `1` a host the
route refused for some other reason, `2` no sign-in (wrong password, no password, no server).
A sign-in failure is never silent, because the operator's next move is `install-agent.sh`.
"""

from __future__ import annotations

import argparse
import email.message
import json
import os
import re
import sys
import urllib.error
import urllib.parse
import urllib.request
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
#: The non-committed tree `TODO.md:3` asks for. `data/` is in `.gitignore`, so no key this
#: writes can be committed by a scoped `git add` of a phase's work.
DEFAULT_OUT_DIR = ROOT / "data" / "certs"
TIMEOUT_SECONDS = 10.0

try:  # the package owns the cookie name; importing it is what stops that drifting
    from dashboards.auth import SESSION_COOKIE
except ImportError:  # a bare interpreter outside the project venv can still run this
    SESSION_COOKIE = "dash_session"

#: `--names` accepts the shape `HOST_TARGETS` has (space or comma separated) so the caller can
#: hand over what it already parsed instead of this script re-parsing a config it must not
#: disagree with.
_SPLIT_RE = re.compile(r"[,\s]+")


class ExportError(Exception):
    """One sentence for the operator: what failed, and where to look next."""


def split_names(values: list[str] | None) -> list[str]:
    """The host names to export, from one space/comma-separated string or several."""
    return [part for part in _SPLIT_RE.split(" ".join(values or []).strip()) if part]


def _base(url: str) -> str:
    """A base URL with no trailing slash, so joining a route path cannot double one."""
    return url.strip().rstrip("/")


def _json(body: bytes) -> dict[str, object] | None:
    try:
        parsed = json.loads(body.decode("utf-8", "replace"))
    except ValueError:
        return None
    return parsed if isinstance(parsed, dict) else None


def _detail(body: bytes) -> str:
    """The server's own sentence, or the first bytes of whatever it sent instead.

    Every refusal this script can print is one the app writes for an operator
    (`wrong password`, `too many failed sign-ins — try again in 120s`) and carries no
    credential, which is the only reason echoing it is allowed at all (AGENTS.md invariant 6).
    The response body of a *successful* sign-in never reaches the output.
    """
    parsed = _json(body)
    if parsed is not None and isinstance(parsed.get("detail"), str):
        return str(parsed["detail"])
    text = body.decode("utf-8", "replace").strip().replace("\n", " ")
    return text[:160] or "no body"


def _request(
    url: str,
    *,
    method: str = "GET",
    body: bytes | None = None,
    headers: dict[str, str] | None = None,
) -> tuple[int, bytes, email.message.Message | None]:
    """One HTTP call answered as `(status, body, headers)`.

    urllib raises for every 4xx/5xx, and here a 404 **is** the answer: `{"configured": false}`
    and `{"configured": true, "issued": false}` are two different facts about a host. So the
    status comes back instead of an exception, and only a call that never reached the dashboard
    raises.
    """
    request = urllib.request.Request(url, data=body, headers=headers or {}, method=method)
    try:
        with urllib.request.urlopen(request, timeout=TIMEOUT_SECONDS) as response:
            return response.status, response.read(), response.headers
    except urllib.error.HTTPError as exc:  # a refusal from the app: read it, do not raise
        return exc.code, exc.read(), exc.headers
    except (urllib.error.URLError, TimeoutError, OSError) as exc:
        reason = getattr(exc, "reason", None) or exc
        raise ExportError(f"no dashboard answered at {url}: {reason}") from exc


def _cookie(headers: email.message.Message | None, name: str) -> str:
    """The value of one `Set-Cookie` from a response, with no cookie jar."""
    if headers is None:
        return ""
    for line in headers.get_all("set-cookie") or []:
        for chunk in line.split(";"):
            key, sep, value = chunk.strip().partition("=")
            if sep and key == name:
                return value
    return ""


def sign_in(base: str, password: str) -> str:
    """The session token for this password, or an :class:`ExportError`.

    `/api/auth/login` is the one route an anonymous caller may POST (AGENTS.md invariant 12),
    and the token it mints travels **only** in its `Set-Cookie` — never in a body field a page
    could log — so reading it back out of the header is the whole job. A `429` says so with its
    `Retry-After` in the sentence: a dev loop that trips its own throttle should wait, not
    retype.
    """
    status, body, headers = _request(
        f"{base}/api/auth/login",
        method="POST",
        body=json.dumps({"password": password}).encode("utf-8"),
        headers={"content-type": "application/json"},
    )
    if status != 200:
        raise ExportError(f"sign-in at {base} refused ({status}): {_detail(body)}")
    token = _cookie(headers, SESSION_COOKIE)
    if not token:
        raise ExportError(f"sign-in at {base} answered 200 with no {SESSION_COOKIE} cookie")
    return token


def fetch_bundle(base: str, name: str, token: str) -> bytes | str:
    """The bundle **bytes** for an issued host, or the one-line **note** for one with nothing.

    The two 404 shapes are the invariant-3 pair and both mean "write no file": a host with no
    certificate yet, and a name this dashboard does not dial at all. The second cannot happen
    when the caller passes `HOST_TARGETS` names, so it is handled as a line rather than a
    crash — a config typo must not stop the other hosts' bundles.
    """
    route = f"/api/certs/{urllib.parse.quote(name, safe='')}/bundle"
    status, body, _headers = _request(f"{base}{route}",
                                      headers={"authorization": f"Bearer {token}"})
    if status == 200 and body:
        return body
    payload = _json(body)
    if status == 404 and payload is not None:
        if payload.get("configured") is True:
            return (f"{name}: no bundle issued yet — issue one on /certs "
                    f"(POST /api/certs/{name}/rotate mints it)")
        return f"{name}: not in this dashboard's HOST_TARGETS — nothing to export"
    raise ExportError(f"{name}: GET {route} answered {status}: {_detail(body)}")


def export_one(base: str, name: str, token: str, out_dir: Path) -> str:
    """One host's output line: where its bundle went, or why nothing was written.

    Raises :class:`ExportError` for a host the route refused for a reason that is not "nothing
    issued" — the caller counts those and exits non-zero, because a bundle nobody downloaded is
    a host that stays on `http://`.
    """
    result = fetch_bundle(base, name, token)
    if isinstance(result, str):
        return f"  - {result}"
    path, unchanged = write_bundle(out_dir, name, result)
    return (f"  ✓ {name}: {path} ({len(result)} bytes, mode 0600"
            f"{', unchanged' if unchanged else ''})")


def write_bundle(out_dir: Path, name: str, data: bytes) -> tuple[Path, bool]:
    """Write one bundle at `0600`; returns (path, whether the bytes were already there).

    `os.open` rather than `open`, so `0600` is the file's **birth** mode: there is no window in
    which a private key exists on disk at the umask default. The `chmod` after covers the
    already-existing file, for which `O_CREAT`'s mode argument is ignored.
    """
    if not name or Path(name).name != name:
        # Names come from HOST_TARGETS, which the route has already validated, but this is the
        # one function that decides where a key lands on disk. Say it plainly and write nothing.
        raise ExportError(f"{name!r} is not a usable host name — nothing written")
    out_dir.mkdir(mode=0o700, parents=True, exist_ok=True)
    path = out_dir / f"{name}.bundle"
    previous = path.read_bytes() if path.exists() else None
    descriptor = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
    with os.fdopen(descriptor, "wb") as handle:
        handle.write(data)
    os.chmod(path, 0o600)
    return path, previous == data


def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Export the dev dashboard's stored agent bundles to a non-committed "
                    "directory (one file per configured host that has an issued bundle).",
    )
    parser.add_argument("--url", required=True,
                        help="the dashboard's base URL, e.g. http://127.0.0.1:8090")
    parser.add_argument("--password", default=os.environ.get("DASH_ADMIN_PASSWORD", ""),
                        help="the sign-in password (defaults to $DASH_ADMIN_PASSWORD, which "
                             "dev-server.sh already loads from .env.dev)")
    parser.add_argument("--names", nargs="*", default=None, metavar="HOST",
                        help="the configured host names, space or comma separated (from "
                             "HOST_TARGETS — parsed by the caller, never re-parsed here)")
    parser.add_argument("--out-dir", default=str(DEFAULT_OUT_DIR),
                        help=f"where to write <name>.bundle (default: {DEFAULT_OUT_DIR}; keep "
                             "it inside the gitignored data/ tree)")
    return parser.parse_args(argv)


def main(argv: list[str] | None = None) -> int:
    args = parse_args(argv)
    names = split_names(args.names)
    if not names:
        print("  - no host names given (HOST_TARGETS unset?) — nothing to export")
        return 0
    if not args.password:
        print("  ! no sign-in password: pass --password or set DASH_ADMIN_PASSWORD "
              "(dev-server.sh loads it from .env.dev)", file=sys.stderr)
        return 2

    base = _base(args.url)
    try:
        token = sign_in(base, args.password)
    except ExportError as exc:
        print(f"  ! {exc}", file=sys.stderr)
        return 2

    out_dir = Path(args.out_dir)
    failed = 0
    for name in names:
        try:
            print(export_one(base, name, token, out_dir))
        except ExportError as exc:
            print(f"  ! {exc}", file=sys.stderr)
            failed += 1
    return 1 if failed else 0


if __name__ == "__main__":
    sys.exit(main())