後量子錢包 API 設計：REST 和 WebSocket 模式

# 位址端點實作 (FastAPI) 從 快速API 進口 FastAPI、取決於、HTTPException 從 迂腐的 進口 基礎模型 從 打字 進口 列表，可選 進口 Base64 應用程式 = FastAPI(標題=“SynX 皮夾 API”，版本="1.0.0") 班級 地址回應（基本模型）： """具有後量子公鑰的位址""" 位址：str 路徑：str kyber_public_key：str # Base64 編碼（1,184 位元組） sphincs_public_key：str # Base64編碼（32位元組） 餘額：int未決餘額：intcreated_at：str 班級 派生地址請求（BaseModel）：帳戶：int = 0 變更：int = 0 索引：可選[int] = None # 如果沒有則自動遞增 @app.get(“/API/v1/地址”,response_model=列表[地址回應]) 非同步定義 清單位址（wallet_id：str = Depends（get_authenticated_wallet），跳過：int = 0，限制：int = 50）： """ 列出錢包位址及其餘額 注意：Kyber 公鑰很大（1.2KB）。為了列出，請考慮排除金鑰並單獨取得。""" 地址 = 等待 address_service.list_addresses( wallet_id,skip=skip,limit=limit ) 返回 [ 地址回應（地址=addr.address，路徑=addr.path，kyber_public_key=base64.b64encode（addr.kyber_pk）.decode（），sphincs_public_key=base64.b64encode（addr.sphincs_pbal).decode（），baseance=addance. created_at=addr.created_at.isoformat() ) 為了 地址 in 地址] @app.post(“/API/v1/地址/派生”，響應模型=地址回應) 非同步定義 派生地址（ 要求： 派生地址請求, wallet_id: str = Depends(get_authenticated_wallet) ): """在指定的派生路徑派生新位址""" 地址= 等待 address_service.derive_address（ wallet_id，account=request.account，change=request.change，index=request.index ） 返回 地址回應(...)

班級 交易摘要（基本模型）： """沒有完整簽名資料的交易（對於清單）""" tx_id：str 時間戳記：str 輸入計數：int 輸出計數：int 金額：int 費用：int 確認：int 狀態：str #“待處理”、“已確認”、“失敗” 班級 交易已滿（基本模型）： """包含簽名的完整交易""" tx_id: str 版本: int 時間戳記: str 輸入: List['交易輸入架構'] 輸出：列表['交易輸出模式'] 費用：int 確認：int block_hash：可選[str] raw_hex：str # 完整的序列化交易 班級 交易輸入模式(BaseModel): prev_tx_id: str prev_output_index: int 金額: int 位址: str 簽章: str # Base64（SPHINCS+-SHAKE-128s 約 10.5KB，7,856 個原始位元組） 公鑰：str # Base64（SPHINCS+ 為 44 位元組） 班級 建立交易請求（BaseModel）：輸出：列表['輸出規格'] Fee_rate：可選[int] = None # 如果沒有則自動計算 更改位址：可選[str] =無 # 如果沒有則自動選擇 班級 輸出規格(BaseModel): 收件者: str 金額: int 備註: 可選[str] = None @app.get(“/API/v1/交易”,response_model=列表[交易摘要]) 非同步定義 交易清單( wallet_id: str = Depends(get_authenticated_wallet), 跳過: int = 0, 限制: int = 20, 狀態: 可選[str] = None ): """ 列出交易（僅摘要） 簽名從清單回應中排除以減少有效負載。使用 GET /transactions/{tx_id} 取得具有簽名的完整交易。""" 發送 = 等待 transaction_service.list_transactions( wallet_id,skip=skip,limit=limit,status=status ) 返回 [tx.to_summary() 為了 tx in TXS] @app.get(“/API/v1/交易/{tx_id}”，響應模型=交易已滿) 非同步定義 取得交易( tx_id: str, wallet_id: str = Depends(get_authenticated_wallet), include_signatures: bool = 真 ): """ 取得完整的交易詳細資料 如果您只需要交易元數據，請設定 include_signatures=false 以減少回應大小。 """ 發送 = 等待 transaction_service.get_transaction(wallet_id, tx_id) 如果不 發送： 增加 HTTPException（狀態代碼=404，詳細資料=“未找到交易”) 返回 tx.to_full_schema(include_signatures=include_signatures) @app.post(“/API/v1/交易/建置”) 非同步定義 建構交易（ 要求： 建立交易請求, wallet_id: str = Depends(get_authenticated_wallet) ): """ 建立未簽署的交易 傳回準備好進行客戶端簽署的交易資料。簽章發生在客戶端上，以使私鑰遠離伺服器。""" 無符號_tx = 等待 transaction_service.build_transaction（ wallet_id，outputs = request.outputs，fee_rate = request.fee_rate，change_address = request.change_address） 返回 { “無符號_tx”：base64.b64encode（unsigned_tx.serialize_for_signing（））.decode（）， “簽名訊息”：base64.b64encode(unsigned_tx.tx_hash()).decode(), “輸入簽名”: [ { “指數”: i, “地址”: inp.地址, “數量”：輸入金額， “派生路徑”: inp.path } 為了 我，inp in 列舉（unsigned_tx.inputs）]， “預計費用”: unsigned_tx.fee, “估計大小”: unsigned_tx.estimated_size() }

