포스트 퀀텀 지갑 API 설계: REST 및 WebSocket 패턴

# 주소 엔드포인트 구현(FastAPI) ~에서 패스트피 수입 FastAPI, 종속됨, HTTPException ~에서 피단틱한 수입 기본 모델 ~에서 타자 수입 목록, 선택사항 수입 base64 앱 = FastAPI(제목="SynX 지갑 API", 버전="1.0.0") 수업 주소응답(기본 모델): """포스트 퀀텀 공개 키가 있는 주소""" 주소: str 경로: str kyber_public_key: str # Base64로 인코딩됨(1,184바이트) sphincs_public_key: 문자열 # Base64로 인코딩됨(32바이트) 잔액: int Pending_balance: int Create_at: str 수업 파생주소요청(BaseModel): 계정: int = 0 변경: int = 0 인덱스: Optional[int] = None # 없으면 자동 증가 @app.get("/API/v1/주소", response_model=목록[주소응답]) 비동기 정의 목록_주소( wallet_id: str = 종속(get_authenticated_wallet), 건너뛰기: int = 0, 제한: int = 50 ): """ 잔액이 있는 지갑 주소 나열 참고: Kyber 공개 키는 대용량(1.2KB)입니다. 나열하려면 키를 제외하고 별도로 가져오는 것을 고려하십시오. """ 주소 = 기다리다 address_service.list_addresses( 지갑_id, 건너뛰기=건너뛰기, 제한=제한) 반품 [ 주소응답( 주소=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, 보류 중인_balance=addr.pending_balance, created_at=addr.created_at.isoformat() ) ~을 위한 주소 in 주소 ] @app.post("/API/v1/주소/파생", 응답_모델=주소응답) 비동기 정의 파생_주소( 요구: 파생주소요청, wallet_id: str = 종속(get_authenticated_wallet) ): """지정된 파생 경로에서 새 주소 파생""" 주소 = 기다리다 address_service.derive_address( wallet_id, account=request.account, 변경=request.change, index=request.index ) 반품 주소응답(...)

수업 거래요약(기본 모델): """전체 서명 데이터가 없는 거래(목록용)""" tx_id: str 타임스탬프: str inputs_count: int Outputs_count: int 금액: int 수수료: int 확인: int 상태: str # "보류 중", "확인됨", "실패함" 수업 거래가 가득 찼습니다.(기본 모델): """서명을 포함한 전체 거래""" tx_id: str 버전: int timestamp: 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 원시 바이트) public_key: 문자열 # Base64(SPHINCS+의 경우 44바이트) 수업 BuildTransactionRequest(BaseModel): 출력: 목록['출력 사양'] fee_rate: 선택사항[int] = 없음 # 없으면 자동 계산 변경 주소: 선택사항[str] = 없음 # 없으면 자동 선택 수업 출력 사양(BaseModel): 수신자: str 금액: int 메모: Optional[str] = None @app.get("/API/v1/트랜잭션", response_model=목록[거래요약]) 비동기 정의 목록 거래( wallet_id: str = 종속(get_authenticated_wallet), 건너뛰기: int = 0, 제한: int = 20, 상태: Optional[str] = None ): """ 목록 트랜잭션(요약만) 페이로드를 줄이기 위해 목록 응답에서 서명이 제외됩니다. 서명이 있는 전체 트랜잭션에는 GET /transactions/{tx_id}를 사용하세요. """ 전송 = 기다리다 transaction_service.list_transactions( 지갑_id, 건너뛰기=건너뛰기, 한도=한계, 상태=상태) 반품 [tx.to_summary() ~을 위한 tx in TX] @app.get("/API/v1/트랜잭션/{tx_id}", 응답_모델=거래가 가득 찼습니다.) 비동기 정의 get_transaction( tx_id: str, wallet_id: str = 종속(get_authenticated_wallet), include_signatures: bool = True ): """ 전체 거래 세부정보 가져오기 거래 메타데이터만 필요한 경우 응답 크기를 줄이려면 include_signatures=false를 설정하세요. """ 텍사스 = 기다리다 transaction_service.get_transaction(wallet_id, tx_id) 그렇지 않다면 텍사스: 들어올리다 HTTPException(status_code=404, 세부정보="거래를 찾을 수 없습니다") 반품 tx.to_full_schema(include_signatures=include_signatures) @app.post("/API/v1/트랜잭션/빌드") 비동기 정의 build_transaction( 요구: BuildTransactionRequest, wallet_id: str = 종속(get_authenticated_wallet) ): """ 서명되지 않은 트랜잭션 빌드 클라이언트측 서명을 위해 준비된 트랜잭션 데이터를 반환합니다. 서명은 서버에서 개인 키를 유지하기 위해 클라이언트에서 발생합니다. """ unsigned_tx = 기다리다 transaction_service.build_transaction( wallet_id, 출력=request.outputs, fee_rate=request.fee_rate,change_address=request.change_address) 반품 { "서명되지 않은_tx": base64.b64encode(unsigned_tx.serialize_for_signing()).decode(), "signing_message": base64.b64encode(unsigned_tx.tx_hash()).decode(), "inputs_to_sign": [ { "색인": i, "주소": 입력 주소, "양": 입력량, "파생_경로": 입력.경로 } ~을 위한 나, 인피 in 열거(unsigned_tx.inputs) ], "예상_수수료": unsigned_tx.fee, "추정_크기": unsigned_tx.estimated_size() }

