#!/usr/bin/env python3
# requires: python >=3.10
"""dashboards-routectl — the route-switch root helper (phase 33, task 01).

Turns one `connections[].id` off or back on in one of the gateway's two proxy configs,
validates the file with the proxy's own checker, reloads that one proxy, and refuses
everything else. This process is the phase's only writer: the dashboard's
`POST /api/gateway/route` reaches a config file only through it (via the agent's
`POST /route`, which reaches this helper only through the systemd-activated socket,
task 02). Run by hand on the gateway it is the operator's rehearsal: `diff` shows
what the dashboard would do, and `disable --id ID` without `--apply` is a dry run.

It never `restart`s anything. A restart severs every live connection through the
fleet's only ingress, and a reload does not, so the only verb this file executes is
`systemctl reload <unit>`; a failed reload leaves the applied file in place and says
so in the answer (`reloaded: false` + the unit's status line) — the config is right,
the reload is the operator's next action.

The one write protocol (the order is the safety property):

1. re-read the file, index it with the agent's parser, map the `id` to the line
   (an unknown id is a refusal that writes nothing);
2. already in the requested state → `{"applied": false, "changed": false}`, no reload;
3. `--apply` absent (or the wire's `dry_run: true`) → answer with the unified `diff`
   only — no write, no checker, no reload, no state change;
4. write the new text to a `mkstemp` in the same directory, `fsync`, `os.replace`;
5. run the proxy's own checker (`nginx -t` / `haproxy -c -f <file>`); on failure
   `os.replace` the previous bytes back and answer
   `{"applied": false, "validated": false, "restored": true}` with the checker's own
   first line — and reload nothing;
6. `systemctl reload <unit>`; a failure is `reloaded: false` with the status line;
7. re-read and re-index the file to prove the marker state is live; record the state.

Per-kind edit (decision 5; a fourth arm is a new decision, not a fall-through):

* a haproxy `server` line is commented with the marker (`#off <name>`): the backend
  stops receiving, and the marker keeps the original restorable by eye and by
  machine;
* a vhost `proxy_pass` line is commented the same way: the vhost keeps `listen` +
  `server_name` and gains no target, a shape the dashboard's warnings already treat
  as quiet, so "off" does not read as "broken";
* a stream `map` entry is **rewritten to the sink address** and marked: commenting
  it would fall through to the map's `default` and mis-route the domain instead of
  blocking it. The sink file is created on first need (the marker's header, a stream
  `server` on `127.0.0.1:44450`), is loaded by the main config's existing
  `stream { include /etc/nginx/stream.d/*.conf; }` glob, and the **main config is
  never edited**; it is removed only when this helper wrote it (its header present)
  and no `map` entry still points at it;
* the `map` `default` entry is **refused** (`"refused": "map-default"`): it is the
  fallback for every unmapped SNI, not a service.

`enable` restores **only** when the current line is byte-identical to the line this
helper wrote (the state sidecar's `written` row, recorded at write time); a hand edit
in the meantime is `{"applied": false, "drift": true}` and no byte changes. The map
arm's original value is not in the file after a disable (it was rewritten), so the
sidecar also carries the `original` line.

Nothing in this file takes a path, a line number, a pattern, or a command from the
wire. `--serve` accepts exactly the keys `{id, action, dry_run}` and every other key
is a refusal; the checker and unit names come from the flags the operator installed,
never from a request; every subprocess call is a fixed argv list (no `shell=True`),
and a `--file` option does not exist. The config parser is the agent's:
`--agent-module` is imported by file path (the helper is installed beside it at
/usr/local/lib/dashboards-agent/) and a helper that cannot import it refuses to run
at all — two parsers reading one config are two opinions about one file.

`--serve` (the socket-activated mode, task 02's `StandardInput=socket`): the chosen
idiom is the connected socket on **stdin** — systemd's `StandardInput=socket` dups
the per-connection socket onto fd 0 (and `StandardOutput=socket` onto fd 1), so the
process wraps fd 0 (or fd 3, the classic socket-activation fd, when the platform
hands the connection there instead — the live box put it on fd 0), checks
`SO_PEERCRED` against `--allow-uid` (the **second** look: the socket file's `0600`
+ `SocketUser` is the first) — the uid is read at the field slot a same-process
probe calibrates on this kernel, because the 12-byte ucred buffer's field order is
not portable (the live box's el10 kernel delivers it as `{pid, uid, gid}`, the
classic layout reversed; in the root service the calibration shifts the probe's
own uid with `setuid` while the gid stays 0 — a privilege drop, restored before
any state file is touched — so a matching gid on a foreign-uid caller cannot
open the gate; an unshifted `euid == egid` probe requires both remaining fields
to match) — reads one newline-delimited JSON request, dispatches
one verb, writes one JSON line back to the socket, and exits. One request per
activation is the whole threading policy: the socket's `Accept=yes` spawns the
template instance per connection, so there is no accept loop, no thread, no timeout
to invent.
"""

import argparse
import contextlib
import datetime
import difflib
import hashlib
import importlib.util
import json
import os
import pwd
import socket
import struct
import subprocess
import sys
import tempfile

# The managed sink: the nginx `stream` block a disabled `map` entry is pointed at. A
# file this helper creates (the marker arms never delete it), a header that identifies
# it as machine-managed, and an address that is the loopback by construction.
SINK_HEADER = "# managed by dashboards-routectl — do not edit"
SINK_FILE_DEFAULT = "/etc/nginx/stream.d/00-dashboards-sink.conf"
SINK_ADDRESS_DEFAULT = "127.0.0.1:44450"

