862 lines
35 KiB
Python
862 lines
35 KiB
Python
#!/usr/bin/env python3
|
|
"""panel-api — HTTP wrapper around panelctl with a web UI."""
|
|
|
|
import json
|
|
import os
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
from http.server import BaseHTTPRequestHandler, HTTPServer
|
|
from urllib.parse import urlparse, parse_qs
|
|
import urllib.request
|
|
import urllib.error
|
|
|
|
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 get_app_volumes(name):
|
|
result = run_panelctl(["inspect-volumes", name])
|
|
volumes = {}
|
|
if result["ok"]:
|
|
for line in result["stdout"].splitlines():
|
|
if "|" in line:
|
|
vname, vpath = line.split("|", 1)
|
|
volumes[vname.strip()] = vpath.strip()
|
|
return volumes
|
|
|
|
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, "<h1>Panel</h1><p>Frontend not found.</p>")
|
|
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=", ""),
|
|
"repo_url": fields[4] if len(fields) >= 5 else "",
|
|
})
|
|
self._json(200, {"ok": True, "apps": apps})
|
|
return
|
|
|
|
# /apps/<name>/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/<name>/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/<name>/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/<name>/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/<name>/backups/<filename> — 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 <name>-<timestamp>.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/<name> — 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
|
|
|
|
# /apps/<name>/volumes
|
|
if len(parts) == 3 and parts[0] == "apps" and parts[2] == "volumes":
|
|
name = parts[1]
|
|
if not is_safe_name(name):
|
|
self._json(400, {"ok": False, "error": "invalid app name"})
|
|
return
|
|
volumes = get_app_volumes(name)
|
|
self._json(200, {"ok": True, "name": name, "volumes": volumes})
|
|
return
|
|
|
|
# /apps/<name>/volume/files
|
|
if len(parts) == 4 and parts[0] == "apps" and parts[2] == "volume" and parts[3] == "files":
|
|
name = parts[1]
|
|
app, err = read_app_info(name)
|
|
if err is not None:
|
|
self._json(404, err)
|
|
return
|
|
|
|
volumes = get_app_volumes(name)
|
|
vol_key = query.get("vol", ["default"])[0]
|
|
if vol_key not in volumes:
|
|
self._json(400, {"ok": False, "error": "invalid volume specified"})
|
|
return
|
|
|
|
subpath = query.get("path", [""])[0].strip("/")
|
|
data_dir = volumes[vol_key]
|
|
target_dir = os.path.abspath(os.path.join(data_dir, subpath))
|
|
|
|
# Ensure traversal didn't escape data_dir
|
|
if not target_dir.startswith(os.path.abspath(data_dir)):
|
|
self._json(403, {"ok": False, "error": "path traversal denied"})
|
|
return
|
|
|
|
if not os.path.exists(target_dir):
|
|
self._json(404, {"ok": False, "error": "directory not found"})
|
|
return
|
|
|
|
if not os.path.isdir(target_dir):
|
|
self._json(400, {"ok": False, "error": "target is not a directory"})
|
|
return
|
|
|
|
files = []
|
|
for item in os.listdir(target_dir):
|
|
if item == "." or item == "..":
|
|
continue
|
|
item_path = os.path.join(target_dir, item)
|
|
try:
|
|
stat = os.stat(item_path)
|
|
files.append({
|
|
"name": item,
|
|
"is_dir": os.path.isdir(item_path),
|
|
"size": stat.st_size,
|
|
"mtime": stat.st_mtime
|
|
})
|
|
except OSError:
|
|
continue
|
|
|
|
files.sort(key=lambda x: (not x["is_dir"], x["name"].lower()))
|
|
self._json(200, {"ok": True, "path": subpath, "files": files})
|
|
return
|
|
|
|
# /apps/<name>/volume/download
|
|
if len(parts) == 4 and parts[0] == "apps" and parts[2] == "volume" and parts[3] == "download":
|
|
name = parts[1]
|
|
app, err = read_app_info(name)
|
|
if err is not None:
|
|
self._json(404, err)
|
|
return
|
|
|
|
volumes = get_app_volumes(name)
|
|
vol_key = query.get("vol", ["default"])[0]
|
|
if vol_key not in volumes:
|
|
self._json(400, {"ok": False, "error": "invalid volume specified"})
|
|
return
|
|
|
|
subpath = query.get("path", [""])[0].strip("/")
|
|
if not subpath:
|
|
self._json(400, {"ok": False, "error": "path parameter required"})
|
|
return
|
|
|
|
data_dir = volumes[vol_key]
|
|
target_file = os.path.abspath(os.path.join(data_dir, subpath))
|
|
|
|
if not target_file.startswith(os.path.abspath(data_dir)):
|
|
self._json(403, {"ok": False, "error": "path traversal denied"})
|
|
return
|
|
|
|
if not os.path.isfile(target_file):
|
|
self._json(404, {"ok": False, "error": "file not found"})
|
|
return
|
|
|
|
self.send_response(200)
|
|
self.send_header("Content-Type", "application/octet-stream")
|
|
self.send_header("Content-Disposition", f'attachment; filename="{os.path.basename(target_file)}"')
|
|
size = os.path.getsize(target_file)
|
|
self.send_header("Content-Length", str(size))
|
|
self.end_headers()
|
|
with open(target_file, "rb") as fh:
|
|
while True:
|
|
chunk = fh.read(65536)
|
|
if not chunk:
|
|
break
|
|
self.wfile.write(chunk)
|
|
return
|
|
|
|
self._json(404, {"ok": False, "error": "not found"})
|
|
|
|
# ── PUT ──
|
|
def do_PUT(self):
|
|
path, parts, query = self._parse_path()
|
|
if len(parts) == 4 and parts[0] == "apps" and parts[2] == "volume" and parts[3] == "files":
|
|
name = parts[1]
|
|
app, err = read_app_info(name)
|
|
if err is not None:
|
|
self._json(404, err)
|
|
return
|
|
|
|
volumes = get_app_volumes(name)
|
|
vol_key = query.get("vol", ["default"])[0]
|
|
if vol_key not in volumes:
|
|
self._json(400, {"ok": False, "error": "invalid volume specified"})
|
|
return
|
|
|
|
subpath = query.get("path", [""])[0].strip("/")
|
|
if not subpath:
|
|
self._json(400, {"ok": False, "error": "path parameter required"})
|
|
return
|
|
|
|
data_dir = volumes[vol_key]
|
|
target_file = os.path.abspath(os.path.join(data_dir, subpath))
|
|
if not target_file.startswith(os.path.abspath(data_dir)):
|
|
self._json(403, {"ok": False, "error": "path traversal denied"})
|
|
return
|
|
|
|
try:
|
|
os.makedirs(os.path.dirname(target_file), exist_ok=True)
|
|
length = int(self.headers.get("Content-Length", "0"))
|
|
with open(target_file, "wb") as fh:
|
|
bytes_read = 0
|
|
while bytes_read < length:
|
|
chunk = self.rfile.read(min(65536, length - bytes_read))
|
|
if not chunk:
|
|
break
|
|
fh.write(chunk)
|
|
bytes_read += len(chunk)
|
|
self._json(200, {"ok": True, "path": subpath})
|
|
except Exception as exc:
|
|
self._json(500, {"ok": False, "error": str(exc)})
|
|
return
|
|
|
|
self._json(404, {"ok": False, "error": "not found"})
|
|
|
|
# ── DELETE ──
|
|
def do_DELETE(self):
|
|
path, parts, query = self._parse_path()
|
|
|
|
if len(parts) == 4 and parts[0] == "apps" and parts[2] == "volume" and parts[3] == "files":
|
|
name = parts[1]
|
|
app, err = read_app_info(name)
|
|
if err is not None:
|
|
self._json(404, err)
|
|
return
|
|
|
|
volumes = get_app_volumes(name)
|
|
vol_key = query.get("vol", ["default"])[0]
|
|
if vol_key not in volumes:
|
|
self._json(400, {"ok": False, "error": "invalid volume specified"})
|
|
return
|
|
|
|
subpath = query.get("path", [""])[0].strip("/")
|
|
if not subpath:
|
|
self._json(400, {"ok": False, "error": "path parameter required"})
|
|
return
|
|
|
|
data_dir = volumes[vol_key]
|
|
target_file = os.path.abspath(os.path.join(data_dir, subpath))
|
|
|
|
if not target_file.startswith(os.path.abspath(data_dir)):
|
|
self._json(403, {"ok": False, "error": "path traversal denied"})
|
|
return
|
|
|
|
if not os.path.exists(target_file):
|
|
self._json(404, {"ok": False, "error": "file or directory not found"})
|
|
return
|
|
|
|
try:
|
|
if os.path.isdir(target_file):
|
|
import shutil
|
|
shutil.rmtree(target_file)
|
|
else:
|
|
os.remove(target_file)
|
|
self._json(200, {"ok": True, "deleted": subpath})
|
|
except Exception as exc:
|
|
self._json(500, {"ok": False, "error": str(exc)})
|
|
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()
|
|
source_type = payload.get("source_type", "default")
|
|
except Exception as exc:
|
|
self._json(400, {"ok": False, "error": f"invalid payload: {exc}"})
|
|
return
|
|
|
|
if source_type not in ["default", "raw", "github"]:
|
|
self._json(400, {"ok": False, "error": "invalid source_type"})
|
|
return
|
|
|
|
try:
|
|
result = run_panelctl(["init", name, domain, port, auth])
|
|
if not result["ok"]:
|
|
self._json(400, result)
|
|
return
|
|
|
|
app, err = read_app_info(name)
|
|
if err is not None or app is None:
|
|
run_panelctl(["remove", name])
|
|
err_msg = (err or {}).get("error", "unknown error") if err else "app state unavailable"
|
|
self._json(500, {"ok": False, "error": f"failed to read app state: {err_msg}"})
|
|
return
|
|
|
|
if source_type == "raw":
|
|
content = payload.get("compose_content", "")
|
|
try:
|
|
with open(app["APP_COMPOSE_FILE"], "w", encoding="utf-8") as fh:
|
|
fh.write(content)
|
|
except OSError as exc:
|
|
run_panelctl(["remove", name])
|
|
self._json(500, {"ok": False, "error": f"failed to write compose: {exc}"})
|
|
return
|
|
|
|
elif source_type == "github":
|
|
repo_url = payload.get("github_url", "").strip()
|
|
branch = payload.get("github_branch", "main").strip()
|
|
pat = payload.get("github_pat", "").strip()
|
|
|
|
if not repo_url:
|
|
run_panelctl(["remove", name])
|
|
self._json(400, {"ok": False, "error": "github_url is required"})
|
|
return
|
|
|
|
# Validate URL and extract owner/repo
|
|
match = re.match(r'^https?://(?:www\.)?github\.com/([^/]+)/([^/]+?)(?:\.git)?$', repo_url)
|
|
if not match:
|
|
run_panelctl(["remove", name])
|
|
self._json(400, {"ok": False, "error": "invalid github_url format"})
|
|
return
|
|
|
|
owner, repo = match.groups()
|
|
|
|
if pat:
|
|
# Check GitHub API access
|
|
api_url = f"https://api.github.com/repos/{owner}/{repo}"
|
|
req = urllib.request.Request(api_url, headers={"Authorization": f"Bearer {pat}"})
|
|
try:
|
|
urllib.request.urlopen(req)
|
|
except urllib.error.URLError as e:
|
|
run_panelctl(["remove", name])
|
|
self._json(400, {"ok": False, "error": f"github api check failed: {e.reason}"})
|
|
return
|
|
|
|
# Clone the repository
|
|
git_bin = shutil.which("git")
|
|
if not git_bin:
|
|
run_panelctl(["remove", name])
|
|
self._json(400, {"ok": False, "error": "git is not installed or not in PATH"})
|
|
return
|
|
|
|
target_dir = os.path.join(app["APP_STACK_DIR"], "repo")
|
|
if os.path.exists(target_dir):
|
|
shutil.rmtree(target_dir)
|
|
|
|
auth_url = repo_url
|
|
if pat:
|
|
auth_url = auth_url.replace("://", f"://{pat}@")
|
|
|
|
if not auth_url.endswith(".git"):
|
|
auth_url += ".git"
|
|
|
|
clone_result = subprocess.run(
|
|
[git_bin, "clone", "--branch", branch, auth_url, target_dir],
|
|
capture_output=True, text=True
|
|
)
|
|
|
|
if clone_result.returncode != 0:
|
|
run_panelctl(["remove", name])
|
|
self._json(400, {"ok": False, "error": f"git clone failed: {clone_result.stderr.strip()}"})
|
|
return
|
|
|
|
# Find compose file
|
|
compose_path = None
|
|
for fname in ["compose.yaml", "docker-compose.yml", "compose.yml", "docker-compose.yaml"]:
|
|
candidate = os.path.join(target_dir, fname)
|
|
if os.path.isfile(candidate):
|
|
compose_path = candidate
|
|
break
|
|
|
|
if not compose_path:
|
|
run_panelctl(["remove", name])
|
|
self._json(400, {"ok": False, "error": "could not find compose file in repository root"})
|
|
return
|
|
|
|
# Update manifest
|
|
manifest_path = os.path.join(BASE_DIR, "state", "apps", f"{name}.env")
|
|
try:
|
|
with open(manifest_path, "r", encoding="utf-8") as fh:
|
|
lines = fh.readlines()
|
|
with open(manifest_path, "w", encoding="utf-8") as fh:
|
|
for line in lines:
|
|
if line.startswith("APP_COMPOSE_FILE="):
|
|
fh.write(f'APP_COMPOSE_FILE="{compose_path}"\n')
|
|
else:
|
|
fh.write(line)
|
|
fh.write(f'APP_REPO_URL="{repo_url}"\n')
|
|
fh.write(f'APP_REPO_BRANCH="{branch}"\n')
|
|
except OSError as exc:
|
|
run_panelctl(["remove", name])
|
|
self._json(500, {"ok": False, "error": f"failed to update manifest: {exc}"})
|
|
return
|
|
|
|
self._json(200, {"ok": True, "code": 0, "stdout": "initialized successfully"})
|
|
except Exception as exc:
|
|
run_panelctl(["remove", name])
|
|
self._json(500, {"ok": False, "error": f"init failed: {exc}"})
|
|
return
|
|
|
|
if len(parts) >= 3 and parts[0] == "apps":
|
|
name = parts[1]
|
|
action = parts[2]
|
|
|
|
# POST /apps/<name>/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/<name>/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/<name>/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/<name>/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", "volume-clear"}:
|
|
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/<name>/repo-pull — re-clone/pull repo and redeploy
|
|
if action == "repo-pull":
|
|
if not is_safe_name(name):
|
|
self._json(400, {"ok": False, "error": "invalid app name"})
|
|
return
|
|
|
|
try:
|
|
app, err = read_app_info(name)
|
|
if err is not None or app is None:
|
|
self._json(404, {"ok": False, "error": "app not found"})
|
|
return
|
|
|
|
repo_url = app.get("APP_REPO_URL", "").strip()
|
|
branch = app.get("APP_REPO_BRANCH", "main").strip()
|
|
|
|
if not repo_url:
|
|
self._json(400, {"ok": False, "error": "app has no APP_REPO_URL"})
|
|
return
|
|
|
|
git_bin = shutil.which("git")
|
|
if not git_bin:
|
|
self._json(400, {"ok": False, "error": "git is not installed"})
|
|
return
|
|
|
|
target_dir = os.path.join(app["APP_STACK_DIR"], "repo")
|
|
pat = ""
|
|
|
|
auth_url = repo_url
|
|
if pat:
|
|
auth_url = auth_url.replace("://", f"://{pat}@")
|
|
if not auth_url.endswith(".git"):
|
|
auth_url += ".git"
|
|
|
|
if os.path.exists(target_dir):
|
|
# Already cloned — try git pull
|
|
pull_result = subprocess.run(
|
|
[git_bin, "-C", target_dir, "pull", "origin", branch],
|
|
capture_output=True, text=True
|
|
)
|
|
if pull_result.returncode != 0:
|
|
# Fall back to re-clone
|
|
shutil.rmtree(target_dir)
|
|
clone_result = subprocess.run(
|
|
[git_bin, "clone", "--branch", branch, auth_url, target_dir],
|
|
capture_output=True, text=True
|
|
)
|
|
if clone_result.returncode != 0:
|
|
self._json(400, {"ok": False, "error": f"git clone failed: {clone_result.stderr.strip()}"})
|
|
return
|
|
else:
|
|
clone_result = subprocess.run(
|
|
[git_bin, "clone", "--branch", branch, auth_url, target_dir],
|
|
capture_output=True, text=True
|
|
)
|
|
if clone_result.returncode != 0:
|
|
self._json(400, {"ok": False, "error": f"git clone failed: {clone_result.stderr.strip()}"})
|
|
return
|
|
|
|
# Find compose file
|
|
compose_path = None
|
|
for fname in ["compose.yaml", "docker-compose.yml", "compose.yml", "docker-compose.yaml"]:
|
|
candidate = os.path.join(target_dir, fname)
|
|
if os.path.isfile(candidate):
|
|
compose_path = candidate
|
|
break
|
|
|
|
if not compose_path:
|
|
self._json(400, {"ok": False, "error": "compose file not found in repository"})
|
|
return
|
|
|
|
# Update manifest compose path
|
|
manifest_path = os.path.join(BASE_DIR, "state", "apps", f"{name}.env")
|
|
try:
|
|
with open(manifest_path, "r", encoding="utf-8") as fh:
|
|
lines = fh.readlines()
|
|
with open(manifest_path, "w", encoding="utf-8") as fh:
|
|
for line in lines:
|
|
if line.startswith("APP_COMPOSE_FILE="):
|
|
fh.write(f'APP_COMPOSE_FILE="{compose_path}"\n')
|
|
else:
|
|
fh.write(line)
|
|
except OSError as exc:
|
|
self._json(500, {"ok": False, "error": f"failed to update manifest: {exc}"})
|
|
return
|
|
|
|
# Redeploy
|
|
result = run_panelctl(["deploy", name])
|
|
self._json(200 if result["ok"] else 400, result)
|
|
except Exception as exc:
|
|
self._json(500, {"ok": False, "error": f"repo-pull failed: {exc}"})
|
|
return
|
|
|
|
# POST /apps/<name>/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()
|