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

Когда 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-код не пишет ни строчки для аутентификации. Всё происходит на уровне транспорта, до приложения.

order-service → [istio-proxy sidecar] → зашифрованный mTLS → [istio-proxy] → payment-service
                 cert: order-service-prod                        проверяет cert

# 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 — не токен пользователя, а именно машинный — и прикладывает его к каждому запросу.

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

order-service → 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 → POST payment-service/charge
                Authorization: Bearer <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 — он сам кеширует токен, следит за сроком его жизни и при необходимости запрашивает новый. Вручную ничего реализовывать не нужно:

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

ensure_active_token() проверяет, что токен не просрочен. Если просрочен — запрашивает новый у IdP. Ручное кеширование и refresh не нужны.

Отдельный 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:
        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()

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

Клиенты регистрируются в 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)

Коротко

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

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