app-catalog-felhom.eu/scripts/check-probe-matches-compose.py

#!/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
`<container-name>:<port><path>` 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=<dir>  # 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=<dir>`.

    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:]))