diff --git a/agentd/cli.py b/agentd/cli.py index 90784ec..5356c8c 100644 --- a/agentd/cli.py +++ b/agentd/cli.py @@ -38,6 +38,14 @@ def cmd_tick(cfg, a): d.tick(project=a.project) except Busy as exc: print("skip: %s" % exc) + sys.exit(75) # same code as the cron script's flock conflict: the webhook receiver retries later + + +def cmd_webhook_serve(cfg, a): + if not cfg.webhook: + sys.exit("no webhook section in %s" % cfg.path) + from .webhook import serve + serve(cfg, a.bind, a.port) def cmd_new(cfg, a, pl=None, quiet=False): @@ -159,6 +167,9 @@ def cmd_check_config(cfg, a): print("terminal %-10s %-13s %-8s executors=%s token %s: %s" % ( name, t["transport"], t["os"], ",".join(t["executors"]), env, "set" if os.environ.get(env) else "MISSING")) print("cron/exec secret args: %s" % " ".join(cfg.secret_env_args())) + if cfg.webhook: + k = cfg.webhook["secret_key"] + print("webhook-serve secret arg: --secret-env %s=%s (%s)" % (k, k, "set" if os.environ.get(k) else "not in this env")) def parser(): @@ -205,6 +216,10 @@ def parser(): help="do not tick here; only wait for the scheduled dispatcher (use when it runs on another host)") s.add_argument("--as", dest="as_") s.set_defaults(fn=cmd_smoke) + s = sub.add_parser("webhook-serve", help="Plane webhook receiver: each delivery triggers a tick (agentd/webhook.py)") + s.add_argument("--bind", help="address (default: webhook.bind)") + s.add_argument("--port", type=int, help="port (default: webhook.port)") + s.set_defaults(fn=cmd_webhook_serve) sub.add_parser("check-config", help="validate the config and list token env vars").set_defaults(fn=cmd_check_config) return ap diff --git a/agentd/config.py b/agentd/config.py index 052c3af..73ab95d 100644 --- a/agentd/config.py +++ b/agentd/config.py @@ -148,6 +148,8 @@ class Config: self.executors = data["executors"] self.projects = data["projects"] self.agentd = data.get("agentd", {}) + self.webhook = data.get("webhook", {}) + self.notify = data.get("notify") or {"kind": "none"} @property def state_dir(self): @@ -176,9 +178,10 @@ class Config: return os.environ.get(self.token_env(terminal)) def secret_keys(self): - """Every env var name that carries a secret (Plane tokens, webhook secrets): never handed to executors.""" + """Every env var name that carries a secret (Plane tokens, the webhook secret): never handed to children.""" keys = {t["token_key"] for t in self.terminals.values()} - keys |= {p["webhook_secret_key"] for p in self.projects.values() if p.get("webhook_secret_key")} + if self.webhook.get("secret_key"): + keys.add(self.webhook["secret_key"]) return keys def scrubbed_env(self): diff --git a/agentd/core.py b/agentd/core.py index 56bb83d..f028141 100644 --- a/agentd/core.py +++ b/agentd/core.py @@ -458,6 +458,13 @@ class Agentd: client.set_state(ctx.pid, iid, state, users) if iid in ctx.issues: # keep this tick's view consistent ctx.issues[iid].update(state=state, assignees=users) + self.notify(ctx, iid, state_key, user) + + def notify(self, ctx, iid, event, user): + """Notification hook (design §7 phase 3), called on every state move (event = todo/waiting/review/...). + Config `notify: {kind: none}` is the only kind today: a no-op. An ntfy sender would go here.""" + if self.cfg.notify.get("kind", "none") == "none": + return def post(self, ctx, client, iid, rec, text, pre=None): c = client.post_comment(ctx.pid, iid, text_to_html(text, pre)) diff --git a/agentd/webhook.py b/agentd/webhook.py new file mode 100644 index 0000000..8f4a11f --- /dev/null +++ b/agentd/webhook.py @@ -0,0 +1,238 @@ +"""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 (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": }, "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 +/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() diff --git a/cli/agentd b/cli/agentd index 948ee48..7307ecc 100755 --- a/cli/agentd +++ b/cli/agentd @@ -1,5 +1,5 @@ #!/usr/bin/env python3 -"""Entry point: agentd tick|new|show|comment|state|assign|smoke|check-config (see agentd/cli.py).""" +"""Entry point: agentd tick|new|show|comment|state|assign|smoke|webhook-serve|check-config (see agentd/cli.py).""" import sys from pathlib import Path diff --git a/config/agentd.example.yaml b/config/agentd.example.yaml index 4853536..babb316 100644 --- a/config/agentd.example.yaml +++ b/config/agentd.example.yaml @@ -13,6 +13,18 @@ agentd: comment_max_chars: 6000 cli_terminal: oregon # identity used by `agentd new/comment/state/assign` +# Plane webhook receiver (`agentd webhook-serve`, docs/runbook.md "Webhook"); leave the section out to run on cron only. +webhook: + bind: 127.0.0.1 + port: 8790 + path: /plane + secret_key: PLANE_WEBHOOK_SECRET # Vault key name of the Plane webhook secret (workspace-level webhook) + tick_command: [cron/agentd-tick.sh] # exits 75 when another tick holds the lock (flock -E 75) + debounce_seconds: 3 + retry_seconds: 15 + retry_window_seconds: 120 +notify: {kind: none} # notification hook: only `none` (no-op) exists today + terminals: win10: transport: warpgate-ssh @@ -72,7 +84,6 @@ projects: GWENT: id: 79cb5a60-aa4f-4138-9864-1dd70426e665 key_prefix: GW - webhook_secret_key: PLANE_WEBHOOK_GWENT states: {backlog: Backlog, todo: Todo, running: In Progress, waiting: Waiting, review: Review, done: Done} reviewer: oregon # Plane user that gets [review] items answerer: oregon # Plane user that gets [question]/[blocked] and failures (default: reviewer) diff --git a/config/agentd.yaml b/config/agentd.yaml index 4a417ec..6585cb8 100644 --- a/config/agentd.yaml +++ b/config/agentd.yaml @@ -13,6 +13,21 @@ agentd: comment_max_chars: 6000 cli_terminal: oregon +# Plane webhook receiver (`agentd webhook-serve`, systemd plane-agentd-webhook.service; docs/runbook.md "Webhook"). +# One workspace-level Plane webhook (work items + comments) -> POST http://10.77.10.249:8790/plane -> one tick. +webhook: + bind: 10.77.10.249 # LAN only (Plane VM 10.77.10.68 is on the same /24) + port: 8790 + path: /plane + secret_key: PLANE_WEBHOOK_SECRET # Vault key name; the value is injected by agent-access.py exec --secret-env + # the cron script holds `flock -n -E 75`; --scope moves the tick out of the receiver's cgroup, so restarting the + # receiver never kills a running tick + tick_command: [systemd-run, --scope, --quiet, --collect, /opt/plane-agentd/cron/agentd-tick-lxc.sh] + debounce_seconds: 3 # deliveries within this window -> one tick + retry_seconds: 15 # lock busy -> retry every 15 s ... + retry_window_seconds: 120 # ... for up to 2 min; then the cron tick covers it +notify: {kind: none} # notification hook, no-op (no ntfy service); core calls it on every state move + terminals: win10: transport: warpgate-ssh @@ -85,4 +100,3 @@ projects: timeout_hours: 4 max_rounds_per_thread: 8 max_progress_rounds: 12 - # notify (ntfy) is phase 3: left out on purpose diff --git a/config/schema.json b/config/schema.json index 8b912d6..087ebfe 100644 --- a/config/schema.json +++ b/config/schema.json @@ -36,6 +36,30 @@ "cli_terminal": {"type": "string"} } }, + "webhook": { + "type": "object", + "required": ["secret_key", "tick_command"], + "additionalProperties": false, + "properties": { + "bind": {"type": "string"}, + "port": {"type": "integer", "minimum": 1}, + "path": {"type": "string"}, + "secret_key": {"type": "string"}, + "tick_command": {"type": "array", "minItems": 1, "items": {"type": "string"}}, + "tick_timeout_seconds": {"type": "integer", "minimum": 10}, + "debounce_seconds": {"type": "number", "minimum": 0}, + "retry_seconds": {"type": "number", "minimum": 1}, + "retry_window_seconds": {"type": "number", "minimum": 0} + } + }, + "notify": { + "type": "object", + "additionalProperties": false, + "properties": { + "kind": {"enum": ["none"]}, + "on": {"$ref": "#/definitions/strings"} + } + }, "terminals": {"type": "object", "minProperties": 1, "additionalProperties": {"$ref": "#/definitions/terminal"}}, "executors": {"type": "object", "minProperties": 1, "additionalProperties": {"$ref": "#/definitions/executor"}}, "projects": {"type": "object", "minProperties": 1, "additionalProperties": {"$ref": "#/definitions/project"}} @@ -101,7 +125,6 @@ "properties": { "id": {"type": "string"}, "key_prefix": {"type": "string"}, - "webhook_secret_key": {"type": "string"}, "states": { "type": "object", "required": ["todo", "running", "waiting", "review", "done"], diff --git a/cron/agentd-tick-lxc.sh b/cron/agentd-tick-lxc.sh index 20bac7d..8f9b51c 100755 --- a/cron/agentd-tick-lxc.sh +++ b/cron/agentd-tick-lxc.sh @@ -2,7 +2,9 @@ # LXC agentd-01 cron entry (*/2, root): one agentd tick under server-access exec (Plane tokens only in the # process env), with a lock. The outer flock skips a tick while the previous one still runs; agentd also holds # its own state lock. Logs: /var/log/plane-agentd/{agentd.log,cron.log}; state: /var/lib/plane-agentd. -exec flock -n /var/lib/plane-agentd/cron.lock \ +# Also the webhook receiver's tick command (agentd/webhook.py): exit 75 = lock held (flock -E 75, or agentd's own +# state lock), the receiver then retries. +exec flock -n -E 75 /var/lib/plane-agentd/cron.lock \ timeout 900 python3 -I /root/.local/share/server-access-client/agent-access.py \ --config https://nas.tail9531b.ts.net:8443/identity/agent.json exec \ --secret-env PLANE_OREGON_API_TOKEN=PLANE_OREGON_API_TOKEN \ diff --git a/cron/agentd-tick.sh b/cron/agentd-tick.sh index 0c384a8..47a7b75 100755 --- a/cron/agentd-tick.sh +++ b/cron/agentd-tick.sh @@ -2,7 +2,7 @@ # oregon cron entry (*/2) -- ROLLBACK ONLY since PAD-11: the dispatcher runs on the LXC agentd-01 # (cron/agentd-tick-lxc.sh). Never run both: two dispatchers double-dispatch. See docs/runbook.md "回滚到 oregon". # One agentd tick under server-access exec (Plane tokens in the process env only), with a lock. -exec flock -n /home/ubuntu/.gwent-board/agentd.lock \ +exec flock -n -E 75 /home/ubuntu/.gwent-board/agentd.lock \ timeout 900 python3 -I /home/ubuntu/.local/share/server-access-client/agent-access.py \ --config https://nas.tail9531b.ts.net:8443/identity/agent.json exec \ --secret-env PLANE_OREGON_API_TOKEN=PLANE_OREGON_API_TOKEN \ diff --git a/systemd/plane-agentd-webhook.service b/systemd/plane-agentd-webhook.service new file mode 100644 index 0000000..d437e63 --- /dev/null +++ b/systemd/plane-agentd-webhook.service @@ -0,0 +1,24 @@ +# Plane webhook receiver on the LXC agentd-01 (docs/runbook.md "Webhook", docs/deploy-lxc.md §8). +# The webhook secret is injected by agent-access.py exec (Vault key PLANE_WEBHOOK_SECRET) into the process env only. +# Install: cp systemd/plane-agentd-webhook.service /etc/systemd/system/ && systemctl daemon-reload && +# systemctl enable --now plane-agentd-webhook +[Unit] +Description=plane-agentd Plane webhook receiver (triggers dispatcher ticks) +Wants=network-online.target +After=network-online.target +StartLimitIntervalSec=0 + +[Service] +Type=simple +Environment=HOME=/root +WorkingDirectory=/opt/plane-agentd +ExecStart=/usr/bin/python3 -I /root/.local/share/server-access-client/agent-access.py \ + --config https://nas.tail9531b.ts.net:8443/identity/agent.json exec \ + --secret-env PLANE_WEBHOOK_SECRET=PLANE_WEBHOOK_SECRET --timeout 315360000 -- \ + /usr/bin/python3 /opt/plane-agentd/cli/agentd --config /opt/plane-agentd/config/agentd.yaml \ + webhook-serve --bind 10.77.10.249 --port 8790 +Restart=always +RestartSec=5 + +[Install] +WantedBy=multi-user.target diff --git a/tests/test_transports.py b/tests/test_transports.py index 9b1f338..145759b 100644 --- a/tests/test_transports.py +++ b/tests/test_transports.py @@ -38,7 +38,7 @@ class LocalSecretsTest(unittest.TestCase): data = config_data() data["agentd"]["state_dir"] = str(Path(tmp) / "state") data["terminals"]["oregon"]["repo"] = tmp - data["projects"]["GWENT"]["webhook_secret_key"] = "HOOK_GW" + data["webhook"] = {"secret_key": "HOOK_GW", "tick_command": ["true"]} cfg = Config(data) self.assertEqual(cfg.secret_keys(), {"TOK_OR", "TOK_W10", "HOOK_GW"}) t = LocalTransport(cfg, "oregon") diff --git a/tests/test_webhook.py b/tests/test_webhook.py new file mode 100644 index 0000000..71e59f4 --- /dev/null +++ b/tests/test_webhook.py @@ -0,0 +1,184 @@ +import json +import threading +import time +import unittest +import urllib.error +import urllib.request +from http.server import ThreadingHTTPServer + +from agentd.webhook import BUSY, Trigger, make_handler, route, sign, verify + +SECRET = "plane_wh_test" +PROJECTS = {"GWENT": {"id": "proj-gw"}, "PAD": {"id": "proj-pad"}} + + +def payload(project="proj-gw", event="issue", action="updated"): + return {"event": event, "action": action, "webhook_id": "w1", "workspace_id": "ws", "workspace_slug": "kappiee", + "data": {"id": "i1", "project": project}, "activity": {"field": "state"}} + + +def plane_body(p): + return json.dumps(p).encode() # what Plane signs (json.dumps(payload)) and sends + + +class SignatureTest(unittest.TestCase): + def test_ok_bad_missing(self): + body = plane_body(payload()) + self.assertTrue(verify(SECRET, body, sign(SECRET, body))) + self.assertTrue(verify(SECRET, body, sign(SECRET, body).upper())) + self.assertFalse(verify(SECRET, body, sign("other", body))) + self.assertFalse(verify(SECRET, body + b" ", sign(SECRET, body + b"x"))) + self.assertFalse(verify(SECRET, body, None)) + self.assertFalse(verify(SECRET, body, "")) + self.assertFalse(verify("", body, sign("", body))) # no secret configured: reject everything + self.assertFalse(verify(SECRET, body, "zzé")) # non-hex garbage + + def test_reencoded_body_matches_plane_serialisation(self): + p = payload() + compact = json.dumps(p, separators=(",", ":")).encode() + self.assertTrue(verify(SECRET, compact, sign(SECRET, plane_body(p)))) + self.assertFalse(verify(SECRET, b"not json", sign(SECRET, plane_body(p)))) + + +class RouteTest(unittest.TestCase): + def test_route(self): + self.assertEqual(route(payload(), PROJECTS), "GWENT") + self.assertEqual(route(payload("proj-pad", "issue_comment", "created"), PROJECTS), "PAD") + self.assertIsNone(route(payload("proj-other"), PROJECTS)) + self.assertIsNone(route({"event": "issue", "action": "deleted", "data": {"id": "i1"}}, PROJECTS)) + self.assertIsNone(route({"event": "issue"}, PROJECTS)) + self.assertEqual(route({"data": {"project": {"id": "proj-gw"}}}, PROJECTS), "GWENT") + + +class FakeTrigger: + def __init__(self): + self.kicks = [] + + def kick(self, reason): + self.kicks.append(reason) + + def status(self): + return {"pending": bool(self.kicks)} + + +class HandlerTest(unittest.TestCase): + def setUp(self): + self.trigger, self.lines = FakeTrigger(), [] + self.httpd = ThreadingHTTPServer(("127.0.0.1", 0), make_handler(SECRET, PROJECTS, self.trigger, self.lines.append)) + threading.Thread(target=self.httpd.serve_forever, daemon=True).start() + self.url = "http://127.0.0.1:%d" % self.httpd.server_address[1] + + def tearDown(self): + self.httpd.shutdown() + self.httpd.server_close() + + def post(self, body, sig, path="/plane"): + headers = {"Content-Type": "application/json", "X-Plane-Delivery": "d-1", "X-Plane-Event": "issue"} + if sig is not None: + headers["X-Plane-Signature"] = sig + req = urllib.request.Request(self.url + path, data=body, headers=headers, method="POST") + try: + with urllib.request.urlopen(req, timeout=5) as r: + return r.status + except urllib.error.HTTPError as e: + return e.code + + def test_signed_tenant_delivery_triggers(self): + body = plane_body(payload()) + self.assertEqual(self.post(body, sign(SECRET, body)), 200) + self.assertEqual(self.trigger.kicks, ["d-1"]) + self.assertIn("project=GWENT", self.lines[-1]) + self.assertIn("decision=tick", self.lines[-1]) + + def test_bad_or_missing_signature_is_401_and_triggers_nothing(self): + body = plane_body(payload()) + self.assertEqual(self.post(body, sign("wrong", body)), 401) + self.assertEqual(self.post(body, None), 401) + self.assertEqual(self.trigger.kicks, []) + self.assertIn("401 missing", self.lines[-1]) + + def test_other_project_is_202(self): + body = plane_body(payload("proj-other")) + self.assertEqual(self.post(body, sign(SECRET, body)), 202) + self.assertEqual(self.trigger.kicks, []) + + def test_wrong_path_and_healthz(self): + body = plane_body(payload()) + self.assertEqual(self.post(body, sign(SECRET, body), "/other"), 404) + with urllib.request.urlopen(self.url + "/healthz", timeout=5) as r: + self.assertEqual((r.status, json.loads(r.read())["ok"]), (200, True)) + + +class FakeRunner: + def __init__(self, codes=(), hold=0.0): + self.codes, self.hold, self.calls = list(codes), hold, 0 + self.started = threading.Event() + + def __call__(self): + self.calls += 1 + self.started.set() + time.sleep(self.hold) + return self.codes.pop(0) if self.codes else 0 + + +class TriggerTest(unittest.TestCase): + def make(self, runner, **kw): + kw = dict(dict(debounce=0.2, retry=0.1, retry_window=1.0), **kw) + self.lines = [] + return Trigger(runner, log=self.lines.append, **kw) + + def test_burst_coalesces_into_one_tick(self): + run = FakeRunner() + t = self.make(run) + for i in range(5): + t.kick("d%d" % i) + self.assertTrue(t.step(wait=2)) + self.assertEqual(run.calls, 1) + self.assertIn("5 deliveries", self.lines[0]) + self.assertFalse(t.step(wait=0.4)) # nothing left + self.assertEqual(run.calls, 1) + + def test_debounce_waits_before_ticking(self): + run = FakeRunner() + t = self.make(run, debounce=0.3) + t0 = time.monotonic() + t.kick("d1") + t.step(wait=2) + self.assertGreaterEqual(time.monotonic() - t0, 0.29) + + def test_delivery_during_a_tick_causes_one_follow_up(self): + run = FakeRunner(hold=0.3) + t = self.make(run, debounce=0.05).start() + t.kick("d1") + self.assertTrue(run.started.wait(2)) + t.kick("d2") # arrives while the first tick runs + t.kick("d3") + time.sleep(1.0) + self.assertEqual(run.calls, 2) + self.assertEqual(t.status()["pending"], False) + + def test_busy_lock_is_retried_until_the_tick_runs(self): + run = FakeRunner(codes=[BUSY, BUSY, 0]) + t = self.make(run, debounce=0.0) + t.kick("d1") + for _ in range(3): + self.assertTrue(t.step(wait=2)) + self.assertEqual(run.calls, 3) + self.assertEqual(sum("retry in" in l for l in self.lines), 2) + self.assertIn("1 deliveries: d1", self.lines[-2]) # the delivery is carried into the retry + self.assertIsNone(t.status()["retrying_since"]) + self.assertFalse(t.step(wait=0.3)) + + def test_busy_gives_up_after_the_window(self): + run = FakeRunner(codes=[BUSY] * 10) + t = self.make(run, debounce=0.0, retry=0.1, retry_window=0.35) + t.kick("d1") + while t.step(wait=1): + pass + self.assertTrue(any("giving up" in l for l in self.lines)) + self.assertLessEqual(run.calls, 5) + self.assertFalse(t.status()["pending"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/tools/plane-webhook.py b/tools/plane-webhook.py new file mode 100755 index 0000000..7c55d26 --- /dev/null +++ b/tools/plane-webhook.py @@ -0,0 +1,121 @@ +#!/usr/bin/env python3 +"""Plane workspace webhooks, as the workspace admin kappiee (session login; the internal /api/workspaces//webhooks/ +endpoint needs admin). Never prints secrets: a new or regenerated secret goes straight into the Vault as +PLANE_WEBHOOK_SECRET (then restart plane-agentd-webhook). Run under + agent-access.py --config exec --secret-env PLANE_KAPPIEE_PASSWORD=PLANE_KAPPIEE_PASSWORD -- python3 -I tools/plane-webhook.py ... +usage: list | create URL (inactive; work item + comment events) | patch ID key=value... (e.g. is_active=true) + | regenerate ID | delete ID | logs ID [n]""" +import http.cookiejar, json, os, ssl, sys, urllib.parse, urllib.request + +BASE = "https://plane.kappiee.com" +WS = "kappiee" +jar = http.cookiejar.CookieJar() +op = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(jar)) +HDR = {"Referer": BASE + "/", "Origin": BASE, "User-Agent": "plane-agentd-admin"} + + +def csrf(): + return next((c.value for c in jar if c.name == "csrftoken"), "") + + +def call(method, path, body=None, form=None): + h = dict(HDR) + data = None + if form is not None: + data = urllib.parse.urlencode(form).encode() + h["Content-Type"] = "application/x-www-form-urlencoded" + elif body is not None: + data = json.dumps(body).encode() + h["Content-Type"] = "application/json" + if method != "GET": + h["X-CSRFToken"] = csrf() + req = urllib.request.Request(BASE + path, data=data, headers=h, method=method) + try: + with op.open(req, timeout=30) as r: + raw = r.read() + return r.status, raw, r.geturl() + except urllib.error.HTTPError as e: + return e.code, e.read(), path + + +def login(): + st, raw, _ = call("GET", "/auth/get-csrf-token/") + tok = json.loads(raw).get("csrf_token") or csrf() + st, raw, url = call("POST", "/auth/sign-in/", form={"csrfmiddlewaretoken": tok, "email": "kappiee@kappiee.com", + "password": os.environ["PLANE_KAPPIEE_PASSWORD"]}) + ok = any(c.name.startswith("session") for c in jar) + print("login", st, "session cookie" if ok else "NO SESSION", urllib.parse.urlparse(url).path) + if not ok: + sys.exit(1) + + +def redact(d): + return {k: ("<%d chars>" % len(v) if k == "secret_key" and v else v) for k, v in d.items()} + + +def vault_store(name, value): + ident = json.load(urllib.request.urlopen("https://nas.tail9531b.ts.net:8443/identity/agent.json", timeout=30)) + v = ident["vault"] + ctx = ssl.create_default_context() + ctx.check_hostname = False + ctx.verify_mode = ssl.CERT_NONE + req = urllib.request.Request(v["url"].rstrip("/") + "/v1/credentials", method="POST", + data=json.dumps({"vault": "default", "credentials": {name: value}}).encode(), + headers={"Authorization": "Bearer " + v["token"], "X-Vault": "default", + "Content-Type": "application/json"}) + with urllib.request.urlopen(req, context=ctx, timeout=30) as r: + print("vault store", name, r.status) + + +def store_secret(w): + if w.get("secret_key"): + vault_store("PLANE_WEBHOOK_SECRET", w["secret_key"]) + + +login() +P = "/api/workspaces/%s/webhooks/" % WS +cmd = sys.argv[1] +if cmd == "list": + st, raw, _ = call("GET", P) + print("list", st) + if st == 200: + for w in json.loads(raw): + print(json.dumps(redact(w))) + else: + print(raw[:300]) +elif cmd == "create": + body = {"url": sys.argv[2], "is_active": False, "project": False, "issue": True, "module": False, "cycle": False, + "issue_comment": True} + st, raw, _ = call("POST", P, body) + print("create", st) + if st in (200, 201): + w = json.loads(raw) + store_secret(w) + print(json.dumps(redact(w))) + else: + print(raw[:500]) +elif cmd == "patch": + body = {} + for kv in sys.argv[3:]: + k, v = kv.split("=", 1) + body[k] = {"true": True, "false": False}.get(v, v) + st, raw, _ = call("PATCH", P + sys.argv[2] + "/", body) + print("patch", st, json.dumps(redact(json.loads(raw))) if st == 200 else raw[:500]) +elif cmd == "regenerate": + st, raw, _ = call("POST", P + sys.argv[2] + "/regenerate/", {}) + print("regenerate", st) + if st in (200, 201): + w = json.loads(raw) + store_secret(w) + print(json.dumps(redact(w))) + else: + print(raw[:500]) +elif cmd == "delete": + st, raw, _ = call("DELETE", P + sys.argv[2] + "/") + print("delete", st, raw[:200]) +elif cmd == "logs": + st, raw, _ = call("GET", "/api/workspaces/%s/webhook-logs/%s/" % (WS, sys.argv[2])) + print("logs", st) + if st == 200: + for l in json.loads(raw)[: int(sys.argv[3]) if len(sys.argv) > 3 else 10]: + print(l.get("created_at"), l.get("event_type"), l.get("request_method"), l.get("response_status"), (l.get("response_body") or "")[:120], l.get("retry_count"))