feat: the configuration that says which services run together (4.17.0)

ServiceClustersMixin.get_service_clusters() -> ServiceClustersDict
{patroni, adguardhome_sync, bunkerweb}: per mechanism the clusters the
host's configuration says it belongs to, each with its members (address,
port, name, role as the mechanism words it, this_host, config_from) and the
unit or container it was read from (source).

- Patroni: scope, name and restapi.connect_address from its YAML, then the
  members and roles /cluster lists from the DCS.
- adguardhome-sync: ORIGIN_URL and REPLICA*_URL; the origin pushes to each
  replica.
- BunkerWeb: a worker's API_WHITELIST_IP names the manager it admits, a
  manager's BUNKERWEB_INSTANCES its workers.

Two hosts answering at one virtual IP prove nothing; this is the evidence
that they run one service (NetOrk/netork#1125). One generic command for
every Linux host, filtered on the host like the virtual IPs' one:
allowlisted fields only, URLs without user:password@, DCS endpoints stay
there. Root first, then without it; None per mechanism is "did not look".

README.md

@@ -154,6 +154,22 @@ A mechanism it could not read completely is `None` ("did not look"), a host with
 `[]`. The command is about 4 kB, so a driver sends it on an exec channel
 (`_run_virtual_ips_command(command, privileged=)`), not typed into a shell.
 
+`ServiceClustersMixin` (`get_service_clusters`) reads the configuration that proves two
+hosts run one service together -- not merely that they answer at one address: Patroni
+(`scope`, `name` and `restapi.connect_address` from its YAML, then the members and roles
+its REST API's `/cluster` lists from the DCS), adguardhome-sync (`ORIGIN_URL` and
+`REPLICA*_URL` of its container: the origin pushes to each replica) and BunkerWeb (a
+worker's `API_WHITELIST_IP` names the manager it takes its configuration from, a manager's
+`BUNKERWEB_INSTANCES` its workers). Each member carries its role as the mechanism words it
+and the index of the member it takes its configuration from (`config_from`). The same
+configuration holds the superuser password, the sync's logins and BunkerWeb's database URI,
+so awk programs filter **on the host**: allowlisted fields only, a URL without its
+`user:password@`, and the DCS endpoints stay on the host. Root first (Patroni's YAML is the
+postgres user's), then without it. `None` per mechanism is "did not look", `[]` "has none".
+The container engine is asked through `_container_engine_binary` where the driver has one.
+A driver sends the command on an exec channel (`_run_service_clusters_command(command,
+privileged=)`).
+
 `get_dhcp_ha` (declared on `DhcpServerMixin`, optional) is what a DHCP server's own HA
 configuration says: whether HA is on, which peer this server is, each peer's name, role --
 as the server words it, not a shared vocabulary -- and, where the server has one, the URL

napalm_device_types/__init__.py

@@ -51,6 +51,7 @@ instead of being restated on every role that happens to need it:
 * :class:`~napalm_device_types.nat_vpn.NatVpnMixin`
 * :class:`~napalm_device_types.packages.PackageManagementMixin`
 * :class:`~napalm_device_types.ping_sweep.PingSweepMixin`
+* :class:`~napalm_device_types.service_clusters.ServiceClustersMixin`
 * :class:`~napalm_device_types.services.ServiceControlMixin`
 * :class:`~napalm_device_types.systemd.SystemdServicesMixin`
 * :class:`~napalm_device_types.updates.UpdateMixin`
@@ -155,6 +156,12 @@ from napalm_device_types.system_identity import (
     is_placeholder_identity,
     parse_system_identity,
 )
+from napalm_device_types.service_clusters import (
+    SERVICE_CLUSTERS_COMMAND,
+    ServiceClustersMixin,
+    parse_service_clusters,
+    service_clusters_command,
+)
 from napalm_device_types.virtual_ips import (
     VIRTUAL_IPS_COMMAND,
     VirtualIpsMixin,
@@ -256,6 +263,10 @@ __all__ = [
     "UpdateMixin",
     "VIRTUAL_IPS_COMMAND",
     "VirtualIpsMixin",
+    "SERVICE_CLUSTERS_COMMAND",
+    "ServiceClustersMixin",
+    "parse_service_clusters",
+    "service_clusters_command",
     "SystemIdentityMixin",
     "SYSTEM_IDENTITY_COMMAND",
     "is_placeholder_identity",

napalm_device_types/models.py

@@ -425,6 +425,68 @@ class VirtualIpDict(TypedDict):
     holds: NotRequired[Optional[bool]]
 
 
+class ServiceClusterMemberDict(TypedDict):
+    """One member of a service cluster, as the configuration names it.
+
+    A member is named by what the configuration has: an address (a URL's host, a
+    whitelisted address -- possibly a host name), a port where it gives one, a
+    name where the mechanism has its own (Patroni's). ``this_host`` marks the
+    member the reading host is when the configuration does not name it by
+    address. Only these fields leave the host -- never a password or a DCS
+    endpoint.
+    """
+
+    name: Optional[str]
+    address: Optional[str]
+    port: Optional[int]
+    #: as the mechanism words it: Patroni's "leader"/"sync_standby"/"replica",
+    #: adguardhome-sync's "origin"/"replica", BunkerWeb's "manager"/"worker"
+    role: Optional[str]
+    #: the member's state where the mechanism reports one (Patroni's "running",
+    #: "streaming"); configuration alone has none
+    state: Optional[str]
+    this_host: bool
+    #: the index in ``members`` of the member whose configuration this one takes:
+    #: a replica's origin, a worker's manager. None where it takes none.
+    config_from: Optional[int]
+
+
+class ServiceClusterDict(TypedDict):
+    """A cluster the host's configuration says it belongs to (netOrk #1125).
+
+    The evidence that two hosts run one service together, not that they merely
+    answer at one address: Patroni's scope and the members its DCS lists, the
+    origin and replicas adguardhome-sync keeps in step, a BunkerWeb worker's
+    manager.
+    """
+
+    mechanism: str  # "patroni" | "adguardhome-sync" | "bunkerweb"
+    #: what the members run, as a host reports it: "patroni", "adguardhome", "bunkerweb"
+    service: str
+    #: what the evidence was read from on this host: the unit without ".service"
+    #: ("patroni", "patroni@16-main") or the container's name ("adguardhome-sync")
+    source: str
+    #: what the members share where the reading host can name it: Patroni's scope,
+    #: the sync's origin ("10.9.100.5:3000"), a worker's manager address. None for a
+    #: manager, which cannot name its own address.
+    cluster: Optional[str]
+    running: Optional[bool]  # the unit is active; a listed container runs
+    members: List[ServiceClusterMemberDict]
+
+
+class ServiceClustersDict(TypedDict):
+    """What a driver's ``get_service_clusters`` read, per mechanism.
+
+    A list is what was looked at -- ``[]`` a host without that mechanism. None is
+    "did not look": a configuration it could not read, a host without systemd, a
+    container engine it may not ask, a sync configured by file.
+    """
+
+    patroni: Optional[List[ServiceClusterDict]]
+    adguardhome_sync: Optional[List[ServiceClusterDict]]
+    bunkerweb: Optional[List[ServiceClusterDict]]
+
+
 class VirtualIpsDict(TypedDict):
     """What a driver's ``get_virtual_ips`` read, per mechanism.
 

napalm_device_types/service_clusters.py

@@ -0,0 +1,528 @@
+# -*- coding: utf-8 -*-
+"""Service clusters: the configuration that says which services run together.
+
+Two hosts that answer at one virtual IP may run the same service by accident.
+What proves a cluster is the configuration its members share. That is read the
+same way on every Linux host, so the command and its parse live here once, and
+a driver only carries the command across (netOrk #1125).
+
+**What is read.**
+
+- *Patroni*: every ``patroni.service`` and ``patroni@*.service`` unit systemd
+  knows. Its configuration file -- the unit's first non-flag argument, else
+  ``/etc/patroni/config.yml`` -- gives ``scope``, ``name`` and
+  ``restapi.connect_address``. The REST API there (``/cluster``, which Patroni
+  answers without a login) lists the members its DCS knows, each with its role:
+  ``leader``, ``sync_standby``, ``replica``. When it does not answer, the host
+  is a member of its scope on its own.
+- *adguardhome-sync*: a running container whose image is adguardhome-sync.
+  ``ORIGIN_URL`` and ``REPLICA*_URL`` name the AdGuard Home instances it keeps in
+  step: the origin pushes to every replica. A sync configured by a YAML file
+  instead of its environment is not read.
+- *BunkerWeb*: a running instance container (``bunkerity/bunkerweb``) admits a
+  manager's configuration pushes from the addresses in ``API_WHITELIST_IP``; a
+  manager (``bunkerweb-all-in-one``, ``bunkerweb-scheduler``) names its
+  instances in ``BUNKERWEB_INSTANCES``. BunkerWeb's own default whitelist,
+  loopback and the docker network, names no one. A manager that keeps its
+  instances in its database names none here: its workers name it.
+
+**Filtered on the host.** The same configuration holds secrets -- Patroni's
+superuser and REST passwords, the sync's AdGuard logins, BunkerWeb's database
+URI. The command never prints a configuration: awk programs print allowlisted
+fields only, and a URL loses its ``user:password@``. The DCS endpoints stay on
+the host too; the members come from ``/cluster``.
+
+**None is "did not look".** A Patroni configuration that exists but cannot be
+read, a host without systemd, a docker the login may not ask, a sync configured
+by file: that mechanism is None, and the caller keeps what it knew. ``[]`` is a
+host that has none.
+
+**Root, and without it.** Patroni's configuration is the postgres user's alone.
+The whole script goes to the host as one ``sh -c`` argument; when the privileged
+call brings no report back, it runs again without privilege.
+"""
+
+from __future__ import annotations
+
+import ipaddress
+import json
+import re
+from shlex import quote
+from typing import Callable, Dict, List, Optional, Tuple, TYPE_CHECKING
+from urllib.parse import urlsplit
+
+from napalm_device_types.models import (
+    ServiceClusterDict,
+    ServiceClusterMemberDict,
+    ServiceClustersDict,
+)
+from napalm_device_types.terminal import strip_terminal_codes
+
+_BEGIN = "SVCC_BEGIN"
+_END = "SVCC_END"
+#: Exists only while systemd runs as PID 1; a container may carry systemctl without.
+_SYSTEMD_RUN = "/run/systemd/system"
+_PATRONI_CONFIG = "/etc/patroni/config.yml"
+#: The container environment that may leave the host.
+_ENV_KEYS = "ORIGIN_URL|REPLICA[0-9]*_URL|API_WHITELIST_IP|BUNKERWEB_INSTANCES|API_HTTP_PORT"
+#: BunkerWeb's instance API, where a manager pushes its configuration.
+_BUNKERWEB_API_PORT = 5000
+
+#: Fed a unit's ``systemctl show`` output: ``active <state>`` and ``config <path>``,
+#: the first argument of the command line that is no flag.
+_UNIT_AWK = " ".join(
+    (
+        r'/^ActiveState=/ { print "active " substr($0, 13); next }',
+        r'/^ExecStart=/ { s = $0; i = index(s, "argv[]="); if (!i) next; s = substr(s, i + 7);'
+        r' j = index(s, " ;"); if (j) s = substr(s, 1, j - 1); n = split(s, t, " ");'
+        r' for (k = 2; k <= n; k++) if (!c && t[k] !~ /^-/) { print "config " t[k]; c = 1 } }',
+    )
+)
+
+#: Fed Patroni's YAML: ``scope``, ``name`` (top level only -- ``tags`` may carry a
+#: ``name`` too), ``restapi`` (its connect_address) and ``tls yes`` when the REST
+#: API has a certificate. Nothing else leaves the file.
+_PATRONI_AWK = " ".join(
+    (
+        r'function v(l) { sub(/^[^:]*:[ \t]*/, "", l); return l }',
+        r'/^[A-Za-z_][A-Za-z0-9_]*:/ { sec = $0; sub(/:.*/, "", sec);'
+        r' if (sec == "scope" || sec == "name") print sec " " v($0); next }',
+        r'sec == "restapi" && /^[ \t]+connect_address:/ { print "restapi " v($0) }',
+        r'sec == "restapi" && /^[ \t]+certfile:/ { print "tls yes" }',
+    )
+)
+
+#: Fed a container's environment, one ``KEY=value`` a line: the allowlisted keys,
+#: each URL without its ``user:password@``.
+_ENV_AWK = " ".join(
+    (
+        r'{ x = index($0, "="); if (!x) next; k = substr($0, 1, x - 1); v = substr($0, x + 1);',
+        rf" if (k !~ /^({_ENV_KEYS})$/) next;",
+        r' gsub(/:\/\/[^@\/ ]*@/, "://", v); print "env " k " " v }',
+    )
+)
+
+_PATRONI_SH = (
+    f"if command -v systemctl >/dev/null 2>&1 && [ -d {_SYSTEMD_RUN} ]; then echo '[systemd]'; "
+    "for u in $(systemctl list-units --all --plain --no-legend --type=service "
+    "patroni.service 'patroni@*' 2>/dev/null | awk '{print $1}'); do "
+    'echo "[patroni $u]"; '
+    'o=$(systemctl show -p ActiveState -p ExecStart "$u" 2>/dev/null | '
+    f"awk {quote(_UNIT_AWK)}); "
+    'echo "$o"; '
+    "f=$(echo \"$o\" | sed -n 's/^config //p' | head -n 1); "
+    f'[ -n "$f" ] || f={_PATRONI_CONFIG}; '
+    'if [ -r "$f" ]; then '
+    f'p=$(awk {quote(_PATRONI_AWK)} "$f"); echo "$p"; '
+    "r=$(echo \"$p\" | sed -n 's/^restapi //p' | head -n 1 | tr -d '\\042\\047 '); "
+    's=http; case "$p" in *"tls yes"*) s=https;; esac; '
+    'if [ -n "$r" ]; then '
+    'b=$(curl -s -k -m 3 "$s://$r/cluster" 2>/dev/null || '
+    'wget -q -T 3 --no-check-certificate -O - "$s://$r/cluster" 2>/dev/null); '
+    "if [ -n \"$b\" ]; then echo \"rest $(echo \"$b\" | tr -d '\\n')\"; "
+    "else echo rest-unreachable; fi; fi; "
+    'elif [ -e "$f" ]; then echo "unreadable $f"; '
+    'else echo "missing $f"; fi; '
+    "done; "
+    "else echo '[no-systemd]'; fi; "
+)
+
+
+def _docker_sh(binary: Optional[str]) -> str:
+    """The containers part: running containers of the clustered images, their
+    allowlisted environment. ``[no-docker]`` for a host without the engine."""
+    if not binary:
+        return "echo '[no-docker]'; "
+    return (
+        f"D={quote(binary)}; "
+        'if command -v "$D" >/dev/null 2>&1; then '
+        "if c=$(\"$D\" ps --format '{{.Names}}' 2>/dev/null); then echo '[docker]'; "
+        "for n in $c; do "
+        "i=$(\"$D\" inspect --format '{{.Config.Image}}' \"$n\" 2>/dev/null); "
+        'case "$i" in *adguardhome-sync*|*bunkerweb*) '
+        'echo "[container $n]"; echo "image $i"; '
+        "\"$D\" inspect --format '{{range .Config.Env}}{{println .}}{{end}}' \"$n\" "
+        f"2>/dev/null | awk {quote(_ENV_AWK)};; esac; done; "
+        "else echo '[docker-unreadable]'; fi; "
+        "else echo '[no-docker]'; fi; "
+    )
+
+
+def service_clusters_command(docker: Optional[str] = "docker") -> str:
+    """One line of POSIX ``sh``, read-only. *docker* is the engine's binary as the
+    driver reaches it (QNAP's lives elsewhere), None for a host without one. The
+    frame markers are printed in two halves so that a transport which echoes the
+    command does not show them early."""
+    return (
+        "PATH=$PATH:/usr/sbin:/sbin; "
+        "printf '%s%s\\n' SVCC_ BEGIN; "
+        + _PATRONI_SH
+        + _docker_sh(docker)
+        + "printf '%s%s\\n' SVCC_ END"
+    )
+
+
+SERVICE_CLUSTERS_COMMAND = service_clusters_command()
+
+_COMMENT_RE = re.compile(r"\s+#.*$")
+_Record = Tuple[str, str]
+_Section = Tuple[str, str, List[_Record]]
+
+
+def _frame(output: str) -> List[str]:
+    lines = [line.strip() for line in strip_terminal_codes(output).splitlines()]
+    try:
+        start = lines.index(_BEGIN)
+        end = lines.index(_END, start)
+    except ValueError:
+        raise ValueError("no intact service cluster report in the output") from None
+    return lines[start + 1 : end]
+
+
+def _sections(lines: List[str]) -> List[_Section]:
+    """``[<kind> <name>]`` headers, each with its records as ``(kind, rest)``."""
+    sections: List[_Section] = []
+    for line in lines:
+        if line.startswith("[") and line.endswith("]"):
+            kind, _, name = line[1:-1].partition(" ")
+            sections.append((kind, name, []))
+        elif line and sections:
+            kind, _, rest = line.partition(" ")
+            sections[-1][2].append((kind, rest.strip()))
+    return sections
+
+
+def _clean(value: str) -> str:
+    """Without a trailing ``# comment`` and the quotes around it."""
+    value = _COMMENT_RE.sub("", value.strip())
+    if len(value) >= 2 and value[0] == value[-1] and value[0] in "\"'":
+        value = value[1:-1]
+    return value.strip()
+
+
+def _member(
+    *,
+    address: Optional[str] = None,
+    port: Optional[int] = None,
+    name: Optional[str] = None,
+    role: Optional[str] = None,
+    state: Optional[str] = None,
+    this_host: bool = False,
+    config_from: Optional[int] = None,
+) -> ServiceClusterMemberDict:
+    return {
+        "name": name,
+        "address": address,
+        "port": port,
+        "role": role,
+        "state": state,
+        "this_host": this_host,
+        "config_from": config_from,
+    }
+
+
+def _cluster(
+    mechanism: str,
+    service: str,
+    source: str,
+    cluster: Optional[str],
+    members: List[ServiceClusterMemberDict],
+    running: Optional[bool] = True,
+) -> ServiceClusterDict:
+    return {
+        "mechanism": mechanism,
+        "service": service,
+        "source": source,
+        "cluster": cluster,
+        "running": running,
+        "members": members,
+    }
+
+
+def _host_port(text: str) -> Tuple[Optional[str], Optional[int]]:
+    """``"10.0.0.5:8008"`` or ``"[fd00::5]:8008"`` -> host and port."""
+    try:
+        split = urlsplit(f"//{text.strip()}")
+        return split.hostname, split.port
+    except ValueError:
+        return None, None
+
+
+def _url(text: str) -> Tuple[Optional[str], Optional[int]]:
+    """A URL's host and port, the scheme's default where it names none."""
+    value = text.strip()
+    if "://" not in value:
+        value = f"http://{value}"
+    try:
+        split = urlsplit(value)
+        port = split.port or {"http": 80, "https": 443}.get(split.scheme)
+    except ValueError:
+        return None, None
+    return split.hostname, port
+
+
+# --- Patroni ------------------------------------------------------------------
+
+
+def _patroni_members(rest: str, own: Optional[str]) -> Optional[List[ServiceClusterMemberDict]]:
+    try:
+        members = json.loads(rest).get("members")
+    except (ValueError, AttributeError):
+        return None
+    if not isinstance(members, list):
+        return None
+    found = []
+    for entry in members:
+        if not isinstance(entry, dict) or not entry.get("name"):
+            continue
+        port = entry.get("port")
+        found.append(
+            _member(
+                name=str(entry["name"]),
+                address=str(entry["host"]) if entry.get("host") else None,
+                port=port if isinstance(port, int) else None,
+                role=str(entry["role"]) if entry.get("role") else None,
+                state=str(entry["state"]) if entry.get("state") else None,
+                this_host=entry["name"] == own,
+            )
+        )
+    return sorted(found, key=lambda m: m["name"] or "")
+
+
+def _patroni(unit: str, records: List[_Record]) -> List[ServiceClusterDict]:
+    values: Dict[str, str] = {}
+    rest: Optional[str] = None
+    for kind, text in records:
+        if kind in ("active", "scope", "name", "restapi") and kind not in values:
+            values[kind] = _clean(text)
+        elif kind == "rest":
+            rest = text
+    scope = values.get("scope")
+    if not scope:
+        return []
+    own = values.get("name") or None
+    members = _patroni_members(rest, own) if rest else None
+    if not members:
+        address, _ = _host_port(values.get("restapi", ""))
+        members = [_member(name=own, address=address, this_host=True)]
+    running = values["active"] == "active" if "active" in values else None
+    source = unit[: -len(".service")] if unit.endswith(".service") else unit
+    return [_cluster("patroni", "patroni", source, scope, members, running)]
+
+
+# --- Containers ----------------------------------------------------------------
+
+
+def _env(records: List[_Record]) -> Dict[str, str]:
+    env: Dict[str, str] = {}
+    for kind, text in records:
+        if kind == "env":
+            key, _, value = text.partition(" ")
+            env[key] = value.strip()
+    return env
+
+
+def _image_name(records: List[_Record]) -> str:
+    """``bunkerity/bunkerweb@sha256:...`` -> ``bunkerweb``."""
+    image = next((text for kind, text in records if kind == "image"), "")
+    return image.split("@")[0].rsplit("/", 1)[-1].split(":")[0]
+
+
+def _adguardhome_sync(name: str, records: List[_Record]) -> Optional[List[ServiceClusterDict]]:
+    env = _env(records)
+    origin = _url(env["ORIGIN_URL"]) if env.get("ORIGIN_URL") else (None, None)
+    if origin[0] is None:
+        return None  # configured by its YAML file, which is not read
+    replicas = sorted(
+        (key for key in env if re.fullmatch(r"REPLICA[0-9]*_URL", key)),
+        key=lambda key: int(key[7:-4] or 0),
+    )
+    members = [_member(address=origin[0], port=origin[1], role="origin")]
+    for key in replicas:
+        host, port = _url(env[key])
+        if host:
+            members.append(_member(address=host, port=port, role="replica", config_from=0))
+    cluster = f"{origin[0]}:{origin[1]}" if origin[1] else origin[0]
+    return [_cluster("adguardhome-sync", "adguardhome", name, cluster, members)]
+
+
+def _peer_address(entry: str) -> Optional[str]:
+    """An entry of a whitelist that names one peer: a single address that is no
+    loopback, link-local or unspecified one -- not a network."""
+    text = entry.strip()
+    if "/" in text:
+        try:
+            network = ipaddress.ip_network(text, strict=False)
+        except ValueError:
+            return None
+        if network.num_addresses != 1:
+            return None
+        text = str(network.network_address)
+    try:
+        address = ipaddress.ip_address(text)
+    except ValueError:
+        return None
+    if address.is_loopback or address.is_link_local or address.is_unspecified:
+        return None
+    return str(address)
+
+
+def _api_port(env: Dict[str, str]) -> int:
+    port = env.get("API_HTTP_PORT", "")
+    return int(port) if port.isdigit() else _BUNKERWEB_API_PORT
+
+
+def _bunkerweb_worker(name: str, env: Dict[str, str]) -> List[ServiceClusterDict]:
+    managers = [
+        address
+        for entry in re.split(r"[\s,]+", env.get("API_WHITELIST_IP", ""))
+        if entry and (address := _peer_address(entry))
+    ]
+    if not managers:
+        return []
+    members = [_member(port=_api_port(env), role="worker", this_host=True, config_from=1)]
+    members += [_member(address=address, role="manager") for address in managers]
+    return [_cluster("bunkerweb", "bunkerweb", name, managers[0], members)]
+
+
+def _bunkerweb_manager(name: str, env: Dict[str, str]) -> List[ServiceClusterDict]:
+    instances = [entry for entry in env.get("BUNKERWEB_INSTANCES", "").split() if entry]
+    if not instances:
+        return []
+    members = [_member(role="manager", this_host=True)]
+    for entry in instances:
+        host, port = _host_port(entry)
+        if host:
+            members.append(
+                _member(address=host, port=port or _BUNKERWEB_API_PORT, role="worker", config_from=0)
+            )
+    return [_cluster("bunkerweb", "bunkerweb", name, None, members)] if len(members) > 1 else []
+
+
+def _bunkerweb(name: str, records: List[_Record]) -> List[ServiceClusterDict]:
+    image = _image_name(records)
+    if image == "bunkerweb":
+        return _bunkerweb_worker(name, _env(records))
+    if image in ("bunkerweb-all-in-one", "bunkerweb-scheduler"):
+        return _bunkerweb_manager(name, _env(records))
+    return []  # the UI, the API, autoconf: no instance and no manager
+
+
+_CONTAINER_READERS: Dict[str, Callable[[str, List[_Record]], Optional[List[ServiceClusterDict]]]] = {
+    "adguardhome_sync": _adguardhome_sync,
+    "bunkerweb": _bunkerweb,
+}
+
+
+def _reader_for(records: List[_Record]) -> Optional[str]:
+    image = next((text for kind, text in records if kind == "image"), "")
+    if "adguardhome-sync" in image:
+        return "adguardhome_sync"
+    if "bunkerweb" in image:
+        return "bunkerweb"
+    return None
+
+
+def _containers(sections: List[_Section]) -> Dict[str, Optional[List[ServiceClusterDict]]]:
+    kinds = {kind for kind, _, _ in sections}
+    if "no-docker" in kinds:
+        return {"adguardhome_sync": [], "bunkerweb": []}
+    if "docker" not in kinds:
+        return {"adguardhome_sync": None, "bunkerweb": None}
+    found: Dict[str, Optional[List[ServiceClusterDict]]] = {"adguardhome_sync": [], "bunkerweb": []}
+    for kind, name, records in sections:
+        key = _reader_for(records) if kind == "container" else None
+        if key is None or found[key] is None:
+            continue
+        clusters = _CONTAINER_READERS[key](name, records)
+        if clusters is None:
+            found[key] = None
+        else:
+            found[key].extend(clusters)  # type: ignore[union-attr]
+    return found
+
+
+def parse_service_clusters(output: str) -> ServiceClustersDict:
+    """Parse what :func:`service_clusters_command` printed.
+
+    :raises ValueError: when the output carries no intact report.
+    """
+    sections = _sections(_frame(output))
+    patroni: Optional[List[ServiceClusterDict]]
+    if any(kind == "no-systemd" for kind, _, _ in sections):
+        patroni = None
+    elif any(kind == "patroni" and any(r[0] == "unreadable" for r in records) for kind, _, records in sections):
+        patroni = None
+    else:
+        patroni = [
+            c for kind, unit, records in sections if kind == "patroni" for c in _patroni(unit, records)
+        ]
+    containers = _containers(sections)
+    return {
+        "patroni": patroni,
+        "adguardhome_sync": containers["adguardhome_sync"],
+        "bunkerweb": containers["bunkerweb"],
+    }
+
+
+class ServiceClustersMixin:
+    """Adds :meth:`get_service_clusters` to a driver that can run a command on a Linux host.
+
+    The template form (README, "Function classes"): the command and its parse are
+    the same everywhere, so they are concrete here, and a driver supplies only
+    :meth:`_run_service_clusters_command` -- how a command reaches its host, and
+    how it gains root there. A driver with a container engine of its own path
+    (``ContainerEngineMixin._container_engine_binary``) has it asked with that.
+    Mixed in by the drivers that can, so ``hasattr(driver,
+    "get_service_clusters")`` stays a truthful answer.
+    """
+
+    if TYPE_CHECKING:  # pragma: no cover - declared for type checkers only
+
+        def _run_service_clusters_command(self, command: str, *, privileged: bool) -> str:
+            """Run *command* on the host and return what it printed; as root when
+            *privileged*. The command is a single ``sh -c`` invocation of a few kB:
+            send it on an exec channel, not typed into a shell."""
+            ...
+
+    def _service_clusters_docker(self) -> str:
+        binary = getattr(self, "_container_engine_binary", None)
+        return binary("docker") if callable(binary) else "docker"
+
+    def get_service_clusters(self) -> ServiceClustersDict:
+        """
+        Returns the clusters the host's configuration says it belongs to, per
+        mechanism: Patroni (its scope and the members its DCS lists),
+        adguardhome-sync (its origin and replicas), BunkerWeb (a worker's manager,
+        a manager's instances). See :class:`~napalm_device_types.models.ServiceClusterDict`.
+
+        None for a mechanism means it could not be read; ``[]`` that the host has none.
+
+        Example::
+
+            {
+                "patroni": [
+                    {"mechanism": "patroni", "service": "patroni", "source": "patroni",
+                     "cluster": "netork-db",
+                     "running": True,
+                     "members": [
+                         {"name": "db-01", "address": "10.7.234.12", "port": 5432,
+                          "role": "leader", "state": "running", "this_host": True,
+                          "config_from": None},
+                         {"name": "db-02", "address": "10.7.234.11", "port": 5432,
+                          "role": "sync_standby", "state": "streaming", "this_host": False,
+                          "config_from": None},
+                     ]},
+                ],
+                "adguardhome_sync": [],
+                "bunkerweb": [],
+            }
+
+        :raises ValueError: if neither reading carried an intact report.
+        """
+        command = f"sh -c {quote(service_clusters_command(self._service_clusters_docker()))}"
+        try:
+            return parse_service_clusters(self._run_service_clusters_command(command, privileged=True))
+        except ValueError:
+            pass
+        return parse_service_clusters(self._run_service_clusters_command(command, privileged=False))

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "napalm-device-types"
-version = "4.16.0"
+version = "4.17.0"
 description = "Abstract device-type base classes for NAPALM drivers"
 readme = "README.md"
 requires-python = ">=3.10"

tests/test_service_clusters.py

@@ -0,0 +1,491 @@
+"""get_service_clusters: the configuration that says which services run together.
+
+Two hosts that answer at one virtual IP may run the same service by accident.
+What proves a cluster is the configuration the members share: Patroni's scope
+and the members its DCS lists, adguardhome-sync's origin and replicas, the
+address a BunkerWeb worker takes its configuration from. Reading that is the
+same on every Linux host, so the command and its parse live here once, and a
+driver only carries the command across (netOrk #1125).
+
+The same configuration holds secrets -- Patroni's superuser password, the sync's
+AdGuard logins, a database URI -- so the command filters on the host: only
+allowlisted fields ever leave it. The tests below run the command for real to
+hold it to that.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import shutil
+import subprocess
+from pathlib import Path
+
+import pytest
+
+from napalm_device_types.service_clusters import (
+    SERVICE_CLUSTERS_COMMAND,
+    ServiceClustersMixin,
+    parse_service_clusters,
+    service_clusters_command,
+)
+
+SECRET = "s3cr3t-do-not-leak"
+
+
+def _wire(*sections: str, systemd: bool = True, docker: str = "[docker]", noise: str = "") -> str:
+    head = "[systemd]" if systemd else "[no-systemd]"
+    body = "\n".join([head, *sections[:1], docker, *sections[1:]]) if sections else f"{head}\n{docker}"
+    return f"{noise}SVCC_BEGIN\n{body}\nSVCC_END\n"
+
+
+DB01_CLUSTER = {
+    "members": [
+        {"name": "db-01", "role": "leader", "state": "running", "host": "10.7.234.12", "port": 5432},
+        {"name": "db-02", "role": "sync_standby", "state": "streaming", "host": "10.7.234.11", "port": 5432, "lag": 0},
+    ],
+    "scope": "netork-db",
+}
+
+
+def _patroni(*records: str, unit: str = "patroni.service") -> str:
+    return "\n".join([f"[patroni {unit}]", *records])
+
+
+def _container(name: str, image: str, *env: str) -> str:
+    return "\n".join([f"[container {name}]", f"image {image}", *(f"env {e}" for e in env)])
+
+
+class TestPatroni:
+    def test_the_members_its_dcs_lists_with_their_roles(self):
+        reading = parse_service_clusters(
+            _wire(
+                _patroni(
+                    "active active",
+                    "config /etc/patroni/config.yml",
+                    "scope netork-db",
+                    "name db-01",
+                    "restapi 10.7.234.12:8008",
+                    "rest " + json.dumps(DB01_CLUSTER),
+                )
+            )
+        )
+
+        assert reading["patroni"] == [
+            {
+                "mechanism": "patroni",
+                "service": "patroni",
+                "source": "patroni",
+                "cluster": "netork-db",
+                "running": True,
+                "members": [
+                    {"name": "db-01", "address": "10.7.234.12", "port": 5432, "role": "leader",
+                     "state": "running", "this_host": True, "config_from": None},
+                    {"name": "db-02", "address": "10.7.234.11", "port": 5432, "role": "sync_standby",
+                     "state": "streaming", "this_host": False, "config_from": None},
+                ],
+            }
+        ]
+
+    def test_without_its_rest_api_it_is_itself_in_its_scope(self):
+        reading = parse_service_clusters(
+            _wire(
+                _patroni(
+                    "active active", "scope 'netork-db'  # the cluster", "name db-01",
+                    "restapi 10.7.234.12:8008", "rest-unreachable",
+                )
+            )
+        )
+
+        (cluster,) = reading["patroni"]
+        assert cluster["cluster"] == "netork-db"
+        assert cluster["members"] == [
+            {"name": "db-01", "address": "10.7.234.12", "port": None, "role": None,
+             "state": None, "this_host": True, "config_from": None}
+        ]
+
+    def test_an_unreadable_config_is_not_looked_at(self):
+        reading = parse_service_clusters(
+            _wire(_patroni("active active", "unreadable /etc/patroni/config.yml"))
+        )
+
+        assert reading["patroni"] is None
+
+    def test_a_config_without_a_scope_declares_no_cluster(self):
+        assert parse_service_clusters(_wire(_patroni("active active", "name db-01")))["patroni"] == []
+
+    def test_a_host_without_systemd_is_not_looked_at(self):
+        assert parse_service_clusters(_wire(systemd=False))["patroni"] is None
+
+    def test_no_patroni_is_looked_and_found_nothing(self):
+        assert parse_service_clusters(_wire())["patroni"] == []
+
+
+class TestAdguardhomeSync:
+    def test_the_origin_pushes_to_each_replica(self):
+        reading = parse_service_clusters(
+            _wire(
+                "",
+                _container(
+                    "adguardhome-sync", "ghcr.io/bakito/adguardhome-sync:v0.9.3",
+                    "ORIGIN_URL http://10.9.100.5:3000", "REPLICA1_URL http://10.9.100.6:3000",
+                    "REPLICA2_URL https://dns-03.example.org",
+                ),
+            )
+        )
+
+        (cluster,) = reading["adguardhome_sync"]
+        assert (cluster["mechanism"], cluster["service"], cluster["cluster"]) == (
+            "adguardhome-sync", "adguardhome", "10.9.100.5:3000",
+        )
+        assert [(m["address"], m["port"], m["role"], m["config_from"]) for m in cluster["members"]] == [
+            ("10.9.100.5", 3000, "origin", None),
+            ("10.9.100.6", 3000, "replica", 0),
+            ("dns-03.example.org", 443, "replica", 0),
+        ]
+        assert not any(m["this_host"] for m in cluster["members"])
+
+    def test_a_single_replica_url_counts(self):
+        reading = parse_service_clusters(
+            _wire("", _container("sync", "bakito/adguardhome-sync", "ORIGIN_URL http://a:3000",
+                                 "REPLICA_URL http://b:3000"))
+        )
+
+        assert [m["address"] for m in reading["adguardhome_sync"][0]["members"]] == ["a", "b"]
+
+    def test_a_sync_configured_by_file_is_not_looked_at(self):
+        """Its YAML is not read: the cluster is there, unseen."""
+        reading = parse_service_clusters(_wire("", _container("sync", "bakito/adguardhome-sync")))
+
+        assert reading["adguardhome_sync"] is None
+
+
+class TestBunkerweb:
+    def test_a_worker_takes_its_configuration_from_the_manager_it_admits(self):
+        reading = parse_service_clusters(
+            _wire("", _container("bunkerweb", "bunkerity/bunkerweb@sha256:2026", "API_WHITELIST_IP 10.50.100.5"))
+        )
+
+        (cluster,) = reading["bunkerweb"]
+        assert (cluster["service"], cluster["cluster"]) == ("bunkerweb", "10.50.100.5")
+        assert cluster["members"] == [
+            {"name": None, "address": None, "port": 5000, "role": "worker", "state": None,
+             "this_host": True, "config_from": 1},
+            {"name": None, "address": "10.50.100.5", "port": None, "role": "manager", "state": None,
+             "this_host": False, "config_from": None},
+        ]
+
+    def test_the_default_whitelist_names_no_manager(self):
+        """127.0.0.0/8 and a docker network are BunkerWeb's default, not a peer."""
+        reading = parse_service_clusters(
+            _wire("", _container("bw", "bunkerity/bunkerweb:1.6", "API_WHITELIST_IP 127.0.0.0/8 10.20.30.0/24"))
+        )
+
+        assert reading["bunkerweb"] == []
+
+    def test_a_manager_names_its_instances(self):
+        reading = parse_service_clusters(
+            _wire("", _container("bw", "bunkerity/bunkerweb-all-in-one:1.6",
+                                 "BUNKERWEB_INSTANCES 10.50.100.6:5000 bw-03")),
+        )
+
+        (cluster,) = reading["bunkerweb"]
+        assert cluster["cluster"] is None  # the manager cannot name its own address
+        assert [(m["address"], m["port"], m["role"], m["this_host"], m["config_from"])
+                for m in cluster["members"]] == [
+            (None, None, "manager", True, None),
+            ("10.50.100.6", 5000, "worker", False, 0),
+            ("bw-03", 5000, "worker", False, 0),
+        ]
+
+    def test_a_manager_without_instances_is_no_cluster(self):
+        """proxy-01's all-in-one keeps its instances in its database."""
+        reading = parse_service_clusters(_wire("", _container("bw", "bunkerity/bunkerweb-all-in-one@sha256:2b34")))
+
+        assert reading["bunkerweb"] == []
+
+    def test_its_ui_and_scheduler_side_containers_are_not_instances(self):
+        reading = parse_service_clusters(
+            _wire("", _container("ui", "bunkerity/bunkerweb-ui:1.6", "API_WHITELIST_IP 10.0.0.9"))
+        )
+
+        assert reading["bunkerweb"] == []
+
+
+class TestTheReport:
+    def test_a_host_without_docker_runs_no_containers(self):
+        reading = parse_service_clusters(_wire(docker="[no-docker]"))
+
+        assert (reading["adguardhome_sync"], reading["bunkerweb"]) == ([], [])
+
+    def test_a_docker_it_may_not_ask_is_not_looked_at(self):
+        reading = parse_service_clusters(_wire(docker="[docker-unreadable]"))
+
+        assert (reading["adguardhome_sync"], reading["bunkerweb"]) == (None, None)
+
+    def test_noise_before_the_report_is_ignored(self):
+        assert parse_service_clusters(_wire(noise="Last login: today\n"))["patroni"] == []
+
+    def test_a_report_cut_short_raises_without_quoting_it(self):
+        with pytest.raises(ValueError) as caught:
+            parse_service_clusters("SVCC_BEGIN\n[systemd]\nscope netork-db\n")
+
+        assert "netork-db" not in str(caught.value)
+
+
+# ---------------------------------------------------------------------------
+# The command itself, run against stubs and real files
+# ---------------------------------------------------------------------------
+
+PATRONI_YML = f"""scope: netork-db
+namespace: /service/
+name: db-01
+
+restapi:
+  listen: 10.7.234.12:8008
+  connect_address: 10.7.234.12:8008
+  authentication:
+    username: patroni
+    password: {SECRET}
+
+etcd3:
+  hosts:
+    - 10.7.234.12:2379
+    - 10.7.234.13:2379
+  password: {SECRET}
+
+postgresql:
+  connect_address: 10.7.234.12:5432
+  authentication:
+    superuser:
+      username: postgres
+      password: {SECRET}
+tags:
+  name: not-the-member-name
+"""
+
+
+def _awks() -> list:
+    found = []
+    for name, argv in (("mawk", ["mawk"]), ("busybox", ["busybox", "awk"]), ("gawk", ["gawk"])):
+        if shutil.which(argv[0]):
+            found.append(pytest.param(argv, id=name))
+    return found or [pytest.param(None, marks=pytest.mark.skip(reason="no awk to run"))]
+
+
+class _Host:
+    """A host as far as the command can tell: units, containers, a REST API, files."""
+
+    def __init__(self, root: Path, awk: list) -> None:
+        self.root = root
+        self.bin = root / "bin"
+        self.bin.mkdir()
+        self.state = root / "state"
+        self.state.mkdir()
+        self.booted = root / "run-systemd-system"
+        self.booted.mkdir()
+        (self.state / "units").write_text("")
+        (self.state / "containers").write_text("")
+        cat = shutil.which("cat")
+        self._stub(
+            "systemctl",
+            'case "$1" in\n'
+            f'  list-units) {cat} "$d/units" ;;\n'
+            f'  show) for last; do :; done; {cat} "$d/show-$last" ;;\n'
+            "esac\n",
+        )
+        self._stub(
+            "docker",
+            'case "$1" in\n'
+            f'  ps) {cat} "$d/containers" ;;\n'
+            '  inspect) for last; do :; done;\n'
+            '    case "$3" in *Config.Image*) '
+            f'{cat} "$d/image-$last" ;; *) {cat} "$d/env-$last" ;; esac ;;\n'
+            "esac\n",
+        )
+        self._stub("curl", f'for last; do :; done; {cat} "$d/rest" 2>/dev/null || exit 7\n')
+        for tool in ("sed", "head", "tr"):
+            if shutil.which(tool):
+                (self.bin / tool).symlink_to(shutil.which(tool))
+        wrapper = self.bin / "awk"
+        program = " ".join([shutil.which(awk[0]) or awk[0], *awk[1:]])
+        wrapper.write_text(f'#!/bin/sh\nexec {program} "$@"\n')
+        wrapper.chmod(0o755)
+
+    def _stub(self, name: str, body: str) -> None:
+        stub = self.bin / name
+        stub.write_text(f"#!/bin/sh\nd={self.state}\n{body}")
+        stub.chmod(0o755)
+
+    def unit(self, name: str, *properties: str) -> None:
+        with (self.state / "units").open("a") as units:
+            units.write(f"{name} loaded active running {name}\n")
+        (self.state / f"show-{name}").write_text("".join(f"{p}\n" for p in properties))
+
+    def container(self, name: str, image: str, *env: str) -> None:
+        with (self.state / "containers").open("a") as containers:
+            containers.write(f"{name}\n")
+        (self.state / f"image-{name}").write_text(f"{image}\n")
+        (self.state / f"env-{name}").write_text("".join(f"{e}\n" for e in env))
+
+    def rest(self, body: dict) -> None:
+        (self.state / "rest").write_text(json.dumps(body))
+
+    def file(self, relative: str, content: str) -> Path:
+        path = self.root / relative
+        path.parent.mkdir(parents=True, exist_ok=True)
+        path.write_text(content)
+        return path
+
+    def run(self, *, docker: bool = True) -> str:
+        command = service_clusters_command("docker" if docker else None)
+        assert command.count("/run/systemd/system") == 1
+        command = command.replace("/run/systemd/system", str(self.booted))
+        return subprocess.run(
+            ["/bin/sh", "-c", command],
+            capture_output=True,
+            text=True,
+            env={"PATH": str(self.bin)},
+            timeout=30,
+        ).stdout
+
+
+def _exec_start(*argv: str) -> str:
+    return f"ExecStart={{ path={argv[0]} ; argv[]={' '.join(argv)} ; ignore_errors=no ; pid=7 ; status=0/0 }}"
+
+
+@pytest.fixture(params=_awks())
+def host(request, tmp_path) -> _Host:
+    return _Host(tmp_path, request.param)
+
+
+class TestTheCommandOnAHost:
+    def test_patroni_as_on_db_01(self, host):
+        config = host.file("etc/patroni/config.yml", PATRONI_YML)
+        host.unit("patroni.service", "ActiveState=active", _exec_start("/usr/bin/patroni", str(config)))
+        host.rest(DB01_CLUSTER)
+
+        out = host.run()
+
+        assert SECRET not in out
+        assert "10.7.234.13" not in out  # the DCS endpoints stay on the host
+        assert "not-the-member-name" not in out
+        (cluster,) = parse_service_clusters(out)["patroni"]
+        assert cluster["cluster"] == "netork-db"
+        assert [(m["name"], m["role"], m["this_host"]) for m in cluster["members"]] == [
+            ("db-01", "leader", True),
+            ("db-02", "sync_standby", False),
+        ]
+
+    def test_patroni_whose_rest_api_does_not_answer(self, host):
+        config = host.file("etc/patroni/config.yml", PATRONI_YML)
+        host.unit("patroni.service", "ActiveState=active", _exec_start("/usr/bin/patroni", str(config)))
+
+        (cluster,) = parse_service_clusters(host.run())["patroni"]
+
+        assert [(m["name"], m["address"], m["role"]) for m in cluster["members"]] == [
+            ("db-01", "10.7.234.12", None)
+        ]
+
+    @pytest.mark.skipif(os.geteuid() == 0, reason="root reads a file of mode 000")
+    def test_a_patroni_config_it_cannot_read_is_not_looked_at(self, host):
+        config = host.file("etc/patroni/config.yml", PATRONI_YML)
+        config.chmod(0)
+        host.unit("patroni.service", "ActiveState=active", _exec_start("/usr/bin/patroni", str(config)))
+
+        out = host.run()
+
+        assert SECRET not in out
+        assert parse_service_clusters(out)["patroni"] is None
+
+    def test_the_sync_and_a_worker_without_their_secrets(self, host):
+        host.container(
+            "adguardhome-sync", "ghcr.io/bakito/adguardhome-sync:v0.9.3",
+            "ORIGIN_URL=http://10.9.100.5:3000", f"ORIGIN_PASSWORD={SECRET}",
+            f"REPLICA1_URL=http://admin:{SECRET}@10.9.100.6:3000", f"REPLICA1_PASSWORD={SECRET}",
+            "CRON=*/5 * * * *",
+        )
+        host.container(
+            "bunkerweb", "bunkerity/bunkerweb@sha256:2026",
+            "API_WHITELIST_IP=10.50.100.5", f"DATABASE_URI=mariadb+pymysql://bw:{SECRET}@db/bw",
+        )
+        host.container("zabbix", "zabbix/zabbix-agent2:alpine-latest", f"ZBX_PSK={SECRET}")
+
+        out = host.run()
+
+        assert SECRET not in out
+        assert "admin" not in out
+        reading = parse_service_clusters(out)
+        assert [m["address"] for m in reading["adguardhome_sync"][0]["members"]] == [
+            "10.9.100.5", "10.9.100.6",
+        ]
+        assert reading["bunkerweb"][0]["cluster"] == "10.50.100.5"
+        assert reading["patroni"] == []
+
+    def test_a_host_without_docker(self, host):
+        reading = parse_service_clusters(host.run(docker=False))
+
+        assert (reading["adguardhome_sync"], reading["bunkerweb"]) == ([], [])
+
+
+class TestTheMixin:
+    class _Driver(ServiceClustersMixin):
+        def __init__(self, answers: dict) -> None:
+            self.answers = answers
+            self.calls: list = []
+
+        def _run_service_clusters_command(self, command: str, *, privileged: bool) -> str:
+            self.calls.append(privileged)
+            return self.answers[privileged]
+
+    def test_root_first(self):
+        driver = self._Driver({True: _wire(), False: "unused"})
+
+        assert driver.get_service_clusters()["patroni"] == []
+        assert driver.calls == [True]
+
+    def test_without_root_when_root_brought_no_report(self):
+        driver = self._Driver({True: "sudo: a password is required", False: _wire()})
+
+        assert driver.get_service_clusters()["bunkerweb"] == []
+        assert driver.calls == [True, False]
+
+    def test_the_container_engine_binary_is_the_drivers(self):
+        class _Qnap(self._Driver):
+            def _container_engine_binary(self, engine: str) -> str:
+                return "/share/CACHEDEV1_DATA/.qpkg/container-station/bin/docker"
+
+        driver = _Qnap({True: _wire(), False: _wire()})
+        seen: list = []
+
+        def run(command: str, *, privileged: bool) -> str:
+            seen.append(command)
+            return _wire()
+
+        driver._run_service_clusters_command = run  # type: ignore[method-assign]
+        driver.get_service_clusters()
+
+        assert "container-station/bin/docker" in seen[0]
+
+    def test_the_default_command_asks_docker(self):
+        assert "docker" in SERVICE_CLUSTERS_COMMAND
+
+
+class TestTheSource:
+    """Where the evidence was read: the unit or container that carries it."""
+
+    def test_a_template_unit_is_named_after_its_instance(self):
+        reading = parse_service_clusters(
+            _wire(_patroni("active active", "scope pg", "name a", unit="patroni@16-main.service"))
+        )
+
+        assert reading["patroni"][0]["source"] == "patroni@16-main"
+
+    def test_a_container_by_its_name(self):
+        reading = parse_service_clusters(
+            _wire("", _container("adguardhome-sync", "bakito/adguardhome-sync", "ORIGIN_URL http://a:3000"),
+                  _container("bw", "bunkerity/bunkerweb", "API_WHITELIST_IP 10.0.0.5"))
+        )
+
+        assert reading["adguardhome_sync"][0]["source"] == "adguardhome-sync"
+        assert reading["bunkerweb"][0]["source"] == "bw"