Service layer

The HTTP surface: the request and response envelope, bearer token verification, the single operation endpoint, and the ASGI application itself.

Request and response models

models

Pydantic data models for the Cryptography Manager API.

These models are the single source of truth for the request envelope, the response envelope and the audit record. They deliberately avoid importing any cryptographic backend so that the schema stays importable even when optional extras such as PyDP, Pyfhel or MPyC are not installed.

Operation

Bases: str, Enum

Cryptographic operation requested by the caller.

Backend

Bases: str, Enum

Cryptographic library used to service an operation.

OutputFormat

Bases: str, Enum

Desired response encoding.

JSON-LD is a plausible future addition, but its vocabulary would have to align with the Compliance Manager's DPV graphs, which are not pinned down yet. Only JSON is accepted for now; requesting anything else is a validation error rather than a silently ignored field.

DPQueryType

Bases: str, Enum

Aggregate queries supported by the differential privacy adapter.

Mirrors the PyDP Laplacian algorithms wired up in the DP adapter. Declared here rather than imported from the adapter so that importing the schema never requires PyDP to be installed.

CryptographyRequest

Bases: BaseModel

Request envelope for POST /cryptography.

The Authorization header is not modelled here; it is handled by the auth dependency before the body is ever parsed.

check_backend_compatible

check_backend_compatible() -> str | None

Verify the requested backend can service the operation.

Returns:

Type	Description
str | None	An error message if the pairing is invalid, otherwise None.

Source code in src/cryptography_manager/api/models.py

def check_backend_compatible(self) -> str | None:
    """Verify the requested backend can service the operation.

    Returns:
        An error message if the pairing is invalid, otherwise ``None``.
    """
    allowed = COMPATIBLE_BACKENDS[self.operation]
    if self.backend in allowed:
        return None
    names = ", ".join(sorted(b.value for b in allowed))
    return (
        f"Backend '{self.backend.value}' cannot service operation "
        f"'{self.operation.value}'; supported: {names}"
    )

DPParameters

Bases: BaseModel

Differential privacy parameters carried in parameters.

Validated separately from the envelope because the required fields depend on the operation. epsilon_cost is the privacy price of the query and is checked against the caller's remaining budget before execution.

CryptographyResponse

Bases: BaseModel

Response envelope for POST /cryptography.

AuditRecord

Bases: BaseModel

Structured audit entry produced for every request.

Sensitive material -- secret and private keys, plaintext values, secret shares and intermediate cryptographic state -- must never reach this model. Sanitisation happens in core.audit before construction.

Authentication

auth

Bearer token verification.

Verification sits behind a small interface with two implementations, because the identity provider this component will ultimately talk to is not settled. The OIDC verifier targets the standard discovery and JWKS endpoints rather than any one vendor's API, so it should work against any conforming provider and needs replacing only if the eventual choice is non-standard.

The development verifier exists so the service is runnable with no identity provider at all. It still rejects a missing or malformed token, so the rejection path is exercised the same way in both modes -- it simply does not verify signatures, and must never be used outside development.

AuthResult

AuthResult(
    user_id: str | None = None,
    claims: dict[str, Any] | None = None,
    error: str | None = None,
    permissions: frozenset[str] | None = None,
)

Outcome of verifying a token.

Source code in src/cryptography_manager/api/auth.py

def __init__(
    self,
    user_id: str | None = None,
    claims: dict[str, Any] | None = None,
    error: str | None = None,
    permissions: frozenset[str] | None = None,
) -> None:
    self.user_id = user_id
    self.claims = claims or {}
    self.error = error
    #: Permissions the token grants, or ``None`` when they are not being
    #: enforced and every action is allowed.
    self.permissions = permissions

client_id property

client_id: str | None

The client the token was issued to, if it says.

issuer property

issuer: str | None

The issuer of the token, if it says.

ok property

ok: bool

Whether verification succeeded.

TokenVerifier

Bases: Protocol

Verifies a bearer token and identifies the caller.

verify

verify(token: str | None) -> AuthResult

Verify a token.

Parameters:

Name	Type	Description	Default
token	str | None	The bearer token, without its scheme prefix.	required

Returns:

Type	Description
AuthResult	The verification outcome. Failures are returned rather than
AuthResult	raised, because a rejected request still has to be audited.

Source code in src/cryptography_manager/api/auth.py

