family_home_lab/deploy/service-controller/service_controller.py

"""service-controller — host-side on/off control for the Family Console.

Only allowlisted services can be started/stopped. Requires a token
(X-Controller-Token). SC_HOST default 127.0.0.1; the systemd unit overrides it
with the docker bridge gateway so the portal container can reach us via
host.docker.internal while staying unreachable from the LAN. Covers Docker containers (sam is in the docker
group) and systemd USER services. No generic passthrough — unknown id => 404.

API (see ON_OFF.md §4.2):
  GET  /health
  GET  /services
  GET  /services/{id}
  POST /services/{id}/start
  POST /services/{id}/stop

Runs as a systemd USER unit: family-service-controller.service
"""
from __future__ import annotations

import json
import os
import shlex
import subprocess
import threading
import time
import tomllib
from pathlib import Path
from typing import Annotated, Any

import uvicorn
from fastapi import FastAPI, Header, HTTPException, Request
from fastapi.responses import JSONResponse

HERE = Path(__file__).resolve().parent
SERVICES_FILE = Path(os.getenv("SC_SERVICES", HERE / "services.toml"))
AUDIT_FILE = Path(os.getenv("SC_AUDIT", HERE / "audit.log"))
AUDIT_MAX_LINES = int(os.getenv("SC_AUDIT_MAX", "500"))
HOST = os.getenv("SC_HOST", "127.0.0.1")
PORT = int(os.getenv("SC_PORT", "8443"))  # in NixOS firewall allowlist; 8099 was not
TOKEN = os.getenv("SC_TOKEN", "")  # set by the unit's EnvironmentFile (10-secrets.conf)

app = FastAPI(title="Family service-controller", docs_url=None)
_audit_lock = threading.Lock()


# --------------------------------------------------------------------------- #
# allowlist
# --------------------------------------------------------------------------- #
def load_services() -> list[dict[str, Any]]:
    with open(SERVICES_FILE, "rb") as fh:
        data = tomllib.load(fh)
    return [dict(s) for s in data.get("services", [])]


SERVICES: list[dict[str, Any]] = load_services()
BY_ID = {s["id"]: s for s in SERVICES}


def _auth(x_token: Annotated[str | None, Header(alias="X-Controller-Token")] = None) -> None:
    if not TOKEN or not x_token or x_token != TOKEN:
        raise HTTPException(status_code=401, detail="unauthorized")


def audit(what: str) -> None:
    """Append one bounded audit line (oldest dropped at AUDIT_MAX_LINES)."""
    line = f"{time.strftime('%Y-%m-%dT%H:%M:%S%z')} {what}"
    with _audit_lock:
        try:
            lines = AUDIT_FILE.read_text().splitlines() if AUDIT_FILE.exists() else []
        except Exception:
            lines = []
        lines.append(line)
        if len(lines) > AUDIT_MAX_LINES:
            lines = lines[-AUDIT_MAX_LINES:]
        AUDIT_FILE.write_text("\n".join(lines) + "\n")


# --------------------------------------------------------------------------- #
# probes / actions
# --------------------------------------------------------------------------- #
def _container_state(target: str) -> str:
    r = subprocess.run(["docker", "inspect", "-f", "{{.State.Status}}", target],
                       capture_output=True, text=True)
    out = r.stdout.strip()
    return out or ("unknown" if r.returncode != 0 else "unknown")


SSH_OPTS = ["-o", "StrictHostKeyChecking=no", "-o", "BatchMode=yes"]


def _remote(cmd: list[str], remote: str | None) -> list[str]:
    """Wrap a command in ssh when the target service lives on another host.

    ssh joins its trailing arguments with spaces before the remote shell parses
    them, so the command is quoted as one string first — otherwise remote
    commands containing quotes or shell syntax (e.g. the `sh -c` used by
    systemd-system) get word-split and fail.
    """
    return ["ssh", *SSH_OPTS, remote, shlex.join(cmd)] if remote else cmd


def _systemd_state(target: str, remote: str | None = None, system: bool = False) -> str:
    """`is-active` needs no privilege, so this never uses sudo."""
    cmd = ["systemctl", "is-active", target] if system else ["systemctl", "--user", "is-active", target]
    r = subprocess.run(_remote(cmd, remote), capture_output=True, text=True)
    out = (r.stdout.strip() or "inactive").lower()
    return "running" if out == "active" else out


