import os
import threading
import time
from html import escape as html_escape
from flask import Blueprint, jsonify, make_response, request
from app.models import (
accept_submission,
add_submission,
find_site_by_token,
get_auto_generate_template,
get_submission,
get_user,
get_user_usage,
list_submissions,
parse_site_fields,
)
from app.services.form_logic import compute_all_fields, validate_submission
from app.services.logging import log
api_bp = Blueprint("api", __name__)
# Load SMTP config
# Config
APP_BASE_URL = os.environ.get("APP_BASE_URL", "https://agentforms.io").rstrip("/")
# CORS allowlist — comma-separated origins (empty = allow any http/https origin for backwards compat)
CORS_ALLOWED_ORIGINS = [
o.strip() for o in os.environ.get("CORS_ALLOWED_ORIGINS", "").split(",") if o.strip()
] or None # None means "allow any http/https" for backwards compatibility
SMTP_HOST = os.environ.get("SMTP_HOST", "")
SMTP_PORT = int(os.environ.get("SMTP_PORT", "587"))
SMTP_USER = os.environ.get("SMTP_USER", "")
SMTP_PASS = os.environ.get("SMTP_PASS", "")
SMTP_FROM = os.environ.get("SMTP_FROM", "")
# ─── CORS ────────────────────────────────────────────────────────────────────
def add_cors_headers(response):
"""Add CORS and security headers for all responses."""
origin = request.headers.get("Origin")
if origin:
# Validate it's a proper URL scheme (http/https only, no javascript:)
if origin.startswith(("http://", "https://")):
# If allowlist is configured, only permit listed origins
if CORS_ALLOWED_ORIGINS is None or origin in CORS_ALLOWED_ORIGINS:
response.headers["Access-Control-Allow-Origin"] = origin
else:
# Origin not in allowlist — don't set CORS header (blocks the request)
pass
else:
response.headers["Access-Control-Allow-Origin"] = "*"
else:
response.headers["Access-Control-Allow-Origin"] = "*"
response.headers["Access-Control-Allow-Methods"] = "POST, OPTIONS"
response.headers["Access-Control-Allow-Headers"] = "Content-Type, Authorization"
response.headers["Access-Control-Max-Age"] = "86400"
# Security headers
response.headers["Content-Security-Policy"] = "default-src 'none'; frame-ancestors 'none';"
response.headers["X-Frame-Options"] = "DENY"
response.headers["X-Content-Type-Options"] = "nosniff"
response.headers["Referrer-Policy"] = "strict-origin-when-cross-origin"
return response
api_bp.after_request(add_cors_headers)
# ─── Rate Limiting (in-memory token bucket) ─────────────────────────────────
class TokenBucket:
"""Simple in-memory token bucket for rate limiting.
Per-IP: max `capacity` requests, refill `refill_rate` tokens/second.
Not distributed — sufficient for single-instance deployment.
"""
def __init__(self, capacity, refill_rate):
self.capacity = capacity
self.refill_rate = refill_rate
self._buckets = {}
self._lock = threading.Lock()
def consume(self, key, tokens=1):
now = time.monotonic()
with self._lock:
if key not in self._buckets:
self._buckets[key] = {"tokens": self.capacity, "last": now}
bucket = self._buckets[key]
elapsed = now - bucket["last"]
bucket["tokens"] = min(self.capacity, bucket["tokens"] + elapsed * self.refill_rate)
bucket["last"] = now
if bucket["tokens"] >= tokens:
bucket["tokens"] -= tokens
return True
return False
def cleanup(self, max_age=3600):
"""Remove stale entries older than max_age seconds."""
now = time.monotonic()
with self._lock:
stale = [k for k, v in self._buckets.items() if now - v["last"] > max_age]
for k in stale:
del self._buckets[k]
def reset(self):
"""Clear all buckets. Intended for tests."""
with self._lock:
self._buckets.clear()
# Configurable via environment: RATE_LIMIT_REQUESTS (burst), RATE_LIMIT_REFILL (per second)
_RATE_LIMIT = TokenBucket(
capacity=int(os.environ.get("RATE_LIMIT_REQUESTS", "30")),
refill_rate=float(os.environ.get("RATE_LIMIT_REFILL", "1.0")),
)
# Per-site/token rate limiting (separate from per-IP global limit)
# Prevents a single form from being spam-submitted even from different IPs
_SITE_RATE_LIMIT = TokenBucket(
capacity=int(os.environ.get("SITE_RATE_LIMIT_REQUESTS", "20")),
refill_rate=float(os.environ.get("SITE_RATE_LIMIT_REFILL", "0.5")),
)
class DynamicTokenBucket:
"""Per-site token bucket that reads capacity/refill from site config on each call.
Each key (site_token) maintains its own bucket state, but the capacity and
refill_rate are looked up dynamically via the settings_fn callback. This lets
site owners configure their own rate limits via the dashboard.
"""
def __init__(self, settings_fn):
"""settings_fn(key) -> (capacity, refill_rate) or None to skip."""
self._settings_fn = settings_fn
self._buckets = {}
self._lock = threading.Lock()
def consume(self, key, tokens=1):
settings = self._settings_fn(key)
if settings is None:
return True # rate limiting disabled
capacity, refill_rate = settings
now = time.monotonic()
with self._lock:
if key not in self._buckets:
self._buckets[key] = {"tokens": capacity, "last": now}
bucket = self._buckets[key]
elapsed = now - bucket["last"]
bucket["tokens"] = min(capacity, bucket["tokens"] + elapsed * refill_rate)
bucket["last"] = now
if bucket["tokens"] >= tokens:
bucket["tokens"] -= tokens
return True
return False
def cleanup(self, max_age=3600):
"""Remove stale entries older than max_age seconds."""
now = time.monotonic()
with self._lock:
stale = [k for k, v in self._buckets.items() if now - v["last"] > max_age]
for k in stale:
del self._buckets[k]
def reset(self):
"""Clear all buckets. Intended for tests."""
with self._lock:
self._buckets.clear()
def _get_site_rate_settings(token):
"""Look up per-site rate limit settings from the DB."""
site = find_site_by_token(token)
if not site:
return None
if not site.get("rate_limit_enabled", True):
return None # disabled by owner
return (
site.get("rate_limit_burst", 20),
site.get("rate_limit_refill", 0.5),
)
_DYNAMIC_SITE_RATE_LIMIT = DynamicTokenBucket(_get_site_rate_settings)
# Periodic cleanup of stale rate limiter entries
_RATE_CLEANUP_INTERVAL = 300 # 5 minutes
_last_cleanup = 0
def _maybe_cleanup_rate_limiter():
"""Periodically clean up stale rate limiter entries."""
global _last_cleanup
now = time.monotonic()
if now - _last_cleanup < _RATE_CLEANUP_INTERVAL:
return
_last_cleanup = now
_RATE_LIMIT.cleanup(max_age=3600)
_SITE_RATE_LIMIT.cleanup(max_age=3600)
_DYNAMIC_SITE_RATE_LIMIT.cleanup(max_age=3600)
def get_client_ip():
"""Extract client IP, respecting X-Forwarded-For (Cloudflare Tunnel)."""
forwarded = request.headers.get("X-Forwarded-For")
if forwarded:
return forwarded.split(",")[0].strip()
return request.remote_addr or "unknown"
# ─── Email Notification ─────────────────────────────────────────────────────
def _sanitize_email_field(value):
"""Strip newlines and carriage returns to prevent header injection."""
if not value:
return ""
return value.replace("\r", "").replace("\n", "").strip()
def _sanitize_submission_data(data):
"""Strip all HTML tags from string values in submission data to prevent XSS."""
from bleach import clean as bleach_clean
sanitized = {}
for key, value in data.items():
if isinstance(value, str):
sanitized[key] = bleach_clean(value, tags=[], strip=True)
else:
sanitized[key] = value
return sanitized
def send_notification(site, submission, field_config=None):
"""Send email notification to site owner with tracking."""
if not SMTP_HOST:
log.info("smtp", "Not configured, skipping")
return False
import json
import secrets
import smtplib
import ssl
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
customer_email = _sanitize_email_field(submission.get("customer_email", ""))
customer_name = _sanitize_email_field(submission.get("customer_name", "Unknown"))
# Generate tracking token for this notification
tracking_token = secrets.token_urlsafe(32)
msg = MIMEMultipart("alternative")
msg["From"] = site.get("smtp_from") or SMTP_FROM
msg["To"] = site["owner_email"]
msg["Subject"] = f"New Form Submission: {customer_name}"
if customer_email:
msg["Reply-To"] = f"{customer_name} <{customer_email}>"
# Build plain text body
lines = [f"New form submission from {site['name']}:", ""]
lines.append(f"Name: {submission.get('customer_name', 'N/A')}")
lines.append(f"Phone: {submission.get('customer_phone', 'N/A')}")
lines.append(f"Email: {submission.get('customer_email', 'N/A')}")
lines.append(f"Equipment: {submission.get('customer_equipment', 'N/A')}")
lines.append(f"\nMessage:\n{submission.get('customer_message', 'N/A')}")
# Add dynamic fields
if submission.get("data"):
try:
dynamic = json.loads(submission["data"])
if dynamic:
lines.append("\nAdditional Fields:")
for key, value in dynamic.items():
lines.append(f"{key}: {value}")
except (json.JSONDecodeError, TypeError):
pass
lines.append(f"\nSubmitted: {submission.get('submitted_at', 'N/A')}")
plain_body = "\n".join(lines)
# Build HTML body with tracking pixel
html_lines = [
"<html><body style='font-family: Arial, sans-serif; padding: 20px;'>",
f"<h2 style='color: #2563eb;'>New Form Submission from {html_escape(site['name'])}</h2>",
"<table style='border-collapse: collapse; width: 100%; max-width: 600px;'>",
f"<tr><td style='padding: 8px; font-weight: bold; width: 120px;'>Name:</td><td style='padding: 8px;'>{html_escape(str(submission.get('customer_name') or 'N/A'))}</td></tr>",
f"<tr><td style='padding: 8px; font-weight: bold;'>Phone:</td><td style='padding: 8px;'>{html_escape(str(submission.get('customer_phone') or 'N/A'))}</td></tr>",
f"<tr><td style='padding: 8px; font-weight: bold;'>Email:</td><td style='padding: 8px;'>{html_escape(str(submission.get('customer_email') or 'N/A'))}</td></tr>",
f"<tr><td style='padding: 8px; font-weight: bold;'>Equipment:</td><td style='padding: 8px;'>{html_escape(str(submission.get('customer_equipment') or 'N/A'))}</td></tr>",
f"<tr><td style='padding: 8px; font-weight: bold;'>Message:</td><td style='padding: 8px;'>{html_escape(str(submission.get('customer_message') or 'N/A'))}</td></tr>",
]
# Add dynamic fields to HTML
if submission.get("data"):
try:
dynamic = json.loads(submission["data"])
if dynamic:
html_lines.append(
"<tr><td colspan='2' style='padding: 12px 8px 4px; font-weight: bold; border-top: 1px solid #e5e7eb;'>Additional Fields:</td></tr>"
)
for key, value in dynamic.items():
html_lines.append(
f"<tr><td style='padding: 8px; font-weight: bold; width: 120px;'>{html_escape(str(key))}:</td><td style='padding: 8px;'>{html_escape(str(value) or '')}</td></tr>"
)
except (json.JSONDecodeError, TypeError):
pass
html_lines.extend(
[
"<tr><td colspan='2' style='padding: 12px 8px 4px; font-weight: bold; border-top: 1px solid #e5e7eb;'>Submitted:</td></tr>",
f"<tr><td colspan='2' style='padding: 8px; color: #6b7280;'>{html_escape(str(submission.get('submitted_at') or 'N/A'))}</td></tr>",
"</table>",
# Tracking pixel (1x1 transparent GIF)
f'<img src="{APP_BASE_URL}/tracking/open/{tracking_token}" width="1" height="1" style="display:none;" alt=""/>',
"</body></html>",
]
)
html_body = "\n".join(html_lines)
msg.attach(MIMEText(plain_body, "plain"))
msg.attach(MIMEText(html_body, "html"))
try:
context = ssl.create_default_context()
with smtplib.SMTP(SMTP_HOST, SMTP_PORT, timeout=15) as server:
server.ehlo()
# Only login if credentials are configured (e.g., SendGrid)
if SMTP_USER and SMTP_PASS:
server.starttls(context=context)
server.ehlo()
server.login(SMTP_USER, SMTP_PASS)
server.send_message(msg)
log.info("smtp", f"Sent to {site['owner_email']}", recipient=site["owner_email"])
# Log the email with tracking token
try:
from app.model_email_campaigns import log_notification_email
log_notification_email(
user_id=site.get("user_id"),
submission_id=submission.get("id"),
to_email=site["owner_email"],
tracking_token=tracking_token,
)
except Exception as log_err:
log.warning("email_log", f"Failed to log notification: {log_err}")
return True
except Exception as e:
log.error("smtp", f"Failed: {e}", error=str(e))
return False
def _send_customer_email(site, submission, pdf_path):
"""Send the auto-generated document PDF to the customer via email."""
if not SMTP_HOST:
log.info("smtp", "Not configured, skipping customer email")
return False
import os
import smtplib
import ssl
from email.mime.application import MIMEApplication
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
customer_email = _sanitize_email_field(submission.get("customer_email", ""))
if not customer_email:
log.warning("auto-gen", "No customer email, skipping customer email")
return False
customer_name = _sanitize_email_field(submission.get("customer_name", "there"))
msg = MIMEMultipart()
msg["From"] = site.get("smtp_from") or SMTP_FROM
msg["To"] = customer_email
msg["Subject"] = f"Document from {site['name']}"
body_lines = [
f"Hello {customer_name},",
"",
f"Thank you for your submission to {site['name']}.",
"",
"Please find your document attached.",
"",
"If you have any questions, please don't hesitate to reach out.",
"",
f"Best regards,\n{site['name']}",
]
msg.attach(MIMEText("\n".join(body_lines), "plain"))
# Attach PDF
if pdf_path and os.path.exists(pdf_path):
with open(pdf_path, "rb") as f:
pdf_part = MIMEApplication(f.read(), _subtype="pdf")
pdf_part.add_header(
"Content-Disposition", "attachment", filename=f"document-{submission.get('id', 'invoice')}.pdf"
)
msg.attach(pdf_part)
try:
context = ssl.create_default_context()
with smtplib.SMTP(SMTP_HOST, SMTP_PORT, timeout=15) as server:
server.ehlo()
if SMTP_USER and SMTP_PASS:
server.starttls(context=context)
server.ehlo()
server.login(SMTP_USER, SMTP_PASS)
server.send_message(msg)
log.info("auto-gen", f"Sent document to {customer_email}", recipient=customer_email)
return True
except Exception as e:
log.error("auto-gen", f"Email to customer failed: {e}", error=str(e))
return False
def _auto_generate_document(site, submission, field_config):
"""Generate document using auto-generate template (runs in background thread)."""
try:
from app.models import store_document
from app.services import documents as doc_service
template = get_auto_generate_template(site["id"])
if not template:
log.info("auto-gen", f"No auto-generate template for site {site['id']}", site_id=site["id"])
return
submission_data = dict(submission)
# Render the document
result = doc_service.render_document(
template_id=template["id"],
submission_id=submission["id"],
template_data=template,
submission_data=submission_data,
style=template.get("style_config", {}),
)
# Save PDF to disk
pdf_path = doc_service.save_pdf(result["pdf_bytes"], result["document_id"])
# Store document record
store_document(
user_id=site["user_id"],
site_id=site["id"],
submission_id=submission["id"],
template_id=template["id"],
document_type=template.get("document_type", "invoice"),
document_id=result["document_id"],
data=result["data"],
pdf_path=pdf_path,
expires_at=None,
)
log.info(
"auto-gen",
f"Document {result['document_id']} generated for submission {submission['id']}",
document_id=result["document_id"],
submission_id=submission["id"],
)
# Send to customer
_send_customer_email(site, submission, pdf_path)
except Exception as e:
log.error("auto-gen", f"Error generating document: {e}", error=str(e))
# Never block the submission — just log the error
@api_bp.route("/api/submit", methods=["POST", "OPTIONS"])
def submit():
"""Public endpoint for client sites to POST form data."""
# Handle CORS preflight
if request.method == "OPTIONS":
return make_response("", 204)
# Content type validation — only accept JSON or form-encoded
ct = request.content_type or ""
if not any(t in ct for t in ("application/json", "application/x-www-form-urlencoded")):
log.warning("submit", f"Rejected unsupported content type: {ct} from {get_client_ip()}")
return jsonify(
{"error": "Unsupported content type. Use application/json or application/x-www-form-urlencoded."}
), 415
# Max payload size: 100KB — prevent resource exhaustion
max_content_length = 100 * 1024
if request.content_length and request.content_length > max_content_length:
log.warning("submit", f"Payload too large: {request.content_length} bytes from {get_client_ip()}")
return jsonify({"error": "Payload too large. Maximum 100KB."}), 413
# Rate limiting — per-IP
_maybe_cleanup_rate_limiter() # Periodic cleanup of stale entries
client_ip = get_client_ip()
if not _RATE_LIMIT.consume(client_ip):
return jsonify({"error": "Rate limit exceeded. Try again in a moment."}), 429
token = (
request.args.get("token")
or (request.json.get("token") if request.is_json else None)
or request.form.get("token")
)
if not token:
return jsonify({"error": "Missing token"}), 400
site = find_site_by_token(token)
if not site:
return jsonify({"error": "Invalid token"}), 404
# Per-site rate limiting — uses site's own configurable settings from the DB
if not _DYNAMIC_SITE_RATE_LIMIT.consume(token):
return jsonify({"error": "Too many submissions for this form. Try again shortly."}), 429
# Accept both form data and JSON
if request.is_json:
data = request.json
else:
data = request.form.to_dict()
# Sanitize all string values — strip HTML tags to prevent XSS in DB/webhooks/emails
data = _sanitize_submission_data(data)
# Honeypot check — if any _fr_* field has a value, bot caught. Return 200 silently.
if site.get("honeypot_enabled", True):
for key, value in data.items():
if key.startswith("_fr_") and value:
log.info(
"honeypot",
f"Bot caught: {key}='{value}' from {client_ip} on site {site['name']}",
key_field=key,
bot_value=value,
client_ip=client_ip,
site_name=site["name"],
)
return jsonify({"success": True}), 200
# Parse field config for validation + downstream use
field_config = parse_site_fields(site)
# Server-side validation (Phase 3 — form_logic)
if field_config:
validation_errors = validate_submission(field_config, data)
if validation_errors:
return jsonify(
{
"error": "Validation failed",
"errors": validation_errors,
}
), 422
# Compute computed fields and merge into data
if field_config:
data = compute_all_fields(field_config, data)
# Tier enforcement — atomic check, insert, and increment in one transaction
owner_id = site.get("user_id")
user_agent = request.headers.get("User-Agent", "")
if owner_id:
owner = get_user(owner_id)
if owner:
allowed, submission, current, max_count, reason = accept_submission(
site["id"],
owner_id,
owner.get("tier", "free"),
data,
site=site,
client_ip=client_ip,
user_agent=user_agent,
)
if not allowed:
return jsonify(
{
"error": "Submission limit reached. Please contact the site owner.",
"usage": {"current": current, "max": max_count},
}
), 429
# Spam detected — return 200 silently (bot thinks it succeeded)
if reason:
log.info(
"spam",
f"Blocked: {reason} from {client_ip} on site {site['name']}",
reason=reason,
client_ip=client_ip,
site_name=site["name"],
)
return jsonify({"success": True, "submission_id": submission["id"]}), 200
else:
# Sites without owner_id bypass tier enforcement (legacy/admin sites)
submission = add_submission(site["id"], data, client_ip=client_ip, user_agent=user_agent)
# Email notification (field_config already parsed above)
send_notification(site, submission, field_config)
# Webhook forwarding (async, non-blocking)
from app.services.webhook import fire_webhook
fire_webhook(site, submission, field_config)
# Auto-generate document (async, non-blocking)
thread = threading.Thread(
target=_auto_generate_document,
args=(site, submission, field_config),
daemon=True,
)
thread.start()
# Execute configured actions (async, non-blocking) — backward compat:
# existing fire_webhook + send_notification + _auto_generate_document
# still run above; execute_actions() runs any additional configured actions
try:
from app.services.action_pipeline import execute_actions_async
execute_actions_async(site, submission, field_config)
except ImportError:
pass # Graceful degradation if action pipeline not yet available
# Build response with usage info
resp = {"success": True, "submission_id": submission["id"]}
if owner_id and owner:
usage = get_user_usage(owner_id, owner.get("tier", "free"))
resp["usage"] = {
"submissions": usage["submissions"],
"max_submissions": usage["max_submissions"],
"sites_count": usage["site_count"],
"max_sites": usage["max_sites"],
}
return jsonify(resp), 201
@api_bp.route("/health")
def health():
"""Lightweight liveness check — no DB required."""
return jsonify({"status": "ok"}), 200
@api_bp.route("/health/ready")
def health_ready():
"""Readiness probe — checks all critical components.
Returns 200 with component details on success,
503 with failure details if any component is unhealthy.
"""
import os
components = {}
healthy = True
# Database connectivity
try:
from app.db import get_db
conn = get_db()
conn.execute("SELECT 1")
conn.close()
components["database"] = "ok"
except Exception as e:
components["database"] = f"error: {str(e)}"
healthy = False
# Disk space (data directory writable)
try:
data_dir = os.environ.get("DATA_DIR", "/app/data")
test_file = os.path.join(data_dir, ".health_check")
with open(test_file, "w") as f:
f.write("ok")
os.remove(test_file)
components["disk"] = "ok"
except Exception as e:
components["disk"] = f"error: {str(e)}"
healthy = False
# Backup directory accessible
try:
backup_dir = os.path.join(os.environ.get("DATA_DIR", "/app/data"), "backups")
if os.path.isdir(backup_dir):
latest = sorted(os.listdir(backup_dir))[-1]
components["backup"] = f"ok (latest: {latest})"
else:
components["backup"] = "no backups found"
except Exception as e:
components["backup"] = f"error: {str(e)}"
status_code = 200 if healthy else 503
return jsonify(
{
"status": "ok" if healthy else "degraded",
"components": components,
}
), status_code