def verify(self, token: str | None) -> AuthResult:
    """Verify a token.

    Args:
        token: The bearer token, without its scheme prefix.

    Returns:
        The verification outcome. Failures are returned rather than
        raised, because a rejected request still has to be audited.
    """
    ...

DevTokenVerifier

Accepts any well-formed token without verifying its signature.

Intended solely for development against no identity provider. It reads the claims of a JWT to identify the caller, and otherwise treats the token string itself as the identity.

verify

verify(token: str | None) -> AuthResult

Identify the caller without cryptographic verification.

Source code in src/cryptography_manager/api/auth.py

def verify(self, token: str | None) -> AuthResult:
    """Identify the caller without cryptographic verification."""
    if not token or not token.strip():
        return AuthResult(error="Missing authorization token")

    token = token.strip()
    try:
        claims = jwt.decode(
            token, options={"verify_signature": False, "verify_exp": False}
        )
    except Exception:
        # Not a JWT. Derive a stable identity from a digest of the token
        # rather than the token itself: the identifier reaches audit
        # records and budget state, and neither may ever hold a
        # credential that could be replayed.
        digest = hashlib.sha256(token.encode("utf-8")).hexdigest()
        return AuthResult(user_id=f"dev:{digest[:16]}")

    user_id = _extract_user_id(claims)
    if not user_id:
        return AuthResult(error="Token contains no usable subject claim")
    return AuthResult(user_id=user_id, claims=claims)

OIDCTokenVerifier

OIDCTokenVerifier(settings: Settings)

Verifies tokens against an OpenID Connect provider.

Keys are discovered through the provider's well-known configuration and cached by the JWKS client, so steady-state verification is local and does not add a network round trip per request.

Initialise the verifier.

Parameters:

Name	Type	Description	Default
settings	Settings	Runtime settings carrying the issuer and client details.	required

Raises:

Type	Description
ValueError	If no issuer is configured.

Source code in src/cryptography_manager/api/auth.py

def __init__(self, settings: Settings) -> None:
    """Initialise the verifier.

    Args:
        settings: Runtime settings carrying the issuer and client details.

    Raises:
        ValueError: If no issuer is configured.
    """
    issuer = settings.issuer_url
    if not issuer:
        raise ValueError("OIDC verification requires an issuer URL")
    self.issuer = issuer
    self.settings = settings
    self._jwks_client: PyJWKClient | None = None
    self._discovered: dict[str, Any] | None = None

verify

verify(token: str | None) -> AuthResult

Verify a token's signature, expiry, issuer and audience.

Source code in src/cryptography_manager/api/auth.py

def verify(self, token: str | None) -> AuthResult:
    """Verify a token's signature, expiry, issuer and audience."""
    if not token or not token.strip():
        return AuthResult(error="Missing authorization token")

    token = token.strip()
    try:
        signing_key = self._client().get_signing_key_from_jwt(token)
        audience = self.settings.audience_to_verify
        claims = jwt.decode(
            token,
            signing_key.key,
            algorithms=ALLOWED_ALGORITHMS,
            issuer=self.issuer,
            audience=audience,
            options={
                "verify_signature": True,
                "verify_exp": True,
                "verify_iss": True,
                "verify_aud": bool(audience),
                "require": ["exp"],
            },
        )
    except jwt.ExpiredSignatureError:
        return AuthResult(error="Token has expired")
    except jwt.InvalidTokenError as exc:
        return AuthResult(error=f"Invalid token: {exc}")
    except Exception as exc:
        # A provider that is unreachable must not be reported as a valid
        # token; failing closed is the only safe direction here.
        logger.warning(f"Token verification could not complete: {exc}")
        return AuthResult(error="Token verification unavailable")

    user_id = _extract_user_id(claims)
    if not user_id:
        return AuthResult(error="Token contains no usable subject claim")

    permissions = None
    if self.settings.oidc_enforce_permissions:
        permissions = extract_permissions(
            claims,
            self.settings.expected_audience,
            self.settings.oidc_permission_prefix,
        )
    return AuthResult(
        user_id=user_id, claims=claims, permissions=permissions
    )

ServiceTokenProvider

ServiceTokenProvider(settings: Settings)

Obtains this service's own access token for calls it makes.

Uses the client credentials grant and holds the token until shortly before it expires, so outbound calls do not each cost a round trip to the provider.

Initialise the provider.

Parameters:

Name	Type	Description	Default
settings	Settings	Runtime settings carrying issuer and client details.	required

Source code in src/cryptography_manager/api/auth.py

