Security Boundary Configuration

A security boundary in a research grant automation stack is not a firewall rule — it is the single place where an untrusted payload becomes a trusted record. Grant documents arrive from sponsor portals, equipment manifests arrive from departmental LIMS feeds, personnel certifications arrive from HR exports, and every one of those sources is, until proven otherwise, hostile to the integrity of the system. The boundary is the layer that decides what crosses into the trusted zone, with what privileges, carrying what compliance classification, and leaving what audit evidence behind. This guide specifies how to configure that layer so the decision is deterministic, least-privilege by default, and reproducible under audit. It is anchored to the foundational principles set out in Core Architecture & Policy Mapping for Research Grants.

For university administrators, research compliance officers, Python automation developers, and laboratory managers, the design goal is direct: every record that enters the trusted zone has been validated against the active rule set, tagged with an explicit data classification, scoped to the credentials that may touch it, and anchored by a field-level hash that an auditor can recompute months later. Nothing crosses on trust. A re-poll of the same payload converges to the same committed state; a malformed or over-privileged payload is diverted to a quarantine queue with an explicit reason and never contaminates downstream systems.

Problem Framing: The Trust-Zone Gap #

The recurring failure is treating ingress as a transport problem rather than a policy problem. A sync job authenticates to a sponsor API, pulls a batch, and writes it straight into the grants database because “the connection is encrypted.” Transport security says nothing about whether the payload is well-formed, whether the calling service account is allowed to write that grant’s records, or whether the row carries human-subjects data that must never sit in an unencrypted column. The connection is secure and the data is still wrong.

A security boundary closes that gap by making three guarantees explicit and enforced in code rather than assumed. First, classification: every field that crosses is tagged with a handling class — controlled unclassified information (CUI), hazardous-material inventory, human-subjects data, or public — and that tag determines storage and access for the rest of the record’s life. Second, least privilege: an automated job inherits exactly the scope of its service account and cannot escalate; a script that syncs Lab A’s inventory cannot write Lab B’s records even if the payload claims it should. Third, auditability: the boundary records why each record was accepted or rejected, and produces a hash an auditor can reproduce. This guide builds those guarantees in layers — the policy constraints that bound them, the canonical schema they operate on, the idempotent implementation that enforces them, the integration contracts with adjacent systems, and the verification and recovery procedures that keep the whole thing defensible. The boundary consumes its rules from the University Policy Mapping Frameworks and protects the records owned by the Grant Lifecycle Architecture Design; it is the subsystem that owns what is allowed to enter the trusted zone, with what privilege, and under what classification.

Policy Constraints on Boundary Enforcement #

The boundary does not invent its rules. It consumes a versioned predicate set published by the policy mapping layer and applies the subset that governs ingress and access. What the boundary owns is which control applies to which classification of data — the translation of regulation into machine-readable predicates lives upstream, and the boundary records the rule version it evaluated against so a decision can be replayed.

• NIH Grants Policy Statement governs human-subjects and protected-health classifications. Any field tagged HUMAN_SUBJECTS at the boundary must be stored encrypted at rest and is readable only by service accounts holding the phi:read scope; the boundary rejects a write that would persist such a field in a public column.
• NSF PAPPG and 2 CFR 200 (Uniform Guidance) set the controls for financial and award-identity fields. Cost and indirect-rate fields are classified CUI and gated by allowable-cost predicates before they may be committed against an active award.
• OSHA 29 CFR 1910.1450 (Laboratory Standard) and EPA RCRA (40 CFR Part 262) apply when a payload carries chemical or hazardous-material inventory. These records are classified HAZMAT, and a boundary write triggers an inventory-reconciliation predicate that links to the downstream Equipment Calibration & Lab Inventory Tracking domain before the record is trusted.
• NIST SP 800-53 Rev. 5 access controls (AC-3, AC-6, AC-17) express as the least-privilege scoping model itself: mutual-TLS identity on every cross-boundary call, scope checks on every write, and key rotation on the signing material that anchors the audit hash.

Two rules hold across every classification. Policy is evaluated before a record is committed, never after — a rejected record never reaches the trusted zone and the rejection is logged. And a policy version change never mutates an already-committed record; the new rule version affects only future boundary decisions, which is what keeps historical audit hashes reproducible. When a degraded primary route forces traffic onto a backup path, the same boundary predicates still apply — the routing logic for that case is specified in the Fallback Routing Protocols.

Data Schema & Field Mapping #

The boundary operates on one canonical ingress record and an append-only decision ledger. Heterogeneous source payloads — an eRA Commons export, a Research.gov feed, a LIMS manifest, an HR certification extract — are mapped to this canonical shape at the moment of ingress; the wire-level transport rules for those source connections are detailed in Configuring secure API boundaries for research data sync. The canonical schema below is the contract every source must be mapped onto before validation runs.

Two field-level rules matter for reproducibility. The idempotency_key is computed only over canonical, validated content — never over a wall-clock timestamp or unsorted dict — so the same logical record always yields the same key on replay. And classification is immutable once committed: a later payload that proposes a lower classification for an existing record (for example, demoting a HUMAN_SUBJECTS row to PUBLIC) is treated as a policy violation, not an update.

