catchhook.py

#!/usr/bin/env python3
"""
catchhook -- Catch and inspect HTTP requests. Local webhook receiver. Zero deps.

Starts a local server that captures ALL incoming requests and displays them.
Perfect for webhook development, API debugging, and testing.

Usage:
    py catchhook.py                                # Listen on :8888
    py catchhook.py -p 9000                        # Custom port
    py catchhook.py --response 201                 # Always respond 201
    py catchhook.py --body '{"ok":true}'           # Custom response body
    py catchhook.py --log requests.jsonl           # Save requests to file
    py catchhook.py --quiet                        # Minimal output (no body)
    py catchhook.py --filter POST                  # Only show POST requests
"""

import argparse
import json
import os
import sys
import time
from datetime import datetime
from http.server import HTTPServer, BaseHTTPRequestHandler
from urllib.parse import urlparse, parse_qs

# Colors
RESET = "\033[0m"
BOLD = "\033[1m"
DIM = "\033[2m"
GREEN = "\033[32m"
YELLOW = "\033[33m"
CYAN = "\033[36m"
RED = "\033[31m"
MAGENTA = "\033[35m"
BLUE = "\033[34m"

METHOD_COLORS = {
    "GET": GREEN, "POST": CYAN, "PUT": YELLOW,
    "PATCH": MAGENTA, "DELETE": RED, "HEAD": DIM,
    "OPTIONS": DIM,
}


def color_supported() -> bool:
    if os.environ.get("NO_COLOR"):
        return False
    if sys.platform == "win32":
        return bool(os.environ.get("TERM") or os.environ.get("WT_SESSION"))
    return hasattr(sys.stdout, "isatty") and sys.stdout.isatty()


USE_COLOR = color_supported()


def c(code: str, text: str) -> str:
    return f"{code}{text}{RESET}" if USE_COLOR else text


def format_json(text: str) -> str:
    """Try to pretty-print JSON, return original if not valid JSON."""
    try:
        data = json.loads(text)
        formatted = json.dumps(data, indent=2, ensure_ascii=False)
        return formatted
    except (json.JSONDecodeError, TypeError):
        return text


def format_size(n: int) -> str:
    if n < 1024:
        return f"{n}B"
    elif n < 1024 * 1024:
        return f"{n / 1024:.1f}KB"
    return f"{n / (1024 * 1024):.1f}MB"


