import json
import logging
import os
import threading
from datetime import UTC, datetime, timezone
import requests
logger = logging.getLogger("agentforms.webhook")
logger.setLevel(logging.INFO)
# Ensure output appears in Docker logs (stderr)
if not logger.handlers:
_h = logging.StreamHandler(__import__("sys").stderr)
_h.setFormatter(logging.Formatter("[webhook] %(message)s"))
logger.addHandler(_h)
# Timeout for webhook delivery attempts (seconds)
WEBHOOK_TIMEOUT = 10
# Max retry attempts with exponential backoff
MAX_RETRIES = 5
# Base delay in minutes: 30, 60(capped), 60, 60, 60
BASE_RETRY_MINUTES = 30
def _sanitize_value(value):
"""Strip all HTML tags from a submission value to prevent XSS in webhooks."""
from bleach import clean as bleach_clean
if isinstance(value, str):
return bleach_clean(value, tags=[], strip=True)
return value
def _build_payload(site, submission, field_config=None):
"""Build webhook payload from site + submission data."""
from app.services.form_logic import evaluate_all_conditions
# Build field payload: hardcoded + dynamic
fields = {}
if submission.get("customer_name"):
fields["name"] = _sanitize_value(submission["customer_name"])
if submission.get("customer_phone"):
fields["phone"] = _sanitize_value(submission["customer_phone"])
if submission.get("customer_email"):
fields["email"] = _sanitize_value(submission["customer_email"])
if submission.get("customer_equipment"):
fields["equipment"] = _sanitize_value(submission["customer_equipment"])
if submission.get("customer_message"):
fields["message"] = _sanitize_value(submission["customer_message"])
# Merge dynamic data from JSON column (also sanitize string values)
if submission.get("data"):
try:
dynamic = json.loads(submission["data"])
sanitized_dynamic = {k: _sanitize_value(v) for k, v in dynamic.items()}
fields.update(sanitized_dynamic)
except (json.JSONDecodeError, TypeError):
pass
# If site has a custom field config, only include those fields
if field_config:
field_names = {f.get("name", f.get("key", "")) for f in field_config}
fields = {k: v for k, v in fields.items() if k in field_names}
# Build field metadata with conditional info
field_metadata = None
if field_config:
field_metadata = evaluate_all_conditions(field_config, fields)
payload = {
"event": "submission",
"site_id": site["id"],
"site_name": site.get("name", ""),
"submission_id": submission["id"],
"submitted_at": submission.get("submitted_at", datetime.now(UTC).isoformat()),
"fields": fields,
}
if field_metadata:
payload["field_metadata"] = {
"visible_fields": field_metadata.get("visible_fields", {}),
"computed_fields": field_metadata.get("computed_fields", {}),
"applied_conditions": field_metadata.get("applied_conditions", []),
}
return payload
def _deliver_webhook(url, payload, headers=None):
"""Deliver a single webhook HTTP POST. Returns (success, error_msg).
Accepts payload as either a dict (sent via json=) or a pre-serialised
JSON string (sent via data=). Custom headers merge with defaults —
custom keys override defaults.
"""
# SSRF protection — validate URL before sending
is_valid, validation_error = validate_webhook_url(url)
if not is_valid:
logger.warning("SSRF blocked: %s (%s)", url, validation_error)
return False, f"blocked: {validation_error}"
default_headers = {
"Content-Type": "application/json",
"X-AgentForms-Event": "submission",
}
if headers:
merged = {**default_headers, **headers}
else:
merged = dict(default_headers)
try:
if isinstance(payload, dict):
resp = requests.post(
url,
json=payload,
headers=merged,
timeout=WEBHOOK_TIMEOUT,
)
else:
# Already serialised string — use data= and strip json=
# (Content-Type should already be set by the formatter)
resp = requests.post(
url,
data=payload,
headers=merged,
timeout=WEBHOOK_TIMEOUT,
)
if 200 <= resp.status_code < 300:
return True, None
else:
return False, f"HTTP {resp.status_code}"
except requests.exceptions.Timeout:
return False, "timeout"
except requests.exceptions.ConnectionError as e:
return False, f"connection_error: {e}"
except Exception as e:
return False, str(e)
def send_webhook(site, submission, field_config=None):
"""Send legacy webhook notification for a new submission.
Called in a background thread so it doesn't block the API response.
Logs the attempt and enables retry for failures.
"""
from app.models import log_webhook_attempt
webhook_url = (site.get("webhook_url") or "").strip()
if not webhook_url:
return
webhook_enabled = site.get("webhook_enabled", 0)
if not webhook_enabled:
return
payload = _build_payload(site, submission, field_config)
success, error = _deliver_webhook(webhook_url, payload)
# Log the attempt for retry tracking
log_webhook_attempt(site["id"], submission["id"], webhook_url, success, error)
if success:
logger.info(
"Webhook delivered for site %s submission %s",
site["id"],
submission["id"],
)
else:
logger.warning(
"Webhook delivery failed for site %s submission %s: %s",
site["id"],
submission["id"],
error,
)
def fire_destination(destination, site, submission, field_config=None):
"""Fire a single integration destination.
Uses the integration formatter to build the correct payload and headers
for the destination type (webhook, slack, discord, telegram, etc.).
Logs the attempt and updates destination stats.
"""
from app.models import log_webhook_attempt, update_destination_stats
from app.services.integrations import format_payload
dest_id = destination.get("id")
dest_type = destination.get("type", "")
site_id = site["id"]
submission_id = submission["id"]
# Format payload via integration formatter
result = format_payload(dest_type, destination, site, submission, field_config)
if result is None:
err = f"unknown destination type: {dest_type}"
logger.error(
"Destination delivery failed for site %s submission %s dest %s: %s",
site_id,
submission_id,
dest_id,
err,
)
if dest_id:
update_destination_stats(dest_id, False, err)
return
url, payload, headers = result
if not url:
err = "missing URL"
logger.error(
"Destination delivery failed for site %s submission %s dest %s: %s",
site_id,
submission_id,
dest_id,
err,
)
if dest_id:
update_destination_stats(dest_id, False, err)
return
# Enrich headers with site/submission identifiers
enriched_headers = dict(headers)
enriched_headers["X-AgentForms-Site-ID"] = str(site_id)
enriched_headers["X-AgentForms-Submission-ID"] = str(submission_id)
success, error = _deliver_webhook(url, payload, headers=enriched_headers)
# Log the attempt for retry tracking
log_webhook_attempt(site_id, submission_id, url, success, error, destination_id=dest_id)
# Update destination delivery stats
if dest_id:
update_destination_stats(dest_id, success, error)
if success:
logger.info(
"Destination delivered [%s] for site %s submission %s",
dest_type,
site_id,
submission_id,
)
else:
logger.warning(
"Destination delivery failed [%s] for site %s submission %s: %s",
dest_type,
site_id,
submission_id,
error,
)
def retry_webhook(log_entry):
"""Retry a single failed webhook log entry.
If the entry has a destination_id, re-fetches the destination and uses
the integration formatter. Otherwise falls back to the legacy format.
Called by the background retry thread.
"""
from app.models import (
get_destination,
get_site_webhook,
get_submission,
mark_webhook_delivered,
release_webhook_retry,
)
site_id = log_entry["site_id"]
submission_id = log_entry["submission_id"]
webhook_url = log_entry["webhook_url"]
destination_id = log_entry.get("destination_id")
# Re-fetch submission data
submission = get_submission(submission_id)
if not submission:
logger.info(
"Webhook retry skipped for site %s submission %s: submission not found",
site_id,
submission_id,
)
release_webhook_retry(log_entry["id"], "delivered")
return
# ── Destination-aware retry ──
if destination_id:
destination = get_destination(destination_id)
if not destination or not destination.get("enabled", True):
logger.info(
"Webhook retry skipped for site %s submission %s: destination %s disabled or deleted",
site_id,
submission_id,
destination_id,
)
release_webhook_retry(log_entry["id"], "delivered")
return
# Re-format using the integration formatter
from app.services.integrations import format_payload
result = format_payload(destination["type"], destination, {"id": site_id, "name": ""}, submission)
if result is None:
err = f"unknown formatter for type {destination['type']}"
logger.warning(
"Webhook retry failed (attempt %d) for site %s submission %s: %s",
log_entry["attempt_count"] + 1,
site_id,
submission_id,
err,
)
release_webhook_retry(log_entry["id"], "failed", err)
return
url, payload, headers = result
if not url:
url = webhook_url # fallback to stored URL
success, error = _deliver_webhook(url, payload, headers=headers)
if success:
release_webhook_retry(log_entry["id"], "delivered")
logger.info(
"Webhook retry delivered (attempt %d) for site %s submission %s [dest %s]",
log_entry["attempt_count"] + 1,
site_id,
submission_id,
destination_id,
)
else:
# Log new failure attempt for next retry cycle
from app.models import log_webhook_attempt
log_webhook_attempt(site_id, submission_id, url, False, error, destination_id=destination_id)
release_webhook_retry(log_entry["id"], "failed", error)
logger.warning(
"Webhook retry failed (attempt %d) for site %s submission %s: %s",
log_entry["attempt_count"] + 1,
site_id,
submission_id,
error,
)
return
# ── Legacy retry (no destination_id) ──
site_cfg = get_site_webhook(site_id)
if not site_cfg or not site_cfg.get("webhook_url"):
logger.info(
"Webhook retry skipped for site %s submission %s: webhook disabled or site deleted",
site_id,
submission_id,
)
release_webhook_retry(log_entry["id"], "delivered")
return
site_dict = {"id": site_id, "name": "", "webhook_url": webhook_url, "webhook_enabled": 1}
payload = _build_payload(site_dict, submission)
success, error = _deliver_webhook(webhook_url, payload)
if success:
release_webhook_retry(log_entry["id"], "delivered")
logger.info(
"Webhook retry delivered (attempt %d) for site %s submission %s",
log_entry["attempt_count"] + 1,
site_id,
submission_id,
)
else:
from app.models import log_webhook_attempt
log_webhook_attempt(site_id, submission_id, webhook_url, False, error)
release_webhook_retry(log_entry["id"], "failed", error)
logger.warning(
"Webhook retry failed (attempt %d) for site %s submission %s: %s",
log_entry["attempt_count"] + 1,
site_id,
submission_id,
error,
)
def _retry_worker():
"""Background thread that periodically retries failed webhooks."""
import threading
import time
from datetime import datetime, timedelta
from app.db import get_db
from app.models import claim_webhook_retries, get_permanently_failed_webhooks
# Alert deduplication - DB-backed (survives restarts)
_ALERTED_MAX_AGE = 86400 # 24 hours
thread_name = f"retry-{threading.current_thread().name}"
while True:
conn = None
try:
conn = get_db()
# 1. Retry pending webhooks
pending = claim_webhook_retries(thread_name, max_retries=MAX_RETRIES)
for entry in pending:
retry_webhook(entry)
# 2. Alert on permanently failed webhooks (MAX_RETRIES exhausted)
permanently_failed = get_permanently_failed_webhooks(max_retries=MAX_RETRIES)
if permanently_failed:
# Deduplicate alerts — only alert once per submission per 24h (DB-backed)
unalerted = []
for entry in permanently_failed:
sub_key = f"{entry['site_id']}:{entry['submission_id']}"
# Check DB for existing alert
dedup_row = conn.execute(
"SELECT alerted_at FROM alert_dedup WHERE alert_key = ?", (sub_key,)
).fetchone()
if dedup_row:
# Check if alert is still within 24h window
alerted_ts = datetime.fromisoformat(dedup_row["alerted_at"]).timestamp()
if time.time() - alerted_ts < _ALERTED_MAX_AGE:
continue # already alerted recently
unalerted.append(entry)
# Upsert dedup record
conn.execute(
"INSERT OR REPLACE INTO alert_dedup (alert_key, alerted_at) VALUES (?, ?)",
(sub_key, datetime.now(UTC).isoformat()),
)
# Cleanup stale dedup entries (>24h old)
cutoff = datetime.now(UTC) - timedelta(seconds=_ALERTED_MAX_AGE)
conn.execute("DELETE FROM alert_dedup WHERE alerted_at < ?", (cutoff.isoformat(),))
if unalerted:
from app.services.alerting import alert
alert.warning(
f"Webhook delivery permanently failed: {len(unalerted)} submission(s)",
max_retries=MAX_RETRIES,
submissions=[
f"site={e['site_id']} sub={e['submission_id']} attempts={e['attempt_count']}"
for e in unalerted
],
)
logger.warning("Permanently failed webhooks: %s entries", len(unalerted))
if conn:
conn.commit()
except Exception as e:
logger.error("Webhook retry worker error: %s", e)
finally:
if conn:
conn.close()
time.sleep(60) # Check every minute
def start_retry_worker():
"""Start the webhook retry background thread (call once on app init)."""
thread = threading.Thread(target=_retry_worker, daemon=True, name="webhook-retry")
thread.start()
logger.info("Webhook retry worker started")
def _enqueue_webhook_delivery(site, submission, field_config=None):
"""Queue webhook delivery via RQ (persists in Redis).
Falls back to background thread if Redis is unavailable.
"""
redis_url = os.environ.get("REDIS_URL")
if redis_url:
try:
import redis as redis_lib
from rq import Queue
r = redis_lib.from_url(redis_url)
r.ping()
q = Queue("webhooks", connection=r)
q.enqueue(
_fire_all_destinations,
args=(site, submission, field_config),
job_timeout=120,
result_ttl=3600,
)
return True
except Exception as e:
logger.warning("RQ unavailable, falling back to thread: %s", e)
# Fallback: background thread (original behavior)
thread = threading.Thread(
target=_fire_all_destinations,
args=(site, submission, field_config),
daemon=True,
)
thread.start()
return False
def fire_webhook(site, submission, field_config=None):
"""Fire webhooks for all destinations + legacy webhook URL via RQ queue.
Returns immediately. Delivery is persisted in Redis and survives
app restarts.
Priority order:
1. Enabled integration destinations (via get_site_destinations).
2. Legacy webhook_url if no destinations configured.
"""
_enqueue_webhook_delivery(site, submission, field_config)
def _fire_all_destinations(site, submission, field_config=None):
"""Synchronous: deliver to all enabled destinations and/or legacy webhook."""
from app.models import get_site_destinations
site_id = site["id"]
# 1. Try integration destinations first
destinations = get_site_destinations(site_id)
enabled_destinations = [d for d in destinations if d.get("enabled", True)]
if enabled_destinations:
for dest in enabled_destinations:
try:
fire_destination(dest, site, submission, field_config)
except Exception as e:
logger.error(
"Uncaught error firing destination %s for site %s submission %s: %s",
dest.get("id"),
site_id,
submission["id"],
e,
)
# 2. Fallback to legacy webhook if no destinations configured
if not enabled_destinations:
send_webhook(site, submission, field_config)
# ─── PII Redaction for Logs ─────────────────────────────────────────────────────
import re
_PII_PATTERNS = [
re.compile(r"[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}", re.IGNORECASE), # Email
re.compile(r"\b\d{3}[-.]?\d{3}[-.]?\d{4}\b"), # Phone
re.compile(r"(?://)([a-zA-Z0-9.-]+)(:[0-9]+)?(/.*)?"), # URL domain
]
def redact_webhook_url(url):
"""Redact sensitive parts of a webhook URL for safe logging."""
if not url:
return url
try:
from urllib.parse import urlparse, urlunparse
parsed = urlparse(url)
# Keep scheme, domain, port, path - redact query params and auth
redacted = urlunparse(
(
parsed.scheme,
parsed.netloc,
parsed.path,
"", # params
"", # query
"", # fragment
)
)
return redacted
except Exception:
return url
def redact_pii(text):
"""Redact PII patterns from log text."""
if not text:
return text
# Redact emails
text = re.sub(r"[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}", "[EMAIL]", text, flags=re.IGNORECASE)
# Redact phones
text = re.sub(r"\b\d{3}[-.]?\d{3}[-.]?\d{4}\b", "[PHONE]", text)
# Redact URL auth/query/fragment
text = re.sub(r"://[^@/]+@", "://[AUTH]@", text)
text = re.sub(r"\?[^\s]*", "?[QUERY]", text)
text = re.sub(r"#[^\s]*", "#[FRAGMENT]", text)
return text
# ─── SSRF Protection ─────────────────────────────────────────────────────────────
import ipaddress
from urllib.parse import urlparse
# IP ranges to block (RFC 1918, link-local, loopback, etc.)
_BLOCKED_RANGES = [
ipaddress.ip_network("10.0.0.0/8"), # Private
ipaddress.ip_network("172.16.0.0/12"), # Private
ipaddress.ip_network("192.168.0.0/16"), # Private
ipaddress.ip_network("127.0.0.0/8"), # Loopback
ipaddress.ip_network("0.0.0.0/8"), # Current
ipaddress.ip_network("100.64.0.0/10"), # CGNAT
ipaddress.ip_network("169.254.0.0/16"), # Link-local
ipaddress.ip_network("224.0.0.0/4"), # Multicast
ipaddress.ip_network("240.0.0.0/4"), # Reserved
ipaddress.ip_network("::1/128"), # IPv6 loopback
ipaddress.ip_network("fc00::/7"), # IPv6 unique local
ipaddress.ip_network("fe80::/10"), # IPv6 link-local
]
def _is_blocked_ip(ip_str):
"""Check if an IP address falls within blocked ranges."""
try:
ip = ipaddress.ip_address(ip_str)
for network in _BLOCKED_RANGES:
if ip in network:
return True
return False
except ValueError:
return True # Treat invalid IPs as blocked
def validate_webhook_url(url):
"""Validate a webhook URL for SSRF protection.
Returns (is_valid, error_message).
"""
if not url:
return False, "Empty URL"
try:
parsed = urlparse(url)
except Exception:
return False, "Invalid URL format"
# Only allow http/https
if parsed.scheme not in ("http", "https"):
return False, f"Scheme {parsed.scheme} not allowed. Use https:// for external URLs."
# Check hostname
hostname = parsed.hostname
if not hostname:
return False, "No hostname in URL"
# Block metadata endpoints (cloud provider)
metadata_endpoints = [
"169.254.169.254", # AWS/GCP/Azure metadata
"metadata.google.internal",
"metadata.azure.com",
]
if hostname in metadata_endpoints:
return False, "Metadata endpoint access blocked"
# Resolve hostname and check IP (both IPv4 and IPv6)
import socket
resolved = False
for af in (socket.AF_INET, socket.AF_INET6):
try:
addr_info = socket.getaddrinfo(hostname, None, af, socket.SOCK_STREAM)
for info in addr_info:
ip = info[4][0]
if _is_blocked_ip(ip):
return False, f"Blocked IP address: {ip} ({hostname})"
resolved = True
except socket.gaierror:
pass # IPv6 may not be available for all hosts — skip
except socket.herror:
return False, f"Invalid hostname: {hostname}"
if not resolved:
return False, f"Could not resolve hostname: {hostname}"
return True, None