Kuantum Sonrası Cüzdan API Tasarımı: REST ve WebSocket Kalıpları

# Adres uç noktaları uygulaması (FastAPI) itibaren fastapi içe aktarmak FastAPI, Bağlıdır, HTTPException itibaren pıdantik içe aktarmak Temel Model itibaren yazarak içe aktarmak Liste, İsteğe Bağlı içe aktarmak base64 uygulaması = FastAPI(başlık="SynX Cüzdan API", sürüm="1.0.0") sınıf AdresYanıt(TemelModel): """Kuantum sonrası ortak anahtarlara sahip adres""" adres: str yol: str kyber_public_key: str # Base64 kodlu (1.184 bayt) sphincs_public_key: str # Base64 kodlu (32 bayt) bakiye: int pending_balance: int created_at: str sınıf DeriveAddressRequest(BaseModel): hesap: int = 0 değişiklik: int = 0 indeks: İsteğe bağlı[int] = Yok # Yok ise otomatik artış @app.get("/API/v1/adresler", yanıt_model=Liste[AdresYanıt]) eşzamansız tanım liste_adresleri( Wallet_id: str = Bağlıdır(get_authenticated_wallet), skip: int = 0, limit: int = 50 ): """ Cüzdan adreslerini bakiyelerle birlikte listeleyin Not: Kyber genel anahtarları büyüktür (1,2 KB). Listelemek için anahtarları hariç tutmayı ve ayrı olarak getirmeyi düşünün. """ adresler = beklemek adres_hizmet.list_adresleri(cüzdan_kimliği, atlama=atla, sınır=sınır) geri dönmek [ AdresYanıt( adres=addr.address, yol=addr.path, kyber_public_key=base64.b64encode(addr.kyber_pk).decode(), sphincs_public_key=base64.b64encode(addr.sphincs_pk).decode(), denge=addr.balance, pending_balance=addr.pending_balance, created_at=addr.created_at.isoformat() ) için adres in adresler] @app.post("/API/v1/adresler/türet", yanıt_modeli=AdresYanıt) eşzamansız tanım türetme_adresi( rica etmek: DeriveAddressRequest, cüzdan_id: str = Bağlıdır(get_authenticated_wallet) ): """Belirtilen türetme yolundan yeni adres türet""" adres = beklemek adres_hizmeti.derive_address( cüzdan_kimliği, hesap=istek.hesap, değişiklik=istek.değişim, indeks=istek.index ) geri dönmek AdresYanıt(...)

