#!/usr/bin/env python3 """ Floor It estimate-form API. Pure Python 3 standard library only — no pip install, no packages. Handles POST /estimate from the static site, validates fields, verifies reCAPTCHA, sends via Resend, with idempotency + rate limiting. """ import hashlib import http.server import json import os import re import socketserver import time import urllib.parse import urllib.request PORT = int(os.environ.get("PORT", "3001")) RESEND_API_KEY = os.environ.get("RESEND_API_KEY", "") RECAPTCHA_SECRET = os.environ.get("RECAPTCHA_SECRET", "") TO_EMAIL = os.environ.get("TO_EMAIL", "floorithardwoodfloors@gmail.com") FROM_EMAIL = os.environ.get("FROM_EMAIL", "Floor It Hardwood Floors ") RECAPTCHA_MIN = float(os.environ.get("RECAPTCHA_MIN", "0.5")) PHONE_RE = re.compile(r"^$?\d{3}$?[\s.\-]?\d{3}[\s.\-]?\d{4}$") EMAIL_RE = re.compile(r"^[^\s@]+@[^\s@]+\.[^\s@]+$") # In-memory rate limit: 5 requests / IP / 15 minutes RATE_MAP = {} RATE_WINDOW = 15 * 60 RATE_MAX = 5 def sanitize(s): if not isinstance(s, str): return "" s = s.replace("&", "&").replace("<", "<").replace(">", ">").replace('"', """) return s.strip()[:2000] def validate_fields(body): errors = [] if not body.get("name") or len((body.get("name") or "").strip()) < 2: errors.append("name") if not body.get("email") or not EMAIL_RE.match((body.get("email") or "").strip()): errors.append("email") phone_clean = (body.get("phone") or "").replace(" ", "") if not phone_clean or not PHONE_RE.match(phone_clean): errors.append("phone") if not body.get("address") or len((body.get("address") or "").strip()) < 3: errors.append("address") if not (body.get("service") or "").strip(): errors.append("service") return errors def verify_recaptcha(token): if not RECAPTCHA_SECRET or not token: return 0.0 data = urllib.parse.urlencode({"secret": RECAPTCHA_SECRET, "response": token}).encode() req = urllib.request.Request( "https://www.google.com/recaptcha/api/siteverify", data=data, headers={"Content-Type": "application/x-www-form-urlencoded"}, ) try: with urllib.request.urlopen(req, timeout=8) as resp: return float(json.loads(resp.read()).get("score", 0)) except Exception: return 0.0 def send_via_resend(fields): safe = {k: sanitize(fields.get(k, "")) for k in ["name", "email", "phone", "address", "city", "zip", "service", "condition", "message"]} html = f"""

A new estimate request was submitted on floorithardwoodfloors.com.

Name	{safe['name']}
Email	{safe['email']}
Phone	{safe['phone']}
Address	{safe['address']}
City	{safe['city']}
Zip	{safe['zip']}
Service	{safe['service']}
Floor Condition	{safe['condition']}
Message	{safe['message']}

Reply directly to this email to respond to {safe['name']}.

""" text = ( "New estimate request from floorithardwoodfloors.com\n\n" f"Name: {safe['name']}\n" f"Email: {safe['email']}\n" f"Phone: {safe['phone']}\n" f"Address: {safe['address']}\n" f"City: {safe['city']}\n" f"Zip: {safe['zip']}\n" f"Service: {safe['service']}\n" f"Floor Condition: {safe['condition']}\n\n" "Message:\n" f"{safe['message']}\n\n" f"Reply directly to this email to respond to {safe['name']}." ) payload_obj = { "from": FROM_EMAIL, "to": [TO_EMAIL], "reply_to": (fields.get("email") or "").strip(), "subject": f"New estimate request: {safe['name']} ({safe['city'] or 'Buffalo'})", "html": html, "text": text, } payload = json.dumps(payload_obj).encode("utf-8") idem_key = hashlib.sha256(payload).hexdigest()[:64] req = urllib.request.Request( "https://api.resend.com/emails", data=payload, headers={ "Authorization": f"Bearer {RESEND_API_KEY}", "Content-Type": "application/json", "Idempotency-Key": idem_key, "User-Agent": "FloorIt-Estimate-Form/1.0", }, ) try: with urllib.request.urlopen(req, timeout=10) as resp: if resp.status < 200 or resp.status >= 300: raise RuntimeError(f"Resend {resp.status}: {resp.read().decode('utf-8', 'ignore')}") except urllib.error.HTTPError as e: body = e.read().decode("utf-8", "ignore") raise RuntimeError(f"Resend {e.code}: {body}") from None def rate_limit(ip): now = time.time() entry = RATE_MAP.get(ip, {"count": 0, "start": now}) if now - entry["start"] > RATE_WINDOW: entry = {"count": 0, "start": now} entry["count"] += 1 RATE_MAP[ip] = entry return entry["count"] > RATE_MAX class Handler(http.server.BaseHTTPRequestHandler): def log_message(self, fmt, *args): print(f"[api] {fmt % args}", flush=True) def _json(self, code, obj): body = json.dumps(obj).encode("utf-8") self.send_response(code) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(body))) self.send_header("Access-Control-Allow-Origin", "https://floorithardwoodfloors.com") self.send_header("Access-Control-Allow-Methods", "POST, OPTIONS") self.send_header("Access-Control-Allow-Headers", "Content-Type") self.end_headers() self.wfile.write(body) def do_OPTIONS(self): self.send_response(204) self.send_header("Access-Control-Allow-Origin", "https://floorithardwoodfloors.com") self.send_header("Access-Control-Allow-Methods", "POST, OPTIONS") self.send_header("Access-Control-Allow-Headers", "Content-Type") self.end_headers() def do_GET(self): if self.path == "/health": self._json(200, {"ok": True}) return self._json(404, {"error": "not found"}) def do_POST(self): if self.path != "/estimate": self._json(404, {"error": "not found"}) return ip = self.headers.get("x-forwarded-for", "").split(",")[0].strip() or self.client_address[0] if rate_limit(ip): self._json(429, {"error": "Too many requests. Please call us at (716) 602-1429."}) return length = int(self.headers.get("Content-Length", "0")) if length > 16384: self._json(413, {"error": "Payload too large."}) return try: raw = self.rfile.read(length).decode("utf-8") body = json.loads(raw) except Exception: self._json(400, {"error": "Invalid request."}) return errors = validate_fields(body) if errors: self._json(422, {"error": "Validation failed.", "fields": errors}) return score = verify_recaptcha(body.get("token", "")) if RECAPTCHA_SECRET and score < RECAPTCHA_MIN: self._json(403, {"error": "Request could not be verified. Please try again or call us directly."}) return try: send_via_resend(body) self._json(200, {"ok": True}) except Exception as e: print(f"[estimate] {e}", flush=True) self._json(500, {"error": "Could not send your message. Please call us at (716) 602-1429."}) class ThreadedHTTPServer(socketserver.ThreadingMixIn, http.server.HTTPServer): daemon_threads = True allow_reuse_address = True if __name__ == "__main__": print(f"[api] listening on :{PORT}", flush=True) ThreadedHTTPServer(("0.0.0.0", PORT), Handler).serve_forever()