Projekt portfela Post-Quantum API: wzorce REST i WebSocket

# Implementacja adresów końcowych (FastAPI) z szybko import FastAPI, zależy, wyjątek HTTP z pydantyczny import Model podstawowy z pisanie na maszynie import Lista, opcjonalna import aplikacja base64 = FastAPI(tytuł=„SynX Portfel API”, wersja="1.0.0") klasa AdresOdpowiedź(Model podstawowy): """Adres z postkwantowymi kluczami publicznymi""" adres: str ścieżka: str kyber_public_key: str # Zakodowane w Base64 (1184 bajty) sphincs_public_key: str # Kodowanie Base64 (32 bajty) saldo: int pending_balance: int utworzony_at: str klasa Żądanie adresu pochodnego(BaseModel): konto: int = 0 zmiana: int = 0 indeks: Opcjonalne[int] = Brak # Automatyczna inkrementacja, jeśli brak @app.get(„/API/v1/adresy”, model_odpowiedzi=Lista[AdresOdpowiedź]) asynchroniczna def lista_adresów( identyfikator_portfela: str = Zależy(get_authenticated_wallet), pomiń: int = 0, limit: int = 50 ): """ Wyświetl adresy portfeli z saldami Uwaga: klucze publiczne Kyber są duże (1,2 KB). W przypadku wystawiania na liście rozważ wykluczenie kluczy i pobranie ich osobno. """ adresy = czekać na adres_usługi.list_adresy(id_portfela, pomiń=pomiń, limit=limit ) powrót [ AdresOdpowiedź( adres=addr.adres, ścieżka=addr.ścieżka, kyber_public_key=base64.b64encode(addr.kyber_pk).decode(), sphincs_public_key=base64.b64encode(addr.sphincs_pk).decode(), saldo=addr.balance, pending_balance=addr.pending_balance, create_at=addr.created_at.isoformat() ) Do adres in adresy] @aplikacja.post(„/API/v1/adresy/wyprowadzenie”, model_odpowiedzi=AdresOdpowiedź) asynchroniczna def adres_pochodny( wniosek: Żądanie adresu pochodnego, identyfikator_portfela: str = Zależy(get_authenticated_wallet)): """Utwórz nowy adres w określonej ścieżce wyprowadzenia""" adres = czekać na adres_usługi.derive_address(identyfikator_portfela, konto=żądanie.konto, zmiana=żądanie.zmiana, indeks=żądanie.indeks ) powrót AdresOdpowiedź(...)

