xbp 0.9.4

import argparse
import json
import requests
import os
from colored import fg, bg, attr

def print_request_info(method, url, headers, payload, cookies):
    print(f"{fg('yellow')}Request Info:{attr('reset')}")
    print(f"{fg('yellow')}Method:{attr('reset')} {method}")
    print(f"{fg('yellow')}URL:{attr('reset')} {url}")
    print(f"{fg('blue')}Headers:{attr('reset')} {headers}")
    print(f"{fg('magenta')}Payload:\n{attr('reset')}{json.dumps(payload, indent=2)}")
    print(f"{fg('cyan')}Cookies:{attr('reset')} {cookies}")

def add_monitor(
    tag,
    url,
    name,
    icon_url=None,
    description=None,
    cookie=None,
    kener_user_jwt=None,
    monitor_id=None,
    dry_run=False
):
    """
    Add or update a monitor on Kener via the API.

    - tag: The monitor tag to assign
    - url: The endpoint to monitor
    - name: The friendly name for the monitor
    - icon_url: Small image/icon URL (optional)
    - description: Optional description
    - cookie: Raw cookie to use when authenticating the request
    - kener_user_jwt: The 'kener-user' JWT for authentication
    - monitor_id: If updating, specify existing monitor id; for new, leave blank
    - dry_run: If set, will only print the outgoing payload and not send the HTTP request
    """

    api_url = "https://kener.xbp.app/manage/app/api"

    if not cookie and not kener_user_jwt:
        raise ValueError("You must provide either a complete session cookie or at least the `kener-user` JWT")

    headers = {
        "accept": "*/*",
        "content-type": "application/json",
        "origin": "https://kener.xbp.app",
        "referer": "https://kener.xbp.app/manage/app/monitors",
        "user-agent": "kener-cli/1.0",
    }

    cookies = {}
    if cookie is not None:
        cookies = dict([c.strip().split('=', 1) for c in cookie.split(';') if '=' in c])
    elif kener_user_jwt:
        cookies['kener-user'] = kener_user_jwt

    monitor_data = {
        "tag": tag,
        "url": url,
        "name": name,
    }
    if icon_url:
        monitor_data["icon_url"] = icon_url
    if description:
        monitor_data["description"] = description

    if monitor_id:
        monitor_data["id"] = monitor_id

    down_trigger = {
        "failureThreshold": 1,
        "trigger_type": "DOWN",
        "successThreshold": 1,
        "description": "The monitor is down",
        "createIncident": "NO",
        "active": True,
        "triggers": [1],
        "severity": "critical"
    }
    degraded_trigger = {
        "failureThreshold": 1,
        "trigger_type": "DEGRADED",
        "successThreshold": 1,
        "active": False,
        "description": "The monitor is degraded",
        "createIncident": "NO",
        "triggers": [],
        "severity": "warning"
    }

    payload = {
        "action": "updateMonitorTriggers",
        "data": {
            "id": monitor_id or 1,
            "down_trigger": json.dumps(down_trigger),
            "degraded_trigger": json.dumps(degraded_trigger),
        }
    }
    payload["data"].update(monitor_data)

    if dry_run:
        print_request_info("POST", api_url, headers, payload, cookies)
        print(f"{fg('yellow')}Dry run mode: No request sent.{attr('reset')}")
        return

    # Print request before sending
    print_request_info("POST", api_url, headers, payload, cookies)

    try:
        resp = requests.post(api_url, headers=headers, json=payload, cookies=cookies)
    except Exception as e:
        print(f"{fg('red')}Request error:{attr('reset')} {e}")
        return

    status_color = fg('green') if resp.status_code == 200 else fg('red')
    print(f"{status_color}Status code: {resp.status_code}{attr('reset')}")
    
    print(f"{fg('blue')}Response headers:{attr('reset')}")
    for h, v in resp.headers.items():
        print(f"  {fg('blue')}{h}:{attr('reset')} {v}")

    print(f"{fg('magenta')}Response body:{attr('reset')}")
    try:
        response_json = resp.json()
        print(f"{fg('white')}{json.dumps(response_json, indent=2)}{attr('reset')}")
    except Exception:
        print(f"{fg('yellow')}API responded with non-JSON content:{attr('reset')}")
        try:
            print(f"{fg('white')}{resp.content.decode()}{attr('reset')}")
        except Exception:
            print(f"{fg('red')}(unable to decode response bytes){attr('reset')}")

    if resp.status_code == 200:
        print(f"{fg('green')} Successfully communicated with Kener API.{attr('reset')}")
    else:
        print(f"{fg('red')} API returned an error status.{attr('reset')}")
        if resp.text:
            print(f"{fg('red')}Raw response text:{attr('reset')}\n{resp.text}")

