andaquenosabesna/server.py

"""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()