# The `id` grammar (`source:in-port->host:out-port`, both ports required — the
# dashboard's own id always carries them) is bounded by construction so a wire
# body stays a parse-and-refuse matter, not a resource one.
ROUTE_ID_MAX = 256
# `--serve` reads at most this many bytes; more is a malformed refusal.
SERVE_REQUEST_MAX = 65536
# The checker and the reload are both bounded: a wedged `nginx -t` is not an
# indefinite hold on the write path.
COMMAND_TIMEOUT_SECONDS = 60

EXIT_ANSWERED = 0    # the request was answered (it may be a refusal: the JSON says)
EXIT_REFUSED = 1     # a refusal, or the state could not be recorded
EXIT_USAGE = 2       # a flag was used wrongly


def _port(value):
    """The dashboard's own `_port`: an int or None, never a string."""
    if isinstance(value, int):
        return value
    if isinstance(value, str) and value.isdigit():
        return int(value)
    return None


def _directive_part(line):
    """The line up to its end-of-line comment: the agent's tokenizer starts a comment
    at `#`, and the tokens matched here (a `proxy_pass` target, a `map` entry value)
    never contain one."""
    return line.split("#", 1)[0]


def _read(path):
    try:
        with open(path, encoding="utf-8") as handle:
            return handle.read()
    except OSError:
        return None


def _replace_file(path, text):
    """`mkstemp` in the same directory (one filesystem, so the replace is atomic),
    `fsync`, `os.replace`; the temp file is removed on the failure paths."""
    directory = os.path.dirname(path) or "."
    os.makedirs(directory, exist_ok=True)
    fd, tmp = tempfile.mkstemp(prefix=".dashboards-routectl.", dir=directory)
    try:
        with os.fdopen(fd, "w", encoding="utf-8") as handle:
            handle.write(text)
            handle.flush()
            os.fsync(handle.fileno())
        os.replace(tmp, path)
    except BaseException:
        with contextlib.suppress(OSError):
            os.remove(tmp)
        raise


# -- the agent's parser --------------------------------------------------------- #

def load_agent(path):
    """`(module, None)` or `(None, refusal)`: the helper's parser is the agent's, by
    file path. A failure names the path (the operator sees which install is wrong)
    and the exception's *type* — the module's own error text is not this process's
    to print verbatim."""
    if not os.path.isfile(path):
        return None, f"cannot import the agent module {path}: not a file"
    try:
        spec = importlib.util.spec_from_file_location("dashboards_agent_module", path)
        if spec is None or spec.loader is None:
            return None, f"cannot import the agent module {path}: no loader"
        module = importlib.util.module_from_spec(spec)
        spec.loader.exec_module(module)
        return module, None
    except Exception as exc:  # noqa: BLE001 — every import failure is a refusal, by type
        return None, f"cannot import the agent module {path}: {type(exc).__name__}"


def under_root(root, path):
    """Resolve a config path the way the agent's `--root` does: absolute under
    `root`. A path that is already under `root` (an operator's absolute path for
    the sink, say) comes back unchanged — `--root` is a prefix, not a remap."""
    if not path.startswith("/"):
        return path
    root = str(root)
    if root in ("", "/"):
        return path
    if path == root or path.startswith(root.rstrip("/") + "/"):
        return path
    return os.path.join(root, path.lstrip("/"))


# -- the id and the locate ----------------------------------------------------- #

def parse_route_id(route_id):
    """`(source, in_port, host, out_port)` or None: the id's own grammar, nothing else.

    `out_port` is the target's port — for a `map` entry that is the value's port, the
    way the dashboard's claim writes it (a switched-off entry is addressed by the sink
    value it carries, which is how `enable` finds the entry again).
    """
    head, sep, tail = route_id.partition("->")
    if not sep:
        return None
    source, _, in_port_text = head.partition(":")
    if source not in ("haproxy", "nginx") or not in_port_text.isdigit():
        return None
    host, sep2, port_text = tail.rpartition(":")
    if not sep2 or not port_text.isdigit():
        return None
    return source, int(in_port_text), host, int(port_text)


def _haproxy_candidates(agent, text, in_port, host, port):
    """`(candidates, refusal)`: the `server` lines whose target is `host:port`, in a
    backend the frontend on `in_port` reaches (its `default_backend` plus its
    `use_backend` names). A candidate is `(index, name)`. Two backends reaching the
    same named server are two candidates — the caller's `ambiguous` arm, not a
    silent first match."""
    parsed = agent.parse_haproxy_config(text)
    if not parsed.get("recognised"):
        return [], "not-found"
    reachable = set()
    for frontend in parsed.get("frontends", ()):
        ports = [_port(b.get("port")) for b in frontend.get("binds", ())]
        if _port(in_port) not in ports:
            continue
        default = frontend.get("default_backend")
        if default:
            reachable.add(default)
        reachable.update(frontend.get("backends", ()))
    reachable.discard(None)
    candidates = []
    for backend in parsed.get("backends", ()):
        if backend.get("name") not in reachable:
            continue
        for server in backend.get("servers", ()):
            if server.get("target") != host or _port(server.get("port")) != _port(port):
                continue
            name = server.get("name")
            for index, line in enumerate(text.splitlines()):
                words = line.split()
                if len(words) < 3 or words[0] != "server" or words[1] != name:
                    continue
                # the line's own target token, parsed by the agent's own helper — the
                # grammar is the agent's, and a token that does not parse to the id's
                # target is not this line.
                if agent._haproxy_addr_port(words[2]) != (host, _port(port)):
                    continue
                candidates.append((index, name))
    return candidates, None


