Межсервисная аутентификация в FastAPI: mTLS и Client Credentials

← назад к разделу

Когда order-service обращается к payment-service внутри кластера, кажется, что это безопасно: закрытая сеть, VPC, никто посторонний не зайдёт. Но если один контейнер окажется скомпрометирован через уязвимость в зависимости — атакующий сможет ходить на все соседние сервисы без ограничений, потому что они его «не спрашивают».

Решение — каждый вызов между сервисами должен нести доказательство личности вызывающего. Есть два способа организовать это в Python: mTLS через Service Mesh или Client Credentials Flow с токеном от IdP.

Способ 1: mTLS через Service Mesh

mTLS расшифровывается как mutual TLS — двусторонняя TLS-аутентификация. Обычный HTTPS проверяет только сервер (его сертификат). mTLS проверяет обе стороны: и сервер, и клиент предъявляют сертификаты.

В Kubernetes это реализует Service Mesh — Istio или Linkerd. Каждому поду автоматически выдаётся уникальный сертификат (стандарт SPIFFE). Sidecar-контейнер рядом с вашим приложением перехватывает весь входящий и исходящий трафик, шифрует его и проверяет сертификаты.

Главное преимущество: ваш Python-код не пишет ни строчки для аутентификации. Всё происходит на уровне транспорта, до приложения.

# adapters/out/http/payment_client.py
import httpx
from app.config import settings


class PaymentClient:
    def __init__(self) -> None:
        self._client = httpx.AsyncClient(
            base_url=settings.payment_service_url,
        )

    async def charge(self, order_id: str, amount_cents: int) -> dict:
        resp = await self._client.post(
            "/charge",
            json={"order_id": order_id, "amount_cents": amount_cents},
        )
        resp.raise_for_status()
        return resp.json()

Никаких заголовков Authorization добавлять не нужно — sidecar делает это за вас. Принимающий сервис получает идентификатор вызывающего через заголовок X-Forwarded-Client-Cert (Istio) или через политику PeerAuthentication.

Istio автоматически обновляет сертификаты каждые 24 часа — ротация бесплатная. И способ одинаково работает для Java-, Go- и Python-сервисов в одном кластере.

Единственный минус: нужна инфраструктура Service Mesh. В локальной разработке или в окружении без Istio/Linkerd — нужен второй способ.

Способ 2: Client Credentials Flow с authlib

Это стандартный OAuth2-поток для машинных клиентов. Сервис запрашивает у IdP (например, Keycloak) свой собственный access_token — не токен пользователя, а именно машинный — и прикладывает его к каждому запросу.

Схема выглядит так:

Секреты через pydantic-settings

client_secret нельзя писать прямо в коде. Читаем из переменных окружения через pydantic-settings:

# config.py
from pydantic_settings import BaseSettings


class Settings(BaseSettings):
    payment_service_url: str
    idp_token_url: str
    payment_client_id: str
    payment_client_secret: str

    model_config = {"env_file": ".env"}


settings = Settings()

OAuth2-клиент с authlib

authlib предоставляет AsyncOAuth2Client — он хранит токен, следит за сроком его жизни и перед каждым запросом перезапрашивает его у IdP, если срок вышел. Единственное, что остаётся на вас, — получить первый токен: без него клиент откажется делать запрос с MissingTokenError:

# adapters/out/http/payment_client.py
from authlib.integrations.httpx_client import AsyncOAuth2Client
from app.config import settings


class PaymentClient:
    def __init__(self) -> None:
        self._client = AsyncOAuth2Client(
            client_id=settings.payment_client_id,
            client_secret=settings.payment_client_secret,
            token_endpoint=settings.idp_token_url,
            grant_type="client_credentials",
            scope="payment:charge",
        )

    async def _ready(self) -> None:
        if not self._client.token:
            await self._client.fetch_token()

    async def charge(self, order_id: str, amount_cents: int) -> dict:
        await self._ready()
        resp = await self._client.post(
            f"{settings.payment_service_url}/charge",
            json={"order_id": order_id, "amount_cents": amount_cents},
        )
        resp.raise_for_status()
        return resp.json()

    async def aclose(self) -> None:
        await self._client.aclose()

