后量子钱包 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_pk）.decode（），balance=addr.balance，pending_balance=addr.pending_balance， 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 = True ): """ 获取完整的交易详细信息 如果您只需要交易元数据，请设置 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() )