"""Public contact form endpoint — no authentication required.""" from __future__ import annotations import html import logging import time from email.mime.multipart import MIMEMultipart from email.mime.text import MIMEText from fastapi import APIRouter, Request from pydantic import BaseModel, EmailStr, Field from backend.config import settings from backend.services.email import _send_raw logger = logging.getLogger(__name__) router = APIRouter() # Simple in-memory rate limiting (per-instance, resets on deploy) _recent: dict[str, float] = {} _RATE_LIMIT_SECONDS = 60 class ContactRequest(BaseModel): name: str = Field(..., min_length=1, max_length=200) email: EmailStr = Field(..., max_length=254) message: str = Field(..., min_length=1, max_length=5000) company: str = Field("", max_length=200) subject: str = Field("", max_length=200) honeypot: str = Field("", alias="_honey") class ContactResponse(BaseModel): success: bool message: str def _build_contact_message(data: ContactRequest) -> MIMEMultipart: """Build the contact form email.""" msg = MIMEMultipart("alternative") msg["From"] = f"Periscope <{settings.email_sender}>" msg["To"] = settings.contact_recipient msg["Reply-To"] = data.email msg["Subject"] = f"[Periscope Contact] {data.subject or 'New message'} from {data.name}" # Plain text lines = [ f"Name: {data.name}", f"Email: {data.email}", ] if data.company: lines.append(f"Company: {data.company}") if data.subject: lines.append(f"Subject: {data.subject}") lines += ["", data.message, "", "— Sent from the Periscope contact form"] msg.attach(MIMEText("\n".join(lines), "plain")) # HTML name = html.escape(data.name) email = html.escape(data.email) company = html.escape(data.company) subject = html.escape(data.subject) message = html.escape(data.message) rows = f"""\ Name {name} Email {email} """ if data.company: rows += f"""\ Company {company} """ if data.subject: rows += f"""\ Subject {subject} """ html_body = f"""\

New contact form submission

{rows}
{message}

Sent from the Periscope contact form

""" msg.attach(MIMEText(html_body, "html")) return msg @router.post("/contact", response_model=ContactResponse) async def submit_contact(data: ContactRequest, request: Request): # Honeypot check — bots fill hidden fields if data.honeypot: return ContactResponse(success=True, message="Message sent! We'll get back to you soon.") # Rate limiting by IP ip = request.headers.get("x-forwarded-for", "").split(",")[0].strip() or request.client.host now = time.time() last = _recent.get(ip) if last and now - last < _RATE_LIMIT_SECONDS: return ContactResponse(success=False, message="Please wait a minute before submitting again.") _recent[ip] = now # Clean up old entries if len(_recent) > 1000: cutoff = now - _RATE_LIMIT_SECONDS for key in [k for k, v in _recent.items() if v < cutoff]: del _recent[key] # Check email is configured if not settings.use_email or not settings.contact_recipient: logger.warning("Contact form submitted but email is not configured") return ContactResponse( success=False, message="Email is not configured on this server.", ) msg = _build_contact_message(data) await _send_raw(settings.contact_recipient, msg, "Contact form") return ContactResponse(success=True, message="Message sent! We'll get back to you soon.")