def _nginx_documents(agent, collector, main_label):
    """`{label: text}`: the main config plus its one level of `include`, the agent's
    way (its bounds, its glob expansion, its labels). None when the main file cannot
    be read: that is `not-found` for every nginx id, not an error."""
    main_text = _read(under_root(collector.root, main_label))
    if main_text is None:
        return None
    errors = []
    includes, _dropped, _unread = collector.nginx_include_files(
        agent.nginx_include_paths(main_text), errors)
    documents = {main_label: main_text}
    documents.update(includes)
    return documents


def _line_matches_token(agent, lines, index, node):
    """True when line `index` carries the node's token: the `proxy_pass <token>`
    whose token the agent's own `_nginx_proxy_pass` parses back to the node's three
    facts (scheme, target, variable). The grammar is the agent's, not this
    file's; the search is text over the lines the parser already parsed."""
    words = _directive_part(lines[index]).strip().split()
    if len(words) < 2 or words[0] != "proxy_pass":
        return False
    # the `;` glues to the token in the file; the agent's tokenizer takes it
    # off, so it comes off before the token goes to the agent's helper.
    token = words[1].rstrip(";")
    pp = agent._nginx_proxy_pass(token, None)
    return (pp.get("scheme") == node.get("scheme")
            and pp.get("target") == node.get("target")
            and pp.get("variable") == node.get("variable"))


def _nginx_candidates(agent, documents, in_port, host, port):
    """`(candidates, refusal)` over the documents: a candidate is
    `(label, index, kind, name)`, kind `vhost` (a `proxy_pass`) or `map` (a `map`
    entry). `refusal` is `map-default`: the id addressed the map's `default`
    entry, which is refused, not located. The line is the *unique* line whose
    token the agent's own helper parses to the id's target — a target written on
    two lines (two vhosts, a repeated key) is `ambiguous`, never a silent first
    match: with no line number on the wire, a guess between two lines is a write
    to the wrong one."""
    candidates = []
    for label, text in documents.items():
        parsed = agent.parse_nginx_config(text, None)
        if parsed.get("error"):
            continue  # the agent drops an unparseable document; it names it in `unparsed`
        lines = text.splitlines()
        for server in parsed.get("servers", ()):
            if (_port(in_port)
                    not in {_port(b.get("port")) for b in server.get("listen", ())}):
                continue
            for node in server.get("proxy_passes", ()):
                variable = node.get("variable")
                if variable:
                    for block in parsed.get("maps", ()):
                        if block.get("variable") != variable:
                            continue
                        for entry in block.get("entries", ()):
                            target = entry.get("target")
                            if target is None:
                                continue
                            if (target.get("host") != host
                                    or _port(target.get("port")) != _port(port)):
                                continue
                            if entry.get("key") == "default":
                                return [], "map-default"
                            index, count = _map_entry_line(lines, entry)
                            if count != 1:
                                return [], f"ambiguous: {entry.get('key')}"
                            candidates.append(
                                (label, index, "map", entry.get("key")))
                else:
                    target = node.get("target")
                    if target is None:
                        continue
                    if (target.get("host") != host
                            or _port(target.get("port")) != _port(port)):
                        continue
                    found = [i for i in range(len(lines))
                            if _line_matches_token(agent, lines, i, node)]
                    if len(found) != 1:
                        return [], f"ambiguous: {_marker_name_for_target(target)}"
                    candidates.append((label, found[0], "vhost",
                                      _marker_name_for_target(target)))
    return candidates, None


def _map_entry_line(lines, entry):
    """`(index, count)` for the entry's line: `<key> <value>;` (or a bare
    `<key> ;` when the value was deleted by hand — the parser then reads a
    `None` target, so that arm is the id's `not-found`, and the restore value is
    the sidecar's, not the file's). `count` is over the whole document: a key
    repeated in one block is the caller's `ambiguous` arm, never a silent first
    match. The value word is what the agent's own `_nginx_addr_port` read
    (host, port), re-joined as written."""
    target = entry.get("target")
    if target is None:
        value = ""
    elif _port(target.get("port")) is not None:
        value = f"{target.get('host')}:{target.get('port')}"
    else:
        value = target.get("host") or ""
    found = []
    for i, line in enumerate(lines):
        words = _directive_part(line).strip().rstrip(";").split()
        if not words or words[0] != entry.get("key"):
            continue
        if len(words) > 1:
            if words[1] == value:
                found.append(i)
        elif value == "":
            found.append(i)
    return (found[0] if found else None), len(found)


def _marker_name_for_target(target):
    """The vhost arm's marker name, pinned by the captured fixture: the target host's
    first label plus the port (`nextcloud.reeselink.com:11000` → `nextcloud-11000`),
    the label alone when the target names no port."""
    host = target.get("host") or ""
    label = host.split(".", 1)[0] or "unnamed"
    port = _port(target.get("port"))
    return f"{label}-{port}" if port is not None else label


# -- the state sidecar --------------------------------------------------------- #

def load_state(path):
    """`(dict, None)` or `(None, refusal)`. A missing file is the empty state; a
    corrupt one is a refusal, not a `{}` — a state this helper cannot read is not a
    state it may write over."""
    if not os.path.exists(path):
        return {}, None
    try:
        with open(path, encoding="utf-8") as handle:
            state = json.load(handle)
    except (OSError, ValueError):
        return None, f"state-corrupt: {path}"
    if not isinstance(state, dict):
        return None, f"state-corrupt: {path}"
    return state, None


def save_state(path, state):
    """The sidecar: written with `os.open(..., O_CREAT, 0o600)` so the mode is its
    birth mode (the `dev-certs-export.py` idiom), in a directory created `0700`."""
    directory = os.path.dirname(path)
    if directory:
        os.makedirs(directory, mode=0o700, exist_ok=True)
    fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
    try:
        with os.fdopen(fd, "w", encoding="utf-8") as handle:
            json.dump(state, handle, indent=1, sort_keys=True)
            handle.write("\n")
    except OSError as exc:
        return False, type(exc).__name__
    return True, None


