Исправляем ошибку PEM приватного ключа в Coinbase Advanced Trade API на FastAPI

Разбираем сбой cryptography при интеграции Coinbase Advanced Trade с FastAPI: ошибка PEM приватного ключа, причины и решение с генерацией корректного JWT.

Интеграция Coinbase Advanced Trade API в сервис на FastAPI может неожиданно уткнуться в проблему криптографии: приватный ключ не загружается. Типичный симптом — сервер запускается без ошибок, но первый аутентифицированный вызов падает с ошибкой разбора PEM. Ниже — краткий разбор сценария сбоя, его причин и способа починки, причём менять саму схему аутентификации не придётся.

Обзор проблемы

Приложение FastAPI считывает учетные данные API из переменных окружения и инициализирует клиент Coinbase REST. Сервер стартует нормально, но первый запрос к защищённой конечной точке падает с ошибкой криптографии, указывающей на приватный ключ. В трассировке стека выделяется путь формирования JWT внутри клиента и упоминаются проблемы разбора PEM.

Ключевая ошибка выглядит так:

Не удалось загрузить PEM-файл. Подробнее: https://cryptography.io/en/latest/faq/#why-can-t-i-import-my-pem-file. MalformedFraming Вы уверены, что сгенерировали ключ на https://cloud.coinbase.com/access/api ?

Код для воспроизведения (что приводит к ошибке)

Приложение подключает клиент Coinbase REST, используя значения из файла .env, а затем вызывает endpoint accounts. Логика ниже в общих чертах повторяет эту схему.

# app/services/trade_api.py
import logging
from app.core.config import CB_API_KEY, CB_API_SECRET
from coinbase.rest import RESTClient
logger = logging.getLogger(__name__)
try:
    cb_client = RESTClient(api_key=CB_API_KEY, api_secret=CB_API_SECRET, verbose=True)
except Exception as exc:
    logger.error("Could not create RESTClient with provided credentials")
    logger.exception(exc)
    raise
def fetch_accounts():
    logger.info(f"Key in use: {CB_API_KEY}")
    try:
        return cb_client.get_accounts()
    except Exception as exc:
        logger.error("Failed to call Coinbase accounts endpoint")
        logger.exception(exc)
        return {"error": str(exc)}

Когда конечная точка вызывает fetch_accounts, клиент пытается подписать запрос, внутренне собирает JWT, и затем cryptography падает на загрузке PEM. Трассировка показывает путь сбоя через coinbase.rest и coinbase.jwt_generator к функции load_pem_private_key библиотеки cryptography.

Что на самом деле идёт не так

Клиент Coinbase ожидает корректный PEM-кодированный приватный ключ EC, который cryptography может загрузить. Если в COINBASE_API_SECRET попало значение, не разбираемое как PEM, cryptography выдаёт ошибку вроде MalformedFraming. Это возможно, если ключевой материал не в формате PEM или при сохранении в окружении потерялись важные детали форматирования — переносы строк и разделители. Результат один: сборщик JWT не может десериализовать ключ, и запрос так и не подписывается.

Исправление: преобразуйте секрет в корректный PEM и подписывайте запросы

Чтобы разблокировать процесс, преобразуйте секрет в правильный SECP256K1 PEM и используйте его для генерации JWT. Сниппет ниже делает именно это: декодирует секрет из base64, нормализует длину до 32 байт под скаляр, выводит приватный ключ EC на SECP256K1, экспортирует его в PEM и передаёт в jwt_generator из Coinbase.

# app/integrations/jwt_tools.py
import os
import base64
import logging
from coinbase import jwt_generator
from cryptography.hazmat.primitives.asymmetric import ec
from cryptography.hazmat.primitives import serialization
log = logging.getLogger(__name__)
API_KEY_VALUE = os.getenv("COINBASE_API_KEY")
SECRET_B64_VALUE = os.getenv("COINBASE_API_SECRET")
def wrap_secret_as_pem(b64_secret: str) -> str:
    try:
        decoded = base64.b64decode(b64_secret)
        # Если буфер 64 байта, оставляем первые 32 байта как скаляр
        if len(decoded) == 64:
            decoded = decoded[:32]
        if len(decoded) != 32:
            raise ValueError(f"Expected 32 bytes for SECP256K1 secret, got {len(decoded)} bytes")
        scalar = int.from_bytes(decoded, "big")
        ec_priv = ec.derive_private_key(scalar, ec.SECP256K1())
        pem_bytes = ec_priv.private_bytes(
            encoding=serialization.Encoding.PEM,
            format=serialization.PrivateFormat.TraditionalOpenSSL,
            encryption_algorithm=serialization.NoEncryption()
        )
        return pem_bytes.decode("utf-8")
    except Exception as err:
        log.error(f"Failed to produce PEM from secret: {err}")
        raise
PEM_MATERIAL = wrap_secret_as_pem(SECRET_B64_VALUE)
def issue_jwt(
    http_method: str = "GET",
    resource_path: str = "/v2/accounts"
) -> str:
    log.info(f"Using API key: {API_KEY_VALUE}")
    log.info(f"PEM material prepared: {PEM_MATERIAL}")
    uri = jwt_generator.format_jwt_uri(http_method, resource_path)
    token = jwt_generator.build_rest_jwt(uri, API_KEY_VALUE, PEM_MATERIAL)
    return token

Суть проста: передайте cryptography валидный EC-ключ в формате PEM. Как только такой PEM получен, jwt_generator.build_rest_jwt его принимает, и клиент сможет успешно авторизовать вызовы.

Почему это важно

Когда вы работаете с API, где запросы подписываются, корректность формата приватного ключа — не вопрос эстетики. Слой подписи находится на критическом пути каждого защищённого вызова. Если ключ неверный, библиотека будет падать рано и снова, и вместо бизнес-логики вы будете отлаживать инфраструктуру и криптографию. В данном случае сбой исходит от строгого загрузчика ключей в cryptography — это хорошо для безопасности, но означает, что значение в окружении должно быть точным до символа.

Итоги

Убедитесь, что секрет, который вы передаёте, — это загружаемый PEM приватного ключа SECP256K1. Если значение приходит как base64‑кодированные байты, сначала преобразуйте его в корректный PEM, а уже потом передавайте сборщику JWT. Следите за целостностью ключевого материала при хранении в переменных окружения, чтобы разделители и переводы строк сохранялись на всём пути. С валидным PEM клиент Coinbase сможет собирать JWT, и конечная точка accounts будет отвечать как ожидается.