Post-Quantum Wallet API Design: REST- och WebSocket-mönster

# Implementering av adressändpunkter (FastAPI) från fastapi importera FastAPI, Depends, HTTPException från pydantisk importera Basmodell från skriver importera Lista, valfritt importera base64 app = FastAPI(titel="SynX Plånbok API", version="1.0.0") klass Adresssvar(Basmodell): """Adress med post-quantum offentliga nycklar""" adress: str sökväg: str kyber_public_key: str # Base64-kodad (1 184 byte) sphincs_public_key: str # Base64-kodad (32 byte) saldo: int pending_balance: int create_at: str klass DeriveAddressRequest(Basmodell): konto: int = 0 ändring: int = 0 index: Valfritt[int] = Ingen # Öka automatiskt om ingen @app.get("/API/v1/adresser", response_model=Lista[Adresssvar]) asynkron def list_adresser( wallet_id: str = Depends(get_authenticated_wallet), hoppa över: int = 0, limit: int = 50 ): """ Lista plånboksadresser med saldon Obs: Kyber publika nycklar är stora (1,2KB). För listning, överväg att utesluta nycklar och hämta separat. """ adresser = invänta address_service.list_addresses( wallet_id, skip=hopp, limit=limit ) återvända [ Adresssvar( 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. create_at=addr.created_at.isoformat() ) för adr in adresser ] @app.post("/API/v1/adresser/derive", response_model=Adresssvar) asynkron def härleda_adress(förfrågan: DeriveAddressRequest, wallet_id: str = Depends(get_authenticated_wallet) ): """Härled ny adress vid specificerad härledningsväg""" addr = invänta address_service.derive_address( wallet_id, account=request.account, change=request.change, index=request.index ) återvända Adresssvar(...)

klass Transaktionsöversikt(Basmodell): """Transaktion utan fullständig signaturdata (för listor)""" tx_id: str tidsstämpel: str inputs_count: int outputs_count: int summa: int avgift: int bekräftelser: int status: str # "väntande", "bekräftad", "misslyckades" klass TransaktionFull(Basmodell): """Fullständig transaktion inklusive signaturer""" tx_id: str version: int tidsstämpel: str ingångar: List['TransactionInputSchema'] utgångar: List['TransactionOutputSchema'] avgift: int bekräftelser: int block_hash: Valfritt[str] raw_hex: str # Full serialiserad transaktion klass TransactionInputSchema(BaseModel): prev_tx_id: str prev_output_index: int mängd: int adress: str signatur: str # Base64 (~10,5KB för SPHINCS+-SHAKE-128s, 7 856 råbyte) public_key: str # Base64 (44 byte för SPHINCS+) klass BuildTransactionRequest(BaseModel): utgångar: List['OutputSpec'] fee_rate: Valfritt[int] = Ingen # Beräkna automatiskt om ingen change_address: Valfritt[str] = Ingen # Välj automatiskt om Ingen klass OutputSpec(BaseModel): mottagare: str belopp: int memo: Valfritt[str] = Ingen @app.get("/API/v1/transaktioner", response_model=Lista[Transaktionsöversikt]) asynkron def list_transactions( wallet_id: str = Depends(get_authenticated_wallet), skip: int = 0, limit: int = 20, status: Optional[str] = None ): """ Listtransaktioner (endast sammanfattningar) Signaturer exkluderas från listsvar för att minska nyttolasten. Använd GET /transactions/{tx_id} för fullständig transaktion med signaturer. """ txs = invänta transaction_service.list_transactions( wallet_id, skip=hopp, limit=limit, status=status ) återvända [tx.to_summary() för tx in txs] @app.get("/API/v1/transactions/{tx_id}", response_model=TransaktionFull) asynkron def get_transaction( tx_id: str, wallet_id: str = Depends(get_authenticated_wallet), include_signatures: bool = True ): """ Få fullständig transaktionsinformation Ange include_signatures=false för att minska svarsstorleken om du bara behöver transaktionsmetadata. """ tx = invänta transaction_service.get_transaction(wallet_id, tx_id) om inte tx: höja HTTPException(status_kod=404, detalj="Transaktionen hittades inte") återvända tx.to_full_schema(include_signatures=include_signatures) @app.post("/API/v1/transactions/build") asynkron def build_transaction(förfrågan: BuildTransactionRequest, wallet_id: str = Depends(get_authenticated_wallet) ): """ Bygg osignerad transaktion Returnerar transaktionsdata redo för signering på klientsidan. Signering sker på klienten för att hålla privata nycklar från servern. """ unsigned_tx = invänta transaction_service.build_transaction( wallet_id, outputs=request.outputs, fee_rate=request.fee_rate, change_address=request.change_address ) återvända { "unsigned_tx": base64.b64encode(unsigned_tx.serialize_for_signing()).decode(), "signeringsmeddelande": base64.b64encode(unsigned_tx.tx_hash()).decode(), "inputs_to_sign": [ { "index": i, "adress": inp.adress, "belopp": inp.amount, "avledningsväg": inp.sökväg } för jag, inp in enumerate(unsigned_tx.inputs) ], "uppskattad_avgift": unsigned_tx.fee, "uppskattad_storlek": unsigned_tx.estimated_size() }

