Security Boundaries for Collection APIs

Workflow Context #

Collection APIs sit at the intersection of public access, institutional compliance, and digital preservation, and they are the last gate a record passes before it leaves the building. Within the Core Architecture & Collection Taxonomy pipeline this is the delivery boundary: everything upstream has already turned raw ingest into a canonical, rights-tagged record, and this stage decides which consumer may see which fields of it. The failure this page prevents is the quiet one — a donor restriction, a conservation note, or an unpublished provenance chain serialized into a public manifest because the delivery layer trusted the caller to ask only for what it was entitled to.

The security boundary is a stateless validation gate, not a firewall rule. External aggregators, harvesters, and internal DAM clients never query raw tables; every request traverses a gate that authenticates the consumer, resolves an institutional rights tier, strips fields that tier is not entitled to, and only then serializes an IIIF-compliant payload. That decoupling is what lets a public discovery portal and a rights-holder’s private reconciliation client read the same underlying record without the schema leaking privileged data across the boundary. The role that owns this code is the Python automation engineer; the role that audits its field-stripping ratios is the DAMS administrator answerable for a rights breach.

This gate consumes the rights fields set upstream — the rights_tier and rights_uri produced when designing museum object schemas and the RightsStatements.org URIs asserted during rights metadata mapping and licensing automation. It never re-derives rights state; it enforces the state the record already carries.

Prerequisites #

Before deploying the enforcement layer, confirm the following are in place:

• Python 3.9+ with from __future__ import annotations where PEP 604 unions appear in evaluated positions.
• httpx 0.24+ for the async internal client (connection pooling, timeouts, HTTP/2).
• pydantic v2 for request and rights-tier models (model_config, field_validator, model_dump).
• An identity source — OAuth2 introspection endpoint or a JWKS URL for JWT verification — that resolves an opaque access token to a set of rights scopes. The boundary caches the result; it does not invent tiers.
• The canonical record contract from the core taxonomy layer, so the boundary knows which fields exist and which are rights-sensitive.
• An internal fetch endpoint (behind mutual TLS) that returns full records to the boundary and to nothing else. Public clients must have no network path to it.
• A read replica for the repository, so public reads never contend with ingest writes.

Schema Reference: Request and Rights-Tier Contracts #

The boundary is defined by two Pydantic v2 models. The inbound CollectionRequest is what a consumer is allowed to send; the RightsTier is what a validated token resolves to. Field-level access is the intersection of the fields a request asks for and the allowed_fields its tier grants — never their union.

from __future__ import annotations

from pydantic import BaseModel, ConfigDict, Field


class CollectionRequest(BaseModel):
    """What a consumer is permitted to send across the boundary."""

    model_config = ConfigDict(strict=True, extra="forbid")

    record_id: str = Field(min_length=1, max_length=64)
    fields: list[str] = Field(default_factory=list)
    access_token: str = Field(min_length=1)
    idempotency_key: str = Field(min_length=1)


class RightsTier(BaseModel):
    """What a validated token resolves to — the ceiling on what may be served."""

    model_config = ConfigDict(strict=True, frozen=True)

    tier: str = Field(pattern="^(public|restricted|embargoed|internal)$")
    allowed_fields: list[str] = Field(min_length=1)
    max_resolution: int | None = None   # IIIF Image API pixel cap; None = unbounded

The extra="forbid" on CollectionRequest means an unexpected key — a hand-crafted include_internal: true, say — fails validation at the edge rather than being silently ignored and possibly honoured by a downstream handler.

Step-by-Step Implementation #

The enforcement layer executes an ordered pipeline: authenticate, validate, redact, deliver. Each step below is a gate that the previous step’s output must clear before the next runs.

Step 1 — Resolve the token to a rights tier (fail closed) #

Token validation is where the boundary either fails closed or leaks. The resolver admits only tokens with a known rights tier; an unrecognized token raises before any internal fetch is attempted. Production wiring populates the cache from OAuth2 introspection or JWT verification — the point is that an unknown token is rejected, never defaulted to public.

import httpx
from pydantic import BaseModel


class APIBoundary:
    def __init__(
        self,
        base_url: str,
        api_key: str,
        token_tiers: dict[str, RightsTier] | None = None,
    ) -> None:
        self.client = httpx.AsyncClient(base_url=base_url, timeout=10.0)
        self.api_key = api_key
        # Seeded from OAuth2 introspection / JWT validation in production.
        self._rights_cache: dict[str, RightsTier] = dict(token_tiers or {})

    async def validate_token(self, token: str) -> RightsTier:
        # Fail closed: only tokens with a resolved rights tier are admitted.
        # An unknown token is rejected, never defaulted to a public tier.
        tier = self._rights_cache.get(token)
        if tier is None:
            raise PermissionError("Invalid or unrecognized access token")
        return tier