# Kyber + SPHINCS+를 이용한 인증 흐름 수입 오크스 수입 해시립 수입 hmac ~에서 날짜시간 수입 날짜시간, 시간델타 수업 QuantumSafe인증: """ 양자 안전 API 인증 흐름: 1. 클라이언트가 Kyber 공개 키를 보냅니다. 2. 서버가 세션 키를 캡슐화합니다. 3. 클라이언트가 캡슐화를 해제하여 세션 키를 얻습니다. 4. 세션 키를 사용하여 HMAC로 서명된 요청 """ 데프 __초기화__(자체): self.session_store = {} # 프로덕션에서는 Redis를 사용합니다. self.session_duration = timedelta(시간=24) 비동기 정의 개시_세션( self, wallet_id: str, client_kyber_pk: 바이트 ) -> 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: bytes, request_data: bytes, timestamp: int ) -> 선택사항[str]: """ 세션 키를 사용하여 요청 서명을 확인합니다. 유효한 경우 wallet_id를 반환하고 그렇지 않으면 없음을 반환합니다. """ 세션 = self.session_store.get(session_id) 그렇지 않다면 세션: 반품 없음 # 만료 확인 if datetime.utcnow() > 세션["만료_일"]: 델 self.session_store[세션_ID] 반품 없음 # 타임스탬프 확인(재생 방지) request_time = 날짜시간.fromtimestamp(타임스탬프) if abs((datetime.utcnow() - request_time).total_seconds()) > 300: 반품 없음 # 5분 이상 지난/미래 # HMAC 서명 확인 예상_시그 = hmac.new(세션["세션_키"], request_data + str(timestamp).encode(), hashlib.Blake2b ).digest() if hmac.compare_digest(request_signature, Expect_sig): 반품 세션["지갑_ID"] 반품 없음 # 인증된 경로에 대한 FastAPI 종속성 auth_service = QuantumSafe인증() 비동기 정의 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(status_code=401, 세부정보="잘못된 인증") 반품 wallet_id

