← Back to the section

When order-service calls payment-service inside the cluster, it seems safe: a closed network, a VPC, no outsider gets in. But if one container is compromised through a vulnerability in a dependency — the attacker can reach all the neighboring services without restriction, because they do not "ask" it anything.

The solution is that every call between services must carry proof of the caller's identity. There are two ways to arrange this in Python: mTLS via a Service Mesh, or Client Credentials Flow with a token from the IdP.

Option 1: mTLS via a Service Mesh

mTLS stands for mutual TLS — two-way TLS authentication. Ordinary HTTPS verifies only the server (its certificate). mTLS verifies both sides: the server and the client both present certificates.

In Kubernetes this is implemented by a Service Mesh — Istio or Linkerd. Each pod is automatically issued a unique certificate (the SPIFFE standard). A sidecar container next to your application intercepts all inbound and outbound traffic, encrypts it, and verifies the certificates.

The main advantage: your Python code writes not a single line for authentication. Everything happens at the transport level, before the application.

order-service → [istio-proxy sidecar] → encrypted mTLS → [istio-proxy] → payment-service
                 cert: order-service-prod                        verifies 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()

You do not need to add any Authorization headers — the sidecar does it for you. The receiving service gets the caller's identity via the X-Forwarded-Client-Cert header (Istio) or through the PeerAuthentication policy.

Istio automatically rotates certificates every 24 hours — rotation is free. And the approach works the same for Java, Go, and Python services in one cluster.

The only downside: it requires Service Mesh infrastructure. In local development or in an environment without Istio/Linkerd — you need the second approach.

Option 2: Client Credentials Flow with authlib

This is the standard OAuth2 flow for machine clients. The service requests its own access_token from the IdP (for example, Keycloak) — not a user's token, but specifically a machine one — and attaches it to every request.

The scheme looks like this:

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>

Secrets via pydantic-settings

The client_secret must not be written directly in the code. We read it from environment variables via 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()

An OAuth2 client with authlib

authlib provides AsyncOAuth2Client — it caches the token itself, tracks its lifetime, and requests a new one when needed. There is nothing to implement manually:

# 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() checks that the token is not expired. If it is expired — it requests a new one from the IdP. Manual caching and refresh are not needed.

A separate scope for each operation

A common mistake is one token with a broad scope for all operations. If that token leaks, the attacker gains access to everything. It is better to use a 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 only, not 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 only
        )

This is called blast-radius containment: if something goes wrong with one client, the damage is limited to just its operations.

Verification on the receiving side

payment-service receives a request with a token and must understand that this is a machine call (not a user). The system role is mapped from the JWT's scope:

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

The endpoint declares the required role explicitly:

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

Anonymous traffic — a common mistake

Here is what the problematic code looks like:

# Incorrect — any pod in the cluster can call without restrictions
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()

One compromised container — and the attacker moves across the whole cluster. The correct version is either mTLS (the sidecar adds authentication) or 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()

Registering clients as singletons

Clients are registered in lifespan rather than created on every request. This lets you reuse HTTP connections and not lose the token cache:

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

In short

  • Authentication is needed inside the cluster too — an isolated network does not protect against a compromised container.
  • Two approaches: mTLS via a Service Mesh (the code does nothing — the sidecar takes it on) or Client Credentials Flow (the service obtains a machine token from the IdP).
  • In Client Credentials Flow use authlib.AsyncOAuth2Client — it caches the token and refreshes it itself. Manual refresh is not needed.
  • The secret (client_secret) — only via environment variables, pydantic-settings. Not in code, not in git.
  • One client — one scope per operation (payment:charge, not payment:*). This limits the damage from a compromise.
  • Clients live as singletons in app.state via lifespan — to reuse connections and the token cache.
  • The receiving service maps the JWT's scope to the system role and declares it explicitly in require_roles.

Further reading