Add Python form API + harden nginx web root
- New api/ Python service (stdlib only — no pip install, no packages): validates fields server-side, verifies reCAPTCHA, sends via Resend with idempotency key, rate-limited (5 req/IP/15min) - Matches existing project tooling (build_locations.py, build_services.py) - Front-end form.js stays vanilla JS, no JS frameworks anywhere - docker-compose runs nginx + python:3.13-alpine api with healthcheck - nginx proxies /api/ to Python service, strips prefix - Dockerfile now copies only public folders into web root (was copying everything, exposing /Dockerfile, /build_*.py, /api/.env) - nginx.conf denies dotfiles, .env, .conf, .yml, .py, .md, .txt and Dockerfile as defense in depth - .dockerignore keeps sensitive files out of build context - .gitignore protects api/.env and __pycache__ from being committed
This commit is contained in:
Concept Agent
+225
@@ -0,0 +1,225 @@
|
||||
#!/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 <webleads@floorithardwoods.com>")
|
||||
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"""<!DOCTYPE html>
|
||||
<html><body style="font-family:Arial,sans-serif;color:#222;max-width:600px;margin:0 auto;padding:24px;">
|
||||
<p>A new estimate request was submitted on floorithardwoodfloors.com.</p>
|
||||
<table cellpadding="8" style="border-collapse:collapse;border:1px solid #e5e5e5;width:100%;">
|
||||
<tr><td style="border:1px solid #e5e5e5;background:#f7f7f7;"><strong>Name</strong></td><td style="border:1px solid #e5e5e5;">{safe['name']}</td></tr>
|
||||
<tr><td style="border:1px solid #e5e5e5;background:#f7f7f7;"><strong>Email</strong></td><td style="border:1px solid #e5e5e5;">{safe['email']}</td></tr>
|
||||
<tr><td style="border:1px solid #e5e5e5;background:#f7f7f7;"><strong>Phone</strong></td><td style="border:1px solid #e5e5e5;">{safe['phone']}</td></tr>
|
||||
<tr><td style="border:1px solid #e5e5e5;background:#f7f7f7;"><strong>Address</strong></td><td style="border:1px solid #e5e5e5;">{safe['address']}</td></tr>
|
||||
<tr><td style="border:1px solid #e5e5e5;background:#f7f7f7;"><strong>City</strong></td><td style="border:1px solid #e5e5e5;">{safe['city']}</td></tr>
|
||||
<tr><td style="border:1px solid #e5e5e5;background:#f7f7f7;"><strong>Zip</strong></td><td style="border:1px solid #e5e5e5;">{safe['zip']}</td></tr>
|
||||
<tr><td style="border:1px solid #e5e5e5;background:#f7f7f7;"><strong>Service</strong></td><td style="border:1px solid #e5e5e5;">{safe['service']}</td></tr>
|
||||
<tr><td style="border:1px solid #e5e5e5;background:#f7f7f7;"><strong>Floor Condition</strong></td><td style="border:1px solid #e5e5e5;">{safe['condition']}</td></tr>
|
||||
<tr><td style="border:1px solid #e5e5e5;background:#f7f7f7;vertical-align:top;"><strong>Message</strong></td><td style="border:1px solid #e5e5e5;white-space:pre-wrap;">{safe['message']}</td></tr>
|
||||
</table>
|
||||
<p style="margin-top:24px;color:#666;font-size:13px;">Reply directly to this email to respond to {safe['name']}.</p>
|
||||
</body></html>"""
|
||||
|
||||
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()
|
||||
Reference in New Issue
Block a user