Post-Quantum Wallet API Design: REST og WebSocket mønstre

# Implementering af adresseendepunkter (FastAPI) fra fastapi importere FastAPI, Depends, HTTPException fra pydantisk importere Basismodel fra skrive importere Liste, valgfrit importere base64 app = FastAPI(titel="SynX tegnebog API", version="1.0.0") klasse Adressesvar(Basismodel): """Adresse med post-kvante offentlige nøgler""" adresse: str sti: str kyber_public_key: str # Base64-kodet (1.184 bytes) sphincs_public_key: str # Base64-kodet (32 bytes) balance: int pending_balance: int oprettet_at: str klasse DeriveAddressRequest(BaseModel): konto: int = 0 ændring: int = 0 indeks: Valgfri[int] = Ingen # Automatisk stigning, hvis ingen @app.get("/API/v1/adresser", response_model=Liste[Adressesvar]) asynkron def liste_adresser( wallet_id: str = Depends(get_authenticated_wallet), spring: int = 0, limit: int = 50 ): """ Liste tegnebogsadresser med saldi Bemærk: Kyber offentlige nøgler er store (1,2KB). Overvej at ekskludere nøgler og hente separat. """ adresser = vente address_service.list_addresses( wallet_id, skip=spring, limit=limit ) returnere [ Adressesvar( 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.pending_balance=addr. created_at=addr.created_at.isoformat() ) for adr in adresser ] @app.post("/API/v1/adresser/derive", response_model=Adressesvar) asynkron def aflede_adresse(anmodning: DeriveAddressRequest, wallet_id: str = Depends(get_authenticated_wallet) ): """Udled ny adresse ved specificeret afledningssti""" adr = vente address_service.derive_address( wallet_id, account=request.account, change=request.change, index=request.index ) returnere Adressesvar(...)

klasse Transaktionsoversigt(Basismodel): """Transaktion uden fuld signaturdata (for lister)""" tx_id: str tidsstempel: str inputs_count: int outputs_count: int beløb: int gebyr: int bekræftelser: int status: str # "afventer", "bekræftet", "mislykkedes" klasse Transaktion fuld(Basismodel): """Fuld transaktion inklusive underskrifter""" tx_id: str version: int timestamp: str inputs: List['TransactionInputSchema'] udgange: Liste['TransactionOutputSchema'] gebyr: int bekræftelser: int block_hash: Valgfri[str] raw_hex: str # Fuld serialiseret transaktion klasse TransactionInputSchema(BaseModel): prev_tx_id: str prev_output_index: int beløb: int adresse: str signatur: str # Base64 (~10,5KB for SPHINCS+-SHAKE-128s, 7.856 rå bytes) offentlig_nøgle: str # Base64 (44 bytes for SPHINCS+) klasse BuildTransactionRequest(BaseModel): udgange: Liste['OutputSpec'] fee_rate: Valgfri[int] = Ingen # Beregn automatisk hvis ingen change_address: Valgfri[str] = Ingen # Vælg automatisk hvis Ingen klasse OutputSpec(BaseModel): modtager: str beløb: int memo: Valgfrit[str] = Ingen @app.get("/API/v1/transaktioner", response_model=Liste[Transaktionsoversigt]) asynkron def liste_transaktioner( wallet_id: str = Depends(get_authenticated_wallet), spring: int = 0, limit: int = 20, status: Optional[str] = Ingen ): """ Listetransaktioner (kun oversigter) Signaturer er udelukket fra listesvar for at reducere nyttelast. Brug GET /transactions/{tx_id} til fuld transaktion med signaturer. """ txs = vente transaction_service.list_transactions( wallet_id, skip=spring, limit=limit, status=status ) returnere [tx.to_summary() for tx in txs] @app.get("/API/v1/transactions/{tx_id}", response_model=Transaktion fuld) asynkron def get_transaction(tx_id: str, wallet_id: str = Depends(get_authenticated_wallet), include_signatures: bool = True): """ Få alle transaktionsoplysninger Indstil include_signatures=false for at reducere svarstørrelsen, hvis du kun har brug for transaktionsmetadata. """ tx = vente transaction_service.get_transaction(wallet_id, tx_id) hvis ikke tx: hæve HTTPException(status_kode=404, detalje="Transaktion ikke fundet") returnere tx.to_full_schema(include_signatures=include_signatures) @app.post("/API/v1/transactions/build") asynkron def build_transaction(anmodning: BuildTransactionRequest, wallet_id: str = Depends(get_authenticated_wallet) ): """ Byg usigneret transaktion Returnerer transaktionsdata klar til signering på klientsiden. Signering sker på klienten for at holde private nøgler væk fra serveren. """ unsigned_tx = vente transaction_service.build_transaction( wallet_id, outputs=request.outputs, fee_rate=request.fee_rate, change_address=request.change_address ) returnere { "unsigned_tx": base64.b64encode(unsigned_tx.serialize_for_signing()).decode(), "signeringsmeddelelse": base64.b64encode(unsigned_tx.tx_hash()).decode(), "input_to_sign": [ { "indeks": i, "adresse": inp.adresse, "beløb": inp.beløb, "afledningssti": inp.sti } for jeg, inp in enumerate(unsigned_tx.inputs) ], "estimeret_gebyr": unsigned_tx.fee, "estimeret_størrelse": unsigned_tx.estimated_size() }

