Ports and Services

Context

With the domain layer in place, this document defines the ports (Protocol-based interfaces) and services (business logic) following hexagonal architecture.

• Ports define boundaries using Protocols (static duck typing)
• Services implement business logic, depending only on domain types and ports
• Dependencies flow inward: adapters → services → ports → domain

File structure

src/hdc_auth_proxy/
├── ports/
│   ├── __init__.py           # Re-export all Protocols
│   ├── authentication.py     # MethodAuthenticator, TokenValidator, PrincipalResolver, ApiKeyValidator, OidcProviderSettings, ClaimsMapper
│   ├── authorization.py      # PolicyRepository, ResourceRepository
│   ├── oidc.py               # PKCEStore, OIDCTokenExchanger (async ports for OIDC PKCE flow)
│   └── input.py              # Authenticator, Authorizer (input ports / use cases)
│
├── services/
│   ├── __init__.py           # Re-export service classes
│   ├── authentication.py     # TokenMethodAuthenticator, ApiKeyMethodAuthenticator, AuthenticationService
│   ├── authorization.py      # AuthorizationService
│   └── oidc.py               # OIDCFlowService (PKCE initiate + complete)
│
├── wiring.py                 # Composition root: wire_services() + wire_oidc_flow() → OIDCWiringResult
├── dev.py                    # Dev mode: build_dev_services() with pre-populated test data

Output Ports (driven side — what adapters implement)

ports/authentication.py

ports/authorization.py

ports/oidc.py

These async ports are used by OIDCFlowService for the PKCE browser login flow. See ADR-0014 and ADR-0015.

Input Ports (driving side — what services expose)

ports/input.py

Services

Authentication — Strategy Pattern

Instead of a monolithic service that branches on AuthMethod, each auth method has its own MethodAuthenticator implementation with only the dependencies it needs.

                        ┌─────────────────────────────┐
                        │   AuthenticationService      │
                        │   strategies: Mapping[        │
                        │     AuthMethod,               │
                        │     MethodAuthenticator       │
                        │   ]                           │
                        └──────────┬──────────────────┘
                                   │ dispatches via Mapping
                    ┌──────────────┴──────────────┐
                    │                             │
        ┌───────────▼──────────┐     ┌────────────▼─────────┐
        │ TokenMethod          │     │ ApiKeyMethod          │
        │ Authenticator        │     │ Authenticator         │
        │                      │     │                       │
        │ deps:                │     │ deps:                 │
        │  - TokenValidator    │     │  - ApiKeyValidator    │
        │  - PrincipalResolver │     │                       │
        └──────────────────────┘     └───────────────────────┘

TokenMethodAuthenticator (OAUTH2_OIDC, JWT_BEARER):

1. Validate token → TokenValidationResult
2. Check expiration (domain logic)
3. Resolve principal → Principal | None
4. Return AuthenticationResult

ApiKeyMethodAuthenticator (API_KEY):

1. Validate key → KeyValidationResult
2. Return AuthenticationResult (no claims step)

AuthenticationService (dispatcher):

• Receives Mapping[AuthMethod, MethodAuthenticator]
• Dispatches credential.method to the right strategy
• Returns "unsupported method" failure for unknown methods

Authorization — Default Deny

AuthorizationService:

1. Look up resource → if public=True, ALLOW immediately
2. Load PolicyRules matching the resource URI
3. Evaluate rules: glob pattern + permission + role + scopes
4. First matching rule → ALLOW; no match → DENY

Rule matching uses fnmatch for glob-style URI patterns.

OIDC PKCE Flow — OIDCFlowService

OIDCFlowService (services/oidc.py) — async service orchestrating the OIDC authorization code + PKCE S256 flow:

• initiate(callback_uri) → generates PKCE challenge, state, nonce; stores verifier in PKCEStore; builds authorization URL
• complete(callback, callback_uri) → retrieves verifier (atomic delete); exchanges code via OIDCTokenExchanger; returns TokenExchangeResult
• refresh(refresh_token) → exchanges refresh token for new access/id tokens via OIDCTokenExchanger; returns TokenRefreshResult

Dependencies: PKCEStore, OIDCTokenExchanger (async ports), provider settings, client_id, scopes. No HTTP/framework dependency — testable with stubs.

See ADR-0015 for PKCE S256 and session cookie design.

Design Decisions

HTTP Adapter (Driving Side)

The HTTP adapter is the driving adapter that receives HTTP requests and orchestrates the full proxy flow. Implemented with Starlette (ADR-0011).

File structure

src/hdc_auth_proxy/adapters/http/
├── __init__.py               # Re-export create_app
├── app.py                    # create_app(ProxySettings) → Starlette
├── credential_extractor.py   # Authorization/X-API-Key → Credential | None
├── permission_mapper.py      # HTTP method → Permission
├── resource_mapper.py        # Request path → ResourceIdentifier
├── proxy_handler.py          # Orchestrates extract → authn → authz → forward
├── upstream_client.py        # httpx.AsyncClient wrapper with streaming
├── error_responses.py        # 401, 403, 502 JSONResponse builders
└── headers.py                # Hop-by-hop filter, X-Forwarded-* builder

ProxyHandler flow

Client → ProxyHandler.handle(request)
           ├── extract_credential(request) → Credential | None
           │     └── None → 401 "No credential provided"
           ├── authenticator.authenticate(credential) → AuthenticationResult
           │     └── not authenticated → 401
           ├── map_request_to_resource(path) → ResourceIdentifier
           ├── map_http_method_to_permission(method) → Permission
           ├── authorizer.authorize(AccessRequest) → AccessResult
           │     └── DENY → 403
           └── upstream_client.forward(request) → StreamingResponse
                 └── HTTPError → 502

ProxyHandler depends only on Authenticator and Authorizer Protocols — it has no knowledge of which adapters are behind them.

Composition root (wiring.py)

Adapter selection happens in wire_services(settings), the composition root that evaluates a fixed priority (ADR-0012):

wire_services() returns (AuthenticationService, AuthorizationService) which are passed to create_app(). This keeps the HTTP adapter decoupled from adapter selection logic.

OIDC PKCE browser login (wire_oidc_flow)

A separate wire_oidc_flow(settings) function wires the OIDC PKCE browser login flow when COGNITO_DOMAIN is set:

The AuthRouteHandler provides three routes added before the catch-all proxy:

Configuration is read from PKCEConfig (PKCE_* env vars) and SessionConfig (SESSION_* env vars).

Key design choices

Testing approach

Three-tier strategy covering unit, integration OIDC, and integration Cognito-specific tests. No Docker required — all tiers run in-process. See ADR-0016 for the evaluation of Cognito testing tools.

Tier 1 — Unit tests (pytest-httpx + stubs)

Fast, isolated tests. Stubs implement output port Protocols. pytest-httpx mocks HTTP responses for adapters that do I/O.

Tier 2 — Integration OIDC (pytest-iam)

Tests the full OIDC PKCE S256 flow against a real OIDC provider (Canaille + Authlib) running in-thread. Validates standard compliance independently from Cognito.

Tier 3 — Integration Cognito (moto)

Tests Cognito-specific JWT behavior using moto (in-process AWS mock). Verifies that our adapters work with Cognito's specific token format, issuer convention, and claim names.

Tier 4 — E2E Cognito OIDC (magnito + testcontainers)

Tests the full Cognito OIDC PKCE flow against magnito (Docker-based Cognito emulator). Requires Docker. Marked @pytest.mark.e2e, runs as a separate CI job.