Edge case: a token that was valid at cache-seed time but has since been revoked. For short-lived JWTs, verify the exp claim on every call rather than trusting the cache; for opaque tokens, introspect with a short TTL so a revocation propagates within seconds, not hours.

Step 2 — Redact fields by tier intersection #

Redaction happens before data leaves the internal network, and it is an intersection, never a union. The request may ask for donor_name; it is served only if donor_name is in the tier’s allowed_fields. Anything not on both lists is dropped at the projection, so the internal fetch never even asks the repository for fields the consumer cannot receive.

    async def _process_single(
        self, req: CollectionRequest, tier: RightsTier
    ) -> dict:
        # Intersection filter: only fields the request asked for AND the tier grants.
        allowed = [f for f in req.fields if f in tier.allowed_fields]
        payload = {"ids": [req.record_id], "fields": allowed}
        resp = await self.client.post(
            "/internal/fetch",
            json=payload,
            headers={"Authorization": f"Bearer {self.api_key}"},
        )
        resp.raise_for_status()
        return resp.json()

Edge case — the empty projection: if req.fields is empty, allowed is empty. Decide deliberately whether an empty projection means “the tier’s full default set” or “nothing”; defaulting to the full set is a common accidental over-share. Prefer returning the tier’s default public projection explicitly rather than inferring intent from an empty list.

Step 3 — Bound concurrency for high-volume harvests #

Aggregators harvest in bursts. A semaphore caps in-flight internal fetches so a single research query cannot exhaust the connection pool or the replica. Each request in the batch is authenticated independently — one poisoned token in a batch of ten fails only its own record, not the batch.

    async def batch_process(
        self, requests: list[CollectionRequest], batch_size: int = 10
    ) -> list[dict]:
        semaphore = asyncio.Semaphore(batch_size)

        async def _bounded(req: CollectionRequest) -> dict:
            async with semaphore:
                tier = await self.validate_token(req.access_token)
                return await self._process_single(req, tier)

        return await asyncio.gather(*(_bounded(r) for r in requests))

    async def aclose(self) -> None:
        await self.client.aclose()

Payload variants: the same intersection logic applies whether the consumer sends a single JSON body, an OAI-PMH ListRecords batch, or a CSV of record_ids uploaded for reconciliation. Parse the transport into CollectionRequest instances first, then run the identical gate — never special-case a transport into a bypass.

Step 4 — Serialize through the IIIF-compliant delivery channel #

Only after redaction does the payload become an IIIF Presentation API 3.0 response. The delivery step injects the @context, normalizes controlled-vocabulary URIs to their Getty authority form, and caps image delivery at the tier’s max_resolution through the IIIF Image API. Rights metadata maps to the record’s rights block per institutional policy. A validation failure at any earlier step never reaches here; it returns a deterministic HTTP 403 with a machine-readable error code instead.

Boundary Data Flow #

The gate is a strict four-stage sequence. An invalid token short-circuits to 403 before any internal fetch executes, so a rejected consumer never causes a repository read.

Rights and Access Routing #

Rights state is the routing key at this boundary, and it is read, never re-derived. The four tiers map to concrete serialization behaviour:

• public — the default discovery projection: title, object type, resolved AAT/TGN terms, a rights statement URI, and image derivatives capped at a public max_resolution. Donor and conservation fields are absent from allowed_fields, so they can never be requested into existence.
• restricted — adds provenance and condition fields for authenticated institutional clients, with image caps raised but still bounded by donor agreement.
• embargoed — a record whose rights block carries a future release date. The boundary treats it as public minus the embargoed fields until the embargo lifts; the routing that sets that date is owned by implementing embargo workflows.
• internal — full-fidelity access for reconciliation and audit, served only over mutual TLS to named service accounts, never to a public route.

The max_resolution cap is the mechanism that enforces a donor’s image restriction without per-request branching: because it lives on the RightsTier, every IIIF Image API request inherits the ceiling structurally. The rights URI itself is assigned upstream — see mapping RightsStatements.org to collection fields and the Creative Commons routing in routing Creative Commons licenses — and the boundary simply honours whatever the record carries.

IIIF and LIDO Serialization Compliance #

Output payloads must conform to the IIIF Presentation API 3.0 specification and to the LIDO 1.1 rights conventions carried through from the interchange layer described in mapping LIDO to internal databases. Concretely, the delivery step:

• injects the IIIF @context and emits a valid Manifest with items/Canvas/AnnotationPage structure — a payload validated against Presentation API 2.1 will pass a lenient checker yet fail in a compliant 3.0 viewer, so validate against the exact major version your viewer serves;
• normalizes every controlled term to its Getty authority URI rather than a free-text label, matching the JSON-LD contract in how to structure JSON-LD for museum objects;
• maps rights to the manifest rights property as a single RightsStatements.org or Creative Commons URI, and surfaces human-readable terms in requiredStatement;
• caps width/height on image services at the tier’s max_resolution so the Image API never offers a derivative larger than the donor agreement permits.

Verification and Testing #

The boundary earns trust only if fail-closed behaviour and intersection redaction are asserted, not assumed. The test below exercises both the rejection path and the projection path with no live network.

import asyncio

import pytest


@pytest.mark.asyncio
async def test_unknown_token_fails_closed() -> None:
    boundary = APIBoundary(base_url="https://internal.invalid", api_key="k")
    with pytest.raises(PermissionError):
        await boundary.validate_token("not-a-real-token")
    await boundary.aclose()


def test_intersection_never_over_serves() -> None:
    tier = RightsTier(tier="public", allowed_fields=["title", "object_type"])
    requested = ["title", "donor_name", "object_type"]
    allowed = [f for f in requested if f in tier.allowed_fields]
    # donor_name is asked for but not granted — it must not survive.
    assert allowed == ["title", "object_type"]
    assert "donor_name" not in allowed


def test_request_rejects_unknown_keys() -> None:
    from pydantic import ValidationError

    with pytest.raises(ValidationError):
        CollectionRequest.model_validate(
            {
                "record_id": "1",
                "access_token": "t",
                "idempotency_key": "k",
                "include_internal": True,  # extra="forbid" rejects this
            }
        )

Run the suite with pytest -q. In production, wire the two invariants — zero unknown-token fetches and a non-zero field-stripping ratio on public tiers — into monitoring, and alert if either drifts. A field-stripping ratio that suddenly drops to zero on a public endpoint means a tier’s allowed_fields widened by mistake.

Related Workflows at This Boundary #

This gate is the delivery end of the taxonomy pipeline, so its closest neighbours are the stages that hand it a record and the rights logic that sets what it must enforce.

• The record contract it serves is defined in designing museum object schemas — the boundary can only redact fields the canonical model names.
• The interchange payload it flattens and re-serializes originates in mapping LIDO to internal databases, including the Dublin Core crosswalk validation that keeps aggregator payloads conformant.
• The vocabulary URIs it normalizes on output are resolved in implementing Getty AAT and TGN.
• The rights state it enforces is assigned across rights metadata mapping and licensing automation, including embargo dates and RightsStatements.org URIs.

Frequently Asked Questions #

Why fail closed instead of defaulting an unknown token to the public tier? #

A public default turns every bug in the token pipeline into a silent over-share: a mistyped introspection URL, an expired cache, or a network blip would all resolve to “serve the public projection” instead of “deny.” Failing closed means those conditions surface as 403s and alerts rather than as leaked records. The cost — a legitimate consumer briefly denied during an outage — is recoverable; a leaked donor restriction is not.

Should redaction filter fields, or should the internal fetch just never return them? #

Both, layered. The intersection filter builds the projection the internal fetch requests, so the repository is never even asked for disallowed fields — that is the primary control. Keep a second redaction pass at serialization as defence in depth, so a future code path that fetches a wider projection still cannot serialize a field outside the tier. Rely on one control only if you are comfortable a single regression can breach it.

How do embargoed records differ from restricted ones at this boundary? #

A restricted record is permanently gated to authenticated tiers. An embargoed record is public-eligible but time-locked: the boundary serves it as public-minus-embargoed-fields until a release date passes, after which it becomes fully public with no code change. The date logic lives upstream in implementing embargo workflows; the boundary only reads the resulting tier.

What stops a bulk harvester from exhausting the repository? #

A concurrency semaphore caps in-flight internal fetches per batch, and the idempotency_key de-duplicates retried requests so a client re-issuing a timed-out call never triggers duplicate asset generation. Pair these with per-token rate limits at the edge; the semaphore protects the replica, the rate limit protects the boundary itself.

Why cap resolution on the rights tier rather than per request? #

Because a per-request cap is a rule someone can forget to apply, while a cap on the RightsTier model is inherited by every IIIF Image API request that tier makes. Structural enforcement removes the possibility of a code path that serves a full-resolution derivative to a tier a donor agreement restricts. The ceiling is data, not a conditional.

Related #

• Core Architecture & Collection Taxonomy — parent pipeline overview
• Designing Museum Object Schemas — the redactable record contract
• Mapping LIDO to Internal Databases — interchange source payload
• Implementing Getty AAT & TGN — output vocabulary URIs
• Scoping OAuth2 Tokens for Collection APIs — least-privilege token design
• Rights Metadata Mapping & Licensing — assigns enforced rights state