Design API da carteira pós-quântica: padrões REST e WebSocket

# Implementação de endpoints de endereço (FastAPI) de fastapi importação FastAPI, depende, HTTPException de pydantico importação Modelo Base de digitando importação Lista, opcional importação aplicação base64 = FastAPI(título="Carteira SynX API", versão ="1.0.0") classe Resposta de endereço(Modelo Base): """Morada com chaves públicas pós-quânticas""" endereço: str caminho: str kyber_public_key: str # Codificado em Base64 (1.184 bytes) sphincs_public_key:str # Codificado em Base64 (32 bytes) saldo: int saldo_pendente: int criado_em: str classe DerivarAddressRequest(BaseModel): conta: int = 0 alteração: int = 0 índice: Opcional[int] = Nenhum #Incremento automático se nenhum @app.get("/API/v1/endereços", resposta_model=Lista[Resposta de endereço]) definição assíncrona lista_endereços( wallet_id: str = Depende (get_authenticated_wallet), skip: int = 0, limite: int = 50): """ Listar endereços de carteira com saldos Nota: as chaves públicas Kyber são grandes (1,2 KB). Para a listagem, considere eliminar as chaves e procurá-las separadamente. """ endereços = espere address_service.list_addresses( wallet_id, skip=skip, limit=limit ) devolver [ Resposta de endereço(endereço=addr.address, caminho=addr.path, kyber_public_key=base64.b64encode(addr.kyber_pk).decode(), sphincs_public_key=base64.b64encode(addr.sphincs_pk).decode(), balance=addr.balance, pendente_balance=addr.pending_balance, criado_at=addr.created_at.isoformat() ) para endereço in endereços] @app.post("/API/v1/endereços/derivar", modelo_resposta=Resposta de endereço) definição assíncrona endereço_derivado( pedido: DerivarAddressRequest, carteira_id: str = Depende(get_authenticated_wallet)): """Deriva o novo endereço no caminho de derivação especificado""" endereço = espere endereço_service.derive_address( wallet_id, conta=request.account, change=request.change, index=request.index ) devolver Resposta de endereço(...)

classe Resumo da transação(Modelo Base): """Transação sem dados completos de assinatura (para listas)""" tx_id: str timestamp: str inputs_count: int outputs_count: int valor: int taxa: int confirmações: int estado: str # "pendente", "confirmado", "falhou" classe Transação completa(Modelo Base): """Transação completa incluindo assinaturas""" tx_id: str versão: int timestamp: str entradas: Lista['TransactionInputSchema'] saídas: Lista['TransactionOutputSchema'] taxa: confirmações int: int block_hash: Opcional[str] raw_hex: str # Transação serializada completa classe Esquema de entrada de transações(Modelo Base): prev_tx_id: str prev_output_index: quantidade int: endereço int: assinatura str: str # Base64 (~10,5 KB para SPHINCS+-SHAKE-128s, 7.856 bytes brutos) chave_pública: str # Base64 (44 bytes para SPHINCS+) classe ConstruirTransactionRequest(ModeloBase): saídas: Lista['Especificação de saída'] taxa_taxa: Opcional[int] = Nenhum # Calcula automaticamente se nenhum change_address: Opcional[str] = Nenhum # Seleção automática se nenhum classe Especificação de saída(ModeloBase): destinatário: str valor: int memo: Opcional[str] = Nenhum @app.get("/API/v1/transações", resposta_model=Lista[Resumo da transação]) definição assíncrona lista_transações( wallet_id: str = Depende (get_authenticated_wallet), skip: int = 0, limite: int = 20, status: Opcional[str] = Nenhum): """ Transações de lista (apenas resumos) As assinaturas são eliminadas das respostas da lista para reduzir a carga útil. Utilize GET /transactions/{tx_id} para transações completas com assinaturas. """ txs = espere transaction_service.list_transactions( wallet_id, skip=skip, limit=limit, status=status ) devolver [tx.to_summary() para tx in txs] @app.get("/API/v1/transações/{tx_id}", modelo_resposta=Transação completa) definição assíncrona get_transaction( tx_id: str, carteira_id: str = Depende (get_authenticated_wallet), include_signatures: bool = True ): """ Obter detalhes completos da transação Defina include_signatures=false para reduzir o tamanho da resposta se apenas necessitar de metadados da transação. """ tx = espere transação_service.get_transaction(wallet_id, tx_id) se não tx: aumento HTTPException(codigo_estado=404, detalhe="Transação não encontrada") devolver tx.to_full_schema(include_signatures=include_signatures) @app.post("/API/v1/transações/construir") definição assíncrona build_transaction( pedido: ConstruirTransactionRequest, carteira_id: str = Depende(get_authenticated_wallet)): """ Construir transação não assinada Retorna os dados de transação prontos para assinatura do lado do cliente. A assinatura acontece no cliente para manter as chaves privadas fora do servidor. """ unsigned_tx = espere transaction_service.build_transaction( wallet_id, saídas=request.outputs, fee_rate=request.fee_rate, change_address=request.change_address ) devolver { "unsigned_tx": base64.b64encode(unsigned_tx.serialize_for_signing()).decode(), "signing_message": base64.b64encode(unsigned_tx.tx_hash()).decode(), "entradas_para_assinar": [ { "índice": i, "morada": inp.endereço, "montante": inp. quantidade, "caminho_derivação": inp.caminho } para eu, inp in enumerar(unsigned_tx. inputs)], "taxa_estimada": unsigned_tx.fee, "tamanho_estimado": unsigned_tx.estimated_size() }

