#!/usr/bin/env python3 # -*- coding: utf-8 -*- """check-probe-matches-compose.py — the `.felhom.yml` probe must dial where the app LISTENS (R-618). WHAT WENT WRONG, and why a static gate can catch it. `.felhom.yml`'s `healthcheck.checks` tells the controller where to knock: it dials `:` on the compose network. The port is therefore **the port the process listens on INSIDE the container** — not the published port, not the Traefik port. Three templates named a port or a path the app does not answer. On its own that is only a wrong badge; what makes it P1 is that **the guarded update's `verifying` phase waits on that same probe**, so a SUCCESSFUL update ends with `failAndHold` stopping a working app. Measured on 2026-09-21: tandoor served HTTP 200 on the new version at four samples over five minutes, docker's own healthcheck green, and the controller stopped it at +361.9 s. THE ORACLE. The compose `healthcheck.test` of the very same service already dials the app on loopback — `wget http://127.0.0.1:80/accounts/login/`. That line is written by whoever added the template, is exercised by docker on every start, and is therefore known-good. So the two can be compared, statically, with no network and no container: probe port == the port the compose healthcheck dials on 127.0.0.1 probe path == the path it dials (when the probe can FAIL on it) WHICH SERVICE. The one the controller would actually probe. `findProbeContainer` (`controller/internal/stacks/healthprobe.go:297`) takes the container whose name EQUALS the stack name, else the first whose name has it as a PREFIX. This gate resolves the same way, off `container_name`. THE TWO VERDICTS, and why the path rule is narrower than the port rule — this is a measurement, not a compromise: * **A wrong PORT is always fatal.** Nothing listens there, the dial is refused, every check fails. REFUSE, for every check type including `tcp`. * **A wrong PATH is fatal only if the probe can fail on it.** `probeHTTP` returns healthy for ANY response when the type is `http`, and also when the type is `api` with **no `expect` block** (healthprobe.go:253-262). So a path mismatch bites exactly when the type is `api` AND `expect` is present. REFUSE there; WARN otherwise. Run over all 53 on 2026-09-22, that is the difference between convicting `zipline` (`api` + `expect.status: 200`, path `/api/health` vs the app's real `/api/healthcheck`) and merely warning about `home-assistant` (`api`, no `expect`, `/api/` vs `/manifest.json`) — which is honest: home-assistant's badge is right today and would break the day someone adds an `expect`. WHAT IT CANNOT SEE — named, not implied: 1. **A template whose compose healthcheck is ALSO wrong.** The oracle is the compose line; if both agree and both are wrong, this gate passes. Only a live deploy decides that. 2. **A path that is right but answers the wrong STATUS.** `expect.status: 200` against an endpoint that 302s is a real fault and looks identical to a correct template here. 3. **Anything about a service other than the probed one.** Each of these three is a WARN category below rather than a silent pass, wherever it is visible. USAGE python3 scripts/check-probe-matches-compose.py # every app python3 scripts/check-probe-matches-compose.py tandoor wger # only these python3 scripts/check-probe-matches-compose.py --root= # judge another checkout Exit: 0 clean (warnings do not convict) · 1 convicted · 2 inconclusive (unreadable template). """ import os import re import sys try: import yaml except ImportError: print("check-probe-matches-compose: PyYAML is not installed — INCONCLUSIVE") sys.exit(2) ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) TEMPLATES = os.path.join(ROOT, "templates") def set_root(root): """Point the gate at a different checkout — `--root=`. Without this the decoy suite would run the gate against the REAL repo while editing a clone, so every case would read the same unchanged files and every case would pass. That is the `constant-for-measurement` decoy shape this repo's own suite warns about, and it has fooled a gate here before. """ global TEMPLATES TEMPLATES = os.path.join(root, "templates") # The names that mean "this container, from inside it". `0.0.0.0` appears as a dial target in a # couple of templates; it is a bind address rather than a destination, but it reaches the same # listener and the PORT is what this gate compares. LOOPBACK = ("127.0.0.1", "localhost", "0.0.0.0", "::1", "[::1]") def flatten_test(test): """The compose healthcheck test as one string, whatever shape it was written in. Both `["CMD", "wget", …]` and `["CMD-SHELL", "curl … || exit 1"]` and a bare string occur in this catalog. The leading CMD/CMD-SHELL word is dropped; everything else is joined, because the URL may sit in any argument (node's `-e` script puts it inside quotes). """ if test is None: return None if isinstance(test, str): return test parts = list(test) if parts and parts[0] in ("CMD", "CMD-SHELL", "NONE"): parts = parts[1:] return " ".join(str(p) for p in parts) URL_RE = re.compile(r'https?://([A-Za-z0-9_.\-\[\]:]+?)(?::(\d+))?(/[^\s\'"`)\\,]*)?(?=[\s\'"`)\\,]|$)') NC_RE = re.compile(r'\bnc\b[^\n;|&]*?(\S+)\s+(\d+)\b') def endpoints(test_str): """Every loopback endpoint the compose healthcheck dials: (scheme, port, path).""" found = [] for m in URL_RE.finditer(test_str): host, port, path = m.group(1), m.group(2), m.group(3) if host not in LOOPBACK: continue p = int(port) if port else (443 if m.group(0).startswith("https") else 80) found.append(("http", p, path or "/")) for m in NC_RE.finditer(test_str): if m.group(1) in LOOPBACK: found.append(("tcp", int(m.group(2)), None)) return found def probed_service(app, services): """The service the controller would probe — `findProbeContainer`'s rule, statically.""" for name, svc in services.items(): if (svc or {}).get("container_name") == app: return name, svc for name, svc in services.items(): cn = str((svc or {}).get("container_name") or "") if cn.startswith(app): return name, svc return None, None def check_app(app): """Returns (verdicts, warnings) — each a list of one-line strings.""" d = os.path.join(TEMPLATES, app) bad, warn = [], [] try: fy = yaml.safe_load(open(os.path.join(d, ".felhom.yml"), encoding="utf-8")) or {} cy = yaml.safe_load(open(os.path.join(d, "docker-compose.yml"), encoding="utf-8")) or {} except Exception as e: # noqa: BLE001 — report, never crash return None, ["%s: template unreadable (%s)" % (app, e)] checks = ((fy.get("healthcheck") or {}).get("checks")) or [] if not checks: warn.append("%s: no `healthcheck.checks` in .felhom.yml — nothing to compare" % app) return bad, warn services = cy.get("services") or {} svc_name, svc = probed_service(app, services) if svc is None: # Not a conviction, but it is worse than a mismatch: `findProbeContainer` returns "" and # the stack is SKIPPED, so no probe ever runs and the badge can never go red. Loud WARN. warn.append("%s: NO container_name equals or begins with the stack name — the controller " "would find no container to probe and SKIP this stack silently " "(healthprobe.go:62). Containers: %s" % (app, ", ".join(sorted(str((s or {}).get("container_name")) for s in services.values())))) return bad, warn test = flatten_test((svc.get("healthcheck") or {}).get("test")) if not test: warn.append("%s: service %r has no compose healthcheck — this gate has no oracle for it; " "only a live deploy can judge the probe" % (app, svc_name)) return bad, warn eps = endpoints(test) if not eps: has_ports = bool(svc.get("ports") or svc.get("expose")) warn.append("%s: the compose healthcheck of %r dials no loopback URL or port (%s) — " "no oracle.%s" % (app, svc_name, test.strip()[:90], (" Only `expose:`/`ports:` are declared, and a published " "port is NOT the port the app listens on inside the " "container, so it cannot stand in.") if has_ports else "")) return bad, warn for c in checks: ctype = (c or {}).get("type") cport = (c or {}).get("port") cpath = (c or {}).get("path") or "/" # `tcp` is compared on the port alone — it dials a socket and has no path. same_scheme = [e for e in eps if (e[0] == "tcp") == (ctype == "tcp")] or eps ports = sorted({e[1] for e in same_scheme}) if cport not in ports: bad.append("%s: probe %s port %s, but the compose healthcheck of %r dials %s on " "loopback. Nothing listens on %s inside the container, so EVERY probe fails " "and a successful update is stopped by failAndHold (R-618). test: %s" % (app, ctype, cport, svc_name, "/".join(map(str, ports)), cport, test.strip()[:90])) continue if ctype == "tcp": continue paths = sorted({e[2] for e in same_scheme if e[1] == cport and e[2] is not None}) if paths and cpath not in paths: can_fail = ctype == "api" and (c or {}).get("expect") msg = ("%s: probe %s path %r, but the compose healthcheck of %r dials %s on the same " "port. test: %s" % (app, ctype, cpath, svc_name, " or ".join(repr(p) for p in paths), test.strip()[:90])) if can_fail: bad.append(msg + " This probe CAN fail on it: type `api` with an `expect` block " "compares the response (healthprobe.go:253-262).") else: warn.append(msg + " Harmless TODAY — type %r without `expect` calls ANY response " "healthy — and fatal the day someone adds `expect`." % ctype) return bad, warn def main(argv): for a in argv: if a.startswith("--root="): set_root(a[len("--root="):]) unknown = [a for a in argv if a.startswith("-") and not a.startswith("--root=")] if unknown: print("unknown option(s): %s" % " ".join(unknown)) return 2 apps = [a for a in argv if not a.startswith("-")] if not apps: apps = sorted(d for d in os.listdir(TEMPLATES) if os.path.isdir(os.path.join(TEMPLATES, d))) print("probe-matches-compose: %d app(s)" % len(apps)) convicted, warnings, unreadable = [], [], 0 for app in sorted(apps): bad, warn = check_app(app) if bad is None: unreadable += 1 warnings += warn continue convicted += bad warnings += warn if warnings: print("\n%d WARNING(S) — not a conviction, and not a pass either:" % len(warnings)) for w in warnings: print(" WARN " + w) if unreadable: print("\nprobe-matches-compose: %d template(s) unreadable — INCONCLUSIVE" % unreadable) return 2 if convicted: print("\nprobe-matches-compose: CONVICTED — %d probe(s) do not match the app" % len(convicted)) for b in convicted: print(" FAIL " + b) return 1 print("\nprobe-matches-compose: OK — every probe dials the port (and, where it can fail, the " "path) that the app's own compose healthcheck dials") return 0 if __name__ == "__main__": sys.exit(main(sys.argv[1:]))