sınıf İşlem Özeti(TemelModel): """Tam imza verileri olmadan işlem (listeler için)""" tx_id: str zaman damgası: str inputs_count: int çıktılar_sayımı: int miktarı: int ücreti: int onayları: int durumu: str # "beklemede", "onaylandı", "başarısız oldu" sınıf İşlemTam(TemelModel): """İmzalar dahil tüm işlem""" tx_id: str version: int timestamp: str girişler: Liste['İşlem Giriş Şeması'] çıktılar: Liste['İşlemÇıktıŞeması'] ücret: int onaylar: int blok_hash: İsteğe bağlı[str] raw_hex: str # Tam serileştirilmiş işlem sınıf İşlemGiriş Şeması(BaseModel): prev_tx_id: str prev_output_index: int miktarı: int adresi: str imzası: str # Base64 (SPHINCS+-SHAKE-128'ler için ~10,5KB, 7.856 ham bayt) public_key: dizi # Base64 (SPHINCS+ için 44 bayt) sınıf İşlem İsteği Oluştur(BaseModel): çıktılar: Liste['ÇıktıBelirtimi'] ücret_oranı: İsteğe bağlı[int] = Yok # Yok ise otomatik hesapla change_address: İsteğe bağlı[str] = Yok # Yok ise otomatik seç sınıf Çıkış Özellikleri(BaseModel): alıcı: str miktarı: int memo: Optional[str] = Yok @app.get("/API/v1/işlemler", yanıt_model=Liste[İşlem Özeti]) eşzamansız tanım list_transactions( Wallet_id: str = Depends(get_authenticated_wallet), skip: int = 0, limit: int = 20, status: Optional[str] = Yok ): """ İşlemleri listele (yalnızca özetler) Yükü azaltmak için imzalar liste yanıtlarından hariç tutulur. İmzalarla tam işlem için GET /transactions/{tx_id} kullanın. """ teşekkürler = beklemek transaction_service.list_transactions( Wallet_id, skip=atla, limit=limit, durum=durum ) geri dönmek [tx.to_summary() için tx in teşekkürler] @app.get("/API/v1/transactions/{tx_id}", yanıt_modeli=İşlemTam) eşzamansız tanım get_transaction( tx_id: str, Wallet_id: str = Bağlıdır(get_authenticated_wallet), include_signatures: bool = Doğru ): """ Tüm işlem ayrıntılarını alın Yalnızca işlem meta verilerine ihtiyacınız varsa yanıt boyutunu azaltmak için include_signatures=false ayarını yapın. """ tx = beklemek transaction_service.get_transaction(wallet_id, tx_id) değilse tx: artırmak HTTPException(status_code=404, detay="İşlem bulunamadı") geri dönmek tx.to_full_schema(include_signatures=include_signatures) @app.post("/API/v1/işlemler/derleme") eşzamansız tanım build_transaction( rica etmek: İşlem İsteği Oluştur, cüzdan_id: str = Bağlıdır(get_authenticated_wallet) ): """ İmzasız işlem oluştur İstemci tarafı imzalamaya hazır işlem verilerini döndürür. İmzalama, özel anahtarları sunucudan uzak tutmak için istemcide gerçekleşir. """ unsigned_tx = beklemek transaction_service.build_transaction( cüzdan_kimliği, çıktılar=istek.çıkışlar, ücret_oranı=istek.fee_oranı, change_address=request.change_address ) geri dönmek { "imzasız_tx": base64.b64encode(unsigned_tx.serialize_for_signing()).decode(), "imza_mesajı": base64.b64encode(unsigned_tx.tx_hash()).decode(), "imzaya_girdiler": [ { "dizin": i, "adres": giriş adresi, "miktar": giriş tutarı, "türetme_yolu": inp.yol } için ben, inp in numaralandır(unsigned_tx.inputs)], "tahmini_ücret": unsigned_tx.fee, "tahmini_boyut": unsigned_tx.estimated_size() }

# Kyber + SPHINCS+ kullanarak kimlik doğrulama akışı içe aktarmak tamam içe aktarmak hashlib içe aktarmak hmac itibaren tarihsaat içe aktarmak tarihsaat, zaman deltası sınıf QuantumSafeKimlik Doğrulama: """ Kuantum açısından güvenli API kimlik doğrulama Akışı: 1. İstemci Kyber genel anahtarını gönderir 2. Sunucu oturum anahtarını kapsüller 3. İstemci oturum anahtarını almak için kapsülden çıkarır 4. İstekler """ oturum anahtarını kullanarak HMAC ile imzalanır def __init__(kendisi): self.session_store = {} # Üretimde Redis'i kullanın self.session_duration = timedelta(saat=24) eşzamansız tanım startup_session( self, Wallet_id: str, client_kyber_pk: bytes ) -> dict: """ Adım 1: İstemci oturumu Kyber genel anahtarıyla başlatır Sunucu, istemcinin """ anahtarına bir oturum sırrı saklar kem = oqs.KeyEncapsulation("Kyber768") şifreli metin, paylaşılan_secret = kem.encap_secret(client_kyber_pk) # Oturum anahtarını paylaşılan sırdan türetin oturum_anahtarı = hashlib.shake_256( paylaşılan_secret + b"oturum anahtarı" ).digest(32) # Oturum kimliği oluştur session_id = hashlib.Blake2b( paylaşılan_secret + str(datetime.utcnow()).encode(), sindirim_size=16 ).hexdigest() # Oturumu sakla (sunucu tarafı) self.session_store[session_id] = { "cüzdan_id": cüzdan_kimliği, "oturum_anahtarı": oturum_anahtarı, "süresi doluyor_at": datetime.utcnow() + self.session_duration, "oluşturuldu_at": tarihsaat.utcnow() } geri dönmek { "oturum_id": oturum_kimliği, "şifreli metin": base64.b64encode(ciphertext).decode(), "süresi doluyor_at": (datetime.utcnow() + self.session_duration).isoformat() } def doğrulama_request( self, session_id: str, request_signature: bytes, request_data: bytes, timestamp: int ) -> İsteğe bağlı[str]: """ Oturum anahtarını kullanarak istek imzasını doğrulayın Geçerliyse cüzdan_id değerini döndürür, aksi halde Yok """ oturum = self.session_store.get(session_id) değilse oturum: geri dönmek Hiçbiri # Son kullanma tarihini kontrol et if datetime.utcnow() > oturum["süresi doluyor_at"]: del self.session_store[session_id] geri dönmek Hiçbiri # Zaman damgasını kontrol edin (tekrar oynatmayı önleyin) request_time = datetime.fromtimestamp(timestamp) if abs((datetime.utcnow() - request_time).total_seconds()) > 300: geri dönmek Hiçbiri # 5 dakikadan fazla eski/gelecek # HMAC imzasını doğrulayın beklenen_sig = hmac.new( oturum["oturum_anahtarı"], request_data + str(timestamp).encode(), hashlib.Blake2b ).digest() if hmac.compare_digest(request_signature, beklenen_sig): geri dönmek oturum["cüzdan_id"] geri dönmek Hiçbiri # Kimliği doğrulanmış rotalar için FastAPI bağımlılığı kimlik_hizmeti = QuantumSafeKimlik Doğrulama() eşzamansız tanım get_authenticated_wallet( x_session_id: str = Başlık(...), x_signature: str = Başlık(...), x_timestamp: str = Başlık(...), request: İstek = Yok ) -> str: """Kuantum güvenli kimlik doğrulamayı doğrulayan bağımlılık""" vücut = beklemek 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) ) değilse cüzdan_kimliği: artırmak HTTPException(status_code=401, detay="Geçersiz kimlik doğrulama") geri dönmek cüzdan_kimliği