# -- the one write protocol ---------------------------------------------------- #

class Write:
    """One request's whole life: locate, state, edit, validate, reload, prove,
    record. `ctx` carries the flags (paths, units, the actor); the agent module is
    the parser (and the config-reader's include machinery)."""

    def __init__(self, ctx, agent, source, in_port, host, port, route_id):
        self.ctx = ctx
        self.agent = agent
        self.source = source
        self.in_port = in_port
        self.host = host
        self.port = port
        self.route_id = route_id

    def locate(self, source, in_port, host, port):
        """Step 1: `(route, None)` or `(None, refusal)`. `route` is everything the
        protocol needs: the file label and its text, the line index, the marker
        name, the unit, the checker."""
        ctx = self.ctx
        if source == "haproxy":
            label = ctx.haproxy_config
            text = _read(under_root(ctx.root, label))
            if text is None:
                return None, "not-found"
            candidates, refusal = _haproxy_candidates(
                self.agent, text, in_port, host, port)
            if refusal:
                return None, refusal
            if not candidates:
                return None, "not-found"
            if len(candidates) > 1:
                return None, f"ambiguous: {sorted({c[1] for c in candidates})}"
            index, name = candidates[0]
            return self._route("haproxy", label, {label: text}, index, name,
                              ctx.haproxy_unit, "haproxy"), None
        collector = self.agent.Collector(ctx.root, smart=False, proxy=False,
                                        resolve=False)
        documents = _nginx_documents(self.agent, collector, ctx.nginx_config)
        if documents is None:
            return None, "not-found"
        candidates, refusal = _nginx_candidates(self.agent, documents, in_port,
                                              host, port)
        if refusal:
            return None, refusal
        if not candidates:
            return None, "not-found"
        distinct = {(c[0], c[1]) for c in candidates}
        if len(distinct) > 1:
            return None, f"ambiguous: {sorted({c[3] for c in candidates})}"
        label, index, kind, name = candidates[0]
        return self._route(kind, label, documents, index, name, ctx.nginx_unit,
                          "nginx"), None

    def _route(self, kind, label, documents, index, name, unit, checker):
        return {"kind": kind, "file": label,
                "path": under_root(self.ctx.root, label),
                "documents": documents, "text": documents[label],
                "index": index, "name": name,
                "unit": unit, "checker": checker}

    def current_line(self, route):
        return route["text"].splitlines()[route["index"]]

    def state_of(self, route):
        """Step 2's state: `off` when the line carries the marker, or (the map arm
        only) when the value was re-pointed at the sink by hand — the dashboard's
        `sink_endpoint` rule, where a bare loopback address alone is not off, the
        *address* must be; `on` otherwise. The agent's `split_marker` is the
        reader: the grammar is one."""
        line = self.current_line(route)
        _clean, name = self.agent.split_marker(line)
        if name is not None:
            return "off"
        if route["kind"] == "map":
            words = _directive_part(line).strip().rstrip(";").split()
            if len(words) > 1:
                host, _, port = words[1].rpartition(":")
                sink_host, _, sink_port = self.ctx.sink_address.rpartition(":")
                if host == sink_host and _port(port) == _port(sink_port):
                    return "off"
        return "on"

    def edited_line(self, route, wanted, original):
        """The edited line: `off` sets the marker (the map arm rewrites the value to
        the sink first); `on` strips the marker (the map arm restores the sidecar's
        `original`, because the value was rewritten away)."""
        line = self.current_line(route)
        clean, _name = self.agent.split_marker(line)
        if wanted == "off":
            if route["kind"] == "map":
                indent = line[:len(line) - len(line.lstrip())]
                return "{}{} {}; {}".format(indent, route["name"],
                                       self.ctx.sink_address,
                                       self.agent.marker_for(route["name"]))
            return clean + " " + self.agent.marker_for(route["name"])
        if route["kind"] == "map":
            return original
        return clean

    def validate(self, route):
        """Step 5: the proxy's own checker. `(ok, first_line)`: on failure the
        checker's first non-empty line, bounded — a checker that dumps a stack
        trace is not this process's to reprint. A missing binary is a failure naming
        the binary: a checker this host does not have is not a pass."""
        argv = ["nginx", "-t"] if route["checker"] == "nginx" else ["haproxy", "-c", "-f", route["path"]]
        try:
            proc = subprocess.run(argv, capture_output=True, text=True,
                                 timeout=COMMAND_TIMEOUT_SECONDS)
        except FileNotFoundError:
            return False, f"{argv[0]}: not found"
        except (OSError, subprocess.SubprocessError) as exc:
            return False, f"{argv[0]}: {type(exc).__name__}"
        if proc.returncode == 0:
            return True, ""
        lines = [line_ for line_ in (proc.stdout + "\n" + proc.stderr).splitlines()
                 if line_.strip()]
        return False, (lines[0] if lines else f"checker failed: {argv[0]}")[:200]

    def reload(self, route):
        """Step 6: `systemctl reload <unit>` — the only verb this file executes.
        `(ok, status_line)`: on failure the unit's own first status line, so the
        answer carries the fact and the operator the next action."""
        unit = route["unit"]
        proc = subprocess.run(["systemctl", "reload", unit],
                             capture_output=True, text=True,
                             timeout=COMMAND_TIMEOUT_SECONDS)
        if proc.returncode == 0:
            return True, ""
        status = subprocess.run(["systemctl", "status", unit],
                               capture_output=True, text=True,
                               timeout=COMMAND_TIMEOUT_SECONDS)
        lines = [line_ for line_ in (status.stdout + "\n" + status.stderr).splitlines()
                 if line_.strip()]
        detail = (lines[0] if lines
                 else f"systemctl reload {unit} failed")[:200]
        return False, detail

    def _sink_path(self):
        return under_root(self.ctx.root, self.ctx.sink_conf)

    def _sink_block(self):
        # the captured fixture's shape: a bare `server` block (the file is loaded by
        # the main config's `stream { include …; }` glob, so it must not carry its
        # own `stream` wrapper), no `server_name`, no `proxy_pass` — the block, not
        # the port, is what `sink_endpoint()` recognises.
        return "\n".join([
            SINK_HEADER,
            "server {",
            f"    listen {self.ctx.sink_address};",
            "}",
        ]) + "\n"

    def ensure_sink(self):
        """The sink before the checker runs (a `map` entry rewritten to the sink is a
        reference the checker must be able to resolve): create it, or refuse a file
        that exists without the header — an operator's file is not this helper's to
        overwrite."""
        path = self._sink_path()
        existing = _read(path)
        if existing is not None:
            lines = existing.splitlines()
            if not lines or lines[0] != SINK_HEADER:
                return False, f"sink-file-not-ours: {self.ctx.sink_conf}"
            return True, None
        _replace_file(path, self._sink_block())
        return True, None

    def remove_sink_if_empty(self, documents):
        """After an `enable`: remove the sink only when this helper wrote it (its
        header is present) and no `map` entry still points at it. A file without
        the header is left alone, quietly: it is not this helper's."""
        path = self._sink_path()
        if not os.path.exists(path):
            return
        text = _read(path)
        lines = (text or "").splitlines()
        if not lines or lines[0] != SINK_HEADER:
            return
        sink_host, _, sink_port = self.ctx.sink_address.rpartition(":")
        for _label, doc_text in documents.items():
            parsed = self.agent.parse_nginx_config(doc_text, None)
            for block in parsed.get("maps", ()):
                for entry in block.get("entries", ()):
                    target = entry.get("target")
                    if target is None:
                        continue
                    if (target.get("host") == sink_host
                            and _port(target.get("port")) == _port(sink_port)):
                        return  # a route still points at the sink: it stays
        os.remove(path)

    def run(self, body):
        """The whole protocol for one validated body: `(answer, code)`. `body` is
        `{id, action, dry_run}` — the body has already been shape-checked by the
        caller (the wire's gate is the same one)."""
        ctx = self.ctx
        write = self
        source, in_port, host, port = self.source, self.in_port, self.host, self.port
        wanted = "off" if body["action"] == "disable" else "on"
        route, refusal = write.locate(source, in_port, host, port)
        if refusal:
            if refusal == "map-default":
                return {"ok": False, "applied": False,
                        "refused": "map-default"}, EXIT_REFUSED
            return {"ok": False, "detail": refusal}, EXIT_REFUSED
        original = write.current_line(route)
        current = write.state_of(route)
        # `enable` restores only from the sidecar: the line must be byte-identical
        # to what this helper wrote, and the map arm's original value comes from the
        # row (it is not in the file after a disable). A missing or drifted row is
        # a refusal, and no byte changes.
        if wanted == "on":
            state, refusal = load_state(ctx.state_file)
            if refusal:
                return {"ok": False, "detail": refusal}, EXIT_REFUSED
            # a `map` route is addressed by its sink value while off, so the
            # enable's id is not the write's: the row is found by the line this
            # helper wrote in this file, never by the id alone.
            row = next((value for value in state.values()
                       if value.get("file") == route["file"]
                       and value.get("line") == route["index"]), None)
            if row is None:
                return {"ok": False, "applied": False, "drift": True,
                        "detail": "no state row: this helper did not write the line"}, \
                    EXIT_REFUSED
            if original != row.get("written"):
                return {"ok": False, "applied": False, "drift": True,
                        "detail": "the line was edited after the helper wrote it"}, \
                    EXIT_REFUSED
            original = row["original"]
        if current == wanted:
            # step 2: already there. No write, no checker, no reload, no state change.
            return {"ok": True, "applied": False, "changed": False,
                    "id": body["id"], "state": current}, EXIT_ANSWERED
        new_line = write.edited_line(route, wanted, original)
        lines = route["text"].splitlines()
        new_lines = list(lines)
        new_lines[route["index"]] = new_line
        before = route["text"]
        after = "\n".join(new_lines)
        if before.endswith("\n"):
            after += "\n"
        if body["dry_run"]:
            # the dry run: the unified diff and nothing else — no file (not even
            # the sink's), no checker, no reload, no state.
            diff = list(difflib.unified_diff(lines, new_lines,
                                            fromfile=route["file"],
                                            tofile=route["file"] + ".new",
                                            lineterm=""))
            return {"ok": True, "applied": False, "changed": True,
                    "dry_run": True, "id": body["id"], "diff": diff}, EXIT_ANSWERED
        # the sink must exist before the checker runs (a `map` entry rewritten to
        # it is a reference the checker resolves); an operator's file at that
        # path is a refusal.
        if wanted == "off" and route["kind"] == "map":
            ok, sink_refusal = write.ensure_sink()
            if not ok:
                return {"ok": False, "detail": sink_refusal}, EXIT_REFUSED
        # step 4: the atomic write.
        _replace_file(route["path"], after)
        # step 5: the checker; a failure restores the previous bytes and reloads
        # nothing.
        ok, first = write.validate(route)
        if not ok:
            _replace_file(route["path"], before)
            return {"ok": False, "applied": False, "validated": False,
                    "restored": True, "detail": first}, EXIT_REFUSED
        # step 6: the reload (the only verb); a failure leaves the file applied.
        reloaded, status = write.reload(route)
        # step 7: prove the marker state is live on the re-read file, then record.
        text = _read(route["path"])
        proof = "on"
        if text is not None:
            _clean, name = self.agent.split_marker(
                text.splitlines()[route["index"]])
            proof = "off" if name is not None else "on"
        if route["kind"] == "map" and wanted == "on":
            # the re-pointed entry no longer names the sink: the sink stays only
            # while some route does (every document is scanned, not just this one).
            new_documents = dict(route["documents"])
            new_documents[route["file"]] = text or ""
            write.remove_sink_if_empty(new_documents)
        state, refusal = load_state(ctx.state_file)
        if refusal:
            return {"ok": True, "applied": True, "validated": True,
                    "reloaded": reloaded, "status": status,
                    "detail": refusal}, EXIT_REFUSED
        if body["action"] == "disable":
            state[body["id"]] = {
                "actor": ctx.actor,
                # `datetime.timezone.utc`, not the `datetime.UTC` alias (3.11+): the
                # helper runs on a host's Python 3.10.
                "at": datetime.datetime.now(datetime.timezone.utc).isoformat(),  # noqa: UP017
                "file": route["file"],
                # the file as the helper left it: `verify` compares against it, so a
                # hand edit since the write is a fact the operator can see.
                "sha256": hashlib.sha256(after.encode("utf-8")).hexdigest(),
                "action": "disable",
                "line": route["index"],
                "written": new_line,
                "original": original,
            }
        else:
            # the row is keyed by the write's id, which for a `map` route is not
            # the enable's id: drop the row this line came from.
            for key in [k for k, value in state.items()
                       if value.get("file") == route["file"]
                       and value.get("line") == route["index"]]:
                state.pop(key, None)
        ok, refusal = save_state(ctx.state_file, state)
        if not ok:
            return {"ok": True, "applied": True, "validated": True,
                    "reloaded": reloaded, "status": status,
                    "detail": f"state-write-failed: {refusal}"}, EXIT_REFUSED
        return {"ok": True, "applied": True, "validated": True,
                "reloaded": reloaded, "status": status, "state": proof}, \
            EXIT_ANSWERED