# 使用 Kyber + SPHINCS+ 的身份驗證流程 進口 奧克斯 進口 哈希庫 進口 哈馬奇 從 日期時間 進口 日期時間、時間增量 班級 量子安全認證: """ 量子安全 API 驗證流程： 1. 用戶端傳送 Kyber 公鑰 2. 伺服器封裝會話金鑰 3. 用戶端解封裝以取得會話金鑰 4. 使用會話金鑰透過 HMAC 簽署的請求 """ 定義 __初始化__(自我): self.session_store = {} # 在生產中，使用Redis self.session_duration = timedelta(小時=24) 非同步定義 發起會話( self, wallet_id: str, client_kyber_pk: bytes ) -> 字典: """ 第 1 步：客戶端使用 Kyber 公鑰發起會話 伺服器將會話秘密封裝到客戶端的金鑰 """ kem = oqs.KeyEncapsulation(“凱伯768”) 密文，shared_secret = kem.encap_secret(client_kyber_pk) # 從共用金鑰匯出會話金鑰 session_key = hashlib.shake_256(shared_secret + b“會話密鑰” ).摘要(32) # 建立會話ID session_id = hashlib.Blake2b（shared_secret + str（datetime.utcnow（））.encode（），digest_size = 16）.hexdigest（） # 儲存會話（伺服器端） self.session_store[session_id] = { “錢包_id”: 錢包_id, “會話密鑰”：會話金鑰， “到期時間”: datetime.utcnow() + self.session_duration, “創建於”: 日期時間.utcnow() } 返回 { “會話 ID”：會話 ID， “密文”：base64.b64encode(密文).decode(), “到期時間”: (datetime.utcnow() + self.session_duration).isoformat() } 定義 驗證請求( self, session_id: str, request_signature: bytes, request_data: bytes, timestamp: int ) -> 可選[str]: """ 使用會話金鑰驗證請求簽名 如果有效則傳回 wallet_id，否則傳回 None """ 會話 = self.session_store.get(session_id) 如果不 會議： 返回 沒有任何 # 檢查過期時間 if datetime.utcnow() > 會話[“到期時間”]: 德爾 self.session_store[session_id] 返回 沒有任何 # 檢查時間戳記（防止重播） request_time = datetime.fromtimestamp(時間戳記) if 絕對((datetime.utcnow() - request_time).total_seconds()) > 300: 返回 沒有任何 # 過去/未來超過 5 分鐘 # 驗證HMAC簽名 Expected_sig = hmac.new( 會話[“會話密鑰”], request_data + str(時間戳).encode(), hashlib.Blake2b ).digest() if hmac.compare_digest（request_signature，expected_sig）： 返回 會議[“錢包_id”] 返回 沒有任何 # 經過驗證的路由的 FastAPI 依賴項 授權服務= 量子安全認證() 非同步定義 取得驗證錢包( x_session_id: str = Header(...), x_signature: str = Header(...), x_timestamp: str = Header(...), request: Request = None ) -> str: """驗證量子安全身份驗證的依賴關係""" 身體= 等待 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) ) 如果不 錢包ID： 增加 HTTPException（狀態代碼=401，詳細資料=“驗證無效”) 返回 錢包ID

