claudetools/.claude/skills/mailprotector/scripts/mp.py

#!/usr/bin/env python3
"""CLI for the mailprotector skill — Mailprotector CloudFilter REST API.

Read subcommands run freely. Write subcommands (release, release-many, add-rule,
enable-release, raw with a non-GET method) refuse to run unless --confirm is
passed; without it they print what they WOULD do and exit non-zero.

NOTE: find-user is a READ even though it is a POST under the hood — it is NOT
gated.

Read examples:
  py mp.py status
  py mp.py domains
  py mp.py domain <domain_id>
  py mp.py customers <reseller_id>
  py mp.py users <scope> <id>
  py mp.py find-user user@client.com
  py mp.py logs <scope> <id> --sender boss@vendor.com --decision quarantine_spam
  py mp.py messages <scope> <id> --recipient ceo@client.com
  py mp.py config <scope> <id>
  py mp.py rules <scope> <id>

Write examples (all require --confirm):
  py mp.py release <message_id> --confirm
  py mp.py release-many <scope> <id> --ids 111,222,333 --confirm
  py mp.py add-rule <scope> <id> --value vendor.com --type allow --confirm
  py mp.py enable-release <scope> <id> --confirm

Escape hatch (raw request against any path; non-GET requires --confirm):
  py mp.py raw GET domains/123/logs
  py mp.py raw POST messages/999/deliver --body '{...}' --confirm

`scope` values are validated against:
  resellers, customers, domains, user_groups, users
"""
from __future__ import annotations

import argparse
import json
import sys

from mp_client import MailprotectorClient, MailprotectorError, VALID_SCOPES


def _emit(obj) -> None:
    print(json.dumps(obj, indent=2, ensure_ascii=False, default=str))


def _parse_body(raw: str | None) -> dict | None:
    if raw is None:
        return None
    try:
        parsed = json.loads(raw)
    except json.JSONDecodeError as exc:
        raise SystemExit(f"--body is not valid JSON: {exc}")
    if not isinstance(parsed, dict):
        raise SystemExit("--body must be a JSON object")
    return parsed


def _require_confirm(args, action: str, detail: str) -> None:
    if not getattr(args, "confirm", False):
        print(f"[DRY RUN] Would {action}: {detail}")
        print("Refusing to perform a write without --confirm. Re-run with --confirm.")
        raise SystemExit(2)


def _add_log_filters(sp) -> None:
    """Attach the shared logs/messages filter flags to a subparser."""
    sp.add_argument("scope", choices=VALID_SCOPES)
    sp.add_argument("id")
    sp.add_argument("--sender")
    sp.add_argument("--recipient")
    sp.add_argument("--subject")
    sp.add_argument(
        "--decision",
        choices=[
            "default",
            "deliver",
            "quarantine_spam",
            "quarantine_virus",
            "quarantine_policy",
            "bounce",
            "encrypt",
            "delete",
        ],
    )
    sp.add_argument(
        "--sort-field",
        dest="sort_field",
        choices=[
            "@timestamp",
            "prime.direction",
            "prime.from_header_raw",
            "prime.recipient",
            "prime.subject",
            "prime.decision",
            "prime.score",
        ],
    )
    sp.add_argument(
        "--sort-direction", dest="sort_direction", choices=["desc", "asc"]
    )
    sp.add_argument("--page", type=int)
    sp.add_argument("--page-size", dest="page_size", type=int)


