ポスト量子ウォレット API 設計: REST および WebSocket パターン

# アドレスエンドポイントの実装 (FastAPI) から ファスタピ 輸入 FastAPI、依存、HTTPException から 卑劣な 輸入 ベースモデル から タイピング 輸入 リスト、オプション 輸入 Base64 アプリ = FastAPI(title=「SynX ウォレット API」、バージョン="1.0.0") クラス アドレス応答(ベースモデル): """ポスト量子公開鍵を使用したアドレス""" アドレス: str パス: str kyber_public_key: str # Base64 エンコード (1,184 バイト) sphincs_public_key: str # Base64 エンコード (32 バイト) バランス: int pending_balance: int created_at: str クラス DeriveAddressRequest(BaseModel): アカウント: int = 0 変更: int = 0 インデックス: Optional[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) 戻る [ アドレス応答( 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.pending_balance、 created_at=addr.created_at.isoformat() ) のために アドレス in アドレス] @app.post(「/API/v1/アドレス/派生」、response_model=アドレス応答) 非同期定義 派生アドレス（ リクエスト： DeriveAddressRequest、wallet_id: str = 依存(get_authenticated_wallet) ): """指定された導出パスで新しいアドレスを導出します""" アドレス = 待つ address_service.derive_address(wallet_id, account=request.account,change=request.change,index=request.index) 戻る アドレス応答(...)

クラス 取引概要(ベースモデル): """完全な署名データのないトランザクション (リスト用)""" tx_id: str タイムスタンプ: str inputs_count: int Outputs_count: int 金額: int 料金: int 確認: int ステータス: str # 「保留中」、「確認済み」、「失敗」 クラス トランザクションフル(ベースモデル): """署名を含む完全な取引""" tx_id: str バージョン: int タイムスタンプ: str 入力: List['トランザクション入力スキーマ'] 出力: List['トランザクション出力スキーマ'] 料金: int 確認: int block_hash: Optional[str] raw_hex: str # 完全なシリアル化されたトランザクション クラス トランザクション入力スキーマ(BaseModel): prev_tx_id: str prev_output_index: int 量: int アドレス: str 署名: str # Base64 (SPHINCS+-SHAKE-128 の場合は約 10.5KB、7,856 raw バイト) 公開キー: str # Base64 (SPHINCS+ の場合は 44 バイト) クラス BuildTransactionRequest(BaseModel): 出力: List['出力仕様'] 料金レート: オプション[int] = なし # なしの場合は自動計算 Change_address: オプション[str] = なし # なしの場合は自動選択 クラス 出力仕様(BaseModel): 受信者: str 金額: int メモ: オプション[str] = なし @app.get(「/API/v1/トランザクション」、response_model=リスト[取引概要]) 非同期定義 リスト_トランザクション(wallet_id: str = depends(get_authenticated_wallet)、スキップ: int = 0、制限: int = 20、ステータス: Optional[str] = None ): """ リスト トランザクション (概要のみ) ペイロードを減らすために、署名はリストの応答から除外されます。署名のある完全なトランザクションには GET /transactions/{tx_id} を使用してください。 """ txs = 待つ transaction_service.list_transactions(wallet_id、skip=スキップ、limit=制限、status=ステータス) 戻る [tx.to_summary() のために tx in TXS] @app.get(「/API/v1/transactions/{tx_id}」、response_model=トランザクションフル) 非同期定義 get_transaction( tx_id: str、wallet_id: str = depends(get_authenticated_wallet)、include_signatures: bool = True ): """ トランザクションの完全な詳細を取得する トランザクションのメタデータのみが必要な場合は、 include_signatures=false を設定して応答サイズを削減します。 """ tx = 待つ transaction_service.get_transaction(wallet_id, tx_id) そうでない場合 送信: 上げる HTTPException(ステータスコード=404、詳細=「トランザクションが見つかりません」) 戻る tx.to_full_schema(include_signatures=include_signatures) @app.post(「/API/v1/トランザクション/ビルド」) 非同期定義 ビルド_トランザクション（ リクエスト： BuildTransactionRequest、wallet_id: str = 依存(get_authenticated_wallet) ): """ 未署名のトランザクションを構築します。クライアント側で署名できる状態のトランザクション データを返します。署名はクライアントで行われ、秘密キーがサーバーから遠ざかるようにします。 """ unsigned_tx = 待つ transaction_service.build_transaction(wallet_id,outputs=request.outputs,fee_rate=request.fee_rate,change_address=request.change_address) 戻る { 「unsigned_tx」:base64.b64encode(unsigned_tx.serialize_for_signing()).decode()、 「署名メッセージ」:base64.b64encode(unsigned_tx.tx_hash()).decode()、 「署名への入力」: [ { "索引": i, "住所": 入力アドレス、 "額": 入金金額、 「導出パス」: inp.パス } のために 私、インプ in enumerate(unsigned_tx.inputs) ], 「推定料金」: unsigned_tx.fee、 「推定サイズ」: unsigned_tx.estimated_size() }