def __init__(self, settings: Settings) -> None:
    """Initialise the provider.

    Args:
        settings: Runtime settings carrying issuer and client details.
    """
    self.settings = settings
    self._token: str | None = None
    self._expires_at = 0.0
    self._token_endpoint: str | None = None
    self._lock = threading.Lock()

token

token() -> str | None

Return a valid access token, or None if none can be had.

Never raises: the callers are best-effort deliveries that must go ahead, unauthenticated if need be, rather than fail.

Source code in src/cryptography_manager/api/auth.py

def token(self) -> str | None:
    """Return a valid access token, or ``None`` if none can be had.

    Never raises: the callers are best-effort deliveries that must go
    ahead, unauthenticated if need be, rather than fail.
    """
    with self._lock:
        if self._token and time.monotonic() < self._expires_at:
            return self._token
        try:
            response = httpx.post(
                self._endpoint(),
                data={
                    "grant_type": "client_credentials",
                    "client_id": self.settings.oidc_client_id,
                    "client_secret": self.settings.oidc_client_secret,
                },
                timeout=10.0,
            )
            response.raise_for_status()
            body = response.json()
            self._token = str(body["access_token"])
            lifetime = float(body.get("expires_in", 60))
            self._expires_at = (
                time.monotonic() + max(lifetime - self.LEEWAY, 0.0)
            )
        except Exception as exc:
            logger.warning(f"Could not obtain a service token: {exc}")
            self._token = None
        return self._token

extract_permissions

extract_permissions(
    claims: dict[str, Any],
    audience: str | None,
    prefix: str,
) -> frozenset[str]

Collect the permissions a token grants.

Providers disagree on where authorization lives, so every usual place is read: the space-separated scope claim, realm-level roles, and the roles granted on this service specifically. Roles of that last kind are already scoped to this service by where they sit, so a bare encrypt there is taken to mean <prefix>.encrypt.

Parameters:

Name	Type	Description	Default
claims	dict[str, Any]	The verified token claims.	required
audience	str | None	This service's client ID, keying its client-level roles.	required
prefix	str	Namespace of this service's permissions.	required

Returns:

Type	Description
frozenset[str]	Every permission name found.

Source code in src/cryptography_manager/api/auth.py

def extract_permissions(
    claims: dict[str, Any], audience: str | None, prefix: str
) -> frozenset[str]:
    """Collect the permissions a token grants.

    Providers disagree on where authorization lives, so every usual place is
    read: the space-separated ``scope`` claim, realm-level roles, and the
    roles granted on this service specifically. Roles of that last kind are
    already scoped to this service by where they sit, so a bare ``encrypt``
    there is taken to mean ``<prefix>.encrypt``.

    Args:
        claims: The verified token claims.
        audience: This service's client ID, keying its client-level roles.
        prefix: Namespace of this service's permissions.

    Returns:
        Every permission name found.
    """
    granted: set[str] = set()

    scope = claims.get("scope")
    if isinstance(scope, str):
        granted.update(scope.split())

    realm = claims.get("realm_access")
    if isinstance(realm, dict) and isinstance(realm.get("roles"), list):
        granted.update(str(role) for role in realm["roles"])

    resources = claims.get("resource_access")
    if audience and isinstance(resources, dict):
        own = resources.get(audience)
        if isinstance(own, dict) and isinstance(own.get("roles"), list):
            for role in own["roles"]:
                name = str(role)
                granted.add(name)
                if not name.startswith(f"{prefix}."):
                    granted.add(f"{prefix}.{name}")

    return frozenset(granted)

build_service_token_provider

build_service_token_provider(
    settings: Settings,
) -> ServiceTokenProvider | None

Construct the outbound token provider, if one is configured.

Parameters:

Name	Type	Description	Default
settings	Settings	Runtime settings.	required

Returns:

Type	Description
ServiceTokenProvider | None	A provider when the service has an identity provider and client
ServiceTokenProvider | None	credentials of its own, otherwise None.

Source code in src/cryptography_manager/api/auth.py

def build_service_token_provider(
    settings: Settings,
) -> ServiceTokenProvider | None:
    """Construct the outbound token provider, if one is configured.

    Args:
        settings: Runtime settings.

    Returns:
        A provider when the service has an identity provider and client
        credentials of its own, otherwise ``None``.
    """
    if (
        settings.auth_mode == "oidc"
        and settings.oidc_client_id
        and settings.oidc_client_secret
    ):
        return ServiceTokenProvider(settings)
    return None