# -- one request --------------------------------------------------------------- #

def refused(detail, code=EXIT_REFUSED):
    """The one refusal shape: `ok` false and the `detail` (never a traceback,
    never a credential — the detail names the arm, the file, the reason, and
    nothing else)."""
    return {"ok": False, "detail": detail}, code


def show(write):
    """`show`: the live state of one id — the file, the line as it stands, and
    `state` (`on`/`off`) read through the agent's own marker helper. No write,
    no checker, no reload. An id the file no longer carries is `not-found`."""
    route, refusal = write.locate(write.source, write.in_port, write.host, write.port)
    if refusal:
        if refusal == "map-default":
            return refused("map-default")[0]
        return refused(refusal)[0]
    return {"ok": True, "id": write.route_id, "state": write.state_of(route),
            "file": route["file"]}


def handle_request(ctx, agent, body, actor="manual"):
    """The one wire body: exactly `{id, action, dry_run}`. Anything wider or
    narrower is a `detail` refusal before any file is read: the wire is the
    dashboard's one narrow request, and `dry_run` here means exactly what it
    means in the task — the diff, no write, no checker, no reload, no state.
    `action` is the request's verb; the read-only verbs (`show`, `verify`)
    take no action at all, so `action` is optional for them."""
    if not isinstance(body, dict):
        return refused("body: one JSON object")
    if set(body) - {"id", "action", "dry_run"}:
        return refused("body: exactly {id, action, dry_run}")
    route_id = body.get("id")
    if not isinstance(route_id, str) or not route_id or len(route_id) > ROUTE_ID_MAX:
        return refused("id: one non-empty string of at most 256 characters")
    action = body.get("action")
    if action is None and "action" not in body:
        action = "show"
    if action not in ("show", "disable", "enable"):
        return refused("action: one of show|disable|enable")
    if "dry_run" in body and not isinstance(body["dry_run"], bool):
        return refused("dry_run: a boolean")
    dry_run = bool(body.get("dry_run", False))
    parsed = parse_route_id(route_id)
    if parsed is None:
        return refused("id: `source:in-port->host:out-port`")
    source, in_port, host, out_port = parsed
    write = Write(ctx, agent, source, in_port, host, out_port, route_id)
    if action == "show":
        return show(write), EXIT_ANSWERED
    body = {"id": route_id, "action": action, "dry_run": dry_run}
    return write.run(body)


