Дизайн пост-квантового кошелька API: шаблоны REST и WebSocket

# Реализация адресных конечных точек (FastAPI) от фастапи импортировать FastAPI, Зависит, HTTPException от пидантический импортировать Базовая модель от печатание импортировать Список, необязательно импортировать приложение base64 = FastAPI(title="SynX Кошелек API", версия="1.0.0") сорт АдресОтвет(Базовая модель): """Адрес с постквантовыми открытыми ключами""" адрес: str путь: str kyber_public_key: str # Кодировка Base64 (1184 байта) sphincs_public_key: стр # Кодировка Base64 (32 байта) баланс: int pending_balance: int созданный_at: str сорт Деривеадресзапрос(BaseModel): учетная запись: int = 0 изменение: int = 0 индекс: необязательный [int] = нет # Автоинкремент, если нет @app.get("/API/v1/адреса", response_model=Список[АдресОтвет]) асинхронная защита список_адресов(wallet_id: str = Depends(get_authenticated_wallet), пропуск: int = 0, предел: int = 50): """ Перечислите адреса кошельков с балансами. Примечание. Открытые ключи Kyber имеют большой размер (1,2 КБ). Для включения в список рассмотрите возможность исключения ключей и их выборки отдельно. """ адреса = ждать адрес_сервис.список_адресов(кошелек_ид, пропустить=пропустить, лимит=лимит) возвращаться [ АдресОтвет( адрес=адрес.адрес, путь=адрес.путь, kyber_public_key=base64.b64encode(addr.kyber_pk).decode(), sphincs_public_key=base64.b64encode(addr.sphincs_pk).decode(), баланс=addr.balance, pending_balance=addr.pending_balance, созданный_at=addr.created_at.isoformat() ) для адрес in адреса] @app.post("/API/v1/адреса/производные", модель_ответа=АдресОтвет) асинхронная защита производный_адрес( запрос: Деривеадресзапрос, Wallet_id: str = Depends(get_authenticated_wallet) ): """Получить новый адрес по указанному пути деривации""" адрес = ждать адрес_сервис.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 временная метка: входы str: List['ТранзакцияИнпутСхема'] выходные данные: Список[«ТранзакцияВыходнаяСхема»] плата: int подтверждения: int block_hash: Необязательный[str] raw_hex: str # Полная сериализованная транзакция сорт Транзакционинпутсхема(BaseModel): prev_tx_id: str prev_output_index: int сумма: int адрес: str подпись: str # Base64 (~10,5 КБ для SPHINCS+-SHAKE-128, 7856 необработанных байт) открытый_ключ: ул # Base64 (44 байта для SPHINCS+) сорт Билдтранзакционрекуест(Базовая модель): выходные данные: Список[«Выходная спецификация»] Fee_rate: Необязательный[int] = Нет # Автоматический расчет, если нет Change_address: Необязательный[str] = Нет # Автоматический выбор, если нет сорт Выходная спецификация(BaseModel): получатель: str сумма: int memo: необязательно [str] = нет @app.get("/API/v1/транзакции", response_model=Список[Сводка транзакции]) асинхронная защита список_транзакций( Wallet_id: str = Depends(get_authenticated_wallet), пропуск: int = 0, предел: int = 20, статус: Необязательно[str] = Нет): """ Транзакции списка (только сводки). Подписи исключаются из ответов списка, чтобы уменьшить полезную нагрузку. Используйте GET /transactions/{tx_id} для полной транзакции с подписями. """ ТХС = ждать транзакция_service.list_transactions( Wallet_id, Skip=skip, Limit=Limit, Status=Status) возвращаться [tx.to_summary() для tx in спасибо] @app.get("/API/v1/транзакции/{tx_id}", модель_ответа=ТранзакцияПолная) асинхронная защита get_transaction(tx_id: str, Wallet_id: str = Depends(get_authenticated_wallet), include_signatures: bool = True): """ Получение полной информации о транзакции. Установите include_signatures=false, чтобы уменьшить размер ответа, если вам нужны только метаданные транзакции. """ передача = ждать транзакция_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( запрос: Билдтранзакционрекуест, Wallet_id: str = Depends(get_authenticated_wallet) ): """ Создать неподписанную транзакцию. Возвращает данные транзакции, готовые для подписания на стороне клиента. Подписание происходит на клиенте, чтобы сохранить закрытые ключи на сервере. """ unsigned_tx = ждать транзакция_service.build_transaction(wallet_id, выходы=запрос.выходы, плата_ставка=запрос.комиссия_ставка, изменение_адрес=запрос.изменить_адрес) возвращаться { "беззнаковый_tx": base64.b64encode(unsigned_tx.serialize_for_signing()).decode(), "подписное_сообщение": 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 от дата и время импортировать дата-время, дельта времени сорт QuantumSafeAuth: """ Квантовобезопасная аутентификация API. Поток: 1. Клиент отправляет открытый ключ Kyber 2. Сервер инкапсулирует сеансовый ключ 3. Клиент декапсулирует для получения сеансового ключа 4. Запросы подписываются с помощью HMAC с использованием сеансового ключа """ защита __init__(сам): self.session_store = {} # В производстве используйте Redis self.session_duration = timedelta(часы=24) асинхронная защита инициировать_сессию(self, Wallet_id: str, client_kyber_pk: байты) -> dict: """ Шаг 1: Клиент инициирует сеанс с открытым ключом Kyber. Сервер инкапсулирует секрет сеанса в ключ клиента """ кем = oqs.KeyEncapsulation("Кибер768") зашифрованный текст,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(), ignore_size=16).hexdigest() # Сохранение сеанса (на стороне сервера) self.session_store[session_id] = { "идентификатор_кошелька": идентификатор_кошелька, "ключ_сессии": ключ_сеанса, "истекает_в": datetime.utcnow() + self.session_duration, "создано_at": datetime.utcnow() } возвращаться { "ИД_сеанса": идентификатор сеанса, "зашифрованный текст": base64.b64encode(зашифрованный текст).decode(), "истекает_в": (datetime.utcnow() + self.session_duration).isoformat() } защита проверить_запрос(self, session_id: str, request_signature: байты, request_data: байты, метка времени: int) -> Необязательный[str]: """ Проверка подписи запроса с использованием сеансового ключа. Возвращает Wallet_id, если он действителен, в противном случае None """ сеанс = self.session_store.get(session_id) если не сессия: возвращаться Никто # Проверить срок действия if datetime.utcnow() > сеанс["истекает_в"]: дель self.session_store[session_id] возвращаться Никто # Проверить временную метку (предотвратить повтор) request_time = datetime.fromtimestamp(временная метка) if abs((datetime.utcnow() - request_time).total_секунды()) > 300: возвращаться Никто # Более 5 минут назад/в будущем # Проверяем подпись HMAC ожидаемый_сиг = hmac.new(сессия["ключ_сессии"], request_data + str(timestamp).encode(), hashlib.Blake2b ).digest() if hmac.compare_digest(request_signature, ожидаемая_подпись): возвращаться сессия["идентификатор_кошелька"] возвращаться Никто # Зависимость FastAPI для аутентифицированных маршрутов auth_service = QuantumSafeAuth() асинхронная защита get_authenticated_wallet( 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) ) если не идентификатор_кошелька: поднимать HTTPException(status_code=401, подробно=«Неверная аутентификация») возвращаться Wallet_id

