From 2bd7bcc77c601808402bca9346a5064269c2e7f3 Mon Sep 17 00:00:00 2001 From: Christian Manivong Date: Sun, 18 Oct 2026 15:28:34 +0200 Subject: [PATCH] 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 | 16 + napalm_device_types/__init__.py | 11 + napalm_device_types/models.py | 62 +++ napalm_device_types/service_clusters.py | 528 ++++++++++++++++++++++++ pyproject.toml | 2 +- tests/test_service_clusters.py | 491 ++++++++++++++++++++++ 6 files changed, 1109 insertions(+), 1 deletion(-) create mode 100644 napalm_device_types/service_clusters.py create mode 100644 tests/test_service_clusters.py diff --git a/README.md b/README.md index e6fb8e3..c12f40d 100644 --- a/README.md +++ b/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 diff --git a/napalm_device_types/__init__.py b/napalm_device_types/__init__.py index 596cc8c..4b1cfd6 100644 --- a/napalm_device_types/__init__.py +++ b/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", diff --git a/napalm_device_types/models.py b/napalm_device_types/models.py index cb9e8d9..7cbe4d9 100644 --- a/napalm_device_types/models.py +++ b/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. diff --git a/napalm_device_types/service_clusters.py b/napalm_device_types/service_clusters.py new file mode 100644 index 0000000..fc22784 --- /dev/null +++ b/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 `` and ``config ``, +#: 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]: + """``[ ]`` 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)) diff --git a/pyproject.toml b/pyproject.toml index 25af5dc..84aa952 100644 --- a/pyproject.toml +++ b/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" diff --git a/tests/test_service_clusters.py b/tests/test_service_clusters.py new file mode 100644 index 0000000..2bf8a7f --- /dev/null +++ b/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"