def get_monitors(
    status="ACTIVE",
    category_name="All Categories",
    cookie=None,
    kener_user_jwt=None,
    dry_run=False
):
    """
    Fetch monitors from Kener via the API.

    - status: Monitor status to filter (e.g. 'ACTIVE')
    - category_name: Category name filter (e.g. "All Categories")
    - cookie: Raw cookie string for authentication
    - kener_user_jwt: The 'kener-user' JWT for authentication
    - dry_run: If set, will only print the outgoing payload and not send the HTTP request
    """
    api_url = "https://kener.xbp.app/manage/app/api"

    if not cookie and not kener_user_jwt:
        raise ValueError("You must provide either a complete session cookie or at least the `kener-user` JWT")

    headers = {
        "accept": "*/*",
        "content-type": "application/json",
        "origin": "https://kener.xbp.app",
        "referer": "https://kener.xbp.app/manage/app/monitors",
        "user-agent": "kener-cli/1.0",
    }

    cookies = {}
    if cookie is not None:
        cookies = dict([c.strip().split('=', 1) for c in cookie.split(';') if '=' in c])
    elif kener_user_jwt:
        cookies['kener-user'] = kener_user_jwt

    payload = {
        "action": "getMonitors",
        "data": {
            "status": status,
            "category_name": category_name
        }
    }

    if dry_run:
        print_request_info("POST", api_url, headers, payload, cookies)
        print(f"{fg('yellow')}Dry run mode: No request sent.{attr('reset')}")
        return

    # Print request before sending
    print_request_info("POST", api_url, headers, payload, cookies)

    try:
        resp = requests.post(api_url, headers=headers, json=payload, cookies=cookies)
    except Exception as e:
        print(f"{fg('red')}Request error:{attr('reset')} {e}")
        return

    status_color = fg('green') if resp.status_code == 200 else fg('red')
    print(f"{status_color}Status code: {resp.status_code}{attr('reset')}")
    
    print(f"{fg('blue')}Response headers:{attr('reset')}")
    for h, v in resp.headers.items():
        print(f"  {fg('blue')}{h}:{attr('reset')} {v}")

    print(f"{fg('magenta')}Response body:{attr('reset')}")
    try:
        response_json = resp.json()
        print(f"{fg('white')}{json.dumps(response_json, indent=2)}{attr('reset')}")
    except Exception:
        print(f"{fg('yellow')}API responded with non-JSON content:{attr('reset')}")
        try:
            print(f"{fg('white')}{resp.content.decode()}{attr('reset')}")
        except Exception:
            print(f"{fg('red')}(unable to decode response bytes){attr('reset')}")

    if resp.status_code == 200:
        print(f"{fg('green')} Successfully retrieved monitors from Kener API.{attr('reset')}")
    else:
        print(f"{fg('red')} API returned an error status.{attr('reset')}")
        if resp.text:
            print(f"{fg('red')}Raw response text:{attr('reset')}\n{resp.text}")

def main():
    parser = argparse.ArgumentParser(description="Kener CLI utility")
    subparsers = parser.add_subparsers(dest="command", required=True)

    # Subparser for add_monitor
    parser_add = subparsers.add_parser("add", help="Add or update a monitor to Kener")
    parser_add.add_argument("--tag", required=True, help="Monitor tag")
    parser_add.add_argument("--url", required=True, help="URL to monitor")
    parser_add.add_argument("--name", required=True, help="Monitor name")
    parser_add.add_argument("--icon-url", help="Icon URL for the monitor")
    parser_add.add_argument("--description", help="Description for the monitor")
    parser_add.add_argument("--cookie", help="Provide full session cookie string")
    parser_add.add_argument("--jwt", help="Provide value of 'kener-user' JWT")
    parser_add.add_argument("--monitor-id", type=int, help="ID of monitor to update (omit to create a new one)")
    parser_add.add_argument("--dry-run", action="store_true", help="Just print payload and exit")
    
    # Subparser for get_monitors
    parser_get = subparsers.add_parser("list", help="List monitors on Kener")
    parser_get.add_argument("--status", default="ACTIVE", help="Monitor status to filter (default: ACTIVE)")
    parser_get.add_argument("--category", default="All Categories", help="Monitor category name (default: All Categories)")
    parser_get.add_argument("--cookie", help="Provide full session cookie string")
    parser_get.add_argument("--jwt", help="Provide value of 'kener-user' JWT")
    parser_get.add_argument("--dry-run", action="store_true", help="Just print payload and exit")

    args = parser.parse_args()

    if args.command == "add":
        add_monitor(
            tag=args.tag,
            url=args.url,
            name=args.name,
            icon_url=args.icon_url,
            description=args.description,
            cookie=args.cookie,
            kener_user_jwt=args.jwt,
            monitor_id=args.monitor_id,
            dry_run=args.dry_run
        )
    elif args.command == "list":
        get_monitors(
            status=args.status,
            category_name=args.category,
            cookie=args.cookie,
            kener_user_jwt=args.jwt,
            dry_run=args.dry_run
        )

if __name__ == "__main__":
    main()