# 실시간 업데이트를 위한 WebSocket 구현 ~에서 패스트피 수입 웹소켓, 웹소켓연결 끊기 수입 JSON 수입 비동기 수업 연결 관리자: """지갑별 WebSocket 연결 관리""" 데프 __초기화__(자체): self.active_connections: dict[str, List[WebSocket]] = {} 비동기 정의 연결하다(자체, 웹소켓: ​​WebSocket, wallet_id: str): 기다리다 websocket.accept() if wallet_id 안에는 없어 self.active_connections: self.active_connections[wallet_id] = [] self.active_connections[wallet_id].append(websocket) 데프 연결을 끊다(자체, 웹소켓: ​​WebSocket, wallet_id: str): if wallet_id in self.active_connections: self.active_connections[wallet_id].remove(websocket) 비동기 정의 Broadcast_to_wallet(self, wallet_id: str, 메시지: dict): if wallet_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, wallet_id: str): """ 실시간 지갑 업데이트를 위한 WebSocket 이벤트: - Balance_update: 잔액 변경 - transaction_received: 들어오는 거래 - transaction_confirmed: TX가 확인에 도달했습니다. - transaction_sent: 나가는 TX 브로드캐스트 """ # WebSocket 연결 인증 auth_token = websocket.query_params.get("토큰") 기다리지 않으면 verify_ws_token(auth_token, wallet_id): 기다리다 websocket.close(코드=4001) 반품 기다리다 Manager.connect(websocket, wallet_id) 노력하다: # 초기 상태 보내기 기다리다 websocket.send_json({ "유형": "연결됨", "지갑_ID": 지갑_ID, "타임스탬프": datetime.utcnow().isoformat() }) # 들어오는 메시지(구독, 핑)를 처리합니다. ~하는 동안 참: 데이터 = 기다리다 websocket.receive_json() if 데이터.get("유형") == "핑": 기다리다 websocket.send_json({"유형": "퐁"}) 엘리프 데이터.get("유형") == "구독하다": # 특정 주소를 구독하세요 주소 = data.get("구애", []) 기다리다 구독_서비스.구독(지갑_ID, 주소) 제외하고 WebSocketDisconnect: Manager.disconnect(websocket, wallet_id) # 이벤트 방송(블록체인 모니터에 의해 호출) 비동기 정의 방송_균형_업데이트(wallet_id: str, 주소: str, new_balance: int): 기다리다 Manager.broadcast_to_wallet(wallet_id, { "유형": "균형_업데이트", "주소": 주소, "균형": new_balance, "타임스탬프": datetime.utcnow().isoformat() }) 비동기 정의 방송_거래_수신(wallet_id: str, tx_summary: dict): 기다리다 Manager.broadcast_to_wallet(wallet_id, { "유형": "거래_수신", "거래": tx_요약, # 전체 서명이 아닌 요약만 가능 "타임스탬프": datetime.utcnow().isoformat() })

# FastAPI에서 gzip 압축을 활성화합니다. ~에서 fastapi.middleware.gzip 수입 GZipMiddleware app.add_middleware(GZipMiddleware, 최소_크기=1000) # 선택사항: 모바일/임베디드 클라이언트에 대한 MessagePack 응답 ~에서 fastapi.responses 수입 응답 수입 msgpack 수업 MsgPack응답(응답): media_type = "응용 프로그램/msgpack" 데프 세우다(자체, 내용) -> 바이트: 반품 msgpack.packb(content, use_bin_type=True) @app.get("/API/v1/트랜잭션/{tx_id}/바이너리") 비동기 정의 get_transaction_binary(tx_id: 문자열): """MessagePack 형식으로 트랜잭션 가져오기(JSON보다 작음)""" 텍사스 = 기다리다 transaction_service.get_transaction(tx_id) 반품 MsgPack응답(콘텐츠=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/균형") @limiter.limit("100/분") 비동기 정의 get_balance(요청: 요청): ... @app.post("/API/v1/트랜잭션/브로드캐스트") @limiter.limit("5/분") 비동기 정의 방송 거래(요청: 요청): """스팸 방지를 위해 방송 제한을 강화했습니다""" ...

# 표준화된 오류 응답 ~에서 열거형 수입 열거형 수업 오류코드(str, 열거형): INVALID_ADDRESS = "INVALID_ADDRESS" INSUFFICIENT_BALANCE = "INSUFFICIENT_BALANCE" 잘못된_서명 = "INVALID_SIGNATURE" 거래가 거부됨 = "트랜잭션_거부됨" RATE_LIMITED = "RATE_LIMITED" 세션_만료됨 = "세션_만료됨" 파생_실패 = "DERIVATION_FAILED" 수업 API오류(기본 모델): 코드: 오류코드 메시지: 문자열 세부정보: Optional[dict] = 없음 @app.Exception_handler(불충분한 잔액 오류) 비동기 정의 불충분_균형_처리자(요청, 제외): 반품 JSONResponse(상태_코드=400, 내용=API오류( 코드=오류코드.INSUFFICIENT_BALANCE, 메시지="거래 잔액이 부족합니다.", 세부정보={ "사용 가능": 제외 가능, "필수의": 제외필수, "부족": exc.required - ex.available } ).dict() )