# Autentiseringsflöde med Kyber + SPHINCS+ importera oqs importera hashlib importera hmac från datumtid importera datetime, timedelta klass QuantumSafeAuth: """ Kvantsäkert API-autentiseringsflöde: 1. Klienten skickar Kyber offentlig nyckel 2. Servern kapslar in sessionsnyckeln 3. Klienten dekapslar för att få sessionsnyckeln 4. Begäranden signerade med HMAC med sessionsnyckeln """ def __init__(self): self.session_store = {} # I produktionen, använd Redis self.session_duration = timedelta(timmar=24) asynkron def initiate_session( self, wallet_id: str, client_kyber_pk: bytes ) -> dict: """ Steg 1: Klient initierar session med Kyber public key Server kapslar in en sessionshemlighet för klientens nyckel """ kem = oqs.KeyEncapsulation("Kyber768") chiffertext, shared_secret = kem.encap_secret(client_kyber_pk) # Härled sessionsnyckeln från delad hemlighet session_key = hashlib.shake_256( shared_secret + b"session-nyckel" ).digest(32) # Skapa sessions-ID session_id = hashlib.Blake2b( shared_secret + str(datetime.utcnow()).encode(), digest_size=16 ).hexdigest() # Butikssession (serversidan) self.session_store[session_id] = { "wallet_id": wallet_id, "session_key": session_key, "expires_at": datetime.utcnow() + self.session_duration, "skapad_vid": datetime.utcnow() } återvända { "session_id": session_id, "chiffertext": 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 ) -> Valfritt[str]: """ Verifiera begäransignaturen med sessionsnyckel Returnerar wallet_id om giltigt, inget annat """ session = self.session_store.get(session_id) om inte session: återvända Ingen # Kontrollera utgångsdatum if datetime.utcnow() > session["expires_at"]: del self.session_store[session_id] återvända Ingen # Kontrollera tidsstämpel (förhindra uppspelning) request_time = datetime.fromtimestamp(timestamp) if abs((datetime.utcnow() - request_time).total_seconds()) > 300: återvända Ingen # Mer än 5 minuter gammal/framtida # Verifiera HMAC-signaturen förväntat_sig = hmac.new( session["session_key"], request_data + str(timestamp).encode(), hashlib.Blake2b ).digest() if hmac.compare_digest(request_signature, expected_sig): återvända session["wallet_id"] återvända Ingen # FastAPI-beroende för autentiserade rutter 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: """Beroende som validerar kvantsäker autentisering""" kropp = invänta 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) ) om inte wallet_id: höja HTTPException(status_kod=401, detalj="Ogiltig autentisering") återvända wallet_id