Implementation: Validating and Committing Across the Boundary #

The implementation pattern is the same one used throughout this architecture: a Pydantic model normalizes and validates the payload, a deterministic key is derived from canonical content, and a SQLAlchemy upsert commits the record exactly once while routing every rejection to a quarantine queue. The boundary processor below is idempotent — repeated runs over the same batch converge to the same committed state — and it never raises out of the batch loop, so one bad record cannot halt a multi-million-dollar instrumentation grant’s sync.

The flow each payload follows across the boundary:

Every payload crosses a single validation boundary — non-compliant or over-privileged records divert to the quarantine queue, never the trusted zone.

from __future__ import annotations

import hashlib
import json
import logging
from datetime import datetime, timezone
from enum import Enum

from pydantic import BaseModel, Field, field_validator
from sqlalchemy import String, JSON, DateTime
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session
from sqlalchemy.dialects.postgresql import insert as pg_insert

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)s | %(name)s | %(message)s",
)
logger = logging.getLogger("security_boundary")


class Classification(str, Enum):
    PUBLIC = "PUBLIC"
    CUI = "CUI"
    HUMAN_SUBJECTS = "HUMAN_SUBJECTS"
    HAZMAT = "HAZMAT"


# Ordered from least to most sensitive; a record may never be re-committed
# at a lower rank than it already holds in the trusted zone.
_RANK: dict[Classification, int] = {
    Classification.PUBLIC: 0,
    Classification.CUI: 1,
    Classification.HUMAN_SUBJECTS: 2,
    Classification.HAZMAT: 2,
}


class IngressRecord(BaseModel):
    """Canonical shape every source payload is mapped onto before it may cross."""

    grant_id: str
    source_system: str
    classification: Classification
    owner_scope: str
    payload: dict
    policy_version: str

    @field_validator("grant_id", "owner_scope")
    @classmethod
    def _normalize(cls, value: str) -> str:
        normalized = value.strip().upper()
        if not normalized:
            raise ValueError("identity field must be non-empty")
        return normalized

    def idempotency_key(self) -> str:
        """SHA-256 over canonical identity + payload + policy version.

        Pydantic's model_dump_json with sorted keys fixes field order, so the
        same logical record always yields the same key on replay.
        """
        anchor = {
            "grant_id": self.grant_id,
            "source_system": self.source_system,
            "classification": self.classification.value,
            "payload": self.payload,
            "policy_version": self.policy_version,
        }
        canonical = json.dumps(anchor, sort_keys=True, separators=(",", ":"))
        return hashlib.sha256(canonical.encode("utf-8")).hexdigest()


class Base(DeclarativeBase):
    pass


class BoundaryLedger(Base):
    __tablename__ = "boundary_ledger"

    idempotency_key: Mapped[str] = mapped_column(String, primary_key=True)
    grant_id: Mapped[str] = mapped_column(String, index=True)
    classification: Mapped[str] = mapped_column(String)
    owner_scope: Mapped[str] = mapped_column(String)
    state_hash: Mapped[str] = mapped_column(String)
    policy_version: Mapped[str] = mapped_column(String)
    committed_at: Mapped[datetime] = mapped_column(DateTime(timezone=True))
    payload: Mapped[dict] = mapped_column(JSON)


class SecurityBoundary:
    """Idempotent ingress controller for grant and lab-inventory payloads."""

    def __init__(self, session: Session, granted_scopes: set[str]) -> None:
        self.session = session
        # The scopes the calling service account actually holds. A payload
        # claiming an owner_scope outside this set is an escalation attempt.
        self.granted_scopes = granted_scopes
        self.quarantine: list[dict] = []

    def _quarantine(self, raw: dict, reason: str) -> None:
        logger.warning("Boundary rejected (%s): %s", reason, raw.get("grant_id", "UNKNOWN"))
        self.quarantine.append({"payload": raw, "reason": reason})

    def _existing_rank(self, grant_id: str) -> int | None:
        row = (
            self.session.query(BoundaryLedger)
            .filter(BoundaryLedger.grant_id == grant_id)
            .order_by(BoundaryLedger.committed_at.desc())
            .first()
        )
        return _RANK[Classification(row.classification)] if row else None

    def ingest(self, raw_payloads: list[dict]) -> dict[str, int]:
        stats = {"committed": 0, "idempotent_hit": 0, "quarantined": 0}

        for raw in raw_payloads:
            # 1. Schema boundary — malformed payloads never reach scope checks.
            try:
                record = IngressRecord.model_validate(raw)
            except Exception as exc:  # pydantic.ValidationError and friends
                self._quarantine(raw, f"schema_violation: {exc.__class__.__name__}")
                stats["quarantined"] += 1
                continue

            # 2. Least-privilege boundary — no escalation past the account's scope.
            if record.owner_scope not in self.granted_scopes:
                self._quarantine(raw, "scope_denied")
                stats["quarantined"] += 1
                continue

            # 3. Classification boundary — sensitivity may never be downgraded.
            prior_rank = self._existing_rank(record.grant_id)
            if prior_rank is not None and _RANK[record.classification] < prior_rank:
                self._quarantine(raw, "classification_downgrade")
                stats["quarantined"] += 1
                continue

            key = record.idempotency_key()

            # 4. Idempotent commit — INSERT ... ON CONFLICT DO NOTHING means a
            # re-polled payload is a no-op, not a duplicate row.
            state_hash = hashlib.sha256(
                record.model_dump_json().encode("utf-8")
            ).hexdigest()
            stmt = (
                pg_insert(BoundaryLedger)
                .values(
                    idempotency_key=key,
                    grant_id=record.grant_id,
                    classification=record.classification.value,
                    owner_scope=record.owner_scope,
                    state_hash=state_hash,
                    policy_version=record.policy_version,
                    committed_at=datetime.now(timezone.utc),
                    payload=record.payload,
                )
                .on_conflict_do_nothing(index_elements=["idempotency_key"])
            )
            result = self.session.execute(stmt)
            if result.rowcount == 0:
                stats["idempotent_hit"] += 1
                logger.info("Idempotent hit: %s", key[:12])
            else:
                stats["committed"] += 1
                logger.info("Boundary accepted: %s | %s", record.grant_id, key[:12])

        self.session.commit()
        return stats

