leocrm/app/services/webhook_service.py

"""CRUD service for Webhook subscriptions and delivery."""

from __future__ import annotations

import hashlib
import hmac
import ipaddress
import json
import logging
import socket
import uuid
from datetime import UTC, datetime
from typing import Any
from urllib.parse import urlparse

import httpx
from sqlalchemy import cast, select
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.visibility import apply_visibility_filter, check_single_entity_access
from app.models.webhook import Webhook

logger = logging.getLogger(__name__)


def _validate_webhook_url(url: str) -> None:
    """Validate a webhook URL to prevent SSRF attacks.

    Blocks:
    - Non-http(s) schemes
    - Private/internal IP ranges (10.x, 172.16-31.x, 192.168.x, 127.x, 169.254.x, ::1)
    - Hostnames that resolve to private IPs (DNS rebinding)
    - Redirects (handled by httpx follow_redirects=False)
    """
    parsed = urlparse(url)

    # Protocol allowlist
    if parsed.scheme not in ("http", "https"):
        raise ValueError(f"Webhook URL must use http or https, got: {parsed.scheme}")

    hostname = parsed.hostname
    if not hostname:
        raise ValueError("Webhook URL has no hostname")

    # Check if hostname is an IP address
    try:
        ip = ipaddress.ip_address(hostname)
        if ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_reserved:
            raise ValueError(f"Webhook URL points to private/reserved IP: {ip}")
    except ValueError as exc:
        if "points to private" in str(exc) or "must use" in str(exc):
            raise
        # Not an IP — resolve hostname and check
        try:
            resolved = socket.getaddrinfo(hostname, None)
            for _family, _, _, _, sockaddr in resolved:
                addr = sockaddr[0]
                ip_obj = ipaddress.ip_address(addr)
                if ip_obj.is_private or ip_obj.is_loopback or ip_obj.is_link_local or ip_obj.is_reserved:
                    raise ValueError(f"Webhook hostname '{hostname}' resolves to private IP: {addr}") from exc
        except socket.gaierror:
            raise ValueError(f"Cannot resolve webhook hostname: {hostname}") from None


async def list_webhooks(
    db: AsyncSession,
    tenant_id: uuid.UUID,
    event: str | None = None,
    user_id: uuid.UUID | None = None,
    is_system_admin: bool = False,
) -> list[Webhook]:
    """List webhooks for a tenant, optionally filtered by event."""
    stmt = select(Webhook).where(
        Webhook.tenant_id == tenant_id,
    )
    if event:
        # Webhook.events is a JSONB array column (NOT a relationship):
        # events @> '["<event>"]' — JSONB containment instead of the invalid
        # relationship .any() call (same fix as webhook_dispatcher).
        stmt = stmt.where(cast(Webhook.events, JSONB).contains([event]))
    if user_id and not is_system_admin:
        stmt = await apply_visibility_filter(
            db, stmt, "webhook", Webhook, user_id, tenant_id, is_system_admin
        )
    stmt = stmt.order_by(Webhook.created_at.desc())
    result = await db.execute(stmt)
    return list(result.scalars().all())


async def get_webhook(
    db: AsyncSession,
    tenant_id: uuid.UUID,
    webhook_id: uuid.UUID,
    user_id: uuid.UUID | None = None,
    is_system_admin: bool = False,
) -> Webhook | None:
    """Get a single webhook by ID."""
    stmt = select(Webhook).where(
        Webhook.id == webhook_id,
        Webhook.tenant_id == tenant_id,
    )
    result = await db.execute(stmt)
    webhook = result.scalar_one_or_none()
    if webhook is None:
        return None
    if user_id and not is_system_admin:
        has_access = await check_single_entity_access(
            db, "webhook", webhook.id, user_id, tenant_id, "read", is_system_admin
        )
        if not has_access:
            raise PermissionError("No access")
    return webhook


async def create_webhook(
    db: AsyncSession,
    tenant_id: uuid.UUID,
    user_id: uuid.UUID,
    data: dict[str, Any],
    is_system_admin: bool = False,
) -> Webhook:
    """Create a new webhook subscription."""
    webhook = Webhook(
        tenant_id=tenant_id,
        url=data["url"],
        events=data["events"],
        secret=data.get("secret"),
        is_active=data.get("is_active", True),
        retry_count=data.get("retry_count", 3),
        timeout_seconds=data.get("timeout_seconds", 30),
        created_by=user_id,
        updated_by=user_id,
        owner_id=user_id,
    )
    db.add(webhook)
    await db.flush()
    await db.refresh(webhook)
    return webhook