def serve(ctx, agent):
    """`--serve`: one request per process, from the socket-activated stdin. The
    request is one newline-delimited JSON line — exactly the agent write's wire
    body, `{id, action, dry_run}` (wider or narrower is a refusal) — and the
    answer is one JSON line back on the socket. The
    peer check is the second look on top of the unit's `SocketUser`/`SocketMode`:
    `SO_PEERCRED` names the caller's uid and it must be `--allow-uid`'s. The
    transport is the socket when systemd handed one: fd 3 (the classic
    socket-activation contract) or fd 0 (with `Accept=yes` and
    `StandardInput=socket`, the accepted connection lands on stdin — that is
    what the live box delivered, phase 33 task 07); otherwise a connected pipe,
    and the peer check runs on it too — a transport without a readable peer (an
    unconnected pipe) is a refusal, not a skipped gate.
    """
    sock = None
    for fd in (3, 0):
        with contextlib.suppress(OSError, ValueError, AttributeError):
            sock = socket.fromfd(fd, socket.AF_UNIX, socket.SOCK_STREAM)
            if sock is not None:
                break
    peer_ok, peer_detail = _peer_check(sock, ctx.allow_uid)
    if not peer_ok:
        # a refusal the dialer cannot see is a timeout in their code: the
        # refusal goes on the wire before the process exits.
        if sock is not None:
            with contextlib.suppress(OSError):
                sock.sendall(json.dumps({"ok": False, "detail": peer_detail},
                                       separators=(",", ":")).encode("utf-8") + b"\n")
        return {"ok": False, "detail": peer_detail}, EXIT_REFUSED
    # the socket path reads with `recv(2)`: on this fleet's kernel, `read(2)`
    # on a socket fd does not wake on data the peer already sent (the
    # socketpair test in tests/test_deploy_scripts.py pins it — the helper
    # sat in `read` with the request in the buffer, `recv` got it at once).
    next_chunk = sock.recv if sock is not None else sys.stdin.buffer.read
    buffer = b""
    while b"\n" not in buffer:
        chunk = next_chunk(4096)
        if not chunk:
            break
        buffer += chunk
        if len(buffer) > SERVE_REQUEST_MAX:
            return {"ok": False,
                    "detail": f"body: longer than {SERVE_REQUEST_MAX} bytes"}, EXIT_REFUSED
    line = buffer.split(b"\n", 1)[0]
    try:
        body = json.loads(line.decode("utf-8"))
    except (ValueError, UnicodeDecodeError):
        return {"ok": False, "detail": "body: not JSON"}, EXIT_REFUSED
    answer, code = handle_request(ctx, agent, body,
                                 actor=f"agent:{ctx.allow_uid}")
    # the answer goes back to the socket: with `StandardOutput=socket` that is
    # also the process's stdout, but the socket is the contract and stdout is
    # not.
    if sock is not None:
        try:
            sock.sendall(json.dumps(answer, separators=(",", ":")).encode("utf-8")
                        + b"\n")
        except OSError as exc:
            return {"ok": False, "detail": f"send: {type(exc).__name__}"}, EXIT_REFUSED
    return answer, code


