proxmox/services/sankofa-it-read-api/server.py

#!/usr/bin/env python3
"""
Read-only HTTP API for IT inventory JSON (Phase 0 BFF stub).

Serves latest reports/status/live_inventory.json and drift.json from the repo tree.
Optional IT_READ_API_KEY: when set, /v1/* requires header X-API-Key (GET and POST).

Usage (from repo root):
  IT_READ_API_KEY=secret python3 services/sankofa-it-read-api/server.py
  # or
  python3 services/sankofa-it-read-api/server.py   # open /v1 without key

Env:
  IT_READ_API_HOST  (default 127.0.0.1)
  IT_READ_API_PORT  (default 8787)
  IT_READ_API_KEY   (optional)
  IT_READ_API_CORS_ORIGINS  (optional, comma-separated; enables CORS for browser direct calls)
"""
from __future__ import annotations

import json
import os
import subprocess
import sys
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path


def _project_root() -> Path:
    here = Path(__file__).resolve()
    for p in [here.parent.parent.parent, *here.parents]:
        if (p / "config" / "ip-addresses.conf").is_file():
            return p
    return Path.cwd()


ROOT = _project_root()
REPORTS = ROOT / "reports" / "status"
EXPORT_SCRIPT = ROOT / "scripts" / "it-ops" / "export-live-inventory-and-drift.sh"
API_KEY = os.environ.get("IT_READ_API_KEY", "").strip()
HOST = os.environ.get("IT_READ_API_HOST", "127.0.0.1")
PORT = int(os.environ.get("IT_READ_API_PORT", "8787"))
# Comma-separated origins for Access-Control-Allow-Origin (optional; portal should proxy via Next.js).
_CORS_RAW = os.environ.get("IT_READ_API_CORS_ORIGINS", "").strip()
CORS_ORIGINS = {o.strip() for o in _CORS_RAW.split(",") if o.strip()}


class Handler(BaseHTTPRequestHandler):
    server_version = "SankofaITReadAPI/0.1"

    def log_message(self, fmt: str, *args) -> None:
        sys.stderr.write("%s - %s\n" % (self.address_string(), fmt % args))

    def _maybe_cors(self) -> None:
        if not CORS_ORIGINS:
            return
        origin = (self.headers.get("Origin") or "").strip()
        if origin in CORS_ORIGINS:
            self.send_header("Access-Control-Allow-Origin", origin)
            self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS")
            self.send_header(
                "Access-Control-Allow-Headers",
                "Content-Type, X-API-Key",
            )
            self.send_header("Vary", "Origin")

    def do_OPTIONS(self) -> None:
        if CORS_ORIGINS:
            self.send_response(204)
            self._maybe_cors()
            self.end_headers()
            return
        self._text(404, "Not found\n")

    def _json(self, code: int, obj: object) -> None:
        body = json.dumps(obj, indent=2).encode("utf-8")
        self.send_response(code)
        self.send_header("Content-Type", "application/json")
        self._maybe_cors()
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def _text(self, code: int, text: str, ctype: str = "text/plain") -> None:
        body = text.encode("utf-8")
        self.send_response(code)
        self.send_header("Content-Type", f"{ctype}; charset=utf-8")
        self._maybe_cors()
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def _auth_ok(self) -> bool:
        if not API_KEY:
            return True
        return self.headers.get("X-API-Key") == API_KEY

    def do_GET(self) -> None:
        path = self.path.split("?", 1)[0].rstrip("/") or "/"
        if path == "/health":
            self._json(
                200,
                {
                    "ok": True,
                    "service": "sankofa-it-read-api",
                    "project_root": str(ROOT),
                    "auth_required_for_v1": bool(API_KEY),
                },
            )
            return

        if path.startswith("/v1"):
            if not self._auth_ok():
                self._json(401, {"error": "unauthorized"})
                return
            if path == "/v1/inventory/live":
                f = REPORTS / "live_inventory.json"
            elif path == "/v1/inventory/drift":
                f = REPORTS / "drift.json"
            else:
                self._json(404, {"error": "not_found"})
                return
            if not f.is_file():
                self._json(404, {"error": "file_missing", "path": str(f)})
                return
            try:
                data = json.loads(f.read_text(encoding="utf-8"))
            except json.JSONDecodeError as e:
                self._json(500, {"error": "invalid_json", "detail": str(e)})
                return
            self._json(200, data)
            return

        self._text(404, "Not found. GET /health or /v1/inventory/live\n")

    def do_POST(self) -> None:
        path = self.path.split("?", 1)[0].rstrip("/") or "/"
        if path != "/v1/inventory/refresh":
            self._json(404, {"error": "not_found"})
            return
        if not API_KEY or not self._auth_ok():
            self._json(401, {"error": "unauthorized"})
            return
        if not EXPORT_SCRIPT.is_file():
            self._json(500, {"error": "export_script_missing"})
            return
        try:
            subprocess.run(
                ["bash", str(EXPORT_SCRIPT)],
                cwd=str(ROOT),
                check=True,
                timeout=600,
                capture_output=True,
                text=True,
            )
        except subprocess.CalledProcessError as e:
            self._json(
                500,
                {
                    "error": "export_failed",
                    "returncode": e.returncode,
                    "stderr": (e.stderr or "")[-4000:],
                },
            )
            return
        except subprocess.TimeoutExpired:
            self._json(500, {"error": "export_timeout"})
            return
        self._json(200, {"ok": True, "refreshed": True})


def main() -> None:
    httpd = ThreadingHTTPServer((HOST, PORT), Handler)
    print(f"sankofa-it-read-api listening on http://{HOST}:{PORT}", file=sys.stderr)
    print(f"  project_root={ROOT}", file=sys.stderr)
    print(f"  GET /health  GET /v1/inventory/live  GET /v1/inventory/drift", file=sys.stderr)
    if API_KEY:
        print("  X-API-Key required for /v1/*", file=sys.stderr)
    if CORS_ORIGINS:
        print(f"  CORS origins: {sorted(CORS_ORIGINS)}", file=sys.stderr)
    try:
        httpd.serve_forever()
    except KeyboardInterrupt:
        print("\nshutdown", file=sys.stderr)


if __name__ == "__main__":
    main()