# Gerçek zamanlı güncellemeler için WebSocket uygulaması itibaren fastapi içe aktarmak WebSocket, WebSocketBağlantıyı Kes içe aktarmak json içe aktarmak uyumsuz sınıf Bağlantı Yöneticisi: """Cüzdan başına WebSocket bağlantılarını yönetin""" def __init__(self): self.active_connections: dict[str, List[WebSocket]] = {} eşzamansız tanım bağlamak(self, websocket: WebSocket, Wallet_id: str): beklemek websocket.accept() if cüzdan_kimliği içinde değil self.active_connections: self.active_connections[wallet_id] = [] self.active_connections[wallet_id].append(websocket) def bağlantıyı kes(self, websocket: WebSocket, Wallet_id: str): if cüzdan_kimliği in self.active_connections: self.active_connections[wallet_id].remove(websocket) eşzamansız tanım yayın_to_wallet(self, Wallet_id: str, message: dict): if cüzdan_kimliği in self.active_connections: ölü_bağlantılar = [] için bağlantı in self.active_connections[wallet_id]: denemek: beklemek bağlantı.send_json(mesaj) hariç: dead_connections.append(bağlantı) # Ölü bağlantıları temizleyin için bağlantı in ölü_bağlantılar: self.active_connections[wallet_id].remove(bağlantı) yöneticisi = Bağlantı Yöneticisi() @app.websocket("/ws/{wallet_id}") eşzamansız tanım websocket_endpoint(websocket: WebSocket, Wallet_id: str): """ Gerçek zamanlı cüzdan güncellemeleri için WebSocket Olaylar: -balance_update: Bakiye değişti - transaction_received: Gelen işlem - transaction_confirmed: TX onaylara ulaştı - transaction_sent: Giden TX yayını """ # WebSocket bağlantısını doğrulayın auth_token = websocket.query_params.get("jeton") eğer beklemezsen validate_ws_token(auth_token, cüzdan_kimliği): beklemek websocket.close(kod=4001) geri dönmek beklemek Manager.connect(websocket, cüzdan_id) denemek: # Başlangıç ​​durumunu gönder beklemek websocket.send_json({ "tip": "bağlı", "cüzdan_id": cüzdan_kimliği, "zaman damgası": datetime.utcnow().isoformat() }) # Gelen mesajları yönetin (abonelikler, pingler) sırasında Doğru: veri = beklemek websocket.receive_json() if data.get("tip") == "ping": beklemek websocket.send_json({"tip": "pong"}) elif data.get("tip") == "abone": # Belirli adreslere abone olun adresler = data.get("adresler", []) beklemek abonelik_hizmeti.subscribe(cüzdan_kimliği, adresler) hariç WebSocketDisconnect: yönetici.disconnect(websocket, cüzdan_id) # Olay yayını (blockchain monitörü tarafından çağrılır) eşzamansız tanım yayın_denge_update(wallet_id: str, adres: str, new_balance: int): beklemek Manager.broadcast_to_wallet(wallet_id, { "tip": "denge_güncelleme", "adres": adres, "denge": yeni_denge, "zaman damgası": datetime.utcnow().isoformat() }) eşzamansız tanım yayın_transaction_received(wallet_id: str, tx_summary: dict): beklemek Manager.broadcast_to_wallet(wallet_id, { "tip": "işlem_alındı", "işlem": tx_summary, # Yalnızca özet, tam imza değil "zaman damgası": datetime.utcnow().isoformat() })

