Постквантовий дизайн гаманця API: шаблони REST і WebSocket

# Реалізація кінцевих точок адреси (FastAPI) від фастапі імпорт FastAPI, залежно, HTTPException від пидантичний імпорт BaseModel від введення тексту імпорт Список, необов'язковий імпорт програма base64 = FastAPI(title="Гаманець SynX API", версія="1.0.0") клас AddressResponse(базова модель): """Адреса з постквантовими відкритими ключами""" адреса: str шлях: str kyber_public_key: вул # Закодовано Base64 (1184 байти) sphincs_public_key: вул # Кодування Base64 (32 байти) баланс: int pending_balance: int created_at: вул клас DeriveAddressRequest(BaseModel): обліковий запис: int = 0 зміна: int = 0 index: необов’язковий [int] = None # Автоматичне збільшення, якщо немає @app.get("/API/v1/адреси", response_model=Список[AddressResponse]) асинхронне визначення список_адрес(wallet_id: str = Depends(get_authenticated_wallet), skip: int = 0, limit: int = 50): """ Перелік адрес гаманців із балансами. Примітка: відкриті ключі Kyber великі (1,2 КБ). Для переліку подумайте про виключення ключів і вибірку окремо. """ адреси = чекати address_service.list_addresses( wallet_id, skip=пропустити, limit=ліміт) повернення [ AddressResponse( адреса=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/addresses/derive", модель_відповіді=AddressResponse) асинхронне визначення отримати_адресу( запит: DeriveAddressRequest, wallet_id: str = Depends(get_authenticated_wallet)): """Отримати нову адресу за вказаним шляхом отримання""" адреса = чекати address_service.derive_address( wallet_id, account=request.account, change=request.change, index=request.index ) повернення AddressResponse(...)

клас TransactionSummary(базова модель): """Трансакція без повних даних підпису (для списків)""" tx_id: str timestamp: str inputs_count: int outputs_count: int сума: int комісія: int підтвердження: int статус: str # "очікує", "підтверджено", "не вдалося" клас TransactionFull(базова модель): """Повна транзакція, включаючи підписи""" tx_id: str версія: int timestamp: str inputs: список["TransactionInputSchema"] виводить: List["TransactionOutputSchema"] плата: int підтвердження: int block_hash: необов’язковий [str] raw_hex: str # Повна серіалізована транзакція клас TransactionInputSchema(BaseModel): prev_tx_id: str prev_output_index: int сума: int адреса: str підпис: str # Base64 (~10,5 КБ для SPHINCS+-SHAKE-128s, 7856 необроблених байтів) відкритий_ключ: вул # Base64 (44 байти для SPHINCS+) клас BuildTransactionRequest(BaseModel): виводить: List['OutputSpec'] fee_rate: необов’язковий [int] = немає # Автоматичний розрахунок, якщо немає change_address: необов’язковий [str] = немає # Автоматичний вибір, якщо немає клас OutputSpec(BaseModel): одержувач: str сума: int memo: необов’язковий [str] = немає @app.get("/API/v1/транзакції", response_model=Список[TransactionSummary]) асинхронне визначення list_transactions(wallet_id: str = Depends(get_authenticated_wallet), skip: int = 0, ліміт: int = 20, status: Optional[str] = None): """ Список транзакцій (тільки підсумки) Підписи виключено з відповідей списку, щоб зменшити корисне навантаження. Використовуйте GET /transactions/{tx_id} для повної транзакції з підписами. """ txs = чекати transaction_service.list_transactions( wallet_id, skip=пропустити, limit=ліміт, status=status ) повернення [tx.to_summary() для tx in txs] @app.get("/API/v1/transactions/{tx_id}", модель_відповіді=TransactionFull) асинхронне визначення 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(status_code=404, detail="Транзакція не знайдена") повернення tx.to_full_schema(include_signatures=include_signatures) @app.post("/API/v1/transactions/build") асинхронне визначення build_transaction( запит: BuildTransactionRequest, wallet_id: str = Depends(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(), "inputs_to_sign": [ { "індекс": i, "адреса": вхідна адреса, "сума": вхідна сума, "дериваційний_шлях": inp.path } для я, вп in enumerate(unsigned_tx.inputs)], "орієнтована_плата": unsigned_tx.fee, "приблизний_розмір": unsigned_tx.estimated_size() }

