Formulario de contacto (SMTP vía Poste), Quiénes somos e invitación a venir al programa
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01TSyR5Zw5eGj6oPTRVjvP1b
This commit is contained in:
co-authored by
Claude Sonnet 5.5
parent
4be49b0c91
commit
89bb7b4d34
@@ -0,0 +1,117 @@
|
||||
"""Endpoint del formulario de contacto: POST /api/contacto -> correo por SMTP.
|
||||
|
||||
Configuración solo por variables de entorno (nunca en el repo):
|
||||
SMTP_HOST, SMTP_PORT (587), SMTP_USER, SMTP_PASS, SMTP_TLS_NAME (nombre del certificado, por defecto SMTP_HOST),
|
||||
MAIL_FROM, MAIL_TO
|
||||
No guarda los mensajes en disco: solo los reenvía por correo.
|
||||
"""
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import smtplib
|
||||
import ssl
|
||||
import sys
|
||||
import time
|
||||
from collections import defaultdict, deque
|
||||
from email.message import EmailMessage
|
||||
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
||||
|
||||
MOTIVOS = {
|
||||
'venir': 'Quiero venir al programa',
|
||||
'tema': 'Propongo un tema',
|
||||
'otro': 'Otro asunto',
|
||||
}
|
||||
EMAIL_RE = re.compile(r'^[^@\s]{1,64}@[^@\s]{1,190}\.[^@\s]{2,}$')
|
||||
MAX_BODY = 16 * 1024
|
||||
RATE_MAX, RATE_WINDOW = 5, 3600 # 5 mensajes por IP y hora
|
||||
hits = defaultdict(deque)
|
||||
|
||||
|
||||
def clean(s, n):
|
||||
return re.sub(r'[\r\n\t]+', ' ', str(s or '')).strip()[:n]
|
||||
|
||||
|
||||
def rate_ok(ip):
|
||||
now = time.time()
|
||||
q = hits[ip]
|
||||
while q and now - q[0] > RATE_WINDOW:
|
||||
q.popleft()
|
||||
if len(q) >= RATE_MAX:
|
||||
return False
|
||||
q.append(now)
|
||||
return True
|
||||
|
||||
|
||||
def send(nombre, email, motivo, mensaje):
|
||||
host, user, pwd = os.environ.get('SMTP_HOST'), os.environ.get('SMTP_USER'), os.environ.get('SMTP_PASS')
|
||||
if not (host and user and pwd and os.environ.get('MAIL_TO')):
|
||||
raise RuntimeError('SMTP no configurado')
|
||||
msg = EmailMessage()
|
||||
msg['Subject'] = f'[Web AQNSN] {MOTIVOS[motivo]} · {nombre}'
|
||||
msg['From'] = os.environ.get('MAIL_FROM', user)
|
||||
msg['To'] = os.environ['MAIL_TO']
|
||||
msg['Reply-To'] = email
|
||||
msg.set_content(f'Motivo: {MOTIVOS[motivo]}\nNombre: {nombre}\nEmail: {email}\n\n{mensaje}\n')
|
||||
s = smtplib.SMTP(host, int(os.environ.get('SMTP_PORT', '587')), timeout=20)
|
||||
try:
|
||||
s.ehlo()
|
||||
s._host = os.environ.get('SMTP_TLS_NAME', host) # el certificado se valida contra este nombre
|
||||
s.starttls(context=ssl.create_default_context())
|
||||
s.ehlo()
|
||||
s.login(user, pwd)
|
||||
s.send_message(msg)
|
||||
finally:
|
||||
try:
|
||||
s.quit()
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
||||
class H(BaseHTTPRequestHandler):
|
||||
server_version = 'aqnsn'
|
||||
|
||||
def log_message(self, fmt, *a): # sin IPs ni contenido en los logs
|
||||
pass
|
||||
|
||||
def reply(self, code, obj):
|
||||
body = json.dumps(obj).encode()
|
||||
self.send_response(code)
|
||||
self.send_header('Content-Type', 'application/json')
|
||||
self.send_header('Content-Length', str(len(body)))
|
||||
self.send_header('Cache-Control', 'no-store')
|
||||
self.end_headers()
|
||||
self.wfile.write(body)
|
||||
|
||||
def do_POST(self):
|
||||
if self.path != '/api/contacto':
|
||||
return self.reply(404, {'ok': False})
|
||||
try:
|
||||
n = int(self.headers.get('Content-Length', '0'))
|
||||
if n <= 0 or n > MAX_BODY:
|
||||
return self.reply(413, {'ok': False, 'error': 'Mensaje demasiado largo.'})
|
||||
d = json.loads(self.rfile.read(n))
|
||||
except Exception:
|
||||
return self.reply(400, {'ok': False, 'error': 'Petición no válida.'})
|
||||
ip = (self.headers.get('X-Forwarded-For') or self.client_address[0]).split(',')[0].strip()
|
||||
# antispam: campo trampa relleno o envío en menos de 3 s => se responde "ok" sin enviar
|
||||
if d.get('web') or (isinstance(d.get('t'), (int, float)) and d['t'] < 3000):
|
||||
return self.reply(200, {'ok': True})
|
||||
if not rate_ok(ip):
|
||||
return self.reply(429, {'ok': False, 'error': 'Demasiados mensajes. Inténtalo más tarde.'})
|
||||
nombre, email = clean(d.get('nombre'), 100), clean(d.get('email'), 200)
|
||||
motivo = d.get('motivo') if d.get('motivo') in MOTIVOS else 'otro'
|
||||
mensaje = str(d.get('mensaje') or '').replace('\r', '').strip()[:3000]
|
||||
if not nombre or not EMAIL_RE.match(email) or len(mensaje) < 10 or d.get('consent') is not True:
|
||||
return self.reply(400, {'ok': False, 'error': 'Revisa los campos del formulario.'})
|
||||
try:
|
||||
send(nombre, email, motivo, mensaje)
|
||||
except RuntimeError:
|
||||
return self.reply(503, {'ok': False, 'error': 'El formulario no está disponible ahora mismo.'})
|
||||
except Exception as exc:
|
||||
print('error enviando correo:', type(exc).__name__, file=sys.stderr, flush=True)
|
||||
return self.reply(502, {'ok': False, 'error': 'No se pudo enviar. Inténtalo más tarde.'})
|
||||
self.reply(200, {'ok': True})
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
ThreadingHTTPServer(('127.0.0.1', 8081), H).serve_forever()
|
||||
Reference in New Issue
Block a user