# FastAPI'de gzip sıkıştırmasını etkinleştirin itibaren fastapi.middleware.gzip içe aktarmak GZipMiddleware app.add_middleware(GZipMiddleware, minimum_size=1000) # İsteğe bağlı: Mobil/yerleşik istemciler için Mesaj Paketi yanıtları itibaren fastapi.responses içe aktarmak Cevap içe aktarmak mesaj paketi sınıf MsgPack Yanıtı(Yanıt): media_type = "uygulama/msgpack" def oluşturmak(kendi, içerik) -> bayt: geri dönmek msgpack.packb(içerik, use_bin_type=Doğru) @app.get("/API/v1/transactions/{tx_id}/binary") eşzamansız tanım get_transaction_binary(tx_id:str): """İşlemi Mesaj Paketi biçiminde al (JSON'dan küçük)""" tx = beklemek transaction_service.get_transaction(tx_id) geri dönmek MsgPack Yanıtı(içerik=tx.to_dict())

# Cüzdan API için hız sınırlaması itibaren yavaşapi içe aktarmak Sınırlayıcı, _rate_limit_exceeded_handler itibaren yavaşapi.hataları içe aktarmak RateLimitExceeded sınırlayıcı = Sınırlayıcı(key_func=get_wallet_id_from_request) app.state.limiter = sınırlayıcı app.add_ Exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler) # Farklı işlemler için farklı limitler RATE_LIMITS = { "Okumak": "100/dakika", # Bakiye kontrolleri, tx listeleri "yazmak": "20/dakika", # Adres türetme "yayın": "5/dakika", # İşlem yayını } @app.get("/API/v1/denge") @sınırlayıcı.limit("100/dakika") eşzamansız tanım get_balance(talep: Talep): ... @app.post("/API/v1/işlemler/yayın") @sınırlayıcı.limit("5/dakika") eşzamansız tanım yayın_transaction(talep: Talep): """Spam'i önlemek için yayın sınırı daha sıkı""" ...

# Standartlaştırılmış hata yanıtları itibaren Enum içe aktarmak Enum sınıf Hata Kodu(str, Enum): INVALID_ADDRESS = "GEÇERSİZ_ADDRESS" YETERSİZ_BALANCE = "YETERLİ_BALANS" INVALID_SIGNATURE = "GEÇERSİZ_İMZA" TRANSACTION_REJECTED = "TRANSACTION_REDDEDİLDİ" RATE_LIMITED = "RATE_LIMITED" SESSION_EXPIRED = "SESSION_EXPIRED" DERIVATION_FAILED = "DERIVATION_FAILED" sınıf API Hatası(BaseModel): kod: Hata Kodu mesaj: str ayrıntıları: İsteğe bağlı[dict] = Yok @app.istisna_handler(Yetersiz Bakiye Hatası) eşzamansız tanım yetersiz_denge_işleyicisi(istek, hariç): geri dönmek JSONResponse( durum_kodu=400, içerik=API Hatası( kod=Hata Kodu.INSUFFICIENT_BALANCE, mesaj="İşlem için yeterli bakiye yok", ayrıntılar={ "mevcut": mevcut olanlar hariç, "gerekli": hariç gerekli, "eksiklik": hariç.gerekli - hariç.mevcut } ).dict() )