# Потік автентифікації за допомогою Kyber + SPHINCS+ імпорт oqs імпорт хешліб імпорт hmac від датачас імпорт datetime, timedelta клас QuantumSafeAuth: """ Потік квантово-безпечної автентифікації API: 1. Клієнт надсилає відкритий ключ Kyber 2. Сервер інкапсулює ключ сеансу 3. Клієнт декапсулює, щоб отримати ключ сеансу 4. Запити, підписані за допомогою HMAC за допомогою ключа сеансу """ зах __init__(self): self.session_store = {} # У виробництві використовуйте Redis self.session_duration = дельта часу (годин=24) асинхронне визначення initiate_session( self, wallet_id: str, client_kyber_pk: bytes ) -> dict: """ Крок 1: Клієнт ініціює сеанс за допомогою відкритого ключа Kyber. Сервер інкапсулює секрет сеансу в ключ клієнта """ kem = oqs.KeyEncapsulation("Kyber768") зашифрований текст, shared_secret = kem.encap_secret(client_kyber_pk) # Отримати ключ сеансу зі спільного секрету session_key = hashlib.shake_256( shared_secret + b"сеансовий ключ" ).дайджест(32) # Створити ідентифікатор сеансу session_id = hashlib.Blake2b( shared_secret + str(datetime.utcnow()).encode(), digest_size=16).hexdigest() # Сеанс магазину (на стороні сервера) self.session_store[session_id] = { "wallet_id": wallet_id, "session_key": session_key, "expires_at": datetime.utcnow() + self.session_duration, "created_at": datetime.utcnow() } повернення { "ідентифікатор_сесії": session_id, "зашифрований текст": base64.b64encode(ciphertext).decode(), "expires_at": (datetime.utcnow() + self.session_duration).isoformat() } зах verify_request( self, session_id: str, request_signature: bytes, request_data: bytes, timestamp: int ) -> Додатково [str]: """ Перевірити підпис запиту за допомогою ключа сеансу. Повертає wallet_id, якщо дійсний, в іншому випадку немає """ session = self.session_store.get(session_id) якщо ні сесія: повернення Жодного # Перевірити термін дії if datetime.utcnow() > session["expires_at"]: дел self.session_store[session_id] повернення Жодного # Перевірити позначку часу (запобігти повторному відтворенню) запит_час = datetime.fromtimestamp(часова позначка) if abs((datetime.utcnow() - request_time).total_seconds()) > 300: повернення Жодного # Понад 5 хвилин тому/в майбутньому # Перевірити підпис HMAC очікуваний_сиг = hmac.new( сеанс["session_key"], запит_даних + str(мітка часу).encode(), hashlib.Blake2b ).digest() if hmac.compare_digest(request_signature, expect_sig): повернення сесія["wallet_id"] повернення Жодного # Залежність FastAPI для автентифікованих маршрутів auth_service = QuantumSafeAuth() асинхронне визначення 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) ) якщо ні wallet_id: підняти HTTPException(status_code=401, detail="Недійсна автентифікація") повернення wallet_id

