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 = [ "", f"

New Form Submission from {html_escape(site['name'])}

", "", f"", f"", f"", f"", f"", ] # Add dynamic fields to HTML if submission.get("data"): try: dynamic = json.loads(submission["data"]) if dynamic: html_lines.append( "" ) for key, value in dynamic.items(): html_lines.append( f"" ) except (json.JSONDecodeError, TypeError): pass html_lines.extend( [ "", f"", "
Name:{html_escape(str(submission.get('customer_name') or 'N/A'))}
Phone:{html_escape(str(submission.get('customer_phone') or 'N/A'))}
Email:{html_escape(str(submission.get('customer_email') or 'N/A'))}
Equipment:{html_escape(str(submission.get('customer_equipment') or 'N/A'))}
Message:{html_escape(str(submission.get('customer_message') or 'N/A'))}
Additional Fields:
{html_escape(str(key))}:{html_escape(str(value) or '')}
Submitted:
{html_escape(str(submission.get('submitted_at') or 'N/A'))}
", # Tracking pixel (1x1 transparent GIF) f'', "", ] ) 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