klasa Podsumowanie transakcji(Model podstawowy): """Transakcja bez pełnych danych podpisu (dla list)""" tx_id: str znacznik czasu: str liczba_wejść: int liczba_wyjść: int kwota: int opłata: int potwierdzenia: int status: str # „oczekuje”, „potwierdzono”, „nie powiodło się” klasa Transakcja pełna(Model podstawowy): """Pełna transakcja łącznie z podpisami""" tx_id: str wersja: int znacznik czasu: str wejścia: Lista [„Schemat wejściowy transakcji”] wyjścia: Lista[„Schemat wyjściowy transakcji”] opłata: int potwierdzenia: int block_hash: Opcjonalne[str] raw_hex: str # Pełna serializowana transakcja klasa Schemat wejściowy transakcji(BaseModel): prev_tx_id: str prev_output_index: int ilość: int adres: str podpis: str # Base64 (~10,5 KB dla SPHINCS+-SHAKE-128s, 7856 surowych bajtów) klucz_publiczny: str # Base64 (44 bajty dla SPHINCS+) klasa Zbuduj żądanie transakcji(BaseModel): wyjścia: Lista[„Specyfikacja wyjściowa”] stawka_opłaty: Opcjonalne[int] = Brak # Oblicz automatycznie, jeśli brak zmiana_adresu: Opcjonalne[str] = Brak # Wybierz automatycznie, jeśli Brak klasa Specyfikacja wyjściowa(BaseModel): odbiorca: str kwota: int memo: Opcjonalne[str] = Brak @app.get(„/API/v1/transakcje”, model_odpowiedzi=Lista[Podsumowanie transakcji]) asynchroniczna def lista_transakcji( identyfikator_portfela: str = Zależy(get_authenticated_wallet), pomiń: int = 0, limit: int = 20, status: Opcjonalny[str] = Brak ): """ Lista transakcji (tylko podsumowania) Podpisy są wykluczone z odpowiedzi na listy, aby zmniejszyć ładunek. Użyj GET /transactions/{tx_id}, aby uzyskać pełną transakcję z podpisami. """ txs = czekać na transakcja_usługa.list_transactions(id_portfela, pomiń=pomiń, limit=limit, status=status) powrót [tx.to_summary() Do tx in TX] @app.get(„/API/v1/transakcje/{tx_id}”, model_odpowiedzi=Transakcja pełna) asynchroniczna def pobierz_transakcję( tx_id: str, portfel_id: str = zależy(get_authenticated_wallet), include_signatures: bool = True ): """ Uzyskaj pełne szczegóły transakcji Ustaw include_signatures=false, aby zmniejszyć rozmiar odpowiedzi, jeśli potrzebujesz tylko metadanych transakcji. """ tx = czekać na transakcja_usługa.get_transakcja(id_portfela, identyfikator tx) w przeciwnym razie tx: wznosić Wyjątek HTTP(kod_stanu=404, szczegół=„Nie znaleziono transakcji”) powrót tx.to_full_schema(include_signatures=include_signatures) @aplikacja.post(„/API/v1/transakcje/kompilacja”) asynchroniczna def transakcja_kompilacyjna( wniosek: Zbuduj żądanie transakcji, identyfikator_portfela: str = Zależy(get_authenticated_wallet)): """ Kompiluj niepodpisaną transakcję. Zwraca dane transakcji gotowe do podpisania po stronie klienta. Podpisywanie odbywa się na kliencie, aby klucze prywatne nie znajdowały się poza serwerem. """ bez znaku_tx = czekać na transakcja_usługa.build_transaction(identyfikator_portfela, wyjścia=request.outputs, opłata_rate=request.fee_rate, zmiana_adresu=request.change_address ) powrót { „bez znaku_tx”: base64.b64encode(unsigned_tx.serialize_for_signing()).decode(), „wiadomość_podpisywania”: base64.b64encode(unsigned_tx.tx_hash()).decode(), "wejścia_do_podpisu": [ { "indeks": i, "adres":adres wejściowy, "kwota": kwota wejściowa, „ścieżka_wyprowadzenia”: ścieżka.inp } Do ja, inp in enumerate(unsigned_tx.inputs) ], „szacowana_opłata”: unsigned_tx.fee, „szacowany_rozmiar”: unsigned_tx.estimated_size() }

