Service-to-service — mTLS и Client Credentials Flow в FastAPI

Service-to-service auth в FastAPI: mTLS (Service Mesh) или Client Credentials Flow с authlib. Анонимный inter-service трафик запрещён. AUTH-13..AUTH-14.

Опубликовано 20 июня 2026 ~5 мин чтения Автор: Вадим Викулин

Опирается на правила: AUTH-13 … AUTH-14 из Auth Patterns — раздел 5. Service-to-service.

Важно знать

Два способа межсервисной аутентификации: mTLS (рекомендуется) или Client Credentials Flow.

mTLS — двусторонний TLS на K8s Service Mesh (Istio, Linkerd). Identity = CN client-сертификата; application-код аутентификации не пишет.

Client Credentials Flow — grant_type=client_credentials, сервис получает access_token от IdP с scope=service:operation; authlib кеширует токен и обновляет при истечении.

Анонимный inter-service трафик — критическое нарушение (AUTH-14). httpx.AsyncClient без mTLS или Bearer — на ревью.

scope per operation (payment:charge, inventory:reserve) — не один широкий scope=service. Это blast-radius containment.

Hard-coded client_secret в коде или конфиге — никогда. Только env / Vault через pydantic-settings.

Один s2s-клиент — один домен: PaymentClient, InventoryClient — отдельные адаптеры в adapters/out/http/, не общий «HTTP-клиент на всё».

Service-to-service вызовы внутри кластера кажутся безопасными: VPC isolated, internal network. Это иллюзия. Один скомпрометированный pod через уязвимость в образе — и атакующий ходит на все соседние сервисы без ограничений. AUTH-13..14 формулируют zero trust: каждый межсервисный вызов аутентифицирован.

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

AUTH-13: Istio или Linkerd автоматически выдают каждому pod-у уникальный client-сертификат (SPIFFE identity), шифруют весь internal-трафик TLS 1.2+ и верифицируют сертификат принимающей стороны.

order-service pod → istio-proxy sidecar → mTLS encrypted → istio-proxy → payment-service pod
                    (cert: order-service-prod)              (verifies cert)

Fastapi-приложение ничего не пишет для аутентификации — sidecar обрабатывает mTLS до приложения. Принимающий сервис получает identity через заголовок X-Forwarded-Client-Cert (Istio) или через policy PeerAuthentication.

# 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()

При mTLS на Service Mesh никаких Authorization-заголовков добавлять не нужно — sidecar добавляет mTLS-идентификацию автоматически. Аутентификация встроена в transport.

Преимущества mTLS:

Identity встроена в transport — нечего «забыть» добавить в headers.
Rotation автоматическая — Istio обновляет сертификаты каждые 24 часа.
Cross-language — работает одинаково для Java, Go, Python сервисов.

Недостаток: требует Service Mesh-инфры (Istio/Linkerd). В dev/test без Service Mesh нужен Client Credentials Flow.

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

AUTH-13: OAuth2 standard. Сервис получает собственный access_token от IdP — не перекладывает токен пользователя.

order-service → IdP: POST /realms/main/protocol/openid-connect/token
                grant_type=client_credentials
                client_id=order-service-prod
                client_secret=$ORDER_SERVICE_CLIENT_SECRET
                scope=payment:charge

← IdP: { "access_token": "...", "expires_in": 3600 }

order-service → payment-service:
                POST /charge
                Authorization: Bearer <token>

payment-service: Depends(require_roles("system")) валидирует токен,
                 видит role=system, разрешает.

Конфигурация через pydantic-settings

AUTH-17 — секреты не в git, только через env:

# 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 httpx-клиент с authlib

authlib предоставляет AsyncOAuth2Client — он кеширует access_token до exp, обновляет при истечении и добавляет Authorization: Bearer к каждому запросу:

# 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 charge(self, order_id: str, amount_cents: int) -> dict:
        await self._client.ensure_active_token()
        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 refund(self, order_id: str, amount_cents: int) -> dict:
        await self._client.ensure_active_token()
        resp = await self._client.post(
            f"{settings.payment_service_url}/refund",
            json={"order_id": order_id, "amount_cents": amount_cents},
        )
        resp.raise_for_status()
        return resp.json()

ensure_active_token() проверяет, что токен не просрочен, и при необходимости запрашивает новый у IdP. Ручное кеширование и refresh не нужны.

Раздельные scope per operation

order-service может делать charge, но не refund — разные scope, разные client-регистрации, если нужно жёсткое разделение:

# 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
        )

# 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
        )

Один широкий scope на все операции — антипаттерн: компрометация одного клиента даёт доступ ко всем операциям (нет blast-radius containment).

Получение s2s-токена в принимающем сервисе

На стороне payment-service s2s-вызов отличается от user-запроса ролью: system. AUTH-9 обязывает каждый 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)

Роль system маппится из scope-клейма JWT (AUTH-7). Пример маппинга в _extract_roles:

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", [])

Запрет анонимного трафика

AUTH-14: httpx.AsyncClient без mTLS или Authorization — критическое нарушение.

# ЗАПРЕЩЕНО — AUTH-14
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()

Любой pod в кластере вызывает product-service без ограничений. Один скомпрометированный контейнер — lateral movement по всему кластеру.

Корректно — либо mTLS (sidecar добавляет auth автоматически), либо AsyncOAuth2Client с Client Credentials:

# 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:
        await self._client.ensure_active_token()
        resp = await self._client.get(
            f"{settings.product_service_url}/products/{product_id}"
        )
        resp.raise_for_status()
        return resp.json()

Dependency Injection клиентов

Клиенты регистрируются как синглтоны через FastAPI 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._client.aclose()
    await app.state.inventory_client._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)

Что запрещено

Антипаттерн	Правило	Что взамен
`httpx.AsyncClient` без mTLS или `Bearer`	`AUTH-14`	mTLS или `AsyncOAuth2Client`
Анонимный inter-service трафик	`AUTH-14`	zero trust: каждый вызов аутентифицирован
Hard-coded `client_secret` в коде	`AUTH-17`	env через `pydantic-settings`
Один токен с широким scope на все операции	`AUTH-13`	scope per operation (`payment:charge`, `inventory:reserve`)
`client_secret` в `settings.py` напрямую	`AUTH-17`	`${PAYMENT_CLIENT_SECRET}` через env / Vault
Ручное кеширование и refresh токена	`AUTH-13`	`AsyncOAuth2Client.ensure_active_token()` делает сам
Перекладывание user-токена в s2s-вызов	`AUTH-13`	собственный токен сервиса через Client Credentials
mTLS только для public-facing эндпоинтов	`AUTH-13`	zero trust: internal-трафик тоже

Куда дальше

JWT validation — payment-service валидирует токен от order-service.
PII и секреты — client_secret через env и Vault.
Где какая проверка — роль system для s2s на принимающей стороне.
RBAC: маппинг ролей — маппинг scope в роль system.
ABAC: владение ресурсом — s2s-вызов с system обходит ABAC.
Аудит admin-команд — декоратор аудита при system-вызовах.
Хранение токенов на клиенте — HttpOnly cookie, refresh-token rotation.
Идемпотентность — Idempotency-Key для денежных s2s-команд.