Ticket ID: CP-SDK-004
CP-IAM-003: Service-to-Service Auth
Background
Services in the ITL platform must authenticate with each other via OAuth2 client credentials. The SDK provides a ServiceAuthClient that automatically manages tokens and enriches HTTP requests with the correct Bearer token.
File
src/itl_sdk/iam/service_auth.py
Implementation
"""
ITL SDK: Service-to-Service authentication.
Provides an httpx client that automatically adds Keycloak tokens
to outgoing requests. Built on top of TokenManager.
"""
from __future__ import annotations
import httpx
from dataclasses import dataclass
from .token_manager import TokenManager
@dataclass
class ServiceConfig:
"""Configuration for a service-to-service connection."""
keycloak_base_url: str
realm: str
client_id: str
client_secret: str
target_service_url: str
class ServiceAuthTransport(httpx.AsyncBaseTransport):
"""
httpx transport that automatically adds Bearer tokens.
Hooks into the httpx request lifecycle to inject the Authorization
header for every request.
"""
def __init__(self, token_manager: TokenManager, config: ServiceConfig):
self._token_manager = token_manager
self._config = config
self._inner = httpx.AsyncHTTPTransport()
async def handle_async_request(self, request: httpx.Request) -> httpx.Response:
token = await self._token_manager.get_token(
realm= self._config.realm,
client_id= self._config.client_id,
client_secret= self._config.client_secret,
)
request.headers["Authorization"] = f"Bearer {token}"
return await self._inner.handle_async_request(request)
class ServiceAuthClient:
"""
HTTP client with automatic S2S authentication.
Use as a drop-in replacement for httpx.AsyncClient for inter-service calls.
Usage:
config = ServiceConfig(
keycloak_base_url="https://auth.itlusions.com",
realm="itl-platform",
client_id="braincell-api",
client_secret=os.environ["BRAINCELL_CLIENT_SECRET"],
target_service_url="https://ingest.itlusions.com",
)
async with ServiceAuthClient(config) as client:
resp = await client.get("/api/v1/sources")
resp.raise_for_status()
"""
def __init__(self, config: ServiceConfig):
self._config = config
self._token_manager = TokenManager(config.keycloak_base_url)
self._client = httpx.AsyncClient(
base_url=config.target_service_url,
transport=ServiceAuthTransport(self._token_manager, config),
)
async def get(self, url: str, **kwargs) -> httpx.Response:
return await self._client.get(url, **kwargs)
async def post(self, url: str, **kwargs) -> httpx.Response:
return await self._client.post(url, **kwargs)
async def put(self, url: str, **kwargs) -> httpx.Response:
return await self._client.put(url, **kwargs)
async def delete(self, url: str, **kwargs) -> httpx.Response:
return await self._client.delete(url, **kwargs)
async def __aenter__(self) -> ServiceAuthClient:
return self
async def __aexit__(self, *args) -> None:
await self.close()
async def close(self) -> None:
await self._token_manager.close()
await self._client.aclose()Usage in BrainCell.Api → Ingest service
# In BrainCell.Api — call to Ingest service
ingest_config = ServiceConfig(
keycloak_base_url=settings.KEYCLOAK_URL,
realm=settings.KEYCLOAK_REALM,
client_id="braincell-api",
client_secret=settings.BRAINCELL_CLIENT_SECRET,
target_service_url=settings.INGEST_SERVICE_URL,
)
async def get_ingest_client() -> ServiceAuthClient:
return ServiceAuthClient(ingest_config)Acceptance Criteria
Related
Ticket ID:
CP-SDK-004CP-IAM-003: Service-to-Service Auth
Background
Services in the ITL platform must authenticate with each other via OAuth2 client credentials. The SDK provides a
ServiceAuthClientthat automatically manages tokens and enriches HTTP requests with the correct Bearer token.File
src/itl_sdk/iam/service_auth.pyImplementation
Usage in BrainCell.Api → Ingest service
Acceptance Criteria
ServiceAuthTransportadds Bearer token to every requestTokenManagerServiceAuthClientworks as async context managerget(),post(),put(),delete()implementedServiceConfigcontains all required connection settingsRelated