# 用於即時更新的 WebSocket 實現 從 快速API 進口 WebSocket、WebSocketDisconnect 進口 json 進口 非同步 班級 連線管理器: """管理每個錢包的 WebSocket 連線""" 定義 __初始化__(self): self.active_connections: dict[str, List[WebSocket]] = {} 非同步定義 連接（自我，websocket：WebSocket，wallet_id：str）： 等待 websocket.accept() if 錢包ID 不在 self.active_connections: self.active_connections[wallet_id] = [] self.active_connections[wallet_id].append(websocket) 定義 斷開（自我，websocket：WebSocket，wallet_id：str）： if 錢包ID in self.active_connections: self.active_connections[wallet_id].remove(websocket) 非同步定義 廣播到錢包（自我，wallet_id：str，訊息：dict）： if 錢包ID in self.active_connections: dead_connections = [] 為了 聯繫 in self.active_connections[wallet_id]: 嘗試: 等待 連接.send_json(訊息) 除了: dead_connections.append(連接) # 清理死連接 為了 康乃狄克州 in dead_connections: self.active_connections[wallet_id].remove(conn) manager = 連線管理器() @app.websocket(“/ws/{wallet_id}”) 非同步定義 websocket_端點（websocket：WebSocket，wallet_id：str）： """ 用於即時錢包更新的 WebSocket 事件： - Balance_update：餘額已更改 - transaction_received：傳入交易 - transaction_confirmed：TX 達到確認 - transaction_sent：傳出 TX 廣播 """ # 驗證 WebSocket 連接 auth_token = websocket.query_params.get(“令牌”) 如果不等待 validate_ws_token（auth_token，wallet_id）： 等待 websocket.close（代碼=4001） 返回 等待 manager.connect(websocket, wallet_id) 嘗試: # 傳送初始狀態 等待 websocket.send_json({ “類型”: “已連線”, “錢包_id”: 錢包_id, “時間戳”: datetime.utcnow().isoformat() }) # 處理傳入訊息（訂閱、ping） 儘管 正確：數據 = 等待 websocket.receive_json() if 數據.獲取(“類型”) == “平”: 等待 websocket.send_json({“類型”: “乒”}) 埃利夫 數據.獲取(“類型”) == “訂閱”: # 訂閱特定位址 地址 = data.get(“地址”, []) 等待 subscription_service.subscribe( wallet_id, 位址 ) 除了 WebSocketDisconnect: manager.disconnect(websocket, wallet_id) # 事件廣播（由區塊鏈監視器呼叫） 非同步定義 廣播平衡更新（wallet_id：str，地址：str，new_balance：int）： 等待 manager.broadcast_to_wallet(wallet_id, { “類型”: “平衡更新”, “地址”： 地址， “平衡”：新平衡， “時間戳”: datetime.utcnow().isoformat() }) 非同步定義 廣播交易接收(wallet_id: str, tx_summary: dict): 等待 manager.broadcast_to_wallet(wallet_id, { “類型”: “交易已收到”, “交易”: tx_summary, # 僅摘要，不完整簽名 “時間戳”: datetime.utcnow().isoformat() })

# 在 FastAPI 中啟用 gzip 壓縮 從 fastapi.middleware.gzip 進口 GZipMiddleware app.add_middleware（GZipMiddleware，minimum_size = 1000） # 可選：行動/嵌入式用戶端的 MessagePack 回應 從 fastapi.回應 進口 回覆 進口 訊息包 班級 訊息包回應（響應）：media_type = “應用程式/訊息包” 定義 使成為（自身，內容）->位元組： 返回 msgpack.packb(內容, use_bin_type=True) @app.get(“/API/v1/交易/{tx_id}/binary”) 非同步定義 取得交易二進位文件（tx_id：str）： """取得MessagePack格式的交易（小於JSON）""" 發送 = 等待 transaction_service.get_transaction(tx_id) 返回 訊息包回應(內容=tx.to_dict())

# 錢包API的速率限制 從 慢速API 進口 限制器，_rate_limit_exceeded_handler 從 慢速API.錯誤 進口 RateLimitExceeded 限制器 = 限制器(key_func=get_wallet_id_from_request) app.state.limiter = 限制器 app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler) # 不同的操作有不同的限制 速率限制 = { “讀”: “100/分鐘”, # 餘額檢查、交易列表 “寫”: “20/分鐘”, # 地址推導 “播送”: “5/分鐘”, # 交易廣播 } @app.get(“/API/v1/餘額”) @limiter.limit(“100/分鐘”) 非同步定義 獲取餘額（請求：請求）：... @app.post(“/API/v1/交易/廣播”) @limiter.limit(“5/分鐘”) 非同步定義 廣播交易（請求：請求）： """更嚴格的廣播限制以防止垃圾郵件""" ...

# 標準化錯誤回應 從 列舉 進口 列舉 班級 錯誤代碼（str，枚舉）：INVALID_ADDRESS = “INVALID_ADDRESS” 餘額不足 = “餘額不足” 無效簽名 = “INVALID_SIGNATURE” TRANSACTION_REJECTED = “交易被拒絕” 速率限制 = “RATE_LIMITED” 會話過期= “SESSION_EXPIRED” 派生失敗 = “推導_失敗” 班級 API錯誤（基礎模型）：程式碼： 錯誤代碼 訊息：str 詳細資料：可選[dict] = None @app.exception_handler（餘額不足錯誤） 非同步定義 餘額不足處理程序（請求，例外）： 返回 JSONResponse( 狀態碼=400，內容=API錯誤（代碼=錯誤代碼.INSUFFICIENT_BALANCE，訊息=“交易餘額不足”，詳細資料={ “可用的”: 除外可用， “必需的”: 排除必需的， “短缺”: exc.required - exc.available } ).dict() )