Post-Quantum Wallet API Design: REST and WebSocket Patterns

# Address endpoints implementation (FastAPI) from fastapi import FastAPI, Depends, HTTPException from pydantic import BaseModel from typing import List, Optional import base64 app = FastAPI(title="SynX Wallet API", version="1.0.0") class AddressResponse(BaseModel): """Address with post-quantum public keys""" address: str path: str kyber_public_key: str # Base64 encoded (1,184 bytes) sphincs_public_key: str # Base64 encoded (32 bytes) balance: int pending_balance: int created_at: str class DeriveAddressRequest(BaseModel): account: int = 0 change: int = 0 index: Optional[int] = None # Auto-increment if None @app.get("/api/v1/addresses", response_model=List[AddressResponse]) async def list_addresses( wallet_id: str = Depends(get_authenticated_wallet), skip: int = 0, limit: int = 50 ): """ List wallet addresses with balances Note: Kyber public keys are large (1.2KB). For listing, consider excluding keys and fetching separately. """ addresses = await address_service.list_addresses( wallet_id, skip=skip, limit=limit ) return [ AddressResponse( address=addr.address, path=addr.path, kyber_public_key=base64.b64encode(addr.kyber_pk).decode(), sphincs_public_key=base64.b64encode(addr.sphincs_pk).decode(), balance=addr.balance, pending_balance=addr.pending_balance, created_at=addr.created_at.isoformat() ) for addr in addresses ] @app.post("/api/v1/addresses/derive", response_model=AddressResponse) async def derive_address( request: DeriveAddressRequest, wallet_id: str = Depends(get_authenticated_wallet) ): """Derive new address at specified derivation path""" addr = await address_service.derive_address( wallet_id, account=request.account, change=request.change, index=request.index ) return AddressResponse(...)

class TransactionSummary(BaseModel): """Transaction without full signature data (for lists)""" tx_id: str timestamp: str inputs_count: int outputs_count: int amount: int fee: int confirmations: int status: str # "pending", "confirmed", "failed" class TransactionFull(BaseModel): """Full transaction including signatures""" tx_id: str version: int timestamp: str inputs: List['TransactionInputSchema'] outputs: List['TransactionOutputSchema'] fee: int confirmations: int block_hash: Optional[str] raw_hex: str # Full serialized transaction class TransactionInputSchema(BaseModel): prev_tx_id: str prev_output_index: int amount: int address: str signature: str # Base64 (~10.5KB for SPHINCS+-128s) public_key: str # Base64 (44 bytes for SPHINCS+) class BuildTransactionRequest(BaseModel): outputs: List['OutputSpec'] fee_rate: Optional[int] = None # Auto-calculate if None change_address: Optional[str] = None # Auto-select if None class OutputSpec(BaseModel): recipient: str amount: int memo: Optional[str] = None @app.get("/api/v1/transactions", response_model=List[TransactionSummary]) async def list_transactions( wallet_id: str = Depends(get_authenticated_wallet), skip: int = 0, limit: int = 20, status: Optional[str] = None ): """ List transactions (summaries only) Signatures are excluded from list responses to reduce payload. Use GET /transactions/{tx_id} for full transaction with signatures. """ txs = await transaction_service.list_transactions( wallet_id, skip=skip, limit=limit, status=status ) return [tx.to_summary() for tx in txs] @app.get("/api/v1/transactions/{tx_id}", response_model=TransactionFull) async def get_transaction( tx_id: str, wallet_id: str = Depends(get_authenticated_wallet), include_signatures: bool = True ): """ Get full transaction details Set include_signatures=false to reduce response size if you only need transaction metadata. """ tx = await transaction_service.get_transaction(wallet_id, tx_id) if not tx: raise HTTPException(status_code=404, detail="Transaction not found") return tx.to_full_schema(include_signatures=include_signatures) @app.post("/api/v1/transactions/build") async def build_transaction( request: BuildTransactionRequest, wallet_id: str = Depends(get_authenticated_wallet) ): """ Build unsigned transaction Returns transaction data ready for client-side signing. Signing happens on client to keep private keys off server. """ unsigned_tx = await transaction_service.build_transaction( wallet_id, outputs=request.outputs, fee_rate=request.fee_rate, change_address=request.change_address ) return { "unsigned_tx": base64.b64encode(unsigned_tx.serialize_for_signing()).decode(), "signing_message": base64.b64encode(unsigned_tx.tx_hash()).decode(), "inputs_to_sign": [ { "index": i, "address": inp.address, "amount": inp.amount, "derivation_path": inp.path } for i, inp in enumerate(unsigned_tx.inputs) ], "estimated_fee": unsigned_tx.fee, "estimated_size": unsigned_tx.estimated_size() }

