Files
pipu-music/deploy/server-agent.py
T
mrleo1nid 24707a0bfc
ci / docker / docker (образы собираются — их сервер собирает сам при обновлении) (push) Successful in 2m15s
ci / ts / ts (packages + web) (push) Successful in 2m19s
fix(admin): mask all stored configuration secrets and credential URLs
2026-10-23 00:27:20 +03:00

499 lines
24 KiB
Python

#!/usr/bin/env python3
"""Host-side management worker. No network listener or Docker socket in the API.
Only fixed operations cross the shared mailbox. Private state stays outside it.
Run with Python >= 3.10; installed by install-management.sh.
"""
import argparse
import hashlib
import json
import os
import re
import subprocess
import tempfile
import threading
import time
import uuid
from datetime import datetime, timezone
from pathlib import Path
from urllib.parse import urlsplit
class ManagementError(Exception):
def __init__(self, code, status=400):
super().__init__(code)
self.code, self.status = code, status
def atomic(path, value, mode=0o600, preserve_owner=False):
fd, name = tempfile.mkstemp(dir=path.parent, prefix=".management-")
try:
with os.fdopen(fd, "w", encoding="utf-8") as out:
os.chmod(name, mode)
if preserve_owner and path.exists() and hasattr(os, "chown"):
previous = path.stat()
os.chown(name, previous.st_uid, previous.st_gid)
out.write(value)
out.flush()
os.fsync(out.fileno())
os.replace(name, path)
finally:
if os.path.exists(name):
os.unlink(name)
def read_env(path):
"""Read literal dotenv values, including the quoted format we write.
Compose performs interpolation; this reader never executes dotenv contents.
"""
values = {}
if not path.exists():
return values
for line in path.read_text(encoding="utf-8").splitlines():
match = re.match(r"^(?:export\s+)?([A-Z][A-Z0-9_]*)\s*=(.*)$", line.strip())
if not match:
continue
key, value = match.groups()
value = value.strip()
if value.startswith("'") and value.endswith("'"):
value = value[1:-1].replace("\\'", "'")
elif value.startswith('"') and value.endswith('"'):
try:
value = json.loads(value)
except ValueError:
value = value[1:-1]
else:
value = value.split(" #", 1)[0].strip()
values[key] = value
return values
def validate_url(url):
if not isinstance(url, str) or len(url) > 2048 or re.search(r"[\s\x00-\x1f]", url):
raise ManagementError("management.source_invalid")
if re.fullmatch(r"git@[A-Za-z0-9.-]+:[A-Za-z0-9_./-]+", url):
return
parsed = urlsplit(url)
if (parsed.scheme not in ("https", "http", "ssh") or not parsed.hostname
or parsed.password or parsed.query or parsed.fragment
or (parsed.username and not (parsed.scheme == "ssh" and parsed.username == "git"))
or not parsed.path.strip("/")):
raise ManagementError("management.source_invalid")
def version_key(tag):
match = re.fullmatch(r"v?(\d+)\.(\d+)\.(\d+)(?:-rc\.(\d+))?", tag)
if not match:
return None
major, minor, patch, rc = match.groups()
return int(major), int(minor), int(patch), 1 if rc is None else 0, int(rc or 0)
def now():
return datetime.now(timezone.utc).isoformat()
def sensitive_url(key, value):
"""URL credentials also occur in userinfo, query strings and fragments."""
if not key.endswith("_URL") or not value:
return False
try:
parsed = urlsplit(value)
return bool(parsed.username or parsed.password or parsed.query or parsed.fragment)
except ValueError:
# A malformed URL must not bypass masking.
return True
class Agent:
def __init__(self, deploy, mailbox, private):
self.deploy, self.mailbox, self.private = deploy, mailbox, private
self.lock = threading.Lock()
self.cache = {}
self.job = self.load("job.json", None)
# systemd KillMode=control-group stops descendants too. Never claim success
# after a reboot or interrupted deployment.
if self.job and self.job["state"] == "running":
self.job.update(state="failed", finished_at=now(), log="Agent interrupted; inspect deployment before retrying.")
self.save("job.json", self.job)
def load(self, name, default):
path = self.private / name
return json.loads(path.read_text(encoding="utf-8")) if path.exists() else default
def save(self, name, value):
atomic(self.private / name, json.dumps(value, ensure_ascii=False))
def git_environment(self):
return {"GIT_TERMINAL_PROMPT": "0", "GIT_SSH_COMMAND": "ssh -oBatchMode=yes",
"GIT_CONFIG_COUNT": "1", "GIT_CONFIG_KEY_0": "safe.directory", "GIT_CONFIG_VALUE_0": str(self.deploy.parent)}
def command(self, args, timeout=30, **kwargs):
return subprocess.run(args, cwd=self.deploy, timeout=timeout, check=True,
stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True,
env={**os.environ, **self.git_environment()}, **kwargs).stdout
def sources(self):
sources = self.load("sources.json", None)
if sources is None:
url = self.command(["git", "remote", "get-url", "origin"]).strip()
validate_url(url)
sources = [{"id": "origin", "name": "origin", "url": url}]
self.save("sources.json", sources)
return sources
def refs(self, url, fresh=False):
cached = self.cache.get(url)
if cached and not fresh and time.monotonic() - cached[0] < 60:
return cached[1]
output = self.command(["git", "-c", "protocol.file.allow=never", "-c", "protocol.ext.allow=never",
"ls-remote", "--heads", "--tags", "--", url], timeout=20)
refs = dict(line.split()[::-1] for line in output.splitlines() if len(line.split()) == 2)
self.cache[url] = time.monotonic(), refs
return refs
def schema(self):
return json.loads((self.deploy / "config.schema.json").read_text(encoding="utf-8"))
def revision(self):
return hashlib.sha256((self.deploy / "config.local").read_bytes()).hexdigest()
def effective_revision(self, ignore_local=False):
env = read_env(self.deploy / ".env")
local = {} if ignore_local else read_env(self.deploy / "config.local")
effective = {item["key"]: env.get(item["key"], local.get(item["key"], item["default_value"]))
for item in self.schema() if not item["readonly"]}
return hashlib.sha256(json.dumps(effective, sort_keys=True).encode()).hexdigest()
def config(self):
env = read_env(self.deploy / ".env")
local = read_env(self.deploy / "config.local")
entries = []
for item in self.schema():
key = item["key"]
source = "env" if key in env else "local" if key in local else "default"
effective = env.get(key, local.get(key, item["default_value"]))
secret = item["secret"] or any(sensitive_url(key, value) for value in
(env.get(key), local.get(key), item["default_value"]))
def display(value):
return "••••••••" if secret and value else value
entries.append({**item, "secret": secret, "default_value": display(item["default_value"]),
"env": display(env.get(key)), "local": display(local.get(key)),
"effective": display(effective), "source": source})
effective = self.effective_revision()
applied = self.load("applied-effective.json", None)
if applied is None and (not local or self.load("applied.json", None) == self.revision()):
self.save("applied-effective.json", effective)
applied = effective
# Changes shadowed by .env do not change the running configuration.
empty = self.effective_revision(ignore_local=True)
return {"entries": entries, "revision": self.revision(),
"restart_required": effective != (applied or empty)}
def installed_commit(self, body):
"""Trust the running API's build identity, never the checkout or .env.
The checkout can already point at a failed update. Legacy builds contain
git-describe/main hashes or a release tag, which we resolve locally.
"""
revision = body.get("_running_commit", "")
if re.fullmatch(r"[a-f0-9]{40,64}", revision):
return revision
version = body.get("_running_version", "")
match = re.fullmatch(r"(?:.*-g|main-)([a-f0-9]{7,64})", version)
candidates = [match[1]] if match else []
if version_key(version):
candidates = ["refs/tags/v" + version.removeprefix("v"), "refs/tags/" + version.removeprefix("v")]
for candidate in candidates:
try:
commit = self.command(["git", "rev-parse", "--verify", candidate + "^{commit}"]).strip()
if re.fullmatch(r"[a-f0-9]{40,64}", commit):
return commit
except (subprocess.SubprocessError, OSError):
pass
return ""
def update_target(self, refs, body):
channel = body.get("channel") or "latest"
if channel not in ("latest", "version", "main"):
raise ManagementError("management.target_invalid")
if channel == "latest":
tags = [r[10:] for r in refs if r.startswith("refs/tags/") and version_key(r[10:]) and "-rc." not in r]
if not tags:
return "", ""
target = max(tags, key=version_key)
else:
target = "main" if channel == "main" else body.get("version") or ""
if not target:
return "", ""
if target != "main" and not version_key(target):
raise ManagementError("management.target_invalid")
ref = "refs/heads/main" if target == "main" else "refs/tags/" + target
commit = refs.get(ref + "^{}", refs.get(ref, ""))
if not re.fullmatch(r"[a-f0-9]{40,64}", commit):
return target, ""
return target, commit
def update_state(self, target, commit, current, body):
if body.get("channel") == "version" and not body.get("version"):
return "select_version"
if not commit:
return "unavailable"
if not current:
return "unknown"
if current == commit:
return "current"
try:
self.command(["git", "merge-base", "--is-ancestor", commit, current])
return "ahead"
except (subprocess.SubprocessError, OSError):
# A selected source's commit may not have been fetched locally yet.
pass
# A stable tag behind a git-describe build is not a newer release.
running = body.get("_running_version", "")
base = re.match(r"v?(\d+\.\d+\.\d+)(?:-rc\.\d+)?", running)
target_key = version_key(target)
if target_key and base:
running_key = version_key(base[0])
if target_key[:3] < running_key[:3] or (target_key[:3] == running_key[:3] and re.search(r"-\d+-g[0-9a-f]+$", running)):
return "ahead"
return "available"
def save_config(self, body):
if self.job and self.job["state"] == "running":
raise ManagementError("management.busy", 409)
if body.get("revision") != self.revision():
raise ManagementError("management.config_conflict", 409)
values = body.get("values")
if not isinstance(values, dict) or len(values) > 200:
raise ManagementError("management.config_invalid")
schema = {v["key"]: v for v in self.schema()}
local = read_env(self.deploy / "config.local")
for key, value in values.items():
item = schema.get(key)
if not item or item["readonly"]:
raise ManagementError("management.config_readonly")
if value is None:
local.pop(key, None)
continue
if not isinstance(value, str) or len(value) > 8192 or any(c in value for c in "\r\n\x00"):
raise ManagementError("management.config_invalid")
if item["type"] == "boolean" and value not in ("true", "false", "1", "0"):
raise ManagementError("management.config_invalid")
if item["type"] == "integer" and not re.fullmatch(r"-?\d+", value):
raise ManagementError("management.config_invalid")
if key == "LOG_LEVEL" and value not in ("debug", "info", "warn", "error"):
raise ManagementError("management.config_invalid")
if item["type"] == "integer" and key != "TRAIN_HOUR":
minimum = -1 if key in ("PIPU_FILE_UID", "PIPU_FILE_GID") else 0
if int(value) < minimum:
raise ManagementError("management.config_invalid")
if key == "TRAIN_HOUR" and not -1 <= int(value) <= 23:
raise ManagementError("management.config_invalid")
if key.endswith("_PORT") and not 1 <= int(value) <= 65535:
raise ManagementError("management.config_invalid")
if item["type"] == "duration" and not re.fullmatch(r"(?:\d+(?:\.\d+)?(?:ns|us|µs|ms|s|m|h))+", value):
raise ManagementError("management.config_invalid")
local[key] = value
# Single quotes make dollar signs literal to Compose. Newlines are rejected.
content = "# Managed by Pipu. Environment and .env override these values.\n"
content += "".join(k + "='" + v.replace("'", "\\'") + "'\n" for k, v in sorted(local.items()))
path = self.deploy / "config.local"
previous = path.read_text(encoding="utf-8")
atomic(path, content, preserve_owner=True)
try:
self.command([*self.compose(), "config", "--quiet"])
except (subprocess.SubprocessError, OSError):
atomic(path, previous, preserve_owner=True)
raise ManagementError("management.config_invalid") from None
self.audit("config.save", list(values))
return self.config()
def compose(self):
return ["docker", "compose", "--env-file", "config.local", "--env-file", ".env"]
def audit(self, action, details):
# Deliberately no configuration values or credentials in this log.
with (self.private / "audit.jsonl").open("a", encoding="utf-8") as out:
out.write(json.dumps({"at": now(), "actor": getattr(self, "actor", ""), "action": action, "details": details}) + "\n")
def start_job(self, kind, target, command, extra_env=None):
if self.job and self.job["state"] == "running":
raise ManagementError("management.busy", 409)
job = {"id": str(uuid.uuid4()), "kind": kind, "state": "running", "target": target,
"started_at": now(), "finished_at": "", "log": ""}
self.job = job
self.save("job.json", job)
revision = self.revision()
self.audit(kind + ".start", {"id": job["id"], "target": target})
threading.Thread(target=self.run_job, args=(job, command, extra_env or {}, revision), daemon=True).start()
return dict(job)
def run_job(self, job, command, extra_env, revision):
effective_revision = self.effective_revision()
# Let the API deliver the accepted job before recreating its container.
time.sleep(2)
# Capture full output privately, expose only a bounded redacted tail.
secrets = [v for k, v in {**read_env(self.deploy / "config.local"), **read_env(self.deploy / ".env")}.items()
if re.search(r"TOKEN|PASSWORD|KEY|SECRET|DATABASE_URL|REDIS_URL", k) and len(v) >= 4]
def redact(text):
for value in secrets:
text = text.replace(value, "[redacted]")
return re.sub(r"(https?://)[^\s/@]+:[^\s/@]+@", r"\1[redacted]@", text)[-64000:]
logpath = self.private / (job["id"] + ".log")
result = 1
try:
with logpath.open("w", encoding="utf-8") as output:
process = subprocess.Popen(command, cwd=self.deploy, stdout=output, stderr=subprocess.STDOUT,
env={**os.environ, **self.git_environment(), **extra_env}, start_new_session=True)
deadline = time.monotonic() + 7200
while process.poll() is None:
if time.monotonic() > deadline:
import signal
os.killpg(process.pid, signal.SIGKILL)
process.wait()
break
with self.lock:
with logpath.open("rb") as output:
output.seek(max(0, logpath.stat().st_size - 128000))
job["log"] = redact(output.read().decode("utf-8", errors="replace"))
self.save("job.json", job)
time.sleep(1)
result = process.returncode
with logpath.open("rb") as output:
output.seek(max(0, logpath.stat().st_size - 128000))
job["log"] = redact(output.read().decode("utf-8", errors="replace"))
if result == 0:
self.save("applied.json", revision)
self.save("applied-effective.json", effective_revision)
except (OSError, subprocess.SubprocessError):
result = 1
job["log"] += "\nCould not execute deployment command. Inspect journalctl -u pipu-management."
finally:
with self.lock:
job.update(state="succeeded" if result == 0 else "failed", finished_at=now())
self.save("job.json", job)
def handle(self, op, body):
if op == "config.get":
return self.config()
if op == "config.save":
return self.save_config(body)
if op == "config.apply":
return self.start_job("config", "config.local", ["bash", str(self.deploy / "apply-config.sh")])
if op == "sources.save":
if self.job and self.job["state"] == "running":
raise ManagementError("management.busy", 409)
sources = body.get("sources")
if not isinstance(sources, list) or not 1 <= len(sources) <= 20:
raise ManagementError("management.source_invalid")
ids = set()
for source in sources:
if (not isinstance(source, dict) or not re.fullmatch(r"[A-Za-z0-9_-]{1,64}", source.get("id", ""))
or not isinstance(source.get("name"), str) or not 1 <= len(source["name"].strip()) <= 100
or source["id"] in ids):
raise ManagementError("management.source_invalid")
validate_url(source.get("url"))
ids.add(source["id"])
self.save("sources.json", sources)
self.audit("sources.save", list(ids))
return {"sources": sources}
sources = self.sources()
selected = next((s for s in sources if s["id"] == (body.get("source_id") or sources[0]["id"])), None)
if not selected:
raise ManagementError("management.source_invalid")
if op == "updates.get":
versions, error, refs = [], "", {}
try:
refs = self.refs(selected["url"], fresh=bool(body.get("fresh")))
versions = sorted([r.removeprefix("refs/tags/") for r in refs if r.startswith("refs/tags/") and version_key(r[10:])], key=version_key, reverse=True)
except (subprocess.SubprocessError, OSError):
error = "management.source_unavailable"
current = self.installed_commit(body)
target, commit = self.update_target(refs, body)
return {"sources": sources, "versions": versions, "source_error": error,
"current_commit": current, "target_version": target, "target_commit": commit,
"update_state": self.update_state(target, commit, current, body),
"current_version": read_env(self.deploy / ".env").get("PIPU_VERSION", ""), "job": self.job}
if op == "updates.start":
if self.job and self.job["state"] == "running":
raise ManagementError("management.busy", 409)
refs = self.refs(selected["url"], fresh=True)
target, commit = self.update_target(refs, body)
ref = "refs/heads/main" if target == "main" else "refs/tags/" + target
if not commit:
raise ManagementError("management.target_invalid")
state = self.update_state(target, commit, self.installed_commit(body), body)
if state != "available":
code = {"current": "management.already_installed", "ahead": "management.downgrade", "unknown": "management.version_unknown"}.get(state, "management.target_invalid")
raise ManagementError(code, 409)
if body.get("expected_commit") and body["expected_commit"] != commit:
raise ManagementError("management.target_changed", 409)
return self.start_job("update", target, ["bash", str(self.deploy / "update.sh"), target],
{"PIPU_UPDATE_SOURCE": selected["url"], "PIPU_UPDATE_COMMIT": commit,
"PIPU_UPDATE_REF": ref})
raise ManagementError("management.operation_invalid")
def tick(self):
atomic(self.mailbox / "heartbeat", str(time.time()), 0o644)
for path in sorted(self.mailbox.glob("*.request.json")):
if path.is_symlink() or not re.fullmatch(r"[a-f0-9-]{36}\.request\.json", path.name):
continue
response = self.mailbox / path.name.replace(".request.json", ".response.json")
if response.exists():
path.unlink(missing_ok=True)
continue
try:
if path.stat().st_size > 131072:
raise ManagementError("management.request_invalid")
request = json.loads(path.read_text(encoding="utf-8"))
if time.time() - request["created_at"] > 45:
raise ManagementError("management.request_expired", 409)
# Claim before executing: a worker crash must never replay a mutation.
path.unlink(missing_ok=True)
with self.lock:
self.actor = request.get("actor", "")
result = {"data": self.handle(request["operation"], request.get("body") or {})}
except ManagementError as error:
result = {"code": error.code, "status": error.status}
except (OSError, ValueError, KeyError, TypeError, subprocess.SubprocessError):
result = {"code": "management.agent_error", "status": 502}
atomic(response, json.dumps(result, ensure_ascii=False), 0o644)
path.unlink(missing_ok=True)
for path in self.mailbox.glob("*.response.json"):
if not path.is_symlink() and time.time() - path.stat().st_mtime > 120:
path.unlink(missing_ok=True)
def main():
import fcntl # Linux host only; tests import this module on Windows too.
parser = argparse.ArgumentParser()
parser.add_argument("--deploy-dir", type=Path, required=True)
parser.add_argument("--mailbox", type=Path, required=True)
parser.add_argument("--state-dir", type=Path, required=True)
args = parser.parse_args()
args.state_dir.mkdir(parents=True, exist_ok=True, mode=0o700)
with (args.state_dir / "lock").open("w") as lock:
fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
agent = Agent(args.deploy_dir.resolve(), args.mailbox.resolve(), args.state_dir.resolve())
while True:
# Refresh the installed worker only after a successful deployment has
# persisted its final status. The running worker survives checkout/build.
with agent.lock:
installed = Path(__file__).resolve()
candidate = agent.deploy / "server-agent.py"
if (agent.job and agent.job["kind"] == "update" and agent.job["state"] == "succeeded"
and installed == Path("/usr/local/lib/pipu/server-agent.py") and candidate.is_file()
and candidate.read_bytes() != installed.read_bytes()):
atomic(installed, candidate.read_text(encoding="utf-8"), 0o700)
# The singleton lock is non-inheritable and released by exec.
import sys
os.execv(sys.executable, [sys.executable, str(installed), *sys.argv[1:]])
agent.tick()
time.sleep(0.2)
if __name__ == "__main__":
main()