VAP Media · Unified MCP Server for AI Agents (Flux · Veo · Suno)

"""
VAP Webhook Helpers
Utilities for webhook signature verification
"""

import hmac
import hashlib
import json
import time
from typing import Union


def verify_webhook_signature(
    payload: Union[dict, str, bytes],
    signature: str,
    timestamp: Union[str, int],
    secret: str,
    max_age_seconds: int = 300,
) -> bool:
    """
    Verify VAP webhook signature.

    VAP sends webhooks with HMAC-SHA256 signatures for security.
    Use this function to verify that webhooks are authentic.

    Args:
        payload: Webhook payload (dict, JSON string, or bytes)
        signature: Value from X-VAP-Signature header
        timestamp: Value from X-VAP-Timestamp header
        secret: Your webhook secret (from webhook registration)
        max_age_seconds: Maximum age of signature (default: 5 minutes)

    Returns:
        True if signature is valid and not expired, False otherwise

    Example:
        from vap_client import verify_webhook_signature

        @app.post("/webhooks/vape")
        def handle_webhook(request):
            is_valid = verify_webhook_signature(
                payload=request.body,
                signature=request.headers["X-VAP-Signature"],
                timestamp=request.headers["X-VAP-Timestamp"],
                secret="your_webhook_secret"
            )

            if not is_valid:
                return {"error": "Invalid signature"}, 401

            # Process webhook...
    """
    try:
        # Convert timestamp to int
        ts = int(timestamp)

        # Check timestamp age
        current_time = int(time.time())
        if abs(current_time - ts) > max_age_seconds:
            return False

        # Normalize payload to JSON string
        if isinstance(payload, bytes):
            payload = payload.decode("utf-8")
        if isinstance(payload, dict):
            payload_json = json.dumps(payload, separators=(",", ":"), sort_keys=True)
        else:
            # Try to parse and re-serialize for consistent formatting
            try:
                parsed = json.loads(payload)
                payload_json = json.dumps(parsed, separators=(",", ":"), sort_keys=True)
            except:
                payload_json = payload

        # Create signed payload
        signed_payload = f"{ts}.{payload_json}"

        # Generate expected signature
        expected_signature = hmac.new(
            secret.encode("utf-8"),
            signed_payload.encode("utf-8"),
            hashlib.sha256
        ).hexdigest()

        # Constant-time comparison
        return hmac.compare_digest(signature, expected_signature)

    except (ValueError, TypeError, AttributeError):
        return False


def generate_webhook_signature(
    payload: Union[dict, str],
    secret: str,
    timestamp: int = None,
) -> tuple:
    """
    Generate webhook signature (for testing purposes).

    Args:
        payload: Webhook payload
        secret: Webhook secret
        timestamp: Unix timestamp (default: current time)

    Returns:
        Tuple of (signature, timestamp)

    Example:
        signature, timestamp = generate_webhook_signature(
            payload={"event": "test"},
            secret="your_secret"
        )
    """
    if timestamp is None:
        timestamp = int(time.time())

    # Normalize payload
    if isinstance(payload, dict):
        payload_json = json.dumps(payload, separators=(",", ":"), sort_keys=True)
    else:
        try:
            parsed = json.loads(payload)
            payload_json = json.dumps(parsed, separators=(",", ":"), sort_keys=True)
        except:
            payload_json = payload

    # Create signed payload
    signed_payload = f"{timestamp}.{payload_json}"

    # Generate signature
    signature = hmac.new(
        secret.encode("utf-8"),
        signed_payload.encode("utf-8"),
        hashlib.sha256
    ).hexdigest()

    return signature, timestamp


class WebhookEvent:
    """
    Parsed webhook event.

    Usage:
        event = WebhookEvent.from_request(payload, signature, timestamp, secret)
        if event.is_valid:
            print(f"Event type: {event.event_type}")
            print(f"Data: {event.data}")
    """

    def __init__(
        self,
        payload: dict,
        is_valid: bool,
        event_type: str = None,
        timestamp: str = None,
    ):
        self.payload = payload
        self.is_valid = is_valid
        self.event_type = event_type or payload.get("event", "unknown")
        self.timestamp = timestamp or payload.get("timestamp")
        self.data = payload

    @classmethod
    def from_request(
        cls,
        payload: Union[dict, str, bytes],
        signature: str,
        timestamp: Union[str, int],
        secret: str,
        max_age_seconds: int = 300,
    ) -> "WebhookEvent":
        """
        Create WebhookEvent from request data.

        Args:
            payload: Request body
            signature: X-VAP-Signature header
            timestamp: X-VAP-Timestamp header
            secret: Your webhook secret
            max_age_seconds: Max signature age

        Returns:
            WebhookEvent with is_valid flag
        """
        # Verify signature
        is_valid = verify_webhook_signature(
            payload=payload,
            signature=signature,
            timestamp=timestamp,
            secret=secret,
            max_age_seconds=max_age_seconds,
        )

        # Parse payload
        if isinstance(payload, bytes):
            payload = payload.decode("utf-8")
        if isinstance(payload, str):
            try:
                payload = json.loads(payload)
            except:
                payload = {"raw": payload}

        return cls(
            payload=payload,
            is_valid=is_valid,
            timestamp=str(timestamp),
        )

    def __repr__(self):
        return f"WebhookEvent(event_type={self.event_type!r}, is_valid={self.is_valid})"


# Webhook event types
WEBHOOK_EVENTS = {
    "generation.complete": "Image generation completed",
    "balance.low": "Account balance is low",
    "balance.depleted": "Account balance is depleted",
    "rate.limit.exceeded": "Rate limit exceeded",
    "test": "Test webhook",
}