# Przebieg uwierzytelniania przy użyciu Kyber + SPHINCS+ import OK import hashlib import hmac z datagodzina import data/godzina, delta czasu klasa QuantumSafeAuth: „”” Quantum-safe uwierzytelnianie API Przebieg: 1. Klient wysyła klucz publiczny Kyber 2. Serwer hermetyzuje klucz sesji 3. Klient dekapsuluje, aby uzyskać klucz sesji 4. Żądania podpisane za pomocą HMAC przy użyciu klucza sesji „”” def __inicj__(self): self.session_store = {} # W środowisku produkcyjnym użyj Redis self.session_duration = timedelta(godziny=24) asynchroniczna def inicjacja_sesja( self, identyfikator_portfela: str, klient_kyber_pk: bajty ) -> dykt: """ Krok 1: Klient inicjuje sesję z kluczem publicznym Kyber. Serwer hermetyzuje sekret sesji w kluczu klienta """ kem = oqs.KeyEncapsulation(„Kyber768”) szyfrogram, wspólna_secret = kem.encap_secret(client_kyber_pk) # Uzyskaj klucz sesji ze wspólnego sekretu session_key = hashlib.shake_256(share_secret + b„klucz sesji” ).streszczenie(32) # Utwórz identyfikator sesji identyfikator_sesji = hashlib.Blake2b(shared_secret + str(datetime.utcnow()).encode(), rozmiar_rozmiaru = 16 ).hexdigest() # Sesja sklepu (po stronie serwera) self.session_store[session_id] = { „id_portfela”: identyfikator_portfela, „klucz_sesji”: klucz_sesji, „wygasa o”: datetime.utcnow() + self.session_duration, „utworzono_w”: datetime.utcnow() } powrót { „id_sesji”: identyfikator sesji, „zaszyfrowany tekst”: base64.b64encode(tekst zaszyfrowany).decode(), „wygasa o”: (datetime.utcnow() + self.session_duration).isoformat() } def zweryfikować_prośbę( self, session_id: str, request_signature: bytes, request_data: bytes, timestamp: int ) -> Opcjonalne[str]: """ Zweryfikuj podpis żądania przy użyciu klucza sesji. Zwraca identyfikator portfela, jeśli jest prawidłowy, Brak w przeciwnym razie """ sesja = self.session_store.get(id_sesji) w przeciwnym razie sesja: powrót Nic # Sprawdź datę ważności if datetime.utcnow() > sesja[„wygasa o”]: del self.session_store[session_id] powrót Nic # Sprawdź znacznik czasu (zapobiegaj powtarzaniu) request_time = datetime.fromtimestamp(timestamp) if abs((data.utcnow() - czas_żądania).total_sekundy()) > 300: powrót Nic # Ponad 5 minut starych/przyszłych # Sprawdź podpis HMAC oczekiwany_sig = hmac.new(sesja[„klucz_sesji”], dane_żądania + str(znacznik czasu).encode(), hashlib.Blake2b .digest() if hmac.compare_digest(podpis_żądania, oczekiwany_sig): powrót sesja[„id_portfela”] powrót Nic # Zależność FastAPI dla uwierzytelnionych tras usługa_auta = QuantumSafeAuth() asynchroniczna def get_authenticated_wallet( x_session_id: str = Nagłówek(...), x_signature: str = Nagłówek(...), x_timestamp: str = Nagłówek(...), żądanie: Żądanie = Brak ) -> str: """Zależność weryfikująca uwierzytelnianie bezpieczne kwantowo""" ciało = czekać na request.body() portfel_id = auth_service.verify_request( session_id=x_session_id, request_signature=base64.b64decode(x_signature), request_data=body, timestamp=int(x_timestamp) ) w przeciwnym razie identyfikator_portfela: wznosić Wyjątek HTTP(kod_stanu=401, szczegół=„Nieprawidłowe uwierzytelnienie”) powrót identyfikator_portfela

# Implementacja WebSocket do aktualizacji w czasie rzeczywistym z szybko import WebSocket, WebSocketRozłącz import json import asyncio klasa Menedżer połączeń: """Zarządzaj połączeniami WebSocket na portfel""" def __inicj__(self): self.active_connections: dict[str, List[WebSocket]] = {} asynchroniczna def łączyć(self, websocket: WebSocket, portfel_id: str): czekać na websocket.accept() if identyfikator_portfela nie w self.active_connections: self.active_connections[wallet_id] = [] self.active_connections[wallet_id].append(websocket) def odłączyć(self, websocket: WebSocket, portfel_id: str): if identyfikator_portfela in self.active_connections: self.active_connections[wallet_id].remove(websocket) asynchroniczna def transmisja_do_portfela(self, identyfikator_portfela: str, wiadomość: dykt): if identyfikator_portfela in self.active_connections: dead_connections = [] Do połączenie in self.active_connections[wallet_id]: próbować: czekać na połączenie.send_json(wiadomość) z wyjątkiem: dead_connections.append(połączenie) # Oczyść martwe połączenia Do połączenie in dead_connections: self.active_connections[wallet_id].remove(conn) manager = Menedżer połączeń() @app.websocket(„/ws/{id_portfela}”) asynchroniczna def punkt końcowy_websocket(websocket: WebSocket, identyfikator_portfela: str): """ WebSocket do aktualizacji portfela w czasie rzeczywistym Zdarzenia: - Balance_update: Saldo zmienione - transakcja_received: Transakcja przychodząca - transakcja_confirmed: TX osiągnęła potwierdzenia - transakcja_sent: Wychodząca transmisja TX """ # Uwierzytelnij połączenie WebSocket auth_token = websocket.query_params.get("znak") jeśli nie, poczekaj valid_ws_token(token_auth, identyfikator_portfela): czekać na websocket.close(code=4001) powrót czekać na manager.connect(websocket, identyfikator_portfela) próbować: # Wyślij stan początkowy czekać na websocket.send_json({ "typ": "połączony", „id_portfela”: identyfikator_portfela, „znacznik czasu”: datetime.utcnow().isoformat() }) # Obsługuj wiadomości przychodzące (subskrypcje, pingi) chwila Prawda: dane = czekać na websocket.receive_json() if dane.get("typ") == "świst": czekać na websocket.send_json({"typ": „pong”}) Elif dane.get("typ") == "subskrybować": # Subskrybuj określone adresy adresy = dane.get("adresy", []) czekać na usługa_subskrypcji.subskrybuj(identyfikator_portfela, adresy) z wyjątkiem WebSocketDisconnect: manager.disconnect(websocket, identyfikator_portfela) # Transmisja zdarzeń (wywoływana przez monitor blockchain) asynchroniczna def transmisja_balance_update(id_portfela: str, adres: str, nowe saldo: int): czekać na manager.broadcast_to_wallet(walet_id, { "typ": „aktualizacja_bilansu”, "adres": adres, "balansować": nowe_bilanse, „znacznik czasu”: datetime.utcnow().isoformat() }) asynchroniczna def transmisja_transakcja_odebrana(wallet_id: str, tx_summary: dict): czekać na manager.broadcast_to_wallet(walet_id, { "typ": „transakcja_otrzymana”, "transakcja": podsumowanie_tx, # Tylko podsumowanie, a nie pełny podpis „znacznik czasu”: datetime.utcnow().isoformat() })