def main(argv=None) -> int:
    p = argparse.ArgumentParser(
        prog="mp.py", description="Mailprotector CloudFilter REST API CLI"
    )
    p.add_argument("--json", action="store_true", help="emit raw JSON (default)")
    sub = p.add_subparsers(dest="cmd", required=True)

    # --- read ---
    sub.add_parser("status", help="validate token (GET /domains per_page=1)")

    sp = sub.add_parser("domains", help="list domains (global or scoped)")
    sp.add_argument("--scope", choices=VALID_SCOPES)
    sp.add_argument("--id", help="entity id (required if --scope given)")
    sp.add_argument("--page", type=int, default=1)
    sp.add_argument("--per-page", dest="per_page", type=int, default=25)

    sp = sub.add_parser("domain", help="one domain")
    sp.add_argument("domain_id")

    sp = sub.add_parser("customers", help="customers under a reseller")
    sp.add_argument("reseller_id")
    sp.add_argument("--page", type=int, default=1)
    sp.add_argument("--per-page", dest="per_page", type=int, default=25)

    sp = sub.add_parser("customer", help="one customer")
    sp.add_argument("customer_id")

    sp = sub.add_parser("users", help="users under an entity")
    sp.add_argument("scope", choices=VALID_SCOPES)
    sp.add_argument("id")
    sp.add_argument("--page", type=int, default=1)
    sp.add_argument("--per-page", dest="per_page", type=int, default=25)

    sp = sub.add_parser("user", help="one user")
    sp.add_argument("user_id")

    sp = sub.add_parser("find-user", help="find a user/alias by email address")
    sp.add_argument("address")

    sp = sub.add_parser("logs", help="mail-flow logs for an entity")
    _add_log_filters(sp)

    sp = sub.add_parser("messages", help="held/quarantined messages for an entity")
    _add_log_filters(sp)

    sp = sub.add_parser("config", help="entity configuration")
    sp.add_argument("scope", choices=VALID_SCOPES)
    sp.add_argument("id")

    sp = sub.add_parser("rules", help="allow/block rules for an entity")
    sp.add_argument("scope", choices=VALID_SCOPES)
    sp.add_argument("id")

    # --- write (gated) ---
    sp = sub.add_parser("release", help="release one held message")
    sp.add_argument("message_id")
    sp.add_argument("--recipients", help="optional csv of override recipients")
    sp.add_argument("--confirm", action="store_true")

    sp = sub.add_parser("release-many", help="bulk-release held messages")
    sp.add_argument("scope", choices=VALID_SCOPES)
    sp.add_argument("id")
    sp.add_argument("--ids", help="csv of message ids to release")
    sp.add_argument("--all", action="store_true", help="release all selected")
    sp.add_argument("--recipients", help="optional csv of override recipients")
    sp.add_argument("--confirm", action="store_true")

    sp = sub.add_parser("add-rule", help="add an allow/block rule")
    sp.add_argument("scope", choices=VALID_SCOPES)
    sp.add_argument("id")
    sp.add_argument("--value", required=True)
    sp.add_argument("--type", dest="rule_type", required=True, choices=["allow", "block"])
    sp.add_argument("--confirm", action="store_true")

    sp = sub.add_parser(
        "enable-release", help="enable spam release on an entity (allow_spam_release)"
    )
    sp.add_argument("scope", choices=VALID_SCOPES)
    sp.add_argument("id")
    sp.add_argument("--confirm", action="store_true")

    # --- raw escape hatch ---
    sp = sub.add_parser("raw", help="raw request against any path")
    sp.add_argument("method", choices=["GET", "POST", "PUT", "PATCH", "DELETE"])
    sp.add_argument("path", help="relative path, e.g. domains/123/logs")
    sp.add_argument("--body")
    sp.add_argument("--confirm", action="store_true")

    args = p.parse_args(argv)
    client = MailprotectorClient()

    try:
        if args.cmd == "status":
            result = client.status()
            _emit({"status": "ok", "auth": "valid", "sample": result})
        elif args.cmd == "domains":
            if args.scope and not args.id:
                raise SystemExit("--id is required when --scope is given")
            _emit(
                client.domains(
                    scope=args.scope,
                    entity_id=args.id,
                    page=args.page,
                    per_page=args.per_page,
                )
            )
        elif args.cmd == "domain":
            _emit(client.domain(args.domain_id))
        elif args.cmd == "customers":
            _emit(
                client.customers(
                    args.reseller_id, page=args.page, per_page=args.per_page
                )
            )
        elif args.cmd == "customer":
            _emit(client.customer(args.customer_id))
        elif args.cmd == "users":
            _emit(
                client.users(
                    args.scope, args.id, page=args.page, per_page=args.per_page
                )
            )
        elif args.cmd == "user":
            _emit(client.user(args.user_id))
        elif args.cmd == "find-user":
            _emit(client.find_user(args.address))
        elif args.cmd == "logs":
            _emit(
                client.logs(
                    args.scope,
                    args.id,
                    sort_direction=args.sort_direction,
                    sort_field=args.sort_field,
                    page=args.page,
                    page_size=args.page_size,
                    sender=args.sender,
                    recipient=args.recipient,
                    subject=args.subject,
                    decision=args.decision,
                )
            )
        elif args.cmd == "messages":
            _emit(
                client.messages(
                    args.scope,
                    args.id,
                    sort_direction=args.sort_direction,
                    sort_field=args.sort_field,
                    page=args.page,
                    page_size=args.page_size,
                    sender=args.sender,
                    recipient=args.recipient,
                    subject=args.subject,
                    decision=args.decision,
                )
            )
        elif args.cmd == "config":
            _emit(client.configuration(args.scope, args.id))
        elif args.cmd == "rules":
            _emit(client.allow_block_rules(args.scope, args.id))

        elif args.cmd == "release":
            detail = args.message_id
            if args.recipients:
                detail += f" -> {args.recipients}"
            _require_confirm(args, "RELEASE held message", detail)
            _emit(client.release_message(args.message_id, recipients=args.recipients))
        elif args.cmd == "release-many":
            if not args.ids and not args.all:
                raise SystemExit("release-many requires --ids <csv> or --all")
            target = "ALL selected" if args.all else f"ids={args.ids}"
            _require_confirm(
                args, "BULK RELEASE held messages", f"{args.scope}/{args.id}: {target}"
            )
            _emit(
                client.release_many(
                    args.scope,
                    args.id,
                    ids=args.ids,
                    all_selected=args.all,
                    recipients=args.recipients,
                )
            )
        elif args.cmd == "add-rule":
            _require_confirm(
                args,
                f"add {args.rule_type} rule",
                f"{args.scope}/{args.id}: {args.value}",
            )
            _emit(
                client.add_rule(args.scope, args.id, args.value, args.rule_type)
            )
        elif args.cmd == "enable-release":
            _require_confirm(
                args,
                "enable spam release (allow_spam_release=true)",
                f"{args.scope}/{args.id}",
            )
            _emit(client.enable_release(args.scope, args.id))

        elif args.cmd == "raw":
            body = _parse_body(args.body)
            if args.method != "GET":
                _require_confirm(args, f"{args.method} {args.path}", json.dumps(body))
            _emit(client.request(args.method, args.path, json_body=body))
        else:
            p.error(f"unknown command {args.cmd}")
    except MailprotectorError as exc:
        print(f"[ERROR] {exc}", file=sys.stderr)
        return 1
    return 0


if __name__ == "__main__":
    raise SystemExit(main())