# WebSocket-implementering för realtidsuppdateringar från fastapi importera WebSocket, WebSocketDisconnect importera json importera asyncio klass Connection Manager: """Hantera WebSocket-anslutningar per plånbok""" def __init__(self): self.active_connections: dict[str, List[WebSocket]] = {} asynkron def ansluta(self, websocket: WebSocket, wallet_id: str): invänta websocket.accept() if wallet_id inte in self.active_connections: self.active_connections[wallet_id] = [] self.active_connections[wallet_id].append(websocket) def koppla ur(self, 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, meddelande: dict): if wallet_id in self.active_connections: dead_connections = [] för förbindelse in self.active_connections[wallet_id]: försök: invänta connection.send_json(meddelande) utom: dead_connections.append(anslutning) # Rensa upp döda anslutningar för anslutning 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 för uppdateringar av plånbok i realtid Händelser: - saldo_uppdatering: saldo ändrad - transaktion_mottagen: Inkommande transaktion - transaktion_bekräftad: TX nådde bekräftelser - transaktion_sänt: Utgående sändning """ # Autentisera WebSocket-anslutning auth_token = websocket.query_params.get("tecken") om inte vänta validate_ws_token(auth_token, wallet_id): invänta websocket.close(kod=4001) återvända invänta manager.connect(websocket, wallet_id) försök: # Skicka initialtillstånd invänta websocket.send_json({ "typ": "ansluten", "wallet_id": wallet_id, "tidsstämpel": datetime.utcnow().isoformat() }) # Hantera inkommande meddelanden (prenumerationer, pingar) medan Sant: data = invänta websocket.receive_json() if data.get("typ") == "pinga": invänta websocket.send_json({"typ": "pong"}) elif data.get("typ") == "prenumerera": # Prenumerera på specifika adresser adresser = data.get("adresser", []) invänta subscription_service.subscribe( wallet_id, addresses) utom WebSocketDisconnect: manager.disconnect(websocket, wallet_id) # Event sändning (kallas av blockchain monitor) asynkron def broadcast_balance_update(plånboks-id: str, adress: str, ny_balans: int): invänta manager.broadcast_to_wallet(wallet_id, { "typ": "balans_uppdatering", "adress": adress, "balans": new_balance, "tidsstämpel": datetime.utcnow().isoformat() }) asynkron def broadcast_transaction_received(wallet_id: str, tx_summary: dict): invänta manager.broadcast_to_wallet(wallet_id, { "typ": "transaktion_mottagen", "transaktion": tx_summary, # Endast sammanfattning, inte fullständig signatur "tidsstämpel": datetime.utcnow().isoformat() })

# Aktivera gzip-komprimering i FastAPI från fastapi.middleware.gzip importera GZipMiddleware app.add_middleware(GZipMiddleware, minimum_size=1000) # Valfritt: MessagePack-svar för mobila/inbäddade klienter från fastapi.responses importera Svar importera msgpack klass MsgPackResponse(Svar): media_type = "applikation/msgpack" def göra(själv, innehåll) -> bytes: återvända msgpack.packb(content, use_bin_type=True) @app.get("/API/v1/transactions/{tx_id}/binary") asynkron def get_transaction_binary(tx_id: str): """Hämta transaktion i MessagePack-format (mindre än JSON)""" tx = invänta transaktionstjänst.get_transaktion(tx_id) återvända MsgPackResponse(content=tx.to_dict())

# Prisbegränsning för plånbok API från slowapi importera Limiter, _rate_limit_exceeded_handler från slowapi.errors importera RateLimitExceeded limiter = Limiter(key_func=get_wallet_id_from_request) app.state.limiter = limiter app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler) # Olika gränser för olika operationer RATE_LIMITS = { "läsa": "100/minut", # Balanskontroller, tx-listor "skriva": "20/minut", # Adressavledning "utsända": "5/minut", # Transaktionssändning } @app.get("/API/v1/balans") @limiter.limit("100/minut") asynkron def få_balans(begäran: Begäran): ... @app.post("/API/v1/transaktioner/broadcast") @limiter.limit("5/minut") asynkron def broadcast_transaction(request: Request): """Strängare gräns för sändning för att förhindra spam""" ...

# Standardiserade felsvar från uppräkning importera Enum klass Felkod(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" klass API-fel(Basmodell): kod: Felkod meddelande: str detaljer: Valfritt[dict] = Ingen @app.exception_handler(InsufficientBalanceError) asynkron def insufficient_balance_handler(förfrågan, ex): återvända JSONResponse( status_code=400, content=API-fel(kod=Felkod.INSUFFICIENT_BALANCE, meddelande="Otillräckligt saldo för transaktionen", detaljer={ "tillgänglig": ex.tillgänglig, "nödvändig": ex.required, "underskott": exc.required - exc.available } ).dict() )