# Реалізація WebSocket для оновлень у реальному часі від фастапі імпорт WebSocket, WebSocketDisconnect імпорт json імпорт asyncio клас ConnectionManager: """Керувати підключеннями WebSocket для кожного гаманця""" зах __init__(self): self.active_connections: dict[str, List[WebSocket]] = {} асинхронне визначення підключити(self, 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) зах відключити(self, 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, message: dict): if wallet_id in self.active_connections: dead_connections = [] для підключення in self.active_connections[wallet_id]: try: чекати connection.send_json(повідомлення) крім: dead_connections.append(з'єднання) # Очищення мертвих з'єднань для конн in dead_connections: self.active_connections[wallet_id].remove(conn) manager = ConnectionManager() @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) try: # Надіслати початковий стан чекати websocket.send_json({ "тип": "підключений", "wallet_id": wallet_id, "мітка часу": datetime.utcnow().isoformat() }) # Обробляти вхідні повідомлення (підписки, ping) поки Правда: дані = чекати websocket.receive_json() if data.get("тип") == "ping": чекати websocket.send_json({"тип": "понг"}) еліф data.get("тип") == "підписатися": # Підпишіться на певні адреси адреси = data.get("адреси", []) чекати subscription_service.subscribe( wallet_id, адреси ) крім WebSocketDisconnect: manager.disconnect(websocket, wallet_id) # Трансляція подій (викликається монітором блокчейну) асинхронне визначення broadcast_balance_update(wallet_id: str, адреса: str, new_balance: int): чекати manager.broadcast_to_wallet(wallet_id, { "тип": "balance_update", "адреса": адреса, "баланс": новий_баланс, "мітка часу": datetime.utcnow().isoformat() }) асинхронне визначення broadcast_transaction_received(wallet_id: str, tx_summary: dict): чекати manager.broadcast_to_wallet(wallet_id, { "тип": "transaction_received", "транзакція": tx_summary, # Лише резюме, а не повний підпис "мітка часу": datetime.utcnow().isoformat() })

# Увімкніть стиснення gzip у FastAPI від fastapi.middleware.gzip імпорт GZipMiddleware app.add_middleware(GZipMiddleware, minimum_size=1000) # Додатково: відповіді MessagePack для мобільних/вбудованих клієнтів від fastapi.відповіді імпорт Відповідь імпорт msgpack клас MsgPackResponse(Відповідь): media_type = "програма/msgpack" зах рендерити(власний, вміст) -> байти: повернення msgpack.packb(вміст, use_bin_type=True) @app.get("/API/v1/transactions/{tx_id}/binary") асинхронне визначення get_transaction_binary(tx_id: str): """Отримати транзакцію у форматі MessagePack (менше JSON)""" tx = чекати transaction_service.get_transaction(tx_id) повернення MsgPackResponse(content=tx.to_dict())

# Обмеження швидкості для гаманця API від slowapi імпорт Обмежувач, _rate_limit_exceeded_handler від slowapi.errors імпорт RateLimitExceeded limiter = Limiter(key_func=get_wallet_id_from_request) app.state.limiter = 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/transactions/broadcast") @limiter.limit("5/хвилина") асинхронне визначення широкомовна_транзакція(запит: запит): """Суворіше обмеження для трансляції для запобігання спаму""" ...

# Стандартизовані відповіді на помилки від перелік імпорт Enum клас ErrorCode(str, Enum): INVALID_ADDRESS = "INVALID_ADDRESS" INSUFFICIENT_BALANCE = "INSUFFICIENT_BALANCE" INVALID_SIGNATURE = "INVALID_SIGNATURE" TRANSACTION_REJECTED = "TRANSACTION_REJECTED" RATE_LIMITED = "RATE_LIMITED" SESSION_EXPIRED = "SESSION_EXPIRED" DERIVATION_FAILED = "DERIVATION_FAILED" клас APIError(Базова модель): код: ErrorCode повідомлення: str деталі: Необов’язковий [dict] = немає @app.exception_handler(InsufficientBalanceError) асинхронне визначення insufficient_balance_handler(запит, викл): повернення JSONResponse( status_code=400, content=APIError(код=ErrorCode.INSUFFICIENT_BALANCE, повідомлення=«Недостатній баланс для транзакції», деталі={ "доступний": відмінно доступний, "потрібно": необов. "дефіцит": exc.required - exc.available } ).dict() )