def _ucred_slots(allow_uid):
    """Calibrate which field of `SO_PEERCRED` is the uid, because the 12-byte
    ucred buffer's field order is not portable: this fleet's el10 kernel
    (6.12.0-211.69.1.el10_2) delivers it as `{pid, uid, gid}` — the classic
    layout `{uid, gid, pid}` reversed (probed on the live box, phase 33 task
    07). The probe is a socketpair whose peer is this process: the field equal
    to our pid names the pid slot. For a root process (the service mode) the
    probe *shifts the uid* to the expected value with `setuid` while the gid
    stays 0 (a privilege drop, restored in `finally` before the state file is
    touched): the field that becomes `expected` is the uid slot, the one that
    stays 0 is the gid slot — a matching gid on a foreign-uid caller cannot
    open the gate. A non-root process cannot shift: with `euid == egid` its
    two remaining fields are equal, so it gets **both** back and the gate
    then requires both to hold the expected value (the service account has
    uid == gid, so a true match still passes); `euid != egid` separates the
    two by uniqueness. An inconclusive layout refuses: one ambiguous instance
    must not open the gate for everyone.
    """
    try:
        expected = pwd.getpwnam(allow_uid).pw_uid
    except KeyError:
        return None
    if expected is None or expected < 0:
        return None
    shifted = os.geteuid() == 0 and expected != 0
    if shifted:
        try:
            os.setuid(expected)  # gid stays 0: only the uid field moves
        except OSError:
            shifted = False
    a, b = socket.socketpair(socket.AF_UNIX, socket.SOCK_STREAM)
    try:
        data = a.getsockopt(socket.SOL_SOCKET, socket.SO_PEERCRED, 12)
    finally:
        if shifted:
            with contextlib.suppress(OSError):
                os.setuid(0)
        a.close()
        b.close()
    fields = list(struct.unpack_from("<3i", data))
    if len(fields) != 3:
        return None
    pid_slots = {i for i, v in enumerate(fields) if v == os.getpid()}
    if len(pid_slots) != 1:
        return None
    pid_slot = pid_slots.pop()
    non_pid = [i for i in range(3) if i != pid_slot]
    if shifted:
        uid_slots = tuple(i for i in non_pid if fields[i] == expected)
        if len(uid_slots) != 1:
            return None
    elif expected == 0 or os.geteuid() == os.getegid():
        # unshifted and the two remaining fields are equal (or `root` is
        # allowed from an unprivileged probe): require both to hold the value
        uid_slots = tuple(non_pid)
    else:
        uid_slots = tuple(i for i in non_pid if fields[i] == os.geteuid())
        if len(uid_slots) != 1:
            return None
    return pid_slot, uid_slots


def _peer_check(sock, allow_uid):
    """`SO_PEERCRED` on the connected socket: `(ok, detail)`. The uid field is
    read at the slot `_ucred_slots` calibrated on this kernel (the buffer's
    field order is not portable — the live box delivers it reversed); a
    transport that cannot report a peer (a plain pipe) or whose layout is
    inconclusive fails the check — the gate never opens on a caller the kernel
    will not name, and never on an unreadable one."""
    if sock is None:
        return False, "peer: no connected socket on this transport"
    try:
        data = sock.getsockopt(socket.AF_UNIX, socket.SO_PEERCRED, 12)
    except OSError:
        return False, "peer: the transport reported no peer"
    expected = None
    try:
        expected = pwd.getpwnam(allow_uid).pw_uid
    except KeyError:
        return False, f"peer: allow-uid {allow_uid} is not a user"
    if expected is None or expected < 0:
        return False, f"peer: allow-uid {allow_uid} is not a user"
    slots = _ucred_slots(allow_uid)
    if slots is None:
        return False, "peer: the credential layout is inconclusive on this kernel"
    pid_slot, uid_slots = slots
    fields = struct.unpack_from("<3i", data)
    if fields[pid_slot] <= 0:
        return False, "peer: the transport reported no peer"
    held = [fields[i] for i in uid_slots]
    if any(v != expected for v in held):
        return False, f"peer: uid {held[0]} is not {allow_uid}"
    return True, None


