157 lines
4.5 KiB
Python
Executable File
157 lines
4.5 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""odysseus-webhook — shell wrapper for scheduled-task webhook tokens.
|
|
|
|
Tasks in the scheduled-task system can carry a `webhook_token`. Any
|
|
HTTP POST to `/api/webhook/<token>` fires the task. This CLI lists,
|
|
rotates, and revokes those tokens.
|
|
|
|
odysseus-webhook list # tasks that have a token
|
|
odysseus-webhook show TASK_ID
|
|
odysseus-webhook rotate TASK_ID # generate a fresh token
|
|
odysseus-webhook revoke TASK_ID # remove the token
|
|
odysseus-webhook url TASK_ID --base https://app.example.com
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
import sys
|
|
import os, sys
|
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "_lib"))
|
|
from cli import quiet_logs, emit, fail, common_parser, run, REPO_ROOT as _REPO_ROOT
|
|
quiet_logs()
|
|
|
|
import argparse, json, logging, os, secrets, sys
|
|
from pathlib import Path
|
|
|
|
try:
|
|
from core.database import SessionLocal, ScheduledTask
|
|
quiet_logs()
|
|
except ModuleNotFoundError as e:
|
|
sys.stderr.write(f"error: {e}\nhint: run from repo root with venv active.\n")
|
|
sys.exit(2)
|
|
|
|
|
|
def _mask_token(token: str, reveal: bool = False) -> str:
|
|
token = token or ""
|
|
if reveal:
|
|
return token
|
|
if not token:
|
|
return ""
|
|
if len(token) <= 10:
|
|
return "***"
|
|
return token[:6] + "…" + token[-4:]
|
|
|
|
|
|
def _summary(t: "ScheduledTask", reveal: bool = False) -> dict:
|
|
tok = t.webhook_token or ""
|
|
return {
|
|
"task_id": t.id,
|
|
"name": t.name,
|
|
"status": t.status,
|
|
"task_type": t.task_type,
|
|
"webhook_token": _mask_token(tok, reveal),
|
|
"has_token": bool(tok),
|
|
}
|
|
|
|
|
|
def cmd_list(args):
|
|
db = SessionLocal()
|
|
try:
|
|
rows = db.query(ScheduledTask).filter(ScheduledTask.webhook_token.isnot(None)).all()
|
|
emit([_summary(t, args.reveal) for t in rows], args)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def cmd_show(args):
|
|
db = SessionLocal()
|
|
try:
|
|
t = db.get(ScheduledTask, args.id)
|
|
if not t:
|
|
fail(f"no task with id {args.id!r}")
|
|
emit(_summary(t, args.reveal), args)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def cmd_rotate(args):
|
|
db = SessionLocal()
|
|
try:
|
|
t = db.get(ScheduledTask, args.id)
|
|
if not t:
|
|
fail(f"no task with id {args.id!r}")
|
|
# 32 bytes urlsafe → 43-char token, plenty of entropy.
|
|
t.webhook_token = secrets.token_urlsafe(32)
|
|
db.commit()
|
|
emit({"ok": True, "task_id": t.id, "webhook_token": t.webhook_token}, args)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def cmd_revoke(args):
|
|
db = SessionLocal()
|
|
try:
|
|
t = db.get(ScheduledTask, args.id)
|
|
if not t:
|
|
fail(f"no task with id {args.id!r}")
|
|
old = t.webhook_token
|
|
t.webhook_token = None
|
|
db.commit()
|
|
emit({"ok": True, "task_id": t.id, "revoked": bool(old)}, args)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def cmd_url(args):
|
|
db = SessionLocal()
|
|
try:
|
|
t = db.get(ScheduledTask, args.id)
|
|
if not t:
|
|
fail(f"no task with id {args.id!r}")
|
|
if not t.webhook_token:
|
|
fail(f"task {args.id!r} has no webhook token (rotate one first)")
|
|
base = (args.base or "http://localhost:7000").rstrip("/")
|
|
url = f"{base}/api/webhook/{t.webhook_token}"
|
|
emit({
|
|
"task_id": t.id,
|
|
"name": t.name,
|
|
"url": url,
|
|
"curl": f"curl -X POST {url}",
|
|
}, args)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def _build_parser():
|
|
common = argparse.ArgumentParser(add_help=False)
|
|
common.add_argument("--pretty", action="store_true")
|
|
p = argparse.ArgumentParser(prog="odysseus-webhook", parents=[common])
|
|
sub = p.add_subparsers(dest="cmd", required=True)
|
|
|
|
pl = sub.add_parser("list", parents=[common])
|
|
pl.add_argument("--reveal", action="store_true", help="show full tokens")
|
|
pl.set_defaults(func=cmd_list)
|
|
|
|
psh = sub.add_parser("show", parents=[common])
|
|
psh.add_argument("id")
|
|
psh.add_argument("--reveal", action="store_true")
|
|
psh.set_defaults(func=cmd_show)
|
|
|
|
pr = sub.add_parser("rotate", parents=[common])
|
|
pr.add_argument("id")
|
|
pr.set_defaults(func=cmd_rotate)
|
|
|
|
prv = sub.add_parser("revoke", parents=[common])
|
|
prv.add_argument("id")
|
|
prv.set_defaults(func=cmd_revoke)
|
|
|
|
pu = sub.add_parser("url", parents=[common])
|
|
pu.add_argument("id")
|
|
pu.add_argument("--base", help="base URL (default http://localhost:7000)")
|
|
pu.set_defaults(func=cmd_url)
|
|
|
|
return p
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(run(_build_parser()))
|