sasy-common 0.1.0

Common utilities for observability platform: auth, TLS, credentials, and proto stubs

Observability Common

Shared utilities for the observability platform, providing authentication, TLS configuration, and protocol buffer stubs used by all services.

Design

This package contains shared code including that for authentication, TLS handling, and gRPC communication. It includes:

• Server-side auth providers: Validate incoming requests (JWT, mTLS, API keys)
• Client-side auth hooks: Add credentials to outgoing requests
• TLS configuration: Consistent TLS/mTLS setup across services
• gRPC interceptors: Authentication enforcement for gRPC services
• Protocol buffer stubs: Generated Python code for all service protos

Components

Server-Side Authentication (auth.py)

Auth providers validate incoming requests and extract entity/role information:

from observability_common import JWTAuthProvider, MTLSAuthProvider, APIKeyAuthProvider

# JWT validation with JWKS endpoint
jwt_provider = JWTAuthProvider(
    jwks_url="https://keycloak:8443/realms/myapp/protocol/openid-connect/certs",
    issuer="https://keycloak:8443/realms/myapp",
    entity_claim="preferred_username",
    roles_claim="roles",
)

# mTLS - extract entity from certificate
mtls_provider = MTLSAuthProvider(entity_source="cn")

# API key authentication
apikey_provider = APIKeyAuthProvider(
    metadata_key="x-api-key",
    static_keys={"dev-key": "developer", "prod-key": "service"},
)

# Chain multiple providers (try in order)
from observability_common import ChainAuthProvider
chain = ChainAuthProvider([jwt_provider, apikey_provider, mtls_provider])

# Validate a request
result = provider.authenticate(gRPC_context)
if result.authenticated:
    print(f"Entity: {result.entity}, Roles: {result.roles}")

Client-Side Authentication (auth_hooks.py)

Auth hooks add credentials to outgoing gRPC requests:

from observability_common import (
    NoAuthHook,
    APIKeyAuthHook,
    OIDCAuthHook,
    KeycloakAuthHook,
    BrowserAuthHook,
    DeviceAuthHook,
)

# No authentication
hook = NoAuthHook()

# Static API key
hook = APIKeyAuthHook(api_key="my-secret-key")

# OAuth2 client credentials
hook = OIDCAuthHook(
    token_url="https://keycloak:8443/realms/myapp/protocol/openid-connect/token",
    client_id="my-service",
    client_secret="secret",
    scope="openid profile",
)

# Browser-based OAuth2 with PKCE
hook = BrowserAuthHook(
    authorization_url="https://keycloak:8443/realms/myapp/protocol/openid-connect/auth",
    token_url="https://keycloak:8443/realms/myapp/protocol/openid-connect/token",
    client_id="my-cli",
    redirect_port=8400,
)

# Device authorization grant
hook = DeviceAuthHook(
    device_authorization_url="https://keycloak:8443/realms/myapp/protocol/openid-connect/auth/device",
    token_url="https://keycloak:8443/realms/myapp/protocol/openid-connect/token",
    client_id="my-device",
)

# Get metadata for gRPC calls
metadata = hook.get_metadata()  # Returns [("authorization", "Bearer ...")]

TLS Configuration (tls.py)

Consistent TLS setup for clients and servers:

from observability_common import TLSConfig, ServerTLSConfig, get_shared_tls_config

# Client TLS config
client_tls = TLSConfig(
    ca_path="certs/ca.crt",
    cert_path="certs/client.crt",  # For mTLS
    key_path="certs/client.key",
)
credentials = client_tls.to_grpc_credentials()

# Server TLS config
server_tls = ServerTLSConfig(
    cert_path="certs/server.crt",
    key_path="certs/server.key",
    ca_path="certs/ca.crt",  # For mTLS client verification
    require_client_auth=True,
)
credentials = server_tls.to_grpc_credentials()

# Shared config from environment
tls = get_shared_tls_config()  # Uses TLS_CA_PATH, TLS_CERT_PATH, TLS_KEY_PATH

gRPC Interceptors (grpc_interceptors.py)

Enforce authentication and authorization for gRPC services:

from observability_common import AuthInterceptor, require_role, get_auth_context

# Create interceptor with auth provider
interceptor = AuthInterceptor(auth_provider)

# Use in gRPC server
server = grpc.server(
    futures.ThreadPoolExecutor(),
    interceptors=[interceptor],
)

# In service methods, check roles
@require_role("admin")
def AdminOnlyMethod(self, request, context):
    auth = get_auth_context(context)
    print(f"Called by: {auth.entity}")

Authorization Config (auth_config.yaml)

Central entity-to-role mapping loaded from YAML:

from observability_common import load_auth_config, get_auth_config

# Load from file
config = load_auth_config("config/auth_config.yaml")

# Get roles for an entity
roles = config.get_roles_for_entity("my-service")  # ["admin", "reader"]

Protocol Buffer Stubs

Generated stubs for all platform services are in observability_common/proto/:

• observability_pb2.py - Observability server messages
• credential_server_pb2.py - Credential server messages
• policy_engine_pb2.py - Policy engine messages
• reference_monitor_pb2.py - Reference monitor messages

Regenerate with:

make proto  # From repo root