#!/usr/bin/env python3
"""panel-api — HTTP wrapper around panelctl with a web UI."""
import json
import os
import re
import subprocess
from http.server import BaseHTTPRequestHandler, HTTPServer
from urllib.parse import urlparse, parse_qs
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", "/var/lib/containers")
FRONTEND_DIR = os.environ.get(
"PANEL_FRONTEND_DIR",
os.path.join(os.path.dirname(os.path.abspath(__file__)), "frontend"),
)
def is_safe_name(name):
return re.match(r"^[a-z0-9]([a-z0-9-]*[a-z0-9])?$", name) is not 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
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 parse_status_output(stdout):
"""Try to determine if any container is running from panelctl status output."""
text = stdout.lower()
if not text or "no containers" in text:
return {"running": False, "raw": stdout}
# podman compose ps --format json returns JSON array
try:
containers = json.loads(stdout)
if isinstance(containers, list):
running = any(
c.get("State", "").lower() == "running"
or c.get("status", "").lower().startswith("up")
for c in containers
)
return {
"running": running,
"containers": [
{
"name": c.get("Name", c.get("name", "?")),
"state": c.get("State", c.get("status", "unknown")),
"image": c.get("Image", c.get("image", "")),
}
for c in containers
],
}
except (json.JSONDecodeError, TypeError):
pass
# Fallback: check for "Up" or "running" in text
running = "up" in text or "running" in text
return {"running": running, "raw": stdout}
def parse_backups_output(stdout):
"""Parse panelctl list-backups output into structured data."""
backups = []
for line in stdout.splitlines():
line = line.strip()
if not line or "no backups" in line.lower():
continue
parts = line.split()
if len(parts) >= 1:
entry = {"name": parts[0]}
if len(parts) >= 2:
entry["size"] = parts[1]
if len(parts) >= 3:
try:
entry["mtime"] = int(parts[2])
except ValueError:
pass
backups.append(entry)
return backups
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 _file(self, code, filepath, content_type):
try:
with open(filepath, "rb") as fh:
data = fh.read()
self.send_response(code)
self.send_header("Content-Type", content_type)
self.send_header("Content-Length", str(len(data)))
self.end_headers()
self.wfile.write(data)
except OSError:
self._json(500, {"ok": False, "error": "failed to read file"})
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):
# Log to stdout (goes to systemd journal)
print(f"[panel-api] {self.address_string()} {fmt % args}")
# ── Routing helpers ──
def _parse_path(self):
parsed = urlparse(self.path)
path = parsed.path.rstrip("/") or "/"
query = parse_qs(parsed.query)
parts = [p for p in path.split("/") if p]
return path, parts, query
# ── GET ──
def do_GET(self):
path, parts, query = self._parse_path()
if path == "/":
index = os.path.join(FRONTEND_DIR, "index.html")
if os.path.isfile(index):
self._file(200, index, "text/html; charset=utf-8")
else:
self._html(200, "Panel
Frontend not found.
")
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
fields = line.split()
if len(fields) < 4:
continue
apps.append({
"name": fields[0],
"domain": fields[1].split(",")[0],
"domains": fields[1],
"upstream": fields[2],
"auth": fields[3].replace("auth=", ""),
})
self._json(200, {"ok": True, "apps": apps})
return
# /apps//compose
if len(parts) == 3 and parts[0] == "apps" and parts[2] == "compose":
name = parts[1]
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: {exc}"})
return
self._json(200, {"ok": True, "name": name, "content": content})
return
# /apps//status
if len(parts) == 3 and parts[0] == "apps" and parts[2] == "status":
name = parts[1]
if not is_safe_name(name):
self._json(400, {"ok": False, "error": "invalid app name"})
return
result = run_panelctl(["status", name])
status = parse_status_output(result["stdout"])
self._json(200, {"ok": True, "name": name, **status})
return
# /apps//logs
if len(parts) == 3 and parts[0] == "apps" and parts[2] == "logs":
name = parts[1]
if not is_safe_name(name):
self._json(400, {"ok": False, "error": "invalid app name"})
return
tail = query.get("tail", ["100"])[0]
try:
tail = str(int(tail))
except ValueError:
tail = "100"
result = run_panelctl(["logs", name, "--tail", tail])
self._json(200, {"ok": True, "name": name, "logs": result["stdout"]})
return
# /apps//backups
if len(parts) == 3 and parts[0] == "apps" and parts[2] == "backups":
name = parts[1]
if not is_safe_name(name):
self._json(400, {"ok": False, "error": "invalid app name"})
return
result = run_panelctl(["list-backups", name])
backups = parse_backups_output(result["stdout"])
self._json(200, {"ok": True, "name": name, "backups": backups})
return
# /apps//backups/ — download backup zip
if len(parts) == 4 and parts[0] == "apps" and parts[2] == "backups":
name = parts[1]
filename = parts[3]
if not is_safe_name(name):
self._json(400, {"ok": False, "error": "invalid app name"})
return
# Validate filename: must match -.zip
if not re.match(r"^[a-z0-9-]+-\d{8}-\d{6}\.zip$", filename):
self._json(400, {"ok": False, "error": "invalid backup filename"})
return
backup_path = os.path.join(BASE_DIR, "backups", filename)
norm_path = os.path.abspath(backup_path)
norm_backups = os.path.abspath(os.path.join(BASE_DIR, "backups")) + os.sep
if not norm_path.startswith(norm_backups):
self._json(403, {"ok": False, "error": "path traversal denied"})
return
if not os.path.isfile(norm_path):
self._json(404, {"ok": False, "error": "backup not found"})
return
self.send_response(200)
self.send_header("Content-Type", "application/zip")
self.send_header("Content-Disposition", f'attachment; filename="{filename}"')
size = os.path.getsize(norm_path)
self.send_header("Content-Length", str(size))
self.end_headers()
with open(norm_path, "rb") as fh:
while True:
chunk = fh.read(65536)
if not chunk:
break
self.wfile.write(chunk)
return
# /apps/ — show single app
if len(parts) == 2 and parts[0] == "apps":
name = parts[1]
if not is_safe_name(name):
self._json(400, {"ok": False, "error": "invalid 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"})
# ── POST ──
def do_POST(self):
path, parts, query = self._parse_path()
# POST /apps/init
if path == "/apps/init":
try:
payload = self._read_json()
name = payload["name"]
# Support both "domain" (string, possibly comma-separated) and "domains" (array)
if "domains" in payload and isinstance(payload["domains"], list):
domain = ",".join(payload["domains"])
else:
domain = str(payload.get("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
if len(parts) >= 3 and parts[0] == "apps":
name = parts[1]
action = parts[2]
# POST /apps//compose — save compose file
if action == "compose":
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: {exc}"})
return
self._json(200, {"ok": True, "name": name, "saved": True})
return
# POST /apps//validate-compose
if action == "validate-compose":
if not is_safe_name(name):
self._json(400, {"ok": False, "error": "invalid app name"})
return
result = run_panelctl(["validate-compose", name])
self._json(200 if result["ok"] else 400, result)
return
# POST /apps//backup
if action == "backup":
if not is_safe_name(name):
self._json(400, {"ok": False, "error": "invalid app name"})
return
result = run_panelctl(["backup", name])
self._json(200 if result["ok"] else 400, result)
return
# POST /apps//restore
if action == "restore":
if not is_safe_name(name):
self._json(400, {"ok": False, "error": "invalid app name"})
return
try:
payload = self._read_json()
except Exception:
payload = {}
backup_file = payload.get("file", "")
if not backup_file:
self._json(400, {"ok": False, "error": "backup file name is required"})
return
result = run_panelctl(["restore", name, backup_file])
self._json(200 if result["ok"] else 400, result)
return
# Simple panelctl pass-through actions
if action in {"deploy", "stop", "restart", "render-route"}:
if not is_safe_name(name):
self._json(400, {"ok": False, "error": "invalid app name"})
return
result = run_panelctl([action, name])
self._json(200 if result["ok"] else 400, result)
return
# POST /apps//remove
if action == "remove":
if not is_safe_name(name):
self._json(400, {"ok": False, "error": "invalid app name"})
return
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}")
print(f"frontend dir: {FRONTEND_DIR}")
server.serve_forever()
if __name__ == "__main__":
main()