"""Middleware, hooks, and error handlers for AgentForms.
Registers before/after request hooks, error handlers, context processors,
template filters, and security headers on the Flask app instance.
"""
import secrets
import traceback
from flask import abort, render_template, request, session
from markupsafe import Markup
from werkzeug.middleware.proxy_fix import ProxyFix
from app.config import Config
from app.csrf import csrf_context_processor, csrf_protect
from app.services.logging import log
from app.services.ratelimit import limiter
def init_middleware(app):
"""Register all middleware hooks, error handlers, and template utilities."""
# Proxy fix — trust Caddy reverse proxy
app.wsgi_app = ProxyFix(app.wsgi_app, x_for=1, x_proto=1)
# ─── CSRF protection ────────────────────────────────────────────────────────
# Middleware extracted to app/csrf.py for clarity.
# API endpoints are exempted — they use tokens, not form submissions.
# Webhooks, public endpoints, login/registration entry points are skipped.
# JSON API from authenticated sessions uses origin validation (not Content-Type trust).
app.before_request(csrf_protect)
# Inject csrf_token + CSP nonce into all templates
@app.context_processor
def inject_csrf():
return csrf_context_processor()
# Generate per-request CSP nonce
@app.before_request
def _generate_csp_nonce():
request.csp_nonce = secrets.token_urlsafe(16)
# ─── Custom template filters ──────────────────────────────────────────────
# nl2br_escape — escape HTML first, then convert newlines to <br> (XSS-safe)
@app.template_filter("nl2br")
def nl2br_filter(value):
if not value:
return ""
from markupsafe import escape
escaped = escape(str(value))
return Markup(str(escaped).replace("\n", "<br>\n"))
# ─── Security headers ─────────────────────────────────────────────────────
@app.after_request
def _security_headers(response):
# Allow document previews to be iframed
if request.path.startswith("/documents/") and (
request.path.endswith("/preview") or request.path.endswith("/embed")
):
return response
# Content Security Policy — strict script-src 'self', nonce for styles only
nonce = getattr(request, "csp_nonce", None) or secrets.token_urlsafe(16)
response.headers["Content-Security-Policy"] = (
"default-src 'self'; "
"script-src 'self'; "
f"style-src 'self' 'nonce-{nonce}'; "
"img-src 'self' data: https:; "
"font-src 'self' data:; "
"connect-src 'self'; "
"frame-ancestors 'none';"
)
# HSTS — enforce HTTPS for 1 year, no subdomains
response.headers["Strict-Transport-Security"] = "max-age=31536000; includeSubDomains"
# Prevent clickjacking
response.headers["X-Frame-Options"] = "DENY"
# Prevent MIME sniffing
response.headers["X-Content-Type-Options"] = "nosniff"
# Referrer policy
response.headers["Referrer-Policy"] = "strict-origin-when-cross-origin"
# Permissions Policy — restrict unused browser features
response.headers["Permissions-Policy"] = (
"camera=(), microphone=(), geolocation=(), "
"payment=(), usb=(), magnetometer=(), gyroscope=(), accelerometer=()"
)
return response
# ─── Error handling ──────────────────────────────────────────────────────
@app.errorhandler(404)
def not_found(e):
return render_template("error_404.html"), 404
@app.errorhandler(403)
def forbidden(e):
return render_template("error_403.html"), 403
@app.errorhandler(413)
def payload_too_large(e):
"""Handle oversized request payloads (Phase 2)."""
if request.is_json:
return {"error": "Payload too large (max 1MB)"}, 413
return "Payload too large", 413
@app.errorhandler(500)
def internal_error(e):
"""Handle 5xx errors with alerting."""
from app.services.alerting import alert
tb = traceback.format_exc()
log.error("app", f"Internal server error: {e}", path=request.path, method=request.method)
alert.error(
"500 Internal Server Error",
path=request.path,
method=request.method,
error=str(e),
)
return render_template("error_500.html"), 500
@app.errorhandler(502)
def bad_gateway(e):
"""Handle 502 errors with alerting."""
from app.services.alerting import alert
log.error("app", "Bad gateway", path=request.path, method=request.method)
alert.error(
"502 Bad Gateway",
path=request.path,
method=request.method,
error=str(e),
)
return {"error": "Bad gateway"}, 502
@app.errorhandler(503)
def service_unavailable(e):
"""Handle 503 errors with alerting."""
from app.services.alerting import alert
log.error("app", "Service unavailable", path=request.path, method=request.method)
alert.error(
"503 Service Unavailable",
path=request.path,
method=request.method,
error=str(e),
)
return {"error": "Service unavailable"}, 503
# ─── Rate limiting middleware ──────────────────────────────────────────────
# API endpoints: 60 req/min per IP
# Auth endpoints: 10 req/min per IP (brute-force protection)
AUTH_RATE_LIMIT = Config.AUTH_RATE_LIMIT
API_RATE_LIMIT = Config.API_RATE_LIMIT
USER_API_RATE_LIMIT = Config.USER_API_RATE_LIMIT
RATE_LIMIT_WINDOW = Config.RATE_LIMIT_WINDOW
@app.before_request
def rate_limit_check():
path = request.path
# Auth endpoints — stricter limit
if path.startswith("/auth/") or path.startswith("/settings/"):
ip_key = f"rate:ip:{request.remote_addr}:auth"
allowed, remaining, reset_at = limiter.check(ip_key, max_requests=AUTH_RATE_LIMIT, window=RATE_LIMIT_WINDOW)
request.rate_limit_info = {
"limit": AUTH_RATE_LIMIT,
"remaining": remaining,
"reset": int(reset_at),
}
if not allowed:
from flask import jsonify as jf
resp = jf({"error": "Too many requests. Try again later."})
resp.status_code = 429
resp.headers["Retry-After"] = "60"
return resp
# API endpoints — per-IP limit
if path.startswith("/api/"):
ip_key = f"rate:ip:{request.remote_addr}:api"
allowed, remaining, reset_at = limiter.check(ip_key, max_requests=API_RATE_LIMIT, window=RATE_LIMIT_WINDOW)
request.rate_limit_info = {
"limit": API_RATE_LIMIT,
"remaining": remaining,
"reset": int(reset_at),
}
if not allowed:
from flask import jsonify as jf
resp = jf({"error": "Too many requests. Try again later."})
resp.status_code = 429
resp.headers["Retry-After"] = "60"
return resp
# Per-user API rate limiting (authenticated users)
# Prevents a compromised account from spamming
if "user_id" in session and path.startswith("/api/"):
user_key = f"rate:user:{session['user_id']}:api"
user_allowed, user_remaining, user_reset = limiter.check(user_key, max_requests=USER_API_RATE_LIMIT, window=RATE_LIMIT_WINDOW)
if not user_allowed:
from flask import jsonify as jf
resp = jf(
{
"error": "User rate limit exceeded. Try again later.",
"usage": {"current": USER_API_RATE_LIMIT, "max": USER_API_RATE_LIMIT, "window": RATE_LIMIT_WINDOW},
}
)
resp.status_code = 429
resp.headers["Retry-After"] = "60"
return resp
@app.after_request
def add_rate_limit_headers(response):
"""Add X-RateLimit-* headers to API responses."""
if hasattr(request, "rate_limit_info") and request.rate_limit_info:
info = request.rate_limit_info
response.headers["X-RateLimit-Limit"] = str(info["limit"])
response.headers["X-RateLimit-Remaining"] = str(info["remaining"])
response.headers["X-RateLimit-Reset"] = str(info["reset"])
return response