"""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 hmac as _hmac import json import os from urllib.parse import urlparse, parse_qs, quote 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}' def cita_corta(ts): d = datetime.datetime.fromtimestamp(ts, ZoneInfo('Europe/Madrid')) return f'el {DIAS[d.weekday()]} {d.day} de {MESES[d.month - 1]} 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 def _gen_token(offset=0): """Token HMAC diario para la página generadora. Válido 48h (slot actual + anterior).""" key = (os.environ.get('LINK_PASS') or '').encode() if not key: return None slot = str(int(time.time()) // 86400 - offset).encode() return _hmac.new(key, slot, 'sha256').hexdigest() def parse_dias(d_str): """Valida una lista de timestamps separados por comas. Devuelve lista de ints o None.""" if not d_str: return None parts = str(d_str).replace(' ', '').split(',') if not 1 <= len(parts) <= 10: return None today = datetime.date.today() result = [] for p in parts: try: ts = int(p) day = datetime.datetime.fromtimestamp(ts, datetime.timezone.utc).date() if not (-1 <= (day - today).days <= 60): return None result.append(ts) except (ValueError, OverflowError, OSError): return None return result if result 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): path = urlparse(self.path).path if path == '/api/gen-verify': tok = parse_qs(urlparse(self.path).query).get('t', [''])[0] valid = tok and tok in filter(None, [_gen_token(0), _gen_token(1)]) return self.reply(200, {'ok': bool(valid)}) if 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) n_raw = q.get('n', [''])[0] f_raw = q.get('f', [''])[0] d_raw = q.get('d', [''])[0] if f_raw: inv = invitado(n_raw, f_raw) return self.reply(200, {'ok': True, 'nombre': inv[0], 'cita': inv[1]} if inv else {'ok': False}) if d_raw: n = ' '.join(str(n_raw or '').split()) if not n or not NAME_RE.match(n): return self.reply(200, {'ok': False}) dias = parse_dias(d_raw) if not dias: return self.reply(200, {'ok': False}) return self.reply(200, {'ok': True, 'nombre': n, 'dias': [{'ts': ts, 'txt': cita_txt(ts)} for ts in dias]}) return self.reply(200, {'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.'}) f_raw = str(d.get('f') or '').strip() d_raw = str(d.get('d') or '').strip() dia_raw = str(d.get('dia') or '').strip() if f_raw: inv = invitado(d.get('n'), f_raw) nombre, ts = inv if inv else (None, None) elif d_raw and dia_raw: n_val = ' '.join(str(d.get('n') or '').split()) if not n_val or not NAME_RE.match(n_val): nombre, ts = None, None else: dias = parse_dias(d_raw) try: dia_ts = int(dia_raw) except (ValueError, TypeError): dia_ts = None if dias and dia_ts is not None and dia_ts in dias: nombre, ts = n_val, dia_ts else: nombre, ts = None, None else: nombre, ts = 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.'}) tel = ' '.join(str(d.get('tel') or '').split()) if not re.match(r'^\+?[\d ().-]{7,20}$', tel): return self.reply(400, {'ok': False, 'error': 'El teléfono 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}', f'Teléfono móvil: {tel}', '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) lugar = os.environ.get('STUDIO_ADDRESS', '').strip() mapa = os.environ.get('STUDIO_MAP', '').strip() or 'https://www.google.com/maps/search/?api=1&query=' + quote(lugar) # solo en el servidor, nunca en el código if key not in confirmados: # un solo recordatorio por invitado y cita confirmados.add(key) try: send(f'¡Grabamos {cita_corta(ts)}!', None, f'Hola, {nombre}:\n\n¡Ya tenemos tu ficha, gracias! Ahora toca lo bueno: grabamos {cita_corta(ts)}.\n\n' + (f'Nos vemos en {lugar}.\nCómo llegar: {mapa}\n\n' if lugar else 'En breve te mandamos por correo electrónico la dirección del estudio.\n\n') + 'Ven con ganas de contar historias, que de lo demás nos encargamos nosotros. ' 'Si te surge cualquier imprevisto, responde a este correo y lo arreglamos.\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): p = self.path.split('?')[0] if p == '/api/gen-auth': try: n = int(self.headers.get('Content-Length', '0')) d = json.loads(self.rfile.read(n)) if 0 < n <= 1024 else {} except Exception: return self.reply(400, {'ok': False}) link_pass = os.environ.get('LINK_PASS', '') if not link_pass: return self.reply(503, {'ok': False, 'error': 'No configurado.'}) if _hmac.compare_digest(str(d.get('pass', '')), link_pass): return self.reply(200, {'ok': True, 'token': _gen_token(0)}) return self.reply(401, {'ok': False, 'error': 'Contraseña incorrecta.'}) if p == '/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()