def _do(kind: str, action: str, target: str, remote: str | None = None) -> str:
    """Run start/stop. Returns the state after a short settle.

    Kind semantics:
      container       -> docker <action> <target>          (sam is in the docker group)
      systemd-user    -> systemctl --user <action> <target> (local user units on .13)
      systemd-system  -> sudo -n <canonical systemctl> <action> <target>
                         (system units; `remote` sends it over ssh)

    `systemd-system` requires a NOPASSWD sudoers rule for that exact
    command+argument on the target host. On .27 this is granted for
    `systemctl start|stop|restart ollama` (see hosts/sam-4screen-desktop),
    where `target` must match the rule verbatim: "ollama", not "ollama.service".

    Note: sudo matches the *literal* PATH-resolved path, and NixOS puts a
    symlink (`/run/current-system/sw/bin/systemctl`) on PATH, so a bare
    `sudo systemctl ...` does not match a store-path rule. We therefore resolve
    `systemctl` to its canonical store path first. This also keeps working after
    a nixpkgs bump changes the systemd store hash.
    """
    if kind == "container":
        subprocess.run(["docker", action, target], capture_output=True, text=True)
        time.sleep(1.5)
        return _container_state(target)
    if kind == "systemd-system":
        shell = f'sudo -n "$(readlink -f "$(command -v systemctl)")" {action} {target}'
        cmd = ["sh", "-c", shell]
    else:
        cmd = ["systemctl", "--user", action, target]
    res = subprocess.run(_remote(cmd, remote), capture_output=True, text=True)
    if res.returncode != 0 and kind == "systemd-system":
        audit(f"WARN {kind} {action} {target} exit={res.returncode} {res.stderr.strip()[:160]}")
    time.sleep(1.5)
    return _systemd_state(target, remote=remote, system=(kind == "systemd-system"))


def _targets(s: dict[str, Any]) -> list[str]:
    """A service is either a single `target` or a group `targets` list."""
    t = s.get("targets")
    if isinstance(t, list) and t:
        return [str(x) for x in t]
    single = s.get("target")
    return [single] if single else []


def _probe(s: dict[str, Any]) -> tuple[str, list[str]]:
    """Collect the state of every target. Returns (aggregate, per-target list).
    Aggregate: running if ANY is running; stopped if ALL stopped; else partial."""
    kind = s["kind"]
    remote = s.get("remote")
    states = []
    for tgt in _targets(s):
        if kind == "container":
            states.append(_container_state(tgt))
        else:
            states.append(_systemd_state(tgt, remote=remote, system=(kind == "systemd-system")))
    n_ok = sum(1 for st in states if st in ("running", "active", "started"))
    n_none = sum(1 for st in states if st in ("unknown",)) or 0
    if n_ok == len(states):
        agg = "running"
    elif n_ok == 0 and len(states):
        agg = "stopped"
    else:
        agg = "partial" if len(states) else "unknown"
    return agg, states


def _describe(s: dict[str, Any]) -> dict[str, Any]:
    agg, states = _probe(s)
    return {
        "id": s["id"],
        "kind": s["kind"],
        "target": s.get("target", ""),
        "targets": s.get("targets", []),
        "group": s.get("group", ""),
        "label": s.get("label", s["id"]),
        "ram_mb": s.get("ram_mb"),
        "default_state": s.get("default_state", "stopped"),
        "who": s.get("who", []),
        "state": agg,
        "error": None,
    }


# --------------------------------------------------------------------------- #
# routes
# --------------------------------------------------------------------------- #
@app.get("/health")
async def health(request: Request) -> dict:
    return {"ok": True, "services": len(SERVICES)}


@app.get("/services")
async def services_list(x_token: Annotated[str | None, Header(alias="X-Controller-Token")] = None) -> dict:
    _auth(x_token)
    return {"ok": True, "services": [_describe(s) for s in SERVICES]}


@app.get("/services/{sid}")
async def service_get(sid: str, x_token: Annotated[str | None, Header(alias="X-Controller-Token")] = None) -> dict:
    _auth(x_token)
    s = BY_ID.get(sid)
    if s is None:
        raise HTTPException(status_code=404, detail="unknown service")
    return {"ok": True, "service": _describe(s)}


@app.post("/services/{sid}/{action}")
async def service_action(sid: str, action: str,
                         x_token: Annotated[str | None, Header(alias="X-Controller-Token")] = None,
                         who: Annotated[str | None, Header()] = None) -> dict:
    """Start or stop one service. `who` is the acting user (admin gate is done
    portal-side; the controller logs it and enforces the allowlist)."""
    _auth(x_token)
    if action not in ("start", "stop"):
        raise HTTPException(status_code=400, detail="action must be start|stop")
    s = BY_ID.get(sid)
    if s is None:
        raise HTTPException(status_code=404, detail="unknown service")
    actor = (who or "unknown-user").strip()[:40]
    remote = s.get("remote")
    results = {}
    for tgt in _targets(s):
        results[tgt] = _do(s["kind"], action, tgt, remote=remote)
    agg, _ = _probe(s)
    audit(f"{actor} {action} {sid} -> {agg} ({len(_targets(s))} targets)")
    return {"ok": True, "id": sid, "state": agg, "targets": results}


if __name__ == "__main__":
    uvicorn.run(app, host=HOST, port=PORT)