# Реализация WebSocket для обновлений в реальном времени от фастапи импортировать Вебсокет, Вебсокетдисконнект импортировать JSON импортировать асинхронный сорт Диспетчер соединений: """Управление соединениями WebSocket для каждого кошелька""" защита __init__(self): self.active_connections: dict[str, List[WebSocket]] = {} асинхронная защита соединять(я, веб-сокет: WebSocket, Wallet_id: str): ждать вебсокет.принять() 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(я, Wallet_id: str, сообщение: dict): if Wallet_id in self.active_connections: Dead_connections = [] для связь in self.active_connections[wallet_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: баланс изменен - ​​транзакции_получены: входящая транзакция - транзакции_подтверждены: подтверждены достижения TX - транзакции_sent: исходящая трансляция TX """ # Аутентификация соединения WebSocket auth_token = websocket.query_params.get("жетон") если не ждать validate_ws_token(auth_token, Wallet_id): ждать вебсокет.закрыть (код = 4001) возвращаться ждать Manager.connect(websocket, Wallet_id) пытаться: # Отправляем исходное состояние ждать websocket.send_json({ "тип": "связанный", "идентификатор_кошелька": идентификатор_кошелька, "метка времени": datetime.utcnow().isoformat() }) # Обработка входящих сообщений (подписки, пинги) пока Верно: данные = ждать вебсокет.receive_json() if data.get("тип") == "пинг": ждать websocket.send_json({"тип": "понг"}) Элиф data.get("тип") == "подписаться": # Подписаться на определенные адреса адреса = data.get("адреса", []) ждать подписка_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, { "тип": "транзакция_получена", "сделка": tx_summary, # Только краткое описание, а не полная подпись "метка времени": datetime.utcnow().isoformat() })

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

# Ограничение скорости для кошелька API от медленный импортировать Ограничитель, _rate_limit_exceeded_handler от медленный API.errors импортировать Ограничитель RateLimitExceeded = Limiter(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, Enum): INVALID_ADDRESS = "INVALID_ADDRESS" НЕДОСТАТОЧНЫЙ_БАЛАНС = «НЕДОСТАТОЧНЫЙ_БАЛАНС» INVALID_SIGNATURE = "INVALID_SIGNATURE" TRANSACTION_REJECTED = "TRANSACTION_REJECTED" RATE_LIMITED = "RATE_LIMITED" СЕССИЯ_EXPIRED = "СЕССИЯ_EXPIRED" DERIVATION_FAILED = "DERIVATION_FAILED" сорт Ошибка API(Базовая модель): код: Код ошибки сообщение: информация о строке: Необязательный[dict] = Нет @app.Exception_handler(Ошибка недостаточного баланса) асинхронная защита недостаточный_баланс_обработчик(запрос, исключение): возвращаться JSONResponse(status_code=400, content=Ошибка API(код=Код ошибки.INSUFFICIENT_BALANCE, сообщение=«Недостаточно баланса для транзакции», подробности={ "доступный": доступно, "необходимый": обязательно, "дефицит": exc.required - exc.available } ).dict() )