# Autentificeringsflow ved hjælp af Kyber + SPHINCS+ importere oks importere hashlib importere hmac fra datotid importere datetime, timedelta klasse QuantumSafeAuth: """ Kvantesikker API-godkendelsesflow: 1. Klient sender Kyber offentlig nøgle 2. Server indkapsler sessionsnøgle 3. Klient dekapsler for at få sessionsnøgle 4. Anmodninger signeret med HMAC ved hjælp af sessionsnøgle """ def __init__(selv): self.session_store = {} # Brug Redis i produktionen self.session_duration = timedelta(timer=24) asynkron def initiate_session( self, wallet_id: str, client_kyber_pk: bytes ) -> dict: """ Trin 1: Klient starter session med Kyber public key Server indkapsler en sessionshemmelighed til klientens nøgle """ kem = oqs.KeyEncapsulation("Kyber768") ciphertext, shared_secret = kem.encap_secret(client_kyber_pk) # Udled sessionsnøgle fra delt hemmelighed session_key = hashlib.shake_256( shared_secret + b"session-nøgle" ).digest(32) # Opret sessions-id session_id = hashlib.Blake2b( shared_secret + str(datetime.utcnow()).encode(), digest_size=16 ).hexdigest() # Butikssession (server-side) self.session_store[session_id] = { "wallet_id": wallet_id, "session_key": session_key, "udløber_kl": datetime.utcnow() + self.session_duration, "created_at": datetime.utcnow() } returnere { "session_id": session_id, "chiffertekst": base64.b64encode(ciphertext).decode(), "udløber_kl": (datetime.utcnow() + self.session_duration).isoformat() } def verify_request( self, session_id: str, request_signature: bytes, request_data: bytes, timestamp: int ) -> Valgfri[str]: """ Bekræft anmodningssignatur ved hjælp af sessionsnøgle Returnerer wallet_id hvis gyldigt, ingen ellers """ session = self.session_store.get(session_id) hvis ikke session: returnere Ingen # Tjek udløb if datetime.utcnow() > session["udløber_kl"]: del self.session_store[session_id] returnere Ingen # Tjek tidsstempel (forhindrer genafspilning) request_time = datetime.fromtimestamp(timestamp) if abs((datetime.utcnow() - request_time).total_seconds()) > 300: returnere Ingen # Mere end 5 minutter gammel/fremtidig # Bekræft HMAC-signatur forventet_sig = hmac.new( session["session_key"], request_data + str(timestamp).encode(), hashlib.Blake2b ).digest() if hmac.compare_digest(request_signature, forventet_sig): returnere session["wallet_id"] returnere Ingen # FastAPI-afhængighed for autentificerede ruter auth_service = QuantumSafeAuth() asynkron def get_authenticated_wallet( x_session_id: str = Header(...), x_signature: str = Header(...), x_timestamp: str = Header(...), request: Request = None ) -> str: """Afhængighed, der validerer kvantesikker godkendelse""" krop = vente 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) ) hvis ikke wallet_id: hæve HTTPException(status_kode=401, detalje="Ugyldig godkendelse") returnere wallet_id