# Fluxo de autenticação utilizando Kyber + SPHINCS+ importação ok importação hashlib importação hmac de datahora importação datahora, horadelta classe QuantumSafeAuth: """ Fluxo de autenticação API Quantum-safe: 1. Cliente envia chave pública Kyber 2. Servidor encapsula chave de sessão 3. Cliente desencapsula para obter chave de sessão 4. Pedidos assinados com HMAC utilizando chave de sessão """ def __iniciar__(auto): self.session_store = {} # Em produção, utilize Redis self.session_duration = timedelta(horas=24) definição assíncrona sessão_iniciativa( self, wallet_id: str, client_kyber_pk: bytes ) -> dict: """ Passo 1: Cliente inicia sessão com a chave pública Kyber Servidor encapsula um segredo de sessão para a chave do cliente """ kem = oqs.KeyEncapsulation("Kyber768") texto cifrado, shared_secret = kem.encap_secret(client_kyber_pk) # Deduza a chave de sessão do segredo partilhado session_key = hashlib.shake_256(shared_secret + b"chave de sessão" ).digerir(32) # Cria ID de sessão session_id = hashlib.Blake2b(shared_secret + str(datetime.utcnow()).encode(), digest_size=16 .hexdigest() # Armazenar sessão (lado do servidor) self.session_store[session_id] = { "carteira_id": carteira_id, "session_key": chave_sessão, "expira_em": datetime.utcnow() + self.session_duration, "criado_em": datetime.utcnow() } devolver { "id_sessão": ID_sessão, "texto cifrado": base64.b64encode(texto cifrado).decode(), "expira_em": (datetime.utcnow() + self.session_duration).isoformat() } def verificar_solicitação(self, session_id: str, request_signature: bytes, request_data: bytes, timestamp: int) -> Opcional[str]: """ Verificar a assinatura do pedido utilizando a chave de sessão Retorna wallet_id se for válido, Nenhum caso contrário """ sessão = self.session_store.get(session_id) se não sessão: devolver Nenhum #Verifique a expiração if datetime.utcnow() > sessão["expira_em"]: del self.session_store[session_id] devolver Nenhum # Verifique o carimbo de data e hora (evita a repetição) request_time = datetime.fromtimestamp(timestamp) if abs((datetime.utcnow() - request_time).total_seconds()) > 300: devolver Nenhum # Mais de 5 minutos de idade/futuro #Verifique a assinatura HMAC esperado_sig = hmac.new(sessão["session_key"], request_data + str(timestamp).encode(), hashlib.Blake2b ).digest() if hmac.compare_digest(request_signature, esperado_sig): devolver sessão["carteira_id"] devolver Nenhum # Dependência FastAPI para rotas autenticadas serviço_auth = QuantumSafeAuth() definição assíncrona get_authenticated_wallet( x_session_id: str = Cabeçalho(...), x_signature: str = Cabeçalho(...), x_timestamp: str = Cabeçalho(...), pedido: Pedido = Nenhum) -> str: """Dependência que valida a autenticação quântica segura""" corpo = espere 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) ) se não carteira_id: aumento HTTPException(codigo_estado=401, detalhe="Autenticação inválida") devolver carteira_id

