364 lines
16 KiB
Python
364 lines
16 KiB
Python
"""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()
|