Security Boundary Configuration
A change order is a financial instrument before it is a document. By the time a field-initiated request reaches an executive approver, it has touched a subcontractor portal, an estimator’s recalculation engine, and at least one accounting export — and at every hop a different party is entitled to see a different slice of the record. The specific sub-problem this page solves is how a pipeline decides, deterministically and on every payload, which fields and cost lines a submitting tier is allowed to write or read, and what happens when a payload reaches past that boundary. When that decision is left implicit, a subcontractor portal leaks general-contractor markup into a trade view, an aggregation job rolls a privileged contingency line into a report the owner sees, and a privilege-escalated payload silently overwrites an approved baseline. Inside a deterministic construction data architecture and taxonomy, the security boundary is the contract that pins every record to an origin tier and refuses to advance it past the scope that tier holds. This page details the boundary subsystem end to end — the typed payload contract, the access matrix that maps tiers to authorized scope, the boundary-aware recalculation step, and the routing that auto-advances, holds for review, or quarantines each record. It targets Python automation builders, project engineers, and estimators who need tier isolation to survive real-world input variance.
Prerequisites
The boundary subsystem sits downstream of document extraction and upstream of the change order ledger and the approval router. It assumes the inbound payload has already cleared structural schema validation rules at the API gateway, so the work here is authorization and tier isolation, not parsing. Before implementing the patterns below, you need:
- Python 3.11+ with
pydanticv2 for typed validation, plus the standard-librarydecimal,re,enum, andloggingmodules. No floating-point money touches a cost field; every monetary value is aDecimal. - A canonical scope taxonomy to authorize against. Boundary rules resolve a payload’s location to the same Work Breakdown Structure that drives WBS mapping strategies; a tier is granted scope over WBS subtrees, not over arbitrary strings.
- A standardized cost vocabulary so the access matrix can name what a tier may touch. Authorized cost codes reference the canonical keys defined by budget code standardization rather than ad-hoc labels, which is what lets one matrix entry cover an entire division.
- An identity provider that issues tokens carrying tier and project claims. The
access_tieron every payload must be derived from a verified token scope at the gateway, never trusted from the request body. - A task queue — Celery on a Redis or RabbitMQ broker — so quarantined payloads can be parked in a dead-letter queue and replayed after review instead of dropped. The escalation policy for parked records is owned by fallback alert routing, and the structured-logging conventions follow the shared error handling protocols.
Architecture: tiers, the access matrix, and routing branches
A security boundary is two orthogonal concerns held apart on purpose. The first is a field-and-scope contract: which cost codes and WBS subtrees a given tier may write, expressed as data rather than code so it can be audited and versioned. The second is a routing decision: given a validated payload, does it auto-advance, wait for a human, or get quarantined. Keeping these separate is what lets you tighten a matrix entry without touching routing logic, and re-tune routing thresholds without re-deriving who-can-see-what. The diagram below traces a single payload from the gateway through both concerns to one of three terminal states.
The three quarantine branches are not failure paths to be smoothed over — they are the audit trail. Every quarantined record is logged with its origin tier, the boundary it breached, and an immutable timestamp, which is exactly the evidence a payment dispute or a security review needs. The confidence thresholds (0.92 auto-route, 0.75–0.92 human-review, below 0.75 quarantine) are the same site-canonical bands used across the routing layer, so a boundary decision reads consistently against an RFI or submittal decision.
Step-by-step implementation
Step 1 — Model the payload as a typed contract
The boundary cannot be enforced on a dictionary. Model the change order as a Pydantic v2 contract whose fields carry the construction-domain constraints directly: the WBS element matches the PROJ-NNN-DIV-NN pattern, the cost code matches a MasterFormat XX XX XX division pattern, the discipline is a closed enum, and the tier is a Literal. Monetary fields are Decimal so quantity-times-rate aggregations never drift. Constraining at the type boundary means a malformed record is rejected before any authorization logic runs.
from __future__ import annotations
import logging
import re
from decimal import Decimal
from enum import Enum
from typing import Literal
from pydantic import BaseModel, ConfigDict, Field, field_validator
logger = logging.getLogger("boundary")
WBS_PATTERN = re.compile(r"^PROJ-\d{3}-(ARCH|STR|MEP|CIV|ELEC|PLMB)-\d{2}$")
COST_CODE_PATTERN = re.compile(r"^\d{2} \d{2} \d{2}$") # MasterFormat XX XX XX
Discipline = Literal["ARCH", "STR", "MEP", "CIV", "ELEC", "PLMB"]
class AccessTier(str, Enum):
SUBCONTRACTOR = "subcontractor"
ESTIMATOR = "estimator"
EXECUTIVE = "executive"
class ChangeOrderLineItem(BaseModel):
model_config = ConfigDict(extra="forbid") # reject smuggled fields
cost_code: str = Field(..., description="MasterFormat division code, XX XX XX")
description: str = Field(..., max_length=512)
quantity: Decimal = Field(..., ge=0)
unit_rate: Decimal = Field(..., ge=0)
@field_validator("cost_code")
@classmethod
def cost_code_is_masterformat(cls, v: str) -> str:
if not COST_CODE_PATTERN.match(v):
raise ValueError(f"cost_code '{v}' is not a MasterFormat XX XX XX code")
return v
@property
def line_total(self) -> Decimal:
return self.quantity * self.unit_rate
class ChangeOrderPayload(BaseModel):
model_config = ConfigDict(extra="forbid")
project_id: str = Field(..., pattern=r"^PROJ-\d{3}$")
wbs_element: str
cost_center: str
discipline: Discipline
access_tier: AccessTier
line_items: list[ChangeOrderLineItem] = Field(..., min_length=1)
@field_validator("wbs_element")
@classmethod
def wbs_matches_pattern(cls, v: str) -> str:
if not WBS_PATTERN.match(v):
raise ValueError(f"wbs_element '{v}' must match PROJ-NNN-DIV-NN")
return vThe extra="forbid" config is load-bearing: it is what stops an attacker (or a careless integration) from smuggling a privileged markup_pct or internal_note field into a subcontractor payload. Any unmodeled key raises a validation error before the record is trusted.
Step 2 — Express the access matrix as versioned data
The authorization rules belong in data, not in branching code, so they can be reviewed, diffed, and version-pinned alongside the schema. Each tier maps to the cost-code prefixes it may write and the WBS disciplines it owns. A subcontractor is scoped to its trade’s divisions; an estimator owns the markup and contingency codes; an executive sees everything but, by separation of duties, does not author trade line items.
from pydantic import BaseModel, ConfigDict
class TierGrant(BaseModel):
model_config = ConfigDict(extra="forbid")
cost_code_prefixes: tuple[str, ...] # e.g. ("03", "04") for concrete/masonry
disciplines: tuple[Discipline, ...]
may_apply_markup: bool
# Version this object; pin the version into every quarantine log line.
ACCESS_MATRIX_VERSION = "2026.06"
ACCESS_MATRIX: dict[AccessTier, TierGrant] = {
AccessTier.SUBCONTRACTOR: TierGrant(
cost_code_prefixes=("03", "04", "05"),
disciplines=("STR", "CIV"),
may_apply_markup=False,
),
AccessTier.ESTIMATOR: TierGrant(
cost_code_prefixes=("01", "03", "04", "05", "23", "26"),
disciplines=("ARCH", "STR", "MEP", "CIV", "ELEC", "PLMB"),
may_apply_markup=True,
),
AccessTier.EXECUTIVE: TierGrant(
cost_code_prefixes=("01",), # executives approve, they don't author trade lines
disciplines=("ARCH", "STR", "MEP", "CIV", "ELEC", "PLMB"),
may_apply_markup=False,
),
}Step 3 — Enforce the boundary deterministically
With the contract and the matrix in place, enforcement is a single pure function that takes a parsed payload and the matrix, and returns either a sanitized payload or a typed violation. It checks every cost code against the tier’s authorized prefixes and the WBS discipline against the tier’s owned disciplines. Crucially, the function never silently drops a bad line — a single out-of-scope code fails the whole payload, because partial acceptance is how privilege escalation hides.
class BoundaryViolation(Exception):
"""Raised when a payload reaches past its tier's authorized scope."""
def __init__(self, reason: str, tier: AccessTier) -> None:
self.reason = reason
self.tier = tier
super().__init__(reason)
def enforce_boundary(
payload: ChangeOrderPayload,
matrix: dict[AccessTier, TierGrant],
) -> ChangeOrderPayload:
grant = matrix.get(payload.access_tier)
if grant is None:
raise BoundaryViolation("unknown tier", payload.access_tier)
# 1. WBS discipline must fall within the tier's owned disciplines.
if payload.discipline not in grant.disciplines:
raise BoundaryViolation(
f"discipline {payload.discipline} outside tier scope", payload.access_tier
)
# 2. Every cost code must match an authorized MasterFormat division prefix.
for item in payload.line_items:
division = item.cost_code[:2]
if division not in grant.cost_code_prefixes:
raise BoundaryViolation(
f"cost_code {item.cost_code} (div {division}) not granted",
payload.access_tier,
)
# 3. Re-emit a clean copy; extra="forbid" already guarantees no smuggled fields.
return payload.model_copy(deep=True)Step 4 — Recalculate within authority, then route by confidence
Once a payload clears the boundary, financial recalculation must respect the same scope. The two-phase rule is: validate that each line maps to an active, unfrozen budget code, then apply markup and tax only if the submitting tier holds may_apply_markup. A subcontractor may revise a material quantity, but the system must never auto-apply overhead or profit on that tier’s behalf — that authority belongs to the estimator. The recalculated total then drives a routing decision against the canonical confidence bands.
def recalc_and_route(
payload: ChangeOrderPayload,
matrix: dict[AccessTier, TierGrant],
boundary_confidence: float,
markup_pct: Decimal = Decimal("0.15"),
) -> str:
"""Return a routing state: 'auto_route', 'human_review', or 'quarantine'."""
grant = matrix[payload.access_tier]
subtotal = sum((li.line_total for li in payload.line_items), Decimal("0"))
if grant.may_apply_markup:
total = subtotal * (Decimal("1") + markup_pct)
else:
total = subtotal # boundary forbids auto-markup for this tier
logger.info(
"boundary.recalc tier=%s wbs=%s total=%s matrix_version=%s",
payload.access_tier.value, payload.wbs_element, total, ACCESS_MATRIX_VERSION,
)
if boundary_confidence >= 0.92:
return "auto_route"
if boundary_confidence >= 0.75:
return "human_review"
return "quarantine"The full ingestion entrypoint wires these together: parse, enforce, recalc-and-route, and on any BoundaryViolation or ValidationError, quarantine with a structured log line rather than raising into the request path. The same dead-letter handoff used elsewhere in the pipeline applies, so a parked record can be reviewed and replayed; the async fan-out for high-volume bid periods follows the shared async batching workflows.
Configuration reference
The fields and configuration keys that govern this subsystem:
| Field / key | Type / pattern | Rule |
|---|---|---|
project_id |
^PROJ-\d{3}$ |
Identifies the project; must match the token’s project claim |
wbs_element |
PROJ-NNN-DIV-NN |
Resolves to a node in the canonical WBS; discipline segment is authorization-relevant |
cost_code |
MasterFormat XX XX XX |
First two digits (division) checked against the tier’s cost_code_prefixes |
discipline |
ARCH|STR|MEP|CIV|ELEC|PLMB |
Must fall within the tier’s owned disciplines |
access_tier |
subcontractor|estimator|executive |
Derived from a verified token scope, never trusted from the body |
line_items[].quantity / unit_rate |
Decimal, >= 0 |
Never float; protects aggregation integrity |
may_apply_markup |
matrix flag | Gates automatic markup/tax application per tier |
ACCESS_MATRIX_VERSION |
string | Pinned into every log line so a decision can be reproduced |
| boundary confidence | float | >= 0.92 auto-route, 0.75–0.92 human-review, < 0.75 quarantine |
The extra="forbid" model config and the Literal/regex constraints are themselves configuration: tightening a pattern or a tier grant is a reviewable, version-controlled change, not a code rewrite.
Verification and testing
Boundary code is exactly the kind of logic that must be tested against adversarial inputs, not just happy-path payloads. A minimal assertion suite confirms that authorized records pass, scope violations raise, and smuggled fields are rejected.
import pytest
from pydantic import ValidationError
def make_payload(**over) -> dict:
base = {
"project_id": "PROJ-014",
"wbs_element": "PROJ-014-STR-02",
"cost_center": "CC-STR",
"discipline": "STR",
"access_tier": "subcontractor",
"line_items": [
{"cost_code": "03 30 00", "description": "CIP concrete",
"quantity": "120", "unit_rate": "145.50"}
],
}
base.update(over)
return base
def test_authorized_subcontractor_passes():
p = ChangeOrderPayload(**make_payload())
assert enforce_boundary(p, ACCESS_MATRIX).wbs_element == "PROJ-014-STR-02"
def test_out_of_scope_cost_code_quarantines():
bad = make_payload(line_items=[
{"cost_code": "26 05 00", "description": "electrical",
"quantity": "1", "unit_rate": "1000"}])
p = ChangeOrderPayload(**bad)
with pytest.raises(BoundaryViolation):
enforce_boundary(p, ACCESS_MATRIX)
def test_smuggled_markup_field_rejected():
bad = make_payload()
bad["line_items"][0]["markup_pct"] = "0.30" # not in the model
with pytest.raises(ValidationError):
ChangeOrderPayload(**bad)
def test_subcontractor_total_has_no_markup():
p = ChangeOrderPayload(**make_payload())
state = recalc_and_route(p, ACCESS_MATRIX, boundary_confidence=0.97)
assert state == "auto_route" # and no markup was applied to the logged totalFor a fast manual check, serialize a sanitized payload with model_dump_json(indent=2) and confirm it contains no field the tier was not entitled to write — the cleanest evidence that extra="forbid" and the matrix are doing their jobs together.
Troubleshooting
European decimal formats fail validation. A payload from an EU accounting export sends "1.450,00" and quantity/unit_rate raise a ValidationError. Root cause: Pydantic does not parse locale-formatted decimals. Fix: normalize the thousands and decimal separators in the gateway’s pre-parse step before the value reaches the Decimal field — never relax the field type to str, which would re-open the float-drift hole.
A legitimate cross-trade change order is quarantined. A design-build subcontractor with structural and electrical scope submits an ELEC line and gets a scope violation. Root cause: the matrix grant for that tier lists only STR/CIV disciplines. Fix: this is a data change, not a code change — extend the tier’s disciplines/cost_code_prefixes grant, bump ACCESS_MATRIX_VERSION, and let the audit log record the widened scope.
Tier appears authorized but markup is silently missing. An estimator’s recalculated totals omit overhead. Root cause: the payload’s access_tier was populated from the request body, where a portal defaulted it to subcontractor, so may_apply_markup is False. Fix: derive access_tier exclusively from the verified token scope at the gateway, and assert token-tier equals body-tier (the first branch in the architecture diagram) before parsing.
Confidence hovers at the 0.92 boundary and records oscillate. Borderline payloads flip between auto-route and human-review across retries. Root cause: a non-deterministic confidence input. Fix: make boundary confidence a pure function of the validated payload so a broker retry reproduces the identical score; pair it with an idempotent commit keyed on project_id plus change-order number so a redelivered message is a no-op.
extra="forbid" breaks an upstream integration. A partner system that appends a tracking field now fails every payload. Root cause: the integration relies on pass-through extra fields. Fix: do not loosen the model; map the partner field explicitly into the schema if it is legitimate, or strip it in the gateway adapter so the boundary contract stays closed.
Frequently Asked Questions
For role-by-role implementation detail — evaluating user claims against project-specific boundaries and routing authorized requests to scoped endpoints — see Setting up role-based access control for subcontractor portals.
The access matrix from Step 2, read as scope per tier — which cost divisions each tier may author (write), see only (read), or not touch at all: