#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Versell - nodo rendezvous (conector) en Python puro.

Reimplementacion del nodo Kotlin/Ktor con el MISMO protocolo, para que la app
Versell funcione sin cambios. El nodo NO puede descifrar nada: solo mueve sobres
cifrados de extremo a extremo. Sin cuentas, sin admin.

Requisitos: solo Python 3.8+ (biblioteca estandar). Sin pip.
  - La verificacion Ed25519 va en Python puro; si estan instaladas 'cryptography'
    o 'pynacl', se usan automaticamente (mucho mas rapidas).

Variables de entorno:
  PORT             puerto de escucha (por defecto 8080)
  VERSELL_DATA     carpeta de datos (por defecto ./data junto a este archivo)
  VERSELL_BIND     interfaz de escucha (por defecto 0.0.0.0; 127.0.0.1 tras proxy TLS)

  --- Endurecimiento (todas opcionales, con valores por defecto sensatos) ---
  VERSELL_RATE_RPM        peticiones/min por IP (por defecto 240; 0 = sin limite)
  VERSELL_MAX_CONN        conexiones concurrentes maximas (por defecto 256)
  VERSELL_MAILBOX_MAX     sobres maximos por buzon (por defecto 500)
  VERSELL_ENVELOPE_TTL_H  horas que se guarda un sobre sin recoger (por defecto 336 = 14d)
  VERSELL_FILES_TTL_H     horas que se guarda un blob subido (por defecto 72 = 3d)
  VERSELL_FILES_MAX_MB    tamaño total maximo de la carpeta de archivos (por defecto 2048)
  VERSELL_TRUST_PROXY     1 si hay un reverse proxy delante (usa X-Forwarded-For; por defecto 1)

  --- GIFs/stickers vía KLIPY (opcional; sin VERSELL_KLIPY_KEY, el endpoint da 503) ---
  VERSELL_KLIPY           1/0, activa/desactiva el proxy (por defecto 1; sin efecto sin clave)
  VERSELL_KLIPY_KEY       clave de API de KLIPY (NUNCA en el repo; drop-in systemd en cada VPS)
  VERSELL_KLIPY_CACHE_TTL_S     segundos que se cachea cada búsqueda/trending (por defecto 600)
  VERSELL_KLIPY_MAX_PER_HOUR   tope propio de llamadas/hora a KLIPY por nodo (por defecto 40;
                                la clave real está compartida entre los 2 VPS y su límite es 100/h)

  --- Promo: puntos canjeables por bitcoin (opcional; sin VERSELL_PROMO_ENABLED, 503) ---
  VERSELL_PROMO_ENABLED         1 = activa los endpoints /promo/*. Es el interruptor.
  VERSELL_PROMO_DAILY_CAP_SATS  tope GLOBAL de sats/dia para TODO el mundo junto. OPCIONAL y
                                APAGADO por defecto (0 o sin definir = sin tope global; mandan
                                solo los limites por persona). Ponerlo >0 lo reactiva.
                                Este nodo SOLO lleva la contabilidad (ledger + cola de pagos
                                pendientes); el pago real de bitcoin lo procesa aparte
                                usb/promo_pay.py, el UNICO sitio con la clave privada -- ver ese
                                fichero y el README §14 para el porque de la separacion.
                                Constantes en codigo (no env, iguales para todo el mundo):
                                PROMO_MIN_WITHDRAW_POINTS=125 (1 EUR, minimo para PEDIR un retiro),
                                PROMO_MAX_WITHDRAW_PER_DAY_POINTS=250 (2 EUR, tope de RETIRO por
                                persona y dia -- el sobrante se paga en dias siguientes, no se
                                pierde), PROMO_DAILY_CAP_POINTS=375 (3 EUR, tope de GANANCIA por
                                persona y dia -- independiente del de retiro, sin tocar).

Uso:
  python node.py            -> arranca el nodo
  python node.py --selftest -> comprueba la criptografia y el protocolo y sale
"""

import base64
import collections
import hashlib
import hmac
import html
import json
import os
import re
import struct
import subprocess
import sys
import threading
import time
import unicodedata
import urllib.request
import uuid
import xml.etree.ElementTree as ET
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from html.parser import HTMLParser as _HTMLParser
from urllib.parse import urlparse, parse_qs, urljoin, quote
import html as _htmllib
import ipaddress as _ipaddr
import socket as _socket

SIG_CONTEXT = b"versell-v1"
HKDF_INFO = b"versell-e2e-v1"  # no usado por el nodo; documental
MAX_FILE_BYTES = 50 * 1024 * 1024
MAX_SKEW_MS = 5 * 60 * 1000
WS_GUID = b"258EAFA5-E914-47DA-95CA-C5AB0DC85B11"

# --- Limites de endurecimiento (configurables por entorno; ver docstring) ---
def _envint(name, default):
    try:
        return int(os.environ.get(name, default))
    except Exception:
        return default

# Cuerpos JSON (announce/send/inbox/ack): los sobres solo llevan metadatos +
# ct pequeño (los archivos van aparte por /files), asi que 1 MB es muy holgado.
MAX_JSON_BYTES = 1 * 1024 * 1024
# Trama WebSocket entrante: el cliente casi no envia; corta abusos de memoria.
MAX_WS_FRAME_BYTES = 1 * 1024 * 1024
# Conexiones WS simultaneas por peer (varios dispositivos, no miles).
MAX_WS_PER_PEER = 8
# Anti-slowloris: corta sockets que no completan la peticion a tiempo.
SOCKET_TIMEOUT_S = 30
# WS legitimo: sin trafico del cliente durante mucho -> se recicla.
WS_IDLE_TIMEOUT_S = 300

# --- Mensajes PROGRAMADOS (el nodo retiene el sobre hasta la hora elegida) ---
# El sobre llega al nodo YA cifrado y firmado como cualquier otro; lo unico nuevo es
# el campo `notBefore` (epoch ms) que dice a partir de cuando puede entregarse. Hasta
# entonces no sale por /inbox, ni por WebSocket, ni dispara aviso de UnifiedPush.
# `notBefore` NO va firmado, y no hace falta que lo vaya: el nodo controla la entrega
# de todas formas (podria retener cualquier sobre), asi que no gana ningun poder nuevo.
SCHED_ENABLED = os.environ.get("VERSELL_SCHED", "1") not in ("0", "false", "no")
SCHED_MAX_MS = _envint("VERSELL_SCHED_MAX_DAYS", 365) * 24 * 3600 * 1000

RATE_RPM = _envint("VERSELL_RATE_RPM", 240)
MAX_CONN = _envint("VERSELL_MAX_CONN", 256)
MAILBOX_MAX = _envint("VERSELL_MAILBOX_MAX", 500)
ENVELOPE_TTL_MS = _envint("VERSELL_ENVELOPE_TTL_H", 336) * 3600 * 1000
FILES_TTL_S = _envint("VERSELL_FILES_TTL_H", 72) * 3600
FILES_MAX_BYTES = _envint("VERSELL_FILES_MAX_MB", 2048) * 1024 * 1024
TRUST_PROXY = os.environ.get("VERSELL_TRUST_PROXY", "1") == "1"
MAINTENANCE_EVERY_S = 600  # purga periodica de buzones y archivos caducados

# Version de esta build del nodo Python. NO es el protocolo del sec. 4 (que no
# cambia byte a byte); solo identifica la implementacion. La consume "GET /".
NODE_VERSION = "1.41.0-py"
# Momento de arranque (epoch ms): lo unico que hace falta para que la herramienta de
# operador pueda decir "lleva X horas en pie" sin guardar ningun historial.
_STARTED_AT = int(time.time() * 1000)

# --- Ocultacion ante sondeo activo (resistencia a censura) ---
# OPT-IN: VERSELL_DECOY=1 (por defecto OFF -> comportamiento IDENTICO al de hoy,
# ningun cliente desplegado se entera). Algunos cortafuegos nacionales hacen
# "sondeo activo": se conectan al servidor y miran como responde para identificar
# proxies/nodos. Con DECOY activo, una peticion SIN autenticar no delata que esto
# es un nodo Versell:
#   - "GET /" y las rutas GET desconocidas -> 404 generico estilo nginx (sin la
#     cabecera {"service":"versell-rendezvous"} ni el JSON {"error":...}, sin CORS).
#   - El header Server no revela Python/BaseHTTP.
# El operador sigue viendo la salud en "/" presentando la clave de admin
# (X-Admin-Key o ?key=). NO toca el protocolo E2E del sec. 4 y la app funciona
# igual (los endpoints funcionales responden normal). Reduce la huella MAS obvia
# (banner + errores); el gateo autenticado de /lookup//online//news es la fase 2.
DECOY = os.environ.get("VERSELL_DECOY", "0") not in ("0", "false", "no", "")

# --- Fase 2: gateo autenticado de los GET que hoy van SIN auth ---
# OPT-IN: VERSELL_STRICT_GET=1 (por defecto OFF). Con el activo, /lookup, /online,
# /news, /news/img y /files exigen un token firmado Ed25519 del solicitante
# ANUNCIADO (cabeceras X-Versell-Peer/Ts/Sig, mismo framing que /inbox:
# auth_data + verify_auth + frescura). Sin token valido -> 404 señuelo, asi que
# cierra la ultima via por la que un sondeo (que ya conozca las rutas del repo
# publico) distinguia el nodo. La app firma esos GET siempre y en silencio; un
# nodo SIN este flag los ignora -> TRANSICION sin dejar a nadie fuera. Por
# defecto OFF -> los GET siguen abiertos (clientes viejos y Versell Web incluidos).
# ⚠️ Encenderlo rechaza a quien NO firme (apps pre-fase2 y el cliente web) ->
# hacerlo SOLO tras adopcion alta o en nodos endurecidos opt-in. No toca el sec. 4.
STRICT_GET = os.environ.get("VERSELL_STRICT_GET", "0") not in ("0", "false", "no", "")

# --- Transportes conmutables: relleno de respuestas (anti-analisis de tamaño) ---
# Un observador de red no ve el contenido (TLS + E2E), pero SI el TAMAÑO de cada
# respuesta -> el largo de un /inbox o un /lookup filtra cuantos/que sobres hay.
# Con el "modo sigiloso" de la app, el cliente manda la cabecera `X-Versell-Pad: 1`
# y el nodo rellena el JSON hasta el siguiente escalon de tamaño (bucket) con un
# campo `_pad` IGNORADO (la app usa ignoreUnknownKeys, como con `_rx`), de bytes
# ALEATORIOS para que un gzip intermedio no lo comprima y deshaga el relleno.
# Es OPT-IN extremo a extremo y por-peticion: solo se rellena a quien lo pide, asi
# que NO penaliza el ancho de banda del resto. Un nodo viejo ignora la cabecera
# (la app sigue funcionando) y un cliente viejo nunca la manda. No toca el sec. 4.
# VERSELL_PAD=0 lo desactiva en el nodo (por defecto ON, pero inerte sin la cabecera).
PAD_ENABLED = os.environ.get("VERSELL_PAD", "1") not in ("0", "false", "no", "")
# Escalones de tamaño del cuerpo JSON (bytes). Un cuerpo se rellena hasta el primero
# que lo supere; por encima del ultimo, no se rellena (respuestas ya grandes, como
# /news con imagenes, no tiene sentido cuantizarlas y costaria ancho de banda).
PAD_BUCKETS = (512, 2048, 8192, 32768, 131072)


def pad_json_body(body, want_pad):
    """Rellena un cuerpo JSON hasta el siguiente bucket con un campo `_pad` de bytes
    ALEATORIOS (no comprimibles), SOLO si `want_pad` y el cuerpo es un objeto JSON
    (empieza por '{' y acaba en '}'). El campo se inyecta antes de la '}' final para
    no reparsear ni reserializar. Idempotente respecto al contenido: la app ignora
    `_pad` (ignoreUnknownKeys). Modulo aparte para poder probarlo en --selftest."""
    if not want_pad:
        return body
    if not body or body[:1] != b"{" or body[-1:] != b"}":
        return body  # una lista u otra cosa: no se toca
    target = next((b for b in PAD_BUCKETS if b > len(body)), None)
    if target is None:
        return body  # ya mayor que el ultimo bucket: no cuantizar
    sep = b"" if body[-2:-1] == b"{" else b","  # objeto vacio {} -> sin coma inicial
    # partes fijas que añadimos: sep + '"_pad":"' + pad + '"}' (y quitamos la '}' final).
    fixed = len(sep) + len(b'"_pad":"') + len(b'"}')
    fill = max(0, target - (len(body) - 1) - fixed)
    pad = os.urandom((fill + 1) // 2).hex()[:fill].encode("ascii")
    return body[:-1] + sep + b'"_pad":"' + pad + b'"}'


# --- Llamadas de audio 1-a-1 (WebRTC): credenciales TURN/STUN efimeras ---
# El endpoint GET /turncreds (SIEMPRE autenticado con la firma del solicitante,
# scope "turn") devuelve credenciales de corta vida para que dos moviles tras
# CGNAT puedan hablar por un relay TURN propio. El nodo NO ve el audio (va
# cifrado DTLS-SRTP extremo a extremo); el relay solo mueve paquetes opacos.
# Config (todo opt-in; sin secreto -> el endpoint devuelve solo STUN o vacio):
#   VERSELL_TURN_SECRET : el mismo 'static-auth-secret' de coturn (REST auth).
#   VERSELL_TURN_URLS   : coma-separado, p. ej. "turn:node1.dicra.io:3478?transport=udp,turns:node1.dicra.io:5349".
#   VERSELL_STUN_URLS   : coma-separado (por defecto, el propio coturn en :3478).
#   VERSELL_TURN_TTL    : validez en segundos de las credenciales (def. 3600).
# --- STUN propio embebido (sin coturn, sin terceros) ---
# Sin STUN, WebRTC solo genera candidatos "host" (IPs locales) y las llamadas SOLO
# funcionan dentro de la misma LAN: el movil no llega a saber con que IP:puerto le
# ve el mundo, asi que no hay hole-punching posible. Antes esto lo tenia que dar
# coturn; ahora el propio nodo responde Binding Requests (RFC 5389) en UDP, que son
# 60 lineas y cero dependencias. Con esto las llamadas cruzan entre redes distintas
# siempre que el NAT lo permita (mayoria de fibra/wifi domesticas).
# ⚠️ NO sustituye al relay TURN: tras CGNAT estricto (habitual en datos moviles)
# sigue haciendo falta coturn. Poner VERSELL_STUN_PORT=0 lo desactiva (p. ej. si
# coturn ya escucha en ese puerto y hace el STUN por su cuenta).
STUN_PORT = _envint("VERSELL_STUN_PORT", 3478)

TURN_SECRET = os.environ.get("VERSELL_TURN_SECRET", "")
TURN_URLS = [u.strip() for u in os.environ.get("VERSELL_TURN_URLS", "").split(",") if u.strip()]
STUN_URLS = [u.strip() for u in os.environ.get("VERSELL_STUN_URLS", "").split(",") if u.strip()]
TURN_TTL = int(os.environ.get("VERSELL_TURN_TTL", "3600"))

# --- Modo lector "Navegar usando Versell" (POST /fetch) ---
# El nodo descarga una pagina web, la SANEA (quita JavaScript, CSS del sitio,
# formularios, inputs, botones y ENLACES) y devuelve HTML autocontenido con solo
# TEXTO + IMAGENES (estas incrustadas como data: URIs, con topes). El movil del
# usuario NUNCA contacta con el sitio (lo hace el nodo, que esta fuera de la red
# restringida) ni ejecuta scripts. Requiere token firmado (scope "fetch") -> nunca
# es un proxy abierto; protegido contra SSRF (bloquea IPs privadas), con topes de
# tamaño/tiempo/numero de imagenes. Opt-out con VERSELL_FETCH=0. No toca el sec. 4.
FETCH_ENABLED = os.environ.get("VERSELL_FETCH", "1") not in ("0", "false", "no", "")
FETCH_HTML_MAX = _envint("VERSELL_FETCH_HTML_KB", 1500) * 1024   # HTML de origen
FETCH_IMG_MAX = 300 * 1024        # por imagen antes de reducir
FETCH_IMG_COUNT = 20              # nº maximo de imagenes incrustadas
FETCH_INLINE_TOTAL = 4 * 1024 * 1024   # tope total de imagenes incrustadas
FETCH_TIMEOUT = 12
# Modo ARTICULO: quedarse solo con el cuerpo de la noticia (+ su titular) en vez
# de con todo el texto de la pagina. El saneado de arriba quita scripts y enlaces,
# pero conservaba menus, "lo mas leido", pies y avisos de cookies: justo el ruido
# que sobra en el lector. Si la heuristica no encuentra un cuerpo solido se cae al
# saneado de la pagina entera (mejor de mas que de menos). VERSELL_FETCH_ARTICLE=0
# vuelve al comportamiento anterior.
FETCH_ARTICLE = os.environ.get("VERSELL_FETCH_ARTICLE", "1") not in ("0", "false", "no", "")
# Suelo de cordura, no un filtro: quien de verdad descarta menus y sidebars es la
# densidad de enlaces. 80 caracteres son ~2 frases chinas (un 简讯 corto ya cuenta);
# subirlo dejaria las noticias breves cayendo al saneado ruidoso de pagina entera.
FETCH_ARTICLE_MIN = _envint("VERSELL_FETCH_ARTICLE_MIN", 80)    # chars minimos del cuerpo

# --- UnifiedPush: "hay correo" a un endpoint elegido por el usuario ---
# La app registra (firmado, scope "push") la URL de SU distribuidor UnifiedPush
# (ntfy, NextPush...); cuando /send deposita un sobre para ese peer y NO tiene un
# WebSocket vivo, el nodo hace un POST minusculo al endpoint con un cuerpo
# CONSTANTE ("new"): NUNCA contenido, ni remitente, ni peerIds -> el servidor de
# push solo aprende CUANDO hay actividad (metadato de tiempo, declarado en el
# README/sec.14). La app, al despertar, lee su buzon por el camino normal E2E.
# Sin registro (o VERSELL_PUSH=0) todo sigue exactamente como hoy.
#   VERSELL_PUSH=0              : desactiva registro y avisos.
#   VERSELL_PUSH_ALLOW_HTTP=1   : permite endpoints http:// (autoalojados; por
#                                 defecto solo https). El SSRF-guard aplica SIEMPRE
#                                 (nunca a IPs privadas/loopback/metadatos).
#   VERSELL_PUSH_MIN_INTERVAL_S : throttle de avisos por peer (def. 8 s).
#   VERSELL_PUSH_TTL_DAYS       : caducidad del endpoint sin refrescar (def. 30).
PUSH_ENABLED = os.environ.get("VERSELL_PUSH", "1") not in ("0", "false", "no", "")
PUSH_ALLOW_HTTP = os.environ.get("VERSELL_PUSH_ALLOW_HTTP", "0") not in ("0", "false", "no", "")
PUSH_MIN_INTERVAL_S = _envint("VERSELL_PUSH_MIN_INTERVAL_S", 8)
PUSH_TTL_DAYS = _envint("VERSELL_PUSH_TTL_DAYS", 30)
PUSH_TIMEOUT = 5
PUSH_MAX_URL = 512

# --- Prekeys (X3DH ligero, 1.32.0-py) ---
# Un bundle por peer: clave X25519 publica + su firma Ed25519. Sirve para que el PRIMER
# mensaje a alguien ya vaya con Double Ratchet, en vez de en v1 (sin forward secrecy)
# hasta que completa el handshake. El nodo solo almacena y sirve: no puede falsificarlo
# (va firmado y lo verifica el destinatario con la clave que ya conoce del contacto), y
# si lo NIEGA el emisor cae al handshake de siempre. Apagable con VERSELL_PREKEY=0.
PREKEY_ENABLED = os.environ.get("VERSELL_PREKEY", "1") not in ("0", "false", "no", "")
PREKEY_TTL_DAYS = _envint("VERSELL_PREKEY_TTL_DAYS", 90)
PREKEY_MAX_ID = 64
PREKEY_MAX_B64 = 128     # una X25519 en base64 son 44 caracteres; el tope es holgura

# --- Ocultacion del nodo ante sondeo (VERSELL_DECOY, mas abajo) ---
# ADMIN_KEY ya NO abre ningun panel de operador (el antiabuso por IP se retiro el
# 2026-08-02: costaba privacidad -nombre visible, version, pais/ciudad de cada
# usuario, 6 meses- y no aportaba nada que el rate-limit no hiciera ya). Lo unico
# que le queda: con VERSELL_DECOY=1, dejar ver la salud en GET / presentando esta
# clave (si no, un sondeo ve un 404 generico). No toca el protocolo E2E del sec. 4.
ADMIN_KEY = os.environ.get("VERSELL_ADMIN_KEY", "")  # vacio = sin clave -> GET / no distingue sondeo de operador

# --- Rescate de fondos (POST /promo/sweep) ---
# SEGUNDA llave, aparte de la de operador, para el unico endpoint que mueve dinero. Sin
# esta variable el endpoint NO EXISTE (404): se arma a mano cuando se quiere poder rescatar
# desde el movil, y se desarma quitandola. La de operador ya no basta por si sola porque
# viaja dentro del APK de DicraAdmin.
SWEEP_PASS = os.environ.get("VERSELL_SWEEP_PASS", "").strip()
SWEEP_PYTHON = os.environ.get("VERSELL_SWEEP_PYTHON", "/opt/versell/promo-venv/bin/python")
# Ensayo: con esto a 1, el endpoint recorre TODO el camino (auth, lanzar el script, leer el
# saldo por Electrum, devolver la salida) pero NO pasa `--si-vaciar`, asi que no mueve un
# satoshi. Existe para poder comprobar que el boton funciona de verdad sin gastar dinero
# real, que es lo unico que no se puede probar dos veces.
SWEEP_DRYRUN = os.environ.get("VERSELL_SWEEP_DRYRUN", "0") in ("1", "true", "yes")
SWEEP_TIMEOUT_S = _envint("VERSELL_SWEEP_TIMEOUT_S", 180)   # Electrum + difusion, con holgura

# --- Promo: puntos por usar la app, canjeables por bitcoin (2026-08-06) ---
# El nodo SOLO lleva la contabilidad (ledger de puntos por peer + cola de pagos
# pendientes, en disco, JSON); NUNCA ve, firma ni difunde nada de bitcoin -- eso
# vive aparte, en usb/promo_pay.py, el UNICO proceso que importa una libreria de
# Bitcoin y el UNICO que ve la clave privada. Separacion deliberada (ver ese
# fichero y el README §14): node.py sigue siendo "sin pip", y el codigo que
# toca dinero real queda fuera del proceso que retransmite los sobres cifrados
# de todo el mundo. Sin VERSELL_PROMO_DAILY_CAP_SATS, /promo/* responde 503
# (mismo patron que /klipy/* sin VERSELL_KLIPY_KEY).
#
# TOPE GLOBAL: OPCIONAL y APAGADO por defecto desde 2026-08-08, por decision de
# Jorge ("no quiero un tope global, solo personal"). 0 o sin definir = sin tope
# global: solo mandan los limites POR PERSONA (minimo de retiro, tope de ganancia
# diaria y tope de retiro diario, aqui debajo).
#
# ⚠️ Lo que se pierde al apagarlo, dicho sin adornos: en Dicra la identidad es
# gratis e ilimitada (sin cuentas, por diseño), asi que un tope POR PERSONA no
# frena a quien fabrique identidades falsas -- cada una con su propio cupo. El
# tope global era el unico techo absoluto de cuanto podia salir de la cartera en
# un dia, pasara lo que pasara. Sin el, el limite real es lo que haya en la
# cartera. Se deja el mecanismo entero por si se quiere recuperar: basta volver a
# poner VERSELL_PROMO_DAILY_CAP_SATS con un valor > 0.
PROMO_DAILY_CAP_SATS = _envint("VERSELL_PROMO_DAILY_CAP_SATS", 0)
# El encendido de Promo ya NO depende del tope global (antes era la misma
# variable, asi que quitar el tope apagaba Promo entero sin querer). Se acepta
# tambien el criterio viejo para no dejar a ciegas a un nodo que aun no tenga la
# variable nueva puesta.
PROMO_ENABLED = (os.environ.get("VERSELL_PROMO_ENABLED", "").strip().lower()
                 in ("1", "true", "si", "yes")) or PROMO_DAILY_CAP_SATS > 0
PROMO_MIN_WITHDRAW_POINTS = 125     # 1 EUR (25 puntos = 20 centimos) -- minimo para PEDIR un retiro
PROMO_DAILY_CAP_POINTS = 375        # 3 EUR por persona y dia -- tope de GANANCIA (cuanto se puede acumular)
PROMO_MAX_WITHDRAW_PER_DAY_POINTS = 250  # 2 EUR por persona y dia -- tope de RETIRO, independiente del de
# arriba: aunque el saldo acumulado sea mayor (varios dias de uso), cada retiro paga como mucho esto y
# el resto se queda en el ledger para el dia siguiente -- un "goteo", no un bloqueo total con saldo grande.
# contact = añadir a alguien; friend = que TE añadan a ti (reciproco, automatico
# al recibir un "intro" de alguien nuevo); share = compartir el APK; call =
# hacer una llamada de audio/video que llega a conectar (no solo marcar);
# update = instalar una version nueva de la app; compensation = reparacion UNICA
# (una sola vez por peer en toda la vida del ledger, ver PromoStore.claim) por los
# puntos que se perdieron antes de 1.0.165; media = mandar una foto o un video
# (vale mas que un mensaje suelto: cuesta mas y es lo que de verdad se comparte);
# story = publicar una historia en "Hoy"; news = abrir una noticia para leerla.
PROMO_POINTS = {"contact": 10, "group": 2, "message": 1, "share": 7, "friend": 10, "call": 3,
                "update": 8, "compensation": 120, "media": 2, "story": 5, "news": 1}
# Tope de ritmo por tipo (claims/hora por peer, en memoria): ni un cliente
# modificado puede ráfagar mas rapido que un uso humano razonable. `news` va
# especialmente corto: abrir titulares es lo mas barato de repetir de toda la
# lista, asi que 10/h son de sobra para leer y muy poco para farmear.
PROMO_CLAIM_MAX_PER_HOUR = {"contact": 20, "group": 10, "message": 60, "share": 5, "friend": 20, "call": 10,
                            "update": 3, "compensation": 2, "media": 30, "story": 6, "news": 10}

# =====================================================================
#  Ed25519 (verificacion). Backend rapido si existe; si no, Python puro.
# =====================================================================

def _pure_ed25519():
    b = 256
    q = 2 ** 255 - 19
    l = 2 ** 252 + 27742317777372353535851937790883648493

    def H(m):
        return hashlib.sha512(m).digest()

    def expmod(bb, e, m):
        if e == 0:
            return 1
        t = expmod(bb, e // 2, m) ** 2 % m
        if e & 1:
            t = (t * bb) % m
        return t

    def inv(x):
        return expmod(x, q - 2, q)

    d = (-121665 * inv(121666)) % q
    I = expmod(2, (q - 1) // 4, q)

    def xrecover(y):
        xx = (y * y - 1) * inv(d * y * y + 1)
        x = expmod(xx, (q + 3) // 8, q)
        if (x * x - xx) % q != 0:
            x = (x * I) % q
        if x % 2 != 0:
            x = q - x
        return x

    By = (4 * inv(5)) % q
    Bx = xrecover(By)
    B = [Bx % q, By % q]

    def edwards(P, Q):
        x1, y1 = P
        x2, y2 = Q
        x3 = (x1 * y2 + x2 * y1) * inv(1 + d * x1 * x2 * y1 * y2)
        y3 = (y1 * y2 + x1 * x2) * inv(1 - d * x1 * x2 * y1 * y2)
        return [x3 % q, y3 % q]

    def scalarmult(P, e):
        if e == 0:
            return [0, 1]
        Q = scalarmult(P, e // 2)
        Q = edwards(Q, Q)
        if e & 1:
            Q = edwards(Q, P)
        return Q

    def bit(h, i):
        return (h[i // 8] >> (i % 8)) & 1

    def encodeint(y):
        bits = [(y >> i) & 1 for i in range(b)]
        return bytes(sum(bits[i * 8 + j] << j for j in range(8)) for i in range(b // 8))

    def encodepoint(P):
        x, y = P
        bits = [(y >> i) & 1 for i in range(b - 1)] + [x & 1]
        return bytes(sum(bits[i * 8 + j] << j for j in range(8)) for i in range(b // 8))

    def decodeint(s):
        return sum(2 ** i * bit(s, i) for i in range(0, b))

    def isoncurve(P):
        x, y = P
        return (-x * x + y * y - 1 - d * x * x * y * y) % q == 0

    def decodepoint(s):
        y = sum(2 ** i * bit(s, i) for i in range(0, b - 1))
        x = xrecover(y)
        if x & 1 != bit(s, b - 1):
            x = q - x
        P = [x, y]
        if not isoncurve(P):
            raise ValueError("punto fuera de la curva")
        return P

    def Hint(m):
        # Reduccion del hash SHA-512 COMPLETO (512 bits), no truncado a 256.
        h = H(m)
        return sum(2 ** i * bit(h, i) for i in range(2 * b))

    def verify(pub, sig, msg):
        if len(sig) != 64 or len(pub) != 32:
            return False
        try:
            R = decodepoint(sig[:32])
            A = decodepoint(pub)
            S = decodeint(sig[32:])
            h = Hint(sig[:32] + pub + msg)
            return scalarmult(B, S) == edwards(R, scalarmult(A, h))
        except Exception:
            return False

    def publickey(sk):
        h = H(sk)
        a = 2 ** (b - 2) + sum(2 ** i * bit(h, i) for i in range(3, b - 2))
        return encodepoint(scalarmult(B, a))

    def sign(sk, msg):
        h = H(sk)
        a = 2 ** (b - 2) + sum(2 ** i * bit(h, i) for i in range(3, b - 2))
        pk = encodepoint(scalarmult(B, a))
        r = Hint(h[b // 8:b // 4] + msg)
        R = scalarmult(B, r)
        S = (r + Hint(encodepoint(R) + pk + msg) * a) % l
        return encodepoint(R) + encodeint(S)

    return verify, sign, publickey


_PURE_VERIFY, _PURE_SIGN, _PURE_PUBKEY = _pure_ed25519()


def _make_verifier():
    try:
        from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey

        def v(pub, sig, msg):
            try:
                Ed25519PublicKey.from_public_bytes(pub).verify(sig, msg)
                return True
            except Exception:
                return False

        return v, "cryptography"
    except Exception:
        pass
    try:
        from nacl.signing import VerifyKey

        def v(pub, sig, msg):
            try:
                VerifyKey(pub).verify(msg, sig)
                return True
            except Exception:
                return False

        return v, "pynacl"
    except Exception:
        pass
    return _PURE_VERIFY, "python-puro"


VERIFY, VERIFY_BACKEND = _make_verifier()

# =====================================================================
#  Framing y helpers de protocolo (identicos al nodo Kotlin / a la app)
# =====================================================================

def b64d(s):
    if isinstance(s, str):
        s = s.encode()
    return base64.b64decode(s + b"=" * (-len(s) % 4))


def _consteq(a, b):
    """Comparacion en tiempo constante (evita fugas por timing en la clave admin)."""
    return hmac.compare_digest(str(a), str(b))


def _field(bb):
    return struct.pack(">I", len(bb)) + bb


def sig_data(frm, to, eph, nonce, ct):
    return b"".join(_field(x) for x in [SIG_CONTEXT, frm.encode(), to.encode(), eph, nonce, ct])


def announce_data(peer, xpub, edpub, ts):
    return b"".join(_field(x) for x in [b"announce", peer.encode(), xpub.encode(), edpub.encode(), str(ts).encode()])


def auth_data(scope, peer, ts):
    return b"".join(_field(x) for x in [scope.encode(), peer.encode(), str(ts).encode()])


def peer_id_for(xpub_b64, edpub_b64):
    raw = b64d(xpub_b64) + b64d(edpub_b64)
    digest = hashlib.sha256(raw).digest()
    return base64.urlsafe_b64encode(digest).rstrip(b"=").decode()


def fresh(ts):
    try:
        return abs(int(time.time() * 1000) - int(ts)) < MAX_SKEW_MS
    except Exception:
        return False


def verify_announce(a):
    try:
        if peer_id_for(a["x25519Pub"], a["ed25519Pub"]) != a["peerId"]:
            return False
        return VERIFY(b64d(a["ed25519Pub"]), b64d(a["sig"]),
                      announce_data(a["peerId"], a["x25519Pub"], a["ed25519Pub"], a["ts"]))
    except Exception:
        return False


def verify_envelope(env, sender_ed_b64):
    try:
        data = sig_data(env["from"], env["to"], b64d(env["ephPub"]), b64d(env["nonce"]), b64d(env["ct"]))
        return VERIFY(b64d(sender_ed_b64), b64d(env["sig"]), data)
    except Exception:
        return False


def verify_auth(scope, peer, ts, sig_b64, ed_b64):
    try:
        return VERIFY(b64d(ed_b64), b64d(sig_b64), auth_data(scope, peer, ts))
    except Exception:
        return False


def prekey_data(peer, pk_id, pk_b64, ts):
    """Framing firmado del BUNDLE de prekey (X3DH ligero, 1.32.0-py). Mismo estilo que el
    resto del sec. 4: campos con prefijo de longitud de 4 bytes big-endian.

    Esta firma NO es autenticacion de transporte: es la que verifica **el destinatario**
    con la clave Ed25519 que ya conoce del contacto (la tiene de su tarjeta/QR). Eso es lo
    que impide que el nodo cuele SU prekey y se ponga en medio: un nodo puede negar el
    bundle, pero no falsificarlo. El nodo tambien la verifica al publicar, pero solo para
    no almacenar basura -- su veredicto no es en el que confia nadie."""
    return b"".join(_field(x) for x in [b"prekey", peer.encode(), pk_id.encode(),
                                        pk_b64.encode(), str(ts).encode()])


def verify_prekey(bundle, ed_b64):
    try:
        return VERIFY(b64d(ed_b64), b64d(bundle["sig"]),
                      prekey_data(bundle["peerId"], bundle["pkId"], bundle["pk"], bundle["ts"]))
    except Exception:
        return False

# =====================================================================
#  Estado (directorio de claves + buzones + archivos), con persistencia
# =====================================================================

class Store:
    def __init__(self, data_dir):
        self.data_dir = data_dir
        self.files_dir = os.path.join(data_dir, "files")
        os.makedirs(self.files_dir, exist_ok=True)
        # Un fichero POR BUZON (1.39.0-py). Antes los buzones vivian dentro de
        # state.json y CADA /send reescribia el estado entero: en nodemax eso eran
        # 21,8 MB por mensaje (11.107 sobres acumulados por el TTL de 14 dias), con
        # el nodo al 65% de CPU y ~40% de peticiones fallando. Diagnosticado midiendo
        # produccion el 2026-08-21: el tamaño de state.json correlacionaba exactamente
        # con la latencia (nodemax 21,8 MB -> 1263 ms/envio; node1 6,3 MB -> 486 ms).
        # Ahora un /send escribe solo el buzon del destinatario (unos KB).
        self.mail_dir = os.path.join(data_dir, "mail")
        os.makedirs(self.mail_dir, exist_ok=True)
        self.state_file = os.path.join(data_dir, "state.json")
        self.lock = threading.RLock()
        self.directory = {}   # peerId -> {peerId,x25519Pub,ed25519Pub}  (lo que devuelve /lookup)
        self.mailboxes = {}   # peerId -> [envelope,...]
        # Metadatos de moderacion, SEPARADOS del directorio para que /lookup NO los
        # filtre nunca (ip/dispositivo son datos sensibles del operador).
        self.meta = {}        # peerId -> {lastSeen}   (desde 1.15.0-py, nada mas)
        # Conclusiones que el operador manda al reportante (obviado/aceptado/controversial).
        # UnifiedPush: peerId -> {"endpoint": url, "ts": ms del registro/refresco}.
        # Separado del directorio (nunca se expone por /lookup ni /admin/peers).
        self.push = {}
        # Prekeys X3DH ligero (1.32.0-py): peerId -> {"pkId","pk","ts","sig"}. Publico y
        # FIRMADO por su dueño: se sirve a quien lo pida (es lo que permite cifrar con
        # trinquete el PRIMER mensaje, sin esperar al handshake). Un solo bundle por peer,
        # rotado por el propio movil; no hay one-time prekeys (ver /prekey).
        self.prekeys = {}
        self._load()

    def _box_path(self, peer):
        """Ruta del fichero de buzon de un peer. El peerId es base64url sin relleno
        (A-Za-z0-9_-), pero se filtra igualmente: un peerId manipulado NO puede
        escaparse del directorio ni pisar otro fichero (path traversal)."""
        safe = re.sub(r"[^A-Za-z0-9_-]", "", peer or "")
        # Fail-closed: un peerId legitimo es EXACTAMENTE base64url sin relleno, asi que
        # si el filtrado ha cambiado algo (o viene vacio/larguisimo) no se toca el disco.
        if not safe or safe != peer or len(safe) > 128:
            return None
        return os.path.join(self.mail_dir, safe + ".json")

    def _save_box(self, peer):
        """Guarda SOLO el buzon de este peer (unos KB), no el estado entero. Un buzon
        vacio se borra en vez de dejar un fichero con `[]`. Llamar SIEMPRE con el lock
        tomado, como el resto de escrituras."""
        p = self._box_path(peer)
        if p is None:
            return
        box = self.mailboxes.get(peer)
        if not box:
            try:
                os.remove(p)
            except OSError:
                pass
            return
        tmp = p + ".tmp"
        with open(tmp, "w", encoding="utf-8") as f:
            json.dump(box, f)
        os.replace(tmp, p)

    def _load_boxes(self):
        """Carga todos los buzones desde mail/. Un fichero corrupto se ignora (se
        pierde ESE buzon, no el arranque del nodo)."""
        try:
            names = os.listdir(self.mail_dir)
        except OSError:
            return
        for name in names:
            if not name.endswith(".json") or name.endswith(".tmp"):
                continue
            peer = name[:-5]
            try:
                with open(os.path.join(self.mail_dir, name), "r", encoding="utf-8") as f:
                    box = json.load(f)
                if isinstance(box, list) and box:
                    self.mailboxes[peer] = box
            except Exception:
                continue

    def _load(self):
        legacy_boxes = None
        if os.path.isfile(self.state_file):
            try:
                with open(self.state_file, "r", encoding="utf-8") as f:
                    data = json.load(f)
                self.directory = data.get("directory", {})
                # MIGRACION (1.39.0-py): los buzones vivian aqui dentro. Si este
                # state.json todavia los trae, se reparten a mail/<peer>.json y se
                # reescribe el estado ya sin ellos. Se hace una sola vez; a partir de
                # ahi `_load_boxes` es la unica fuente. NO se pierde ningun sobre.
                legacy_boxes = data.get("mailboxes")
                self.meta = data.get("meta", {})
                # Minimización de datos: borrar de golpe, al arrancar, todo lo heredado
                # de versiones anteriores. `lastIp` nunca se guardó por usuario; `name` y
                # `deviceId` se guardaban para el panel de operador y desde 1.15.0-py ya
                # no tienen finalidad (ver record_meta). Purgar aquí y no solo en
                # record_meta es lo que limpia también a los peers que no vuelvan a
                # conectarse nunca: con un solo reinicio del servicio, state.json queda
                # sin un solo nickname ni identificador de dispositivo.
                for _m in self.meta.values():
                    if isinstance(_m, dict):
                        _m.pop("lastIp", None)
                        _m.pop("name", None)
                        _m.pop("deviceId", None)
                self.push = data.get("push", {})
                self.prekeys = data.get("prekeys", {})
                # `abuse` y `blocklist` de versiones anteriores (contador antiabuso por
                # IP, retirado el 2026-08-02) se ignoran a proposito: no se cargan a
                # memoria, y el siguiente _save() los deja fuera de state.json para
                # siempre. Minimizacion de datos, mismo criterio que con `meta` arriba.
            except Exception:
                pass
        # Buzones: primero los ficheros propios y despues, si este arranque viene de una
        # version anterior, se vuelcan los que quedaran dentro de state.json. Se hace en
        # este orden para que un fichero ya migrado NUNCA lo pise el state.json viejo.
        self._load_boxes()
        if legacy_boxes:
            migrated = 0
            for peer, box in legacy_boxes.items():
                if not isinstance(box, list) or not box or peer in self.mailboxes:
                    continue
                self.mailboxes[peer] = box
                self._save_box(peer)
                migrated += 1
            # Reescribe state.json ya SIN los buzones (el _save nuevo no los incluye).
            self._save()
            emit_log("Buzones migrados a mail/: %d" % migrated)

    def _save(self):
        """Guarda el estado SIN los buzones (esos van a mail/<peer>.json, uno por peer).
        Sacarlos de aqui es lo que hace que un /send no reescriba decenas de MB."""
        tmp = self.state_file + ".tmp"
        with open(tmp, "w", encoding="utf-8") as f:
            json.dump({"directory": self.directory,
                       "meta": self.meta, "push": self.push, "prekeys": self.prekeys}, f)
        os.replace(tmp, self.state_file)

    def upsert(self, a):
        with self.lock:
            self.directory[a["peerId"]] = {
                "peerId": a["peerId"], "x25519Pub": a["x25519Pub"], "ed25519Pub": a["ed25519Pub"]}
            self._save()

    # ----- Ultima actividad de cada peer -----

    def record_meta(self, peer):
        """Anota la ultima actividad de un peer. NO va al directorio -> nunca se
        expone por /lookup.

        MINIMIZACION DE DATOS (1.16.0-py, 2026-08-01): de los cuatro datos que esta
        funcion llego a guardar, solo sobrevive `lastSeen` — y existe para saber si un
        buzon sigue vivo, no para saber quien eres.
          - `lastIp`: nunca se guardo por usuario (la IP se veia transitoriamente para
            el rate-limit y solo se conservaba dentro de un reporte, que ya no existe).
          - `deviceId` y `name`: se guardaban SOLO para que el operador identificara
            cuentas en admin.php. Retirada la moderacion se quedaron sin finalidad, y
            tratar un dato personal sin finalidad es la peor posicion posible frente al
            principio de minimizacion: todo el coste, cero beneficio.
        El `pop` de abajo no es defensivo: es lo que borra lo heredado de versiones
        anteriores segun cada peer vuelve a conectarse (y `_load` hace lo mismo de golpe
        al arrancar, para los que no vuelvan nunca)."""
        if not peer:
            return
        with self.lock:
            m = self.meta.get(peer) or {}
            m["lastSeen"] = int(time.time() * 1000)
            # Purga de lo heredado de versiones anteriores, no solo "dejar de escribir":
            # sin esto, el nickname y el deviceId de quien no vuelva a conectarse se
            # quedarian en state.json para siempre.
            m.pop("lastIp", None)
            m.pop("deviceId", None)
            m.pop("name", None)
            self.meta[peer] = m
            self._save()

    def purge_peer(self, peer):
        """Elimina por completo a un peer: directorio, metadatos, buzon y push."""
        with self.lock:
            existed = (peer in self.directory or peer in self.meta
                       or peer in self.mailboxes or peer in self.push
                       or peer in self.prekeys)
            self.directory.pop(peer, None)
            self.meta.pop(peer, None)
            had_box = self.mailboxes.pop(peer, None) is not None
            self.push.pop(peer, None)
            self.prekeys.pop(peer, None)
            if had_box:
                self._save_box(peer)   # buzon vacio -> borra su fichero
            if existed:
                self._save()
        return existed

    # ----- Prekeys (X3DH ligero): bundle publico y firmado por peer -----

    def set_prekey(self, peer, bundle):
        with self.lock:
            self.prekeys[peer] = bundle
            self._save()

    def prekey_of(self, peer):
        with self.lock:
            return self.prekeys.get(peer)

    def purge_prekeys(self):
        """Caduca bundles sin refrescar en PREKEY_TTL_DAYS. Un bundle viejo no es
        inseguro (sigue firmado), pero si su dueño lleva meses sin aparecer tampoco
        va a poder descifrar nada: mejor que el emisor caiga al handshake normal."""
        cutoff = int(time.time() * 1000) - PREKEY_TTL_DAYS * 86400 * 1000
        with self.lock:
            stale = [p for p, b in self.prekeys.items() if b.get("ts", 0) < cutoff]
            for p in stale:
                self.prekeys.pop(p, None)
            if stale:
                self._save()
        return len(stale)

    # ----- UnifiedPush: endpoint de aviso por peer -----

    def set_push(self, peer, endpoint):
        with self.lock:
            self.push[peer] = {"endpoint": endpoint, "ts": int(time.time() * 1000)}
            self._save()

    def clear_push(self, peer):
        with self.lock:
            if self.push.pop(peer, None) is not None:
                self._save()

    def push_of(self, peer):
        with self.lock:
            e = self.push.get(peer)
            return e["endpoint"] if e else None

    def purge_push(self):
        """Caduca endpoints sin refrescar en PUSH_TTL_DAYS (app desinstalada, etc.)."""
        cutoff = int(time.time() * 1000) - PUSH_TTL_DAYS * 86400 * 1000
        with self.lock:
            stale = [p for p, e in self.push.items() if e.get("ts", 0) < cutoff]
            for p in stale:
                self.push.pop(p, None)
            if stale:
                self._save()
        return len(stale)

    # ----- Resoluciones (conclusión del reporte -> reportante) -----

    def keys_of(self, peer):
        with self.lock:
            return self.directory.get(peer)

    class MailboxFull(Exception):
        """Ya no la lanza `queue` (el buzon es una cola circular desde 1.40.0-py). Se
        conserva porque el manejador de /send todavia la captura: si alguna vez se vuelve
        a rechazar por lleno, el 429 sigue estando bien construido."""
        pass

    def queue(self, env):
        """Encola un sobre. Si el buzon esta a tope, se DESCARTA EL MAS ANTIGUO en vez de
        rechazar el nuevo (cola circular).

        Por que (medido en produccion el 2026-08-21): rechazar dejaba el envio en fallo
        PERMANENTE, y la app no distingue ese 429 de un fallo transitorio -> reintentaba
        cada 20 s durante los 14 dias del TTL. En nodemax habia 14 buzones clavados en 500
        (7.000 sobres, el 63% de todo el correo) y sus duenos llevaban en su mayoria mas de
        una semana sin aparecer: el resultado era que el **98,5% de los /send se rechazaban**
        (5.953 de 6.041 en 10 minutos) en un bucle que no podia terminar nunca.

        Contrapartida, dicha sin adornos: se pierde el correo mas viejo sin entregar de un
        buzon saturado. Es correo que de todas formas iba a caducar por TTL, y a cambio los
        mensajes NUEVOS —los que a alguien le importan ahora— si entran. Un buzon lleno deja
        de ser una via muerta."""
        with self.lock:
            box = self.mailboxes.setdefault(env["to"], [])
            # Marca de recepcion del servidor (para el TTL). El cliente la ignora;
            # no afecta a la firma, que solo cubre from/to/ephPub/nonce/ct.
            env["_rx"] = int(time.time() * 1000)
            box.append(env)
            if len(box) > MAILBOX_MAX:
                # Se tira por delante: el mas antiguo es el que lleva mas tiempo sin que
                # nadie lo recoja. `del box[:n]` por si el tope se bajara en caliente.
                del box[:len(box) - MAILBOX_MAX]
            self._save_box(env["to"])

    def purge_expired(self):
        """Elimina sobres sin recoger mas viejos que el TTL y buzones vacios.

        El TTL de un sobre PROGRAMADO cuenta desde su hora de entrega, no desde que
        llego: si contara desde la llegada, un mensaje programado a mas de 14 dias se
        borraria antes de poder entregarse — justo lo contrario de lo que se pidio."""
        if ENVELOPE_TTL_MS <= 0:
            return 0
        cutoff = int(time.time() * 1000) - ENVELOPE_TTL_MS
        removed = 0

        def due_since(e):
            return max(int(e.get("_rx") or 0), int(e.get("notBefore") or 0))

        with self.lock:
            for peer in list(self.mailboxes.keys()):
                box = self.mailboxes[peer]
                kept = [e for e in box if (due_since(e) or cutoff + 1) >= cutoff]
                if len(kept) == len(box):
                    continue
                removed += len(box) - len(kept)
                if kept:
                    self.mailboxes[peer] = kept
                else:
                    self.mailboxes.pop(peer, None)
                # Solo se reescribe el buzon que ha cambiado, no todos.
                self._save_box(peer)
        return removed

    def purge_files(self):
        """Borra blobs mas viejos que el TTL y, si se excede la cuota total,
        los mas antiguos hasta volver por debajo del limite."""
        try:
            entries = []
            total = 0
            now = time.time()
            for name in os.listdir(self.files_dir):
                p = os.path.join(self.files_dir, name)
                try:
                    st = os.stat(p)
                except OSError:
                    continue
                entries.append((st.st_mtime, st.st_size, p))
                total += st.st_size
            removed = 0
            # 1) por antiguedad
            if FILES_TTL_S > 0:
                for mt, sz, p in entries:
                    if now - mt > FILES_TTL_S:
                        try:
                            os.remove(p); removed += 1; total -= sz
                        except OSError:
                            pass
            # 2) por cuota total (borra los mas antiguos primero)
            if FILES_MAX_BYTES > 0 and total > FILES_MAX_BYTES:
                survivors = [(mt, sz, p) for (mt, sz, p) in entries if os.path.exists(p)]
                survivors.sort()
                for mt, sz, p in survivors:
                    if total <= FILES_MAX_BYTES:
                        break
                    try:
                        os.remove(p); removed += 1; total -= sz
                    except OSError:
                        pass
            return removed
        except OSError:
            return 0

    def inbox(self, peer):
        """Sobres entregables AHORA. Los programados (`notBefore` en el futuro) se
        quedan dentro: este filtro es el unico sitio por el que salen los sobres
        —lo usan tanto POST /inbox como el volcado al abrir el WebSocket—, asi que
        con retenerlos aqui no se escapan por ninguna via."""
        now = int(time.time() * 1000)
        with self.lock:
            return [e for e in self.mailboxes.get(peer, []) if int(e.get("notBefore") or 0) <= now]

    def due_now(self):
        """Sobres programados que ACABAN de vencer y a los que aun no se ha avisado.
        Se marcan con `_pushed` (campo interno, la app ignora los extra) para no
        volver a empujarlos en cada vuelta del bucle."""
        now = int(time.time() * 1000)
        out = []
        with self.lock:
            touched = set()
            for peer, box in self.mailboxes.items():
                for e in box:
                    nb = int(e.get("notBefore") or 0)
                    if nb and nb <= now and not e.get("_pushed"):
                        e["_pushed"] = True
                        out.append((peer, e))
                        touched.add(peer)
            for peer in touched:
                self._save_box(peer)
        return out

    def ack(self, peer, ids):
        with self.lock:
            box = self.mailboxes.get(peer)
            if box:
                kept = [e for e in box if e["id"] not in ids]
                if kept:
                    self.mailboxes[peer] = kept
                else:
                    self.mailboxes.pop(peer, None)
                self._save_box(peer)

    def save_file(self, data):
        fid = str(uuid.uuid4())
        with open(os.path.join(self.files_dir, fid), "wb") as f:
            f.write(data)
        return fid

    def read_file(self, fid):
        if not re.fullmatch(r"[0-9a-fA-F-]{36}", fid or ""):
            return None
        path = os.path.join(self.files_dir, fid)
        if os.path.isfile(path):
            with open(path, "rb") as f:
                return f.read()
        return None

# =====================================================================
#  Promo: ledger de puntos + cola de pagos pendientes (SOLO contabilidad)
# =====================================================================

class PromoStore:
    """Puntos Promo, canjeables por bitcoin. Esta clase SOLO hace contabilidad:
    lleva el ledger de puntos por peer y escribe la cola de pagos pendientes a
    disco para que `usb/promo_pay.py` (proceso APARTE, el unico con la clave
    privada) los recoja y pague. Nunca importa nada de bitcoin, nunca ve una
    clave. Mismo patron de persistencia que `Store`: JSON, escritura atomica
    (tmp + os.replace), todo bajo un RLock."""

    def __init__(self, data_dir):
        self.data_dir = data_dir
        self.queue_dir = os.path.join(data_dir, "promo_queue")
        os.makedirs(self.queue_dir, exist_ok=True)
        self.state_file = os.path.join(data_dir, "promo.json")
        self.lock = threading.RLock()
        self.ledger = {}           # peerId -> {points, pointsToday, dayEpoch}
        self.claims_hour = {}      # (peerId, action) -> deque[epoch_s], EN MEMORIA, no persistido
        self.global_paid_today_sats = 0
        self.global_day_epoch = 0
        self.withdrawals = {}      # id -> {peerId, addr, sats, points, ts, status, txid}
        self._load()

    def _load(self):
        if os.path.isfile(self.state_file):
            try:
                with open(self.state_file, "r", encoding="utf-8") as f:
                    data = json.load(f)
                self.ledger = data.get("ledger", {})
                self.global_paid_today_sats = data.get("globalPaidTodaySats", 0)
                self.global_day_epoch = data.get("globalDayEpoch", 0)
                self.withdrawals = data.get("withdrawals", {})
            except Exception:
                pass

    def _save(self):
        tmp = self.state_file + ".tmp"
        with open(tmp, "w", encoding="utf-8") as f:
            json.dump({"ledger": self.ledger, "globalPaidTodaySats": self.global_paid_today_sats,
                       "globalDayEpoch": self.global_day_epoch, "withdrawals": self.withdrawals}, f)
        os.replace(tmp, self.state_file)

    @staticmethod
    def _today():
        return int(time.time()) // 86400

    def _entry(self, peer):
        """Fila del ledger de `peer`, reseteando `pointsToday` (tope de
        GANANCIA) y `withdrawnToday` (tope de RETIRO, independiente) si ha
        cambiado el dia -- mismo limite de dia para los dos, sin necesidad de
        dos epochs distintos. NO llama a _save() -- lo hace el llamante, que
        es quien sabe si de verdad hubo un cambio que merezca escribir a
        disco. Entradas viejas sin `withdrawnToday` (de antes de que existiera
        este tope) lo tratan como False via `.get()` en el llamante -- sin
        migracion, sin arriesgar el saldo de nadie."""
        e = self.ledger.get(peer)
        today = self._today()
        if e is None:
            e = {"points": 0, "pointsToday": 0, "dayEpoch": today, "withdrawnToday": False}
            self.ledger[peer] = e
        elif e.get("dayEpoch") != today:
            e["pointsToday"] = 0
            e["dayEpoch"] = today
            e["withdrawnToday"] = False
        return e

    def _rate_ok(self, peer, action):
        """Tope de ritmo por tipo de accion: ni un cliente modificado puede
        ráfagar claims mas rapido que un uso humano razonable. Salvaguarda
        adicional -- la barrera real contra Sybil es el tope GLOBAL diario en
        `queue_withdrawal`, no esto."""
        key = (peer, action)
        now = time.time()
        dq = self.claims_hour.setdefault(key, collections.deque())
        while dq and now - dq[0] > 3600:
            dq.popleft()
        if len(dq) >= PROMO_CLAIM_MAX_PER_HOUR.get(action, 0):
            return False
        dq.append(now)
        return True

    def claim(self, peer, action):
        """Aplica puntos por una accion reclamada (firmada, ver el llamante).
        Devuelve (aceptado, puntos_totales_ahora, motivo).

        El `motivo` existe para que el cliente sepa si tiene sentido REINTENTAR
        (desde 1.0.165 los claims viven en una cola persistente en el movil y no
        se dan por cerrados hasta que este metodo los acepta):
          "ok"             -> concedido
          "rate"           -> tope de ritmo/hora, reintentar mas tarde
          "daily_cap"      -> tope diario de ganancia lleno, reintentar mañana
          "already"        -> nada que deber (compensacion ya dada): NO reintentar
          "unknown_action" -> accion que este nodo no conoce: NO reintentar
        Un cliente viejo (pre-1.0.165) ignora el campo y se comporta igual que antes.

        TODO O NADA a proposito: antes esto concedia `min(valor, hueco_del_tope)`,
        asi que una accion de 10 puntos con 3 de hueco daba 3 y los otros 7 se
        perdian PARA SIEMPRE (el cliente la daba por entregada al ver
        accepted=true). Ahora, si no cabe entera, se rechaza entera y el cliente
        la reintenta cuando el tope se libere -- ningun punto se evapora."""
        if action not in PROMO_POINTS:
            return False, 0, "unknown_action"
        with self.lock:
            e = self._entry(peer)
            # Compensacion: UNA sola vez por peer, y fuera de los dos topes. No es
            # actividad del usuario sino una reparacion nuestra, asi que no debe
            # comerse el hueco diario de puntos que si ha ganado de verdad hoy.
            if action == "compensation":
                if e.get("compensated"):
                    return False, e["points"], "already"
                e["compensated"] = True
                e["points"] += PROMO_POINTS["compensation"]
                self._save()
                return True, e["points"], "ok"
            want = PROMO_POINTS[action]
            # El tope diario se mira ANTES que el de ritmo: comprobarlo al reves
            # gastaria un hueco del ritmo en un claim que va a rechazarse igual.
            if want > max(0, PROMO_DAILY_CAP_POINTS - e["pointsToday"]):
                return False, e["points"], "daily_cap"
            if not self._rate_ok(peer, action):
                return False, e["points"], "rate"
            e["points"] += want
            e["pointsToday"] += want
            self._save()
            return True, e["points"], "ok"

    def points_of(self, peer):
        with self.lock:
            e = self._entry(peer)
            self._save()
            return e["points"], e["pointsToday"], bool(e.get("withdrawnToday"))

    def queue_withdrawal(self, peer, address, price_eur):
        """Comprueba elegibilidad (minimo de puntos), el tope de RETIRO diario
        POR PERSONA y el tope GLOBAL diario, todo BAJO EL MISMO LOCK que la
        resta de puntos (sin eso, dos retiros a la vez podrian colarse los dos
        por cualquiera de los topes). El pago de este retiro es PARCIAL si el
        saldo supera el tope diario de retiro: se paga como mucho
        PROMO_MAX_WITHDRAW_PER_DAY_POINTS y el resto se queda en el ledger
        para el dia siguiente -- nunca se pierde, solo se reparte en varios
        dias (independiente del tope de GANANCIA diaria, que sigue intacto).
        Calcula el importe en sats con el precio EUR/BTC vigente (`PRICE`, ya
        cacheado por el nodo) y encola el pago escribiendo
        `data/promo_queue/<id>.json` para que `promo_pay.py` lo recoja.
        Devuelve (id, motivo_de_rechazo, sats): id=None si se rechaza --
        'not_eligible' (menos del minimo), 'daily_withdraw_capped' (ya se
        retiro hoy, vuelve mañana; el saldo NO se toca) o 'capped' (tope
        global diario agotado; tampoco se tocan los puntos del usuario, no
        pierde nada por que otros hayan retirado antes que el hoy)."""
        with self.lock:
            e = self._entry(peer)
            points = e["points"]
            if points < PROMO_MIN_WITHDRAW_POINTS:
                return None, "not_eligible", 0
            if e.get("withdrawnToday"):
                return None, "daily_withdraw_capped", 0
            payout_points = min(points, PROMO_MAX_WITHDRAW_PER_DAY_POINTS)
            euro_cents = payout_points * 4 // 5
            sats = int(round(euro_cents / 100.0 / price_eur * 100_000_000))
            today = self._today()
            if self.global_day_epoch != today:
                self.global_day_epoch = today
                self.global_paid_today_sats = 0
            # Tope GLOBAL solo si esta configurado (>0). Apagado por defecto desde
            # 2026-08-08: mandan unicamente los limites por persona. Se sigue
            # LLEVANDO LA CUENTA de lo pagado hoy aunque no se aplique -- cuesta
            # nada, sirve para saber cuanto sale al dia y hace que volver a poner
            # un tope sea cambiar una variable de entorno, sin tocar codigo.
            if PROMO_DAILY_CAP_SATS > 0 and self.global_paid_today_sats + sats > PROMO_DAILY_CAP_SATS:
                return None, "capped", 0
            wid = uuid.uuid4().hex
            entry = {"peerId": peer, "addr": address, "sats": sats,
                     "points": payout_points, "ts": int(time.time() * 1000),
                     "status": "queued", "txid": None}
            # El fichero de cola PRIMERO, y solo si se escribe de verdad se descuentan
            # los puntos. Al reves (que es como estaba) un disco lleno o un fallo de
            # E/S dejaba al usuario sin puntos y sin pago encolado: nadie los habria
            # devuelto nunca, porque para el ledger el retiro constaba como hecho.
            try:
                self._write_queue_file(wid, entry)
            except Exception as ex:
                emit_log("Promo: no se pudo encolar el pago %s (%s); no se tocan los puntos" % (wid, ex))
                return None, "error", 0
            self.global_paid_today_sats += sats
            e["points"] -= payout_points
            e["withdrawnToday"] = True
            self.withdrawals[wid] = entry
            self._save()
            return wid, None, sats

    def _write_queue_file(self, wid, entry):
        path = os.path.join(self.queue_dir, wid + ".json")
        tmp = path + ".tmp"
        with open(tmp, "w", encoding="utf-8") as f:
            json.dump({"id": wid, "addr": entry["addr"], "sats": entry["sats"]}, f)
        os.replace(tmp, path)

    def refresh_done(self):
        """Lee los `.done` que `promo_pay.py` haya dejado en la cola (resultado
        de un intento de pago) y actualiza el estado de cada retiro. Se llama
        desde el bucle de mantenimiento, igual que el resto de purgas
        periodicas de `Store`."""
        with self.lock:
            changed = False
            try:
                names = os.listdir(self.queue_dir)
            except Exception:
                return
            for fn in names:
                if not fn.endswith(".done"):
                    continue
                wid = fn[:-5]
                w = self.withdrawals.get(wid)
                if w is None or w["status"] != "queued":
                    continue
                try:
                    with open(os.path.join(self.queue_dir, fn), "r", encoding="utf-8") as f:
                        result = json.load(f)
                except Exception:
                    continue
                if result.get("txid"):
                    w["status"] = "paid"
                    w["txid"] = result["txid"]
                else:
                    w["status"] = "failed"
                    # Fallo PERMANENTE (no "sin saldo todavia, se reintenta"): se
                    # devuelven los puntos y el hueco del tope global gastado --
                    # no hay que penalizar al usuario por un fallo de la propia
                    # infraestructura de pago.
                    if result.get("permanent"):
                        e = self._entry(w["peerId"])
                        e["points"] += w["points"]
                        self.global_paid_today_sats = max(0, self.global_paid_today_sats - w["sats"])
                    else:
                        w["status"] = "queued"  # sigue en cola, se reintentara
                changed = True
            if changed:
                self._save()

    def withdrawals_of(self, peer):
        with self.lock:
            return [dict(w, id=wid) for wid, w in self.withdrawals.items() if w["peerId"] == peer]


# =====================================================================
#  Registro de conexiones WebSocket (entrega en tiempo real)
# =====================================================================

class WsConn:
    def __init__(self, wfile):
        self.wfile = wfile
        self.lock = threading.Lock()
        self.alive = True

    def send_text(self, text):
        payload = text.encode("utf-8")
        header = bytearray([0x81])
        n = len(payload)
        if n < 126:
            header.append(n)
        elif n < 65536:
            header.append(126)
            header += struct.pack(">H", n)
        else:
            header.append(127)
            header += struct.pack(">Q", n)
        with self.lock:
            if not self.alive:
                return
            try:
                self.wfile.write(bytes(header) + payload)
                self.wfile.flush()
            except Exception:
                self.alive = False

    def send_pong(self, data=b""):
        with self.lock:
            if not self.alive:
                return
            try:
                self.wfile.write(bytes([0x8A, len(data)]) + data)
                self.wfile.flush()
            except Exception:
                self.alive = False


class WsRegistry:
    def __init__(self):
        self.lock = threading.Lock()
        self.by_peer = {}  # peerId -> set(WsConn)

    def add(self, peer, conn):
        with self.lock:
            self.by_peer.setdefault(peer, set()).add(conn)

    def remove(self, peer, conn):
        with self.lock:
            s = self.by_peer.get(peer)
            if s:
                s.discard(conn)
                if not s:
                    self.by_peer.pop(peer, None)

    def online(self, peer):
        with self.lock:
            return bool(self.by_peer.get(peer))

    def live(self):
        """Cuantas apps hay conectadas AHORA (conexiones WS vivas) y cuantos peers
        distintos. Numeros AGREGADOS de salud del nodo: cuantas, nunca cuales -- el
        panel no debe poder senalar a una persona."""
        with self.lock:
            return sum(len(v) for v in self.by_peer.values()), len(self.by_peer)

    def count(self, peer):
        with self.lock:
            return len(self.by_peer.get(peer, ()))

    def push(self, peer, frame_text):
        with self.lock:
            conns = list(self.by_peer.get(peer, ()))
        for c in conns:
            c.send_text(frame_text)


STORE = None
PROMO = None
WSREG = WsRegistry()


class RateLimiter:
    """Ventana deslizante de 60 s por IP. Barato y sin dependencias."""
    def __init__(self, rpm):
        self.rpm = rpm
        self.lock = threading.Lock()
        self.hits = {}  # ip -> deque[timestamps]

    def allow(self, ip):
        if self.rpm <= 0:
            return True
        now = time.time()
        with self.lock:
            dq = self.hits.get(ip)
            if dq is None:
                dq = collections.deque()
                self.hits[ip] = dq
            while dq and now - dq[0] > 60:
                dq.popleft()
            if len(dq) >= self.rpm:
                return False
            dq.append(now)
            return True

    def sweep(self):
        """Elimina IPs sin actividad reciente para que el dict no crezca."""
        now = time.time()
        with self.lock:
            for ip in list(self.hits.keys()):
                dq = self.hits[ip]
                while dq and now - dq[0] > 60:
                    dq.popleft()
                if not dq:
                    self.hits.pop(ip, None)


RATE = RateLimiter(RATE_RPM)


# --- UnifiedPush: aviso de "hay correo" ---
_PUSH_LAST = {}   # peerId -> time.monotonic() del ultimo aviso (throttle)
_PUSH_LOCK = threading.Lock()


def _push_wake(peer):
    """POST minusculo y CONSTANTE ("new") al endpoint UnifiedPush del peer, si lo
    registro y NO tiene un WebSocket vivo (entrega en tiempo real -> sobra avisar).
    Fire-and-forget en hilo aparte: jamas bloquea ni tumba el /send. El endpoint se
    valido al registrarse (https + SSRF-guard); el cuerpo nunca lleva contenido."""
    if not PUSH_ENABLED or STORE is None:
        return
    try:
        if WSREG.online(peer):
            return
        ep = STORE.push_of(peer)
        if not ep:
            return
        now = time.monotonic()
        with _PUSH_LOCK:
            if now - _PUSH_LAST.get(peer, 0.0) < PUSH_MIN_INTERVAL_S:
                return
            _PUSH_LAST[peer] = now
            # Poda del throttle para que el dict no crezca sin limite.
            if len(_PUSH_LAST) > 4096:
                cutoff = now - 3600
                for k in [k for k, v in _PUSH_LAST.items() if v < cutoff]:
                    _PUSH_LAST.pop(k, None)
    except Exception:
        return

    def _post():
        try:
            req = urllib.request.Request(
                ep, data=b"new", headers={"Content-Type": "text/plain"}, method="POST")
            with urllib.request.urlopen(req, timeout=PUSH_TIMEOUT) as r:
                r.read(64)
        except Exception:
            pass
    threading.Thread(target=_post, name="versell-push", daemon=True).start()


# Limita conexiones concurrentes totales (anti-agotamiento de hilos/FD).
CONN_SEM = threading.BoundedSemaphore(MAX_CONN)

# Sumidero de log opcional (lo usa la GUI para mostrar la actividad en vivo).
LOG_SINK = None


def set_log_sink(fn):
    global LOG_SINK
    LOG_SINK = fn


def emit_log(msg):
    if LOG_SINK:
        try:
            LOG_SINK(msg)
            return
        except Exception:
            pass
    try:
        sys.stdout.write(msg + "\n")
        sys.stdout.flush()
    except Exception:
        pass


def stats():
    """Estadisticas para la GUI: pares anunciados, pendientes, conexiones WS, backend."""
    peers = 0
    pending = 0
    if STORE is not None:
        with STORE.lock:
            peers = len(STORE.directory)
            pending = sum(len(v) for v in STORE.mailboxes.values())
    with WSREG.lock:
        ws = sum(len(s) for s in WSREG.by_peer.values())
    return {"peers": peers, "pending": pending, "ws": ws, "backend": VERIFY_BACKEND}


class GuardedServer(ThreadingHTTPServer):
    daemon_threads = True

    def handle_error(self, request, client_address):
        # Silencia los errores de red normales (timeouts, conexiones cortadas)
        # para no llenar el log; solo interesa cuando se depura.
        pass


_MAINT_STARTED = False

# Cada cuanto se mira si ha vencido algun mensaje programado. Sin este aviso, el
# sobre se quedaria en el buzon hasta que el destinatario sondease por su cuenta
# —hasta 20 s con la app despierta, y bastante mas en Doze—, cuando lo natural es
# que un mensaje programado a las 09:00 llegue a las 09:00.
SCHED_TICK_S = 30


def _sched_loop():
    while True:
        time.sleep(SCHED_TICK_S)
        try:
            if STORE is None:
                continue
            for peer, env in STORE.due_now():
                WSREG.push(peer, json.dumps({"type": "envelope", "envelope": env}))
                _push_wake(peer)
        except Exception:
            pass


def _maintenance_loop():
    while True:
        time.sleep(MAINTENANCE_EVERY_S)
        try:
            if STORE is not None:
                n = STORE.purge_expired()
                f = STORE.purge_files()
                p = STORE.purge_push()
                k = STORE.purge_prekeys()
                _turn_sweep()   # cierra relays de llamadas terminadas
                if n or f or p or k:
                    emit_log("[mantenimiento] sobres caducados: %d, archivos purgados: %d, push caducados: %d, prekeys caducadas: %d"
                             % (n, f, p, k))
            if PROMO is not None:
                PROMO.refresh_done()   # recoge resultados que promo_pay.py haya dejado en la cola
            RATE.sweep()
        except Exception:
            pass


# =====================================================================
#  Servidor STUN embebido (RFC 5389, solo Binding Request)
#  Le dice a cada movil con que IP:puerto publicos le ve el nodo. Con eso
#  WebRTC genera candidatos "srflx" y dos moviles en redes distintas pueden
#  perforar el NAT y hablar DIRECTAMENTE (sin pasar el audio por nadie).
# =====================================================================

STUN_MAGIC = 0x2112A442
_STUN_SOCK = None


def _stun_response(data, addr):
    """Devuelve la respuesta a un Binding Request, o None si no lo es."""
    if len(data) < 20:
        return None
    mtype, mlen, magic = struct.unpack(">HHI", data[:8])
    if mtype != 0x0001 or magic != STUN_MAGIC:      # solo Binding Request valido
        return None
    tid = data[8:20]
    ip, port = addr[0], addr[1]
    try:
        raw_ip = _socket.inet_aton(ip)
    except Exception:
        return None
    # XOR-MAPPED-ADDRESS: puerto y direccion enmascarados con el magic cookie.
    xport = port ^ (STUN_MAGIC >> 16)
    xip = bytes(b ^ m for b, m in zip(raw_ip, struct.pack(">I", STUN_MAGIC)))
    value = struct.pack(">BBH", 0, 0x01, xport) + xip
    attrs = struct.pack(">HH", 0x0020, len(value)) + value
    return struct.pack(">HHI", 0x0101, len(attrs), STUN_MAGIC) + tid + attrs


# =====================================================================
#  TURN embebido (RFC 5766, el subconjunto que usa WebRTC)
#  Cuando los DOS moviles estan tras CGNAT (tipico en datos moviles) no hay
#  perforacion de NAT posible: hace falta que alguien reenvie el audio. Eso es
#  un relay TURN. Aqui va uno minimo dentro del propio nodo -- mismo puerto que
#  el STUN, sin instalar coturn -- para que las llamadas funcionen en CUALQUIER
#  red. El nodo solo mueve bytes: el audio va cifrado DTLS-SRTP extremo a
#  extremo y el relay NO puede oirlo (ve IPs y duracion, como con los sobres).
# =====================================================================

TURN_ENABLED = os.environ.get("VERSELL_TURN", "1") not in ("0", "false", "no", "")
TURN_MIN_PORT = _envint("VERSELL_TURN_MIN_PORT", 49160)
TURN_MAX_PORT = _envint("VERSELL_TURN_MAX_PORT", 49200)
TURN_REALM = os.environ.get("VERSELL_TURN_REALM", "dicra")
TURN_ALLOC_LIFETIME = _envint("VERSELL_TURN_LIFETIME", 600)
TURN_MAX_ALLOCS = _envint("VERSELL_TURN_MAX_ALLOCS", 200)

# Metodos STUN/TURN
_M_BINDING, _M_ALLOCATE, _M_REFRESH = 0x0001, 0x0003, 0x0004
_M_CREATEPERM, _M_CHANBIND = 0x0008, 0x0009
# Send/Data son INDICATIONS: el valor que viaja lleva la clase en los bits altos
# (0x0016 / 0x0017), no el numero de metodo pelado (0x0006 / 0x0007).
_M_SEND_IND, _M_DATA_IND = 0x0016, 0x0017
# Atributos
_A_MAPPED, _A_USERNAME, _A_MSGINT, _A_ERROR = 0x0001, 0x0006, 0x0008, 0x0009
_A_REALM, _A_NONCE, _A_XPEER, _A_DATA = 0x0014, 0x0015, 0x0012, 0x0013
_A_XRELAY, _A_XMAPPED, _A_LIFETIME = 0x0016, 0x0020, 0x000D
_A_REQTRANSPORT, _A_CHANNEL = 0x0019, 0x000C


def _attr(t, v):
    return struct.pack(">HH", t, len(v)) + v + b"\x00" * ((-len(v)) % 4)


def _parse_attrs(body):
    out, i = {}, 0
    while i + 4 <= len(body):
        t, ln = struct.unpack(">HH", body[i:i + 4])
        out[t] = (body[i + 4:i + 4 + ln], i)
        i += 4 + ln + ((-ln) % 4)
    return out


def _xor_addr(ip, port):
    xport = port ^ (STUN_MAGIC >> 16)
    xip = bytes(b ^ m for b, m in zip(_socket.inet_aton(ip), struct.pack(">I", STUN_MAGIC)))
    return struct.pack(">BBH", 0, 0x01, xport) + xip


def _unxor_addr(v):
    port = struct.unpack(">H", v[2:4])[0] ^ (STUN_MAGIC >> 16)
    ip = _socket.inet_ntoa(bytes(b ^ m for b, m in zip(v[4:8], struct.pack(">I", STUN_MAGIC))))
    return ip, port


def _msg(mtype, tid, attrs=b"", key=None):
    """Construye un mensaje; con `key` añade MESSAGE-INTEGRITY (HMAC-SHA1)."""
    if key:
        hdr = struct.pack(">HHI", mtype, len(attrs) + 24, STUN_MAGIC) + tid
        mi = hmac.new(key, hdr + attrs, hashlib.sha1).digest()
        return hdr + attrs + _attr(_A_MSGINT, mi)
    return struct.pack(">HHI", mtype, len(attrs), STUN_MAGIC) + tid + attrs


def _turn_key(username):
    """Clave de long-term credentials para el username REST (expiry:peerId).
    La contraseña es la misma que emite /turncreds: HMAC-SHA1(secreto, username)."""
    if not TURN_SECRET:
        return None
    pwd = base64.b64encode(hmac.new(TURN_SECRET.encode(), username.encode(), hashlib.sha1).digest()).decode()
    return hashlib.md5(("%s:%s:%s" % (username, TURN_REALM, pwd)).encode()).digest()


def _check_integrity(data, attrs, key):
    """Verifica el MESSAGE-INTEGRITY del cliente re-calculando el HMAC sobre el
    mensaje truncado justo antes del atributo (con la longitud ajustada)."""
    got = attrs.get(_A_MSGINT)
    if not got or not key:
        return False
    value, off = got
    start = 20 + off
    hdr = data[:2] + struct.pack(">H", (start + 24) - 20) + data[4:20]
    return hmac.compare_digest(hmac.new(key, hdr + data[20:start], hashlib.sha1).digest(), value)


class _Allocation:
    __slots__ = ("client", "sock", "username", "expires", "perms", "channels", "server_sock")

    def __init__(self, client, sock, username, server_sock):
        self.client = client
        self.sock = sock                 # socket del relay (puerto publico asignado)
        self.username = username
        self.expires = time.time() + TURN_ALLOC_LIFETIME
        self.perms = {}                  # ip del peer -> caducidad
        self.channels = {}               # nº de canal -> (ip, port)
        self.server_sock = server_sock   # socket 3478 por donde hablamos con el movil


_ALLOCS = {}          # (ip, port) del movil -> _Allocation
_ALLOCS_LOCK = threading.Lock()
_NONCE_SECRET = os.urandom(16)


def _nonce():
    return hashlib.sha1(_NONCE_SECRET + str(int(time.time()) // 3600).encode()).hexdigest().encode()


def _relay_reader(alloc):
    """Todo lo que llegue al puerto del relay se reenvia al movil: por canal
    (ChannelData, 4 bytes de cabecera) si lo hay, o como Data indication."""
    while True:
        try:
            data, peer = alloc.sock.recvfrom(2048)
        except Exception:
            return
        try:
            if time.time() > alloc.expires:
                return
            if peer[0] not in alloc.perms:
                continue                 # sin permiso: se descarta (anti-abuso)
            chan = next((c for c, p in alloc.channels.items() if p == peer), None)
            if chan is not None:
                out = struct.pack(">HH", chan, len(data)) + data
                out += b"\x00" * ((-len(data)) % 4)
            else:
                out = _msg(_M_DATA_IND, os.urandom(12),
                           _attr(_A_XPEER, _xor_addr(peer[0], peer[1])) + _attr(_A_DATA, data))
            alloc.server_sock.sendto(out, alloc.client)
        except Exception:
            continue


def _turn_handle(data, addr, sock):
    """Procesa un mensaje TURN del movil. Devuelve True si lo ha atendido."""
    if not TURN_ENABLED or not TURN_SECRET:
        return False
    # ChannelData: no es STUN, va por su propio formato (los 2 bits altos a 01).
    if data and (data[0] & 0xC0) == 0x40:
        chan, ln = struct.unpack(">HH", data[:4])
        with _ALLOCS_LOCK:
            alloc = _ALLOCS.get(addr)
        if alloc:
            peer = alloc.channels.get(chan)
            if peer and peer[0] in alloc.perms:
                try:
                    alloc.sock.sendto(data[4:4 + ln], peer)
                except Exception:
                    pass
        return True
    if len(data) < 20:
        return False
    mtype, mlen, magic = struct.unpack(">HHI", data[:8])
    if magic != STUN_MAGIC or mtype not in (_M_ALLOCATE, _M_REFRESH, _M_SEND_IND, _M_CREATEPERM, _M_CHANBIND):
        return False
    tid = data[8:20]
    attrs = _parse_attrs(data[20:20 + mlen])

    # Indications (Send) no llevan integridad: se resuelven con la allocation viva.
    if mtype == _M_SEND_IND:
        with _ALLOCS_LOCK:
            alloc = _ALLOCS.get(addr)
        if alloc and _A_XPEER in attrs and _A_DATA in attrs:
            ip, port = _unxor_addr(attrs[_A_XPEER][0])
            if ip in alloc.perms:
                try:
                    alloc.sock.sendto(attrs[_A_DATA][0], (ip, port))
                except Exception:
                    pass
        return True

    # El resto exigen credenciales (long-term).
    user = attrs.get(_A_USERNAME)
    key = _turn_key(user[0].decode(errors="replace")) if user else None
    if not user or not _check_integrity(data, attrs, key):
        err = _attr(_A_ERROR, struct.pack(">HBB", 0, 4, 1) + b"Unauthorized")
        sock.sendto(_msg(mtype | 0x0110, tid, err + _attr(_A_REALM, TURN_REALM.encode()) + _attr(_A_NONCE, _nonce())), addr)
        return True

    if mtype == _M_ALLOCATE:
        with _ALLOCS_LOCK:
            existing = _ALLOCS.get(addr)
            if existing:
                existing.expires = time.time() + TURN_ALLOC_LIFETIME
                body = (_attr(_A_XRELAY, _xor_addr(_public_ip(), existing.sock.getsockname()[1])) +
                        _attr(_A_XMAPPED, _xor_addr(addr[0], addr[1])) +
                        _attr(_A_LIFETIME, struct.pack(">I", TURN_ALLOC_LIFETIME)))
                sock.sendto(_msg(0x0103, tid, body, key), addr)
                return True
            if len(_ALLOCS) >= TURN_MAX_ALLOCS:
                sock.sendto(_msg(0x0113, tid, _attr(_A_ERROR, struct.pack(">HBB", 0, 4, 86) + b"Allocation Quota Reached"), key), addr)
                return True
        rsock = None
        for port in range(TURN_MIN_PORT, TURN_MAX_PORT + 1):
            try:
                s = _socket.socket(_socket.AF_INET, _socket.SOCK_DGRAM)
                s.bind(("0.0.0.0", port))
                rsock = s
                break
            except Exception:
                continue
        if rsock is None:
            sock.sendto(_msg(0x0113, tid, _attr(_A_ERROR, struct.pack(">HBB", 0, 5, 8) + b"No relay ports"), key), addr)
            return True
        alloc = _Allocation(addr, rsock, user[0].decode(errors="replace"), sock)
        with _ALLOCS_LOCK:
            _ALLOCS[addr] = alloc
        threading.Thread(target=_relay_reader, args=(alloc,), name="versell-turn-relay", daemon=True).start()
        body = (_attr(_A_XRELAY, _xor_addr(_public_ip(), rsock.getsockname()[1])) +
                _attr(_A_XMAPPED, _xor_addr(addr[0], addr[1])) +
                _attr(_A_LIFETIME, struct.pack(">I", TURN_ALLOC_LIFETIME)))
        sock.sendto(_msg(0x0103, tid, body, key), addr)
        # Sin IP:puerto del cliente, por lo mismo que el access log (ver log_message).
        emit_log("[turn] allocation concedida -> puerto relay %d" % rsock.getsockname()[1])
        return True

    with _ALLOCS_LOCK:
        alloc = _ALLOCS.get(addr)
    if not alloc:
        sock.sendto(_msg(mtype | 0x0110, tid, _attr(_A_ERROR, struct.pack(">HBB", 0, 4, 37) + b"No allocation"), key), addr)
        return True

    if mtype == _M_REFRESH:
        life = struct.unpack(">I", attrs[_A_LIFETIME][0])[0] if _A_LIFETIME in attrs else TURN_ALLOC_LIFETIME
        if life == 0:
            with _ALLOCS_LOCK:
                _ALLOCS.pop(addr, None)
            try:
                alloc.sock.close()
            except Exception:
                pass
            sock.sendto(_msg(0x0104, tid, _attr(_A_LIFETIME, struct.pack(">I", 0)), key), addr)
        else:
            alloc.expires = time.time() + TURN_ALLOC_LIFETIME
            sock.sendto(_msg(0x0104, tid, _attr(_A_LIFETIME, struct.pack(">I", TURN_ALLOC_LIFETIME)), key), addr)
        return True

    if mtype == _M_CREATEPERM:
        if _A_XPEER in attrs:
            ip, _ = _unxor_addr(attrs[_A_XPEER][0])
            alloc.perms[ip] = time.time() + 300
        sock.sendto(_msg(0x0108, tid, b"", key), addr)
        return True

    if mtype == _M_CHANBIND:
        if _A_CHANNEL in attrs and _A_XPEER in attrs:
            chan = struct.unpack(">H", attrs[_A_CHANNEL][0][:2])[0]
            ip, port = _unxor_addr(attrs[_A_XPEER][0])
            alloc.channels[chan] = (ip, port)
            alloc.perms[ip] = time.time() + 300
        sock.sendto(_msg(0x0109, tid, b"", key), addr)
        return True
    return True


_PUBLIC_IP = None


def _public_ip():
    """IP publica del nodo, para la XOR-RELAYED-ADDRESS. Se resuelve una vez
    abriendo un socket UDP hacia fuera (no envia nada) y se puede forzar con
    VERSELL_PUBLIC_IP en maquinas con NAT delante."""
    global _PUBLIC_IP
    if _PUBLIC_IP:
        return _PUBLIC_IP
    forced = os.environ.get("VERSELL_PUBLIC_IP", "").strip()
    if forced:
        _PUBLIC_IP = forced
        return _PUBLIC_IP
    ip = "0.0.0.0"
    try:
        s = _socket.socket(_socket.AF_INET, _socket.SOCK_DGRAM)
        s.connect(("198.51.100.1", 9))     # TEST-NET-2: no sale trafico real
        ip = s.getsockname()[0]
        s.close()
    except Exception:
        pass
    _PUBLIC_IP = ip
    return ip


def _turn_sweep():
    """Cierra las allocations caducadas (y sus puertos de relay)."""
    now = time.time()
    with _ALLOCS_LOCK:
        dead = [a for a, al in _ALLOCS.items() if now > al.expires]
        for a in dead:
            al = _ALLOCS.pop(a, None)
            if al:
                try:
                    al.sock.close()
                except Exception:
                    pass
    return len(dead)


def _stun_loop(sock):
    while True:
        try:
            data, addr = sock.recvfrom(2048)
        except Exception:
            continue
        try:
            # Rate limit compartido con el resto del nodo: un Binding Request se
            # responde con ~32 bytes frente a 20 recibidos (factor ~1,6), asi que
            # no sirve de amplificador, pero no hace falta atender ráfagas locas.
            # Los datos de una llamada YA establecida (ChannelData / allocation
            # viva) no pasan por el limitador: son 50 paquetes/s legitimos.
            known = False
            if data and (data[0] & 0xC0) == 0x40:
                known = True
            else:
                with _ALLOCS_LOCK:
                    known = addr in _ALLOCS
            if not known and not RATE.allow(addr[0]):
                continue
            if _turn_handle(data, addr, sock):
                continue
            resp = _stun_response(data, addr)
            if resp:
                sock.sendto(resp, addr)
        except Exception:
            continue


def _start_stun():
    global _STUN_SOCK
    if STUN_PORT <= 0 or _STUN_SOCK is not None:
        return
    try:
        s = _socket.socket(_socket.AF_INET, _socket.SOCK_DGRAM)
        s.setsockopt(_socket.SOL_SOCKET, _socket.SO_REUSEADDR, 1)
        # SIEMPRE 0.0.0.0, aunque el HTTP este atado a 127.0.0.1 tras Caddy: el
        # STUN tiene que ser alcanzable directamente desde el movil, no hay proxy.
        s.bind(("0.0.0.0", STUN_PORT))
    except Exception as e:
        # Puerto ocupado (p. ej. coturn ya escucha ahi) o sin permisos: el nodo
        # sigue funcionando igual, solo que sin STUN propio.
        emit_log("[stun] no se pudo escuchar en UDP %d: %s" % (STUN_PORT, e))
        return
    _STUN_SOCK = s
    threading.Thread(target=_stun_loop, args=(s,), name="versell-stun", daemon=True).start()
    emit_log("[stun] servidor STUN escuchando en UDP %d" % STUN_PORT)


def _start_maintenance():
    global _MAINT_STARTED
    if _MAINT_STARTED:
        return
    _MAINT_STARTED = True
    t = threading.Thread(target=_maintenance_loop, name="versell-maint", daemon=True)
    t.start()
    # Avisador de mensajes programados que vencen (ver _sched_loop).
    if SCHED_ENABLED:
        threading.Thread(target=_sched_loop, name="versell-sched", daemon=True).start()


# =====================================================================
#  Noticias (RSS cacheado por el nodo)
#  El nodo descarga titulares e imagenes de fuentes RSS publicas y los
#  sirve a las apps: el telefono NUNCA contacta con los medios (no filtra
#  su IP ni sus intereses). Desactivable con VERSELL_NEWS=0; fuentes por
#  idioma configurables via VERSELL_NEWS_ES/EN/FR/ZH.
# =====================================================================

NEWS_ENABLED = os.environ.get("VERSELL_NEWS", "1") not in ("0", "false", "no")
NEWS_REFRESH_S = int(os.environ.get("VERSELL_NEWS_REFRESH_S", "1800"))

# --- Precios de las criptomonedas cacheados por el nodo (GET /price) ---
# Mismo trato que las noticias y por la misma razon: hasta ahora cada movil llamaba
# a CoinGecko por su cuenta al abrir la cartera, o sea que un tercero veia la IP de
# cada usuario CON cartera Bitcoin — justo el metadato que el proyecto evita en todo
# lo demas ("las descarga el emisor / las descarga el nodo, el movil nunca contacta
# con terceros"). Ahora lo pide el NODO una vez cada 10 min y lo sirve a todos: el
# movil no toca CoinGecko, y de paso funciona donde CoinGecko este bloqueado (China).
#
# Desde 1.31.0-py NO es solo Bitcoin: la misma llamada trae TODAS las monedas de la
# cartera (§0), asi que la pestaña Crypto puede enseñar la equivalencia en EUR/USD/CNY
# de cualquiera de ellas sin una peticion mas y sin que el movil hable con nadie.
# Cuesta lo mismo (una sola URL de CoinGecko con varios `ids`).
PRICE_ENABLED = os.environ.get("VERSELL_PRICE", "1") not in ("0", "false", "no")
PRICE_REFRESH_S = int(os.environ.get("VERSELL_PRICE_REFRESH_S", "600"))
# id de CoinGecko -> simbolo tal y como lo llama la app. `polygon-ecosystem-token` es
# el POL actual (el viejo `matic-network` es MATIC, otra cosa: cotiza distinto).
PRICE_IDS = [
    ("bitcoin", "BTC"),
    ("ethereum", "ETH"),
    ("polygon-ecosystem-token", "POL"),
    ("binancecoin", "BNB"),
    ("avalanche-2", "AVAX"),
    ("solana", "SOL"),
    ("litecoin", "LTC"),
    ("dogecoin", "DOGE"),
    ("usd-coin", "USDC"),
    ("tether", "USDT"),
    ("dai", "DAI"),
]
PRICE_URL = os.environ.get(
    "VERSELL_PRICE_URL",
    "https://api.coingecko.com/api/v3/simple/price?ids=%s&vs_currencies=eur,usd,cny"
    % ",".join(cg_id for cg_id, _ in PRICE_IDS),
)

# --- GIFs/stickers vía KLIPY (GET /klipy/{gifs|stickers}/{trending|search}) ---
# Mismo principio que el precio/las noticias (el NODO habla con el tercero, el movil
# no) pero con una diferencia: aqui la auth firmada se exige SIEMPRE, no solo con
# STRICT_GET. Motivo: /price y /news sirven un unico valor cacheado sin coste marginal
# por peticion; una busqueda de KLIPY consume cuota REAL y limitada del proveedor (la
# clave de prueba de Jorge: 100 llamadas/hora), asi que abrirlo a cualquiera en internet
# agotaria la cuota de inmediato. Limitarlo a peers ya anunciados es contencion de
# coste, no (solo) seguridad.
KLIPY_ENABLED = os.environ.get("VERSELL_KLIPY", "1") not in ("0", "false", "no")
KLIPY_API_KEY = os.environ.get("VERSELL_KLIPY_KEY", "").strip()
KLIPY_BASE = "https://api.klipy.com/api/v1"
KLIPY_CACHE_TTL_S = int(os.environ.get("VERSELL_KLIPY_CACHE_TTL_S", "600"))
KLIPY_CACHE_MAX_ENTRIES = 500
KLIPY_MAX_PER_HOUR = int(os.environ.get("VERSELL_KLIPY_MAX_PER_HOUR", "40"))
KLIPY_MAX_RESPONSE_BYTES = 2 * 1024 * 1024
_KLIPY_CATEGORIES = ("gifs", "stickers")
_KLIPY_ACTIONS = ("trending", "search")

# --- Musica para historias via AUDIUS (GET /audius/{search|trending|stream/<id>}) ---
# Mismo patron que KLIPY (el NODO habla con el tercero; clave y cuota en el servidor;
# auth firmada SIEMPRE) con una pieza mas: aqui el nodo proxya TAMBIEN el AUDIO
# (/audius/stream/<id>), porque quien escucha es el RECEPTOR de la historia y dejarle
# ir directo a Audius romperia la promesa de que un receptor nunca contacta con
# terceros (decision de Jorge, 2026-08-19: "todo por el nodo"). Costes asumidos y
# declarados: el ancho de banda del stream pasa por el VPS, y el nodo ve QUE cancion
# se pide y cuando (nunca quien la escucha respecto a otros peers). La clave es
# OPCIONAL (la API v1 de Audius admite keyless con app_name); con VERSELL_AUDIUS_KEY
# puesta se añade a la peticion. v1 SIN portadas: obligarian al movil a contactar el
# CDN de Audius (fuera) o a proxyar imagenes (v2 si hace falta).
AUDIUS_ENABLED = os.environ.get("VERSELL_AUDIUS", "1") not in ("0", "false", "no")
AUDIUS_KEY = os.environ.get("VERSELL_AUDIUS_KEY", "").strip()
AUDIUS_BASE = os.environ.get("VERSELL_AUDIUS_BASE", "https://api.audius.co/v1").rstrip("/")
AUDIUS_APP_NAME = "dicra"
AUDIUS_CACHE_TTL_S = _envint("VERSELL_AUDIUS_CACHE_TTL_S", 600)
AUDIUS_MAX_PER_HOUR = _envint("VERSELL_AUDIUS_MAX_PER_HOUR", 120)
AUDIUS_MAX_RESPONSE_BYTES = 2 * 1024 * 1024
# Un mp3 de 4 min a 128 kbps ronda los 4 MB; 15 deja holgura sin regalar el disco.
AUDIUS_STREAM_MAX_BYTES = _envint("VERSELL_AUDIUS_STREAM_MAX_MB", 15) * 1024 * 1024


def parse_audius_tracks(raw):
    """Reduce la respuesta de Audius (search/trending) a lo que la app pinta:
    id, titulo, artista y duracion en segundos. Funcion PURA (sin red) para poder
    autoprobarla. Devuelve [] si la respuesta no tiene la forma esperada."""
    try:
        obj = json.loads(raw.decode("utf-8", "replace") if isinstance(raw, bytes) else raw)
    except Exception:
        return []
    data = obj.get("data")
    if not isinstance(data, list):
        return []
    out = []
    for t in data[:25]:
        if not isinstance(t, dict):
            continue
        tid = t.get("id")
        title = t.get("title")
        user = t.get("user") or {}
        artist = user.get("name") if isinstance(user, dict) else None
        dur = t.get("duration")
        if not isinstance(tid, str) or not tid or not isinstance(title, str) or not title:
            continue
        out.append({
            "id": tid[:64],
            "title": title[:120],
            "artist": (artist or "?")[:80] if isinstance(artist, (str, type(None))) else "?",
            "durS": dur if isinstance(dur, int) and 0 < dur < 36000 else None,
        })
    return out


def _audius_norm(s):
    """minusculas + sin acentos + espacios colapsados, para comparar de forma laxa
    ('Rosalía' y 'rosalia' cuentan como lo mismo)."""
    if not isinstance(s, str):
        return ""
    s = unicodedata.normalize("NFKD", s)
    s = "".join(c for c in s if not unicodedata.combining(c))
    return " ".join(s.lower().split())


def filter_audius_by_query(tracks, query):
    """Deja SOLO las pistas cuyo TITULO o ARTISTA contienen de verdad lo buscado. La busqueda
    de Audius es difusa: puntua por relevancia+popularidad y matchea tambien tags y campos que
    el usuario no ve, asi que devuelve pistas que nadie pidio. Aqui exigimos que CADA palabra
    del termino aparezca en 'titulo artista' (normalizado sin acentos). Sin terminos (p.ej.
    trending) no filtra nada. Funcion PURA (sin red) para poder autoprobarla."""
    terms = [t for t in _audius_norm(query).split() if t]
    if not terms:
        return tracks
    out = []
    for t in tracks:
        hay = _audius_norm((t.get("title") or "") + " " + (t.get("artist") or ""))
        if all(term in hay for term in terms):
            out.append(t)
    return out

# --- Proxy JSON-RPC para las carteras cripto (Ethereum/EVM + Solana) ---
# El movil NO habla directo con el RPC de Ankr: le pide al nodo, y el nodo reenvia a
# Ankr con la CLAVE en su entorno (`VERSELL_ANKR_KEY`, NUNCA en el repo ni en el APK
# — un secreto en un cliente publico/AGPL se filtra). Doble beneficio: la clave vive
# solo en el servidor, y la IP del usuario NO llega a Ankr (solo la del nodo). Sin la
# env, el endpoint no existe (404) y la app cae a los RPC publicos keyless de siempre.
# Reenvia SOLO a Ankr y SOLO a cadenas de una lista blanca (no es un proxy abierto);
# el metodo JSON-RPC lo elige el cliente, pero acotado a la superficie de Ankr.
ANKR_KEY = os.environ.get("VERSELL_ANKR_KEY", "").strip()
# subruta de la app -> nombre de cadena en Ankr (rpc.ankr.com/<nombre>/<clave>).
# "multichain" no es una cadena: es la API avanzada de Ankr (ankr_getTransactionsByAddress /
# ankr_getTokenTransfers), que es lo unico keyless-imposible del historial de movimientos —
# BNB Chain no tiene ningun explorador publico sin clave. Va por el mismo camino y con la
# misma regla: la clave se queda aqui y a Ankr solo le llega la IP del nodo.
ANKR_CHAINS = {
    "eth": "eth", "polygon": "polygon", "bsc": "bsc",
    "avalanche": "avalanche", "solana": "solana",
    "multichain": "multichain",
}
RPC_MAX_BYTES = 64 * 1024
# La RESPUESTA se acota aparte de la PETICION: una peticion JSON-RPC nunca pasa de unos
# cientos de bytes (y dejar entrar mas solo regala superficie), pero el historial devuelve
# listas de transacciones y con 64 KB una pagina de 20 movimientos se cortaba -> 502.
RPC_MAX_RESPONSE_BYTES = 512 * 1024


def _safe_int(s, default):
    try:
        return int(s)
    except (TypeError, ValueError):
        return default


NEWS_MAX_ITEMS = 20
NEWS_FEED_MAX_BYTES = 3 * 1024 * 1024
NEWS_IMG_MAX_BYTES = 400 * 1024
NEWS_IMG_BIG_BYTES = 6 * 1024 * 1024  # tope transitorio para reducir con Pillow
NEWS_IMG_MIN_BYTES = 2048  # menos que esto = pixel de rastreo/placeholder (RTVE mete GIFs 1x1)

# Pillow es OPCIONAL: si esta (exe/GUI la traen; en el VPS `apt install python3-pil`),
# las imagenes grandes de los feeds se reducen a ~50 KB; si no, se omiten.
try:
    import io as _io
    from PIL import Image as _PIL_Image
except Exception:
    _PIL_Image = None
# Fuentes por defecto (revisadas 2026-07-25: los feeds viejos de RTVE/francetvinfo/
# BBC-zhongwen estaban MUERTOS — RTVE congelado en 2022, los otros dos daban 301 y
# el nodo no seguia el redirect, por eso las noticias no se actualizaban). Elegidos
# frescos y con imagenes (media:content): El Mundo (es), BBC World (en), France24 (fr)
# y RFI 中文 (zh). Todos publicos. Configurables via VERSELL_NEWS_ES/EN/FR/ZH.
NEWS_FEEDS = {
    "es": os.environ.get("VERSELL_NEWS_ES", "https://e00-elmundo.uecdn.es/elmundo/rss/portada.xml"),
    "en": os.environ.get("VERSELL_NEWS_EN", "https://feeds.bbci.co.uk/news/world/rss.xml"),
    "fr": os.environ.get("VERSELL_NEWS_FR", "https://www.france24.com/fr/rss"),
    "zh": os.environ.get("VERSELL_NEWS_ZH", "https://www.rfi.fr/cn/rss"),
}


def _news_shrink_url(u):
    """Variante reducida para resizers conocidos: algunos feeds (El Pais) traen la
    imagen a resolucion completa (varios MB); su resizer suele aceptar `width` y
    la deja en ~40 KB. OJO: en parte de las URLs firmadas devuelve 403 — por eso
    hay ademas un plan B con Pillow en refresh_lang."""
    if "imagenes.elpais.com/resizer" in u and "width=" not in u:
        return u + ("&" if "?" in u else "?") + "width=480"
    return u


def _news_downscale(data):
    """Reduce una imagen grande a JPEG ~640px con Pillow. None si no se puede."""
    if _PIL_Image is None:
        return None, None
    try:
        im = _PIL_Image.open(_io.BytesIO(data))
        im.thumbnail((640, 640))
        out = _io.BytesIO()
        im.convert("RGB").save(out, "JPEG", quality=80)
        return out.getvalue(), "image/jpeg"
    except Exception:
        return None, None


def _news_fetch_image(u):
    """Baja una imagen aplicando: variante reducida del resizer -> original si
    cabe -> reduccion con Pillow. Descarta pixeles de rastreo (tamaño minimo)."""
    data = ctype = None
    su = _news_shrink_url(u)
    if su != u:
        try:
            data, ctype = _news_http_get(su, NEWS_IMG_MAX_BYTES)
        except Exception:
            data = None
    if data is None:
        try:
            big, bctype = _news_http_get(u, NEWS_IMG_BIG_BYTES)
        except Exception:
            big = None
        if big is not None:
            if len(big) <= NEWS_IMG_MAX_BYTES:
                data, ctype = big, bctype
            else:
                data, ctype = _news_downscale(big)
    if data is not None and len(data) < NEWS_IMG_MIN_BYTES:
        return None, None
    return data, ctype


def _news_http_get(url, max_bytes, timeout=15, partial=False, ua=None):
    """GET con tope de tamaño. Devuelve (bytes|None, content-type).
    Con partial=True devuelve lo leido hasta el tope (util para HTML: el head
    con los og: cabe de sobra) en vez de descartar la respuesta. `ua` permite un
    User-Agent distinto (el modo lector usa uno de navegador para que no le bloqueen)."""
    req = urllib.request.Request(url, headers={"User-Agent": ua or "Versell-Node/1.1 (rss-cache)"})
    with urllib.request.urlopen(req, timeout=timeout) as r:
        data = r.read(max_bytes + 1)
        if len(data) > max_bytes:
            if partial:
                return data[:max_bytes], r.headers.get("Content-Type", "")
            return None, None
        return data, r.headers.get("Content-Type", "application/octet-stream")


def _news_og_image(link):
    """og:image de la pagina del articulo, para feeds que no traen imagen (o solo
    video, como parte de la portada de El Pais). La descarga EL NODO."""
    try:
        raw, _ = _news_http_get(link, 512 * 1024, partial=True)
        if not raw:
            return None
        m = re.search(br'property=["\']og:image["\'][^>]*content=["\']([^"\']+)', raw)
        if m is None:
            m = re.search(br'content=["\']([^"\']+)["\'][^>]*property=["\']og:image', raw)
        return m.group(1).decode("utf-8", "replace") if m else None
    except Exception:
        return None


def _strip_tags(s):
    return re.sub(r"\s+", " ", re.sub(r"<[^>]+>", " ", s or "")).strip()


# ===== Modo lector "Navegar usando Versell" (POST /fetch) — config arriba =====

FETCH_UA = ("Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 "
            "(KHTML, like Gecko) Chrome/125.0.0.0 Safari/537.36")

# Etiquetas que se conservan (solo texto y estructura basica + imagenes).
_READER_KEEP = {
    "p", "br", "hr", "h1", "h2", "h3", "h4", "h5", "h6", "ul", "ol", "li",
    "blockquote", "b", "strong", "i", "em", "u", "span", "div", "article",
    "section", "main", "header", "figure", "figcaption", "img", "table",
    "thead", "tbody", "tr", "td", "th", "pre", "code", "small", "sub", "sup",
    "abbr", "time", "mark", "cite", "dl", "dt", "dd",
}
# Elementos cuyo CONTENIDO se descarta entero (scripts, estilos, interactivos...).
# OJO: NO metemos "head" aqui (dentro va el <title>, que si queremos); sus
# elementos peligrosos (script/style) ya caen por su cuenta, y meta/link se ignoran.
_READER_DROP_TREE = {
    "script", "style", "noscript", "iframe", "object", "embed", "svg", "canvas",
    "form", "template", "video", "audio", "select", "textarea", "button",
    "nav", "aside", "map", "math", "picture",
}
_READER_VOID = {"br", "hr", "img"}


def _ssrf_ok(host):
    """True solo si el host resuelve a IPs PUBLICAS. Bloquea SSRF hacia red interna,
    loopback, link-local, reservadas y metadatos cloud. Best-effort (no cubre el
    rebinding DNS entre esta comprobacion y la descarga real)."""
    if not host:
        return False
    try:
        infos = _socket.getaddrinfo(host, None)
    except Exception:
        return False
    if not infos:
        return False
    for info in infos:
        try:
            a = _ipaddr.ip_address(info[4][0])
        except Exception:
            return False
        if (a.is_private or a.is_loopback or a.is_link_local or a.is_reserved
                or a.is_multicast or a.is_unspecified):
            return False
    return True


def _url_ok(u):
    """URL http/https con host publico (SSRF). Devuelve el host o None."""
    try:
        p = urlparse(u)
    except Exception:
        return None
    if p.scheme not in ("http", "https") or not p.hostname:
        return None
    return p.hostname if _ssrf_ok(p.hostname) else None


class _ReaderParser(_HTMLParser):
    """Reconstruye la pagina quedandose SOLO con texto e imagenes. Quita scripts,
    estilos, formularios/inputs/botones y ENLACES (conserva su texto), y descarta
    todos los atributos salvo el src de <img>. Usa el tokenizador de la stdlib."""

    def __init__(self, base_url):
        super().__init__(convert_charrefs=True)
        self.base = base_url
        self.parts = []
        self._skip = 0
        self.imgs = []
        self._in_title = False
        self.title = ""

    def handle_starttag(self, tag, attrs):
        if self._skip:
            if tag in _READER_DROP_TREE and tag not in _READER_VOID:
                self._skip += 1
            return
        if tag in _READER_DROP_TREE:
            if tag not in _READER_VOID:
                self._skip += 1
            return
        if tag == "title":
            self._in_title = True
            return
        if tag == "a":
            return  # neutraliza el enlace, conserva el texto
        if tag == "img":
            src = None
            for k, v in attrs:
                if k == "src":
                    src = v
                    break
            if src:
                self.parts.append("\x00IMG%d\x00" % len(self.imgs))
                self.imgs.append(urljoin(self.base, src))
            return
        if tag in _READER_KEEP:
            self.parts.append("<%s>" % tag)

    def handle_startendtag(self, tag, attrs):
        self.handle_starttag(tag, attrs)

    def handle_endtag(self, tag):
        if self._skip:
            if tag in _READER_DROP_TREE and tag not in _READER_VOID:
                self._skip -= 1
            return
        if tag == "title":
            self._in_title = False
            return
        if tag in _READER_KEEP and tag not in _READER_VOID:
            self.parts.append("</%s>" % tag)

    def handle_data(self, data):
        if self._skip:
            return
        if self._in_title:
            self.title += data
            return
        self.parts.append(_htmllib.escape(data))


# ----- Extraccion del ARTICULO (heuristica tipo readability) -----------------
# _ReaderParser (arriba) sanea la pagina ENTERA: sirve de respaldo. Lo de aqui
# elige ademas QUE trozo es la noticia, para que el lector chino no muestre el
# menu del medio, la lista de "lo mas leido" ni el pie.

# Clases/ids que delatan lo que NO es el cuerpo, y los que delatan que SI lo es.
# Un nodo que casa con ambos se conserva (p. ej. "article-header", "content-nav").
_UNLIKELY_RE = re.compile(
    r"(comment|disqus|share|social|related|recommend|nav|menu|sidebar|side-bar|"
    r"footer|foot-|masthead|banner|promo|advert|sponsor|popup|modal|cookie|consent|"
    r"newsletter|subscribe|breadcrumb|pagination|pager|copyright|disclaimer|"
    r"toolbar|widget|hot-|ranking|rank-list|tag-list|more-news|extra|aside)", re.I)
_MAYBE_RE = re.compile(
    r"(article|artical|content|con-box|entry|hentry|main|body|post|text|story|"
    r"detail|news-txt|zoom|blog|page-cont)", re.I)
# Publicidad: va aparte porque NO admite rescate por _MAYBE_RE. Caso real de RFI:
# el bloque es "m-block-ad__label" (no casa con "advert") y su span interior es
# "m-block-ad__label__text", que lleva "text" y por tanto colaria la palabra
# 广告 al principio del cuerpo. Un anuncio nunca es cuerpo de noticia.
_AD_RE = re.compile(r"(^|[-_ ])ads?([-_ ]|$)|advert|sponsor|banner", re.I)
# Separadores de frase, latinos y CJK: densidad de puntuacion = prosa de verdad,
# frente a un menu, que son palabras sueltas sin puntuar.
_SEP_RE = re.compile(r"[,.;:!?、。，；：！？…—]")
# Etiquetas que aportan puntos a su contenedor.
_PARA_TAGS = ("p", "td", "pre", "blockquote", "h2", "h3", "section")


class _El:
    """Nodo minimo del arbol: lo justo para puntuar y volver a serializar."""

    __slots__ = ("tag", "attr", "src", "kids", "parent")

    def __init__(self, tag, attr="", src="", parent=None):
        self.tag = tag
        self.attr = attr      # class + id en minusculas (alimenta las heuristicas)
        self.src = src        # solo <img>, ya resuelto contra la base
        self.kids = []        # _El | str
        self.parent = parent


class _DomParser(_HTMLParser):
    """Construye el arbol de la pagina y captura <title>/og:title. Descarta enteros
    los subarboles inutiles (script/style/nav/aside/form...) igual que el saneador."""

    def __init__(self, base_url):
        super().__init__(convert_charrefs=True)
        self.base = base_url
        self.root = _El("body")
        self.cur = self.root
        self._skip = 0
        self._in_title = False
        self.title = ""
        self.og_title = ""

    def handle_starttag(self, tag, attrs):
        if self._skip:
            if tag in _READER_DROP_TREE and tag not in _READER_VOID:
                self._skip += 1
            return
        if tag in _READER_DROP_TREE:
            if tag not in _READER_VOID:
                self._skip += 1
            return
        if tag == "title":
            self._in_title = True
            return
        d = dict(attrs)
        if tag == "meta":
            prop = (d.get("property") or d.get("name") or "").lower()
            if prop in ("og:title", "twitter:title") and not self.og_title:
                self.og_title = d.get("content") or ""
            return
        if tag == "img":
            # Muchos medios chinos cargan la imagen de verdad en data-src y dejan
            # un placeholder en src; sin esto el lector se queda con el hueco.
            src = (d.get("data-src") or d.get("data-original")
                   or d.get("data-lazy-src") or d.get("src") or "").strip()
            if src:
                self.cur.kids.append(_El("img", "", urljoin(self.base, src), self.cur))
            return
        attr = ((d.get("class") or "") + " " + (d.get("id") or "") + " "
                + (d.get("itemprop") or "")).lower()
        el = _El(tag, attr, "", self.cur)
        self.cur.kids.append(el)
        if tag not in _READER_VOID:
            self.cur = el

    def handle_startendtag(self, tag, attrs):
        was = self.cur
        self.handle_starttag(tag, attrs)
        self.cur = was          # <div/> no abre nivel

    def handle_endtag(self, tag):
        if self._skip:
            if tag in _READER_DROP_TREE and tag not in _READER_VOID:
                self._skip -= 1
            return
        if tag == "title":
            self._in_title = False
            return
        if tag in _READER_VOID:
            return
        # HTML real viene con etiquetas sin cerrar: subir hasta la que casa, sin
        # pasarse de la raiz (si no aparece, el cierre es huerfano y se ignora).
        n = self.cur
        while n is not None and n.tag != tag:
            n = n.parent
        if n is not None and n.parent is not None:
            self.cur = n.parent

    def handle_data(self, data):
        if self._skip:
            return
        if self._in_title:
            self.title += data
            return
        if data.strip():
            self.cur.kids.append(data)


def _el_text(el):
    """Texto plano del subarbol. Iterativo: hay paginas con anidamiento absurdo y
    una recursion aqui se lleva por delante el hilo entero."""
    if isinstance(el, str):
        return el
    out, stack = [], [el]
    while stack:
        n = stack.pop()
        if isinstance(n, str):
            out.append(n)
        else:
            stack.extend(reversed(n.kids))
    return "".join(out)


def _link_text_len(el):
    """Caracteres que viven dentro de un <a>. Un bloque casi todo enlaces es un
    menu o un 'te puede interesar', no el cuerpo de la noticia."""
    total, stack = 0, [el]
    while stack:
        n = stack.pop()
        if isinstance(n, str):
            continue
        if n.tag == "a":
            total += len(_el_text(n))
            continue          # ya contado entero
        stack.extend(n.kids)
    return total


def _base_score(el):
    """Puntos por la pinta del contenedor, antes de mirar su texto."""
    s = 0.0
    if _MAYBE_RE.search(el.attr):
        s += 12
    if _UNLIKELY_RE.search(el.attr) and not _MAYBE_RE.search(el.attr):
        s -= 20
    if _AD_RE.search(el.attr):
        s -= 25
    if el.tag == "article":
        s += 25
    elif el.tag in ("main", "section", "div"):
        s += 3
    return s


def _extract_article(root):
    """Devuelve el nodo con mas pinta de ser el cuerpo de la noticia, o None.
    Cada parrafo con texto real reparte puntos a su padre (y la mitad al abuelo);
    luego se penaliza por densidad de enlaces y se sube un poco si el padre agrupa
    mas cuerpo (hay medios que parten el articulo en varios divs hermanos)."""
    nodes, stack = [], [root]
    while stack:
        n = stack.pop()
        if isinstance(n, str):
            continue
        nodes.append(n)
        stack.extend(n.kids)

    # Marcado semantico explicito: si el medio lo dice, no hay que adivinar.
    for n in nodes:
        if "articlebody" in n.attr and len(_el_text(n).strip()) >= FETCH_ARTICLE_MIN:
            return n

    scores, by_id = {}, {}
    for n in nodes:
        if n.tag not in _PARA_TAGS:
            continue
        txt = _el_text(n).strip()
        if len(txt) < 25:
            continue
        pts = 1.0 + len(_SEP_RE.findall(txt)) + min(len(txt) / 100.0, 3.0)
        for anc, share in ((n.parent, 1.0), (n.parent.parent if n.parent else None, 0.5)):
            if anc is None:
                continue
            k = id(anc)
            if k not in by_id:
                by_id[k] = anc
                scores[k] = _base_score(anc)
            scores[k] += pts * share

    best, best_sc = None, 0.0
    for k, n in by_id.items():
        txt = _el_text(n)
        if not txt.strip():
            continue
        dens = _link_text_len(n) / float(max(len(txt), 1))
        sc = scores[k] * (1.0 - min(dens, 1.0))
        if sc > best_sc:
            best, best_sc = n, sc
    if best is None:
        return None

    # Subir mientras el padre puntue casi igual y NO se trague media pagina.
    for _ in range(3):
        p = best.parent
        if p is None or id(p) not in scores:
            break
        ptxt, btxt = len(_el_text(p)), len(_el_text(best))
        if scores[id(p)] >= best_sc * 0.9 and ptxt <= btxt * 1.6:
            best, best_sc = p, scores[id(p)]
        else:
            break
    return best


def _render_article(el):
    """Serializa el subarbol elegido con las MISMAS reglas del saneador: solo
    etiquetas de _READER_KEEP, sin atributos, enlaces neutralizados (se conserva su
    texto) e imagenes como marcador. Iterativo, por lo mismo que _el_text."""
    parts, imgs = [], []
    stack = [(el, "open")]
    while stack:
        node, mode = stack.pop()
        if mode == "close":
            parts.append("</%s>" % node)
            continue
        if isinstance(node, str):
            parts.append(_htmllib.escape(node))
            continue
        t = node.tag
        if t == "img":
            if node.src:
                parts.append("\x00IMG%d\x00" % len(imgs))
                imgs.append(node.src)
            continue
        # Cajas de compartir/relacionadas/anuncios incrustadas EN MITAD del cuerpo.
        if node is not el and (_AD_RE.search(node.attr) or
                               (_UNLIKELY_RE.search(node.attr)
                                and not _MAYBE_RE.search(node.attr))):
            continue
        if t in _READER_KEEP:
            if t in _READER_VOID:
                parts.append("<%s>" % t)
            else:
                parts.append("<%s>" % t)
                stack.append((t, "close"))
        for k in reversed(node.kids):
            stack.append((k, "open"))
    return parts, imgs


_TITLE_TAIL_RE = re.compile(r"\s*[|｜_\-–—·]\s*")


def _clean_title(raw):
    """Quita el sufijo del medio: los titulares chinos suelen venir como
    '标题_联合早报' o '标题 - BBC News 中文'. Solo se corta si la ultima pieza es
    corta (nombre de medio) y queda titular de sobra: un guion dentro del titular
    no debe partirlo."""
    t = re.sub(r"\s+", " ", raw or "").strip()
    if not t:
        return ""
    parts = [p for p in _TITLE_TAIL_RE.split(t) if p.strip()]
    if len(parts) > 1 and len(parts[-1]) <= 20:
        head = " ".join(parts[:-1]).strip()
        if len(head) >= 8:
            return head
    return t


def _pick_title(dom, art, url):
    """Titular: el <h1> del articulo manda (es el de la pieza), luego og:title y
    por ultimo el <title> del navegador, ya sin el nombre del medio."""
    if art is not None:
        stack = [art]
        while stack:
            n = stack.pop()
            if isinstance(n, str):
                continue
            if n.tag == "h1":
                h = re.sub(r"\s+", " ", _el_text(n)).strip()
                if h:
                    return h
            stack.extend(reversed(n.kids))
    for cand in (dom.og_title, dom.title):
        t = _clean_title(cand)
        if t:
            return t
    return urlparse(url).hostname or ""


def _reader_inline(imgs):
    """Descarga las imagenes (SSRF + topes) y devuelve {indice: data-URI}. Las
    reduce con Pillow si esta; descarta las que fallen o excedan el presupuesto."""
    out = {}
    total = 0
    for i, src in enumerate(imgs):
        if i >= FETCH_IMG_COUNT or total >= FETCH_INLINE_TOTAL:
            break
        if not _url_ok(src):
            continue
        try:
            data, ctype = _news_http_get(src, FETCH_IMG_MAX, timeout=FETCH_TIMEOUT, ua=FETCH_UA)
        except Exception:
            data = ctype = None
        if data is None:                      # grande -> intentar reducir con Pillow
            try:
                big, _ = _news_http_get(src, NEWS_IMG_BIG_BYTES, timeout=FETCH_TIMEOUT, ua=FETCH_UA)
            except Exception:
                big = None
            if big is not None:
                data, ctype = _news_downscale(big)
        if not data or len(data) < NEWS_IMG_MIN_BYTES:
            continue
        if _PIL_Image is not None and len(data) > FETCH_IMG_MAX:
            data, ctype = _news_downscale(data)
            if not data:
                continue
        if total + len(data) > FETCH_INLINE_TOTAL:
            continue
        total += len(data)
        mime = (ctype or "image/jpeg").split(";")[0].strip() or "image/jpeg"
        out[i] = "data:%s;base64,%s" % (mime, base64.b64encode(data).decode())
    return out


_READER_CSS = (
    "html{-webkit-text-size-adjust:100%}"
    "body{margin:0;padding:18px 16px;background:#0B1512;color:#DCE6E2;"
    "font-family:-apple-system,Roboto,'Noto Sans','Noto Sans CJK SC',sans-serif;"
    "font-size:17px;line-height:1.6;overflow-wrap:break-word;word-break:break-word}"
    "h1,h2,h3{line-height:1.25;color:#F1F6F4}h1{font-size:1.5em}h2{font-size:1.25em}"
    "img{max-width:100%;height:auto;border-radius:8px;margin:10px 0;display:block}"
    "p{margin:0 0 1em}pre,code{white-space:pre-wrap;font-family:monospace;font-size:.9em}"
    "table{max-width:100%;border-collapse:collapse}td,th{border:1px solid #2A3A36;padding:4px 8px}"
    "hr{border:0;border-top:1px solid #2A3A36;margin:1.4em 0}"
)


_META_CHARSET_RE = re.compile(br"""<meta[^>]+charset\s*=\s*["']?\s*([A-Za-z0-9_.:-]+)""", re.I)


def _charset_alias(name):
    """Normaliza a un SUPERCONJUNTO del charset declarado: muchas paginas dicen
    'gb2312' y luego meten caracteres que solo existen en GBK/GB18030 (y otras
    dicen iso-8859-1 usando comillas de cp1252). Decodificar con el superconjunto
    evita fallar por un solo byte."""
    n = (name or "").strip().lower().replace("_", "-")
    return {
        "gb2312": "gb18030", "gbk": "gb18030", "cp936": "gb18030", "ms936": "gb18030",
        "big5": "big5hkscs", "cp950": "big5hkscs",
        "iso-8859-1": "cp1252", "latin-1": "cp1252", "latin1": "cp1252", "ascii": "utf-8",
    }.get(n, n)


def _decode_html(raw, content_type=""):
    """Decodifica HTML respetando SU codificacion real, no la que nos convenga.
    Orden: charset del Content-Type -> <meta charset> -> utf-8 estricto -> chino
    (gb18030 / big5hkscs) -> occidental. Sin esto, una pagina en GBK/Big5 —muy
    comunes en sitios chinos— se leia como utf-8 y cada caracter chino salia como
    un rombo: el lector mostraba basura justo en el idioma donde mas se usa.
    Ojo con `partial=True`: el tope de descarga puede cortar un caracter multibyte
    por la mitad, asi que un fallo SOLO en los ultimos bytes no descarta utf-8."""
    m = re.search(r"charset\s*=\s*([A-Za-z0-9_.:-]+)", content_type or "", re.I)
    declared = _charset_alias(m.group(1)) if m else None
    if not declared:
        m2 = _META_CHARSET_RE.search(raw[:4096])
        if m2:
            declared = _charset_alias(m2.group(1).decode("ascii", "ignore"))
    if declared:
        try:
            return raw.decode(declared, "replace")
        except LookupError:
            pass  # charset inventado o desconocido: seguimos con la deteccion
    # Sin declaracion (o invalida): probar utf-8 estricto, tolerando un corte final.
    for cut in (0, 1, 2, 3):
        try:
            return raw[:len(raw) - cut].decode("utf-8") if cut else raw.decode("utf-8")
        except UnicodeDecodeError:
            continue
    for cand in ("gb18030", "big5hkscs", "cp1252"):
        try:
            return raw.decode(cand)
        except (UnicodeDecodeError, LookupError):
            continue
    return raw.decode("utf-8", "replace")


# Codigos que suelen significar "me has parecido un bot", no "no existe".
_UA_RETRY_CODES = (401, 403, 405, 406, 429, 451, 500, 502, 503)
# Cliente honesto para el reintento: los anti-bot que exigen coherencia entre el UA
# y la huella real del navegador (TLS, orden de cabeceras) rechazan un UA de Chrome
# falseado, pero dejan pasar a un cliente que se identifica como lo que es.
_UA_FALLBACKS = ("curl/8.5.0", None)


def _reader_get(url):
    """Descarga para el lector con REINTENTO DE USER-AGENT. Caso real (RFI 中文, la
    fuente china del nodo): con el UA de Chrome devuelve 403 —su anti-bot ve un
    navegador falso— y con un cliente honesto devuelve 200. Como el lector usaba
    solo el UA de navegador, TODAS las noticias chinas fallaban con 502 al abrirlas."""
    try:
        return _news_http_get(url, FETCH_HTML_MAX, timeout=FETCH_TIMEOUT, partial=True, ua=FETCH_UA)
    except Exception as e:
        if getattr(e, "code", None) not in _UA_RETRY_CODES:
            raise
        for ua in _UA_FALLBACKS:
            try:
                return _news_http_get(url, FETCH_HTML_MAX, timeout=FETCH_TIMEOUT, partial=True, ua=ua)
            except Exception:
                continue
        raise


def _reader_page(url):
    """Descarga y sanea una pagina. Devuelve (title, html_autocontenido). Lanza
    ValueError si la URL no es valida/segura o no hay contenido."""
    if not _url_ok(url):
        raise ValueError("URL no permitida")
    raw, ctype = _reader_get(url)
    if not raw:
        raise ValueError("Sin contenido")
    text = _decode_html(raw, ctype)

    parts = imgs = None
    title = ""
    if FETCH_ARTICLE:
        # Camino preferente: quedarse SOLO con el cuerpo de la noticia.
        dom = _DomParser(url)
        try:
            dom.feed(text)
        except Exception:
            pass
        art = _extract_article(dom.root)
        # Si lo extraido es raquitico, la heuristica se equivoco (portada, muro de
        # pago, maquetado raro): mejor la pagina entera que tres frases sueltas.
        if art is not None and len(_el_text(art).strip()) >= FETCH_ARTICLE_MIN:
            parts, imgs = _render_article(art)
            title = _pick_title(dom, art, url)

    if parts is None:
        parser = _ReaderParser(url)
        try:
            parser.feed(text)
        except Exception:
            pass
        parts, imgs = parser.parts, parser.imgs
        title = _clean_title(parser.title) or (urlparse(url).hostname or "")

    inlined = _reader_inline(imgs)
    body = []
    for part in parts:
        if part.startswith("\x00IMG"):
            uri = inlined.get(int(part[4:-1]))
            body.append('<img src="%s" alt="">' % uri if uri else "")
        else:
            body.append(part)
    body = "".join(body)
    # El titular suele vivir FUERA del contenedor del cuerpo: si no ha entrado, se
    # pone delante para que el lector abra siempre con la noticia identificada.
    head = ""
    if title and "<h1>" not in body:
        head = "<h1>%s</h1>" % _htmllib.escape(title)
    # Nada de cabecera con la URL de origen: era una tira de 250 caracteres
    # percent-encoded delante del titular que ademas PARECIA un enlace (no lo es,
    # aqui no sale ni un <a>). El usuario ya sabe de donde viene: ha tocado ese
    # titular en la pestaña Noticias, que muestra el medio.
    html = (
        "<!doctype html><html><head><meta charset=\"utf-8\">"
        "<meta name=\"viewport\" content=\"width=device-width,initial-scale=1\">"
        "<title>%s</title><style>%s</style></head><body>%s%s</body></html>"
        % (_htmllib.escape(title), _READER_CSS, head, body)
    )
    return title, html


class NewsCache:
    """Titulares e imagenes en memoria, por idioma. Refresco en hilo propio."""

    ATOM = "{http://www.w3.org/2005/Atom}"
    MEDIA = "{http://search.yahoo.com/mrss/}"

    def __init__(self):
        self.lock = threading.Lock()
        self.langs = {}   # lang -> {"updated", "source", "items": [...]}
        self.images = {}  # (lang, id) -> (content-type, bytes)

    def get(self, lang):
        with self.lock:
            return self.langs.get(lang)

    def get_image(self, lang, iid):
        with self.lock:
            return self.images.get((lang, iid))

    def refresh_lang(self, lang, url):
        raw, _ = _news_http_get(url, NEWS_FEED_MAX_BYTES)
        if raw is None:
            return
        root = ET.fromstring(raw)
        nodes = root.findall(".//item") or root.findall(self.ATOM + "entry")
        items = []
        for it in nodes[:NEWS_MAX_ITEMS]:
            def txt(tag):
                e = it.find(tag)
                return (e.text or "").strip() if e is not None and e.text else ""
            title = txt("title") or txt(self.ATOM + "title")
            if not title:
                continue
            link = txt("link")
            if not link:
                e = it.find(self.ATOM + "link")
                link = e.get("href", "") if e is not None else ""
            raw_desc = txt("description") or txt(self.ATOM + "summary")
            desc = _strip_tags(html.unescape(raw_desc))[:220]
            date = txt("pubDate") or txt(self.ATOM + "updated")
            # Imagen: SOLO variantes de imagen (los feeds mezclan media:content de
            # video, p. ej. El Pais). Orden: enclosure -> media:content/thumbnail
            # -> <img> de la descripcion. Si no hay, og:image del articulo (luego).
            img_url = None
            enc = it.find("enclosure")
            if enc is not None and enc.get("url"):
                et = (enc.get("type") or "").lower()
                if "image" in et or (not et and re.search(r"\.(jpe?g|png|webp|gif)(\?|$)", enc.get("url"), re.I)):
                    img_url = enc.get("url")
            if not img_url:
                for m in it.findall(self.MEDIA + "content"):
                    t = ((m.get("type") or "") + (m.get("medium") or "")).lower()
                    if m.get("url") and ("image" in t or not t):
                        img_url = m.get("url")
                        break
            if not img_url:
                mt = it.find(self.MEDIA + "thumbnail")
                if mt is not None:
                    img_url = mt.get("url")
            if not img_url:
                mm = re.search(r'<img[^>]+src="([^"]+)"', raw_desc)
                if mm:
                    img_url = mm.group(1)
            iid = hashlib.sha256((link or title).encode("utf-8")).hexdigest()[:12]
            items.append({"id": iid, "title": html.unescape(title)[:220], "link": link,
                          "summary": desc, "date": date, "img": False, "_img_url": img_url})
        # Descargar las imagenes (con tope de tamaño) FUERA del lock.
        imgs = {}
        for entry in items:
            u = entry.pop("_img_url", None)
            data = ctype = None
            # 1) La imagen que declara el feed (si es utilizable).
            if u and u.startswith("http"):
                data, ctype = _news_fetch_image(u)
            # 2) Si no hay (o era un pixel de rastreo): og:image del articulo.
            if data is None and entry.get("link", "").startswith("http"):
                og = _news_og_image(entry["link"])
                if og and og.startswith("http"):
                    data, ctype = _news_fetch_image(og)
            if data:
                imgs[(lang, entry["id"])] = (ctype or "image/jpeg", data)
                entry["img"] = True
            time.sleep(0.2)  # cortesia con el CDN de la fuente
        with self.lock:
            self.langs[lang] = {"updated": int(time.time() * 1000), "source": url, "items": items}
            vivos = {(lang, e["id"]) for e in items}
            self.images = {k: v for k, v in self.images.items() if k[0] != lang or k in vivos}
            self.images.update(imgs)

    def refresh_all(self):
        for lang, url in NEWS_FEEDS.items():
            if not url:
                continue
            try:
                self.refresh_lang(lang, url)
            except Exception as e:
                emit_log("[noticias] fallo al refrescar %s: %s" % (lang, e))


NEWS = NewsCache()
_NEWS_STARTED = False


def parse_prices(raw, now_ms=None):
    """Convierte la respuesta de CoinGecko en el cuerpo de `GET /price`. Funcion aparte
    (sin red ni estado) para poder probarla en --selftest.

    El formato es RETROCOMPATIBLE a proposito: `eur`/`usd`/`cny` siguen siendo el precio
    de 1 BTC en la raiz —es lo que leen las apps <1.0.201 y el retiro de Promo, que
    convierte puntos a sats con `price["eur"]`— y las demas monedas van en `coins`:

        {"eur":..,"usd":..,"cny":..,"updated":ms,
         "coins":{"BTC":{"eur":..,"usd":..,"cny":..}, "ETH":{...}, ...}}

    Devuelve None si no se pudo sacar ni un precio (respuesta rara o proveedor caido):
    el llamante deja entonces el valor cacheado anterior en pie en vez de borrarlo.
    """
    obj = json.loads(raw.decode("utf-8", "replace"))
    if not isinstance(obj, dict):
        return None
    coins = {}
    for cg_id, sym in PRICE_IDS:
        src = obj.get(cg_id)
        if not isinstance(src, dict):
            continue
        quote = {}
        for c in ("eur", "usd", "cny"):
            v = src.get(c)
            if isinstance(v, (int, float)) and not isinstance(v, bool) and v > 0:
                quote[c] = float(v)
        if quote:
            coins[sym] = quote
    if not coins:
        return None
    out = dict(coins.get("BTC") or {})   # raiz = BTC, como siempre
    out["updated"] = int(now_ms if now_ms is not None else time.time() * 1000)
    out["coins"] = coins
    return out


class PriceCache:
    """Precio de cada cripto de la cartera en EUR/USD/CNY, cacheado en memoria. Una
    llamada por nodo cada PRICE_REFRESH_S, no una por usuario y apertura de cartera."""

    def __init__(self):
        self.lock = threading.Lock()
        self.data = None      # {"eur":..,"usd":..,"cny":..,"updated":ms,"coins":{...}}

    def get(self):
        with self.lock:
            return self.data

    def refresh(self):
        raw, _ = _news_http_get(PRICE_URL, 64 * 1024, timeout=15)
        if not raw:
            return
        out = parse_prices(raw)
        if not out:
            return
        with self.lock:
            self.data = out


PRICE = PriceCache()


def _price_loop():
    while True:
        try:
            PRICE.refresh()
        except Exception:
            pass
        time.sleep(PRICE_REFRESH_S)


def _news_loop():
    while True:
        try:
            NEWS.refresh_all()
        except Exception:
            pass
        time.sleep(NEWS_REFRESH_S)


def _start_news():
    global _NEWS_STARTED
    if not NEWS_ENABLED or _NEWS_STARTED:
        return
    _NEWS_STARTED = True
    emit_log("  Noticias RSS: activadas (%s); refresco cada %ds" % (",".join(NEWS_FEEDS), NEWS_REFRESH_S))
    threading.Thread(target=_news_loop, name="versell-news", daemon=True).start()


_PRICE_STARTED = False


def _start_price():
    global _PRICE_STARTED
    if not PRICE_ENABLED or _PRICE_STARTED:
        return
    _PRICE_STARTED = True
    emit_log("  Precios cripto (%s): activado; refresco cada %ds"
             % ("/".join(sym for _, sym in PRICE_IDS), PRICE_REFRESH_S))
    threading.Thread(target=_price_loop, name="versell-price", daemon=True).start()


class KlipyCache:
    """Cache LRU+TTL por clave de peticion (categoria+accion+busqueda+pagina).

    A diferencia de PriceCache/NewsCache (un unico valor GLOBAL compartido por
    todos), una busqueda de KLIPY es por texto/por usuario -- no hay "un" resultado
    que cachear para todo el mundo. Pero SI vale la pena cachear peticiones repetidas
    (el mismo trending, la misma busqueda popular) un rato, hoy sobre todo por el
    limite real de 100 llamadas/hora de la clave de prueba."""

    def __init__(self, max_entries, ttl_s):
        self.max_entries = max_entries
        self.ttl_s = ttl_s
        self.lock = threading.Lock()
        self.data = collections.OrderedDict()  # clave -> (expira_ts, dict ya parseado)

    def get(self, key):
        with self.lock:
            entry = self.data.get(key)
            if entry is None:
                return None
            expires_at, value = entry
            if time.time() > expires_at:
                return None
            self.data.move_to_end(key)
            return value

    def get_stale(self, key):
        """Devuelve el valor aunque haya caducado: mejor una respuesta vieja que un
        error, si KLIPY falla o se agoto la cuota propia."""
        with self.lock:
            entry = self.data.get(key)
            return entry[1] if entry else None

    def put(self, key, value):
        with self.lock:
            self.data[key] = (time.time() + self.ttl_s, value)
            self.data.move_to_end(key)
            while len(self.data) > self.max_entries:
                self.data.popitem(last=False)


class KlipyRateLimiter:
    """Ventana deslizante de 1h GLOBAL (no por IP, a diferencia de RateLimiter):
    protege la cuota REAL y limitada de la cuenta de KLIPY (compartida entre los 2
    VPS), no reparte cuota entre usuarios. Best-effort: se resetea si el proceso
    reinicia, no es una ventana persistente exacta -- es una salvaguarda, no una
    garantia."""

    def __init__(self, per_hour):
        self.per_hour = per_hour
        self.lock = threading.Lock()
        self.hits = collections.deque()

    def allow(self):
        if self.per_hour <= 0:
            return True
        now = time.time()
        with self.lock:
            while self.hits and now - self.hits[0] > 3600:
                self.hits.popleft()
            if len(self.hits) >= self.per_hour:
                return False
            self.hits.append(now)
            return True


KLIPY_CACHE = KlipyCache(KLIPY_CACHE_MAX_ENTRIES, KLIPY_CACHE_TTL_S)
KLIPY_LIMITER = KlipyRateLimiter(KLIPY_MAX_PER_HOUR)
# Audius reutiliza las MISMAS clases (cache TTL + limitador por hora): la mecanica
# es identica, solo cambian el proveedor y los topes.
AUDIUS_CACHE = KlipyCache(300, AUDIUS_CACHE_TTL_S)
AUDIUS_LIMITER = KlipyRateLimiter(AUDIUS_MAX_PER_HOUR)

# --- Traduccion de mensajes via AZURE TRANSLATOR (POST /translate) ---
# Mismo principio que Ankr/KLIPY: la CLAVE vive en el entorno del nodo, NUNCA en el APK
# (publico/AGPL -> un secreto en el cliente se filtra) ni en el repo. La app manda el texto
# al nodo y el nodo lo traduce contra Azure; asi la clave se queda aqui y la IP del usuario
# no llega a Microsoft (solo la del nodo). Se configura con `node-ops.sh set-azure-key`.
AZURE_TRANSLATE_ENABLED = os.environ.get("VERSELL_TRANSLATE", "1") not in ("0", "false", "no")
AZURE_KEY = os.environ.get("VERSELL_AZURE_KEY", "").strip()
AZURE_REGION = os.environ.get("VERSELL_AZURE_REGION", "francecentral").strip()
AZURE_URL = os.environ.get("VERSELL_AZURE_URL", "https://api.cognitive.microsofttranslator.com").rstrip("/")
AZURE_MAX_TEXTS = _envint("VERSELL_AZURE_MAX_TEXTS", 25)          # elementos por peticion
AZURE_MAX_CHARS = _envint("VERSELL_AZURE_MAX_CHARS", 5000)        # total de caracteres
AZURE_MAX_PER_HOUR = _envint("VERSELL_AZURE_MAX_PER_HOUR", 600)
AZURE_CACHE = KlipyCache(2000, _envint("VERSELL_AZURE_CACHE_TTL_S", 86400))
AZURE_LIMITER = KlipyRateLimiter(AZURE_MAX_PER_HOUR)
# Codigo de idioma valido de Azure: letras, y opcionalmente -Script o -REGION (es, zh-Hans, pt-BR).
_LANG_RE = re.compile(r"^[A-Za-z]{2,3}(-[A-Za-z]{2,8}){0,2}$")


def parse_azure_translations(raw, want):
    """Reduce la respuesta de Azure Translator a la lista de textos traducidos que espera
    la app, en el MISMO orden que se pidieron (`want` = cuantos textos se enviaron).
    Devuelve (lista_de_textos, idioma_detectado_o_None) o None si la forma no cuadra.
    Funcion PURA (sin red) para poder autoprobarla."""
    try:
        arr = json.loads(raw.decode("utf-8", "replace") if isinstance(raw, bytes) else raw)
    except Exception:
        return None
    if not isinstance(arr, list) or len(arr) != want:
        return None
    out, detected = [], None
    for item in arr:
        if not isinstance(item, dict):
            return None
        tr = item.get("translations")
        if not isinstance(tr, list) or not tr or not isinstance(tr[0], dict):
            return None
        txt = tr[0].get("text")
        if not isinstance(txt, str):
            return None
        out.append(txt)
        dl = item.get("detectedLanguage")
        if detected is None and isinstance(dl, dict) and isinstance(dl.get("language"), str):
            detected = dl["language"]
    return out, detected


def build_server(port, data_dir, bind=None):
    """Crea el servidor HTTP (sin arrancarlo) y prepara el almacen. Devuelve httpd.

    `bind` (o el env VERSELL_BIND) permite atarlo a 127.0.0.1 cuando hay un
    reverse proxy TLS delante; por defecto 0.0.0.0 (autoalojados)."""
    global STORE, PROMO
    os.makedirs(data_dir, exist_ok=True)
    STORE = Store(data_dir)
    PROMO = PromoStore(data_dir)
    if bind is None:
        bind = os.environ.get("VERSELL_BIND", "0.0.0.0")
    httpd = GuardedServer((bind, int(port)), Handler)
    httpd.daemon_threads = True
    _start_maintenance()  # purga periodica de buzones y archivos caducados
    _start_news()         # cache RSS de noticias (opcional, VERSELL_NEWS=0 la apaga)
    _start_price()        # cache del precio BTC (opcional, VERSELL_PRICE=0 lo apaga)
    _start_stun()         # STUN propio en UDP (llamadas fuera de la LAN sin coturn)
    if KLIPY_ENABLED and KLIPY_API_KEY:
        emit_log("  GIFs/stickers (KLIPY): activado; tope propio %d/h" % KLIPY_MAX_PER_HOUR)
    elif KLIPY_ENABLED:
        emit_log("  GIFs/stickers (KLIPY): sin VERSELL_KLIPY_KEY, endpoint dara 503")
    if PROMO_ENABLED:
        if PROMO_DAILY_CAP_SATS > 0:
            emit_log("  Promo: activado; tope global %d sats/dia (pago real lo procesa promo_pay.py, aparte)"
                     % PROMO_DAILY_CAP_SATS)
        else:
            emit_log("  Promo: activado; SIN tope global -- solo limites por persona "
                     "(pago real lo procesa promo_pay.py, aparte)")
    else:
        emit_log("  Promo: sin VERSELL_PROMO_ENABLED, endpoint dara 503")
    return httpd

# =====================================================================
#  Servidor HTTP + WebSocket
# =====================================================================

class Handler(BaseHTTPRequestHandler):
    protocol_version = "HTTP/1.1"
    timeout = SOCKET_TIMEOUT_S  # anti-slowloris en la fase HTTP

    def version_string(self):
        # En modo DECOY el header Server no revela Python/BaseHTTP.
        return "nginx" if DECOY else BaseHTTPRequestHandler.version_string(self)

    def handle(self):
        # Limita conexiones vivas simultaneas (anti-agotamiento de hilos/FD).
        if not CONN_SEM.acquire(blocking=False):
            try:
                self.close_connection = True
                self._err(503, "Servidor ocupado")
            except Exception:
                pass
            return
        try:
            BaseHTTPRequestHandler.handle(self)
        finally:
            CONN_SEM.release()

    def log_message(self, fmt, *args):
        """Access log SIN la IP del cliente.

        Hasta 1.16.0-py esto anteponia `self.address_string()`, o sea la IP, a cada
        peticion. El efecto real no era "una linea de log": journald conserva por
        defecto sin limite de tiempo, asi que el nodo acumulaba un REGISTRO DE TRAFICO
        CON IP de todos sus usuarios durante semanas (1,2 GB y 19 dias cuando se
        detecto, el 2026-08-01) -- exactamente lo que la politica de privacidad dice
        que no se hace. Se conservan metodo, ruta y codigo, que es lo que sirve para
        diagnosticar; quien haya hecho la peticion no aporta nada operativo."""
        emit_log(fmt % args)

    @property
    def client_ip(self):
        """IP real del cliente. Solo confia en X-Forwarded-For si la conexion
        llega de loopback (el reverse proxy); asi un cliente directo no puede
        falsificar la cabecera para evadir el rate limit."""
        peer = self.client_address[0] if self.client_address else "?"
        if TRUST_PROXY and peer in ("127.0.0.1", "::1"):
            xff = self.headers.get("X-Forwarded-For")
            if xff:
                return xff.split(",")[0].strip()
        return peer

    def _ratelimited(self):
        """Puerta de entrada de toda peticion: limite de tasa por IP (ventana
        deslizante en memoria, sin persistencia ni identificacion de quien pega)."""
        if not RATE.allow(self.client_ip):
            self._err(429, "Demasiadas peticiones")
            return True
        return False

    def _has_admin_key(self):
        """True si la peticion trae la clave de admin valida. SIN efectos
        secundarios (no cuenta como intento fallido ni escribe respuesta): sirve
        para dejar ver la salud en '/' al operador cuando DECOY esta activo."""
        if not ADMIN_KEY:
            return False
        key = self.headers.get("X-Admin-Key") or parse_qs(urlparse(self.path).query).get("key", [""])[0]
        return bool(key) and _consteq(key, ADMIN_KEY)

    def _get_auth_ok(self, scope):
        """Fase 2 (solo si STRICT_GET): valida el token firmado de un GET
        (cabeceras X-Versell-Peer/Ts/Sig) contra la clave Ed25519 ANUNCIADA del
        solicitante. Mismo framing y verificacion que /inbox (auth_data +
        verify_auth + frescura del ts). Devuelve False si falta, no es fresco, el
        solicitante no esta anunciado o la firma no cuadra -> el llamante responde
        404 señuelo (indistinguible de un servidor web cualquiera)."""
        peer = self.headers.get("X-Versell-Peer")
        ts = self.headers.get("X-Versell-Ts")
        sig = self.headers.get("X-Versell-Sig")
        if not (peer and ts and sig and fresh(ts)):
            return False
        keys = STORE.keys_of(peer)
        if not keys:
            return False
        return verify_auth(scope, peer, ts, sig, keys["ed25519Pub"])

    def _turn_creds(self, peer):
        """Credenciales TURN efimeras estilo 'coturn REST' (time-limited):
        username = '<expiry_ts>:<peerId>', credential = base64(HMAC-SHA1(secret, username)).
        coturn debe ir con 'use-auth-secret' + 'static-auth-secret=<VERSELL_TURN_SECRET>'.
        Si no hay secreto/urls TURN, se devuelve solo STUN (o vacio): la llamada
        seguira funcionando en la misma LAN / sin CGNAT, pero no por relay."""
        servers = []
        # URLs TURN: las configuradas o, si no hay, un default derivado del propio
        # host de la peticion -> basta con poner el SECRETO en el servicio para que
        # el relay funcione (menos config = menos formas de dejarlo a medias).
        # Orden de preferencia: 443/udp PRIMERO (se confunde con QUIC/HTTP-3 y pasa
        # por cortafuegos que solo dejan "web"), 3478 despues como respaldo clasico.
        host = (self.headers.get("Host") or "").split(":", 1)[0].strip()
        turn_urls = TURN_URLS
        if not turn_urls and TURN_SECRET and host:
            # Relay del PROPIO nodo (TURN embebido) en el puerto del STUN. Si hay un
            # coturn aparte, se listan sus URLs con VERSELL_TURN_URLS y mandan esas.
            turn_urls = ["turn:%s:%d?transport=udp" % (host, STUN_PORT)]
        stun = STUN_URLS
        if not stun:
            if STUN_PORT > 0 and host:
                # STUN PROPIO del nodo (embebido): imprescindible para que la llamada
                # salga de la LAN. Va SIEMPRE, haya o no relay TURN configurado; sin
                # esto la app se quedaba sin ningun servidor ICE y solo conectaba en
                # la misma wifi (fallo reportado por Jorge).
                stun = ["stun:%s:%d" % (host, STUN_PORT)]
            elif turn_urls:
                # STUN implicito en el mismo coturn si no se listo aparte.
                stun = ["stun:" + turn_urls[0].split(":", 1)[1].split("?", 1)[0]] if ":" in turn_urls[0] else []
        if stun:
            servers.append({"urls": stun})
        if TURN_SECRET and turn_urls:
            expiry = int(time.time()) + TURN_TTL
            username = "%d:%s" % (expiry, peer)
            cred = base64.b64encode(
                hmac.new(TURN_SECRET.encode(), username.encode(), hashlib.sha1).digest()
            ).decode()
            servers.append({"urls": turn_urls, "username": username, "credential": cred})
        return {"iceServers": servers, "ttl": TURN_TTL}

    def _decoy_404(self):
        """Respuesta para sondeos en modo DECOY: un 404 indistinguible del de un
        servidor web cualquiera (sin CORS, sin JSON de error, sin nada que delate
        un nodo Versell). Cierra la conexion como el resto de errores."""
        self.close_connection = True
        self._suppress_cors = True
        body = (b"<html>\r\n<head><title>404 Not Found</title></head>\r\n"
                b"<body>\r\n<center><h1>404 Not Found</h1></center>\r\n"
                b"<hr><center>nginx</center>\r\n</body>\r\n</html>\r\n")
        self.send_response(404)
        self.send_header("Content-Type", "text/html")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    # ---- CORS (para Versell Web) ----
    # El cliente web (webapp/) corre en un navegador y este exige CORS para
    # llamar a la API. Abrir el origen es seguro aqui: la API ya es publica por
    # diseño (sin cuentas), la autenticacion va por firma Ed25519 (no cookies)
    # y el contenido son sobres cifrados E2E. El rate limiting sigue aplicando.
    def end_headers(self):
        # El 404-señuelo (modo DECOY) suprime CORS para parecer un servidor web
        # cualquiera; el resto de respuestas llevan CORS (lo exige Versell Web).
        if not getattr(self, "_suppress_cors", False):
            self.send_header("Access-Control-Allow-Origin", "*")
            # Señal para la app de que el nodo esta fortificado (insignia PRO). Solo
            # en respuestas normales: el 404-señuelo (con _suppress_cors) NUNCA la
            # lleva, para no delatarse ante un sondeo.
            if DECOY or STRICT_GET:
                self.send_header("X-Versell-Hardened", "1")
        BaseHTTPRequestHandler.end_headers(self)

    def do_OPTIONS(self):
        # Preflight del navegador (POST con Content-Type: application/json).
        self.send_response(204)
        self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS")
        self.send_header("Access-Control-Allow-Headers", "Content-Type")
        self.send_header("Access-Control-Max-Age", "86400")
        self.send_header("Content-Length", "0")
        self.end_headers()

    # ---- utilidades de respuesta ----
    def _pad_body(self, body):
        want = bool(PAD_ENABLED and self.headers.get("X-Versell-Pad"))
        return pad_json_body(body, want)

    def _json(self, code, obj, extra_headers=None):
        body = self._pad_body(json.dumps(obj).encode("utf-8"))
        self.send_response(code)
        self.send_header("Content-Type", "application/json; charset=utf-8")
        self.send_header("Content-Length", str(len(body)))
        for k, v in (extra_headers or {}).items():
            self.send_header(k, v)
        self.end_headers()
        self.wfile.write(body)

    def _err(self, code, msg):
        # Cerramos la conexion: en varios errores (rate limit, cuerpo demasiado
        # grande) no llegamos a leer el body, y dejarlo en el socket corromperia
        # la siguiente peticion keep-alive. Cerrar es simple y correcto.
        self.close_connection = True
        self._json(code, {"error": msg})

    def _read_body(self, max_bytes):
        """Lee el cuerpo respetando un tope. Devuelve None si excede el limite
        (el llamante responde 413) para no cargar cuerpos enormes en memoria."""
        length = int(self.headers.get("Content-Length", 0) or 0)
        if length < 0 or length > max_bytes:
            return None
        return self.rfile.read(length) if length else b""

    def _read_json(self):
        body = self._read_body(MAX_JSON_BYTES)
        if body is None:
            return None
        return json.loads(body.decode("utf-8"))

    def _rpc_proxy(self, chain):
        """Reenvia una peticion JSON-RPC a Ankr con la clave del entorno (ANKR_KEY), para
        las carteras cripto. La clave NO viaja al cliente y la IP del usuario no llega a
        Ankr. Rate-limited por el limitador general (ya aplicado en do_POST). Sin la env,
        404 -> la app usa sus RPC publicos keyless. Solo reenvia a Ankr y a cadenas de la
        lista blanca; el metodo lo elige el cliente (acotado a la superficie de Ankr)."""
        if not ANKR_KEY or chain not in ANKR_CHAINS:
            return self._decoy_404() if DECOY else self._err(404, "No disponible")
        body = self._read_body(RPC_MAX_BYTES)
        if body is None:
            return self._err(413, "Cuerpo demasiado grande")
        upstream = "https://rpc.ankr.com/%s/%s" % (ANKR_CHAINS[chain], ANKR_KEY)
        try:
            req = urllib.request.Request(
                upstream, data=body,
                headers={"Content-Type": "application/json", "User-Agent": "dicra-node"})
            with urllib.request.urlopen(req, timeout=15) as r:
                out = r.read(RPC_MAX_RESPONSE_BYTES + 1)
        except Exception:
            return self._err(502, "RPC upstream no disponible")
        if len(out) > RPC_MAX_RESPONSE_BYTES:
            return self._err(502, "Respuesta RPC demasiado grande")
        self.send_response(200)
        self.send_header("Content-Type", "application/json; charset=utf-8")
        self.send_header("Content-Length", str(len(out)))
        self.end_headers()
        return self.wfile.write(out)

    def _fetch(self):
        """Modo lector "Navegar usando Versell": descarga y sanea una pagina y la
        devuelve como HTML autocontenido (solo texto+imagenes). SIEMPRE autenticado
        (token firmado scope "fetch") -> nunca es un proxy abierto. Ver _reader_page."""
        if not FETCH_ENABLED:
            return self._decoy_404() if DECOY else self._err(404, "No disponible")
        if not self._get_auth_ok("fetch"):
            return self._decoy_404() if DECOY else self._err(403, "No autorizado")
        body = self._read_body(8192)
        if body is None:
            return self._err(413, "Cuerpo demasiado grande")
        try:
            url = (json.loads(body.decode("utf-8")).get("url") or "").strip()
        except Exception:
            return self._err(400, "JSON invalido")
        try:
            title, html = _reader_page(url)
        except ValueError as e:
            return self._err(400, str(e))
        except Exception:
            return self._err(502, "No se pudo leer la pagina")
        return self._json(200, {"title": title, "html": html})

    def _handle_klipy(self, category, action, parsed):
        """Proxy de busqueda/trending de KLIPY (GIFs/stickers): la clave se queda
        aqui (nunca en el APK ni en el repo publico). SIEMPRE autenticado (a
        diferencia de /price y /news, que solo exigen firma con STRICT_GET) -- ver
        el porque en el comentario de KLIPY_ENABLED. Solo pasa metadatos JSON: los
        bytes de imagen/video los baja el propio movil directo del CDN publico de
        KLIPY (static.klipy.com, sin clave), igual que LinkPreview.kt ya baja
        og:image de cualquier URL de terceros -- el nodo nunca los toca."""
        if not KLIPY_ENABLED or not KLIPY_API_KEY:
            return self._err(503, "GIFs/stickers no disponibles")
        if not self._get_auth_ok("klipy"):
            return self._decoy_404() if DECOY else self._err(403, "No autorizado")

        q = parse_qs(parsed.query)
        query = (q.get("q", [""])[0] or "").strip()[:100]
        if action == "search" and not query:
            return self._err(400, "Falta el termino de busqueda")
        page = max(1, _safe_int(q.get("page", ["1"])[0], 1))
        per_page = max(8, min(50, _safe_int(q.get("per_page", ["24"])[0], 24)))

        cache_key = (category, action, query, page, per_page)
        cached = KLIPY_CACHE.get(cache_key)
        if cached is not None:
            return self._json(200, cached)

        if not KLIPY_LIMITER.allow():
            stale = KLIPY_CACHE.get_stale(cache_key)
            if stale is not None:
                return self._json(200, stale)
            return self._err(429, "Demasiadas busquedas, prueba en un rato")

        qs = "page=%d&per_page=%d" % (page, per_page)
        if action == "search":
            qs += "&q=" + quote(query)
        upstream = "%s/%s/%s/%s?%s" % (KLIPY_BASE, KLIPY_API_KEY, category, action, qs)
        try:
            raw, _ = _news_http_get(upstream, KLIPY_MAX_RESPONSE_BYTES, timeout=15)
        except Exception:
            # NUNCA loguear `upstream`: KLIPY exige la clave EN LA URL, no en cabecera.
            emit_log("klipy: fallo la peticion a %s/%s" % (category, action))
            raw = None
        if raw is None:
            stale = KLIPY_CACHE.get_stale(cache_key)
            if stale is not None:
                return self._json(200, stale)
            return self._err(502, "No se pudo contactar con el proveedor")

        try:
            obj = json.loads(raw.decode("utf-8", "replace"))
        except Exception:
            return self._err(502, "Respuesta invalida del proveedor")
        if obj.get("result") is not True or not isinstance(obj.get("data"), dict):
            stale = KLIPY_CACHE.get_stale(cache_key)
            if stale is not None:
                return self._json(200, stale)
            return self._err(502, "El proveedor devolvio un error")

        page_data = obj["data"]
        KLIPY_CACHE.put(cache_key, page_data)
        return self._json(200, page_data)

    def _handle_audius(self, action, parsed, track_id):
        """Proxy de Audius para la musica de las historias. SIEMPRE autenticado
        (scope "audius"): busquedas y streams gastan cuota y ancho de banda reales.
        `stream` descarga el mp3 ENTERO (Audius responde 302 al CDN; urllib lo
        sigue) y lo sirve con su tam totales — node.py no hace chunked streaming y
        un mp3 de historia cabe en memoria de sobra (tope AUDIUS_STREAM_MAX_BYTES)."""
        if not AUDIUS_ENABLED:
            return self._err(503, "Musica no disponible")
        if not self._get_auth_ok("audius"):
            return self._decoy_404() if DECOY else self._err(403, "No autorizado")
        extra = "&api_key=" + quote(AUDIUS_KEY) if AUDIUS_KEY else ""
        if action == "stream":
            if not AUDIUS_LIMITER.allow():
                return self._err(429, "Demasiadas escuchas, prueba en un rato")
            upstream = "%s/tracks/%s/stream?app_name=%s%s" % (AUDIUS_BASE, quote(track_id), AUDIUS_APP_NAME, extra)
            try:
                raw, _ = _news_http_get(upstream, AUDIUS_STREAM_MAX_BYTES, timeout=30)
            except Exception:
                emit_log("audius: fallo el stream de %s" % track_id)
                raw = None
            if not raw:
                return self._err(502, "No se pudo traer la cancion")
            self.send_response(200)
            self.send_header("Content-Type", "audio/mpeg")
            self.send_header("Content-Length", str(len(raw)))
            self.send_header("Cache-Control", "no-store")
            self.end_headers()
            try:
                self.wfile.write(raw)
            except Exception:
                pass
            return
        q = parse_qs(parsed.query)
        query = (q.get("q", [""])[0] or "").strip()[:100]
        if action == "search" and not query:
            return self._err(400, "Falta el termino de busqueda")
        cache_key = (action, query)
        cached = AUDIUS_CACHE.get(cache_key)
        if cached is not None:
            return self._json(200, cached)
        if not AUDIUS_LIMITER.allow():
            stale = AUDIUS_CACHE.get_stale(cache_key)
            if stale is not None:
                return self._json(200, stale)
            return self._err(429, "Demasiadas busquedas, prueba en un rato")
        if action == "search":
            upstream = "%s/tracks/search?query=%s&app_name=%s%s" % (AUDIUS_BASE, quote(query), AUDIUS_APP_NAME, extra)
        else:
            upstream = "%s/tracks/trending?app_name=%s%s" % (AUDIUS_BASE, AUDIUS_APP_NAME, extra)
        # Robustez: Audius se atasca a ratos (timeout / respuesta vacia). Reintentamos la
        # peticion al INSTANTE hasta 3 veces con timeout corto antes de rendirnos, para que
        # una busqueda no vuelva fallida por un hipo transitorio del proveedor. La app hace
        # ademas su propio reintento/failover entre nodos; esto ataja el fallo en el origen.
        raw = None
        for _ in range(3):
            try:
                raw, _ = _news_http_get(upstream, AUDIUS_MAX_RESPONSE_BYTES, timeout=12)
                if raw:
                    break
            except Exception:
                raw = None
        if raw is None:
            emit_log("audius: fallo la peticion %s tras reintentos" % action)
            stale = AUDIUS_CACHE.get_stale(cache_key)
            if stale is not None:
                return self._json(200, stale)
            return self._err(502, "No se pudo contactar con el proveedor")
        tracks = parse_audius_tracks(raw)
        if not tracks:
            # El PROVEEDOR no devolvio nada usable (forma rara / error): cae a cache vieja o 502.
            stale = AUDIUS_CACHE.get_stale(cache_key)
            if stale is not None:
                return self._json(200, stale)
            return self._err(502, "El proveedor devolvio un error")
        # La busqueda de Audius es laxa; recortamos a lo que de verdad coincide con el termino
        # (titulo+artista). Una lista vacia AQUI significa "sin coincidencias", no un error: la
        # app lo pinta como tal (storyMusicEmpty). El trending no lleva termino, no se toca.
        if action == "search":
            tracks = filter_audius_by_query(tracks, query)
        result = {"tracks": tracks}
        AUDIUS_CACHE.put(cache_key, result)
        return self._json(200, result)

    def _handle_translate(self):
        """Proxy de Azure Translator para la auto-traduccion de mensajes. La CLAVE vive en
        el entorno del nodo (VERSELL_AZURE_KEY), NUNCA en la app. SIEMPRE autenticado (token
        firmado scope "translate") porque cada llamada gasta cuota real de Azure. Cachea por
        (from, to, textos) 24 h: el mismo mensaje no se re-traduce ni re-cobra."""
        if not AZURE_TRANSLATE_ENABLED or not AZURE_KEY:
            return self._decoy_404() if DECOY else self._err(404, "No disponible")
        if not self._get_auth_ok("translate"):
            return self._decoy_404() if DECOY else self._err(403, "No autorizado")
        body = self._read_body(64 * 1024)
        if body is None:
            return self._err(413, "Cuerpo demasiado grande")
        try:
            req = json.loads(body.decode("utf-8"))
            to = str(req.get("to") or "").strip()
            frm = str(req.get("from") or "").strip()
            texts = req.get("texts")
        except Exception:
            return self._err(400, "JSON invalido")
        if not _LANG_RE.match(to) or (frm and not _LANG_RE.match(frm)):
            return self._err(400, "Idioma invalido")
        if not isinstance(texts, list) or not texts or len(texts) > AZURE_MAX_TEXTS:
            return self._err(400, "Textos invalidos")
        texts = [str(t) for t in texts]
        if sum(len(t) for t in texts) > AZURE_MAX_CHARS:
            return self._err(413, "Texto demasiado largo")
        cache_key = (frm, to, tuple(texts))
        cached = AZURE_CACHE.get(cache_key)
        if cached is not None:
            return self._json(200, cached)
        if not AZURE_LIMITER.allow():
            stale = AZURE_CACHE.get_stale(cache_key)
            if stale is not None:
                return self._json(200, stale)
            return self._err(429, "Demasiadas traducciones, prueba en un rato")
        qs = "api-version=3.0&to=" + quote(to)
        if frm:
            qs += "&from=" + quote(frm)
        payload = json.dumps([{"Text": t} for t in texts]).encode("utf-8")
        try:
            r = urllib.request.Request("%s/translate?%s" % (AZURE_URL, qs), data=payload, method="POST", headers={
                "Ocp-Apim-Subscription-Key": AZURE_KEY,
                "Ocp-Apim-Subscription-Region": AZURE_REGION,
                "Content-Type": "application/json; charset=utf-8",
                "User-Agent": "dicra-node",
            })
            with urllib.request.urlopen(r, timeout=15) as resp:
                raw = resp.read(512 * 1024)
        except Exception:
            emit_log("translate: fallo el proveedor")
            return self._err(502, "No se pudo traducir")
        parsed = parse_azure_translations(raw, len(texts))
        if parsed is None:
            return self._err(502, "El proveedor devolvio un error")
        result = {"translations": parsed[0], "detected": parsed[1]}
        AZURE_CACHE.put(cache_key, result)
        return self._json(200, result)

    # ---- GET ----
    def do_GET(self):
        parsed = urlparse(self.path)
        path = parsed.path

        if self._ratelimited():
            return

        if self.headers.get("Upgrade", "").lower() == "websocket":
            return self._handle_ws(parsed)

        if path == "/":
            # En modo DECOY, un sondeo sin la clave de admin ve un 404 generico
            # (no la cabecera de salud que delataria el nodo). El operador la ve
            # presentando X-Admin-Key / ?key=. Por defecto (DECOY off), salud normal.
            if DECOY and not self._has_admin_key():
                return self._decoy_404()
            return self._json(200, {"service": "versell-rendezvous", "version": NODE_VERSION,
                                    "backend": VERIFY_BACKEND, "news": NEWS_ENABLED})
        if path == "/admin/stats":
            # Salud AGREGADA para la herramienta de operador (DicraAdmin): cuantas apps
            # hay conectadas ahora mismo y cuantos sobres esperan entrega.
            #
            # ⚠️ Numeros y nada mas: cuantas, NUNCA cuales. Aqui no se devuelve un peerId,
            # ni una IP, ni un buzon concreto — eso volveria a convertir al nodo en algo
            # que puede senalar a una persona, que es justo lo que se desmonto al retirar
            # el panel de operador (§5.5). El maximo por buzon se da porque delata un
            # destinatario atascado sin decir quien es.
            if not self._has_admin_key():
                return self._decoy_404() if DECOY else self._err(403, "No autorizado")
            conexiones, peers_online = WSREG.live()
            with STORE.lock:
                anunciados = len(STORE.directory)
                buzones = sum(1 for v in STORE.mailboxes.values() if v)
                pendientes = sum(len(v) for v in STORE.mailboxes.values())
                mayor = max((len(v) for v in STORE.mailboxes.values()), default=0)
                prekeys = len(STORE.prekeys)
                push = len(STORE.push)
            return self._json(200, {
                "version": NODE_VERSION,
                "conectados": conexiones,        # conexiones WebSocket vivas
                "peersOnline": peers_online,     # apps distintas conectadas ahora
                "anunciados": anunciados,        # peers que alguna vez se anunciaron aqui
                "pendientes": pendientes,        # sobres esperando a que los recojan
                "buzonesConCola": buzones,
                "mayorBuzon": mayor,
                "prekeys": prekeys,
                "push": push,
                "arranque": _STARTED_AT,
            })
        m = re.fullmatch(r"/lookup/(.+)", path)
        if m:
            if STRICT_GET and not self._get_auth_ok("lookup"):
                return self._decoy_404()
            keys = STORE.keys_of(m.group(1))
            return self._json(200, keys) if keys else self._err(404, "Par desconocido")
        m = re.fullmatch(r"/prekey/(.+)", path)
        if m:
            # Bundle de prekey de un peer (X3DH ligero). Mismo trato que /lookup: es
            # material PUBLICO y firmado, gateado por STRICT_GET como el resto de GET.
            # Que el nodo pueda mentir aqui no rompe nada: puede NEGAR el bundle (y el
            # emisor cae al handshake de siempre), pero no puede fabricar uno -- la firma
            # la comprueba el destinatario con la clave Ed25519 que ya tenia del contacto.
            if STRICT_GET and not self._get_auth_ok("prekey"):
                return self._decoy_404()
            b = STORE.prekey_of(m.group(1)) if PREKEY_ENABLED else None
            return self._json(200, b) if b else self._err(404, "Sin prekey")
        m = re.fullmatch(r"/online/(.+)", path)
        if m:
            if STRICT_GET and not self._get_auth_ok("online"):
                return self._decoy_404()
            return self._json(200, {"online": WSREG.online(m.group(1))})
        if path == "/turncreds":
            # Emite credenciales para una llamada. SIEMPRE autenticado (no es un
            # recurso abierto): sin firma valida del solicitante anunciado -> 404
            # señuelo. No filtra si hay TURN configurado a quien no firme.
            if not self._get_auth_ok("turn"):
                return self._decoy_404()
            return self._json(200, self._turn_creds(self.headers.get("X-Versell-Peer", "")))
        m = re.fullmatch(r"/files/(.+)", path)
        if m:
            if STRICT_GET and not self._get_auth_ok("files"):
                return self._decoy_404()
            data = STORE.read_file(m.group(1))
            if data is None:
                return self._err(404, "No encontrado")
            self.send_response(200)
            self.send_header("Content-Type", "application/octet-stream")
            self.send_header("Content-Length", str(len(data)))
            self.end_headers()
            return self.wfile.write(data)
        # Precios cripto cacheados por el nodo: el movil no llama a CoinGecko. Raiz =
        # BTC (retrocompatible) y el resto de monedas en `coins` (>=1.31.0-py).
        if path == "/price":
            if STRICT_GET and not self._get_auth_ok("price"):
                return self._decoy_404()
            data = PRICE.get() if PRICE_ENABLED else None
            if data is None:
                return self._err(404, "Sin precio")
            return self._json(200, data)
        # Noticias cacheadas: titulares por idioma e imagenes.
        m = re.fullmatch(r"/news/([a-z]{2})", path)
        if m:
            if STRICT_GET and not self._get_auth_ok("news"):
                return self._decoy_404()
            data = NEWS.get(m.group(1)) if NEWS_ENABLED else None
            if data is None:
                return self._err(404, "Sin noticias")
            return self._json(200, data)
        m = re.fullmatch(r"/news/img/([a-z]{2})/([0-9a-f]{12})", path)
        if m:
            if STRICT_GET and not self._get_auth_ok("news"):
                return self._decoy_404()
            img = NEWS.get_image(m.group(1), m.group(2)) if NEWS_ENABLED else None
            if img is None:
                return self._err(404, "No encontrado")
            self.send_response(200)
            self.send_header("Content-Type", img[0])
            self.send_header("Content-Length", str(len(img[1])))
            self.send_header("Cache-Control", "max-age=1800")
            self.end_headers()
            return self.wfile.write(img[1])
        # GIFs/stickers (KLIPY): busqueda/trending. Solo metadatos JSON -- las imagenes
        # las baja el movil directo del CDN publico de KLIPY, nunca por aqui.
        m = re.fullmatch(r"/klipy/(gifs|stickers)/(trending|search)", path)
        if m:
            return self._handle_klipy(m.group(1), m.group(2), parsed)
        # Musica de historias (Audius): busqueda/tendencias y el stream proxied.
        m = re.fullmatch(r"/audius/(search|trending)", path)
        if m:
            return self._handle_audius(m.group(1), parsed, None)
        m = re.fullmatch(r"/audius/stream/([A-Za-z0-9]{1,64})", path)
        if m:
            return self._handle_audius("stream", parsed, m.group(1))
        # Promo: puntos + historial de retiros del peer que pregunta. SIEMPRE
        # autenticado (como /turncreds y /klipy/*, nunca abierto). El nodo solo
        # hace contabilidad -- ver PROMO_ENABLED y PromoStore mas arriba.
        if path == "/promo/status":
            if not PROMO_ENABLED:
                return self._err(503, "Promo no disponible")
            if not self._get_auth_ok("promo_status"):
                return self._decoy_404() if DECOY else self._err(403, "No autorizado")
            peer = self.headers.get("X-Versell-Peer")
            points, points_today, withdrawn_today = PROMO.points_of(peer)
            return self._json(200, {
                "points": points, "pointsToday": points_today,
                "dailyCapPoints": PROMO_DAILY_CAP_POINTS,
                "minWithdrawPoints": PROMO_MIN_WITHDRAW_POINTS,
                "maxWithdrawPerDayPoints": PROMO_MAX_WITHDRAW_PER_DAY_POINTS,
                "withdrawnToday": withdrawn_today,
                "withdrawals": PROMO.withdrawals_of(peer),
            })
        # Ruta GET desconocida: en modo DECOY, 404 generico (no delata el nodo).
        if DECOY:
            return self._decoy_404()
        return self._err(404, "Ruta no encontrada")

    # ---- POST ----
    def do_POST(self):
        path = urlparse(self.path).path
        if self._ratelimited():
            return
        try:
            if path == "/fetch":
                return self._fetch()
            if path == "/translate":
                return self._handle_translate()
            m = re.fullmatch(r"/rpc/([a-z]+)", path)
            if m:
                return self._rpc_proxy(m.group(1))
            if path == "/announce":
                a = self._read_json()
                if a is None:
                    return self._err(413, "Cuerpo demasiado grande")
                if not fresh(a.get("ts")) or not verify_announce(a):
                    return self._err(400, "Anuncio invalido")
                STORE.upsert(a)
                STORE.record_meta(a["peerId"])
                return self._json(200, {})
            if path == "/send":
                env = self._read_json()
                if env is None:
                    return self._err(413, "Cuerpo demasiado grande")
                sender = STORE.keys_of(env.get("from"))
                if not sender:
                    return self._err(400, "Emisor no anunciado")
                if not verify_envelope(env, sender["ed25519Pub"]):
                    return self._err(400, "Firma del sobre invalida")
                # Mensaje PROGRAMADO: se retiene hasta `notBefore`. Una fecha ya pasada
                # (o el nodo con la funcion apagada) = entrega normal, sin sorpresas.
                scheduled = 0
                nb = env.get("notBefore")
                if nb is not None:
                    try:
                        nb = int(nb)
                    except (TypeError, ValueError):
                        return self._err(400, "notBefore invalido")
                    now_ms = int(time.time() * 1000)
                    if not SCHED_ENABLED or nb <= now_ms:
                        env.pop("notBefore", None)
                    elif nb > now_ms + SCHED_MAX_MS:
                        return self._err(400, "Programado demasiado lejos")
                    else:
                        env["notBefore"] = nb
                        scheduled = nb
                try:
                    STORE.queue(env)
                except Store.MailboxFull:
                    return self._err(429, "Buzon lleno")
                STORE.record_meta(env.get("from"))
                if not scheduled:
                    WSREG.push(env["to"], json.dumps({"type": "envelope", "envelope": env}))
                    # UnifiedPush: despierta al destinatario si registro endpoint y no
                    # esta conectado por WS (el aviso es un POST constante, sin contenido).
                    _push_wake(env["to"])
                # Cabecera de capacidad: la app la lee para saber si este nodo respeta de
                # verdad la programacion. Un nodo viejo ignora `notBefore` y entrega YA;
                # sin esta señal la app no podria avisar de que no se ha programado nada.
                return self._json(200, {}, {"X-Versell-Sched": "1"} if SCHED_ENABLED else None)
            if path == "/inbox":
                req = self._read_json()
                if req is None:
                    return self._err(413, "Cuerpo demasiado grande")
                keys = STORE.keys_of(req.get("peerId"))
                if not keys or not fresh(req.get("ts")) or not verify_auth(
                        "inbox", req["peerId"], req["ts"], req["sig"], keys["ed25519Pub"]):
                    return self._err(401, "Autenticacion invalida")
                STORE.record_meta(req["peerId"])
                return self._json(200, STORE.inbox(req["peerId"]))
            if path == "/ack":
                req = self._read_json()
                if req is None:
                    return self._err(413, "Cuerpo demasiado grande")
                keys = STORE.keys_of(req.get("peerId"))
                if not keys or not fresh(req.get("ts")) or not verify_auth(
                        "inbox", req["peerId"], req["ts"], req["sig"], keys["ed25519Pub"]):
                    return self._err(401, "Autenticacion invalida")
                STORE.ack(req["peerId"], set(req.get("ids", [])))
                return self._json(200, {})
            if path == "/push/register":
                # UnifiedPush: la app registra SU endpoint de aviso (firmado, scope
                # "push"). Validacion dura: tope de longitud, https (salvo opt-in
                # http para autoalojados) y SSRF-guard (nunca IPs privadas/loopback).
                req = self._read_json()
                if req is None:
                    return self._err(413, "Cuerpo demasiado grande")
                if not PUSH_ENABLED:
                    return self._err(404, "No encontrado")
                keys = STORE.keys_of(req.get("peerId"))
                if not keys or not fresh(req.get("ts")) or not verify_auth(
                        "push", req["peerId"], req["ts"], req["sig"], keys["ed25519Pub"]):
                    return self._err(401, "Autenticacion invalida")
                ep = (req.get("endpoint") or "").strip()
                if (not ep or len(ep) > PUSH_MAX_URL or not _url_ok(ep)
                        or not (ep.startswith("https://") or (PUSH_ALLOW_HTTP and ep.startswith("http://")))):
                    return self._err(400, "Endpoint invalido")
                STORE.set_push(req["peerId"], ep)
                return self._json(200, {})
            if path == "/promo/sweep":
                # RESCATE DE FONDOS: vacia la cartera de premios de este nodo hacia la
                # cartera maestra (la direccion esta FIJA en usb/promo_sweep.py, no se
                # puede elegir por la red — un rescate con destino parametrizable es un
                # robo esperando a que alguien entre).
                #
                # DOS LLAVES, y esto es lo importante: la clave de operador YA NO BASTA,
                # porque desde 2026-08-13 vive dentro del APK de DicraAdmin y quien tenga
                # ese APK la tiene. Hace falta ademas `VERSELL_SWEEP_PASS`, que no esta en
                # ningun APK ni en el repo y se teclea en el momento del rescate.
                #
                # APAGADO POR DEFECTO: sin `VERSELL_SWEEP_PASS` en el entorno del nodo,
                # esta ruta no existe (404). Un endpoint capaz de mover dinero no debe
                # estar disponible "por si acaso": se arma a mano cuando se quiere.
                if not SWEEP_PASS:
                    return self._err(404, "No encontrado")
                if not self._has_admin_key():
                    return self._decoy_404() if DECOY else self._err(403, "No autorizado")
                dada = self.headers.get("X-Sweep-Pass") or ""
                if not dada or not _consteq(dada, SWEEP_PASS):
                    return self._err(403, "No autorizado")
                script = os.path.join(os.path.dirname(os.path.abspath(__file__)), "promo_sweep.py")
                if not os.path.isfile(script):
                    return self._err(500, "Falta promo_sweep.py en el nodo")
                # Se ejecuta con el python del entorno de pagos (el que tiene bdkpython).
                # El nodo NO lee la clave: el script la coge del entorno del servicio.
                cmd = [SWEEP_PYTHON, script]
                if not SWEEP_DRYRUN:
                    cmd.append("--si-vaciar")
                try:
                    res = subprocess.run(
                        cmd, capture_output=True, text=True, timeout=SWEEP_TIMEOUT_S,
                    )
                    salida = (res.stdout or "") + (res.stderr or "")
                except Exception as e:
                    return self._err(500, "No se pudo lanzar el rescate: %s" % e)
                txid = None
                m2 = re.search(r"txid=([0-9a-fA-F]{64})", salida)
                if m2:
                    txid = m2.group(1)
                # La salida se devuelve TAL CUAL para que el operador vea el motivo real
                # cuando no hay txid (saldo cero, Electrum caido, polvo...). El script no
                # imprime la clave en ningun caso.
                return self._json(200, {
                    "ok": txid is not None,
                    "txid": txid,
                    "ensayo": SWEEP_DRYRUN,
                    "salida": salida[-2000:],
                })
            if path == "/prekey":
                # X3DH ligero: el dueño PUBLICA su bundle de prekey firmado. No hay
                # auth_data aparte a proposito -- la firma del propio bundle (framing
                # `prekey_data`, que incluye peerId y ts) autentica al publicador Y al
                # contenido, y es la MISMA que verificara el destinatario. Dos firmas
                # para una cosa serian dos sitios donde equivocarse.
                req = self._read_json()
                if req is None:
                    return self._err(413, "Cuerpo demasiado grande")
                if not PREKEY_ENABLED:
                    return self._err(404, "No encontrado")
                keys = STORE.keys_of(req.get("peerId"))
                if not keys:
                    return self._err(401, "Autenticacion invalida")
                pk_id = str(req.get("pkId") or "")
                pk = str(req.get("pk") or "")
                raw = b""
                if pk and len(pk) <= PREKEY_MAX_B64:
                    try:
                        raw = b64d(pk)
                    except Exception:
                        raw = b""
                if not pk_id or len(pk_id) > PREKEY_MAX_ID or len(raw) != 32:
                    return self._err(400, "Prekey invalida")
                bundle = {"peerId": req["peerId"], "pkId": pk_id, "pk": pk,
                          "ts": req.get("ts"), "sig": req.get("sig") or ""}
                # Frescura al PUBLICAR (no al servir): impide reponer un bundle viejo
                # que su dueño ya rotó. Al leerlo, el destinatario no exige frescura --
                # un bundle legitimo puede llevar semanas ahi.
                if not fresh(bundle["ts"]) or not verify_prekey(bundle, keys["ed25519Pub"]):
                    return self._err(401, "Autenticacion invalida")
                bundle["ts"] = int(bundle["ts"])
                STORE.set_prekey(req["peerId"], bundle)
                return self._json(200, {})
            if path == "/push/unregister":
                req = self._read_json()
                if req is None:
                    return self._err(413, "Cuerpo demasiado grande")
                keys = STORE.keys_of(req.get("peerId"))
                if not keys or not fresh(req.get("ts")) or not verify_auth(
                        "push", req["peerId"], req["ts"], req["sig"], keys["ed25519Pub"]):
                    return self._err(401, "Autenticacion invalida")
                STORE.clear_push(req["peerId"])
                return self._json(200, {})
            if path == "/files":
                data = self._read_body(MAX_FILE_BYTES)
                if data is None:
                    return self._err(413, "Archivo demasiado grande")
                return self._json(200, {"fileId": STORE.save_file(data)})
            if path == "/promo/claim":
                # Puntos por una accion reclamada por el propio cliente (añadir
                # contacto/crear grupo/mandar mensaje). Firmado por tipo de accion
                # (scope "promo_<accion>") para que el framing no cambie del resto
                # de endpoints -- ver verify_auth. El nodo NO puede comprobar que
                # la accion ocurrio de verdad (todo lo demas es E2E, por diseño);
                # el tope de ritmo + el tope diario son la salvaguarda real, no
                # esta verificacion de firma (que solo evita que alguien reclame
                # puntos EN NOMBRE de otro peerId).
                if not PROMO_ENABLED:
                    return self._err(503, "Promo no disponible")
                req = self._read_json()
                if req is None:
                    return self._err(413, "Cuerpo demasiado grande")
                action = req.get("action")
                if action not in PROMO_POINTS:
                    # 200 con motivo (no 400): el cliente guarda los claims en una
                    # cola persistente y solo los borra cuando el nodo se pronuncia.
                    # Con un 400 seco no podria distinguir "esto no lo sabras
                    # conceder nunca" de "fallo de red" y lo reintentaria para
                    # siempre, atascando la cola detras de el.
                    return self._json(200, {"accepted": False, "points": 0,
                                            "reason": "unknown_action"})
                keys = STORE.keys_of(req.get("peerId"))
                if not keys or not fresh(req.get("ts")) or not verify_auth(
                        "promo_" + action, req["peerId"], req["ts"], req["sig"], keys["ed25519Pub"]):
                    return self._err(401, "Autenticacion invalida")
                ok, points, reason = PROMO.claim(req["peerId"], action)
                return self._json(200, {"accepted": ok, "points": points, "reason": reason})
            if path == "/promo/withdraw":
                # Registra un retiro si el ledger DEL SERVIDOR (no lo que diga el
                # cliente) lo permite y el tope global diario tiene hueco. Nunca
                # envia bitcoin aqui mismo -- solo encola para promo_pay.py.
                if not PROMO_ENABLED:
                    return self._err(503, "Promo no disponible")
                req = self._read_json()
                if req is None:
                    return self._err(413, "Cuerpo demasiado grande")
                address = (req.get("address") or "").strip()
                keys = STORE.keys_of(req.get("peerId"))
                if not keys or not address or not fresh(req.get("ts")) or not verify_auth(
                        "promo_withdraw", req["peerId"], req["ts"], req["sig"], keys["ed25519Pub"]):
                    return self._err(401, "Autenticacion invalida")
                if len(address) > 128:
                    return self._err(400, "Direccion invalida")
                price = PRICE.get() if PRICE_ENABLED else None
                if not price or not price.get("eur"):
                    return self._err(503, "Precio de bitcoin no disponible, prueba en un rato")
                wid, reason, sats = PROMO.queue_withdrawal(req["peerId"], address, price["eur"])
                if wid is None:
                    return self._json(200, {"status": reason})
                return self._json(200, {"status": "queued", "id": wid, "sats": sats})
            return self._err(404, "Ruta no encontrada")
        except ValueError:
            # Cuerpo JSON malformado (json.loads lanza ValueError/JSONDecodeError) u
            # otra entrada invalida: es culpa del CLIENTE -> 400 limpio, no un 500.
            # (Un 500 con el texto de la excepcion, ademas, filtraba detalle interno.)
            return self._err(400, "Peticion invalida")
        except Exception:
            # Fallo inesperado del servidor: 500 GENERICO, sin filtrar el detalle.
            return self._err(500, "Error interno")

    # ---- WebSocket ----
    def _handle_ws(self, parsed):
        key = self.headers.get("Sec-WebSocket-Key")
        if not key:
            return self._err(400, "Falta Sec-WebSocket-Key")
        q = parse_qs(parsed.query)
        peer = (q.get("peerId") or [None])[0]
        ts = (q.get("ts") or [None])[0]
        sig = (q.get("sig") or [None])[0]
        keys = STORE.keys_of(peer) if peer else None
        ok = keys and ts and sig and fresh(ts) and verify_auth("ws", peer, ts, sig, keys["ed25519Pub"])
        if not ok:
            return self._err(401, "No autorizado")
        STORE.record_meta(peer)
        if WSREG.count(peer) >= MAX_WS_PER_PEER:
            return self._err(429, "Demasiadas conexiones")

        accept = base64.b64encode(hashlib.sha1(key.encode() + WS_GUID).digest()).decode()
        self.send_response(101)
        self.send_header("Upgrade", "websocket")
        self.send_header("Connection", "Upgrade")
        self.send_header("Sec-WebSocket-Accept", accept)
        self.end_headers()
        self.close_connection = True

        # Ya en modo WS: quitamos el timeout corto anti-slowloris (una conexion
        # WS legitima puede estar inactiva largo rato escuchando).
        try:
            self.connection.settimeout(None)
        except Exception:
            pass

        conn = WsConn(self.wfile)
        WSREG.add(peer, conn)
        try:
            for env in STORE.inbox(peer):
                conn.send_text(json.dumps({"type": "envelope", "envelope": env}))
            while conn.alive:
                frame = self._read_ws_frame()
                if frame is None:
                    break
                opcode, data = frame
                if opcode == 0x8:      # close
                    break
                elif opcode == 0x9:    # ping -> pong
                    conn.send_pong(data)
                # los frames de datos del cliente se ignoran (envia por REST)
        except Exception:
            pass
        finally:
            conn.alive = False
            WSREG.remove(peer, conn)

    def _read_ws_frame(self):
        hdr = self.rfile.read(2)
        if len(hdr) < 2:
            return None
        b1 = hdr[1]
        opcode = hdr[0] & 0x0F
        masked = b1 & 0x80
        ln = b1 & 0x7F
        if ln == 126:
            ln = struct.unpack(">H", self.rfile.read(2))[0]
        elif ln == 127:
            ln = struct.unpack(">Q", self.rfile.read(8))[0]
        if ln > MAX_WS_FRAME_BYTES:
            return None  # trama abusiva -> cerrar la conexion
        mask = self.rfile.read(4) if masked else b""
        data = self.rfile.read(ln) if ln else b""
        if masked and data:
            data = bytes(data[i] ^ mask[i % 4] for i in range(len(data)))
        return opcode, data


def run_server():
    here = os.path.dirname(os.path.abspath(__file__))
    data_dir = os.environ.get("VERSELL_DATA") or os.path.join(here, "data")
    port = int(os.environ.get("PORT", "8080"))
    bind = os.environ.get("VERSELL_BIND", "0.0.0.0")
    httpd = build_server(port, data_dir, bind)
    emit_log("=====================================================")
    emit_log("  Versell - nodo rendezvous (Python)")
    emit_log("  Escuchando en %s:%d" % (bind, port))
    emit_log("  Datos en: %s" % data_dir)
    emit_log("  Verificacion Ed25519: %s" % VERIFY_BACKEND)
    emit_log("  Limites: %s rpm/IP, %d conex, %d sobres/buzon, files TTL %dh/%dMB" % (
        (RATE_RPM or "sin"), MAX_CONN, MAILBOX_MAX,
        FILES_TTL_S // 3600, FILES_MAX_BYTES // (1024 * 1024)))
    emit_log("  Sin cuentas: solo mueve sobres cifrados E2E.")
    emit_log("  Sin moderacion de contenido: no hay /report ni conversaciones guardadas.")
    emit_log("  Sin panel de operador: no hay contadores por IP, ni nombres, ni vetos.")
    if DECOY:
        emit_log("  Ocultacion: DECOY ACTIVO (sondeo sin auth -> 404 generico; salud en / con la clave de VERSELL_ADMIN_KEY).")
    if STRICT_GET:
        emit_log("  Ocultacion: STRICT_GET ACTIVO (/lookup //online //news //files exigen token firmado; el resto -> 404).")
    emit_log("=====================================================")
    try:
        httpd.serve_forever()
    except KeyboardInterrupt:
        emit_log("Detenido.")

# =====================================================================
#  Autocomprobacion (criptografia + framing + protocolo)
# =====================================================================

def selftest():
    ok = True

    # 0) Parser de Audius (musica de historias): puro, sin red.
    buena = json.dumps({"data": [
        {"id": "abc123", "title": "Cancion", "user": {"name": "Artista"}, "duration": 180},
        {"id": "", "title": "sin id"},
        {"title": "sin id 2"},
        "basura",
        {"id": "x" * 200, "title": "t" * 300, "user": {}, "duration": -5},
    ]}).encode()
    t = parse_audius_tracks(buena)
    if (len(t) == 2 and t[0] == {"id": "abc123", "title": "Cancion", "artist": "Artista", "durS": 180}
            and len(t[1]["id"]) == 64 and len(t[1]["title"]) == 120 and t[1]["durS"] is None
            and t[1]["artist"] == "?"):
        print("[ok] parse_audius_tracks: filtra la basura, capa longitudes y conserva lo valido")
    else:
        print("[FALLO] parse_audius_tracks con respuesta valida"); ok = False
    if parse_audius_tracks(b"no-json") == [] and parse_audius_tracks(b'{"data": 7}') == [] \
            and parse_audius_tracks(b'{}') == []:
        print("[ok] parse_audius_tracks: JSON invalido o formas raras -> lista vacia")
    else:
        print("[FALLO] parse_audius_tracks no tolera basura"); ok = False

    # 0b) Filtro de relevancia de la busqueda de Audius (la queja: salen pistas no buscadas).
    muestra = [
        {"id": "1", "title": "Yesterday", "artist": "The Beatles", "durS": 125},
        {"id": "2", "title": "Beatles Tribute Mix", "artist": "DJ X", "durS": 200},
        {"id": "3", "title": "Cancion cualquiera", "artist": "Otro", "durS": 180},
        {"id": "4", "title": "Rosalia en vivo", "artist": "Rosalía", "durS": 190},
    ]
    f1 = filter_audius_by_query(muestra, "beatles")          # titulo+artista: casan 1 (artista) y 2 (titulo)
    f2 = filter_audius_by_query(muestra, "rosalía")          # acento en el termino -> casa por titulo/artista
    f3 = filter_audius_by_query(muestra, "the beatles")      # dos palabras, ambas exigidas -> solo 1
    f4 = filter_audius_by_query(muestra, "reggaeton perreo") # nada coincide -> lista vacia
    f5 = filter_audius_by_query(muestra, "   ")              # termino vacio -> no filtra
    if ([t["id"] for t in f1] == ["1", "2"] and [t["id"] for t in f2] == ["4"]
            and [t["id"] for t in f3] == ["1"] and f4 == [] and len(f5) == len(muestra)):
        print("[ok] filter_audius_by_query: exige todas las palabras en titulo/artista, sin acentos")
    else:
        print("[FALLO] filter_audius_by_query: filtrado de relevancia incorrecto"); ok = False

    # 0c) Parser de Azure Translator (auto-traduccion): puro, sin red.
    az_good = (b'[{"detectedLanguage":{"language":"fr","score":1.0},'
               b'"translations":[{"text":"Hola mundo","to":"es"}]},'
               b'{"translations":[{"text":"Adios","to":"es"}]}]')
    az = parse_azure_translations(az_good, 2)
    if az == (["Hola mundo", "Adios"], "fr"):
        print("[ok] parse_azure_translations: extrae textos en orden y el idioma detectado")
    else:
        print("[FALLO] parse_azure_translations con respuesta valida (%s)" % (az,)); ok = False
    # Formas rotas o con distinto numero de elementos -> None (no se sirve basura).
    if (parse_azure_translations(b'no-json', 1) is None
            and parse_azure_translations(az_good, 3) is None
            and parse_azure_translations(b'[{"translations":[]}]', 1) is None
            and parse_azure_translations(b'[{}]', 1) is None):
        print("[ok] parse_azure_translations: JSON roto / conteo distinto / sin traduccion -> None")
    else:
        print("[FALLO] parse_azure_translations no rechaza formas invalidas"); ok = False
    # El regex de idioma acepta lo valido (es, zh-Hans, pt-BR) y rechaza inyecciones.
    if (_LANG_RE.match("es") and _LANG_RE.match("zh-Hans") and _LANG_RE.match("pt-BR")
            and not _LANG_RE.match("es&x=1") and not _LANG_RE.match("../etc") and not _LANG_RE.match("")):
        print("[ok] Traduccion: el codigo de idioma se valida (no cuela inyeccion en la URL)")
    else:
        print("[FALLO] Traduccion: el regex de idioma es incorrecto"); ok = False

    # 1) Vector RFC 8032 (Test 1): valida la verificacion Ed25519 contra algo externo.
    pub = bytes.fromhex("d75a980182b10ab7d54bfed3c964073a0ee172f3daa62325af021a68f707511a")
    sig = bytes.fromhex(
        "e5564300c360ac729086e2cc806e828a84877f1eb8e5d974d873e065224901555fb8821590a33bacc61e39701cf9b46bd25bf5f0595bbe24655141438e7a100b")
    if VERIFY(pub, sig, b""):
        print("[ok] Ed25519 verifica el vector RFC 8032 (backend: %s)" % VERIFY_BACKEND)
    else:
        print("[FALLO] Ed25519 no verifica el vector RFC 8032"); ok = False
    if not _PURE_VERIFY(pub, sig, b""):
        print("[FALLO] backend python-puro no verifica el vector RFC 8032"); ok = False
    else:
        print("[ok] backend python-puro tambien verifica el vector RFC 8032")

    # 2) Roundtrip de protocolo: firma con Ed25519 puro y verifica el framing.
    seed = hashlib.sha256(b"versell-selftest-seed").digest()  # 32 bytes
    edpub = _PURE_PUBKEY(seed)
    edpub_b64 = base64.b64encode(edpub).decode()
    xpub_b64 = base64.b64encode(hashlib.sha256(b"fake-x25519").digest()).decode()
    peer = peer_id_for(xpub_b64, edpub_b64)

    ts = int(time.time() * 1000)
    a_sig = base64.b64encode(_PURE_SIGN(seed, announce_data(peer, xpub_b64, edpub_b64, ts))).decode()
    announce = {"peerId": peer, "x25519Pub": xpub_b64, "ed25519Pub": edpub_b64, "ts": ts, "sig": a_sig}
    if fresh(ts) and verify_announce(announce):
        print("[ok] announce firmado se verifica")
    else:
        print("[FALLO] announce no se verifica"); ok = False

    # Sobre (ephPub/nonce/ct son opacos para el nodo; solo cuenta la firma).
    frm = peer
    to = peer
    eph = os.urandom(32); nonce = os.urandom(12); ct = os.urandom(40)
    e_sig = base64.b64encode(_PURE_SIGN(seed, sig_data(frm, to, eph, nonce, ct))).decode()
    env = {"id": "x", "from": frm, "to": to, "sentAt": ts,
           "ephPub": base64.b64encode(eph).decode(), "nonce": base64.b64encode(nonce).decode(),
           "ct": base64.b64encode(ct).decode(), "sig": e_sig}
    if verify_envelope(env, edpub_b64):
        print("[ok] sobre firmado se verifica")
    else:
        print("[FALLO] sobre no se verifica"); ok = False

    tampered = dict(env); tampered["ct"] = base64.b64encode(os.urandom(40)).decode()
    if not verify_envelope(tampered, edpub_b64):
        print("[ok] sobre manipulado se rechaza")
    else:
        print("[FALLO] sobre manipulado NO se rechaza"); ok = False

    au_sig = base64.b64encode(_PURE_SIGN(seed, auth_data("inbox", peer, ts))).decode()
    if verify_auth("inbox", peer, ts, au_sig, edpub_b64):
        print("[ok] auth de buzon se verifica")
    else:
        print("[FALLO] auth de buzon no se verifica"); ok = False

    # (Aqui se comprobaba la firma de un reporte de moderacion: retirada en 1.16.0-py
    # junto con /report. El framing firmado del sec. 4 —announce, sobre y auth— sigue
    # cubierto arriba, que es lo que de verdad tiene que seguir cuadrando byte a byte
    # entre la app y las dos implementaciones de nodo.)

    # 3) Ledger Promo: puntos, tope diario por persona y tope global (2026-08-06).
    #    SOLO contabilidad -- no toca bitcoin de verdad, ver PromoStore. El tope
    #    de RITMO se sube para este test: aqui se prueba el tope de PUNTOS/dia y
    #    el tope global, no el de ritmo (ese lo prueba el numero de "accepted").
    global PROMO_DAILY_CAP_SATS, PROMO_CLAIM_MAX_PER_HOUR
    import shutil as _shutil
    import tempfile as _tempfile
    _tmp_dir = _tempfile.mkdtemp(prefix="versell-selftest-promo-")
    _saved_cap = PROMO_DAILY_CAP_SATS
    _saved_rate = PROMO_CLAIM_MAX_PER_HOUR
    try:
        PROMO_DAILY_CAP_SATS = 10_000_000  # holgado: aqui se prueba el ledger, no el tope global
        PROMO_CLAIM_MAX_PER_HOUR = {"contact": 1000, "group": 1000, "message": 1000,
                                    "update": 1000, "compensation": 1000}
        p = PromoStore(_tmp_dir)
        _, pts, _ = p.claim("peerA", "contact")
        accepted, pts, _ = p.claim("peerA", "message")
        if accepted and pts == 11:
            print("[ok] Promo: claim suma puntos (10+1=11)")
        else:
            print("[FALLO] Promo: claim no suma bien (%d, esperaba 11)" % pts); ok = False

        for _ in range(400):
            p.claim("peerA", "message")
        _, pts, _ = p.claim("peerA", "message")
        if pts <= PROMO_DAILY_CAP_POINTS:
            print("[ok] Promo: tope diario por persona se respeta (%d <= %d)" % (pts, PROMO_DAILY_CAP_POINTS))
        else:
            print("[FALLO] Promo: tope diario por persona NO se respeta (%d > %d)" % (pts, PROMO_DAILY_CAP_POINTS)); ok = False

        wid, reason, _ = p.queue_withdrawal("peerB", "bc1qtest", 30000.0)
        if wid is None and reason == "not_eligible":
            print("[ok] Promo: retiro por debajo del minimo se rechaza")
        else:
            print("[FALLO] Promo: retiro por debajo del minimo NO se rechazo"); ok = False

        # Justo hasta el tope diario: 37x10 + 5x1 = 375. Con el "todo o nada" del
        # tope (ver mas abajo) un 38 contacto ya NO cabe, asi que los ultimos 5
        # puntos hay que sumarlos con acciones de 1 -- si no, el saldo se queda en
        # 370 y tras retirar 250 quedarian 120, por debajo del minimo de retiro, y
        # el caso que prueba el bloque siguiente (tope de RETIRO diario) no se daria.
        for _ in range(37):
            p.claim("peerB", "contact")
        for _ in range(5):
            p.claim("peerB", "message")
        pts_before = p.points_of("peerB")[0]
        wid, reason, sats = p.queue_withdrawal("peerB", "bc1qtest", 30000.0)
        pts_after = p.points_of("peerB")[0]
        if wid is not None and sats > 0 and pts_after == pts_before - PROMO_MAX_WITHDRAW_PER_DAY_POINTS:
            print("[ok] Promo: retiro elegible paga como mucho el tope diario de retiro (%d sats, quedan %d puntos)"
                  % (sats, pts_after))
        else:
            print("[FALLO] Promo: retiro elegible no se encolo/pago bien (id=%s, motivo=%s, quedan=%d)"
                  % (wid, reason, pts_after)); ok = False

        wid2, reason2, _ = p.queue_withdrawal("peerB", "bc1qtest", 30000.0)
        if wid2 is None and reason2 == "daily_withdraw_capped" and p.points_of("peerB")[0] == pts_after:
            print("[ok] Promo: segundo retiro el mismo dia se rechaza (tope de retiro diario), saldo intacto")
        else:
            print("[FALLO] Promo: el tope de retiro diario por persona no se respeto (id=%s, motivo=%s)"
                  % (wid2, reason2)); ok = False

        # TODO-O-NADA en el tope diario (1.0.165): con el reparto parcial de antes,
        # una accion de 10 puntos con solo 5 de hueco daba 5 y tiraba los otros 5.
        for _ in range(37):
            p.claim("peerD", "contact")   # 370 de 375
        acc_big, pts_big, why_big = p.claim("peerD", "contact")   # quiere 10, hay 5
        acc_small, pts_small, _ = p.claim("peerD", "message")     # quiere 1, hay 5
        if not acc_big and why_big == "daily_cap" and pts_big == 370 and acc_small and pts_small == 371:
            print("[ok] Promo: una accion que no cabe entera en el tope diario se rechaza ENTERA (no se parte)")
        else:
            print("[FALLO] Promo: el tope diario sigue troceando acciones (acc=%s motivo=%s pts=%d)"
                  % (acc_big, why_big, pts_big)); ok = False

        # Compensacion: una sola vez por peer, y POR ENCIMA del tope diario (peerA
        # ya lo tiene agotado justo arriba) -- si compitiera por el hueco diario,
        # desplazaria puntos que el usuario si ha ganado de verdad hoy.
        pts_pre = p.points_of("peerA")[0]
        acc1, pts1, why1 = p.claim("peerA", "compensation")
        acc2, pts2, why2 = p.claim("peerA", "compensation")
        if acc1 and pts1 == pts_pre + 120 and why1 == "ok" and not acc2 and why2 == "already" and pts2 == pts1:
            print("[ok] Promo: compensacion de 120 se da UNA vez, ignora el tope diario y no se repite")
        else:
            print("[FALLO] Promo: la compensacion unica no se comporta (acc1=%s pts1=%d why2=%s pts2=%d)"
                  % (acc1, pts1, why2, pts2)); ok = False

        acc_u, _, why_u = p.claim("peerE", "update")
        acc_x, _, why_x = p.claim("peerE", "accion_que_no_existe")
        if acc_u and why_u == "ok" and not acc_x and why_x == "unknown_action":
            print("[ok] Promo: 'update' (+8) se concede y una accion desconocida se marca no-reintentable")
        else:
            print("[FALLO] Promo: update/accion desconocida mal (u=%s/%s x=%s/%s)"
                  % (acc_u, why_u, acc_x, why_x)); ok = False

        # SIN tope global (0 = apagado, el defecto desde 2026-08-08): un retiro que
        # antes se habria rechazado por el tope global ahora tiene que pasar.
        PROMO_DAILY_CAP_SATS = 0
        for _ in range(30):
            p.claim("peerF", "contact")
        wid, reason, sats_f = p.queue_withdrawal("peerF", "bc1qtest", 30000.0)
        if wid is not None and sats_f > 0:
            print("[ok] Promo: sin tope global el retiro pasa (mandan solo los limites por persona)")
        else:
            print("[FALLO] Promo: sin tope global el retiro NO deberia rechazarse (motivo=%s)" % reason); ok = False

        # Y si se REACTIVA poniendo un valor > 0, vuelve a frenar: el mecanismo sigue
        # entero, solo esta apagado por defecto.
        PROMO_DAILY_CAP_SATS = 1  # tope global minusculo: el siguiente retiro debe rechazarse
        for _ in range(30):
            p.claim("peerC", "contact")
        wid, reason, _ = p.queue_withdrawal("peerC", "bc1qtest", 30000.0)
        if wid is None and reason == "capped" and p.points_of("peerC")[0] == 300:
            print("[ok] Promo: reactivar el tope global vuelve a frenar el pago SIN tocar los puntos")
        else:
            print("[FALLO] Promo: el tope global reactivado no freno el pago (id=%s, motivo=%s)" % (wid, reason)); ok = False
    finally:
        PROMO_DAILY_CAP_SATS = _saved_cap
        PROMO_CLAIM_MAX_PER_HOUR = _saved_rate
        _shutil.rmtree(_tmp_dir, ignore_errors=True)

    # --- Relleno de respuestas (transportes conmutables, modo sigiloso) ---
    # want_pad=False no toca nada; True cuantiza al bucket con `_pad` de longitud
    # exacta y JSON valido; una lista o un cuerpo ya-grande se dejan intactos.
    import json as _json_mod
    small = b'{"online":true}'
    if pad_json_body(small, False) == small:
        print("[ok] Pad: sin cabecera (want_pad=False) el cuerpo no se toca")
    else:
        print("[FALLO] Pad: sin want_pad no deberia cambiar el cuerpo"); ok = False
    padded = pad_json_body(small, True)
    parsed = _json_mod.loads(padded)
    if len(padded) == 512 and parsed.get("online") is True and "_pad" in parsed:
        print("[ok] Pad: cuerpo pequeño se cuantiza a 512 B, sigue siendo JSON valido y conserva los datos")
    else:
        print("[FALLO] Pad: cuantizacion mal (len=%d, keys=%s)" % (len(padded), list(parsed.keys()))); ok = False
    empty = pad_json_body(b"{}", True)
    if len(empty) == 512 and _json_mod.loads(empty).get("_pad"):
        print("[ok] Pad: el objeto vacio {} se rellena sin coma sobrante")
    else:
        print("[FALLO] Pad: objeto vacio mal (%r)" % empty[:40]); ok = False
    lst = b'[1,2,3]'
    huge = b'{"x":"' + b"y" * 200000 + b'"}'
    if pad_json_body(lst, True) == lst and pad_json_body(huge, True) == huge:
        print("[ok] Pad: una lista y un cuerpo mayor que el ultimo bucket se dejan intactos")
    else:
        print("[FALLO] Pad: lista/cuerpo-grande no deberian tocarse"); ok = False

    # --- Prekeys X3DH ligero (1.32.0-py) ---
    # Lo unico que el nodo tiene que hacer bien aqui: aceptar un bundle firmado por su
    # dueño y RECHAZAR cualquier otra cosa. Si un nodo pudiera colar una prekey propia,
    # se pondria en medio del primer mensaje de cada conversacion nueva.
    pk_priv = os.urandom(32)
    pk_pub_b64 = base64.b64encode(hashlib.sha256(pk_priv).digest()).decode()  # opaca para el nodo
    pk_ts = int(time.time() * 1000)
    good = {"peerId": peer, "pkId": "pk1", "pk": pk_pub_b64, "ts": pk_ts}
    good["sig"] = base64.b64encode(
        _PURE_SIGN(seed, prekey_data(peer, "pk1", pk_pub_b64, pk_ts))).decode()
    if verify_prekey(good, edpub_b64):
        print("[ok] Prekey: un bundle firmado por su dueño se verifica")
    else:
        print("[FALLO] Prekey: el bundle legitimo NO se verifica"); ok = False
    # Suplantacion: otra identidad firma un bundle a nombre de `peer`. Es EXACTAMENTE
    # lo que intentaria un nodo malicioso para hacer MITM del primer mensaje.
    other_seed = hashlib.sha256(b"versell-selftest-mitm").digest()
    forged = dict(good)
    forged["sig"] = base64.b64encode(
        _PURE_SIGN(other_seed, prekey_data(peer, "pk1", pk_pub_b64, pk_ts))).decode()
    if not verify_prekey(forged, edpub_b64):
        print("[ok] Prekey: un bundle firmado por OTRO se rechaza (no hay MITM del nodo)")
    else:
        print("[FALLO] Prekey: se acepto un bundle firmado por otra identidad"); ok = False
    # Manipulacion: cambiar la clave publica conservando la firma original.
    swapped = dict(good)
    swapped["pk"] = base64.b64encode(os.urandom(32)).decode()
    if not verify_prekey(swapped, edpub_b64):
        print("[ok] Prekey: cambiar la clave publica invalida la firma")
    else:
        print("[FALLO] Prekey: la firma no cubre la clave publica"); ok = False
    # El framing tiene que ser el del sec. 4 (campos con prefijo de longitud), no una
    # concatenacion suelta: si no, dos bundles distintos podrian firmar lo mismo.
    if (prekey_data("a", "bc", "d", 1) != prekey_data("ab", "c", "d", 1)
            and prekey_data(peer, "pk1", pk_pub_b64, pk_ts).startswith(struct.pack(">I", 6) + b"prekey")):
        print("[ok] Prekey: framing con prefijo de longitud, sin ambiguedad entre campos")
    else:
        print("[FALLO] Prekey: el framing es ambiguo"); ok = False
    # Caducidad: un bundle viejo se purga (no es inseguro, pero su dueño ya no esta).
    _pk_tmp = _tempfile.mkdtemp(prefix="versell-selftest-prekey-")
    try:
        st = Store(_pk_tmp)
        st.set_prekey(peer, dict(good))
        st.set_prekey("viejo", {"peerId": "viejo", "pkId": "x", "pk": pk_pub_b64,
                                "ts": pk_ts - (PREKEY_TTL_DAYS + 1) * 86400 * 1000, "sig": ""})
        gone = st.purge_prekeys()
        if gone == 1 and st.prekey_of(peer) and not st.prekey_of("viejo"):
            print("[ok] Prekey: se purga la caducada y se conserva la vigente")
        else:
            print("[FALLO] Prekey: la purga por caducidad no se comporta (purgadas=%d)" % gone); ok = False
        if st.purge_peer(peer) and not st.prekey_of(peer):
            print("[ok] Prekey: borrar un peer se lleva tambien su prekey")
        else:
            print("[FALLO] Prekey: purge_peer dejo la prekey atras"); ok = False
    finally:
        _shutil.rmtree(_pk_tmp, ignore_errors=True)

    # --- Buzones en fichero propio (1.39.0-py) ---
    # Lo que importa aqui: (1) un /send escribe SOLO el buzon del destinatario, que es
    # todo el motivo del cambio; (2) los sobres sobreviven a un reinicio; (3) la
    # MIGRACION desde un state.json con los buzones dentro no pierde ni un sobre (habia
    # 11.107 vivos en produccion al hacer el cambio); (4) un peerId manipulado no puede
    # escribir fuera de mail/.
    _mb_tmp = _tempfile.mkdtemp(prefix="versell-selftest-mail-")
    try:
        p1, p2 = "PeerUno_A-1", "PeerDos_B-2"
        st = Store(_mb_tmp)
        st.queue({"id": "e1", "to": p1, "from": p2, "ct": "x"})
        st.queue({"id": "e2", "to": p1, "from": p2, "ct": "y"})
        st.queue({"id": "e3", "to": p2, "from": p1, "ct": "z"})
        box_files = sorted(n for n in os.listdir(os.path.join(_mb_tmp, "mail")) if n.endswith(".json"))
        # state.json puede ni existir aun: encolar sobres ya NO lo toca (ese es el cambio).
        _sf = os.path.join(_mb_tmp, "state.json")
        st_raw = json.load(open(_sf, encoding="utf-8")) if os.path.exists(_sf) else {}
        if box_files == sorted([p1 + ".json", p2 + ".json"]) and "mailboxes" not in st_raw:
            print("[ok] Buzones: un fichero por peer y state.json ya NO lleva los sobres")
        else:
            print("[FALLO] Buzones: reparto en ficheros incorrecto (%s)" % box_files); ok = False
        # Persistencia: un Store nuevo sobre el mismo directorio ve los mismos sobres.
        st2 = Store(_mb_tmp)
        if len(st2.inbox(p1)) == 2 and len(st2.inbox(p2)) == 1:
            print("[ok] Buzones: los sobres sobreviven al reinicio del nodo")
        else:
            print("[FALLO] Buzones: no se releen del disco"); ok = False
        # Ack: se vacia el buzon y su fichero desaparece (no queda un `[]` huerfano).
        st2.ack(p2, ["e3"])
        if not os.path.exists(os.path.join(_mb_tmp, "mail", p2 + ".json")) and st2.inbox(p2) == []:
            print("[ok] Buzones: al vaciarse un buzon se borra su fichero")
        else:
            print("[FALLO] Buzones: el fichero del buzon vacio sigue ahi"); ok = False
        # Path traversal: un peerId con separadores no puede tocar disco.
        if (st2._box_path("../../etc/passwd") is None and st2._box_path("a/b") is None
                and st2._box_path("") is None and st2._box_path("Bien_A-1") is not None):
            print("[ok] Buzones: un peerId manipulado no escribe fuera de mail/")
        else:
            print("[FALLO] Buzones: _box_path acepta rutas peligrosas"); ok = False
        # Cola circular (1.40.0-py): con el buzon a tope entra el nuevo y sale el VIEJO.
        # Antes se rechazaba con 429 y el emisor reintentaba cada 20 s durante 14 dias:
        # en produccion eso era el 98,5% de los /send rechazados en bucle.
        p3 = "PeerTres_C-3"
        st3 = Store(_mb_tmp)
        for i in range(MAILBOX_MAX + 25):
            st3.queue({"id": "r%d" % i, "to": p3, "from": p1, "ct": "x"})
        box3 = st3.inbox(p3)
        ids3 = [e["id"] for e in box3]
        if (len(box3) == MAILBOX_MAX and ids3[-1] == "r%d" % (MAILBOX_MAX + 24)
                and ids3[0] == "r25" and "r0" not in ids3):
            print("[ok] Buzones: buzon lleno = cola circular (entra el nuevo, cae el mas viejo)")
        else:
            print("[FALLO] Buzones: la cola circular no conserva los ultimos %d (n=%d)"
                  % (MAILBOX_MAX, len(box3))); ok = False
        # Y lo importante de cara al bug: encolar con el buzon lleno NO puede fallar.
        try:
            st3.queue({"id": "rZ", "to": p3, "from": p1, "ct": "x"})
            if [e["id"] for e in st3.inbox(p3)][-1] == "rZ":
                print("[ok] Buzones: con el buzon lleno el envio SIGUE aceptandose (sin 429)")
            else:
                print("[FALLO] Buzones: el sobre nuevo no entro en el buzon lleno"); ok = False
        except Exception as _e:
            print("[FALLO] Buzones: encolar en buzon lleno lanzo %r" % _e); ok = False
    finally:
        _shutil.rmtree(_mb_tmp, ignore_errors=True)

    # Migracion desde el formato viejo (buzones DENTRO de state.json).
    _mg_tmp = _tempfile.mkdtemp(prefix="versell-selftest-migra-")
    try:
        pa, pb = "Viejo_A-1", "Viejo_B-2"
        with open(os.path.join(_mg_tmp, "state.json"), "w", encoding="utf-8") as f:
            json.dump({"directory": {pa: {"peerId": pa}}, "meta": {}, "push": {}, "prekeys": {},
                       "mailboxes": {pa: [{"id": "v1", "to": pa, "ct": "1"},
                                          {"id": "v2", "to": pa, "ct": "2"}],
                                     pb: [{"id": "v3", "to": pb, "ct": "3"}]}}, f)
        stm = Store(_mg_tmp)
        st_raw = json.load(open(os.path.join(_mg_tmp, "state.json"), encoding="utf-8"))
        if (len(stm.inbox(pa)) == 2 and len(stm.inbox(pb)) == 1
                and "mailboxes" not in st_raw
                and os.path.exists(os.path.join(_mg_tmp, "mail", pa + ".json"))):
            print("[ok] Buzones: migracion del state.json viejo sin perder sobres")
        else:
            print("[FALLO] Buzones: la migracion pierde sobres o deja el formato viejo"); ok = False
        # Y una segunda pasada no duplica ni resucita nada.
        stm2 = Store(_mg_tmp)
        if len(stm2.inbox(pa)) == 2 and len(stm2.inbox(pb)) == 1:
            print("[ok] Buzones: re-arrancar tras migrar no duplica sobres")
        else:
            print("[FALLO] Buzones: la segunda carga duplica o pierde sobres"); ok = False
    finally:
        _shutil.rmtree(_mg_tmp, ignore_errors=True)

    # --- Precios de TODAS las criptos (GET /price, 1.31.0-py) ---
    # Lo que importa aqui: la raiz sigue siendo BTC (apps viejas y el retiro de Promo
    # leen price["eur"]), las demas monedas salen en `coins`, y una respuesta rota no
    # produce un cuerpo a medias sino None (el cache se queda con el valor anterior).
    sample = (b'{"bitcoin":{"eur":55588,"usd":64144,"cny":432672},'
              b'"ethereum":{"eur":1656.87,"usd":1911.91,"cny":12896.39},'
              b'"polygon-ecosystem-token":{"eur":0.065284,"usd":0.075333,"cny":0.508142},'
              b'"solana":{"eur":66.45,"usd":76.67,"cny":517.2},'
              b'"dogecoin":{"eur":0.062312,"usd":0.071903,"cny":0.485009}}')
    pr = parse_prices(sample, now_ms=1)
    if pr and pr.get("eur") == 55588 and pr.get("usd") == 64144 and pr.get("cny") == 432672:
        print("[ok] Precios: la raiz sigue siendo BTC (retrocompatible con apps <1.0.201 y con Promo)")
    else:
        print("[FALLO] Precios: la raiz ya no es el precio de BTC (%s)" % pr); ok = False
    c = (pr or {}).get("coins") or {}
    if c.get("ETH", {}).get("eur") == 1656.87 and c.get("POL", {}).get("usd") == 0.075333 \
            and c.get("SOL", {}).get("cny") == 517.2 and c.get("DOGE", {}).get("eur") == 0.062312 \
            and c.get("BTC", {}).get("eur") == 55588:
        print("[ok] Precios: cada moneda sale en `coins` con sus tres divisas (BTC/ETH/POL/SOL/DOGE)")
    else:
        print("[FALLO] Precios: `coins` mal (%s)" % c); ok = False
    # Basura por moneda: cero, negativo, texto o booleano no son precios -> fuera, y si
    # no queda NADA util la respuesta entera es None (no se pisa el cache con vacio).
    junk = (b'{"bitcoin":{"eur":55588,"usd":0,"cny":"432672"},'
            b'"litecoin":{"eur":-1,"usd":true},"tether":{}}')
    pj = parse_prices(junk, now_ms=1)
    if pj and pj.get("eur") == 55588 and "usd" not in pj and "cny" not in pj \
            and list((pj.get("coins") or {}).keys()) == ["BTC"]:
        print("[ok] Precios: se descartan cero/negativo/texto/booleano y las monedas sin nada util")
    else:
        print("[FALLO] Precios: no se filtra la basura (%s)" % pj); ok = False
    if parse_prices(b'{"bitcoin":{}}') is None and parse_prices(b'[]') is None:
        print("[ok] Precios: una respuesta sin ningun precio devuelve None (se conserva el cache)")
    else:
        print("[FALLO] Precios: una respuesta vacia deberia devolver None"); ok = False
    # Un operador con VERSELL_PRICE_URL antigua (solo bitcoin) tiene que seguir yendo.
    only_btc = parse_prices(b'{"bitcoin":{"eur":1,"usd":2,"cny":3}}', now_ms=1)
    if only_btc and only_btc["eur"] == 1 and list(only_btc["coins"].keys()) == ["BTC"]:
        print("[ok] Precios: una URL vieja de solo-bitcoin sigue sirviendo (solo BTC en `coins`)")
    else:
        print("[FALLO] Precios: la URL antigua de solo-bitcoin se rompio (%s)" % only_btc); ok = False

    print("RESULTADO:", "TODO OK" if ok else "HAY FALLOS")
    return 0 if ok else 1


if __name__ == "__main__":
    if "--selftest" in sys.argv:
        sys.exit(selftest())
    run_server()