"""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 datetime from zoneinfo import ZoneInfo import json import os from urllib.parse import urlparse, parse_qs 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', } TIPOS = { 'mencion': 'Anunciarme / que me mencionéis en el programa', 'patrocinio': 'Patrocinar episodios o una temporada', 'colab': 'Colaboración o canje', 'otro': 'Otra propuesta', } 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(subject, reply_to, body, to=None): 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'] = subject msg['From'] = os.environ.get('MAIL_FROM', user) msg['To'] = to or os.environ['MAIL_TO'] if reply_to: msg['Reply-To'] = reply_to msg.set_content(body) 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 def load_guest(): """Preguntas de la ficha (copia del formulario de Google) -> [{id, sec, t, req}].""" p = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'guest_form.json') try: form = json.load(open(p, encoding='utf-8')) except Exception: return [] out, sec, n = [], '', 0 for it in form['items']: if it['type'] == 8: sec = it['title'] elif it['type'] in (0, 1): n += 1 out.append({'id': f'q{n:02d}', 'sec': sec, 't': it['title'], 'req': it['required']}) return out GUEST_Q = load_guest() check_hits = defaultdict(deque) guest_hits = defaultdict(deque) def limited(bucket, ip, maxn, window=3600): now = time.time() q = bucket[ip] while q and now - q[0] > window: q.popleft() if len(q) >= maxn: return True q.append(now) return False confirmados = set() DIAS = ['lunes', 'martes', 'miércoles', 'jueves', 'viernes', 'sábado', 'domingo'] MESES = ['enero', 'febrero', 'marzo', 'abril', 'mayo', 'junio', 'julio', 'agosto', 'septiembre', 'octubre', 'noviembre', 'diciembre'] def cita_txt(ts): d = datetime.datetime.fromtimestamp(ts, ZoneInfo('Europe/Madrid')) return f'{DIAS[d.weekday()]} {d.day} de {MESES[d.month - 1]} de {d.year}, a las {d:%H:%M}' NAME_RE = re.compile(r"^[^\W\d_][\w .'\-]{1,59}$", re.UNICODE) def invitado(n, f): """Enlace = nombre + fecha de grabación como marca de tiempo Unix (segundos). Válido de 14 días antes a 10 días después.""" n = ' '.join(str(n or '').split()) if not NAME_RE.match(n): return None try: ts = int(str(f).strip()) dia = datetime.datetime.fromtimestamp(ts, datetime.timezone.utc).date() except (ValueError, OverflowError, OSError): return None delta = (dia - datetime.date.today()).days return (n, ts) if -10 <= delta <= 14 else None 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 client_ip(self): return (self.headers.get('X-Forwarded-For') or self.client_address[0]).split(',')[0].strip() def do_GET(self): if urlparse(self.path).path != '/api/invitados/check': return self.reply(404, {'ok': False}) if limited(check_hits, self.client_ip(), 40): return self.reply(429, {'ok': False}) q = parse_qs(urlparse(self.path).query) inv = invitado(q.get('n', [''])[0], q.get('f', [''])[0]) return self.reply(200, {'ok': True, 'nombre': inv[0], 'cita': inv[1]} if inv else {'ok': False}) def invitados(self): try: n = int(self.headers.get('Content-Length', '0')) if n <= 0 or n > 128 * 1024: return self.reply(413, {'ok': False, 'error': 'La ficha es demasiado larga.'}) d = json.loads(self.rfile.read(n)) except Exception: return self.reply(400, {'ok': False, 'error': 'Petición no válida.'}) if d.get('web') or (isinstance(d.get('t'), (int, float)) and d['t'] < 20000): return self.reply(200, {'ok': True}) # trampa o demasiado rápido: no se envía if limited(guest_hits, self.client_ip(), 8): return self.reply(429, {'ok': False, 'error': 'Demasiados envíos. Inténtalo más tarde.'}) inv = invitado(d.get('n'), d.get('f')) nombre, ts = inv if inv else (None, None) if not nombre: return self.reply(403, {'ok': False, 'error': 'Enlace no válido o caducado.'}) if d.get('consent') is not True: return self.reply(400, {'ok': False, 'error': 'Debes aceptar las condiciones de participación.'}) resp = d.get('resp') if isinstance(d.get('resp'), dict) else {} email = clean(d.get('email'), 200) if not email or not EMAIL_RE.match(email): return self.reply(400, {'ok': False, 'error': 'El correo no parece válido.'}) lineas, seccion = [], None for q in GUEST_Q: ans = str(resp.get(q['id']) or '').replace('\r', '').strip()[:3000] if q['req'] and not ans: return self.reply(400, {'ok': False, 'error': 'Faltan respuestas obligatorias.'}) if q['sec'] != seccion: seccion = q['sec'] lineas += ['', f'== {seccion} =='] lineas += ['', f"• {q['t']}", ' ' + (ans.replace('\n', '\n ') if ans else '(sin respuesta)')] cabecera = [f'Ficha del invitado: {nombre}', 'Recibida: ' + time.strftime('%d/%m/%Y %H:%M', time.localtime()), f'Cita: {cita_txt(ts)}', f'Correo del invitado: {email}', 'Aceptó las condiciones de participación y la política de privacidad: sí'] try: send(f'[Invitado] {clean(nombre, 80)} · ficha del podcast', email, '\n'.join(cabecera + lineas) + '\n') except RuntimeError: return self.reply(503, {'ok': False, 'error': 'El formulario no está disponible ahora mismo.'}) except Exception as exc: print('error enviando ficha:', type(exc).__name__, file=sys.stderr, flush=True) return self.reply(502, {'ok': False, 'error': 'No se pudo enviar. Inténtalo más tarde.'}) key = (nombre, ts) if key not in confirmados: # un solo recordatorio por invitado y cita confirmados.add(key) try: send('Tu cita en Anda Que No Sabes Ná', None, f'Hola, {nombre}:\n\nGracias por rellenar tu ficha. Recuerda que grabamos el ' f'{cita_txt(ts)}, y te enviaremos por correo la dirección del estudio.\n\n' 'Si no puedes venir o necesitas cambiar algo, responde a este correo.\n\n' 'Un abrazo,\nÁlex y Juan\nAnda Que No Sabes Ná\nhttps://andaquenosabesna.com\n', to=email) except Exception as exc: confirmados.discard(key) print('error enviando recordatorio:', type(exc).__name__, file=sys.stderr, flush=True) self.reply(200, {'ok': True}) def do_POST(self): if self.path.split('?')[0] == '/api/invitados': return self.invitados() if self.path not in ('/api/contacto', '/api/patrocinio'): return self.reply(404, {'ok': False}) patro = self.path == '/api/patrocinio' 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) mensaje = str(d.get('mensaje') or '').replace('\r', '').strip()[:3000] bad = {'ok': False, 'error': 'Revisa los campos del formulario.'} if not nombre or not EMAIL_RE.match(email) or len(mensaje) < 10 or d.get('consent') is not True: return self.reply(400, bad) if patro: empresa, sitio = clean(d.get('empresa'), 120), clean(d.get('sitio'), 200) tipo = d.get('tipo') if d.get('tipo') in TIPOS else 'otro' if not empresa or (sitio and not re.match(r'^https?://', sitio)): return self.reply(400, bad) subject = f'[Web AQNSN · Patrocinio] {empresa} · {nombre}' body = (f'Empresa: {empresa}\nContacto: {nombre}\nEmail: {email}\nWeb: {sitio or "—"}\n' f'Interés: {TIPOS[tipo]}\n\n{mensaje}\n') else: motivo = d.get('motivo') if d.get('motivo') in MOTIVOS else 'otro' subject = f'[Web AQNSN] {MOTIVOS[motivo]} · {nombre}' body = f'Motivo: {MOTIVOS[motivo]}\nNombre: {nombre}\nEmail: {email}\n\n{mensaje}\n' try: send(subject, email, body) 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()