Terminal / tools /trigger_webhook.py
Pulka
sync: 100 file da Baida98/AI@440ef81c (2026-06-20 18:35 UTC)
2949015 verified
Raw
History Blame Contribute Delete
7.05 kB
"""
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,
}