# Włącz kompresję gzip w FastAPI z fastapi.middleware.gzip import Aplikacja GZipMiddleware.add_middleware(GZipMiddleware, minimalny rozmiar_rozmiar=1000) # Opcjonalnie: odpowiedzi MessagePack dla klientów mobilnych/wbudowanych z fastapi.response import Odpowiedź import msgpack klasa Odpowiedź MsgPack(Odpowiedź): typ_mediau = „aplikacja/msgpack” def oddać(self, content) -> bajty: powrót msgpack.packb(treść, use_bin_type=True) @app.get(„/API/v1/transakcje/{tx_id}/binary”) asynchroniczna def get_transaction_binary(tx_id: str): """Pobierz transakcję w formacie MessagePack (mniejszym niż JSON)""" tx = czekać na transakcja_usługa.get_transakcja(tx_id) powrót Odpowiedź MsgPack(treść=tx.to_dict())

# Ograniczenie stawki dla portfela API z slowapi import Limiter, _rate_limit_exceeded_handler z 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) # Różne limity dla różnych operacji STAWKA_LIMITS = { "Czytać": „100/minutę”, # Kontrole salda, listy przesyłek "pisać": „20/minutę”, # Wyprowadzenie adresu "audycja": „5/minutę”, # Transmisja transakcji } @app.get(„/API/v1/saldo”) @limiter.limit(„100/minutę”) asynchroniczna def get_balance(prośba: prośba): ... @aplikacja.post(„/API/v1/transakcje/transmisja”) @limiter.limit(„5/minutę”) asynchroniczna def transmisja_transakcja(prośba: prośba): """Bardziej rygorystyczny limit transmisji w celu zapobiegania spamowi""" ...

# Standaryzowane reakcje na błędy z wyliczenie import Wyliczenie klasa Kod błędu(str, Enum): INVALID_ADDRESS = „INVALID_ADDRESS” INSUFFICIENT_BALANCE = „INSUFFICIENT_BALANCE” INVALID_SIGNATURE = „INVALID_SIGNATURE” TRANSAKCJA_ODRZUCONA = „TRANSACTION_ODRZUCONA” RATE_LIMITED = „RATE_LIMITED” SESJA_EXPIRED = „SESJA_WAŻNA” DERIATION_FAILED = „DERIVATION_FAILED” klasa APIEbłąd(Model bazowy): kod: Kod błędu wiadomość: str szczegóły: Opcjonalne[dykt] = Brak @app.exception_handler(Błąd niewystarczającego salda) asynchroniczna def niewystarczający_balance_handler(prośba, wym): powrót Odpowiedź JSON(kod_stanu=400, treść=APIEbłąd( kod=Kod błędu.INSUFFICIENT_BALANCE, wiadomość=„Niewystarczające saldo dla transakcji”, szczegóły={ "dostępny": ex.dostępne, "wymagany": ex.wymagane, „niedobór”: exc.required - exc.available } ).dict() )