odysseus-webhook

#!/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()))