agent
71fed39f17
Deployments: deploy, restart and git sync now run in the background, one at
a time per app (a newer request replaces a queued one). Each run is recorded
in SQLite with its log, streamed to the UI while it runs, and can be
cancelled. Any earlier deployment can be deployed again, which rolls back to
its commit, or to its saved compose file for compose apps.
Auto deploy: POST /hooks/<app>, verified with the app's secret (Forgejo,
Gitea and GitHub HMAC signatures, or the secret as a token for CI). With a
Forgejo token stored, the panel adds the webhook to the repository itself.
The NixOS module routes /hooks/* past Authelia. Caddy matches the cleaned
path but forwards the original, so the panel refuses dot segments and only
accepts webhook deliveries from that route (tagged with X-Panel-Hook).
Domains: a route can point at a compose service's container port
("web:8080"). The panel picks a free 127.0.0.1 port and panelctl publishes
it through a generated .panel-ports.yaml override, so compose files need no
ports: section. Existing host:port upstreams keep working.
Logs stream live over server-sent events, with service and text filters.
A sampler keeps an hour of CPU and memory per container for the new
Monitoring tab. The Terminal tab opens `podman exec` in a container over a
WebSocket, using xterm.js bundled by the Nix package.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01UbWSNkXxZhYf7eqHTyx3Bf
3095 lines
124 KiB
Python
3095 lines
124 KiB
Python
#!/usr/bin/env python3
|
|
"""panel-api — HTTP wrapper around panelctl with a web UI."""
|
|
|
|
import base64
|
|
import collections
|
|
import fcntl
|
|
import hashlib
|
|
import hmac
|
|
import json
|
|
import os
|
|
import pty
|
|
import re
|
|
import secrets
|
|
import select
|
|
import shlex
|
|
import shutil
|
|
import signal
|
|
import socket
|
|
import sqlite3
|
|
import struct
|
|
import subprocess
|
|
import termios
|
|
import threading
|
|
import time
|
|
import traceback
|
|
import urllib.error
|
|
import urllib.request
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
from contextlib import contextmanager
|
|
|
|
try:
|
|
import yaml # optional: lets the panel suggest services and ports from compose files
|
|
except ImportError: # pragma: no cover
|
|
yaml = None
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
from urllib.parse import urlparse, parse_qs, quote, unquote
|
|
|
|
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 by the NixOS module). 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 ""
|
|
|
|
# Public address of the panel (e.g. https://panel.example.com), used for webhook
|
|
# URLs. Without it the address the browser used is taken from the request.
|
|
PUBLIC_URL = os.environ.get("PANEL_PUBLIC_URL", "").rstrip("/")
|
|
|
|
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")
|
|
DB_PATH = os.path.join(PANEL_STATE_DIR, "panel.db")
|
|
DEPLOY_DIR = os.path.join(PANEL_STATE_DIR, "deployments")
|
|
HOOK_DIR = os.path.join(PANEL_STATE_DIR, "hooks")
|
|
|
|
# Host ports the panel hands out for routes that point at a compose service.
|
|
PORT_RANGE = (18000, 19999)
|
|
# Deployments (and their logs) kept per app.
|
|
DEPLOY_KEEP = 50
|
|
# How long a single deployment command may run (image pulls and builds can be slow).
|
|
DEPLOY_TIMEOUT = int(os.environ.get("PANEL_DEPLOY_TIMEOUT", "3600"))
|
|
WEBHOOK_MAX_BODY = 5 * 1024 * 1024
|
|
|
|
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 try_acquire_app(name, action):
|
|
"""Non-blocking version of app_operation for background jobs."""
|
|
with _busy_lock:
|
|
if name in _busy:
|
|
return False
|
|
_busy[name] = action
|
|
return True
|
|
|
|
|
|
def release_app(name):
|
|
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, method="GET", body=None):
|
|
if not FORGEJO_API_URL:
|
|
raise ForgejoError("no Forgejo instance is configured")
|
|
token = forgejo_token() if token is None else token
|
|
data = json.dumps(body).encode("utf-8") if body is not None else None
|
|
req = urllib.request.Request(FORGEJO_API_URL + path, data=data, method=method,
|
|
headers={"Accept": "application/json"})
|
|
if data is not None:
|
|
req.add_header("Content-Type", "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 == 401:
|
|
raise ForgejoError("Forgejo rejected the token") from exc
|
|
if exc.code == 403:
|
|
raise ForgejoError("the Forgejo token isn't allowed to do this") 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 ──
|
|
|
|
_manifest_lock = threading.Lock()
|
|
|
|
|
|
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}")
|
|
with _manifest_lock:
|
|
_update_manifest(name, values)
|
|
|
|
|
|
def read_manifest(name):
|
|
"""The app's manifest as a dict, or None when the app doesn't exist."""
|
|
if not is_safe_name(name):
|
|
return None
|
|
try:
|
|
with open(os.path.join(BASE_DIR, "state", "apps", f"{name}.env"), "r", encoding="utf-8") as fh:
|
|
return parse_env_blob(fh.read())
|
|
except OSError:
|
|
return None
|
|
|
|
|
|
def _update_manifest(name, values):
|
|
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("|", 3)
|
|
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()
|
|
if len(fields) > 3 and fields[3].strip():
|
|
svc, _, port = fields[3].strip().rpartition(":")
|
|
route["service"] = svc
|
|
route["port"] = int(port) if port.isdigit() else None
|
|
routes.append(route)
|
|
return routes
|
|
|
|
|
|
# ── Routes to compose services ──
|
|
# A route either names an upstream directly (host:port, e.g. a service on the
|
|
# host) or targets a compose service's container port. For the latter the panel
|
|
# picks a free 127.0.0.1 port and panelctl publishes the container port on it.
|
|
|
|
_port_lock = threading.Lock()
|
|
_IPV4_RE = re.compile(r"^\d{1,3}(?:\.\d{1,3}){3}$")
|
|
|
|
|
|
def parse_route_target(value):
|
|
"""'web:8080' / '8080' -> ('service', 'web' or '', 8080); '127.0.0.1:9000' -> ('upstream', ...)."""
|
|
value = str(value or "").strip()
|
|
if re.fullmatch(r"\d+", value):
|
|
port = int(value)
|
|
if not 1 <= port <= 65535:
|
|
raise ValueError(f"port {port} must be between 1 and 65535")
|
|
return ("service", "", port)
|
|
m = re.fullmatch(r"(\[[0-9A-Fa-f:]+\]|[A-Za-z0-9._-]+):(\d+)", value)
|
|
if not m:
|
|
raise ValueError(f"'{value}' should be a port, service:port or host:port")
|
|
host, port = m.group(1), int(m.group(2))
|
|
if not 1 <= port <= 65535:
|
|
raise ValueError(f"port {port} must be between 1 and 65535")
|
|
if host == "localhost" or host.startswith("[") or _IPV4_RE.match(host) or "." in host:
|
|
return ("upstream", f"{host}:{port}", None)
|
|
return ("service", host, port)
|
|
|
|
|
|
def used_route_ports(exclude_app=None):
|
|
used = set()
|
|
for app in load_app_summaries():
|
|
if app["name"] == exclude_app:
|
|
continue
|
|
for r in app["routes"]:
|
|
port = r["upstream"].rsplit(":", 1)[-1]
|
|
if port.isdigit():
|
|
used.add(int(port))
|
|
return used
|
|
|
|
|
|
def port_is_free(port):
|
|
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
|
|
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
|
|
try:
|
|
sock.bind(("127.0.0.1", port))
|
|
return True
|
|
except OSError:
|
|
return False
|
|
|
|
|
|
def compose_services(name):
|
|
"""Service names of an app's compose file, or None when they can't be read."""
|
|
result = run_panelctl(["services", name])
|
|
if not result["ok"]:
|
|
return None
|
|
return [line.strip() for line in result["stdout"].splitlines() if re.fullmatch(r"[A-Za-z0-9._-]+", line.strip())]
|
|
|
|
|
|
def resolve_routes(name, routes, services=None):
|
|
"""Validate route dicts from the API and turn them into manifest entries.
|
|
|
|
Each route has a domain, an optional path and either a target (port,
|
|
service:port or host:port) or, as before, an upstream. Routes to a service
|
|
keep the host port they already had, so saving routes doesn't move them.
|
|
"""
|
|
if not isinstance(routes, list) or not routes:
|
|
raise ValueError("at least one route is required")
|
|
current = {}
|
|
manifest = os.path.join(BASE_DIR, "state", "apps", f"{name}.env")
|
|
if os.path.isfile(manifest):
|
|
with open(manifest, "r", encoding="utf-8") as fh:
|
|
for r in manifest_routes(parse_env_blob(fh.read())):
|
|
if r.get("port"):
|
|
current[(r.get("service", ""), r["port"])] = r["upstream"]
|
|
|
|
with _port_lock:
|
|
taken = used_route_ports(exclude_app=name)
|
|
assigned = {}
|
|
entries = []
|
|
for r in routes:
|
|
if not isinstance(r, dict):
|
|
raise ValueError("each route must be an object")
|
|
domain = str(r.get("domain", "")).strip()
|
|
path = str(r.get("path", "") or "").strip()
|
|
if not domain:
|
|
raise ValueError("every route needs a domain")
|
|
if any(c in domain + path for c in "|,\n\"'`$\\ "):
|
|
raise ValueError(f"invalid characters in route {domain}")
|
|
target = r.get("target")
|
|
if target in (None, "") and r.get("port"):
|
|
target = f"{r.get('service', '')}:{r['port']}" if r.get("service") else str(r["port"])
|
|
if target in (None, ""):
|
|
upstream = str(r.get("upstream", "")).strip()
|
|
if not upstream:
|
|
raise ValueError(f"route {domain} needs a target (port or service:port)")
|
|
entries.append(f"{domain}|{upstream}|{path}" if path else f"{domain}|{upstream}")
|
|
continue
|
|
kind, svc_or_upstream, cport = parse_route_target(target)
|
|
if kind == "upstream":
|
|
entries.append(f"{domain}|{svc_or_upstream}|{path}" if path else f"{domain}|{svc_or_upstream}")
|
|
continue
|
|
svc = svc_or_upstream
|
|
if services is not None:
|
|
if not svc and len(services) == 1:
|
|
svc = services[0]
|
|
elif not svc and len(services) > 1:
|
|
raise ValueError(f"route {domain}: the compose file has several services ({', '.join(services)}), say which one, e.g. {services[0]}:{cport}")
|
|
elif svc and services and svc not in services:
|
|
raise ValueError(f"route {domain}: there is no service '{svc}' in the compose file ({', '.join(services)})")
|
|
key = (svc, cport)
|
|
if key not in assigned:
|
|
# Keep the port this target already had (also from before its service was named).
|
|
upstream = current.get(key) or current.get(("", cport))
|
|
in_use = taken | {int(u.rsplit(":", 1)[-1]) for u in assigned.values()}
|
|
if upstream and int(upstream.rsplit(":", 1)[-1]) in in_use:
|
|
upstream = None
|
|
if not upstream:
|
|
port = next((p for p in range(PORT_RANGE[0], PORT_RANGE[1] + 1)
|
|
if p not in in_use and port_is_free(p)), None)
|
|
if port is None:
|
|
raise ValueError("no free port left for the route")
|
|
upstream = f"127.0.0.1:{port}"
|
|
assigned[key] = upstream
|
|
entries.append(f"{domain}|{assigned[key]}|{path}|{svc}:{cport}")
|
|
return ",".join(entries)
|
|
|
|
|
|
def finalize_route_services(name):
|
|
"""Once an app's compose file exists: name the service of routes that only
|
|
gave a port, and check that the services routes name exist.
|
|
Returns an error message, or None."""
|
|
routes = manifest_routes(read_manifest(name) or {})
|
|
if not any(r.get("port") for r in routes):
|
|
return None
|
|
services = compose_services(name)
|
|
if not services:
|
|
return None # compose file unreadable for now; the deploy reports it
|
|
changed = False
|
|
entries = []
|
|
for r in routes:
|
|
fields = [r["domain"], r["upstream"], r.get("path", "")]
|
|
if r.get("port"):
|
|
svc = r.get("service")
|
|
if not svc:
|
|
if len(services) != 1:
|
|
return (f"route {r['domain']}: the compose file has several services ({', '.join(services)}), "
|
|
f"say which one, e.g. {services[0]}:{r['port']}")
|
|
svc, changed = services[0], True
|
|
elif svc not in services:
|
|
return f"route {r['domain']}: there is no service '{svc}' in the compose file ({', '.join(services)})"
|
|
fields.append(f"{svc}:{r['port']}")
|
|
entries.append("|".join(fields) if len(fields) == 4 else "|".join(fields).rstrip("|"))
|
|
if changed:
|
|
result = run_panelctl(["set-routes", name, ",".join(entries)])
|
|
if not result["ok"]:
|
|
return last_line(result["stderr"]) or "failed to update routes"
|
|
return None
|
|
|
|
|
|
def compose_service_ports(compose_text):
|
|
"""Services and the container ports they mention (ports/expose), for suggestions."""
|
|
if yaml is None:
|
|
return None
|
|
try:
|
|
doc = yaml.safe_load(compose_text) or {}
|
|
except yaml.YAMLError:
|
|
return None
|
|
services = doc.get("services") if isinstance(doc, dict) else None
|
|
if not isinstance(services, dict):
|
|
return None
|
|
out = []
|
|
for svc, spec in services.items():
|
|
ports = []
|
|
spec = spec if isinstance(spec, dict) else {}
|
|
for item in list(spec.get("ports") or []) + list(spec.get("expose") or []):
|
|
if isinstance(item, dict):
|
|
p = item.get("target")
|
|
else:
|
|
p = str(item).split("/")[0].rsplit(":", 1)[-1]
|
|
try:
|
|
p = int(str(p).split("-")[0])
|
|
except ValueError:
|
|
continue
|
|
if 0 < p < 65536 and p not in ports:
|
|
ports.append(p)
|
|
out.append({"name": str(svc), "ports": ports, "image": str(spec.get("image") or "")})
|
|
return out
|
|
|
|
|
|
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",
|
|
"autodeploy": env.get("APP_AUTODEPLOY", "false") == "true",
|
|
})
|
|
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
|
|
|
|
|
|
# ── Deployments ──
|
|
# Deploys, git syncs, restarts and rollbacks run in the background, one at a
|
|
# time per app, queued behind whatever else the app is doing. Each one is a row
|
|
# in SQLite plus a log file that the UI streams while it runs.
|
|
|
|
FINAL_STATES = ("success", "failed", "cancelled")
|
|
# What /status reports as the app's running operation, per job kind.
|
|
JOB_BUSY = {"deploy": "deploy", "sync": "repo-pull", "restart": "restart"}
|
|
ANSI_RE = re.compile(r"\x1b\[[0-9;?]*[ -/]*[@-~]|\x1b[()][A-Za-z0-9]|\x1b[=>]")
|
|
|
|
|
|
def deploy_dir(app):
|
|
return os.path.join(DEPLOY_DIR, app)
|
|
|
|
|
|
def deploy_log_path(app, dep_id):
|
|
return os.path.join(deploy_dir(app), f"{dep_id}.log")
|
|
|
|
|
|
def deploy_snapshot_path(app, dep_id):
|
|
return os.path.join(deploy_dir(app), f"{dep_id}.compose.yaml")
|
|
|
|
|
|
def clean_line(text):
|
|
"""Drop terminal escapes and carriage-return progress redraws from command output."""
|
|
text = ANSI_RE.sub("", text)
|
|
if "\r" in text:
|
|
text = "\n".join(part.rsplit("\r", 1)[-1] for part in text.split("\n"))
|
|
return text
|
|
|
|
|
|
class DeployStore:
|
|
def __init__(self, path):
|
|
os.makedirs(os.path.dirname(path), mode=0o700, exist_ok=True)
|
|
self.lock = threading.Lock()
|
|
self.db = sqlite3.connect(path, check_same_thread=False, isolation_level=None, timeout=10)
|
|
self.db.row_factory = sqlite3.Row
|
|
self.db.execute("""CREATE TABLE IF NOT EXISTS deployments (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
app TEXT NOT NULL,
|
|
kind TEXT NOT NULL,
|
|
trigger TEXT NOT NULL,
|
|
title TEXT NOT NULL DEFAULT '',
|
|
status TEXT NOT NULL,
|
|
created REAL NOT NULL,
|
|
started REAL,
|
|
finished REAL,
|
|
commit_sha TEXT,
|
|
commit_subject TEXT,
|
|
commit_author TEXT,
|
|
target TEXT,
|
|
error TEXT,
|
|
snapshot INTEGER NOT NULL DEFAULT 0)""")
|
|
self.db.execute("CREATE INDEX IF NOT EXISTS deployments_app ON deployments (app, id)")
|
|
# Whatever was still running or queued belonged to a previous panel process.
|
|
now = time.time()
|
|
self.db.execute("UPDATE deployments SET status='failed', finished=?, error='interrupted because the panel restarted' "
|
|
"WHERE status='running'", (now,))
|
|
self.db.execute("UPDATE deployments SET status='cancelled', finished=?, error='the panel restarted before it started' "
|
|
"WHERE status='queued'", (now,))
|
|
|
|
@staticmethod
|
|
def _dict(row):
|
|
if row is None:
|
|
return None
|
|
d = dict(row)
|
|
d["snapshot"] = bool(d["snapshot"])
|
|
d["commit_short"] = (d["commit_sha"] or "")[:7] or None
|
|
d["duration"] = round((d["finished"] or time.time()) - d["started"], 1) if d["started"] else None
|
|
return d
|
|
|
|
def _all(self, sql, args=()):
|
|
with self.lock:
|
|
return [self._dict(r) for r in self.db.execute(sql, args).fetchall()]
|
|
|
|
def create(self, app, kind, trigger, title, target=None):
|
|
with self.lock:
|
|
cur = self.db.execute(
|
|
"INSERT INTO deployments (app, kind, trigger, title, status, created, target) VALUES (?, ?, ?, ?, 'queued', ?, ?)",
|
|
(app, kind, trigger, title, time.time(), target))
|
|
dep_id = cur.lastrowid
|
|
return self.get(dep_id)
|
|
|
|
def update(self, dep_id, **fields):
|
|
if not fields:
|
|
return
|
|
cols = ", ".join(f"{k} = ?" for k in fields)
|
|
with self.lock:
|
|
self.db.execute(f"UPDATE deployments SET {cols} WHERE id = ?", (*fields.values(), dep_id))
|
|
|
|
def get(self, dep_id):
|
|
rows = self._all("SELECT * FROM deployments WHERE id = ?", (dep_id,))
|
|
return rows[0] if rows else None
|
|
|
|
def list(self, app, limit=30):
|
|
return self._all("SELECT * FROM deployments WHERE app = ? ORDER BY id DESC LIMIT ?", (app, limit))
|
|
|
|
def latest(self):
|
|
rows = self._all("SELECT * FROM deployments WHERE id IN (SELECT MAX(id) FROM deployments GROUP BY app)")
|
|
return {r["app"]: r for r in rows}
|
|
|
|
def delete_app(self, app):
|
|
with self.lock:
|
|
self.db.execute("DELETE FROM deployments WHERE app = ?", (app,))
|
|
shutil.rmtree(deploy_dir(app), ignore_errors=True)
|
|
|
|
def prune(self, app):
|
|
old = self._all("SELECT * FROM deployments WHERE app = ? AND status IN ('success', 'failed', 'cancelled') "
|
|
"ORDER BY id DESC LIMIT -1 OFFSET ?", (app, DEPLOY_KEEP))
|
|
for d in old:
|
|
for path in (deploy_log_path(app, d["id"]), deploy_snapshot_path(app, d["id"])):
|
|
try:
|
|
os.remove(path)
|
|
except FileNotFoundError:
|
|
pass
|
|
if old:
|
|
with self.lock:
|
|
self.db.execute(f"DELETE FROM deployments WHERE id IN ({','.join('?' * len(old))})", [d["id"] for d in old])
|
|
|
|
|
|
class DeployCancelled(Exception):
|
|
pass
|
|
|
|
|
|
class DeployFailed(Exception):
|
|
pass
|
|
|
|
|
|
class Job:
|
|
def __init__(self, dep_id, app, kind, params):
|
|
self.id = dep_id
|
|
self.app = app
|
|
self.kind = kind
|
|
self.params = params
|
|
self.cancelled = False
|
|
self.timed_out = False
|
|
self.proc = None
|
|
self.log = None
|
|
self.tail = collections.deque(maxlen=60)
|
|
|
|
def write(self, text):
|
|
self.log.write(text)
|
|
self.log.flush()
|
|
|
|
def step(self, text):
|
|
self.write(f"==> {text}\n")
|
|
|
|
def run(self, cmd, env=None, cwd=None, timeout=DEPLOY_TIMEOUT, redact=False):
|
|
"""Run a command, streaming its output into the deployment log. Returns the exit code."""
|
|
if self.cancelled:
|
|
raise DeployCancelled()
|
|
self.tail.clear()
|
|
self.timed_out = False
|
|
proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, stdin=subprocess.DEVNULL,
|
|
env=env, cwd=cwd, start_new_session=True)
|
|
self.proc = proc
|
|
timer = threading.Timer(timeout, self._timeout, args=(proc,))
|
|
timer.daemon = True
|
|
timer.start()
|
|
try:
|
|
for raw in iter(proc.stdout.readline, b""):
|
|
line = clean_line(raw.decode("utf-8", "replace"))
|
|
if redact:
|
|
line = redact_credentials(line)
|
|
self.write(line)
|
|
if line.strip():
|
|
self.tail.append(line.strip())
|
|
code = proc.wait()
|
|
finally:
|
|
timer.cancel()
|
|
proc.stdout.close()
|
|
self.proc = None
|
|
if self.cancelled:
|
|
raise DeployCancelled()
|
|
if self.timed_out:
|
|
raise DeployFailed(f"timed out after {timeout}s")
|
|
return code
|
|
|
|
def _timeout(self, proc):
|
|
self.timed_out = True
|
|
self._kill(proc)
|
|
|
|
@staticmethod
|
|
def _kill(proc):
|
|
try:
|
|
os.killpg(proc.pid, signal.SIGTERM)
|
|
except OSError:
|
|
return
|
|
|
|
def escalate():
|
|
time.sleep(10)
|
|
if proc.poll() is None:
|
|
try:
|
|
os.killpg(proc.pid, signal.SIGKILL)
|
|
except OSError:
|
|
pass
|
|
threading.Thread(target=escalate, daemon=True).start()
|
|
|
|
def cancel(self):
|
|
self.cancelled = True
|
|
proc = self.proc
|
|
if proc:
|
|
self._kill(proc)
|
|
|
|
def error_line(self, fallback):
|
|
"""The most telling error line of the last command."""
|
|
generic = ("compose up failed", "compose down failed")
|
|
for line in reversed(self.tail):
|
|
m = re.match(r"^(?:error|fatal)\b:?\s*(.*)$", line, re.I)
|
|
if m and not any(g in line for g in generic):
|
|
return m.group(1) or line
|
|
return fallback
|
|
|
|
|
|
class DeployRunner:
|
|
def __init__(self, store):
|
|
self.store = store
|
|
self.lock = threading.Lock()
|
|
self.pending = {} # app -> Job waiting to run (a newer request replaces it)
|
|
self.running = {} # app -> Job
|
|
self.workers = set()
|
|
|
|
def submit(self, app, kind, trigger, title, **params):
|
|
dep = self.store.create(app, kind, trigger, title, target=params.get("commit") or params.get("snapshot_from"))
|
|
job = Job(dep["id"], app, kind, params)
|
|
with self.lock:
|
|
replaced = self.pending.get(app)
|
|
self.pending[app] = job
|
|
start = app not in self.workers
|
|
self.workers.add(app)
|
|
if replaced:
|
|
self.store.update(replaced.id, status="cancelled", finished=time.time(), error=f"superseded by #{dep['id']}")
|
|
if start:
|
|
threading.Thread(target=self._worker, args=(app,), daemon=True, name=f"deploy-{app}").start()
|
|
return dep
|
|
|
|
def queued(self, app):
|
|
with self.lock:
|
|
return app in self.pending
|
|
|
|
def cancel(self, dep_id):
|
|
with self.lock:
|
|
for app, job in list(self.pending.items()):
|
|
if job.id == dep_id:
|
|
del self.pending[app]
|
|
self.store.update(dep_id, status="cancelled", finished=time.time(), error="cancelled before it started")
|
|
return True
|
|
for job in self.running.values():
|
|
if job.id == dep_id:
|
|
job.cancel()
|
|
return True
|
|
return False
|
|
|
|
def cancel_app(self, app):
|
|
with self.lock:
|
|
jobs = [j for j in (self.pending.get(app), self.running.get(app)) if j]
|
|
for job in jobs:
|
|
self.cancel(job.id)
|
|
|
|
def wait(self, dep_id, timeout=DEPLOY_TIMEOUT + 60):
|
|
deadline = time.time() + timeout
|
|
while time.time() < deadline:
|
|
dep = self.store.get(dep_id)
|
|
if dep is None or dep["status"] in FINAL_STATES:
|
|
return dep
|
|
time.sleep(0.3)
|
|
return self.store.get(dep_id)
|
|
|
|
def _worker(self, app):
|
|
while True:
|
|
with self.lock:
|
|
job = self.pending.get(app)
|
|
if job is None:
|
|
self.workers.discard(app)
|
|
return
|
|
# Wait for the app's other operations (backup, restore, …) to finish.
|
|
if not try_acquire_app(app, JOB_BUSY.get(job.kind, "deploy")):
|
|
time.sleep(0.5)
|
|
continue
|
|
with self.lock:
|
|
if self.pending.get(app) is not job: # cancelled or superseded meanwhile
|
|
release_app(app)
|
|
continue
|
|
del self.pending[app]
|
|
self.running[app] = job
|
|
try:
|
|
self._run(job)
|
|
finally:
|
|
with self.lock:
|
|
self.running.pop(app, None)
|
|
release_app(app)
|
|
|
|
def _run(self, job):
|
|
os.makedirs(deploy_dir(job.app), mode=0o750, exist_ok=True)
|
|
started = time.time()
|
|
self.store.update(job.id, status="running", started=started)
|
|
status, error = "success", None
|
|
with open(deploy_log_path(job.app, job.id), "w", encoding="utf-8") as log:
|
|
job.log = log
|
|
try:
|
|
JOB_KINDS[job.kind](job, self.store)
|
|
except DeployCancelled:
|
|
status, error = "cancelled", "cancelled"
|
|
except DeployFailed as exc:
|
|
status, error = "failed", str(exc)
|
|
except Exception as exc: # keep the worker alive, but show what happened
|
|
status, error = "failed", f"internal error: {exc}"
|
|
job.write(traceback.format_exc())
|
|
summary = {"success": "Deployment finished", "failed": f"Deployment failed: {error}",
|
|
"cancelled": "Deployment cancelled"}[status]
|
|
job.write(f"\n==> {summary} after {time.time() - started:.1f}s\n")
|
|
# The log is complete before the status turns final (log streams rely on it).
|
|
self.store.update(job.id, status=status, error=error, finished=time.time())
|
|
self.store.prune(job.app)
|
|
|
|
|
|
def job_app_info(job):
|
|
app, err = read_app_info(job.app)
|
|
if err is not None or app is None:
|
|
raise DeployFailed((err or {}).get("error") or (err or {}).get("stderr") or "app not found")
|
|
return app
|
|
|
|
|
|
def job_record_commit(job, store, repo_dir):
|
|
commit = repo_commit(repo_dir)
|
|
if commit:
|
|
store.update(job.id, commit_sha=commit["sha"], commit_subject=commit["subject"], commit_author=commit["author"])
|
|
job.write(f"Commit {commit['short']}: {commit['subject']} ({commit['author']})\n")
|
|
return commit
|
|
|
|
|
|
def job_compose_up(job):
|
|
job.step("Starting containers (podman compose up)")
|
|
if job.run([PANELCTL, "deploy", job.app]) != 0:
|
|
raise DeployFailed(job.error_line("compose up failed"))
|
|
|
|
|
|
def job_deploy(job, store):
|
|
app = job_app_info(job)
|
|
repo_dir = os.path.join(app["APP_STACK_DIR"], "repo")
|
|
snapshot_from = job.params.get("snapshot_from")
|
|
if snapshot_from:
|
|
src = deploy_snapshot_path(job.app, snapshot_from)
|
|
if not os.path.isfile(src):
|
|
raise DeployFailed(f"deployment #{snapshot_from} has no saved compose file")
|
|
job.step(f"Restoring the compose file of deployment #{snapshot_from}")
|
|
shutil.copyfile(src, app["APP_COMPOSE_FILE"])
|
|
if app.get("APP_REPO_URL") and os.path.isdir(os.path.join(repo_dir, ".git")):
|
|
job_record_commit(job, store, repo_dir)
|
|
elif os.path.isfile(app["APP_COMPOSE_FILE"]):
|
|
# Keep the compose file so this deployment can be redeployed later.
|
|
shutil.copyfile(app["APP_COMPOSE_FILE"], deploy_snapshot_path(job.app, job.id))
|
|
store.update(job.id, snapshot=1)
|
|
job_compose_up(job)
|
|
|
|
|
|
def job_sync(job, store):
|
|
app = job_app_info(job)
|
|
repo_url = app.get("APP_REPO_URL", "").strip()
|
|
if not repo_url:
|
|
raise DeployFailed("the app is not linked to a git repository")
|
|
repo_dir = os.path.join(app["APP_STACK_DIR"], "repo")
|
|
branch = app.get("APP_REPO_BRANCH", "").strip()
|
|
commit = job.params.get("commit")
|
|
git = shutil.which("git") or "git"
|
|
env = git_env()
|
|
|
|
if os.path.isdir(os.path.join(repo_dir, ".git")):
|
|
ref = branch or repo_current_branch(repo_dir)
|
|
if not ref:
|
|
raise DeployFailed("cannot determine which branch to sync")
|
|
job.step(f"Fetching {ref} from {redact_credentials(repo_url, '')}")
|
|
if job.run([git, "-C", repo_dir, "fetch", "origin", ref], env=env, timeout=GIT_TIMEOUT, redact=True) != 0:
|
|
raise DeployFailed("git fetch failed: " + git_error("\n".join(job.tail)))
|
|
else:
|
|
job.step(f"Cloning {redact_credentials(repo_url, '')}")
|
|
if os.path.exists(repo_dir):
|
|
shutil.rmtree(repo_dir)
|
|
ssh_public_key()
|
|
args = [git, "clone"] + (["--branch", branch] if branch else []) + ["--", repo_url, repo_dir]
|
|
if job.run(args, env=env, timeout=GIT_TIMEOUT, redact=True) != 0:
|
|
raise DeployFailed("git clone failed: " + git_error("\n".join(job.tail)))
|
|
|
|
target = commit or "FETCH_HEAD"
|
|
if commit or os.path.exists(os.path.join(repo_dir, ".git", "FETCH_HEAD")):
|
|
if commit and not re.fullmatch(r"[0-9a-f]{7,40}", commit):
|
|
raise DeployFailed("invalid commit")
|
|
job.step(f"Checking out {commit[:7] if commit else 'the fetched commit'}")
|
|
if job.run([git, "-C", repo_dir, "reset", "--hard", target], env=env, timeout=120) != 0:
|
|
raise DeployFailed("git reset failed: " + git_error("\n".join(job.tail)))
|
|
job_record_commit(job, store, repo_dir)
|
|
|
|
compose_path = find_compose_file(repo_dir)
|
|
if not compose_path:
|
|
raise DeployFailed("no compose file (compose.yaml or docker-compose.yml) in the repository root")
|
|
update_manifest(job.app, {"APP_COMPOSE_FILE": compose_path})
|
|
job_compose_up(job)
|
|
|
|
|
|
def job_restart(job, store):
|
|
app = job_app_info(job)
|
|
repo_dir = os.path.join(app["APP_STACK_DIR"], "repo")
|
|
if app.get("APP_REPO_URL") and os.path.isdir(os.path.join(repo_dir, ".git")):
|
|
job_record_commit(job, store, repo_dir)
|
|
job.step("Restarting (podman compose down, then up)")
|
|
if job.run([PANELCTL, "restart", job.app]) != 0:
|
|
raise DeployFailed(job.error_line("restart failed"))
|
|
|
|
|
|
JOB_KINDS = {"deploy": job_deploy, "sync": job_sync, "restart": job_restart}
|
|
|
|
deploys = None # DeployStore, set up in main()
|
|
runner = None # DeployRunner
|
|
sampler = None # StatsSampler
|
|
|
|
|
|
def deployment_result(dep):
|
|
"""Old synchronous response shape (for `wait: true` callers such as scripts)."""
|
|
text = ""
|
|
try:
|
|
with open(deploy_log_path(dep["app"], dep["id"]), "r", encoding="utf-8", errors="replace") as fh:
|
|
text = fh.read()[-65536:]
|
|
except OSError:
|
|
pass
|
|
ok = dep["status"] == "success"
|
|
return {"ok": ok, "deployment": dep, "stdout": text if ok else "", "stderr": "" if ok else text,
|
|
"error": None if ok else dep.get("error")}
|
|
|
|
|
|
# ── Webhooks (auto deploy) ──
|
|
# POST /hooks/<app> is reachable without Authelia (the NixOS module routes it
|
|
# past forward_auth) and authenticated with a per-app secret instead: a
|
|
# Forgejo/Gitea/GitHub HMAC signature, or the secret itself as a token.
|
|
|
|
def hook_secret_path(name):
|
|
return os.path.join(HOOK_DIR, f"{name}.secret")
|
|
|
|
|
|
def hook_last_path(name):
|
|
return os.path.join(HOOK_DIR, f"{name}.last.json")
|
|
|
|
|
|
def read_hook_secret(name):
|
|
try:
|
|
with open(hook_secret_path(name), "r", encoding="utf-8") as fh:
|
|
return fh.read().strip()
|
|
except OSError:
|
|
return ""
|
|
|
|
|
|
def ensure_hook_secret(name, regenerate=False):
|
|
secret = "" if regenerate else read_hook_secret(name)
|
|
if not secret:
|
|
secret = secrets.token_hex(24)
|
|
write_private_file(hook_secret_path(name), secret + "\n")
|
|
return secret
|
|
|
|
|
|
def read_hook_last(name):
|
|
try:
|
|
with open(hook_last_path(name), "r", encoding="utf-8") as fh:
|
|
return json.load(fh)
|
|
except (OSError, ValueError):
|
|
return None
|
|
|
|
|
|
def write_hook_last(name, info):
|
|
try:
|
|
write_private_file(hook_last_path(name), json.dumps(info))
|
|
except OSError:
|
|
pass
|
|
|
|
|
|
def verify_hook(secret, body, headers, token=""):
|
|
if not secret:
|
|
return False
|
|
expected = hmac.new(secret.encode(), body, hashlib.sha256).hexdigest()
|
|
for header in ("X-Forgejo-Signature", "X-Gitea-Signature", "X-Gogs-Signature"):
|
|
value = (headers.get(header) or "").strip().lower()
|
|
if value and hmac.compare_digest(value, expected):
|
|
return True
|
|
value = (headers.get("X-Hub-Signature-256") or "").strip().lower()
|
|
if value and hmac.compare_digest(value, "sha256=" + expected):
|
|
return True
|
|
token = headers.get("X-Gitlab-Token") or headers.get("X-Panel-Token") or token
|
|
return bool(token) and hmac.compare_digest(token.strip(), secret)
|
|
|
|
|
|
def forgejo_repo_path(url):
|
|
"""owner/repo of a repository on the configured Forgejo, else None."""
|
|
if not url or repo_provider(url) != "forgejo":
|
|
return None
|
|
_, path = repo_host_and_path(url)
|
|
return path if path and FORGEJO_REPO_RE.match(path) else None
|
|
|
|
|
|
def register_forgejo_hook(repo, url, secret, branch):
|
|
config = {"url": url, "content_type": "json", "secret": secret}
|
|
last = None
|
|
for hook_type in ("forgejo", "gitea"):
|
|
body = {"type": hook_type, "active": True, "events": ["push"], "config": config}
|
|
if branch:
|
|
body["branch_filter"] = branch
|
|
try:
|
|
return (forgejo_api(f"/api/v1/repos/{repo}/hooks", method="POST", body=body) or {}).get("id")
|
|
except ForgejoError as exc:
|
|
last = exc
|
|
if "HTTP 422" not in str(exc) and "HTTP 400" not in str(exc):
|
|
break
|
|
raise last
|
|
|
|
|
|
def delete_forgejo_hook(repo, hook_id):
|
|
try:
|
|
forgejo_api(f"/api/v1/repos/{repo}/hooks/{int(hook_id)}", method="DELETE")
|
|
except (ForgejoError, ValueError):
|
|
pass
|
|
|
|
|
|
# ── Container metrics ──
|
|
# A background thread samples `podman stats` for all containers (every few
|
|
# seconds while someone looks at the metrics, otherwise once a minute) and
|
|
# keeps an hour of CPU and memory history per container.
|
|
|
|
_SIZE_UNITS = {"b": 1, "kb": 1e3, "mb": 1e6, "gb": 1e9, "tb": 1e12,
|
|
"kib": 1024, "mib": 1024 ** 2, "gib": 1024 ** 3, "tib": 1024 ** 4}
|
|
|
|
|
|
def parse_size(text):
|
|
m = re.match(r"^\s*([\d.]+)\s*([kmgt]?i?b)?\s*$", str(text or ""), re.I)
|
|
if not m:
|
|
return None
|
|
return float(m.group(1)) * _SIZE_UNITS.get((m.group(2) or "b").lower(), 1)
|
|
|
|
|
|
def parse_size_pair(text):
|
|
parts = str(text or "").split("/")
|
|
if len(parts) != 2:
|
|
return None, None
|
|
return parse_size(parts[0]), parse_size(parts[1])
|
|
|
|
|
|
def parse_percent(value):
|
|
try:
|
|
return float(str(value).strip().rstrip("%"))
|
|
except ValueError:
|
|
return None
|
|
|
|
|
|
def pick(d, *keys):
|
|
for key in keys:
|
|
if d.get(key) not in (None, ""):
|
|
return d[key]
|
|
return None
|
|
|
|
|
|
class StatsSampler:
|
|
WINDOW = 3600
|
|
|
|
def __init__(self):
|
|
self.lock = threading.Lock()
|
|
self.series = {} # container -> deque of [t, cpu %, memory bytes]
|
|
self.current = {} # container -> latest sample
|
|
self.projects = {} # container -> app
|
|
self.error = None
|
|
self.last_view = 0.0
|
|
self.last_sample = 0.0
|
|
self.wake = threading.Event()
|
|
|
|
def start(self):
|
|
threading.Thread(target=self._loop, daemon=True, name="stats").start()
|
|
return self
|
|
|
|
def viewed(self):
|
|
self.last_view = time.time()
|
|
if time.time() - self.last_sample > 5:
|
|
self.wake.set()
|
|
|
|
def _loop(self):
|
|
while True:
|
|
try:
|
|
if load_app_summaries():
|
|
self.sample()
|
|
except Exception as exc: # never let the sampler die
|
|
self.error = f"metrics sampling failed: {exc}"
|
|
self.wake.wait(5 if time.time() - self.last_view < 120 else 60)
|
|
self.wake.clear()
|
|
|
|
def sample(self):
|
|
ps = run_panelctl(["containers"])
|
|
if not ps["ok"]:
|
|
self.error = last_line(ps["stderr"]) or "podman ps failed"
|
|
return
|
|
projects = {}
|
|
for c in _decode_containers(ps["stdout"]) or []:
|
|
labels = c.get("Labels") or {}
|
|
project = labels.get("com.docker.compose.project") or labels.get("io.podman.compose.project")
|
|
if project:
|
|
projects[_container_name(c)] = project
|
|
st = run_panelctl(["stats"])
|
|
if not st["ok"]:
|
|
with self.lock:
|
|
self.projects = projects
|
|
self.error = last_line(st["stderr"]) or "podman stats failed"
|
|
return
|
|
now = time.time()
|
|
with self.lock:
|
|
self.projects = projects
|
|
for s in _decode_containers(st["stdout"]) or []:
|
|
name = pick(s, "name", "Name") or _container_name(s)
|
|
if name not in projects:
|
|
continue
|
|
mem_used, mem_limit = parse_size_pair(pick(s, "mem_usage", "MemUsage"))
|
|
net_in, net_out = parse_size_pair(pick(s, "net_io", "NetIO"))
|
|
blk_in, blk_out = parse_size_pair(pick(s, "block_io", "BlockIO"))
|
|
pids = pick(s, "pids", "PIDs", "PIDS")
|
|
sample = {
|
|
"cpu": parse_percent(pick(s, "cpu_percent", "CPUPerc", "CPU")),
|
|
"mem": mem_used,
|
|
"mem_limit": mem_limit,
|
|
"mem_percent": parse_percent(pick(s, "mem_percent", "MemPerc")),
|
|
"net_in": net_in, "net_out": net_out,
|
|
"block_in": blk_in, "block_out": blk_out,
|
|
"pids": int(pids) if str(pids or "").isdigit() else None,
|
|
"time": now,
|
|
}
|
|
self.current[name] = sample
|
|
series = self.series.setdefault(name, collections.deque(maxlen=1000))
|
|
series.append([round(now), sample["cpu"], sample["mem"]])
|
|
while series and series[0][0] < now - self.WINDOW:
|
|
series.popleft()
|
|
for name in list(self.series):
|
|
if name not in projects:
|
|
self.series.pop(name, None)
|
|
self.current.pop(name, None)
|
|
self.error = None
|
|
self.last_sample = now
|
|
|
|
def for_app(self, app):
|
|
with self.lock:
|
|
containers = [{
|
|
"name": name,
|
|
"current": self.current.get(name),
|
|
"history": list(self.series.get(name, [])),
|
|
} for name, project in sorted(self.projects.items()) if project == app]
|
|
return {"containers": containers, "error": self.error, "sampled": self.last_sample or None}
|
|
|
|
|
|
# ── WebSocket (web terminal) ──
|
|
|
|
WS_GUID = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11"
|
|
|
|
|
|
class WebSocketClosed(Exception):
|
|
pass
|
|
|
|
|
|
class WebSocket:
|
|
"""Just enough of RFC 6455 for a terminal: reads from the handler's buffered
|
|
input, writes straight to the socket."""
|
|
|
|
def __init__(self, rfile, sock):
|
|
self.rfile = rfile
|
|
self.sock = sock
|
|
self.lock = threading.Lock()
|
|
self.closed = False
|
|
|
|
def _read(self, n):
|
|
data = self.rfile.read(n)
|
|
if data is None or len(data) < n:
|
|
raise WebSocketClosed()
|
|
return data
|
|
|
|
def recv(self):
|
|
"""Next complete message as (opcode, payload); control frames are returned as they come."""
|
|
message, message_op = b"", None
|
|
while True:
|
|
b1, b2 = self._read(2)
|
|
fin, opcode = b1 & 0x80, b1 & 0x0F
|
|
length = b2 & 0x7F
|
|
if length == 126:
|
|
length = struct.unpack("!H", self._read(2))[0]
|
|
elif length == 127:
|
|
length = struct.unpack("!Q", self._read(8))[0]
|
|
if length > 1 << 20:
|
|
raise WebSocketClosed()
|
|
mask = self._read(4) if b2 & 0x80 else None
|
|
payload = self._read(length) if length else b""
|
|
if mask:
|
|
full = (mask * (length // 4 + 1))[:length]
|
|
payload = (int.from_bytes(payload, "big") ^ int.from_bytes(full, "big")).to_bytes(length, "big") if length else b""
|
|
if opcode >= 0x8:
|
|
return opcode, payload
|
|
if opcode:
|
|
message_op = opcode
|
|
message += payload
|
|
if fin:
|
|
return message_op or 0x1, message
|
|
|
|
def send(self, opcode, payload=b""):
|
|
n = len(payload)
|
|
if n < 126:
|
|
header = struct.pack("!BB", 0x80 | opcode, n)
|
|
elif n < 1 << 16:
|
|
header = struct.pack("!BBH", 0x80 | opcode, 126, n)
|
|
else:
|
|
header = struct.pack("!BBQ", 0x80 | opcode, 127, n)
|
|
with self.lock:
|
|
if self.closed:
|
|
raise WebSocketClosed()
|
|
try:
|
|
self.sock.sendall(header + payload)
|
|
except OSError as exc:
|
|
self.closed = True
|
|
raise WebSocketClosed() from exc
|
|
|
|
def close(self, code=1000, reason=""):
|
|
try:
|
|
self.send(0x8, struct.pack("!H", code) + reason.encode()[:120])
|
|
except WebSocketClosed:
|
|
pass
|
|
self.closed = True
|
|
|
|
|
|
def reap_child(pid):
|
|
"""Hang up on a terminal's process, escalating until it has exited."""
|
|
for sig in (signal.SIGHUP, signal.SIGTERM, signal.SIGKILL):
|
|
try:
|
|
os.kill(pid, sig)
|
|
except ProcessLookupError:
|
|
pass
|
|
for _ in range(20):
|
|
try:
|
|
if os.waitpid(pid, os.WNOHANG) != (0, 0):
|
|
return
|
|
except ChildProcessError:
|
|
return
|
|
time.sleep(0.1)
|
|
|
|
|
|
def app_services(name):
|
|
"""Services of an app's compose file with the container ports it mentions."""
|
|
app, err = read_app_info(name)
|
|
if err is None and app:
|
|
try:
|
|
with open(app["APP_COMPOSE_FILE"], "r", encoding="utf-8") as fh:
|
|
parsed = compose_service_ports(fh.read())
|
|
if parsed is not None:
|
|
return parsed
|
|
except OSError:
|
|
pass
|
|
return [{"name": s, "ports": [], "image": ""} for s in compose_services(name) or []]
|
|
|
|
|
|
def autodeploy_info(name, public_base):
|
|
manifest = read_manifest(name) or {}
|
|
repo_url = manifest.get("APP_REPO_URL", "")
|
|
repo = forgejo_repo_path(repo_url)
|
|
enabled = manifest.get("APP_AUTODEPLOY", "false") == "true"
|
|
return {
|
|
"ok": True,
|
|
"name": name,
|
|
"enabled": enabled,
|
|
"url": f"{public_base}/hooks/{name}",
|
|
"secret": read_hook_secret(name) if enabled else "",
|
|
"git": bool(repo_url),
|
|
"branch": manifest.get("APP_REPO_BRANCH", ""),
|
|
"provider": repo_provider(repo_url) if repo_url else "",
|
|
"forgejo": {
|
|
"repo": repo,
|
|
"can_register": bool(repo and forgejo_token()),
|
|
"hook_id": manifest.get("APP_HOOK_ID") or None,
|
|
"hooks_url": f"{FORGEJO_URL}/{repo}/settings/hooks" if repo else None,
|
|
},
|
|
"last": read_hook_last(name),
|
|
}
|
|
|
|
|
|
# Actions that only read state (or queue a background job) and may run
|
|
# alongside anything else.
|
|
LOCK_FREE_ACTIONS = {"validate-compose", "deploy", "restart", "repo-pull", "autodeploy"}
|
|
|
|
|
|
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, cache=False):
|
|
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; vendored
|
|
# libraries only change with the package.
|
|
self.send_header("Cache-Control", "max-age=86400" if cache else "no-cache")
|
|
self.end_headers()
|
|
self.wfile.write(data)
|
|
except OSError:
|
|
self._json(500, {"ok": False, "error": "failed to read file"})
|
|
|
|
def _public_base(self):
|
|
"""Public origin of the panel, for URLs handed to other services."""
|
|
if PUBLIC_URL:
|
|
return PUBLIC_URL
|
|
host = (self.headers.get("X-Forwarded-Host") or self.headers.get("Host") or f"{BIND}:{PORT}").split(",")[0].strip()
|
|
proto = (self.headers.get("X-Forwarded-Proto") or "http").split(",")[0].strip()
|
|
return f"{proto}://{host}"
|
|
|
|
# ── Server-sent events ──
|
|
|
|
def _sse_start(self):
|
|
self.send_response(200)
|
|
self.send_header("Content-Type", "text/event-stream")
|
|
self.send_header("Cache-Control", "no-cache")
|
|
self.send_header("X-Accel-Buffering", "no")
|
|
self.end_headers()
|
|
self.close_connection = True
|
|
|
|
def _sse(self, event, data):
|
|
self.wfile.write(f"event: {event}\ndata: {json.dumps(data)}\n\n".encode("utf-8"))
|
|
|
|
def _client_gone(self, timeout=0):
|
|
"""True once the browser closed the connection (it sends nothing else on an SSE stream)."""
|
|
try:
|
|
ready, _, _ = select.select([self.connection], [], [], timeout)
|
|
# Anything readable is either EOF or bytes nobody asked for (dropped).
|
|
return bool(ready) and not self.connection.recv(4096)
|
|
except OSError:
|
|
return True
|
|
|
|
def _deployment_log(self, dep):
|
|
try:
|
|
with open(deploy_log_path(dep["app"], dep["id"]), "r", encoding="utf-8", errors="replace") as fh:
|
|
text = fh.read()
|
|
except OSError:
|
|
text = ""
|
|
limit = 2 * 1024 * 1024
|
|
self._json(200, {"ok": True, "deployment": dep, "log": text[-limit:], "truncated": len(text) > limit})
|
|
|
|
def _stream_deployment(self, dep, query):
|
|
"""Send the deployment log as it grows, then a final 'done' event."""
|
|
path = deploy_log_path(dep["app"], dep["id"])
|
|
try:
|
|
offset = max(0, int(query.get("offset", ["0"])[0]))
|
|
except ValueError:
|
|
offset = 0
|
|
chunk_max = 256 * 1024
|
|
self._sse_start()
|
|
last_status, last_write = None, time.time()
|
|
try:
|
|
while True:
|
|
# Check the status before reading: once it is final the log is complete.
|
|
cur = deploys.get(dep["id"])
|
|
final = cur is None or cur["status"] in FINAL_STATES
|
|
if cur and cur["status"] != last_status:
|
|
last_status = cur["status"]
|
|
self._sse("status", {"deployment": cur})
|
|
data = b""
|
|
try:
|
|
with open(path, "rb") as fh:
|
|
fh.seek(offset)
|
|
data = fh.read(chunk_max)
|
|
except FileNotFoundError:
|
|
pass
|
|
if data and not final and len(data) < chunk_max:
|
|
data = data[:data.rfind(b"\n") + 1] # only complete lines while it runs
|
|
if data:
|
|
offset += len(data)
|
|
self._sse("log", {"text": data.decode("utf-8", "replace"), "offset": offset})
|
|
last_write = time.time()
|
|
continue
|
|
if final:
|
|
self._sse("done", {"deployment": cur})
|
|
return
|
|
if time.time() - last_write > 15:
|
|
self.wfile.write(b": ping\n\n")
|
|
last_write = time.time()
|
|
if self._client_gone(0.3):
|
|
return
|
|
except (BrokenPipeError, ConnectionResetError):
|
|
return
|
|
|
|
def _stream_container_logs(self, name, tail, service):
|
|
"""Follow `compose logs` and forward new lines as they arrive."""
|
|
args = [PANELCTL, "logs", name, "--tail", tail, "--follow"] + (["--service", service] if service else [])
|
|
proc = subprocess.Popen(args, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, stdin=subprocess.DEVNULL,
|
|
start_new_session=True)
|
|
fd = proc.stdout.fileno()
|
|
buf = b""
|
|
try:
|
|
self._sse_start()
|
|
while True:
|
|
ready, _, _ = select.select([fd, self.connection], [], [], 15)
|
|
if not ready:
|
|
self.wfile.write(b": ping\n\n")
|
|
continue
|
|
if self.connection in ready and self._client_gone():
|
|
return
|
|
if fd not in ready:
|
|
continue
|
|
chunk = os.read(fd, 65536)
|
|
if chunk:
|
|
buf += chunk
|
|
# Give a burst of output a moment to arrive, then send it as one event.
|
|
while len(buf) < 262144 and select.select([fd], [], [], 0.05)[0]:
|
|
more = os.read(fd, 65536)
|
|
if not more:
|
|
break
|
|
buf += more
|
|
*lines, buf = buf.split(b"\n")
|
|
if not chunk and buf:
|
|
lines, buf = lines + [buf], b""
|
|
if lines:
|
|
self._sse("lines", {"lines": [clean_line(l.decode("utf-8", "replace")) for l in lines]})
|
|
if not chunk:
|
|
self._sse("end", {"code": proc.wait(timeout=10)})
|
|
return
|
|
except (BrokenPipeError, ConnectionResetError, subprocess.TimeoutExpired):
|
|
return
|
|
finally:
|
|
if proc.poll() is None:
|
|
try:
|
|
os.killpg(proc.pid, signal.SIGTERM)
|
|
except OSError:
|
|
pass
|
|
try:
|
|
proc.wait(timeout=5)
|
|
except subprocess.TimeoutExpired:
|
|
os.killpg(proc.pid, signal.SIGKILL)
|
|
proc.stdout.close()
|
|
|
|
# ── Web terminal ──
|
|
|
|
def _terminal(self, name, container):
|
|
if self.headers.get("Upgrade", "").lower() != "websocket" or not self.headers.get("Sec-WebSocket-Key"):
|
|
self._json(400, {"ok": False, "error": "expected a WebSocket upgrade"})
|
|
return
|
|
# Browsers send Origin with WebSocket requests; refuse other sites' pages.
|
|
origin = self.headers.get("Origin")
|
|
host = (self.headers.get("X-Forwarded-Host") or self.headers.get("Host") or "").split(",")[0].strip()
|
|
if origin and urlparse(origin).netloc != host:
|
|
self._json(403, {"ok": False, "error": "cross-origin terminal request refused"})
|
|
return
|
|
containers = [c["name"] for c in app_status(name).get("containers", [])]
|
|
if container not in containers:
|
|
self._json(404, {"ok": False, "error": f"container '{container}' is not part of '{name}'"})
|
|
return
|
|
|
|
accept = base64.b64encode(hashlib.sha1((self.headers["Sec-WebSocket-Key"].strip() + WS_GUID).encode()).digest())
|
|
self.wfile.write(b"HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\n"
|
|
b"Sec-WebSocket-Accept: " + accept + b"\r\n\r\n")
|
|
self.close_connection = True
|
|
ws = WebSocket(self.rfile, self.connection)
|
|
print(f"[panel-api] terminal opened: {name}/{container}")
|
|
|
|
pid, fd = pty.fork()
|
|
if pid == 0: # child: become `panelctl exec` on the new terminal
|
|
try:
|
|
os.environ["TERM"] = "xterm-256color"
|
|
os.execv(PANELCTL, [PANELCTL, "exec", name, container])
|
|
finally:
|
|
os._exit(127)
|
|
|
|
def set_size(cols, rows):
|
|
try:
|
|
fcntl.ioctl(fd, termios.TIOCSWINSZ, struct.pack("HHHH", rows, cols, 0, 0))
|
|
except OSError:
|
|
pass
|
|
|
|
def pump():
|
|
try:
|
|
while True:
|
|
data = os.read(fd, 65536)
|
|
if not data:
|
|
break
|
|
ws.send(0x2, data)
|
|
except (OSError, WebSocketClosed):
|
|
pass
|
|
ws.close(1000, "the shell exited")
|
|
try:
|
|
self.connection.shutdown(socket.SHUT_RDWR) # unblock the reader below
|
|
except OSError:
|
|
pass
|
|
|
|
set_size(100, 30)
|
|
reader = threading.Thread(target=pump, daemon=True)
|
|
reader.start()
|
|
try:
|
|
while True:
|
|
opcode, payload = ws.recv()
|
|
if opcode == 0x8:
|
|
break
|
|
if opcode == 0x9:
|
|
ws.send(0xA, payload)
|
|
continue
|
|
if opcode not in (0x1, 0x2):
|
|
continue
|
|
try:
|
|
msg = json.loads(payload.decode("utf-8"))
|
|
except ValueError:
|
|
continue
|
|
if msg.get("type") == "input" and isinstance(msg.get("data"), str):
|
|
os.write(fd, msg["data"].encode("utf-8"))
|
|
elif msg.get("type") == "resize":
|
|
try:
|
|
set_size(max(10, min(int(msg["cols"]), 500)), max(4, min(int(msg["rows"]), 200)))
|
|
except (KeyError, TypeError, ValueError):
|
|
pass
|
|
except (WebSocketClosed, OSError):
|
|
pass
|
|
finally:
|
|
reap_child(pid)
|
|
try:
|
|
os.close(fd)
|
|
except OSError:
|
|
pass
|
|
ws.close()
|
|
reader.join(timeout=2)
|
|
print(f"[panel-api] terminal closed: {name}/{container}")
|
|
|
|
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
|
|
|
|
def _refuse_request(self, parts):
|
|
"""Guard against requests that took the unauthenticated webhook route
|
|
in Caddy but address something else here.
|
|
|
|
Caddy matches `/hooks/*` on the cleaned path (`/apps/x/remove/../../../hooks/x`
|
|
becomes `/hooks/x`) but forwards the original one, so dot segments are
|
|
refused outright, and whatever came through the hooks route (tagged with
|
|
X-Panel-Hook by the NixOS module) may only be a webhook delivery."""
|
|
if any(unquote(p) in (".", "..") or "/" in unquote(p) for p in parts):
|
|
self._json(400, {"ok": False, "error": "invalid path"})
|
|
return True
|
|
if self.headers.get("X-Panel-Hook") and not (
|
|
self.command == "POST" and len(parts) == 2 and parts[0] == "hooks"):
|
|
self._json(403, {"ok": False, "error": "only webhook deliveries are allowed here"})
|
|
return True
|
|
return False
|
|
|
|
# ── GET ──
|
|
|
|
def do_GET(self):
|
|
path, parts, query = self._parse_path()
|
|
if self._refuse_request(parts):
|
|
return
|
|
|
|
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
|
|
|
|
# /vendor/<file> — third-party browser assets (xterm.js) bundled by the package
|
|
if len(parts) == 2 and parts[0] == "vendor":
|
|
fname = parts[1]
|
|
types = {".js": "text/javascript; charset=utf-8", ".css": "text/css; charset=utf-8"}
|
|
ext = os.path.splitext(fname)[1]
|
|
fpath = os.path.join(FRONTEND_DIR, "vendor", fname)
|
|
if not re.fullmatch(r"[A-Za-z0-9._-]+", fname) or ext not in types or not os.path.isfile(fpath):
|
|
self._json(404, {"ok": False, "error": "not found"})
|
|
return
|
|
self._file(200, fpath, types[ext], cache=True)
|
|
return
|
|
|
|
# /deployments/<id>[/log|/stream]
|
|
if parts and parts[0] == "deployments" and len(parts) in (2, 3):
|
|
dep = deploys.get(int(parts[1])) if parts[1].isdigit() else None
|
|
if dep is None:
|
|
self._json(404, {"ok": False, "error": "deployment not found"})
|
|
return
|
|
if len(parts) == 2:
|
|
self._json(200, {"ok": True, "deployment": dep})
|
|
elif parts[2] == "log":
|
|
self._deployment_log(dep)
|
|
elif parts[2] == "stream":
|
|
self._stream_deployment(dep, query)
|
|
else:
|
|
self._json(404, {"ok": False, "error": "not found"})
|
|
return
|
|
|
|
if len(parts) >= 3 and parts[0] == "apps" and (
|
|
parts[2] in ("deployments", "services", "stats", "autodeploy", "terminal")
|
|
or parts[2:4] == ["logs", "stream"]):
|
|
name = parts[1]
|
|
if read_manifest(name) is None:
|
|
self._json(404, {"ok": False, "error": f"app '{name}' does not exist"})
|
|
return
|
|
|
|
# /apps/<name>/deployments?limit=N — deployment history, newest first
|
|
if parts[2] == "deployments" and len(parts) == 3:
|
|
try:
|
|
limit = max(1, min(int(query.get("limit", ["30"])[0]), DEPLOY_KEEP))
|
|
except ValueError:
|
|
limit = 30
|
|
self._json(200, {"ok": True, "name": name, "deployments": deploys.list(name, limit),
|
|
"queued": runner.queued(name)})
|
|
return
|
|
|
|
# /apps/<name>/logs/stream?tail=N&service=S — follow container logs (SSE)
|
|
if parts[2:4] == ["logs", "stream"] and len(parts) == 4:
|
|
tail = query.get("tail", ["300"])[0]
|
|
service = query.get("service", [""])[0]
|
|
if not tail.isdigit() or (service and not re.fullmatch(r"[A-Za-z0-9._-]+", service)):
|
|
self._json(400, {"ok": False, "error": "invalid tail or service"})
|
|
return
|
|
self._stream_container_logs(name, tail, service)
|
|
return
|
|
|
|
# /apps/<name>/services — compose services and the container ports they mention
|
|
if parts[2] == "services" and len(parts) == 3:
|
|
self._json(200, {"ok": True, "name": name, "services": app_services(name)})
|
|
return
|
|
|
|
# /apps/<name>/stats — CPU / memory now and over the last hour, per container
|
|
if parts[2] == "stats" and len(parts) == 3:
|
|
sampler.viewed()
|
|
self._json(200, {"ok": True, "name": name, **sampler.for_app(name)})
|
|
return
|
|
|
|
# /apps/<name>/autodeploy — webhook URL, secret and the last delivery
|
|
if parts[2] == "autodeploy" and len(parts) == 3:
|
|
self._json(200, autodeploy_info(name, self._public_base()))
|
|
return
|
|
|
|
# /apps/<name>/terminal?container=C — WebSocket shell in a container
|
|
if parts[2] == "terminal" and len(parts) == 3:
|
|
self._terminal(name, query.get("container", [""])[0])
|
|
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()
|
|
latest = deploys.latest()
|
|
for app in apps:
|
|
app["status"] = statuses.get(app["name"], {"state": "unknown"})
|
|
app["busy"] = busy.get(app["name"])
|
|
app["last_deployment"] = latest.get(app["name"])
|
|
app["queued"] = runner.queued(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 self._refuse_request(parts):
|
|
return
|
|
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 self._refuse_request(parts):
|
|
return
|
|
|
|
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()
|
|
if self._refuse_request(parts):
|
|
return
|
|
|
|
# POST /hooks/<name> — push webhook (reached without Authelia, signed instead)
|
|
if len(parts) == 2 and parts[0] == "hooks":
|
|
self._webhook(parts[1], query)
|
|
return
|
|
|
|
# POST /deployments/<id>/cancel | /deployments/<id>/redeploy
|
|
if len(parts) == 3 and parts[0] == "deployments" and parts[1].isdigit():
|
|
self._deployment_action(int(parts[1]), parts[2])
|
|
return
|
|
|
|
# POST /compose/inspect {"content": "..."} — services and ports of a compose file
|
|
if path == "/compose/inspect":
|
|
try:
|
|
content = str(self._read_json().get("content", ""))
|
|
except Exception as exc:
|
|
self._json(400, {"ok": False, "error": f"invalid payload: {exc}"})
|
|
return
|
|
services = compose_service_ports(content)
|
|
self._json(200, {"ok": True, "services": services or [], "parsed": services is not None})
|
|
return
|
|
|
|
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 = str(payload["name"])
|
|
auth = str(payload.get("auth", True)).lower()
|
|
source_type = payload.get("source_type", "default")
|
|
|
|
# Routes: [{domain, target | upstream, path}]
|
|
if "routes" in payload and isinstance(payload["routes"], list):
|
|
raw_routes = [r for r in payload["routes"] if isinstance(r, dict) and str(r.get("domain", "")).strip()]
|
|
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"])
|
|
raw_routes = [{"domain": d.strip(), "upstream": f"127.0.0.1:{payload['port']}"}
|
|
for d in domain_str.split(",") if d.strip()]
|
|
else:
|
|
self._json(400, {"ok": False, "error": "missing 'routes' array or 'domain'+'port' fields"})
|
|
return
|
|
except Exception as exc:
|
|
self._json(400, {"ok": False, "error": f"invalid payload: {exc}"})
|
|
return
|
|
|
|
if not is_safe_name(name):
|
|
self._json(400, {"ok": False, "error": f"invalid name '{name}' (use lowercase letters, digits and dashes)"})
|
|
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
|
|
try:
|
|
# The starter compose file has a single service called "app".
|
|
routes_str = resolve_routes(name, raw_routes, ["app"] if source_type == "default" else None)
|
|
except ValueError as exc:
|
|
self._json(400, {"ok": False, "error": str(exc)})
|
|
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 source_type in ("raw", "git"):
|
|
route_error = finalize_route_services(name)
|
|
if route_error:
|
|
run_panelctl(["remove", name])
|
|
self._json(400, {"ok": False, "error": route_error})
|
|
return
|
|
|
|
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"):
|
|
# Queued; it starts once this request releases the app.
|
|
result["deployment"] = runner.submit(name, "deploy", "manual", "Environment variables changed")
|
|
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
|
|
manifest = read_manifest(name)
|
|
if manifest is None:
|
|
self._json(404, {"ok": False, "error": f"app '{name}' does not exist"})
|
|
return
|
|
|
|
def published(routes):
|
|
return sorted((r.get("service", ""), r["port"], r["upstream"]) for r in routes if r.get("port"))
|
|
|
|
before = published(manifest_routes(manifest))
|
|
try:
|
|
routes_str = resolve_routes(name, route_list, compose_services(name))
|
|
except ValueError as exc:
|
|
self._json(400, {"ok": False, "error": str(exc)})
|
|
return
|
|
result = run_panelctl(["set-routes", name, routes_str])
|
|
after = manifest_routes(read_manifest(name) or {})
|
|
result["routes"] = after
|
|
# Newly published ports only exist once the containers are recreated.
|
|
result["needs_deploy"] = result["ok"] and published(after) != before
|
|
self._json(200 if result["ok"] else 400, result)
|
|
return
|
|
|
|
# POST /apps/<name>/deploy | restart | repo-pull — queue a deployment.
|
|
# Returns at once with the deployment; {"wait": true} blocks until it
|
|
# finishes and answers like the old synchronous API.
|
|
if action in {"deploy", "restart", "repo-pull"}:
|
|
if read_manifest(name) is None:
|
|
self._json(404, {"ok": False, "error": f"app '{name}' does not exist"})
|
|
return
|
|
try:
|
|
payload = self._read_json() or {}
|
|
except Exception:
|
|
payload = {}
|
|
if action == "repo-pull" and not (read_manifest(name) or {}).get("APP_REPO_URL"):
|
|
self._json(400, {"ok": False, "error": "app is not linked to a git repository"})
|
|
return
|
|
kind, title = {"deploy": ("deploy", "Deploy"), "restart": ("restart", "Restart"),
|
|
"repo-pull": ("sync", "Sync from git")}[action]
|
|
dep = runner.submit(name, kind, "manual", str(payload.get("title") or title)[:200])
|
|
if payload.get("wait"):
|
|
result = deployment_result(runner.wait(dep["id"]))
|
|
self._json(200 if result["ok"] else 400, result)
|
|
return
|
|
self._json(202, {"ok": True, "deployment": dep})
|
|
return
|
|
|
|
# POST /apps/<name>/autodeploy {"enabled", "register", "regenerate"}
|
|
if action == "autodeploy":
|
|
self._autodeploy(name)
|
|
return
|
|
|
|
# Simple panelctl pass-through actions
|
|
if action in {"stop", "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>/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")
|
|
manifest = read_manifest(name) or {}
|
|
runner.cancel_app(name)
|
|
result = run_panelctl(args)
|
|
if result["ok"]:
|
|
repo = forgejo_repo_path(manifest.get("APP_REPO_URL", ""))
|
|
if repo and manifest.get("APP_HOOK_ID"):
|
|
delete_forgejo_hook(repo, manifest["APP_HOOK_ID"])
|
|
for path in (hook_secret_path(name), hook_last_path(name)):
|
|
try:
|
|
os.remove(path)
|
|
except FileNotFoundError:
|
|
pass
|
|
deploys.delete_app(name)
|
|
self._json(200 if result["ok"] else 400, result)
|
|
return
|
|
|
|
self._json(404, {"ok": False, "error": "not found"})
|
|
|
|
# ── Deployments, webhooks and auto deploy ──
|
|
|
|
def _deployment_action(self, dep_id, action):
|
|
dep = deploys.get(dep_id)
|
|
if dep is None:
|
|
self._json(404, {"ok": False, "error": "deployment not found"})
|
|
return
|
|
if action == "cancel":
|
|
if dep["status"] in FINAL_STATES or not runner.cancel(dep_id):
|
|
self._json(409, {"ok": False, "error": f"deployment #{dep_id} is not running"})
|
|
return
|
|
self._json(200, {"ok": True, "deployment": deploys.get(dep_id)})
|
|
return
|
|
if action != "redeploy":
|
|
self._json(404, {"ok": False, "error": "not found"})
|
|
return
|
|
# Deploy what this deployment deployed: its commit for git apps, its saved
|
|
# compose file otherwise.
|
|
manifest = read_manifest(dep["app"])
|
|
if manifest is None:
|
|
self._json(404, {"ok": False, "error": f"app '{dep['app']}' does not exist"})
|
|
return
|
|
label = f"#{dep_id}"
|
|
if manifest.get("APP_REPO_URL") and dep.get("commit_sha"):
|
|
label = dep["commit_sha"][:7]
|
|
new = runner.submit(dep["app"], "sync", "rollback", f"Redeploy {label}", commit=dep["commit_sha"])
|
|
elif dep.get("snapshot") and os.path.isfile(deploy_snapshot_path(dep["app"], dep_id)):
|
|
new = runner.submit(dep["app"], "deploy", "rollback", f"Redeploy compose file of #{dep_id}", snapshot_from=dep_id)
|
|
else:
|
|
self._json(400, {"ok": False, "error": f"deployment #{dep_id} has no commit or saved compose file to go back to"})
|
|
return
|
|
self._json(202, {"ok": True, "deployment": new})
|
|
|
|
def _webhook(self, name, query):
|
|
try:
|
|
length = int(self.headers.get("Content-Length") or 0)
|
|
except ValueError:
|
|
length = -1
|
|
if length < 0 or length > WEBHOOK_MAX_BODY:
|
|
self._json(413, {"ok": False, "error": "payload too large"})
|
|
return
|
|
body = self.rfile.read(length) if length else b""
|
|
manifest = read_manifest(name)
|
|
if manifest is None or manifest.get("APP_AUTODEPLOY") != "true":
|
|
self._json(404, {"ok": False, "error": "auto deploy is not enabled for this app"})
|
|
return
|
|
if not verify_hook(read_hook_secret(name), body, self.headers, query.get("token", [""])[0]):
|
|
self._json(401, {"ok": False, "error": "invalid signature or token"})
|
|
return
|
|
|
|
event = (self.headers.get("X-Forgejo-Event") or self.headers.get("X-Gitea-Event")
|
|
or self.headers.get("X-GitHub-Event") or self.headers.get("X-Gitlab-Event") or "")
|
|
record = {"time": time.time(), "event": event or "manual"}
|
|
|
|
def done(code, result, **extra):
|
|
record.update(result=result, **extra)
|
|
write_hook_last(name, record)
|
|
self._json(code, {"ok": code < 300, "result": result, **extra})
|
|
|
|
if event.lower() == "ping":
|
|
done(200, "pong")
|
|
return
|
|
try:
|
|
payload = json.loads(body.decode("utf-8")) if body.strip() else {}
|
|
except ValueError:
|
|
payload = {}
|
|
payload = payload if isinstance(payload, dict) else {}
|
|
if event and event.lower() not in ("push", "push hook"):
|
|
done(200, f"ignored {event} event")
|
|
return
|
|
|
|
repo_url = manifest.get("APP_REPO_URL", "")
|
|
branch = manifest.get("APP_REPO_BRANCH", "")
|
|
ref = str(payload.get("ref") or "")
|
|
if repo_url and ref and branch and ref != f"refs/heads/{branch}":
|
|
done(200, f"ignored push to {ref.removeprefix('refs/heads/')} (deploying {branch})")
|
|
return
|
|
if payload.get("deleted"):
|
|
done(200, "ignored branch deletion")
|
|
return
|
|
|
|
head = payload.get("head_commit") or (payload.get("commits") or [{}])[-1] or {}
|
|
message = str(head.get("message") or "").strip().splitlines()
|
|
sha = str(payload.get("after") or head.get("id") or "")[:7]
|
|
pusher = payload.get("pusher") or payload.get("sender") or {}
|
|
who = pusher.get("login") or pusher.get("username") or pusher.get("name") or ""
|
|
if repo_url:
|
|
title = f"Push to {branch or 'the branch'}"
|
|
if sha:
|
|
title += f" · {sha}"
|
|
if message:
|
|
title += f": {message[0]}"
|
|
dep = runner.submit(name, "sync", "webhook", title[:200])
|
|
else:
|
|
dep = runner.submit(name, "deploy", "webhook", "Deploy (webhook)")
|
|
done(202, "deploying", deployment=dep["id"], commit=sha or None, pusher=who or None)
|
|
|
|
def _autodeploy(self, name):
|
|
manifest = read_manifest(name)
|
|
if manifest is None:
|
|
self._json(404, {"ok": False, "error": f"app '{name}' does not exist"})
|
|
return
|
|
try:
|
|
payload = self._read_json() or {}
|
|
except Exception as exc:
|
|
self._json(400, {"ok": False, "error": f"invalid payload: {exc}"})
|
|
return
|
|
enabled = payload.get("enabled", manifest.get("APP_AUTODEPLOY") == "true") is not False
|
|
register = payload.get("register", True) is not False
|
|
regenerate = bool(payload.get("regenerate"))
|
|
base = self._public_base()
|
|
repo = forgejo_repo_path(manifest.get("APP_REPO_URL", ""))
|
|
hook_id = manifest.get("APP_HOOK_ID", "")
|
|
warning = None
|
|
|
|
if enabled:
|
|
secret = ensure_hook_secret(name, regenerate=regenerate)
|
|
values = {"APP_AUTODEPLOY": "true"}
|
|
if repo and register and forgejo_token() and (regenerate or not hook_id):
|
|
if hook_id:
|
|
delete_forgejo_hook(repo, hook_id)
|
|
hook_id = ""
|
|
try:
|
|
hook_id = str(register_forgejo_hook(repo, f"{base}/hooks/{name}", secret,
|
|
manifest.get("APP_REPO_BRANCH", "")) or "")
|
|
except ForgejoError as exc:
|
|
warning = f"Couldn't add the webhook to {repo} on Forgejo ({exc}). Add it by hand with the URL and secret below."
|
|
values["APP_HOOK_ID"] = hook_id
|
|
elif repo and register and not forgejo_token() and not hook_id:
|
|
warning = "Connect a Forgejo token in Settings to add the webhook automatically, or add it by hand."
|
|
update_manifest(name, values)
|
|
else:
|
|
if repo and hook_id:
|
|
delete_forgejo_hook(repo, hook_id)
|
|
update_manifest(name, {"APP_AUTODEPLOY": "false", "APP_HOOK_ID": ""})
|
|
|
|
info = autodeploy_info(name, base)
|
|
if warning:
|
|
info["warning"] = warning
|
|
self._json(200, info)
|
|
|
|
|
|
def main():
|
|
global deploys, runner, sampler
|
|
deploys = DeployStore(DB_PATH)
|
|
runner = DeployRunner(deploys)
|
|
sampler = StatsSampler().start()
|
|
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()
|