# Authentication flow using Kyber + SPHINCS+ import oqs import hashlib import hmac from datetime import datetime, timedelta class QuantumSafeAuth: """ Quantum-safe API authentication Flow: 1. Client sends Kyber public key 2. Server encapsulates session key 3. Client decapsulates to get session key 4. Requests signed with HMAC using session key """ def __init__(self): self.session_store = {} # In production, use Redis self.session_duration = timedelta(hours=24) async def initiate_session( self, wallet_id: str, client_kyber_pk: bytes ) -> dict: """ Step 1: Client initiates session with Kyber public key Server encapsulates a session secret to client's key """ kem = oqs.KeyEncapsulation("Kyber768") ciphertext, shared_secret = kem.encap_secret(client_kyber_pk) # Derive session key from shared secret session_key = hashlib.shake_256( shared_secret + b"session-key" ).digest(32) # Create session ID session_id = hashlib.blake2b( shared_secret + str(datetime.utcnow()).encode(), digest_size=16 ).hexdigest() # Store session (server-side) self.session_store[session_id] = { "wallet_id": wallet_id, "session_key": session_key, "expires_at": datetime.utcnow() + self.session_duration, "created_at": datetime.utcnow() } return { "session_id": session_id, "ciphertext": base64.b64encode(ciphertext).decode(), "expires_at": (datetime.utcnow() + self.session_duration).isoformat() } def verify_request( self, session_id: str, request_signature: bytes, request_data: bytes, timestamp: int ) -> Optional[str]: """ Verify request signature using session key Returns wallet_id if valid, None otherwise """ session = self.session_store.get(session_id) if not session: return None # Check expiration if datetime.utcnow() > session["expires_at"]: del self.session_store[session_id] return None # Check timestamp (prevent replay) request_time = datetime.fromtimestamp(timestamp) if abs((datetime.utcnow() - request_time).total_seconds()) > 300: return None # More than 5 minutes old/future # Verify HMAC signature expected_sig = hmac.new( session["session_key"], request_data + str(timestamp).encode(), hashlib.blake2b ).digest() if hmac.compare_digest(request_signature, expected_sig): return session["wallet_id"] return None # FastAPI dependency for authenticated routes auth_service = QuantumSafeAuth() async def get_authenticated_wallet( x_session_id: str = Header(...), x_signature: str = Header(...), x_timestamp: str = Header(...), request: Request = None ) -> str: """Dependency that validates quantum-safe authentication""" body = await 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) ) if not wallet_id: raise HTTPException(status_code=401, detail="Invalid authentication") return wallet_id