The four checks run in a fixed order — schema, then scope, then classification, then idempotent commit — because each depends on the previous having passed. A payload that fails validation never reaches the scope check; a payload that fails the scope check never touches the trusted zone. Routing exceptions to self.quarantine instead of raising keeps the batch alive: the saturation and replay behavior of that queue is handled the same way as in the Grant Lifecycle Architecture Design, so quarantined records re-enter under fresh keys once their source is corrected.

Integration Points #

The boundary sits between external systems and the trusted zone, so its contract with each adjacent system is part of the configuration, not an afterthought.

• Sponsor portals (eRA Commons, Research.gov). Inbound award and financial records arrive over mutual TLS and are mapped to IngressRecord with source_system set accordingly. Cost fields are classified CUI; the boundary commit feeds the canonical award record owned by the lifecycle model.
• LIMS / lab inventory. Equipment and chemical manifests arrive from departmental systems and are classified HAZMAT when they carry regulated substances, triggering the reconciliation predicate before commit. The committed records flow downstream to Equipment Calibration & Lab Inventory Tracking.
• ERP / finance. Indirect-cost and encumbrance data is exported back out of the trusted zone only for grants the consuming account is scoped to, so the same least-privilege model governs egress as ingress.
• Ingestion orchestration. The batch scheduling, retry, and back-pressure logic that drives the boundary lives in Automated Ingestion & Data Sync Workflows; the boundary is the validation stage that orchestration calls into.

A representative inbound payload, after a source adapter has mapped it onto the canonical shape but before it crosses the boundary:

{
  "grant_id": "5R01GM123456-03",
  "source_system": "ERA_COMMONS",
  "classification": "HUMAN_SUBJECTS",
  "owner_scope": "lab:neuro:write",
  "policy_version": "2026.02.1",
  "payload": {
    "pi_name": "Okafor, A.",
    "irb_protocol": "IRB-2025-0412",
    "indirect_rate": 0.555
  }
}

Verification & Audit #

Verification is a hash comparison, not a trust exercise. Every committed record carries a state_hash computed over its canonical, validated content. To confirm a record in the trusted zone was not tampered with after commit, an auditor re-runs the same canonical serialization over the stored fields and compares:

def verify_ledger_row(row: BoundaryLedger) -> bool:
    """Recompute the state hash from stored canonical content and compare."""
    rebuilt = IngressRecord(
        grant_id=row.grant_id,
        source_system="ERA_COMMONS",  # stored alongside in production
        classification=Classification(row.classification),
        owner_scope=row.owner_scope,
        payload=row.payload,
        policy_version=row.policy_version,
    )
    expected = hashlib.sha256(rebuilt.model_dump_json().encode("utf-8")).hexdigest()
    return expected == row.state_hash

Because the hash is computed only over canonical content — never over committed_at or any mutable column — a reproducible match confirms integrity, and any divergence is itself an audit finding. The decision ledger is append-only: a correction is a new row under a new idempotency key, never an in-place edit, so the full history of what crossed the boundary and why remains intact for the records-retention period required by 2 CFR 200.

Failure Modes & Recovery #

Recovery never edits the trusted zone directly; every correction flows back through ingest so it is validated, scope-checked, and logged like any other crossing.

Frequently Asked Questions #

Related #

• Core Architecture & Policy Mapping for Research Grants — the parent architecture this boundary plugs into.
• University Policy Mapping Frameworks — where the regulatory predicates the boundary evaluates are codified and versioned.
• Grant Lifecycle Architecture Design — the state model whose records this boundary protects on the way in.
• Fallback Routing Protocols — how boundary predicates still apply when a primary route degrades.
• Configuring secure API boundaries for research data sync — the wire-level transport and gateway setup beneath this layer.
• Automated Ingestion & Data Sync Workflows — the orchestration that schedules and retries the batches this boundary validates.