#!/usr/bin/env python3 import json import os import re import subprocess from http.server import BaseHTTPRequestHandler, HTTPServer from urllib.parse import urlparse PANELCTL = os.environ.get("PANELCTL_PATH", "/run/current-system/sw/bin/panelctl") BIND = os.environ.get("PANEL_API_BIND", "127.0.0.1") PORT = int(os.environ.get("PANEL_API_PORT", "9911")) BASE_DIR = os.environ.get("PANEL_BASE_DIR", "/home/reudy/containers") INDEX_HTML = """ Panel

Containers Panel

Rootless Podman + Caddy routes from one place.

Create App

Name Domain

Host Port

Protected

Apps

Name	Domain	Upstream	Actions

Compose Editor

No app selected.

Ready.

""" def is_safe_name(name): return re.match(r"^[a-z0-9]([a-z0-9-]*[a-z0-9])?$", name) is not None def read_app_info(name): if not is_safe_name(name): return None, {"ok": False, "error": "invalid app name"} result = run_panelctl(["show", name]) if not result["ok"]: return None, result app = parse_env_blob(result["stdout"]) compose_file = app.get("APP_COMPOSE_FILE", "") if not compose_file: return None, {"ok": False, "error": "missing APP_COMPOSE_FILE in manifest"} base_stacks = os.path.join(BASE_DIR, "stacks") + os.sep norm_compose = os.path.abspath(compose_file) if not norm_compose.startswith(base_stacks): return None, {"ok": False, "error": "compose path is outside allowed base directory"} app["APP_COMPOSE_FILE"] = norm_compose return app, None def run_panelctl(args): proc = subprocess.run( [PANELCTL, *args], check=False, capture_output=True, text=True, ) return { "ok": proc.returncode == 0, "code": proc.returncode, "stdout": proc.stdout.strip(), "stderr": proc.stderr.strip(), } def parse_env_blob(blob): out = {} for line in blob.splitlines(): line = line.strip() if not line or line.startswith("#") or "=" not in line: continue key, value = line.split("=", 1) out[key] = value.strip().strip('"') return out class Handler(BaseHTTPRequestHandler): def _html(self, code, body): payload = body.encode("utf-8") self.send_response(code) self.send_header("Content-Type", "text/html; charset=utf-8") self.send_header("Content-Length", str(len(payload))) self.end_headers() self.wfile.write(payload) def _json(self, code, payload): body = json.dumps(payload, indent=2).encode("utf-8") self.send_response(code) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def _read_json(self): length = int(self.headers.get("Content-Length", "0")) if length == 0: return {} raw = self.rfile.read(length) return json.loads(raw.decode("utf-8")) def log_message(self, fmt, *args): return def do_GET(self): parsed = urlparse(self.path) path = parsed.path if path == "/": self._html(200, INDEX_HTML) return if path == "/health": self._json(200, {"ok": True, "service": "panel-api"}) return if path == "/apps": result = run_panelctl(["list"]) if not result["ok"]: self._json(500, result) return apps = [] for line in result["stdout"].splitlines(): if not line.strip() or line.strip() == "no apps found": continue # format: name domain upstream auth=true|false fields = line.split() if len(fields) < 4: continue app = { "name": fields[0], "domain": fields[1], "upstream": fields[2], "auth": fields[3].replace("auth=", ""), } apps.append(app) self._json(200, {"ok": True, "apps": apps}) return if path.startswith("/apps/") and path.endswith("/compose"): parts = [p for p in path.split("/") if p] if len(parts) != 3: self._json(404, {"ok": False, "error": "not found"}) return _, name, _ = parts app, err = read_app_info(name) if err is not None: self._json(404, err) return try: with open(app["APP_COMPOSE_FILE"], "r", encoding="utf-8") as fh: content = fh.read() except OSError as exc: self._json(500, {"ok": False, "error": f"failed to read compose file: {exc}"}) return self._json(200, {"ok": True, "name": name, "content": content}) return if path.startswith("/apps/"): name = path.split("/")[-1] if not name: self._json(400, {"ok": False, "error": "missing app name"}) return result = run_panelctl(["show", name]) if not result["ok"]: self._json(404, result) return self._json(200, {"ok": True, "app": parse_env_blob(result["stdout"])}) return self._json(404, {"ok": False, "error": "not found"}) def do_POST(self): parsed = urlparse(self.path) path = parsed.path if path == "/apps/init": try: payload = self._read_json() name = payload["name"] domain = payload["domain"] port = str(payload["port"]) auth = str(payload.get("auth", True)).lower() except Exception as exc: self._json(400, {"ok": False, "error": f"invalid payload: {exc}"}) return result = run_panelctl(["init", name, domain, port, auth]) self._json(200 if result["ok"] else 400, result) return action_prefix = "/apps/" if path.startswith(action_prefix): parts = [p for p in path.split("/") if p] if len(parts) == 3 and parts[2] == "compose": _, name, _ = parts app, err = read_app_info(name) if err is not None: self._json(404, err) return try: payload = self._read_json() except Exception as exc: self._json(400, {"ok": False, "error": f"invalid payload: {exc}"}) return content = payload.get("content", "") if not isinstance(content, str) or not content.strip(): self._json(400, {"ok": False, "error": "compose content must be a non-empty string"}) return try: with open(app["APP_COMPOSE_FILE"], "w", encoding="utf-8") as fh: fh.write(content) except OSError as exc: self._json(500, {"ok": False, "error": f"failed to write compose file: {exc}"}) return self._json(200, {"ok": True, "name": name, "saved": True}) return # /apps// if len(parts) == 3: _, name, action = parts if action in {"deploy", "stop", "render-route"}: result = run_panelctl([action, name]) self._json(200 if result["ok"] else 400, result) return if action == "remove": payload = {} try: payload = self._read_json() except Exception: payload = {} keep = payload.get("keepVolumes", False) args = ["remove", name] if keep: args.append("--keep-volumes") result = run_panelctl(args) self._json(200 if result["ok"] else 400, result) return self._json(404, {"ok": False, "error": "not found"}) def main(): server = HTTPServer((BIND, PORT), Handler) print(f"panel-api listening on http://{BIND}:{PORT}") server.serve_forever() if __name__ == "__main__": main()