class CatchHandler(BaseHTTPRequestHandler):
    """Catches and displays all incoming requests."""

    response_code = 200
    response_body = ""
    response_headers = {}
    log_file = None
    quiet = False
    method_filter = None
    request_count = 0

    def _handle(self, method: str):
        CatchHandler.request_count += 1
        req_num = CatchHandler.request_count
        timestamp = datetime.now().strftime("%H:%M:%S.%f")[:-3]

        # Filter
        if self.method_filter and method not in self.method_filter:
            self._send_response()
            return

        parsed = urlparse(self.path)
        path = parsed.path
        query = parse_qs(parsed.query)

        # Read body
        content_length = int(self.headers.get("Content-Length", 0))
        body_raw = self.rfile.read(content_length) if content_length > 0 else b""
        try:
            body_text = body_raw.decode("utf-8")
        except UnicodeDecodeError:
            body_text = f"<binary {len(body_raw)} bytes>"

        # Collect headers
        headers = dict(self.headers)

        # Build record
        record = {
            "n": req_num,
            "timestamp": datetime.now().isoformat(),
            "method": method,
            "path": path,
            "query": query if query else None,
            "headers": headers,
            "body": body_text if body_text else None,
            "body_size": len(body_raw),
            "client": f"{self.client_address[0]}:{self.client_address[1]}",
        }

        # Display
        mc = METHOD_COLORS.get(method, DIM)
        sep = c(DIM, "-" * 60)

        print(sep)
        print(f"  {c(DIM, f'#{req_num}')}  {c(DIM, timestamp)}  "
              f"{c(mc + BOLD, method)}  {c(BOLD, path)}  "
              f"{c(DIM, 'from ' + record['client'])}",
              flush=True)

        # Query params
        if query:
            print(c(DIM, "  Query:"))
            for k, v in query.items():
                print(f"    {c(CYAN, k)} = {v[0] if len(v) == 1 else v}")

        # Headers (show interesting ones)
        if not self.quiet:
            interesting = ["Content-Type", "Authorization", "User-Agent",
                           "Accept", "X-Forwarded-For", "X-Request-Id",
                           "X-Webhook-Id", "X-GitHub-Event", "Stripe-Signature"]
            shown_headers = []
            for h in interesting:
                val = headers.get(h)
                if val:
                    shown_headers.append((h, val))
            # Also show any custom X- headers not in the list
            for h, val in headers.items():
                if h.lower().startswith("x-") and h not in interesting:
                    shown_headers.append((h, val))

            if shown_headers:
                print(c(DIM, "  Headers:"))
                for h, val in shown_headers:
                    # Truncate auth values
                    if "auth" in h.lower() or "secret" in h.lower() or "signature" in h.lower():
                        if len(val) > 20:
                            display_val = val[:20] + "..."
                        else:
                            display_val = val
                    else:
                        display_val = val[:100] + ("..." if len(val) > 100 else "")
                    print(f"    {c(YELLOW, h)}: {display_val}")

        # Body
        if body_text and not self.quiet:
            content_type = headers.get("Content-Type", "")
            print(f"  {c(DIM, f'Body ({format_size(len(body_raw))}):')}  ")
            if "json" in content_type or body_text.strip().startswith("{"):
                formatted = format_json(body_text)
                for line in formatted.splitlines()[:30]:
                    print(f"    {line}")
                if formatted.count("\n") > 30:
                    print(c(DIM, f"    ... ({formatted.count(chr(10)) - 30} more lines)"))
            elif "form" in content_type:
                for pair in body_text.split("&"):
                    if "=" in pair:
                        k, _, v = pair.partition("=")
                        print(f"    {c(CYAN, k)} = {v}")
                    else:
                        print(f"    {pair}")
            else:
                # Plain text, show up to 500 chars
                display = body_text[:500]
                for line in display.splitlines()[:15]:
                    print(f"    {line}")
                if len(body_text) > 500:
                    print(c(DIM, f"    ... ({len(body_text) - 500} more chars)"))
        elif body_text and self.quiet:
            print(f"  {c(DIM, f'Body: {format_size(len(body_raw))}')}")

        print(flush=True)

        # Log to file
        if self.log_file:
            try:
                with open(self.log_file, "a", encoding="utf-8") as f:
                    f.write(json.dumps(record, default=str) + "\n")
            except Exception:
                pass

        # Send response
        self._send_response()

    def _send_response(self):
        self.send_response(self.response_code)
        self.send_header("Access-Control-Allow-Origin", "*")
        self.send_header("Access-Control-Allow-Methods", "*")
        self.send_header("Access-Control-Allow-Headers", "*")
        for k, v in self.response_headers.items():
            self.send_header(k, v)

        body = self.response_body
        if body:
            body_bytes = body.encode("utf-8")
            ct = "application/json" if body.strip().startswith("{") else "text/plain"
            self.send_header("Content-Type", ct)
            self.send_header("Content-Length", str(len(body_bytes)))
            self.end_headers()
            self.wfile.write(body_bytes)
        else:
            self.send_header("Content-Length", "0")
            self.end_headers()

    def do_GET(self): self._handle("GET")
    def do_POST(self): self._handle("POST")
    def do_PUT(self): self._handle("PUT")
    def do_PATCH(self): self._handle("PATCH")
    def do_DELETE(self): self._handle("DELETE")
    def do_HEAD(self): self._handle("HEAD")
    def do_OPTIONS(self): self._handle("OPTIONS")

    def log_message(self, format, *args):
        pass  # Suppress default logging


def main():
    parser = argparse.ArgumentParser(
        description="catchhook -- catch and inspect HTTP requests",
        formatter_class=argparse.RawDescriptionHelpFormatter,
    )
    parser.add_argument("-p", "--port", type=int, default=8888, help="Port (default: 8888)")
    parser.add_argument("--host", default="0.0.0.0", help="Host (default: 0.0.0.0)")
    parser.add_argument("--response", type=int, default=200, help="Response status code (default: 200)")
    parser.add_argument("--body", default="", help="Response body text")
    parser.add_argument("--log", help="Log requests to JSONL file")
    parser.add_argument("-q", "--quiet", action="store_true", help="Minimal output (no headers/body)")
    parser.add_argument("--filter", nargs="+", help="Only show these HTTP methods (e.g. POST PUT)")

    args = parser.parse_args()

    CatchHandler.response_code = args.response
    CatchHandler.response_body = args.body
    CatchHandler.log_file = args.log
    CatchHandler.quiet = args.quiet
    CatchHandler.method_filter = set(m.upper() for m in args.filter) if args.filter else None

    server = HTTPServer((args.host, args.port), CatchHandler)

    print()
    print(c(BOLD, "  catchhook"))
    print(f"  Listening on {c(CYAN, f'http://{args.host}:{args.port}')}")
    print(f"  Response:    {c(GREEN, str(args.response))}" +
          (f" + body ({len(args.body)}B)" if args.body else ""))
    if args.log:
        print(f"  Logging to:  {c(DIM, args.log)}")
    if args.filter:
        print(f"  Filter:      {c(YELLOW, ', '.join(args.filter))}")
    print()
    print(c(DIM, "  Waiting for requests... (Ctrl+C to stop)"))
    print()

    try:
        server.serve_forever()
    except KeyboardInterrupt:
        print(f"\n  {c(DIM, f'Caught {CatchHandler.request_count} request(s). Stopped.')}")
        server.server_close()


if __name__ == "__main__":
    main()