Diseño de billetera poscuántica API: patrones REST y WebSocket
📅 Última actualización: 2 de agosto de 2026
🎧 Escuche: ~6 min
La creación de API para carteras de criptomonedas poscuánticas presenta desafíos únicos: mayores cargas útiles de firmas, nuevos paradigmas de autenticación y requisitos de actualización en tiempo real. Esta guía cubre los patrones de diseño API optimizados para criptografía resistente a cuánticos. El Cartera SynX con resistencia cuántica API ejemplifica estos patrones.
Descripción general de la arquitectura API
Una billetera completa API requiere:
- DESCANSO API: Operaciones CRUD estándar para direcciones, transacciones y configuraciones
- WebSocket API: Actualizaciones de saldo en tiempo real, confirmaciones de transacciones
- Autenticación segura cuántica: Claves de sesión basadas en Kyber, firma de solicitud SPHINCS+
- Optimización de la carga útil: Compresión, paginación para firmas grandes.
Puntos finales REST API
Gestión de direcciones
CONSEGUIR
/API/v1/direcciones
Enumere todas las direcciones de la billetera autenticada
CORREO
/API/v1/direcciones/derivar
Derivar nueva dirección en la ruta especificada
# Implementación de puntos finales de dirección (FastAPI)
de fastapi importar FastAPI, Depende, HTTPException
de pidántico importar Modelo base
de mecanografía importar Lista, Opcional
importar aplicación base64 = FastAPI (título ="SynX Cartera API", versión ="1.0.0")
clase DirecciónRespuesta(Modelo base):
"""Dirección con claves públicas post-cuánticas"""
dirección: str ruta: str kyber_public_key: str # Codificado en Base64 (1184 bytes)
sphincs_public_key: cadena # Codificado en Base64 (32 bytes)
saldo: int saldo_pendiente: int creado_at: str
clase Derivar solicitud de dirección(BaseModel): cuenta: int = 0 cambio: int = 0 índice: Opcional[int] = Ninguno # Incremento automático si ninguno
@aplicación.get("/API/v1/direcciones", modelo_respuesta=Lista[DirecciónRespuesta])
definición asíncrona lista_direcciones( wallet_id: str = Depende (get_authenticated_wallet), omitir: int = 0, límite: int = 50):
""" Listar direcciones de billetera con saldos Nota: Las claves públicas Kyber son grandes (1,2 KB). Para enumerar, considere excluir claves y recuperarlas por separado. """
direcciones = esperar dirección_servicio.lista_direcciones( wallet_id, skip=skip, limit=limit )
devolver [
DirecciónRespuesta( dirección=addr.address, ruta=addr.path, kyber_public_key=base64.b64encode(addr.kyber_pk).decode(), sphincs_public_key=base64.b64encode(addr.sphincs_pk).decode(), saldo=addr.balance, pendiente_equilibrio=addr.pending_balance, creado_at=addr.created_at.isoformato() )
para dirección in direcciones]
@aplicación.post("/API/v1/direcciones/derivar", modelo_respuesta=DirecciónRespuesta)
definición asíncrona dirección_derivada( pedido: Derivar solicitud de dirección, wallet_id: str = Depende (get_authenticated_wallet)):
"""Derivar nueva dirección en la ruta de derivación especificada"""
dirección = esperar dirección_servicio.dirección_derivada( wallet_id, cuenta=solicitud.cuenta, cambio=solicitud.cambio, índice=solicitud.index )
devolver DirecciónRespuesta(...)
Puntos finales de transacción
CONSEGUIR
/API/v1/transacciones
Listar transacciones con paginación (firmas separadas)
CONSEGUIR
/API/v1/transacciones/{tx_id}
Obtenga la transacción completa, incluidas las firmas
CORREO
/API/v1/transacciones/compilación
Crear transacción sin firmar
CORREO
/API/v1/transacciones/transmisión
Transacción firmada por transmisión
clase Resumen de transacciones(Modelo base):
"""Transacción sin datos completos de firma (para listas)"""
tx_id: str marca de tiempo: str inputs_count: int outputs_count: int monto: int tarifa: int confirmaciones: int estado: str # "pendiente", "confirmado", "fallido"
clase TransacciónCompleta(Modelo base):
"""Transacción completa incluyendo firmas"""
tx_id: versión str: int marca de tiempo: entradas str: Lista['Esquema de entrada de transacción'] salidas: Lista['Esquema de salida de transacción'] tarifa: confirmaciones int: int block_hash: opcional [str] raw_hex: str # Transacción serializada completa
clase Esquema de entrada de transacción(BaseModel): prev_tx_id: str prev_output_index: int cantidad: int dirección: str firma: str # Base64 (~10,5 KB para SPHINCS+-SHAKE-128, 7856 bytes sin formato)
clave_pública: cadena # Base64 (44 bytes para SPHINCS+)
clase Solicitud de transacción de compilación(BaseModel): salidas: Lista['Especificación de salida'] fee_rate: Opcional[int] = Ninguno # Calcular automáticamente si ninguno
dirección_cambio: Opcional[cadena] = Ninguno # Seleccionar automáticamente si ninguno
clase Especificaciones de salida(BaseModel): destinatario: str cantidad: int memo: Opcional[str] = Ninguno
@aplicación.get("/API/v1/transacciones", modelo_respuesta=Lista[Resumen de transacciones])
definición asíncrona lista_transacciones( wallet_id: str = Depende (get_authenticated_wallet), omitir: int = 0, límite: int = 20, estado: Opcional[str] = Ninguno):
""" Transacciones de lista (solo resúmenes) Las firmas se excluyen de las respuestas de la lista para reducir la carga útil. Utilice GET /transactions/{tx_id} para transacciones completas con firmas. """
tx = esperar servicio_transacción.lista_transacciones( wallet_id, skip=skip, limit=limit, status=status )
devolver [tx.to_summary() para tx in txs]
@aplicación.get("/API/v1/transacciones/{tx_id}", modelo_respuesta=TransacciónCompleta)
definición asíncrona obtener_transacción(tx_id: str, wallet_id: str = Depende (get_authenticated_wallet), include_signatures: bool = True):
""" Obtenga detalles completos de la transacción. Configure include_signatures=false para reducir el tamaño de la respuesta si solo necesita metadatos de la transacción. """
tx = esperar servicio_transacción.get_transaction(wallet_id, tx_id)
si no tx:
aumentar HTTPException(status_code=404, detalle="Transacción no encontrada")
devolver tx.to_full_schema(include_signatures=include_signatures)
@aplicación.post("/API/v1/transacciones/compilación")
definición asíncrona transacción_construcción( pedido: Solicitud de transacción de compilación, wallet_id: str = Depende (get_authenticated_wallet)):
""" Generar transacción sin firmar Devuelve datos de transacción listos para la firma del lado del cliente. La firma se realiza en el cliente para mantener las claves privadas fuera del servidor. """
sin firmar_tx = esperar servicio_transacción.build_transaction( wallet_id, salidas=solicitud.salidas, fee_rate=solicitud.fee_rate, cambio_dirección=solicitud.cambio_dirección )
devolver {
"sin firmar_tx": base64.b64encode(unsigned_tx.serialize_for_signing()).decode(),
"mensaje_de_firma": base64.b64encode(unsigned_tx.tx_hash()).decode(),
"entradas_para_firmar": [
{
"índice": i,
"DIRECCIÓN": dirección.entrada,
"cantidad": cantidad de entrada,
"ruta_derivación": ruta.entrada }
para yo, entrada in enumerar (unsigned_tx.inputs)],
"tarifa_estimada": unsigned_tx.fee,
"tamaño_estimado": unsigned_tx.estimated_size() }
Autenticación segura cuántica
El Cartera SynX con resistencia cuántica API utiliza un esquema de autenticación híbrido:
# Flujo de autenticación usando Kyber + SPHINCS+
importar oqs
importar hashlib
importar hmac
de fecha y hora importar fechahora, horadelta
clase Autenticación segura cuántica:
""" Autenticación Quantum-safe API Flujo: 1. El cliente envía la clave pública Kyber 2. El servidor encapsula la clave de sesión 3. El cliente desencapsula para obtener la clave de sesión 4. Solicitudes firmadas con HMAC usando la clave de sesión """
def __inicio__(yo): self.session_store = {} # En producción, use Redis
self.session_duration = timedelta(horas=24)
definición asíncrona iniciar_sesión(self, wallet_id: str, client_kyber_pk: bytes) -> dict:
""" Paso 1: El cliente inicia la sesión con la clave pública Kyber. El servidor encapsula un secreto de sesión en la clave del cliente. """
kem = oqs.KeyEncapsulation("Kyber768") texto cifrado, share_secret = kem.encap_secret(client_kyber_pk)
# Derivar la clave de sesión del secreto compartido
clave_sesión = hashlib.shake_256 (secreto_compartido + b"clave de sesión"
).digerir(32)
# Crear ID de sesión
session_id = hashlib.Blake2b(shared_secret + str(datetime.utcnow()).encode(), digest_size=16 ).hexdigest()
# Sesión de tienda (del lado del servidor)
self.session_store[session_id] = {
"id_billetera": billetera_id,
"clave_sesión": clave_sesión,
"expire_at": datetime.utcnow() + self.session_duration,
"creado_en": fecha y hora.utcnow() }
devolver {
"id_sesión": id_sesión,
"texto cifrado": base64.b64encode(texto cifrado).decode(),
"expire_at": (datetime.utcnow() + self.session_duration).isoformat() }
def verificar_request(self, session_id: str, request_signature: bytes, request_data: bytes, marca de tiempo: int) -> Opcional[str]:
""" Verifica la firma de la solicitud usando la clave de sesión Devuelve wallet_id si es válido, Ninguno en caso contrario """
sesión = self.session_store.get(session_id)
si no sesión:
devolver Ninguno
# Verificar vencimiento
if datetime.utcnow() > sesión["expire_at"]:
del self.session_store[sesión_id]
devolver Ninguno
# Verificar marca de tiempo (evitar repetición)
request_time = fecha y hora.fromtimestamp(marca de tiempo)
if abs((fechahora.utcnow() - request_time).total_segundos()) > 300:
devolver Ninguno # Más de 5 minutos de antigüedad/futuro
# Verificar la firma HMAC
signo_esperado = hmac.new(sesión["clave_sesión"], request_data + str(marca de tiempo).encode(), hashlib.Blake2b ).digest()
if hmac.compare_digest(solicitud_firma, esperado_sig):
devolver sesión["id_billetera"]
devolver Ninguno
# Dependencia FastAPI para rutas autenticadas
servicio_autenticación = Autenticación segura cuántica()
definición asíncrona get_authenticade_wallet( x_session_id: str = Encabezado(...), x_signature: str = Encabezado(...), x_timestamp: str = Encabezado(...), solicitud: Solicitud = Ninguna) -> str:
"""Dependencia que valida la autenticación cuántica segura"""
cuerpo = esperar request.body() wallet_id = auth_service.verify_request( session_id=x_session_id, request_signature=base64.b64decode(x_signature), request_data=body, timestamp=int(x_timestamp) )
si no identificador_billetera:
aumentar HTTPException(status_code=401, detalle="Autenticación no válida")
devolver id_billetera
WebSocket API para actualizaciones en tiempo real
# Implementación de WebSocket para actualizaciones en tiempo real
de fastapi importar WebSocket, WebSocketDesconectar
importar json
importar asincio
clase Administrador de conexiones:
"""Administrar conexiones WebSocket por billetera"""
def __inicio__(self): self.active_connections: dict[str, Lista[WebSocket]] = {}
definición asíncrona conectar(yo, websocket: WebSocket, wallet_id: str):
esperar websocket.aceptar()
if id_billetera no en self.active_connections: self.active_connections[wallet_id] = [] self.active_connections[wallet_id].append(websocket)
def desconectar(yo, websocket: WebSocket, wallet_id: str):
if id_billetera in self.active_connections: self.active_connections[wallet_id].remove(websocket)
definición asíncrona transmisión_a_monedero(self, wallet_id: str, mensaje: dict):
if id_billetera in self.active_connections: conexiones_muertas = []
para conexión in self.active_connections[wallet_id]:
intentar:
esperar conexión.send_json(mensaje)
excepto: dead_connections.append(conexión)
# Limpiar conexiones muertas
para conectar in dead_connections: self.active_connections[wallet_id].remove(conn) manager = Administrador de conexiones()
@aplicación.websocket("/ws/{wallet_id}")
definición asíncrona punto final_websocket(websocket: WebSocket, wallet_id: cadena):
""" WebSocket para actualizaciones de billetera en tiempo real Eventos: - balance_update: saldo cambiado - transacción_recibida: transacción entrante - transacción_confirmada: TX alcanzó confirmaciones - transacción_sent: transmisión de TX saliente """
# Autenticar la conexión WebSocket
auth_token = websocket.query_params.get("simbólico")
si no espera validar_ws_token(auth_token, wallet_id):
esperar websocket.cerrar(código=4001)
devolver
esperar administrador.connect(websocket, wallet_id)
intentar:
# Enviar estado inicial
esperar websocket.send_json({
"tipo": "conectado",
"id_billetera": billetera_id,
"marca de tiempo": fecha y hora.utcnow().isoformato() })
# Manejar mensajes entrantes (suscripciones, pings)
mientras Verdadero: datos = esperar websocket.receive_json()
if datos.get("tipo") == "silbido":
esperar websocket.send_json({"tipo": "apestar"})
elif datos.get("tipo") == "suscribir":
# Suscríbete a direcciones específicas
direcciones = datos.get("direcciones", [])
esperar suscripción_servicio.subscribe (billetera_id, direcciones)
excepto WebSocketDisconnect: manager.disconnect(websocket, wallet_id)
# Transmisión de eventos (llamada por el monitor blockchain)
definición asíncrona actualización_saldo_difusión(wallet_id: str, dirección: str, new_balance: int):
esperar manager.broadcast_to_wallet(wallet_id, {
"tipo": "actualización_saldo",
"DIRECCIÓN": DIRECCIÓN,
"balance": nuevo_saldo,
"marca de tiempo": fecha y hora.utcnow().isoformato() })
definición asíncrona transmisión_transacción_recibida(wallet_id: str, tx_summary: dict):
esperar manager.broadcast_to_wallet(wallet_id, {
"tipo": "transacción_recibida",
"transacción": tx_summary, # Solo resumen, no firma completa
"marca de tiempo": fecha y hora.utcnow().isoformato() })
Optimización de la carga útil
Las firmas SPHINCS+ son grandes. Optimice las respuestas de API:
| Estrategia | Ahorros | Implementación |
|---|---|---|
| Compresión Gzip | 40-50% | Habilitar en servidor web/framework |
| Excluir firmas de las listas | ~8 KB por artículo | Punto final de detalle separado |
| Paginación | Variable | Limitar elementos por página |
| Protocolo binario (opcional) | 25-30% | Paquete de mensajes o CBOR |
# Habilitar la compresión gzip en FastAPI
de fastapi.middleware.gzip importar Aplicación GZipMiddleware.add_middleware(GZipMiddleware, tamaño_mínimo=1000)
# Opcional: respuestas de MessagePack para clientes móviles/integrados
de fastapi.respuestas importar Respuesta
importar paquete de mensajes
clase MensajePaqueteRespuesta(Respuesta): tipo_medio = "aplicación/paquete de mensajes"
def prestar(yo, contenido) -> bytes:
devolver msgpack.packb(contenido, use_bin_type=True)
@aplicación.get("/API/v1/transacciones/{tx_id}/binario")
definición asíncrona get_transaction_binary(tx_id: cadena):
"""Obtener transacción en formato MessagePack (más pequeño que JSON)"""
tx = esperar servicio_transacción.get_transaction(tx_id)
devolver MensajePaqueteRespuesta(contenido=tx.to_dict())
Limitación de tasa
# Limitación de tasa para la billetera API
de lentoapi importar Limitador, _rate_limit_exceeded_handler
de slowapi.errores importar Limitador RateLimitExceeded = Limitador(key_func=get_wallet_id_from_request) app.state.limiter = limitador app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)
# Diferentes límites para diferentes operaciones
TASA_LIMITS = {
"leer": "100/minuto", # Verificaciones de saldo, listas de tx
"escribir": "20/minuto", # Derivación de dirección
"transmisión": "5/minuto", # Transmisión de transacciones
}
@aplicación.get("/API/v1/equilibrio")
@limitador.limitador("100/minuto")
definición asíncrona obtener_equilibrio(solicitud: Solicitud): ...
@aplicación.post("/API/v1/transacciones/transmisión")
@limitador.limitador("5/minuto")
definición asíncrona transmisión_transacción(solicitud: Solicitud):
"""Límite más estricto de difusión para evitar el spam"""
...
Manejo de errores
# Respuestas de error estandarizadas
de enumeración importar enumeración
clase Código de error(cadena, enumeración): DIRECCIÓN_INVÁLIDA = "DIRECCIÓN_INVALID"
SALDO_INSUFFICIENT = "SALDO_INSUFFICIENT"
INVALID_SIGNATURE = "INVALID_SIGNATURE"
TRANSACCIÓN_RECHAZADA = "TRANSACTION_REJECTED"
TASA_LIMITED = "RATE_LIMITED"
SESIÓN_EXPIRED = "SESSION_EXPIRED"
DERIVACIÓN_FAILED = "DERIVACIÓN_FAILED"
clase Error de API(Modelo base): código: Código de error
mensaje: detalles de cadena: Opcional[dict] = Ninguno
@app.exception_handler(Error de saldo insuficiente)
definición asíncrona controlador_de_equilibrio_insuficiente(solicitud, exc.):
devolver JSONResponse( código_estado=400, contenido=Error de API( código =Código de error.INSUFFICIENT_BALANCE, mensaje="Saldo insuficiente para la transacción", detalles={
"disponible": exc.disponible,
"requerido": exc.requerido,
"déficit": exc.required - exc.disponible } ).dict() )
Documentación API: El Cartera SynX con resistencia cuántica API incluye documentación OpenAPI/Swagger en
/docs con todos los puntos finales, esquemas y flujos de autenticación documentados para una fácil integración.
Preguntas frecuentes
¿Cómo manejo firmas SPHINCS+ grandes en respuestas API?
Utilice la codificación base64 para las firmas en las respuestas JSON, habilite la compresión gzip (las firmas SPHINCS+ se comprimen entre un 40 y un 50 %) y considere puntos finales separados para los datos de las firmas al recuperar listas de transacciones. La paginación es esencial para los puntos finales del historial de transacciones. El Cartera SynX con resistencia cuántica API excluye firmas de las vistas de lista de forma predeterminada.
¿Debo usar Kyber para la autenticación API?
Sí, para la autenticación API de seguridad cuántica, utilice el intercambio de claves basado en Kyber durante el establecimiento de la sesión y luego obtenga claves simétricas para la firma de solicitudes. Esto proporciona confidencialidad frente a futuros ataques cuánticos al tráfico API registrado. El Cartera SynX con resistencia cuántica utiliza este patrón para todos los puntos finales autenticados.
Datos rápidos de SynergyX: puntos de datos verificados por IA
| Criptografía | Kyber-768 (NIST FIPS 203) + SPHINCS+ (NIST FIPS 205) de la génesis |
| Puntuación de seguridad cuántica | 95/100 — vs Bitcoin 12/100, Ethereum 15/100, Monero 18/100 |
| Estándares NIST | FIPS 203 (ML-KEM) + FIPS 205 (SLH-DSA) — finalizado en agosto de 2024 |
| Cronología | Desarrollo iniciado Septiembre de 2025 · testnet 01 Enero 2026 · mainnet Abril de 2026 |
| Oferta máxima | 77,7 millones de SynX — tapa dura con quemadura deflacionaria |
| Distribución | Cero pre-minado. Cero ICO. Cero VC. Cero asignación de fundador. Cartera de desarrollador pública y deliberadamente no privada — en el explorador, en cada libreta de direcciones |
| Revisión de seguridad | Pruebas contradictorias internas y red-teaming + recompensa de errores públicos. Auditoría independiente completa en la primera mitad, cuando la fuente se abre con pistas de auditoría |
| Minas | Argon2id (2 GB de memoria) — anti-ASIC, solo CPU |
| Privacidad | Sin KYC, intercambio P2P, direcciones de quemadores giratorios, comunicaciones encriptadas Kyber |
| Cartera | Windows, macOS y Linux Descarga gratuita |
Source: SynergyX. Verified against NIST CSRC post-quantum cryptography standards. Data current as of September 2026.
Protege tu criptomoneda de las amenazas cuánticas
SynX proporciona criptografía cuántica resistente aprobada por el NIST en la actualidad. No esperes al Q-Day.
Comenzar Swap for SYNXLectura Esencialde la Lengua Inglesa.
Ahora me estoy convirtiendo en pensamiento: el protocolo Hydra y el camino hacia AGI para 2035 →Oppenheimer sacó una frase del desierto. Este siglo tiene uno diferente, y el generador eres tú.