# WebSocket-implementering til opdateringer i realtid fra fastapi importere WebSocket, WebSocketDisconnect importere json importere asyncio klasse Connection Manager: """Administrer WebSocket-forbindelser pr. tegnebog""" def __init__(selv): self.active_connections: dict[str, Liste[WebSocket]] = {} asynkron def forbinde(selv, websocket: WebSocket, wallet_id: str): vente websocket.accept() if wallet_id ikke i self.active_connections: self.active_connections[wallet_id] = [] self.active_connections[wallet_id].append(websocket) def afbryde(selv, websocket: WebSocket, wallet_id: str): if wallet_id in self.active_connections: self.active_connections[wallet_id].remove(websocket) asynkron def broadcast_to_wallet(self, wallet_id: str, message: dict): if wallet_id in self.active_connections: dead_connections = [] for forbindelse in self.active_connections[wallet_id]: prøve: vente forbindelse.send_json(besked) undtagen: dead_connections.append(forbindelse) # Ryd op i døde forbindelser for forb in dead_connections: self.active_connections[wallet_id].remove(conn) manager = Connection Manager() @app.websocket("/ws/{wallet_id}") asynkron def websocket_endpoint(websocket: WebSocket, wallet_id: str): """ WebSocket til realtids wallet-opdateringer Hændelser: - balance_update: Saldo ændret - transaktion_modtaget: Indgående transaktion - transaktion_bekræftet: TX nået bekræftelser - transaktion_sent: Udgående TX-udsendelse """ # Godkend WebSocket-forbindelse auth_token = websocket.query_params.get("token") hvis ikke afvent validate_ws_token(auth_token, wallet_id): vente websocket.close(kode=4001) returnere vente manager.connect(websocket, wallet_id) prøve: # Send starttilstand vente websocket.send_json({ "type": "forbundet", "wallet_id": wallet_id, "tidsstempel": datetime.utcnow().isoformat() }) # Håndter indgående beskeder (abonnementer, pings) mens Sandt: data = vente websocket.receive_json() if data.get("type") == "ping": vente websocket.send_json({"type": "pong"}) elif data.get("type") == "abonner": # Abonner på specifikke adresser adresser = data.get("adresser", []) vente subscription_service.subscribe( wallet_id, addresses) undtagen WebSocketDisconnect: manager.disconnect(websocket, wallet_id) # Eventudsendelse (kaldet af blockchain-monitor) asynkron def broadcast_balance_update(wallet_id: str, adresse: str, new_balance: int): vente manager.broadcast_to_wallet(wallet_id, { "type": "balance_update", "adresse": adresse, "balance": ny_balance, "tidsstempel": datetime.utcnow().isoformat() }) asynkron def broadcast_transaction_received(wallet_id: str, tx_summary: dict): vente manager.broadcast_to_wallet(wallet_id, { "type": "transaktion_modtaget", "transaktion": tx_summary, # Kun resumé, ikke fuld signatur "tidsstempel": datetime.utcnow().isoformat() })

# Aktiver gzip-komprimering i FastAPI fra fastapi.middleware.gzip importere GZipMiddleware app.add_middleware(GZipMiddleware, minimum_size=1000) # Valgfrit: MessagePack-svar til mobile/indlejrede klienter fra fastapi.svar importere Svar importere msgpack klasse MsgPackResponse(Svar): media_type = "applikation/msgpack" def gengive(selv, indhold) -> bytes: returnere msgpack.packb(content, use_bin_type=True) @app.get("/API/v1/transactions/{tx_id}/binary") asynkron def get_transaction_binary(tx_id: str): """Hent transaktion i MessagePack-format (mindre end JSON)""" tx = vente transaktionstjeneste.get_transaktion(tx_id) returnere MsgPackResponse(content=tx.to_dict())

# Satsbegrænsning for tegnebog API fra slowapi importere Limiter, _rate_limit_exceeded_handler fra slowapi.fejl importere RateLimitExceeded limiter = Limiter(key_func=get_wallet_id_from_request) app.state.limiter = limiter app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler) # Forskellige grænser for forskellige operationer RATE_LIMITS = { "læse": "100/minut", # Saldokontrol, tx-lister "skrive": "20/minut", # Adresseafledning "udsende": "5/minut", # Transaktionsudsendelse } @app.get("/API/v1/balance") @limiter.limit("100/minut") asynkron def få_balance(anmodning: Anmodning): ... @app.post("/API/v1/transaktioner/broadcast") @limiter.limit("5/minut") asynkron def broadcast_transaction(anmodning: Anmodning): """Strengere grænse for udsendelse for at forhindre spam""" ...

# Standardiserede fejlsvar fra enum importere Enum klasse Fejlkode(str, Enum): INVALID_ADDRESS = "INVALID_ADRESSE" INSUFFICIENT_BALANCE = "INSUFFICIENT_BALANCE" INVALID_SIGNATURE = "INVALID_SIGNATURE" TRANSACTION_REJECTED = "TRANSACTION_REJECTED" RATE_LIMITED = "RATE_LIMITED" SESSION_EXPIRED = "SESSION_EXPIRED" DERIVATION_FAILED = "DERIVATION_FAILED" klasse API-fejl(BaseModel): kode: Fejlkode besked: str detaljer: Valgfri[dict] = Ingen @app.exception_handler(InsufficientBalanceError) asynkron def insufficient_balance_handler(anmodning, ekskl): returnere JSONResponse( status_code=400, content=API-fejl(kode=Fejlkode.INSUFFICIENT_BALANCE, besked="Utilstrækkelig saldo til transaktion", detaljer={ "tilgængelig": ekskl. tilgængelig, "påkrævet": exc.påkrævet, "mangel": exc.required - exc.available } ).dict() )