# Kyber + SPHINCS+を使用した認証フロー 輸入 オークス 輸入 ハッシュリブ 輸入 hmac から 日時 輸入 日付時刻、時刻デルタ クラス 量子安全認証: """ 量子安全な API 認証 フロー: 1. クライアントが Kyber 公開キーを送信 2. サーバーがセッション キーをカプセル化 3. クライアントがカプセル化を解除してセッション キーを取得 4. セッション キーを使用して HMAC で署名されたリクエスト """ 確かに __初期化__(自分自身): self.session_store = {} # 運用環境では Redis を使用します self.session_duration = timedelta(時間=24) 非同期定義 セッションの開始( self、wallet_id: str、client_kyber_pk: bytes ) -> dict: """ ステップ 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[セッションID] = { 「ウォレットID」: ウォレットID、 「セッションキー」: セッションキー、 "有効期限": datetime.utcnow() + self.session_duration、 「作成された場所」: datetime.utcnow() } 戻る { 「セッションID」: セッションID、 「暗号文」:base64.b64encode(暗号文).decode()、 "有効期限": (datetime.utcnow() + self.session_duration).isoformat() } 確かに verify_request( self、session_id: str、request_signature: バイト、request_data: バイト、タイムスタンプ: int ) -> オプション[str]: """ セッションキーを使用してリクエストの署名を検証します。有効な場合はwallet_idを返し、それ以外の場合はなしを返します。 """ session = self.session_store.get(session_id) そうでない場合 セッション： 戻る なし # 有効期限を確認する if datetime.utcnow() > セッション["有効期限"]: デル self.session_store[セッションID] 戻る なし # タイムスタンプをチェックする (リプレイを防止) request_time = datetime.fromtimestamp(タイムスタンプ) if abs((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 依存関係 認証サービス = 量子安全認証() 非同期定義 get_authenticated_wallet( x_session_id: str = ヘッダー(...)、x_signature: str = ヘッダー(...)、x_timestamp: str = ヘッダー(...)、リクエスト: リクエスト = なし ) -> 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 実装 から ファスタピ 輸入 WebSocket、WebSocketDisconnect 輸入 json 輸入 非同期 クラス 接続マネージャー: """ウォレットごとに WebSocket 接続を管理""" 確かに __初期化__(self): self.active_connections: dict[str, List[WebSocket]] = {} 非同期定義 接続する(self、websocket: WebSocket、wallet_id: str): 待つ websocket.accept() if ウォレットID 入っていない self.active_connections: self.active_connections[wallet_id] = [] self.active_connections[wallet_id].append(websocket) 確かに 切断する(self、websocket: WebSocket、wallet_id: str): if ウォレットID in self.active_connections: self.active_connections[wallet_id].remove(websocket) 非同期定義 ウォレットへのブロードキャスト(self、wallet_id: str、メッセージ: dict): if ウォレットID in self.active_connections: Dead_connections = [] のために 繋がり in self.active_connections[ウォレットID]: 試す: 待つ connection.send_json(メッセージ) を除外する: Dead_connections.append(接続) # 死んだ接続をクリーンアップする のために コン in Dead_connections: self.active_connections[wallet_id].remove(conn) マネージャー = 接続マネージャー() @app.websocket(「/ws/{wallet_id}」) 非同期定義 websocket_endpoint(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 データ.get("タイプ") == 「ピン」: 待つ websocket.send_json({"タイプ": 「ポン」}) エリフ データ.get("タイプ") == "購読する": # 特定のアドレスを購読する アドレス = data.get(「住所」, []) 待つ subscription_service.subscribe( ウォレット 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.ミドルウェア.gzip 輸入 GZipMiddleware app.add_middleware(GZipMiddleware, minimum_size=1000) # オプション: モバイル/組み込みクライアント用の MessagePack 応答 から fastapi.responses 輸入 応答 輸入 メッセージパック クラス メッセージパック応答(応答): メディアタイプ = 「アプリケーション/メッセージパック」 確かに 与える(自己、コンテンツ) -> バイト: 戻る msgpack.packb(コンテンツ、use_bin_type=True) @app.get(「/API/v1/transactions/{tx_id}/binary」) 非同期定義 get_transaction_binary(tx_id: 文字列): """MessagePack 形式 (JSON より小さい) でトランザクションを取得します""" tx = 待つ トランザクションサービス.get_transaction(tx_id) 戻る メッセージパック応答(content=tx.to_dict())

# ウォレット 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) # 操作ごとに異なる制限 RATE_LIMITS = { "読む": 「100/分」, # 残高チェック、送信リスト "書く": 「20/分」, # アドレス導出 "放送": 「5/分」, # トランザクションブロードキャスト } @app.get(「/API/v1/バランス」) @リミッター.リミット(「100/分」) 非同期定義 バランスを取る(リクエスト: リクエスト): ... @app.post(「/API/v1/トランザクション/ブロードキャスト」) @リミッター.リミット(「5/分」) 非同期定義 ブロードキャスト_トランザクション(リクエスト: リクエスト): """スパム防止のためブロードキャストの制限を厳格化""" ...

# 標準化されたエラー応答 から 列挙型 輸入 列挙型 クラス エラーコード(文字列、列挙型): INVALID_ADDRESS = 「無効なアドレス」 不足残高 = 「残高が不足しています」 INVALID_SIGNATURE = 「INVALID_SIGNATURE」 トランザクション拒否 = 「TRANSACTION_REJECTED」 RATE_LIMITED = 「RATE_LIMITED」 SESSION_EXPIRED = 「セッション_期限切れ」 DERIVATION_FAILED = 「DERIVATION_FAILED」 クラス APIエラー(ベースモデル): コード: エラーコード メッセージ: str 詳細: オプション[dict] = なし @app.例外ハンドラー(残高不足エラー) 非同期定義 不十分な残高ハンドラー(リクエスト、除外): 戻る JSONResponse(ステータスコード=400、コンテンツ=APIエラー(コード=エラーコード.INSUFFICIENT_BALANCE、メッセージ=「取引に必要な残高が不足しています」、詳細={ "利用可能": 利用可能以外、 "必須": 必須を除く、 「不足」: exc.required - exc.available } ).dict() )