# WebSocket implementation for real-time updates from fastapi import WebSocket, WebSocketDisconnect import json import asyncio class ConnectionManager: """Manage WebSocket connections per wallet""" def __init__(self): self.active_connections: dict[str, List[WebSocket]] = {} async def connect(self, websocket: WebSocket, wallet_id: str): await websocket.accept() if wallet_id not in self.active_connections: self.active_connections[wallet_id] = [] self.active_connections[wallet_id].append(websocket) def disconnect(self, websocket: WebSocket, wallet_id: str): if wallet_id in self.active_connections: self.active_connections[wallet_id].remove(websocket) async def broadcast_to_wallet(self, wallet_id: str, message: dict): if wallet_id in self.active_connections: dead_connections = [] for connection in self.active_connections[wallet_id]: try: await connection.send_json(message) except: dead_connections.append(connection) # Clean up dead connections for conn in dead_connections: self.active_connections[wallet_id].remove(conn) manager = ConnectionManager() @app.websocket("/ws/{wallet_id}") async def websocket_endpoint(websocket: WebSocket, wallet_id: str): """ WebSocket for real-time wallet updates Events: - balance_update: Balance changed - transaction_received: Incoming transaction - transaction_confirmed: TX reached confirmations - transaction_sent: Outgoing TX broadcast """ # Authenticate WebSocket connection auth_token = websocket.query_params.get("token") if not await validate_ws_token(auth_token, wallet_id): await websocket.close(code=4001) return await manager.connect(websocket, wallet_id) try: # Send initial state await websocket.send_json({ "type": "connected", "wallet_id": wallet_id, "timestamp": datetime.utcnow().isoformat() }) # Handle incoming messages (subscriptions, pings) while True: data = await websocket.receive_json() if data.get("type") == "ping": await websocket.send_json({"type": "pong"}) elif data.get("type") == "subscribe": # Subscribe to specific addresses addresses = data.get("addresses", []) await subscription_service.subscribe( wallet_id, addresses ) except WebSocketDisconnect: manager.disconnect(websocket, wallet_id) # Event broadcasting (called by blockchain monitor) async def broadcast_balance_update(wallet_id: str, address: str, new_balance: int): await manager.broadcast_to_wallet(wallet_id, { "type": "balance_update", "address": address, "balance": new_balance, "timestamp": datetime.utcnow().isoformat() }) async def broadcast_transaction_received(wallet_id: str, tx_summary: dict): await manager.broadcast_to_wallet(wallet_id, { "type": "transaction_received", "transaction": tx_summary, # Summary only, not full signature "timestamp": datetime.utcnow().isoformat() })

# Enable gzip compression in FastAPI from fastapi.middleware.gzip import GZipMiddleware app.add_middleware(GZipMiddleware, minimum_size=1000) # Optional: MessagePack responses for mobile/embedded clients from fastapi.responses import Response import msgpack class MsgPackResponse(Response): media_type = "application/msgpack" def render(self, content) -> bytes: return msgpack.packb(content, use_bin_type=True) @app.get("/api/v1/transactions/{tx_id}/binary") async def get_transaction_binary(tx_id: str): """Get transaction in MessagePack format (smaller than JSON)""" tx = await transaction_service.get_transaction(tx_id) return MsgPackResponse(content=tx.to_dict())

# Rate limiting for wallet API from slowapi import Limiter, _rate_limit_exceeded_handler from slowapi.errors import RateLimitExceeded limiter = Limiter(key_func=get_wallet_id_from_request) app.state.limiter = limiter app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler) # Different limits for different operations RATE_LIMITS = { "read": "100/minute", # Balance checks, tx lists "write": "20/minute", # Address derivation "broadcast": "5/minute", # Transaction broadcast } @app.get("/api/v1/balance") @limiter.limit("100/minute") async def get_balance(request: Request): ... @app.post("/api/v1/transactions/broadcast") @limiter.limit("5/minute") async def broadcast_transaction(request: Request): """Stricter limit for broadcast to prevent spam""" ...

# Standardized error responses from enum import Enum class ErrorCode(str, Enum): INVALID_ADDRESS = "INVALID_ADDRESS" INSUFFICIENT_BALANCE = "INSUFFICIENT_BALANCE" INVALID_SIGNATURE = "INVALID_SIGNATURE" TRANSACTION_REJECTED = "TRANSACTION_REJECTED" RATE_LIMITED = "RATE_LIMITED" SESSION_EXPIRED = "SESSION_EXPIRED" DERIVATION_FAILED = "DERIVATION_FAILED" class APIError(BaseModel): code: ErrorCode message: str details: Optional[dict] = None @app.exception_handler(InsufficientBalanceError) async def insufficient_balance_handler(request, exc): return JSONResponse( status_code=400, content=APIError( code=ErrorCode.INSUFFICIENT_BALANCE, message="Insufficient balance for transaction", details={ "available": exc.available, "required": exc.required, "shortfall": exc.required - exc.available } ).dict() )