c1ff6c8176
Forgejo:
- panel.nix passes the local Forgejo's public, API and ssh URLs (derived
from forgejo.nix) to panel-api.
- Settings dialog: connect a Forgejo access token (verified against
/api/v1/user, stored 0600 in state/panel/forgejo-token).
- New-app dialog gets a Forgejo repository picker with search and a branch
dropdown; private repos are cloned over https with the stored token, or
over ssh with the deploy key when no token is connected. The app name and
domain are filled in from the repository name.
- Commit and compare links in the Source tab point at Forgejo; cards show
the provider ("Forgejo · main").
Git over ssh:
- ssh:// and git@host:owner/repo URLs are accepted; the panel generates an
ed25519 deploy key in state/panel/ssh and uses it for clone/fetch
(BatchMode, accept-new host keys). openssh added to the service path.
- Credential redaction only applies to http(s) URLs, so ssh usernames are
kept; git errors now report the meaningful line instead of git's advice.
Environment variables:
- Stored per app in state/env/<app>.env (0600), outside the repo and stack.
- panelctl passes them to every compose command via env(1), so ${VAR}
interpolation works; by default deploy/restart also generate a compose
override listing the keys under every service's environment (values are
read from compose's environment, never quoted into YAML).
- Environment tab (and a section in the new-app dialog) with .env paste
import, hidden values, validation of names (reserved podman/compose vars
rejected), hints for ${VAR}s the compose file uses but aren't set, and
Save / Save & deploy. Removing an app deletes its variables.
The API still accepts the old source_type "github" / github_* fields.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01UbWSNkXxZhYf7eqHTyx3Bf
1591 lines
63 KiB
Python
1591 lines
63 KiB
Python
#!/usr/bin/env python3
|
|
"""panel-api — HTTP wrapper around panelctl with a web UI."""
|
|
|
|
import json
|
|
import os
|
|
import re
|
|
import shlex
|
|
import shutil
|
|
import socket
|
|
import subprocess
|
|
import threading
|
|
import time
|
|
import urllib.error
|
|
import urllib.request
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
from contextlib import contextmanager
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
from urllib.parse import urlparse, parse_qs, quote
|
|
|
|
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"),
|
|
)
|
|
|
|
# Optional Forgejo instance (set from panel.nix). The API URL may be an internal
|
|
# address; the public URL is what repositories are cloned from and linked to.
|
|
FORGEJO_URL = os.environ.get("PANEL_FORGEJO_URL", "").rstrip("/")
|
|
FORGEJO_API_URL = (os.environ.get("PANEL_FORGEJO_API_URL", "") or FORGEJO_URL).rstrip("/")
|
|
FORGEJO_SSH_URL = os.environ.get("PANEL_FORGEJO_SSH_URL", "").rstrip("/")
|
|
FORGEJO_HOST = urlparse(FORGEJO_URL).hostname or ""
|
|
|
|
PANEL_STATE_DIR = os.path.join(BASE_DIR, "state", "panel")
|
|
FORGEJO_TOKEN_FILE = os.path.join(PANEL_STATE_DIR, "forgejo-token")
|
|
SSH_DIR = os.path.join(PANEL_STATE_DIR, "ssh")
|
|
SSH_KEY = os.path.join(SSH_DIR, "id_ed25519")
|
|
ENV_DIR = os.path.join(BASE_DIR, "state", "env")
|
|
|
|
COMPOSE_FILENAMES = ["compose.yaml", "compose.yml", "docker-compose.yml", "docker-compose.yaml"]
|
|
GIT_TIMEOUT = 300
|
|
# Manifest values are written into a file that panelctl sources with bash, so
|
|
# they must not contain anything that is special inside double quotes.
|
|
_URL_CHARS = r"[^\s\"'`$\\]"
|
|
REPO_URL_RE = re.compile(
|
|
rf"^(?:https?://{_URL_CHARS}+" # https://host/owner/repo.git
|
|
rf"|ssh://{_URL_CHARS}+" # ssh://git@host:port/owner/repo.git
|
|
rf"|[A-Za-z0-9._-]+@[A-Za-z0-9.-]+:{_URL_CHARS}+)$" # git@host:owner/repo.git
|
|
)
|
|
BRANCH_RE = re.compile(r"^[A-Za-z0-9._/][A-Za-z0-9._/-]*$")
|
|
FORGEJO_REPO_RE = re.compile(r"^[A-Za-z0-9._-]+/[A-Za-z0-9._-]+$")
|
|
|
|
ENV_KEY_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$")
|
|
# Variables are set on the compose process itself, so anything that changes how
|
|
# podman/compose run (or where they look for state) is off limits.
|
|
RESERVED_ENV = {"PATH", "HOME", "USER", "LOGNAME", "SHELL", "TMPDIR", "PWD", "OLDPWD", "IFS", "TERM"}
|
|
RESERVED_ENV_PREFIXES = ("XDG_", "DBUS_", "DOCKER_", "CONTAINER_", "CONTAINERS_", "COMPOSE_",
|
|
"PODMAN_", "BUILDAH_", "LD_", "BASH_")
|
|
|
|
|
|
def is_safe_name(name):
|
|
return re.match(r"^[a-z0-9]([a-z0-9-]*[a-z0-9])?$", name) is not None
|
|
|
|
|
|
# ── Per-app operation locks ──
|
|
# Requests are handled concurrently, so two mutating operations on the same app
|
|
# (e.g. a double-clicked deploy, or deploy + restore) must not overlap.
|
|
|
|
_busy = {}
|
|
_busy_lock = threading.Lock()
|
|
|
|
|
|
class AppBusy(Exception):
|
|
def __init__(self, name, action):
|
|
super().__init__(f"another operation ({action}) is already running on '{name}'")
|
|
self.action = action
|
|
|
|
|
|
@contextmanager
|
|
def app_operation(name, action):
|
|
with _busy_lock:
|
|
if name in _busy:
|
|
raise AppBusy(name, _busy[name])
|
|
_busy[name] = action
|
|
try:
|
|
yield
|
|
finally:
|
|
with _busy_lock:
|
|
_busy.pop(name, None)
|
|
|
|
|
|
def busy_snapshot():
|
|
with _busy_lock:
|
|
return dict(_busy)
|
|
|
|
|
|
def redact_credentials(text, replacement="***@"):
|
|
"""Hide user:token@ credentials embedded in http(s) URLs.
|
|
(ssh://git@host is a username, not a secret, and must be kept.)"""
|
|
return re.sub(r"(https?://)[^/@\s]+@", r"\1" + replacement, text or "")
|
|
|
|
|
|
def last_line(text):
|
|
lines = [line.strip() for line in (text or "").splitlines() if line.strip()]
|
|
return lines[-1] if lines else ""
|
|
|
|
|
|
def git_error(stderr):
|
|
"""The informative line of a git failure (git ends with generic advice)."""
|
|
lines = [line.strip() for line in (stderr or "").splitlines() if line.strip()]
|
|
for line in lines:
|
|
if re.match(r"^(ssh|fatal|error|remote):", line, re.I) and "could not read from remote" not in line.lower():
|
|
return re.sub(r"^fatal:\s*", "", line)
|
|
return last_line(stderr)
|
|
|
|
|
|
def write_private_file(path, content):
|
|
"""Atomically write a file only the service user can read."""
|
|
os.makedirs(os.path.dirname(path), mode=0o700, exist_ok=True)
|
|
tmp = f"{path}.tmp"
|
|
fd = os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
|
|
with os.fdopen(fd, "w", encoding="utf-8") as fh:
|
|
fh.write(content)
|
|
os.replace(tmp, path)
|
|
|
|
|
|
# ── SSH deploy key ──
|
|
# One key pair for the panel; add its public half as a (read-only) deploy key
|
|
# to repositories cloned over SSH.
|
|
|
|
def ssh_public_key(create=True):
|
|
pub = SSH_KEY + ".pub"
|
|
if not os.path.isfile(pub) and create:
|
|
keygen = shutil.which("ssh-keygen")
|
|
if not keygen:
|
|
return None
|
|
os.makedirs(SSH_DIR, mode=0o700, exist_ok=True)
|
|
subprocess.run(
|
|
[keygen, "-t", "ed25519", "-N", "", "-q", "-C", f"panel@{socket.gethostname()}", "-f", SSH_KEY],
|
|
capture_output=True, check=False, timeout=30,
|
|
)
|
|
try:
|
|
with open(pub, "r", encoding="utf-8") as fh:
|
|
return fh.read().strip()
|
|
except OSError:
|
|
return None
|
|
|
|
|
|
def is_ssh_url(url):
|
|
return not re.match(r"^https?://", url or "")
|
|
|
|
|
|
# ── Git helpers ──
|
|
|
|
def git_env():
|
|
# Never block on an interactive credential prompt.
|
|
env = dict(os.environ, GIT_TERMINAL_PROMPT="0")
|
|
if os.path.isfile(SSH_KEY):
|
|
env["GIT_SSH_COMMAND"] = " ".join([
|
|
"ssh", "-i", shlex.quote(SSH_KEY),
|
|
"-o", "IdentitiesOnly=yes",
|
|
"-o", "BatchMode=yes",
|
|
"-o", "StrictHostKeyChecking=accept-new",
|
|
"-o", "UserKnownHostsFile=" + shlex.quote(os.path.join(SSH_DIR, "known_hosts")),
|
|
])
|
|
return env
|
|
|
|
|
|
def run_git(args, cwd=None, timeout=GIT_TIMEOUT):
|
|
git_bin = shutil.which("git")
|
|
if not git_bin:
|
|
return {"ok": False, "stdout": "", "stderr": "git is not installed or not in PATH"}
|
|
cmd = [git_bin] + (["-C", cwd] if cwd else []) + args
|
|
try:
|
|
proc = subprocess.run(cmd, capture_output=True, text=True, env=git_env(), timeout=timeout)
|
|
except subprocess.TimeoutExpired:
|
|
return {"ok": False, "stdout": "", "stderr": f"git {args[0]} timed out after {timeout}s"}
|
|
return {
|
|
"ok": proc.returncode == 0,
|
|
"stdout": redact_credentials(proc.stdout.strip()),
|
|
"stderr": redact_credentials(proc.stderr.strip()),
|
|
}
|
|
|
|
|
|
def clone_repo(url, branch, target_dir, token=""):
|
|
auth_url = url
|
|
if token and not is_ssh_url(url):
|
|
auth_url = url.replace("://", f"://{quote(token, safe='')}@", 1)
|
|
elif is_ssh_url(url):
|
|
ssh_public_key() # make sure the deploy key exists before the first clone
|
|
args = ["clone"]
|
|
if branch:
|
|
args += ["--branch", branch]
|
|
return run_git(args + ["--", auth_url, target_dir])
|
|
|
|
|
|
def repo_host_and_path(url):
|
|
url = redact_credentials(url or "", "")
|
|
m = (re.match(r"^https?://([^/:]+)(?::\d+)?/(.+?)(?:\.git)?/?$", url)
|
|
or re.match(r"^ssh://(?:[^@/]+@)?([^/:]+)(?::\d+)?/(.+?)(?:\.git)?/?$", url)
|
|
or re.match(r"^(?:[^@/]+@)?([^/:]+):(?!/)(.+?)(?:\.git)?/?$", url))
|
|
return (m.group(1), m.group(2)) if m else (None, None)
|
|
|
|
|
|
def repo_provider(url):
|
|
host, _ = repo_host_and_path(url)
|
|
if host and FORGEJO_HOST and host == FORGEJO_HOST:
|
|
return "forgejo"
|
|
if host in ("github.com", "www.github.com"):
|
|
return "github"
|
|
return "git"
|
|
|
|
|
|
def repo_web_url(url):
|
|
"""Browser URL of a repository, for commit / compare links."""
|
|
host, path = repo_host_and_path(url)
|
|
if not host:
|
|
return ""
|
|
if FORGEJO_HOST and host == FORGEJO_HOST:
|
|
return f"{FORGEJO_URL}/{path}"
|
|
m = re.match(r"^(https?)://", url or "")
|
|
return f"{m.group(1) if m else 'https'}://{host}/{path}"
|
|
|
|
|
|
# ── Forgejo ──
|
|
|
|
class ForgejoError(Exception):
|
|
pass
|
|
|
|
|
|
def forgejo_token():
|
|
try:
|
|
with open(FORGEJO_TOKEN_FILE, "r", encoding="utf-8") as fh:
|
|
return fh.read().strip()
|
|
except OSError:
|
|
return ""
|
|
|
|
|
|
def forgejo_api(path, token=None, timeout=10):
|
|
if not FORGEJO_API_URL:
|
|
raise ForgejoError("no Forgejo instance is configured")
|
|
token = forgejo_token() if token is None else token
|
|
req = urllib.request.Request(FORGEJO_API_URL + path, headers={"Accept": "application/json"})
|
|
if token:
|
|
req.add_header("Authorization", f"token {token}")
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=timeout) as res:
|
|
return json.loads(res.read().decode("utf-8") or "null")
|
|
except urllib.error.HTTPError as exc:
|
|
if exc.code in (401, 403):
|
|
raise ForgejoError("Forgejo rejected the token") from exc
|
|
if exc.code == 404:
|
|
raise ForgejoError("not found on Forgejo (or no access)") from exc
|
|
raise ForgejoError(f"Forgejo answered HTTP {exc.code}") from exc
|
|
except (urllib.error.URLError, TimeoutError, ValueError) as exc:
|
|
raise ForgejoError(f"can't reach Forgejo: {getattr(exc, 'reason', exc)}") from exc
|
|
|
|
|
|
def is_forgejo_https_url(url):
|
|
return bool(FORGEJO_URL) and not is_ssh_url(url) and repo_provider(url) == "forgejo"
|
|
|
|
|
|
# ── Environment variables ──
|
|
# Stored per app in state/env/<app>.env as KEY=VALUE lines (0600). panelctl
|
|
# passes them to every compose command, so they work for ${VAR} interpolation
|
|
# and — unless disabled — are injected into every service.
|
|
|
|
def env_file_path(name):
|
|
return os.path.join(ENV_DIR, f"{name}.env")
|
|
|
|
|
|
def read_app_env(name):
|
|
items = []
|
|
try:
|
|
with open(env_file_path(name), "r", encoding="utf-8") as fh:
|
|
lines = fh.read().splitlines()
|
|
except OSError:
|
|
return items
|
|
for line in lines:
|
|
if not line or line.startswith("#") or "=" not in line:
|
|
continue
|
|
key, value = line.split("=", 1)
|
|
items.append({"key": key, "value": value})
|
|
return items
|
|
|
|
|
|
def validate_env(items):
|
|
if items is None:
|
|
return []
|
|
if not isinstance(items, list):
|
|
raise ValueError("env must be a list of {key, value}")
|
|
seen = set()
|
|
out = []
|
|
for item in items:
|
|
if not isinstance(item, dict):
|
|
raise ValueError("env must be a list of {key, value}")
|
|
key = str(item.get("key", "")).strip()
|
|
value = item.get("value", "")
|
|
value = "" if value is None else str(value)
|
|
if not key and not value:
|
|
continue
|
|
if not ENV_KEY_RE.match(key):
|
|
raise ValueError(f"'{key}' is not a valid variable name (letters, digits and _, not starting with a digit)")
|
|
if key in RESERVED_ENV or key.startswith(RESERVED_ENV_PREFIXES):
|
|
raise ValueError(f"'{key}' is reserved because it would change how podman/compose run")
|
|
if key in seen:
|
|
raise ValueError(f"'{key}' is set more than once")
|
|
if any(c in value for c in "\n\r\0"):
|
|
raise ValueError(f"the value of '{key}' must be a single line")
|
|
seen.add(key)
|
|
out.append({"key": key, "value": value})
|
|
return out
|
|
|
|
|
|
def write_app_env(name, items):
|
|
path = env_file_path(name)
|
|
if not items:
|
|
try:
|
|
os.remove(path)
|
|
except FileNotFoundError:
|
|
pass
|
|
return
|
|
write_private_file(path, "".join(f"{i['key']}={i['value']}\n" for i in items))
|
|
|
|
|
|
def repo_commit(repo_dir, ref="HEAD"):
|
|
result = run_git(["log", "-1", "--format=%H%x1f%s%x1f%an%x1f%ct", ref], cwd=repo_dir, timeout=15)
|
|
if not result["ok"] or not result["stdout"]:
|
|
return None
|
|
sha, subject, author, ts = (result["stdout"].split("\x1f") + ["", "", "", ""])[:4]
|
|
return {
|
|
"sha": sha,
|
|
"short": sha[:7],
|
|
"subject": subject,
|
|
"author": author,
|
|
"time": int(ts) if ts.isdigit() else None,
|
|
}
|
|
|
|
|
|
def repo_current_branch(repo_dir):
|
|
result = run_git(["rev-parse", "--abbrev-ref", "HEAD"], cwd=repo_dir, timeout=15)
|
|
if result["ok"] and result["stdout"] and result["stdout"] != "HEAD":
|
|
return result["stdout"]
|
|
return ""
|
|
|
|
|
|
def find_compose_file(repo_dir):
|
|
for fname in COMPOSE_FILENAMES:
|
|
candidate = os.path.join(repo_dir, fname)
|
|
if os.path.isfile(candidate):
|
|
return candidate
|
|
return None
|
|
|
|
|
|
# ── Manifest helpers ──
|
|
|
|
def update_manifest(name, values):
|
|
"""Set KEY="value" lines in an app manifest, replacing existing keys."""
|
|
for key, value in values.items():
|
|
if re.search(r'["`$\\\n]', value):
|
|
raise ValueError(f"unsafe characters in {key}")
|
|
manifest_path = os.path.join(BASE_DIR, "state", "apps", f"{name}.env")
|
|
with open(manifest_path, "r", encoding="utf-8") as fh:
|
|
lines = fh.readlines()
|
|
remaining = dict(values)
|
|
out = []
|
|
for line in lines:
|
|
key = line.split("=", 1)[0].strip()
|
|
if key in remaining:
|
|
out.append(f'{key}="{remaining.pop(key)}"\n')
|
|
else:
|
|
out.append(line if line.endswith("\n") else line + "\n")
|
|
for key, value in remaining.items():
|
|
out.append(f'{key}="{value}"\n')
|
|
with open(manifest_path, "w", encoding="utf-8") as fh:
|
|
fh.writelines(out)
|
|
|
|
|
|
def manifest_routes(env):
|
|
routes_raw = env.get("APP_ROUTES", "")
|
|
# Backward compat: build from old APP_DOMAIN/APP_PORT/APP_UPSTREAM
|
|
if not routes_raw and "APP_DOMAIN" in env:
|
|
upstream = env.get("APP_UPSTREAM", f"127.0.0.1:{env.get('APP_PORT', '18080')}")
|
|
domains = env.get("APP_DOMAINS", env["APP_DOMAIN"])
|
|
routes_raw = ",".join(f"{d.strip()}|{upstream}" for d in domains.split(",") if d.strip())
|
|
routes = []
|
|
for entry in routes_raw.split(","):
|
|
entry = entry.strip()
|
|
if not entry:
|
|
continue
|
|
fields = entry.split("|", 2)
|
|
if len(fields) < 2:
|
|
continue
|
|
route = {"domain": fields[0].strip(), "upstream": fields[1].strip()}
|
|
if len(fields) > 2 and fields[2].strip():
|
|
route["path"] = fields[2].strip()
|
|
routes.append(route)
|
|
return routes
|
|
|
|
|
|
def load_app_summaries():
|
|
"""Read every app manifest directly (much faster than shelling out per app)."""
|
|
apps_dir = os.path.join(BASE_DIR, "state", "apps")
|
|
try:
|
|
entries = sorted(os.listdir(apps_dir))
|
|
except OSError:
|
|
return []
|
|
apps = []
|
|
for fname in entries:
|
|
if not fname.endswith(".env"):
|
|
continue
|
|
name = fname[:-4]
|
|
if not is_safe_name(name):
|
|
continue
|
|
try:
|
|
with open(os.path.join(apps_dir, fname), "r", encoding="utf-8") as fh:
|
|
env = parse_env_blob(fh.read())
|
|
except OSError:
|
|
continue
|
|
repo_url = redact_credentials(env.get("APP_REPO_URL", ""), "")
|
|
apps.append({
|
|
"name": name,
|
|
"routes": manifest_routes(env),
|
|
"auth": env.get("APP_AUTH_PROTECTED", "true") == "true",
|
|
"compose_file": env.get("APP_COMPOSE_FILE", ""),
|
|
"repo_url": repo_url,
|
|
"repo_branch": env.get("APP_REPO_BRANCH", ""),
|
|
"repo_provider": repo_provider(repo_url) if repo_url else "",
|
|
"repo_web_url": repo_web_url(repo_url) if repo_url else "",
|
|
"env_count": len(read_app_env(name)),
|
|
"env_inject": env.get("APP_ENV_INJECT", "true") != "false",
|
|
})
|
|
return apps
|
|
|
|
|
|
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 = {}
|
|
# Parse whatever was printed even on a non-zero exit, so one failing
|
|
# `podman volume ls` doesn't hide the app's default data folder.
|
|
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 _decode_containers(stdout):
|
|
"""`compose ps --format json` prints either a JSON array or one object per line,
|
|
sometimes mixed with other output. Returns a list of dicts, or None."""
|
|
lines = stdout.splitlines()
|
|
for i, line in enumerate(lines):
|
|
if line.lstrip().startswith("["):
|
|
try:
|
|
data, _ = json.JSONDecoder().raw_decode("\n".join(lines[i:]).lstrip())
|
|
except ValueError:
|
|
continue
|
|
if isinstance(data, list):
|
|
return [c for c in data if isinstance(c, dict)]
|
|
items = []
|
|
for line in lines:
|
|
line = line.strip()
|
|
if not line.startswith("{"):
|
|
continue
|
|
try:
|
|
obj = json.loads(line)
|
|
except ValueError:
|
|
continue
|
|
if isinstance(obj, dict):
|
|
items.append(obj)
|
|
return items or None
|
|
|
|
|
|
def _container_name(c):
|
|
name = c.get("Name") or c.get("name")
|
|
if not name:
|
|
names = c.get("Names")
|
|
if isinstance(names, list) and names:
|
|
name = names[0]
|
|
elif isinstance(names, str):
|
|
name = names
|
|
return name or "?"
|
|
|
|
|
|
def parse_status_output(stdout):
|
|
"""Summarise panelctl status output as running / partial / stopped / unknown."""
|
|
stdout = stdout or ""
|
|
containers = _decode_containers(stdout)
|
|
if containers is None:
|
|
text = stdout.lower()
|
|
if not text.strip() or "no containers" in text:
|
|
return {"state": "stopped", "running": False, "running_count": 0, "total": 0, "containers": []}
|
|
running = re.search(r"\b(up|running)\b", text) is not None
|
|
return {
|
|
"state": "running" if running else "unknown",
|
|
"running": running,
|
|
"running_count": None,
|
|
"total": None,
|
|
"containers": [],
|
|
"raw": stdout,
|
|
}
|
|
|
|
parsed = []
|
|
for c in containers:
|
|
state = str(c.get("State") or c.get("state") or "").lower()
|
|
status = str(c.get("Status") or c.get("status") or "")
|
|
is_running = state == "running" or status.lower().startswith("up")
|
|
parsed.append({
|
|
"name": _container_name(c),
|
|
"state": state or ("running" if is_running else "unknown"),
|
|
"status": status,
|
|
"image": c.get("Image") or c.get("image") or "",
|
|
"running": is_running,
|
|
})
|
|
running_count = sum(1 for c in parsed if c["running"])
|
|
total = len(parsed)
|
|
if total and running_count == total:
|
|
state = "running"
|
|
elif running_count:
|
|
state = "partial"
|
|
else:
|
|
state = "stopped"
|
|
return {
|
|
"state": state,
|
|
"running": running_count > 0,
|
|
"running_count": running_count,
|
|
"total": total,
|
|
"containers": parsed,
|
|
}
|
|
|
|
|
|
def app_status(name):
|
|
return parse_status_output(run_panelctl(["status", name])["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
|
|
|
|
|
|
# Actions that only read state and may run alongside anything else.
|
|
LOCK_FREE_ACTIONS = {"validate-compose"}
|
|
|
|
|
|
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)))
|
|
# The UI is a single file that changes with every rebuild.
|
|
self.send_header("Cache-Control", "no-cache")
|
|
self.end_headers()
|
|
self.wfile.write(data)
|
|
except OSError:
|
|
self._json(500, {"ok": False, "error": "failed to read file"})
|
|
|
|
def _read_json(self):
|
|
# Cached: do_POST may read the body before dispatching.
|
|
if hasattr(self, "_payload"):
|
|
return self._payload
|
|
length = int(self.headers.get("Content-Length", "0"))
|
|
if length == 0:
|
|
self._payload = {}
|
|
else:
|
|
raw = self.rfile.read(length)
|
|
self._payload = json.loads(raw.decode("utf-8"))
|
|
return self._payload
|
|
|
|
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
|
|
|
|
# /integrations — Forgejo connection and the panel's SSH deploy key
|
|
if path == "/integrations":
|
|
token = forgejo_token()
|
|
forgejo = {
|
|
"configured": bool(FORGEJO_URL),
|
|
"url": FORGEJO_URL,
|
|
"ssh_url": FORGEJO_SSH_URL,
|
|
"has_token": bool(token),
|
|
"user": None,
|
|
}
|
|
if FORGEJO_URL and token:
|
|
try:
|
|
forgejo["user"] = (forgejo_api("/api/v1/user", timeout=5) or {}).get("login")
|
|
except ForgejoError as exc:
|
|
forgejo["error"] = str(exc)
|
|
self._json(200, {"ok": True, "forgejo": forgejo, "ssh": {"public_key": ssh_public_key()}})
|
|
return
|
|
|
|
# /forgejo/repos?q= — repositories visible to the stored token (public ones without)
|
|
if path == "/forgejo/repos":
|
|
q = query.get("q", [""])[0].strip()
|
|
try:
|
|
data = forgejo_api(f"/api/v1/repos/search?q={quote(q)}&limit=50&sort=updated&order=desc")
|
|
except ForgejoError as exc:
|
|
self._json(502, {"ok": False, "error": str(exc)})
|
|
return
|
|
repos = [{
|
|
"full_name": r.get("full_name", ""),
|
|
"description": r.get("description", ""),
|
|
"private": bool(r.get("private")),
|
|
"empty": bool(r.get("empty")),
|
|
"archived": bool(r.get("archived")),
|
|
"default_branch": r.get("default_branch", ""),
|
|
"clone_url": r.get("clone_url", ""),
|
|
"ssh_url": r.get("ssh_url", ""),
|
|
"html_url": r.get("html_url", ""),
|
|
"updated_at": r.get("updated_at", ""),
|
|
} for r in (data or {}).get("data", [])]
|
|
self._json(200, {"ok": True, "repos": repos, "authenticated": bool(forgejo_token())})
|
|
return
|
|
|
|
# /forgejo/branches?repo=owner/name
|
|
if path == "/forgejo/branches":
|
|
repo = query.get("repo", [""])[0].strip()
|
|
if not FORGEJO_REPO_RE.match(repo):
|
|
self._json(400, {"ok": False, "error": "repo must look like owner/name"})
|
|
return
|
|
try:
|
|
data = forgejo_api(f"/api/v1/repos/{repo}/branches?limit=100")
|
|
except ForgejoError as exc:
|
|
self._json(502, {"ok": False, "error": str(exc)})
|
|
return
|
|
self._json(200, {"ok": True, "branches": [b.get("name", "") for b in (data or [])]})
|
|
return
|
|
|
|
# /apps/<name>/env — environment variables used when deploying
|
|
if len(parts) == 3 and parts[0] == "apps" and parts[2] == "env":
|
|
name = parts[1]
|
|
app, err = read_app_info(name)
|
|
if err is not None:
|
|
self._json(404, err)
|
|
return
|
|
self._json(200, {
|
|
"ok": True,
|
|
"name": name,
|
|
"vars": read_app_env(name),
|
|
"inject": app.get("APP_ENV_INJECT", "true") != "false",
|
|
})
|
|
return
|
|
|
|
# /status — every app with routes, container status and running operation.
|
|
# This is what the UI polls, so it is one request regardless of app count.
|
|
if path == "/status":
|
|
apps = load_app_summaries()
|
|
names = [a["name"] for a in apps]
|
|
statuses = {}
|
|
if names:
|
|
with ThreadPoolExecutor(max_workers=min(8, len(names))) as pool:
|
|
statuses = dict(zip(names, pool.map(app_status, names)))
|
|
busy = busy_snapshot()
|
|
for app in apps:
|
|
app["status"] = statuses.get(app["name"], {"state": "unknown"})
|
|
app["busy"] = busy.get(app["name"])
|
|
self._json(200, {"ok": True, "time": int(time.time()), "apps": apps})
|
|
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
|
|
# New format: name domain|upstream routes=N auth=bool [repo_url]
|
|
first_route = fields[1]
|
|
route_parts = first_route.split("|")
|
|
domain = route_parts[0].split(",")[0] if route_parts else first_route
|
|
upstream = route_parts[1] if len(route_parts) > 1 else ""
|
|
route_count_str = fields[2].replace("routes=", "")
|
|
# Backward compat: fields[2] may be upstream if old format
|
|
if not route_count_str.isdigit():
|
|
upstream = fields[2]
|
|
route_count_str = "1"
|
|
apps.append({
|
|
"name": fields[0],
|
|
"domain": domain,
|
|
"domains": domain,
|
|
"upstream": upstream,
|
|
"first_route": first_route,
|
|
"route_count": route_count_str,
|
|
"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>/routes — get parsed routes
|
|
if len(parts) == 3 and parts[0] == "apps" and parts[2] == "routes":
|
|
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
|
|
routes = manifest_routes(parse_env_blob(result["stdout"]))
|
|
self._json(200, {"ok": True, "name": name, "routes": routes})
|
|
return
|
|
|
|
# /apps/<name>/repo[?fetch=1] — git source info; fetch=1 also checks the remote
|
|
if len(parts) == 3 and parts[0] == "apps" and parts[2] == "repo":
|
|
name = parts[1]
|
|
app, err = read_app_info(name)
|
|
if err is not None:
|
|
self._json(404, err)
|
|
return
|
|
repo_url = app.get("APP_REPO_URL", "")
|
|
if not repo_url:
|
|
self._json(404, {"ok": False, "error": "app is not linked to a git repository"})
|
|
return
|
|
repo_dir = os.path.join(app["APP_STACK_DIR"], "repo")
|
|
branch = app.get("APP_REPO_BRANCH", "")
|
|
info = {
|
|
"ok": True,
|
|
"name": name,
|
|
"url": redact_credentials(repo_url, ""),
|
|
"web_url": repo_web_url(repo_url),
|
|
"provider": repo_provider(repo_url),
|
|
"ssh": is_ssh_url(repo_url),
|
|
"branch": branch,
|
|
"cloned": os.path.isdir(os.path.join(repo_dir, ".git")),
|
|
}
|
|
if info["ssh"]:
|
|
info["public_key"] = ssh_public_key()
|
|
if info["cloned"]:
|
|
info["commit"] = repo_commit(repo_dir)
|
|
status = run_git(["status", "--porcelain", "--untracked-files=no"], cwd=repo_dir, timeout=15)
|
|
info["dirty"] = bool(status["stdout"]) if status["ok"] else None
|
|
if query.get("fetch", ["0"])[0] == "1":
|
|
ref = branch or repo_current_branch(repo_dir)
|
|
fetched = run_git(["fetch", "--quiet", "origin", ref], cwd=repo_dir)
|
|
if not fetched["ok"]:
|
|
info["fetch_error"] = git_error(fetched["stderr"]) or "git fetch failed"
|
|
else:
|
|
info["remote"] = repo_commit(repo_dir, "FETCH_HEAD")
|
|
count = run_git(["rev-list", "--count", "HEAD..FETCH_HEAD"], cwd=repo_dir, timeout=15)
|
|
info["behind"] = int(count["stdout"]) if count["ok"] and count["stdout"].isdigit() else None
|
|
self._json(200, info)
|
|
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()
|
|
|
|
name, action = None, None
|
|
if path == "/apps/init":
|
|
try:
|
|
payload = self._read_json()
|
|
except Exception as exc:
|
|
self._json(400, {"ok": False, "error": f"invalid payload: {exc}"})
|
|
return
|
|
name = str(payload.get("name", "")) if isinstance(payload, dict) else ""
|
|
action = "init"
|
|
elif len(parts) >= 3 and parts[0] == "apps" and parts[2] not in LOCK_FREE_ACTIONS:
|
|
name, action = parts[1], parts[2]
|
|
|
|
if not name:
|
|
self._handle_post(path, parts, query)
|
|
return
|
|
try:
|
|
with app_operation(name, action):
|
|
self._handle_post(path, parts, query)
|
|
except AppBusy as exc:
|
|
self._json(409, {"ok": False, "error": str(exc), "busy": exc.action})
|
|
|
|
def _handle_post(self, path, parts, query):
|
|
# POST /apps/init
|
|
if path == "/apps/init":
|
|
try:
|
|
payload = self._read_json()
|
|
name = payload["name"]
|
|
auth = str(payload.get("auth", True)).lower()
|
|
source_type = payload.get("source_type", "default")
|
|
|
|
# Build routes string: "domain|upstream,domain|upstream,..."
|
|
routes_parts = []
|
|
if "routes" in payload and isinstance(payload["routes"], list):
|
|
for r in payload["routes"]:
|
|
d = r.get("domain", "").strip()
|
|
u = r.get("upstream", "").strip()
|
|
p = r.get("path", "").strip()
|
|
if d and u:
|
|
if p:
|
|
routes_parts.append(f"{d}|{u}|{p}")
|
|
else:
|
|
routes_parts.append(f"{d}|{u}")
|
|
elif "domain" in payload and "port" in payload:
|
|
# Backward compat: single domain + port
|
|
domain_str = payload.get("domain", "")
|
|
if "domains" in payload and isinstance(payload["domains"], list):
|
|
domain_str = ",".join(payload["domains"])
|
|
port = str(payload["port"])
|
|
for d in domain_str.split(","):
|
|
d = d.strip()
|
|
if d:
|
|
routes_parts.append(f"{d}|127.0.0.1:{port}")
|
|
else:
|
|
self._json(400, {"ok": False, "error": "missing 'routes' array or 'domain'+'port' fields"})
|
|
return
|
|
|
|
routes_str = ",".join(routes_parts)
|
|
except Exception as exc:
|
|
self._json(400, {"ok": False, "error": f"invalid payload: {exc}"})
|
|
return
|
|
|
|
if source_type == "github": # older clients
|
|
source_type = "git"
|
|
if source_type not in ["default", "raw", "git"]:
|
|
self._json(400, {"ok": False, "error": "invalid source_type"})
|
|
return
|
|
|
|
# Validate everything before creating anything.
|
|
try:
|
|
env_items = validate_env(payload.get("env"))
|
|
except ValueError as exc:
|
|
self._json(400, {"ok": False, "error": str(exc)})
|
|
return
|
|
env_inject = payload.get("env_inject", True) is not False
|
|
|
|
if source_type == "git":
|
|
repo_url = str(payload.get("repo_url") or payload.get("github_url") or "").strip()
|
|
branch = str(payload.get("repo_branch") or payload.get("github_branch") or "").strip()
|
|
token = str(payload.get("repo_token") or payload.get("github_pat") or "").strip()
|
|
if not REPO_URL_RE.match(repo_url) or repo_url.startswith("-"):
|
|
self._json(400, {"ok": False, "error": "repository URL must be an https://, ssh:// or git@host:owner/repo URL"})
|
|
return
|
|
if branch and not BRANCH_RE.match(branch):
|
|
self._json(400, {"ok": False, "error": f"invalid branch name '{branch}'"})
|
|
return
|
|
# Clone a Forgejo repository with the panel's stored token.
|
|
if not token and payload.get("use_forgejo_token") and is_forgejo_https_url(repo_url):
|
|
token = forgejo_token()
|
|
|
|
try:
|
|
result = run_panelctl(["init", name, routes_str, 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
|
|
|
|
summary = "initialized successfully"
|
|
if source_type == "git":
|
|
# Any git host works (Forgejo, GitHub, ...), over https or ssh.
|
|
# An https token is embedded in the clone URL, so later syncs
|
|
# reuse it from .git/config; ssh uses the panel's deploy key.
|
|
target_dir = os.path.join(app["APP_STACK_DIR"], "repo")
|
|
if os.path.exists(target_dir):
|
|
shutil.rmtree(target_dir)
|
|
|
|
cloned = clone_repo(repo_url, branch, target_dir, token)
|
|
if not cloned["ok"]:
|
|
run_panelctl(["remove", name])
|
|
self._json(400, {
|
|
"ok": False,
|
|
"error": f"git clone failed: {git_error(cloned['stderr'])}",
|
|
"stderr": cloned["stderr"],
|
|
})
|
|
return
|
|
branch = branch or repo_current_branch(target_dir) or "main"
|
|
|
|
compose_path = find_compose_file(target_dir)
|
|
if not compose_path:
|
|
run_panelctl(["remove", name])
|
|
self._json(400, {"ok": False, "error": "could not find a compose file in the repository root"})
|
|
return
|
|
|
|
try:
|
|
update_manifest(name, {
|
|
"APP_COMPOSE_FILE": compose_path,
|
|
"APP_REPO_URL": redact_credentials(repo_url, ""),
|
|
"APP_REPO_BRANCH": branch,
|
|
})
|
|
except (OSError, ValueError) as exc:
|
|
run_panelctl(["remove", name])
|
|
self._json(500, {"ok": False, "error": f"failed to update manifest: {exc}"})
|
|
return
|
|
|
|
commit = repo_commit(target_dir)
|
|
summary = f"cloned {branch} at {commit['short']}: {commit['subject']}" if commit else "cloned"
|
|
|
|
if env_items or not env_inject:
|
|
write_app_env(name, env_items)
|
|
update_manifest(name, {"APP_ENV_INJECT": "true" if env_inject else "false"})
|
|
summary += f"\n{len(env_items)} environment variable(s) set"
|
|
|
|
self._json(200, {"ok": True, "code": 0, "stdout": summary})
|
|
except Exception as exc:
|
|
run_panelctl(["remove", name])
|
|
self._json(500, {"ok": False, "error": f"init failed: {exc}"})
|
|
return
|
|
|
|
# POST /integrations/forgejo {"token": "..."} — verify and store ("" clears it)
|
|
if path == "/integrations/forgejo":
|
|
if not FORGEJO_URL:
|
|
self._json(400, {"ok": False, "error": "no Forgejo instance is configured (PANEL_FORGEJO_URL)"})
|
|
return
|
|
try:
|
|
token = str(self._read_json().get("token", "")).strip()
|
|
except Exception as exc:
|
|
self._json(400, {"ok": False, "error": f"invalid payload: {exc}"})
|
|
return
|
|
if not token:
|
|
try:
|
|
os.remove(FORGEJO_TOKEN_FILE)
|
|
except FileNotFoundError:
|
|
pass
|
|
self._json(200, {"ok": True, "has_token": False})
|
|
return
|
|
try:
|
|
user = (forgejo_api("/api/v1/user", token=token) or {}).get("login")
|
|
except ForgejoError as exc:
|
|
self._json(400, {"ok": False, "error": str(exc)})
|
|
return
|
|
write_private_file(FORGEJO_TOKEN_FILE, token + "\n")
|
|
self._json(200, {"ok": True, "has_token": True, "user": user})
|
|
return
|
|
|
|
if len(parts) >= 3 and parts[0] == "apps":
|
|
name = parts[1]
|
|
action = parts[2]
|
|
|
|
# POST /apps/<name>/env — replace environment variables, optionally redeploy
|
|
if action == "env":
|
|
app, err = read_app_info(name)
|
|
if err is not None:
|
|
self._json(404, err)
|
|
return
|
|
try:
|
|
payload = self._read_json()
|
|
items = validate_env(payload.get("vars", []))
|
|
except ValueError as exc:
|
|
self._json(400, {"ok": False, "error": str(exc)})
|
|
return
|
|
inject = payload.get("inject", True) is not False
|
|
try:
|
|
write_app_env(name, items)
|
|
update_manifest(name, {"APP_ENV_INJECT": "true" if inject else "false"})
|
|
except (OSError, ValueError) as exc:
|
|
self._json(500, {"ok": False, "error": f"failed to save variables: {exc}"})
|
|
return
|
|
result = {"ok": True, "name": name, "count": len(items),
|
|
"stdout": f"saved {len(items)} environment variable(s)"}
|
|
if payload.get("deploy"):
|
|
deployed = run_panelctl(["deploy", name])
|
|
deployed["stdout"] = "\n".join(filter(None, [result["stdout"], deployed["stdout"]]))
|
|
deployed["count"] = len(items)
|
|
self._json(200 if deployed["ok"] else 400, deployed)
|
|
return
|
|
self._json(200, result)
|
|
return
|
|
|
|
# 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
|
|
|
|
# POST /apps/<name>/routes — hot update routes
|
|
if action == "routes":
|
|
if not is_safe_name(name):
|
|
self._json(400, {"ok": False, "error": "invalid app name"})
|
|
return
|
|
try:
|
|
payload = self._read_json()
|
|
except Exception as exc:
|
|
self._json(400, {"ok": False, "error": f"invalid payload: {exc}"})
|
|
return
|
|
route_list = payload.get("routes", [])
|
|
if not isinstance(route_list, list) or not route_list:
|
|
self._json(400, {"ok": False, "error": "routes must be a non-empty array"})
|
|
return
|
|
routes_parts = []
|
|
for r in route_list:
|
|
d = r.get("domain", "").strip()
|
|
u = r.get("upstream", "").strip()
|
|
p = r.get("path", "").strip()
|
|
if not d or not u:
|
|
self._json(400, {"ok": False, "error": "each route needs 'domain' and 'upstream'"})
|
|
return
|
|
if p:
|
|
routes_parts.append(f"{d}|{u}|{p}")
|
|
else:
|
|
routes_parts.append(f"{d}|{u}")
|
|
routes_str = ",".join(routes_parts)
|
|
result = run_panelctl(["set-routes", name, routes_str])
|
|
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 — sync the checkout to the remote branch and redeploy.
|
|
# The repository is the source of truth: fetch + hard reset, so local
|
|
# edits or force-pushes never leave the checkout stuck mid-merge.
|
|
if action == "repo-pull":
|
|
if not is_safe_name(name):
|
|
self._json(400, {"ok": False, "error": "invalid app name"})
|
|
return
|
|
|
|
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()
|
|
if not repo_url:
|
|
self._json(400, {"ok": False, "error": "app is not linked to a git repository"})
|
|
return
|
|
|
|
repo_dir = os.path.join(app["APP_STACK_DIR"], "repo")
|
|
branch = app.get("APP_REPO_BRANCH", "").strip()
|
|
git_log = []
|
|
before = None
|
|
|
|
if os.path.isdir(os.path.join(repo_dir, ".git")):
|
|
before = repo_commit(repo_dir)
|
|
ref = branch or repo_current_branch(repo_dir)
|
|
if not ref:
|
|
self._json(400, {"ok": False, "error": "cannot determine which branch to sync"})
|
|
return
|
|
fetched = run_git(["fetch", "origin", ref], cwd=repo_dir)
|
|
if not fetched["ok"]:
|
|
self._json(400, {
|
|
"ok": False,
|
|
"error": f"git fetch failed: {git_error(fetched['stderr'])}",
|
|
"stderr": fetched["stderr"],
|
|
})
|
|
return
|
|
reset = run_git(["reset", "--hard", "FETCH_HEAD"], cwd=repo_dir, timeout=60)
|
|
if not reset["ok"]:
|
|
self._json(400, {
|
|
"ok": False,
|
|
"error": f"git reset failed: {git_error(reset['stderr'])}",
|
|
"stderr": reset["stderr"],
|
|
})
|
|
return
|
|
git_log.append(reset["stdout"])
|
|
else:
|
|
# No checkout yet (e.g. deleted by hand): clone it fresh.
|
|
if os.path.exists(repo_dir):
|
|
shutil.rmtree(repo_dir)
|
|
cloned = clone_repo(repo_url, branch, repo_dir)
|
|
if not cloned["ok"]:
|
|
self._json(400, {
|
|
"ok": False,
|
|
"error": f"git clone failed: {git_error(cloned['stderr'])}",
|
|
"stderr": cloned["stderr"],
|
|
})
|
|
return
|
|
git_log.append("cloned repository")
|
|
|
|
after = repo_commit(repo_dir)
|
|
compose_path = find_compose_file(repo_dir)
|
|
if not compose_path:
|
|
self._json(400, {"ok": False, "error": "compose file not found in repository root"})
|
|
return
|
|
try:
|
|
update_manifest(name, {"APP_COMPOSE_FILE": compose_path})
|
|
except (OSError, ValueError) as exc:
|
|
self._json(500, {"ok": False, "error": f"failed to update manifest: {exc}"})
|
|
return
|
|
|
|
result = run_panelctl(["deploy", name])
|
|
result["stdout"] = "\n".join(filter(None, git_log + [result["stdout"]]))
|
|
result["before"] = before
|
|
result["after"] = after
|
|
result["changed"] = not before or not after or before["sha"] != after["sha"]
|
|
self._json(200 if result["ok"] else 400, result)
|
|
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 = ThreadingHTTPServer((BIND, PORT), Handler)
|
|
server.daemon_threads = True
|
|
print(f"panel-api listening on http://{BIND}:{PORT}")
|
|
print(f"frontend dir: {FRONTEND_DIR}")
|
|
server.serve_forever()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|