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 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")


def _systemd_user_state(target: str) -> str:
    r = subprocess.run(["systemctl", "--user", "is-active", target],
                       capture_output=True, text=True)
    return (r.stdout.strip() or "inactive").lower()


def _do(kind: str, action: str, target: str) -> str:
    """Run start/stop. Returns the state after a short settle."""
    if kind == "container":
        subprocess.run(["docker", action, target], capture_output=True, text=True)
        time.sleep(1.5)
        return _container_state(target)
    subprocess.run(["systemctl", "--user", action, target], capture_output=True, text=True)
    time.sleep(1.5)
    return _systemd_user_state(target)


def _describe(s: dict[str, Any]) -> dict[str, Any]:
    state = "unknown"
    if s["kind"] == "container":
        state = _container_state(s["target"])
    else:
        state = _systemd_user_state(s["target"])
    return {
        "id": s["id"],
        "kind": s["kind"],
        "target": s["target"],
        "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": "running" if state in ("running", "active") else
                ("stopped" if state in ("stopped", "inactive", "exited") else "unknown"),
        "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]
    state = _do(s["kind"], action, s["target"])
    audit(f"{actor} {action} {sid} -> {state}")
    return {"ok": True, "id": sid, "state": state}


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