build_verifier

build_verifier(settings: Settings) -> TokenVerifier

Construct the verifier selected by configuration.

Parameters:

Name	Type	Description	Default
settings	Settings	Runtime settings.	required

Returns:

Type	Description
TokenVerifier	The configured verifier.

Source code in src/cryptography_manager/api/auth.py

def build_verifier(settings: Settings) -> TokenVerifier:
    """Construct the verifier selected by configuration.

    Args:
        settings: Runtime settings.

    Returns:
        The configured verifier.
    """
    if settings.auth_mode == "oidc":
        logger.info(f"Auth mode: OIDC against {settings.issuer_url}")
        if not settings.audience_to_verify:
            logger.warning(
                "Token audience is NOT verified: tokens issued for other "
                "services are accepted. Set CRM_OIDC_AUDIENCE or "
                "CRM_OIDC_CLIENT_ID."
            )
        return OIDCTokenVerifier(settings)
    logger.warning(
        "Auth mode: development. Tokens are NOT verified -- "
        "set CRM_AUTH_MODE=oidc before any real deployment."
    )
    return DevTokenVerifier()

extract_bearer

extract_bearer(header: str | None) -> str | None

Pull the token out of an Authorization header.

Parameters:

Name	Type	Description	Default
header	str | None	Raw header value, if present.	required

Returns:

Type	Description
str | None	The token, or None if the header is absent or not a bearer token.

Source code in src/cryptography_manager/api/auth.py

def extract_bearer(header: str | None) -> str | None:
    """Pull the token out of an Authorization header.

    Args:
        header: Raw header value, if present.

    Returns:
        The token, or ``None`` if the header is absent or not a bearer token.
    """
    if not header:
        return None
    parts = header.split(None, 1)
    if len(parts) != 2 or parts[0].lower() != "bearer":
        return None
    return parts[1].strip() or None

Routes

routes

HTTP routes.

The request body is read and validated inside the handler rather than through a typed signature. That is deliberate: the framework would otherwise consume and parse the body before the handler runs, which would bypass the audit trail, answer an unauthenticated caller with a schema complaint instead of a refusal, and buffer an arbitrarily large payload before anyone had a chance to object. Authorization is decided first, the body is then read under a size ceiling, and every rejection still produces an audit record.

health async

health(request: Request) -> dict[str, Any]

Report service liveness and which capabilities are available.

Source code in src/cryptography_manager/api/routes.py

@router.get("/health", tags=["service"])
async def health(request: Request) -> dict[str, Any]:
    """Report service liveness and which capabilities are available."""
    state = request.app.state
    return {
        "status": "ok",
        "auth_mode": state.settings.auth_mode,
        "operations": {
            "differential_privacy": "available",
            "encryption": "available",
            "key_management": "available",
            "homomorphic_encryption": "not_implemented",
            "smpc": "not_implemented",
        },
    }

cryptography async

cryptography(
    request: Request,
    authorization: str | None = Header(default=None),
) -> JSONResponse

Execute one cryptographic operation.

Parameters:

Name	Type	Description	Default
request	Request	The incoming request, carrying application state.	required
authorization	str | None	Bearer token issued by the identity provider.	Header(default=None)

Returns:

Type	Description
JSONResponse	The response envelope, with a status code reflecting the outcome.

Source code in src/cryptography_manager/api/routes.py

