""" backend/tools/trigger_webhook.py — P17-F4: Generic webhook trigger tool. Permette all'agente di inviare notifiche / callback a URL esterni (Zapier, Make, n8n, webhook Discord/Slack, CI/CD, ecc.) con payload arbitrario. Registrato nel TOOL_REGISTRY come "trigger_webhook". Sicurezza: - URL allowlist via WEBHOOK_ALLOWED_HOSTS env (CSV). Vuoto = tutti consentiti (dev). - Timeout massimo 10s — mai blocca il loop principale. - Payload massimo 32KB — protegge da exfiltration involontaria. - Metodi consentiti: POST, GET, PUT (default POST). - Rate limit per-host via sliding window: WEBHOOK_RATE_LIMIT_MAX (default 10) / WEBHOOK_RATE_LIMIT_WINDOW (default 60s). """ from __future__ import annotations import json import logging import os import asyncio import time from collections import defaultdict, deque from typing import Any from urllib.parse import urlparse import httpx _logger = logging.getLogger("agente_ai.tools.trigger_webhook") _ALLOWED_HOSTS_ENV = os.getenv("WEBHOOK_ALLOWED_HOSTS", "") _ALLOWED_HOSTS: set[str] = ( {h.strip().lower() for h in _ALLOWED_HOSTS_ENV.split(",") if h.strip()} if _ALLOWED_HOSTS_ENV else set() ) _MAX_PAYLOAD_BYTES = 32_768 # 32 KB _TIMEOUT_S = 10.0 # ── P17-F4 rate-limit guard (sliding window per-host) ───────────────────────── _RATE_LIMIT_MAX: int = int(os.getenv("WEBHOOK_RATE_LIMIT_MAX", "10")) # max calls per window _RATE_LIMIT_WIN: float = float(os.getenv("WEBHOOK_RATE_LIMIT_WINDOW", "60")) # window seconds _rate_lock = asyncio.Lock() _rate_windows: dict[str, deque[float]] = defaultdict(deque) # host → timestamps async def _check_rate_limit(host: str) -> bool: """Sliding window rate limiter per host. Ritorna True se entro il limite.""" async with _rate_lock: now = time.monotonic() win = _rate_windows[host] while win and win[0] < now - _RATE_LIMIT_WIN: win.popleft() if len(win) >= _RATE_LIMIT_MAX: return False win.append(now) return True def _check_host(url: str) -> bool: """Verifica che l'host sia nell'allowlist (se configurata).""" if not _ALLOWED_HOSTS: return True # dev mode — tutti consentiti try: host = urlparse(url).hostname or "" return host.lower() in _ALLOWED_HOSTS except Exception: return False async def trigger_webhook( url: str, payload: dict[str, Any] | str | None = None, method: str = "POST", headers: dict[str, str] | None = None, timeout: float = _TIMEOUT_S, ) -> dict[str, Any]: """ Trigger an HTTP webhook with an arbitrary payload. Args: url: Target URL (https://...) payload: JSON-serializable dict, raw string, or None method: HTTP method — POST (default), GET, PUT headers: Additional HTTP headers (optional) timeout: Request timeout in seconds (max 10) Returns: {"ok": bool, "status": int, "body": str, "error": str|None} """ method = method.upper() if method not in ("POST", "GET", "PUT"): return {"ok": False, "status": 0, "body": "", "error": f"Metodo non consentito: {method}"} if not url.startswith(("https://", "http://")): return {"ok": False, "status": 0, "body": "", "error": "URL deve iniziare con http(s)://"} if not _check_host(url): return {"ok": False, "status": 0, "body": "", "error": "Host non nell'allowlist WEBHOOK_ALLOWED_HOSTS"} # P17-F4: rate-limit guard — sliding window per-host _rl_host = urlparse(url).hostname or url if not await _check_rate_limit(_rl_host): _logger.warning( "trigger_webhook rate limit superato: %s (%s req/%ss)", _rl_host, _RATE_LIMIT_MAX, int(_RATE_LIMIT_WIN), ) return { "ok": False, "status": 429, "body": "", "error": f"Rate limit ({_RATE_LIMIT_MAX} chiamate/{_RATE_LIMIT_WIN:.0f}s) superato per {_rl_host}", } # Serializza payload body_bytes: bytes | None = None req_headers: dict[str, str] = {"User-Agent": "agente-ai/1.0"} if headers: req_headers.update(headers) if payload is not None: if isinstance(payload, dict): raw = json.dumps(payload, ensure_ascii=False) req_headers.setdefault("Content-Type", "application/json") else: raw = str(payload) req_headers.setdefault("Content-Type", "text/plain") body_bytes = raw.encode("utf-8")[:_MAX_PAYLOAD_BYTES] _effective_timeout = min(float(timeout), _TIMEOUT_S) try: async with httpx.AsyncClient(timeout=_effective_timeout, follow_redirects=True) as client: if method == "GET": resp = await client.get(url, headers=req_headers) elif method == "PUT": resp = await client.put(url, content=body_bytes, headers=req_headers) else: resp = await client.post(url, content=body_bytes, headers=req_headers) body_str = resp.text[:2000] ok = resp.is_success _logger.info( "trigger_webhook %s %s → %d (%s)", method, url[:80], resp.status_code, "OK" if ok else "FAIL", ) return {"ok": ok, "status": resp.status_code, "body": body_str, "error": None} except httpx.TimeoutException: _logger.warning("trigger_webhook timeout (%.0fs): %s", _effective_timeout, url[:80]) return {"ok": False, "status": 0, "body": "", "error": f"timeout dopo {_effective_timeout:.0f}s"} except Exception as exc: _logger.warning("trigger_webhook error: %s", exc) return {"ok": False, "status": 0, "body": "", "error": str(exc)[:200]} # ── Tool registry descriptor ─────────────────────────────────────────────────── TOOL_DESCRIPTOR = { "name": "trigger_webhook", "description": ( "Invia una richiesta HTTP (POST/GET/PUT) a un URL esterno con payload JSON. " "Usa per notificare webhook Zapier/Make/n8n, CI/CD, Discord, Slack, ecc. " "Restituisce status HTTP + risposta body (max 2000 chars). " "Rate limit: max 10 call/60s per host (configurabile via WEBHOOK_RATE_LIMIT_MAX)." ), "parameters": { "type": "object", "properties": { "url": {"type": "string", "description": "URL destinazione (https://...)"}, "payload": {"type": "object", "description": "Payload JSON da inviare"}, "method": {"type": "string", "enum": ["POST", "GET", "PUT"], "default": "POST"}, "headers": {"type": "object", "description": "Header HTTP aggiuntivi"}, "timeout": {"type": "number", "description": "Timeout secondi (max 10)", "default": 10}, }, "required": ["url"], }, "fn": trigger_webhook, }