umbrel-maxlab-store/maxlab-probe/probe.py

"""Max Lab Probe - verifiche tecniche della Fase 0.

Server HTTP minimale (solo libreria standard) che, a ogni richiesta,
esegue le verifiche e restituisce un report testuale:
  1. header ricevuti (nome host, X-Forwarded-*)
  2. accesso in sola lettura al socket Docker
  3. accesso in sola lettura a ~/umbrel/app-data e lettura dei manifest
  4. raggiungibilita' dei container delle altre app (saltando app_proxy)
  5. rete condivisa con il servizio "sidecar" (network_mode: service:web)
"""

import http.client
import json
import os
import re
import socket
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

DOCKER_SOCK = "/var/run/docker.sock"
APP_DATA = "/umbrel-app-data"
SIDECAR = ("127.0.0.1", 9000)
LISTEN_PORT = 8080


class UnixHTTPConnection(http.client.HTTPConnection):
    def __init__(self, path, timeout=3):
        super().__init__("localhost", timeout=timeout)
        self.unix_path = path

    def connect(self):
        s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
        s.settimeout(self.timeout)
        s.connect(self.unix_path)
        self.sock = s


def docker_get(path):
    conn = UnixHTTPConnection(DOCKER_SOCK)
    try:
        conn.request("GET", path)
        resp = conn.getresponse()
        return json.loads(resp.read())
    finally:
        conn.close()


def http_status(host, port, path="/"):
    try:
        conn = http.client.HTTPConnection(host, int(port), timeout=2)
        conn.request("GET", path, headers={"Host": f"{host}:{port}"})
        resp = conn.getresponse()
        resp.read(256)
        location = resp.getheader("Location") or ""
        conn.close()
        return f"HTTP {resp.status}" + (f" -> {location}" if location else "")
    except Exception as exc:  # noqa: BLE001 - riportiamo qualunque errore
        return f"ERRORE {type(exc).__name__}: {exc}"


def read_text(path):
    try:
        with open(path, encoding="utf-8", errors="replace") as fh:
            return fh.read()
    except OSError:
        return ""


def find(pattern, text):
    m = re.search(pattern, text, re.MULTILINE)
    return m.group(1).strip().strip("\"'") if m else None


def discover_apps():
    apps = []
    for app_id in sorted(os.listdir(APP_DATA)):
        base = os.path.join(APP_DATA, app_id)
        if not os.path.isdir(base):
            continue
        manifest = read_text(os.path.join(base, "umbrel-app.yml"))
        compose = read_text(os.path.join(base, "docker-compose.yml"))
        apps.append(
            {
                "id": app_id,
                "name": find(r"^name:\s*(.+)$", manifest) or "?",
                "port": find(r"^port:\s*(\d+)", manifest),
                "app_host": find(r"APP_HOST[:=]\s*([^\s#]+)", compose),
                "app_port": find(r"APP_PORT[:=]\s*([^\s#]+)", compose),
            }
        )
    return apps


def build_report(handler):
    out = []
    ok = lambda cond: "OK  " if cond else "FAIL"  # noqa: E731

    out.append("MAX LAB PROBE - verifiche Fase 0")
    out.append("=" * 60)

    # 1. Header ricevuti
    out.append("\n[1] Richiesta ricevuta")
    out.append(f"    client: {handler.client_address[0]}")
    for name in ("Host", "X-Forwarded-For", "X-Forwarded-Proto", "X-Forwarded-Host"):
        out.append(f"    {name}: {handler.headers.get(name, '-')}")

    # 2. Socket Docker
    out.append("\n[2] Socket Docker (sola lettura)")
    containers = []
    try:
        containers = docker_get("/containers/json")
        out.append(f"    {ok(True)} {len(containers)} container in esecuzione")
        me = next(
            (c for c in containers if c["Id"].startswith(socket.gethostname())), None
        )
        if me:
            nets = ", ".join(me["NetworkSettings"]["Networks"].keys())
            out.append(f"    questo container: {me['Names'][0]} - reti: {nets}")
    except Exception as exc:  # noqa: BLE001
        out.append(f"    {ok(False)} {type(exc).__name__}: {exc}")

    # 3. app-data
    out.append("\n[3] Cartella app-data (sola lettura)")
    apps = []
    try:
        apps = discover_apps()
        out.append(f"    {ok(bool(apps))} {len(apps)} app trovate")
    except Exception as exc:  # noqa: BLE001
        out.append(f"    {ok(False)} {type(exc).__name__}: {exc}")

    # 4. Raggiungibilita' container delle app, saltando app_proxy
    out.append("\n[4] Container delle app raggiunti direttamente")
    running = {n.lstrip("/") for c in containers for n in c.get("Names", [])}
    for app in apps:
        if not app["app_host"] or not app["app_port"]:
            out.append(f"    ---- {app['id']}: nessun app_proxy nel compose")
            continue
        host, port = app["app_host"], app["app_port"]
        state = "attivo" if host in running else "non trovato tra i container"
        result = http_status(host, port)
        out.append(
            f"    {app['id']} ({app['name']}) porta UI {app['port']} -> "
            f"{host}:{port} [{state}] {result}"
        )

    # 5. Sidecar con rete condivisa
    out.append("\n[5] Servizio sidecar sulla stessa rete (127.0.0.1:9000)")
    res = http_status(*SIDECAR)
    out.append(f"    {ok(res.startswith('HTTP 200'))} {res}")

    out.append("\nFine report. Disinstalla l'app dopo il test: elenca le tue app.")
    return "\n".join(out) + "\n"


class Handler(BaseHTTPRequestHandler):
    def do_GET(self):  # noqa: N802 - nome imposto da BaseHTTPRequestHandler
        body = build_report(self).encode("utf-8")
        self.send_response(200)
        self.send_header("Content-Type", "text/plain; charset=utf-8")
        self.send_header("Content-Length", str(len(body)))
        self.send_header("Cache-Control", "no-store")
        self.end_headers()
        self.wfile.write(body)

    def log_message(self, fmt, *args):
        print(f"{self.client_address[0]} {fmt % args}", flush=True)


if __name__ == "__main__":
    print(f"Max Lab Probe in ascolto su :{LISTEN_PORT}", flush=True)
    ThreadingHTTPServer(("0.0.0.0", LISTEN_PORT), Handler).serve_forever()