@router.post(
    "/cryptography",
    tags=["cryptography"],
    summary="Execute a cryptographic operation",
    response_model=CryptographyResponse,
    responses={
        200: {"description": "Operation completed successfully"},
        400: {"description": "Configuration or parameter validation failed"},
        401: {"description": "Authorization token missing or invalid"},
        403: {"description": "Token lacks the permission for this action"},
        429: {"description": "Privacy budget exhausted"},
        500: {"description": "Unexpected execution error"},
        413: {"description": "Request body exceeds the configured limit"},
        501: {"description": "Operation recognised but not implemented"},
    },
    openapi_extra={
        "requestBody": {
            "required": True,
            "content": {
                "application/json": {
                    "schema": CryptographyRequest.model_json_schema()
                }
            },
        }
    },
)
async def cryptography(
    request: Request,
    authorization: str | None = Header(default=None),
) -> JSONResponse:
    """Execute one cryptographic operation.

    Args:
        request: The incoming request, carrying application state.
        authorization: Bearer token issued by the identity provider.

    Returns:
        The response envelope, with a status code reflecting the outcome.
    """
    state = request.app.state
    workflow = state.workflow

    auth = state.verifier.verify(extract_bearer(authorization))
    if not auth.ok:
        outcome = workflow.run(None, None, auth.error)
        return _respond(outcome.status, outcome.response)

    def reject(status: WorkflowStatus, error: str) -> JSONResponse:
        outcome = workflow.reject(
            status,
            auth.user_id,
            error,
            client_id=auth.client_id,
            token_issuer=auth.issuer,
        )
        return _respond(outcome.status, outcome.response)

    limit = state.settings.max_request_bytes
    body, oversized = await _read_capped(request, limit)
    if oversized:
        return reject(
            WorkflowStatus.PAYLOAD_TOO_LARGE,
            f"Request body exceeds the {limit} byte limit",
        )

    try:
        payload = json.loads(body) if body.strip() else {}
    except json.JSONDecodeError as exc:
        return reject(
            WorkflowStatus.VALIDATION_ERROR,
            f"Request body is not valid JSON: {exc}",
        )

    if not isinstance(payload, dict):
        return reject(
            WorkflowStatus.VALIDATION_ERROR,
            "Request body must be a JSON object",
        )

    try:
        parsed = CryptographyRequest.model_validate(payload)
    except ValidationError as exc:
        return reject(WorkflowStatus.VALIDATION_ERROR, _render(exc))

    outcome = workflow.run(
        parsed,
        auth.user_id,
        permissions=auth.permissions,
        client_id=auth.client_id,
        token_issuer=auth.issuer,
    )
    return _respond(outcome.status, outcome.response)

Application entry point

main

Service entry point.

Builds the FastAPI application and wires together the pieces a request needs: configuration, the manager that dispatches to backends, per-user budget accounting, audit logging and token verification. Everything is assembled once at startup and shared, because rebuilding adapters or re-reading budget state per request would be both slow and, in the budget's case, wrong.

create_app

create_app(settings: Settings | None = None) -> FastAPI

Build the application.

Parameters:

Name	Type	Description	Default
settings	Settings | None	Runtime settings; read from the environment when omitted.	None

Returns:

Type	Description
FastAPI	The configured application.

Source code in src/cryptography_manager/main.py

def create_app(settings: Settings | None = None) -> FastAPI:
    """Build the application.

    Args:
        settings: Runtime settings; read from the environment when omitted.

    Returns:
        The configured application.
    """
    settings = settings or get_settings()
    _configure_logging(settings.log_level)

    config = Config()
    config_path = Path(settings.config_path)
    if config_path.exists():
        config.load_from_file(config_path)
        logger.info(f"Loaded configuration from {config_path}")
    else:
        logger.warning(
            f"No configuration file at {config_path}; using defaults"
        )

    manager = CryptographyManager(config=config)
    service_tokens = build_service_token_provider(settings)
    audit = AuditLogger(
        log_path=settings.audit_log_path,
        targets=settings.audit_targets,
        timeout=settings.forward_timeout_seconds,
        attempts=settings.forward_attempts,
        compliance_use_case_id=settings.compliance_use_case_id,
        token_provider=service_tokens.token if service_tokens else None,
    )
    budget_store = _build_budget_store(settings, config)

    @asynccontextmanager
    async def lifespan(_: FastAPI) -> AsyncIterator[None]:
        """Let in-flight audit forwards finish before the process exits."""
        yield
        audit.shutdown()

    app = FastAPI(
        title=TITLE,
        version=VERSION,
        description=DESCRIPTION,
        lifespan=lifespan,
    )

    app.state.settings = settings
    app.state.config = config
    app.state.manager = manager
    app.state.audit = audit
    app.state.budget_store = budget_store
    app.state.verifier = build_verifier(settings)
    app.state.workflow = CryptographyWorkflow(
        manager,
        audit,
        budget_store,
        permission_prefix=settings.oidc_permission_prefix,
    )

    app.include_router(router)

    targets = ", ".join(settings.audit_targets) or "local file only"
    logger.info(f"{TITLE} ready; audit targets: {targets}")
    return app

main

main() -> None

Console script entry point: serve the API with uvicorn.

Source code in src/cryptography_manager/main.py

def main() -> None:
    """Console script entry point: serve the API with uvicorn."""
    import uvicorn

    settings = get_settings()
    uvicorn.run(
        "cryptography_manager.main:app",
        host="0.0.0.0",
        port=6001,
        log_level=settings.log_level.lower(),
    )