# Implementação WebSocket para atualizações em tempo real de fastapi importação WebSocket, WebSocketDisconnect importação JSON importação assíncio classe Gestor de conexões: """Gerir ligações WebSocket por carteira""" def __iniciar__(self): self.active_connections: dict[str, List[WebSocket]] = {} definição assíncrona ligar(self, websocket: WebSocket, wallet_id: str): espere websocket.accept() if carteira_id não em self.active_connections: self.active_connections[wallet_id] = [] self.active_connections[wallet_id].append(websocket) def desconectar(self, websocket: WebSocket, wallet_id: str): if carteira_id in self.active_connections: self.active_connections[wallet_id].remove(websocket) definição assíncrona broadcast_to_wallet(self, wallet_id: str, mensagem: dict): if carteira_id in self.active_connections: ligações_mortas = [] para conexão in self.active_connections[wallet_id]: tentar: espere ligação.send_json(mensagem) exceto: dead_connections.append(ligação) #Limpa as ligações mortas para conexão in dead_connections: self.active_connections[wallet_id].remove(conn) manager = Gestor de conexões() @app.websocket("/ws/{wallet_id}") definição assíncrona websocket_endpoint(websocket: WebSocket, carteira_id: str): """ WebSocket para atualizações de carteira em tempo real Eventos: - balance_update: Saldo alterado - transaction_received: Transação recebida - transaction_confirmed: TX alcançou confirmações - transaction_sent: Transmissão TX de saída """ #Autenticar ligação WebSocket auth_token=websocket.query_params.get("símbolo") se não esperar validar_ws_token(auth_token, carteira_id): espere websocket.close(código=4001) devolver espere manager.connect(websocket, wallet_id) tentar: # Envia o estado inicial espere websocket.send_json({ "tipo": "conectado", "carteira_id": carteira_id, "carimbo de data/hora": datetime.utcnow().isoformat() }) # Lidar com mensagens recebidas (assinaturas, pings) enquanto Verdadeiro: dados = espere websocket.receive_json() if dados.get("tipo") == "ping": espere websocket.send_json({"tipo": "pongue"}) Elif dados.get("tipo") == "inscrever-se": # Inscreva-se em endereços específicos endereços = dados.get("endereços", []) espere subscription_service.subscribe( wallet_id, endereços ) exceto WebSocketDisconnect: manager.disconnect(websocket, wallet_id) # Transmissão de eventos (chamada pelo monitor blockchain) definição assíncrona broadcast_balance_update(wallet_id: str, address: str, new_balance: int): espere manager.broadcast_to_wallet(wallet_id, { "tipo": "balance_update", "morada": morada, "equilíbrio": novo_equilíbrio, "carimbo de data/hora": datetime.utcnow().isoformat() }) definição assíncrona transmissão_transação_recebida(wallet_id: str, tx_summary: dict): espere manager.broadcast_to_wallet(wallet_id, { "tipo": "transação_recebida", "transação": resumo_tx, # Apenas resumo, não assinatura completa "carimbo de data/hora": datetime.utcnow().isoformat() })

# Active a compressão gzip no FastAPI de fastapi.middleware.gzip importação GZipMiddleware app.add_middleware(GZipMiddleware, tamanho_mínimo=1000) # Opcional: respostas do MessagePack para clientes móveis/embedded de fastapi.responses importação Resposta importação pacote de mensagens classe MsgPackResponse(Resposta): media_type = "aplicação/msgpack" def devolver(próprio, conteúdo) -> bytes: devolver msgpack.packb(conteúdo, use_bin_type=True) @app.get("/API/v1/transações/{tx_id}/binário") definição assíncrona get_transaction_binary(tx_id:str): """Obter transação no formato MessagePack (inferior a JSON)""" tx = espere transação_service.get_transaction(tx_id) devolver MsgPackResponse(conteúdo=tx.to_dict())

# Limitação de taxa para a carteira API de API lenta importação Limitador, _rate_limit_exceeded_handler de slowapi.errors importação RateLimitExceeded limitador = Limitador(key_func=get_wallet_id_from_request) app.state.limiter = limitador app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler) # Limites diferentes para operações diferentes TAXA_LIMITES = { "ler": "100/minuto", # Verificações de saldo, listas de transferências "escrever": "20/minuto", # Derivação de endereços "transmissão": "5/minuto", # Transmissão de transações } @app.get("/API/v1/equilíbrio") @limiter.limit("100/minuto") definição assíncrona obter_equilíbrio(pedido: Pedido): ... @app.post("/API/v1/transações/transmissão") @limiter.limit("5/minuto") definição assíncrona transmissão_transação(solicitação: Solicitação): """Limite de transmissão mais rigoroso para evitar spam""" ...

# Respostas de erro padronizadas de enumeração importação Enum classe Código de erro(str, Enum): INVALID_ADDRESS = "INVALID_ADDRESS" INSUFFICIENT_BALANCE = "INSUFICIENT_BALANCE" INVALID_SIGNATURE = "INVALID_SIGNATURE" TRANSACTION_REJECTED = "TRANSACTION_REJECTED" TAXA_LIMITED = "RATE_LIMITED" SESSÃO_EXPIRED= "SESSÃO_EXPIRED" DERIVAÇÃO_FAILED = "DERIVATION_FAILED" classe Erro API(ModeloBase): código: Código de erro mensagem: detalhes de str: Opcional[dict] = Nenhum @app.exception_handler(Erro de saldo insuficiente) definição assíncrona insuficiente_balance_handler(solicitação, exc): devolver JSONResponse( código_estado=400, conteúdo=Erro API(código=Código de erro.INSUFFICIENT_BALANCE, mensagem="Saldo insuficiente para transação", detalhes={ "disponível": exc.disponível, "obrigatório": exc.obrigatório, "défice": exc.required - exc.available } ).dict() )