# -- the manual arms ----------------------------------------------------------- #

def verify(ctx, agent):
    """`verify`: the sidecar against the files — per id, the file's hash and the
    written line. A drift is a fact, not an error: the answer says which, and the
    operator decides what it means."""
    state, refusal = load_state(ctx.state_file)
    if refusal:
        return {"ok": False, "detail": refusal}, EXIT_REFUSED
    rows = []
    drifted = False
    for key in sorted(state):
        row = state[key]
        path = under_root(ctx.root, row.get("file", ""))
        text = _read(path)
        if text is None:
            rows.append({"id": key, "file": row.get("file"),
                        "hash": "missing", "marker": "absent"})
            drifted = True
            continue
        # the sidecar's `sha256` is the file as the helper left it: a mismatch is
        # a hand edit since the write — a fact, not an error.
        sha = hashlib.sha256(text.encode("utf-8")).hexdigest()
        hash_ok = sha == row.get("sha256")
        written = row.get("written")
        live = written is not None and written in text.splitlines()
        if not hash_ok or not live:
            drifted = True
        rows.append({"id": key, "file": row.get("file"),
                    "hash": "ok" if hash_ok else "drift",
                    "marker": "live" if live else "absent"})
    return {"ok": not drifted, "state": rows}, \
        EXIT_ANSWERED if not drifted else EXIT_REFUSED


def main(argv=None):
    argv = list(sys.argv[1:] if argv is None else argv)
    parser = argparse.ArgumentParser(
        prog="dashboards-routectl",
        description="the route-switch root helper: one id, one file, one reload")
    parser.add_argument("verb", nargs="?", choices=("list", "show", "diff",
                                                   "disable", "enable", "verify"),
                       help="what to do (`--serve` is the socket-activated mode)")
    parser.add_argument("--root", default="/",
                       help="the tree the config paths are read under (rehearsal)")
    parser.add_argument("--agent-module",
                       default="/usr/local/lib/dashboards-agent/host_agent.py",
                       help="the agent's parser, imported by file path (required)")
    parser.add_argument("--haproxy-config", default="/etc/haproxy/haproxy.cfg")
    parser.add_argument("--nginx-config", default="/etc/nginx/nginx.conf")
    parser.add_argument("--state",
                       default="/var/lib/dashboards-routectl/state.json")
    parser.add_argument("--sink-conf", default=SINK_FILE_DEFAULT)
    parser.add_argument("--sink-address", default=SINK_ADDRESS_DEFAULT)
    parser.add_argument("--nginx-unit", default="nginx")
    parser.add_argument("--haproxy-unit", default="haproxy")
    parser.add_argument("--id", default=None, help="the `connections[].id` to act on")
    parser.add_argument("--action", choices=("disable", "enable"), default=None,
                       help="the action to preview (with `diff`)")
    parser.add_argument("--apply", action="store_true",
                       help="write (without it every write verb is a dry run)")
    parser.add_argument("--serve", action="store_true",
                       help="serve one request from the socket-activated stdin")
    parser.add_argument("--allow-uid", default="dashboards-agent",
                       help="the only peer uid --serve accepts (SO_PEERCRED)")
    args = parser.parse_args(argv)

    if args.serve and args.verb:
        print(json.dumps({"ok": False, "detail": "usage: --serve takes no verb"},
                        separators=(",", ":")))
        return EXIT_USAGE
    if not args.serve and args.verb is None:
        print(json.dumps(
            {"ok": False,
             "detail": "usage: a verb (list | show | diff | disable | enable | "
                      "verify) or --serve"}, separators=(",", ":")))
        return EXIT_USAGE
    agent, refusal = load_agent(args.agent_module)
    if agent is None:
        print(json.dumps({"ok": False, "detail": refusal},
                        separators=(",", ":")))
        return EXIT_REFUSED
    ctx = argparse.Namespace(
        root=args.root, haproxy_config=args.haproxy_config,
        nginx_config=args.nginx_config, state_file=args.state,
        sink_conf=args.sink_conf, sink_address=args.sink_address,
        nginx_unit=args.nginx_unit, haproxy_unit=args.haproxy_unit,
        actor="manual", allow_uid=args.allow_uid)
    if args.serve:
        answer, code = serve(ctx, agent)
    elif args.verb == "list":
        state, refusal = load_state(args.state)
        if refusal:
            answer, code = {"ok": False, "detail": refusal}, EXIT_REFUSED
        else:
            answer, code = {"ok": True, "state": [
                {"id": key, **value} for key, value in sorted(state.items())]
            }, EXIT_ANSWERED
    elif args.verb == "verify":
        answer, code = verify(ctx, agent)
    else:
        if not args.id:
            return {"ok": False, "detail": f"usage: {args.verb} needs --id"}, \
                EXIT_USAGE
        if args.verb == "diff" and args.action not in ("disable", "enable"):
            print(json.dumps(
                {"ok": False, "detail": "usage: diff needs --action disable|enable"},
                separators=(",", ":")))
            return EXIT_USAGE
        body = {"id": args.id,
                "action": args.action if args.verb == "diff" else args.verb,
                "dry_run": not args.apply}
        answer, code = handle_request(ctx, agent, body, actor="manual")
    if not args.serve:
        # in `--serve` the answer is already on the socket (the unit's
        # StandardOutput=socket is that same socket); a second write here would
        # put the line on the wire twice.
        sys.stdout.write(json.dumps(answer, separators=(",", ":")) + "\n")
        sys.stdout.flush()
    return code


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