fetch_token() вызывается один раз, при первом обращении. Дальше каждый post и get внутри AsyncOAuth2Client сам сверяет срок токена и для client_credentials перезапрашивает его у IdP — ручное кеширование и обновление не нужны.

Отдельный scope на каждую операцию

Частая ошибка — один токен с широким scope на все операции. Если этот токен утечёт, атакующий получает доступ ко всему. Правильнее — scope per operation:

# adapters/out/http/payment_charge_client.py
class PaymentChargeClient:
    def __init__(self) -> None:
        self._client = AsyncOAuth2Client(
            client_id=settings.payment_client_id,
            client_secret=settings.payment_client_secret,
            token_endpoint=settings.idp_token_url,
            grant_type="client_credentials",
            scope="payment:charge",     # только charge, не refund
        )

# adapters/out/http/inventory_client.py
class InventoryClient:
    def __init__(self) -> None:
        self._client = AsyncOAuth2Client(
            client_id=settings.inventory_client_id,
            client_secret=settings.inventory_client_secret,
            token_endpoint=settings.idp_token_url,
            grant_type="client_credentials",
            scope="inventory:reserve",  # только reserve
        )

Это называется blast-radius containment: если что-то пойдёт не так с одним клиентом, ущерб ограничен только его операциями.

Проверка на принимающей стороне

payment-service получает запрос с токеном и должен понять, что это машинный вызов (не пользователь). Роль system маппится из scope JWT-токена:

def _extract_roles(claims: dict) -> list[str]:
    scope = claims.get("scope", "")
    if "payment:charge" in scope.split():
        return ["system"]
    realm = claims.get("realm_access", {})
    return realm.get("roles", [])

Endpoint объявляет требуемую роль явно:

# adapters/in/http/payment_router.py
from fastapi import APIRouter, Depends
from adapters.in.http.security import Principal, require_roles
from application.use_cases.charge_payment import ChargePaymentUseCase

router = APIRouter(prefix="/charge")


@router.post("/", status_code=200)
async def charge(
    body: ChargeRequest,
    principal: Principal = Depends(require_roles("system")),
    use_case: ChargePaymentUseCase = Depends(),
) -> ChargeResponse:
    return await use_case.execute(body, principal)

Анонимный трафик — частая ошибка

Вот как выглядит проблемный код:

# Неправильно — любой pod в кластере вызовет без ограничений
class ProductClient:
    def __init__(self) -> None:
        self._client = httpx.AsyncClient(base_url="http://product-service")

    async def get_product(self, product_id: str) -> dict:
        resp = await self._client.get(f"/products/{product_id}")
        resp.raise_for_status()
        return resp.json()

Один скомпрометированный контейнер — и атакующий перемещается по всему кластеру. Правильная версия — либо mTLS (sidecar добавляет аутентификацию), либо AsyncOAuth2Client:

# adapters/out/http/product_client.py
from authlib.integrations.httpx_client import AsyncOAuth2Client
from app.config import settings


class ProductClient:
    def __init__(self) -> None:
        self._client = AsyncOAuth2Client(
            client_id=settings.product_client_id,
            client_secret=settings.product_client_secret,
            token_endpoint=settings.idp_token_url,
            grant_type="client_credentials",
            scope="product:read",
        )

    async def get_product(self, product_id: str) -> dict:
        if not self._client.token:
            await self._client.fetch_token()
        resp = await self._client.get(
            f"{settings.product_service_url}/products/{product_id}"
        )
        resp.raise_for_status()
        return resp.json()

Регистрация клиентов как синглтонов

Клиенты регистрируются в lifespan, а не создаются на каждый запрос. Это позволяет переиспользовать HTTP-соединения и не терять кеш токена:

# main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI
from adapters.out.http.payment_client import PaymentClient
from adapters.out.http.inventory_client import InventoryClient


@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.payment_client = PaymentClient()
    app.state.inventory_client = InventoryClient()
    yield
    await app.state.payment_client.aclose()
    await app.state.inventory_client.aclose()


app = FastAPI(lifespan=lifespan)

# adapters/in/http/orders_router.py
from fastapi import APIRouter, Depends, Request

router = APIRouter(prefix="/orders")


def get_payment_client(request: Request) -> PaymentClient:
    return request.app.state.payment_client


@router.post("/", status_code=201)
async def create_order(
    body: CreateOrderRequest,
    principal: Principal = Depends(require_roles("customer", "admin")),
    payment_client: PaymentClient = Depends(get_payment_client),
    use_case: CreateOrderUseCase = Depends(),
) -> OrderResponse:
    return await use_case.execute(body, principal, payment_client)

Глубже: от имени пользователя: обмен токена по RFC 8693расширенное

Статья оставляет открытым вопрос, который встаёт в первый же день: сервис заказов вызывает сервис оплаты, и оплата должна знать, какой покупатель платит. Пробросить дальше токен покупателя нельзя: он выписан для сервиса заказов (aud), у него права покупателя, и оплата по нему не отличит честный вызов от украденного. Машинный токен из Client Credentials не несёт покупателя вовсе.

Решение называется обменом токена: сервис заказов предъявляет IdP токен покупателя и свои учётные данные и получает новый токен, где sub покупателя сохранён, aud равен payment-service, а scope сужен до нужного. В authlib это тот же fetch_token с другим типом гранта:

token = await self._client.fetch_token(
    settings.idp_token_url,
    grant_type="urn:ietf:params:oauth:grant-type:token-exchange",
    subject_token=user_token,
    subject_token_type="urn:ietf:params:oauth:token-type:access_token",
    audience="payment-service",
)

Принимающая сторона проверяет его как обычный JWT: подпись, aud, exp, и видит в sub покупателя, а в claim act или azp сервис, который действовал от его имени; журнал действий пишет обоих.

Цена: вызов к IdP на каждого пользователя, поэтому результат кешируют по паре «пользователь, аудитория» до exp. Keycloak такой обмен поддерживает, его включают на клиенте сервиса заказов. Альтернатива без обмена, положить user_id в тело или заголовок машинного вызова, работает только внутри полностью доверенного контура: оплата верит сервису заказов на слово, и компрометация заказов открывает все кошельки.

Коротко

• Внутри кластера тоже нужна аутентификация — изолированная сеть не защищает от скомпрометированного контейнера.
• Два способа: mTLS через Service Mesh (код ничего не делает — sidecar берёт на себя) или Client Credentials Flow (сервис получает машинный токен от IdP).
• В Client Credentials Flow используй authlib.AsyncOAuth2Client: первый токен получают fetch_token(), дальше клиент сам обновляет его перед запросами.
• Секрет (client_secret) — только через переменные окружения, pydantic-settings. Не в код, не в git.
• Один клиент — один scope per operation (payment:charge, не payment:*). Это ограничивает ущерб при компрометации.
• Клиенты живут как синглтоны в app.state через lifespan — чтобы переиспользовать соединения и кеш токена.
• Принимающий сервис маппит scope из JWT в роль system и явно объявляет её в require_roles.
• Вызов от имени пользователя делают обменом токена по RFC 8693: fetch_token с grant_type обмена и audience получателя даёт токен с тем же sub и узким scope; user_id в заголовке машинного вызова годится только в доверенном контуре.

Что почитать дальше

• JWT validation в FastAPI — как принимающий сервис валидирует токен.
• RBAC: маппинг ролей — как scope превращается в роль system.
• PII и секреты — хранение client_secret через env и Vault.
• Идемпотентность — Idempotency-Key для денежных межсервисных команд.