async def update_webhook(
    db: AsyncSession,
    tenant_id: uuid.UUID,
    webhook_id: uuid.UUID,
    data: dict[str, Any],
    user_id: uuid.UUID | None = None,
    is_system_admin: bool = False,
) -> Webhook | None:
    """Update an existing webhook subscription."""
    webhook = await get_webhook(db, tenant_id, webhook_id)
    if webhook is None:
        return None

    if not is_system_admin:
        has_access = await check_single_entity_access(
            db, "webhook", webhook.id, user_id, tenant_id, "write", is_system_admin
        )
        if not has_access:
            raise PermissionError("No access")

    update_fields = ["url", "events", "secret", "is_active", "retry_count", "timeout_seconds"]
    for field in update_fields:
        if field in data:
            setattr(webhook, field, data[field])

    if user_id is not None:
        webhook.updated_by = user_id

    await db.flush()
    await db.refresh(webhook)
    return webhook


async def delete_webhook(
    db: AsyncSession,
    tenant_id: uuid.UUID,
    webhook_id: uuid.UUID,
    user_id: uuid.UUID | None = None,
    is_system_admin: bool = False,
) -> bool:
    """Delete a webhook subscription."""
    stmt = select(Webhook).where(
        Webhook.id == webhook_id,
        Webhook.tenant_id == tenant_id,
    )
    result = await db.execute(stmt)
    webhook = result.scalar_one_or_none()
    if webhook is None:
        return False

    if not is_system_admin:
        has_access = await check_single_entity_access(
            db, "webhook", webhook.id, user_id, tenant_id, "admin", is_system_admin
        )
        if not has_access:
            raise PermissionError("No access")

    await db.delete(webhook)
    await db.flush()
    return True


def _sign_payload(payload: bytes, secret: str) -> str:
    """Create HMAC-SHA256 signature for a payload."""
    return hmac.new(
        secret.encode("utf-8"),
        payload,
        hashlib.sha256,
    ).hexdigest()


async def send_webhook(
    webhook: Webhook,
    event_name: str,
    payload: dict[str, Any],
) -> dict[str, Any]:
    """Send a webhook HTTP POST with HMAC signature.

    Returns a dict with 'success' (bool), 'status_code' (int | None),
    and 'error' (str | None).
    """
    body = {
        "event": event_name,
        "payload": payload,
        "timestamp": datetime.now(UTC).isoformat().replace("+00:00", "Z"),
    }
    body_bytes = json.dumps(body, default=str).encode("utf-8")

    headers = {
        "Content-Type": "application/json",
        "User-Agent": "LeoCRM-Webhook/1.0",
    }

    if webhook.secret:
        signature = _sign_payload(body_bytes, webhook.secret)
        headers["X-Webhook-Signature"] = f"sha256={signature}"

    # SSRF protection: validate URL before sending
    try:
        _validate_webhook_url(webhook.url)
    except ValueError as exc:
        logger.warning("Webhook URL validation failed for %s: %s", webhook.url, exc)
        return {"success": False, "status_code": None, "error": f"URL validation failed: {exc}"}

    try:
        async with httpx.AsyncClient(
            timeout=webhook.timeout_seconds,
            follow_redirects=False,  # Prevent SSRF via redirect
        ) as client:
            response = await client.post(
                webhook.url,
                content=body_bytes,
                headers=headers,
            )
        return {
            "success": response.is_success,
            "status_code": response.status_code,
            "error": None,
        }
    except httpx.TimeoutException:
        return {"success": False, "status_code": None, "error": "timeout"}
    except httpx.RequestError as exc:
        return {"success": False, "status_code": None, "error": str(exc)}


async def retry_webhook(
    db: AsyncSession,
    tenant_id: uuid.UUID,
    webhook_id: uuid.UUID,
    event_name: str,
    payload: dict[str, Any],
) -> dict[str, Any]:
    """Retry sending a webhook with the given event and payload."""
    webhook = await get_webhook(db, tenant_id, webhook_id)
    if webhook is None:
        return {"success": False, "status_code": None, "error": "webhook_not_found"}
    return await send_webhook(webhook, event_name, payload)