"""User-facing site management — create, list, delete own sites."""
import ipaddress
import os
import re
from urllib.parse import urlparse
from flask import Blueprint, flash, jsonify, redirect, render_template, request, session, url_for
from app.helpers import _get_site_owner_password_hash
from app.models import (
accept_site_creation,
add_site,
can_create_site,
delete_site,
get_site_analytics,
get_site_analytics_summary,
get_site_owner_tier,
get_template_by_slug,
get_user_site_count,
get_user_sites,
get_user_usage,
list_submissions,
list_templates,
parse_site_fields,
site_device_stats,
site_field_dropoff,
site_geo_stats,
site_impression_stats,
site_session_stats,
update_site_fields,
)
from app.routes.auth import current_user, login_required, verified_email_required
user_sites_bp = Blueprint("user_sites", __name__, url_prefix="/sites")
# ─── SSRF protection ──────────────────────────────────────────────────────────
# RFC 1918 + RFC 3927 + RFC 6598 + link-local + multicast + broadcast
_PRIVATE_IP_RANGES = [
"0.0.0.0/8", # current network
"10.0.0.0/8", # private
"100.64.0.0/10", # CGNAT
"127.0.0.0/8", # loopback
"169.254.0.0/16", # link-local
"172.16.0.0/12", # private
"192.0.0.0/24", # IETF protocol
"192.0.2.0/24", # documentation (TEST-NET-1)
"192.168.0.0/16", # private
"198.18.0.0/15", # benchmark
"198.51.100.0/24", # documentation (TEST-NET-2)
"203.0.113.0/24", # documentation (TEST-NET-3)
"224.0.0.0/4", # multicast
"240.0.0.0/4", # reserved
"255.255.255.255/32", # broadcast
]
# Precompile CIDR networks for fast lookup
_BLOCKED_NETWORKS = [ipaddress.ip_network(cidr) for cidr in _PRIVATE_IP_RANGES]
# Blocklist: AWS metadata, GCP metadata, Azure metadata, Kubernetes, etc.
_METADATA_HOSTS = {
"metadata.google.internal",
"metadata.google.com",
"metadata.google",
"169.254.169.254",
"metadata",
"kubernetes",
"kubernetes.default",
"kubernetes.default.svc",
"kubernetes.default.svc.cluster.local",
}
def _is_safe_url(url: str) -> tuple[bool, str]:
"""Validate a URL to prevent SSRF attacks.
Resolves DNS before checking to prevent DNS rebinding attacks.
Returns (is_safe, error_message).
"""
if not url or not url.strip():
return True, ""
try:
parsed = urlparse(url)
except Exception:
return False, "Invalid URL format"
# Only allow http/https schemes
if parsed.scheme not in ("http", "https"):
return False, f"Only http/https schemes allowed (got: {parsed.scheme})"
hostname = (parsed.hostname or "").lower()
if not hostname:
return False, "URL must contain a hostname"
# Check against known metadata hosts
if hostname in _METADATA_HOSTS or hostname.endswith("." + "_".join(_METADATA_HOSTS)):
return False, "Access to metadata services is blocked"
# If it looks like an IP, check against private ranges
ip_match = re.match(r"^(\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3})$", hostname)
if ip_match:
try:
ip = ipaddress.ip_address(ip_match.group(1))
if isinstance(ip, ipaddress.IPv4Address):
for network in _BLOCKED_NETWORKS:
if ip in network:
return False, f"Private/reserved IP addresses are blocked: {ip}"
except ValueError:
return False, "Invalid IP address"
else:
# Check for IP literal in brackets (IPv6)
if hostname.startswith("[") and hostname.endswith("]"):
return False, "IPv6 literal addresses are blocked for safety"
# DNS rebinding protection — resolve hostname and check resolved IPs
try:
import socket
addrinfos = socket.getaddrinfo(hostname, None, socket.AF_INET)
for info in addrinfos:
resolved_ip = info[4][0]
ip = ipaddress.ip_address(resolved_ip)
for network in _BLOCKED_NETWORKS:
if ip in network:
return False, f"Resolved IP is blocked: {resolved_ip}"
except socket.gaierror:
# Hostname doesn't resolve — allow it (might be valid)
pass
except Exception:
# Fail open on DNS errors — don't block valid requests
pass
return True, ""
def validate_webhook_url(url: str) -> tuple[bool, str]:
"""Validate webhook URL to prevent SSRF.
Returns (is_valid, error_message).
"""
return _is_safe_url(url)
@user_sites_bp.route("/")
@login_required
def list_sites():
"""List user's own sites with usage info."""
user = current_user()
sites = get_user_sites(user["id"])
usage = get_user_usage(user["id"], user["tier"])
can_add = can_create_site(user["id"], user["tier"])
return render_template(
"user_sites.html",
user=user,
sites=sites,
usage=usage,
can_add=can_add,
)
@user_sites_bp.route("/new", methods=["GET", "POST"])
@login_required
@verified_email_required
def new_site():
"""Create a new site (form)."""
import copy
from collections import OrderedDict
user = current_user()
templates = list_templates() # for the dropdown
# Group templates by category for the dropdown
categories = OrderedDict()
for tpl in templates:
cat = tpl["category"]
if cat not in categories:
categories[cat] = []
categories[cat].append(tpl)
grouped_templates = list(categories.items())
if request.method == "POST":
name = request.form.get("name", "").strip()
owner_email = request.form.get("owner_email", "").strip()
smtp_from = request.form.get("smtp_from", "").strip()
template_slug = request.form.get("template_slug", "").strip()
errors = []
if not name:
errors.append("Site name is required")
if not owner_email or "@" not in owner_email:
errors.append("Valid owner email is required")
if errors:
return render_template(
"add_site_user.html",
user=user,
errors=errors,
name=name,
owner_email=owner_email,
templates=templates,
grouped_templates=grouped_templates,
), 400
# If a template was selected, clone its field_config
field_config = None
if template_slug:
template = get_template_by_slug(template_slug)
if template:
field_config = copy.deepcopy(template["field_config"])
success, site, current_count, max_sites = accept_site_creation(
name, owner_email, user["id"], user["tier"], smtp_from or None, field_config
)
if not success:
flash(f"Site limit reached ({current_count}/{max_sites} forms). Upgrade to create more.", "warning")
return redirect(url_for("user_sites.list_sites"))
flash(f"Site '{name}' created! Your token: {site['token']}", "success")
return redirect(url_for("user_sites.list_sites"))
return render_template(
"add_site_user.html", user=user, errors=[], templates=templates, grouped_templates=grouped_templates
)
@user_sites_bp.route("/<int:site_id>/delete", methods=["POST"])
@login_required
def delete_site_route(site_id):
"""Delete user's own site."""
user = current_user()
sites = get_user_sites(user["id"])
site = next((s for s in sites if s["id"] == site_id), None)
if not site:
flash("Site not found", "error")
return redirect(url_for("user_sites.list_sites"))
delete_site(site_id)
flash(f"Site '{site['name']}' deleted.", "info")
return redirect(url_for("user_sites.list_sites"))
@user_sites_bp.route("/<int:site_id>/submissions")
@login_required
def view_submissions(site_id):
"""View submissions for one of the user's sites."""
user = current_user()
base_url = os.environ.get("APP_URL", "https://agentforms.io")
sites = get_user_sites(user["id"])
site = next((s for s in sites if s["id"] == site_id), None)
if not site:
flash("Site not found", "error")
return redirect(url_for("user_sites.list_sites"))
submissions = list_submissions(limit=50, site_filter=site_id, password_hash=_get_site_owner_password_hash(site_id))
# Parse dynamic JSON data for display
import json
for sub in submissions:
raw = sub.get("data") or ""
try:
sub["_dynamic"] = json.loads(raw) if raw else {}
except (json.JSONDecodeError, TypeError):
sub["_dynamic"] = {}
return render_template(
"site_submissions.html",
user=user,
site=site,
submissions=submissions,
field_config=parse_site_fields(site),
base_url=base_url,
)
@user_sites_bp.route("/<int:site_id>/api/submissions")
@login_required
def api_submissions(site_id):
"""JSON API for submissions list (used by Documents UI)."""
import json
user = current_user()
sites = get_user_sites(user["id"])
site = next((s for s in sites if s["id"] == site_id), None)
if not site:
return jsonify({"error": "Site not found"}), 404
limit = request.args.get("limit", 50, type=int)
submissions = list_submissions(
limit=limit, site_filter=site_id, password_hash=_get_site_owner_password_hash(site_id)
)
# Parse dynamic JSON data
result = []
for sub in submissions:
raw = sub.get("data") or ""
try:
dynamic = json.loads(raw) if raw else {}
except (json.JSONDecodeError, TypeError):
dynamic = {}
result.append(
{
"id": sub["id"],
"customer_name": dynamic.get("name", "")
or dynamic.get("full_name", "")
or dynamic.get("customer_name", ""),
"customer_email": dynamic.get("email", "") or dynamic.get("customer_email", ""),
"submitted_at": sub["submitted_at"],
"data": dynamic,
}
)
return jsonify({"submissions": result})
@user_sites_bp.route("/<int:site_id>/settings", methods=["GET", "POST"])
@login_required
def site_settings_redirect(site_id):
"""Redirect /sites/<id>/settings to /sites/<id>/edit where form settings live."""
return redirect(url_for("user_sites.edit_site", site_id=site_id))
@user_sites_bp.route("/<int:site_id>/edit", methods=["GET", "POST"])
@login_required
def edit_site(site_id):
"""Edit site settings: form fields builder + webhook config."""
import json
user = current_user()
base_url = os.environ.get("APP_URL", "https://agentforms.io")
sites = get_user_sites(user["id"])
site = next((s for s in sites if s["id"] == site_id), None)
if not site:
flash("Site not found", "error")
return redirect(url_for("user_sites.list_sites"))
if request.method == "POST":
# Parse field config from JSON in form
field_config_json = request.form.get("field_config", "[]")
try:
field_config = json.loads(field_config_json)
except json.JSONDecodeError:
field_config = []
webhook_url = request.form.get("webhook_url", "").strip()
webhook_enabled = request.form.get("webhook_enabled") == "on"
webhook_events = request.form.get("webhook_events", "submission")
honeypot_enabled = request.form.get("honeypot_enabled") == "on"
spam_filter_enabled = request.form.get("spam_filter_enabled") == "on"
rate_limit_enabled = request.form.get("rate_limit_enabled") == "on"
rate_limit_burst = request.form.get("rate_limit_burst", "20").strip() or "20"
rate_limit_refill = request.form.get("rate_limit_refill", "0.5").strip() or "0.5"
# Validate webhook URL if provided and enabled
if webhook_url and webhook_enabled:
is_safe, error_msg = validate_webhook_url(webhook_url)
if not is_safe:
flash(f"Invalid webhook URL: {error_msg}", "error")
field_config = parse_site_fields(site)
return render_template(
"edit_site.html",
user=user,
site=site,
field_config=field_config,
base_url=base_url,
), 400
update_site_fields(
site_id,
field_config=field_config,
webhook_url=webhook_url if webhook_url else None,
webhook_enabled=webhook_enabled,
webhook_events=webhook_events,
honeypot_enabled=honeypot_enabled,
spam_filter_enabled=spam_filter_enabled,
rate_limit_enabled=rate_limit_enabled,
rate_limit_burst=int(rate_limit_burst),
rate_limit_refill=float(rate_limit_refill),
changed_by=user.get("email", session.get("email", "user")),
change_reason="Field config updated via edit form",
)
flash("Site settings saved!", "success")
return redirect(url_for("user_sites.edit_site", site_id=site_id))
field_config = parse_site_fields(site)
return render_template(
"edit_site.html",
user=user,
site=site,
field_config=field_config,
base_url=base_url,
)
@user_sites_bp.route("/<int:site_id>/analytics")
@login_required
def site_analytics_view(site_id):
"""Per-site analytics page with tier-based gating."""
user = current_user()
sites = get_user_sites(user["id"])
site = next((s for s in sites if s["id"] == site_id), None)
if not site:
flash("Site not found", "error")
return redirect(url_for("user_sites.list_sites"))
days = request.args.get("days", 30, type=int)
_, tier_config = get_site_owner_tier(site_id)
analytics_level = tier_config.get("analytics", "basic")
# Always available: submission analytics
analytics = get_site_analytics(site_id, days=days)
summary = get_site_analytics_summary(site_id)
# Standard+ tier: geo + device
geo = None
device = None
if analytics_level in ("standard", "advanced"):
geo = site_geo_stats(site_id, days=days)
device = site_device_stats(site_id, days=days)
# Advanced tier: impressions + conversion + session analytics
impressions = None
session_stats = None
field_dropoff = None
if analytics_level == "advanced":
impressions = site_impression_stats(site_id, days=days)
session_stats = site_session_stats(site_id, days=days)
field_dropoff = site_field_dropoff(site_id, days=days)
return render_template(
"site_analytics.html",
user=user,
site=site,
analytics=analytics,
summary=summary,
geo=geo,
device=device,
impressions=impressions,
days=days,
analytics_level=analytics_level,
tier_config=tier_config,
session_stats=session_stats,
field_dropoff=field_dropoff,
)
@user_sites_bp.route("/<int:site_id>/analytics.json")
@login_required
def site_analytics_api(site_id):
"""Per-site analytics as JSON (for charts). Tier-gated."""
user = current_user()
sites = get_user_sites(user["id"])
site = next((s for s in sites if s["id"] == site_id), None)
if not site:
return jsonify({"error": "Site not found"}), 404
days = request.args.get("days", 30, type=int)
_, tier_config = get_site_owner_tier(site_id)
analytics_level = tier_config.get("analytics", "basic")
result = {
"daily": get_site_analytics(site_id, days=days),
"summary": get_site_analytics_summary(site_id),
}
if analytics_level in ("standard", "advanced"):
result["geo"] = site_geo_stats(site_id, days=days)
result["device"] = site_device_stats(site_id, days=days)
if analytics_level == "advanced":
result["impressions"] = site_impression_stats(site_id, days=days)
# Phase 10: Session analytics (advanced tier only)
result["session_stats"] = site_session_stats(site_id, days=days)
result["field_dropoff"] = site_field_dropoff(site_id, days=days)
return jsonify(result)
@user_sites_bp.route("/templates")
@login_required
def browse_templates():
"""Template Marketplace — browse and clone form templates."""
user = current_user()
return render_template(
"templates_browser.html",
user=user,
)