jj.wang
6dc7e72ac8
agentd webhook-serve (agentd/webhook.py, stdlib): verifies X-Plane-Signature
(HMAC-SHA256 hex of the body, constant-time; 401 on missing/bad), routes by
data.project to a configured tenant (others 202), coalesces deliveries within
3 s into one tick, runs one more tick when deliveries arrive during a tick,
and retries every 15 s for up to 2 min while the lock is held (tick command
exit 75: flock -E 75 in the cron scripts, cli tick on Busy). Ticks run in a
systemd scope so restarting the receiver never kills one. Log:
<log_dir>/webhook.log; GET /healthz.
Config: webhook section (key names only, PLANE_WEBHOOK_SECRET scrubbed from
child envs like the Plane tokens), notify: {kind: none} as a no-op hook
called on every state move; the unused per-project webhook_secret_key is
gone (Plane webhooks are workspace-level). systemd unit for the LXC and
tools/plane-webhook.py to create/rotate the Plane webhook as the workspace
admin (secret straight into the Vault, never printed).
239 lines
11 KiB
Python
239 lines
11 KiB
Python
"""Plane webhook receiver (PAD-7): a delivery for a tenant project triggers a dispatcher tick right away; the
|
|
2-minute cron tick stays as the fallback. Stdlib only.
|
|
|
|
POST <path> (default /plane):
|
|
- X-Plane-Signature = hex HMAC-SHA256 of the request body with the webhook secret (Plane signs
|
|
json.dumps(payload) and sends the same bytes); missing or wrong -> 401, nothing runs.
|
|
- body {"event": "issue"|"issue_comment", "action": ..., "data": {..., "project": <uuid>}, "activity": ...};
|
|
data.project not a configured tenant (or absent, e.g. on delete) -> 202, nothing runs.
|
|
- otherwise -> 200 and the Trigger schedules a tick.
|
|
GET /healthz -> 200 JSON (pending / running / last tick), no secrets.
|
|
|
|
Trigger: deliveries within `debounce_seconds` coalesce into one tick; deliveries that arrive while a tick runs cause
|
|
one more tick after it (the running tick may have read Plane before the change). The tick command exits 75 when the
|
|
cron lock (or agentd's own state lock) is held: then it is retried every `retry_seconds` for up to
|
|
`retry_window_seconds`; after that the next cron tick picks the change up. Every delivery and tick is logged to
|
|
<log_dir>/webhook.log."""
|
|
import hashlib
|
|
import hmac
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import threading
|
|
import time
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
|
|
from . import ROOT
|
|
|
|
BUSY = 75 # tick command exit code: lock held by another tick (flock -E 75, cli tick on Busy)
|
|
MAX_BODY = 2 << 20 # Plane payloads are a few KB
|
|
|
|
|
|
def sign(secret, body):
|
|
return hmac.new(secret.encode("utf-8"), body, hashlib.sha256).hexdigest()
|
|
|
|
|
|
def verify(secret, body, signature):
|
|
"""Constant-time check of the hex signature against the raw body; falls back to Plane's own serialisation
|
|
(json.dumps of the parsed payload) in case a proxy re-encoded the body."""
|
|
if not secret or not signature:
|
|
return False
|
|
sig = signature.strip().lower().encode("ascii", "replace")
|
|
if hmac.compare_digest(sign(secret, body).encode(), sig):
|
|
return True
|
|
try:
|
|
canonical = json.dumps(json.loads(body)).encode("utf-8")
|
|
except ValueError:
|
|
return False
|
|
return canonical != body and hmac.compare_digest(sign(secret, canonical).encode(), sig)
|
|
|
|
|
|
def project_of(payload):
|
|
data = payload.get("data") if isinstance(payload, dict) else None
|
|
pid = (data or {}).get("project") or (data or {}).get("project_id")
|
|
return pid.get("id") if isinstance(pid, dict) else pid
|
|
|
|
|
|
def route(payload, projects):
|
|
"""Configured project name for the delivery, or None."""
|
|
pid = project_of(payload)
|
|
return next((name for name, p in projects.items() if pid and p["id"] == pid), None)
|
|
|
|
|
|
class Logger:
|
|
def __init__(self, path):
|
|
self.path = path
|
|
self.lock = threading.Lock()
|
|
|
|
def __call__(self, msg):
|
|
line = time.strftime("%Y-%m-%dT%H:%M:%S") + " " + msg + "\n"
|
|
with self.lock:
|
|
if self.path:
|
|
os.makedirs(os.path.dirname(self.path), exist_ok=True)
|
|
with open(self.path, "a", encoding="utf-8") as fh:
|
|
fh.write(line)
|
|
else:
|
|
print(line, end="", flush=True)
|
|
|
|
|
|
class Trigger:
|
|
"""Debounced tick trigger. `run()` executes one tick and returns its exit code (BUSY = lock held)."""
|
|
|
|
def __init__(self, run, debounce=3.0, retry=15.0, retry_window=120.0, log=print, clock=time.monotonic):
|
|
self.run, self.debounce, self.retry, self.retry_window = run, debounce, retry, retry_window
|
|
self.log, self.clock = log, clock
|
|
self.cv = threading.Condition()
|
|
self.due = None # clock() when the next tick starts; None = nothing pending
|
|
self.reasons = [] # deliveries folded into the pending tick
|
|
self.busy_since = None # first BUSY of the current retry chain
|
|
self.running = False
|
|
self.last = {}
|
|
self.ticks = 0
|
|
|
|
def kick(self, reason):
|
|
with self.cv:
|
|
if self.due is None:
|
|
self.due = self.clock() + self.debounce
|
|
self.reasons.append(reason)
|
|
self.cv.notify()
|
|
|
|
def status(self):
|
|
with self.cv:
|
|
return {"pending": self.due is not None, "queued": len(self.reasons), "running": self.running,
|
|
"retrying_since": self.busy_since and round(self.clock() - self.busy_since, 1), "ticks": self.ticks,
|
|
"last": dict(self.last)}
|
|
|
|
def start(self):
|
|
threading.Thread(target=self.serve_forever, name="tick-trigger", daemon=True).start()
|
|
return self
|
|
|
|
def serve_forever(self):
|
|
while True:
|
|
self.step()
|
|
|
|
def step(self, wait=None):
|
|
"""Wait until a tick is due (or `wait` seconds pass), run it, handle the result. True if a tick ran."""
|
|
with self.cv:
|
|
end = None if wait is None else self.clock() + wait
|
|
while self.due is None or self.due > self.clock():
|
|
timeout = None if self.due is None else self.due - self.clock()
|
|
if end is not None:
|
|
left = end - self.clock()
|
|
if left <= 0:
|
|
return False
|
|
timeout = left if timeout is None else min(timeout, left)
|
|
self.cv.wait(timeout)
|
|
reasons, self.reasons, self.due, self.running = self.reasons, [], None, True
|
|
start = self.clock()
|
|
self.log("tick start (%d deliveries: %s)" % (len(reasons), ", ".join(reasons[:5]) + (" …" if len(reasons) > 5 else "")))
|
|
try:
|
|
rc = self.run()
|
|
except Exception as exc: # a broken runner must not kill the trigger thread
|
|
rc = "error %r" % (exc,)
|
|
took = round(self.clock() - start, 1)
|
|
with self.cv:
|
|
self.running = False
|
|
self.ticks += 1
|
|
self.last = {"rc": rc, "seconds": took, "deliveries": len(reasons), "at": time.strftime("%Y-%m-%dT%H:%M:%S")}
|
|
if rc != BUSY:
|
|
self.busy_since = None
|
|
self.log("tick done rc=%s in %.1fs" % (rc, took))
|
|
return True
|
|
now = self.clock()
|
|
self.busy_since = self.busy_since or start
|
|
if now - self.busy_since + self.retry > self.retry_window:
|
|
self.log("tick busy for %.0fs, giving up; the next cron tick covers it" % (now - self.busy_since))
|
|
self.busy_since = None
|
|
return True
|
|
self.reasons = reasons + self.reasons
|
|
self.due = min(self.due, now + self.retry) if self.due is not None else now + self.retry
|
|
self.log("tick busy (another tick holds the lock), retry in %gs" % self.retry)
|
|
return True
|
|
|
|
|
|
def command_runner(cmd, env, timeout=1000):
|
|
"""Run the tick command (it logs to cron.log itself); returns its exit code."""
|
|
def run():
|
|
try:
|
|
return subprocess.run(cmd, env=env, stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL, timeout=timeout).returncode
|
|
except subprocess.TimeoutExpired:
|
|
return "timeout"
|
|
return run
|
|
|
|
|
|
def make_handler(secret, projects, trigger, log, path="/plane"):
|
|
class Handler(BaseHTTPRequestHandler):
|
|
server_version = "plane-agentd-webhook"
|
|
|
|
def log_message(self, fmt, *args): # deliveries are logged by us, one line each
|
|
pass
|
|
|
|
def reply(self, code, obj):
|
|
body = json.dumps(obj).encode()
|
|
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 do_GET(self):
|
|
if self.path.split("?")[0] == "/healthz":
|
|
return self.reply(200, dict(ok=True, **trigger.status()))
|
|
self.reply(404, {"error": "not found"})
|
|
|
|
def do_POST(self):
|
|
if self.path.split("?")[0] != path:
|
|
return self.reply(404, {"error": "not found"})
|
|
delivery = self.headers.get("X-Plane-Delivery", "-")
|
|
try:
|
|
n = int(self.headers.get("Content-Length", ""))
|
|
except ValueError:
|
|
n = -1
|
|
if not 0 <= n <= MAX_BODY:
|
|
log("delivery=%s from=%s decision=400 bad length" % (delivery, self.client_address[0]))
|
|
return self.reply(400, {"error": "bad length"})
|
|
body = self.rfile.read(n)
|
|
sig = self.headers.get("X-Plane-Signature")
|
|
if not verify(secret, body, sig):
|
|
log("delivery=%s from=%s decision=401 %s signature" % (
|
|
delivery, self.client_address[0], "bad" if sig else "missing"))
|
|
return self.reply(401, {"error": "signature"})
|
|
try:
|
|
payload = json.loads(body)
|
|
except ValueError:
|
|
log("delivery=%s decision=400 bad json" % delivery)
|
|
return self.reply(400, {"error": "json"})
|
|
event = "%s/%s" % (payload.get("event") or self.headers.get("X-Plane-Event"), payload.get("action"))
|
|
name = route(payload, projects)
|
|
what = "delivery=%s event=%s project=%s item=%s" % (
|
|
delivery, event, name or project_of(payload) or "-", (payload.get("data") or {}).get("id", "-"))
|
|
if not name:
|
|
log(what + " decision=202 ignored (not a tenant project)")
|
|
return self.reply(202, {"ignored": "project"})
|
|
trigger.kick(delivery)
|
|
log(what + " decision=tick")
|
|
self.reply(200, {"tick": "scheduled"})
|
|
|
|
return Handler
|
|
|
|
|
|
def serve(cfg, bind=None, port=None):
|
|
"""`agentd webhook-serve`: the secret comes from the environment (agent-access.py exec --secret-env)."""
|
|
w = cfg.webhook
|
|
secret = os.environ.get(w["secret_key"])
|
|
if not secret:
|
|
raise SystemExit("webhook secret %s not in the environment; run under agent-access.py exec --secret-env %s=%s"
|
|
% (w["secret_key"], w["secret_key"], w["secret_key"]))
|
|
log = Logger(str(cfg.log_dir / "webhook.log"))
|
|
cmd = list(w["tick_command"])
|
|
if "/" in cmd[0] and not os.path.isabs(cmd[0]):
|
|
cmd[0] = str(ROOT / cmd[0]) # relative script path = repo
|
|
run = command_runner(cmd, cfg.scrubbed_env(), w.get("tick_timeout_seconds", 1000))
|
|
trigger = Trigger(run, w.get("debounce_seconds", 3), w.get("retry_seconds", 15), w.get("retry_window_seconds", 120),
|
|
log=log).start()
|
|
bind, port = bind or w.get("bind", "127.0.0.1"), port or w.get("port", 8790)
|
|
httpd = ThreadingHTTPServer((bind, port), make_handler(secret, cfg.projects, trigger, log, w.get("path", "/plane")))
|
|
log("listening on http://%s:%s%s (tenants: %s)" % (bind, port, w.get("path", "/plane"), ", ".join(cfg.projects)))
|
|
httpd.serve_forever()
|