Security & Access Boundaries for Waste Route Optimization

A routing platform that dispatches municipal collection fleets is also a control surface: whoever can reach the solver can reshape where trucks go, what a manifest claims was collected, and which driver is recorded as responsible. When the access boundary is weak — a shared service token, an audit log that can be edited after the fact, a credential that stays valid after a driver’s Hours of Service ceiling is reached — the platform produces routes that are operationally plausible and legally indefensible. This reference defines the perimeter that prevents that: how a request’s credential is verified and propagated, how tenant fleets are isolated from each other, how the authorization gate wraps the solver so an unproven request never mutates topology, and how every accepted decision is written into a tamper-evident chain an auditor can replay. It sits directly under the Core Architecture & Compliance Mapping framework, which requires that the same input always produces the same audit hash — a guarantee that only holds if the set of principals allowed to produce that input is itself provable.

The boundary is enforced in four layers that each fail closed: credential verification at the edge, scope-based authorization around the solver call, Hours-of-Service-linked revocation that expires tokens before they can authorize an illegal dispatch, and an append-only, hash-chained audit record that makes any post-hoc edit detectable.

Prerequisites

This page targets Python 3.10+ and models principals and scopes with Pydantic v2 so that a verified identity is an immutable, typed object rather than a loose dict the rest of the pipeline must re-check. JWT verification uses PyJWT with an asymmetric key so that routing workers can verify tokens without holding signing secrets. Pin the following before proceeding:

python -m pip install \
  pyjwt[crypto]==2.8.0 \
  pydantic==2.7.1 \
  cryptography==42.0.7 \
  tenacity==8.3.0

The verified principal is the only credential object the rest of the request handling is allowed to trust. It carries the tenant boundary, the granted scopes, and the Hours-of-Service deadline after which the token must be treated as revoked. Every field has an explicit type and range so an illegal principal cannot be represented:

from __future__ import annotations

from datetime import datetime, timezone
from enum import Enum

from pydantic import BaseModel, ConfigDict, field_validator


class Scope(str, Enum):
    """Closed set of grantable capabilities. A scope not in this enum
    cannot be represented, so a malformed token cannot smuggle one in."""
    ROUTE_DISPATCH = "route:dispatch"
    GRAPH_MODIFY = "graph:modify"
    MANIFEST_EDIT = "manifest:edit"
    TELEMETRY_READ = "telemetry:read"
    COMPLIANCE_AUDIT = "compliance:audit"


class Principal(BaseModel):
    """An identity proven by a verified token. Immutable once constructed."""
    model_config = ConfigDict(strict=True, frozen=True, extra="forbid")

    subject: str            # stable user/service id (JWT `sub`)
    tenant_id: str          # municipal fleet boundary — the isolation key
    scopes: frozenset[Scope]
    hos_deadline: datetime  # token is void once the driver's HOS ceiling is hit
    issued_at: datetime

    @field_validator("hos_deadline", "issued_at")
    @classmethod
    def enforce_utc(cls, v: datetime) -> datetime:
        if v.tzinfo is None or v.utcoffset() != timezone.utc.utcoffset(v):
            raise ValueError("Temporal claims must be timezone-aware (UTC)")
        return v

    def has(self, scope: Scope) -> bool:
        return scope in self.scopes

The granular role-to-scope registry that populates the scopes set for dashboard users — dispatcher, compliance officer, route engineer, municipal auditor — is defined in the RBAC Setup for Waste Ops Dashboards guide; this page treats the Principal as already resolved and focuses on what the boundary does with it.

Core Implementation

Step 1 — Verify the credential and propagate a typed principal

Verification happens once, at the edge, and produces a Principal. Every downstream function receives that object, never the raw token — credential propagation means passing a proven identity, not re-parsing a bearer string in five places where four of them will get the validation subtly wrong. The verifier checks the signature against the identity provider’s public key, enforces issuer and audience, and rejects anything expired. A dedicated exception type lets the API gateway translate every failure into a deterministic 403 without leaking why.

import jwt  # PyJWT
from jwt import InvalidTokenError


class AccessDenied(Exception):
    """Raised on any authorization failure. The gateway maps this to HTTP 403
    with a generic body — the reason is logged, never returned to the client."""
    def __init__(self, reason: str, subject: str = "unknown") -> None:
        self.reason = reason
        self.subject = subject
        super().__init__(reason)


def verify_token(token: str, public_key: str, *, tenant_id: str) -> Principal:
    try:
        claims = jwt.decode(
            token,
            public_key,
            algorithms=["RS256"],          # never accept "none" or HS* from a client
            audience="route-optimizer",
            issuer="waste-ops-idp",
            options={"require": ["exp", "iat", "sub", "aud", "iss"]},
        )
    except InvalidTokenError as exc:
        raise AccessDenied(f"token_invalid: {type(exc).__name__}") from exc

    # Tenant isolation: the token's tenant must match the fleet being addressed.
    if claims.get("tenant") != tenant_id:
        raise AccessDenied("tenant_mismatch", subject=claims.get("sub", "unknown"))

    try:
        return Principal(
            subject=claims["sub"],
            tenant_id=claims["tenant"],
            scopes=frozenset(Scope(s) for s in claims.get("scopes", [])),
            hos_deadline=datetime.fromtimestamp(claims["hos_exp"], tz=timezone.utc),
            issued_at=datetime.fromtimestamp(claims["iat"], tz=timezone.utc),
        )
    except (KeyError, ValueError) as exc:
        raise AccessDenied(f"claims_malformed: {exc}", subject=claims.get("sub", "unknown"))

Two choices matter here. Restricting algorithms to RS256 prevents the classic algorithm-confusion attack where a client re-signs a token with the public key as an HMAC secret. Enforcing tenant_id at verification is the tenant-isolation boundary: a valid token for fleet A can never authorize an action against fleet B, so one municipality’s dispatch data is unreachable from another’s session even if they share the same worker pool.

Step 2 — Wrap the solver in a scope-checked authorization gate

The solver — the OR-Tools Implementation that turns a validated matrix into a dispatch — must never be callable without an explicit capability check. A decorator binds a required Scope to each protected operation and rejects the call before any topology is touched, so authorization is enforced at the boundary of the function rather than sprinkled through its body.

import functools
from collections.abc import Callable
from typing import ParamSpec, TypeVar

P = ParamSpec("P")
R = TypeVar("R")


def requires(scope: Scope) -> Callable[[Callable[P, R]], Callable[P, R]]:
    """Guard a solver-facing operation. The wrapped function must receive the
    verified Principal as its first positional argument."""
    def decorator(fn: Callable[P, R]) -> Callable[P, R]:
        @functools.wraps(fn)
        def wrapper(principal: Principal, *args: P.args, **kwargs: P.kwargs) -> R:
            if not isinstance(principal, Principal):
                raise AccessDenied("no_principal")
            _assert_token_fresh(principal)          # Step 3: HOS-linked revocation
            if not principal.has(scope):
                raise AccessDenied(f"missing_scope:{scope.value}", subject=principal.subject)
            return fn(principal, *args, **kwargs)     # type: ignore[arg-type]
        return wrapper  # type: ignore[return-value]
    return decorator


@requires(Scope.GRAPH_MODIFY)
def apply_route_plan(principal: Principal, plan: dict) -> dict:
    """Mutate live routing topology. Only reachable with graph:modify scope,
    a matching tenant, and a token that has not passed its HOS deadline."""
    # ... solver invocation and topology commit happen here ...
    return {"committed": True, "tenant_id": principal.tenant_id, "stops": len(plan["stops"])}

Because the gate runs _assert_token_fresh and the scope check before the wrapped body, there is no code path in which the solver mutates state for an unauthorized or stale principal. Hardcoded service accounts are deliberately absent — every caller, human or machine, arrives as a Principal through the same gate, which eliminates the privilege-escalation path where a “system” identity bypasses the scope model.

Step 3 — Expire tokens against the Hours-of-Service ceiling

A driver whose Hours-of-Service ceiling has been reached must not be assignable to a new leg, and the cleanest place to enforce that is the credential itself. The token carries an hos_exp claim set to the moment the driver hits the federal on-duty limit; once that passes, the token is void regardless of its own exp. This binds identity freshness to the same duty-time rules the DOT/FMCSA Rule Mapping reference translates into solver constraints, so an expired-duty driver cannot be dispatched even if the routing math would allow it.

def _assert_token_fresh(principal: Principal, *, now: datetime | None = None) -> None:
    now = now or datetime.now(timezone.utc)
    if now >= principal.hos_deadline:
        raise AccessDenied("hos_deadline_passed", subject=principal.subject)

Revocation is therefore automatic and needs no distributed blocklist for the common case: the deadline is inside the signed token, so every worker enforces it independently. A revocation list is still maintained for out-of-band events (a suspended CDL, a compromised credential), but the HOS path — the high-frequency one — stays stateless.

Step 4 — Append every decision to a hash-chained audit log

Every accepted and rejected authorization decision is written to an append-only log whose records are chained by hash, so that removing or editing any past record breaks every hash after it. This is the signed compliance chain an auditor replays to prove that a given route was authorized by a specific principal at a specific time. The JSONFormatter below is the custom formatter used site-wide — the standard logging module has no built-in JSON formatter, so structured audit output requires one.

import hashlib
import json
import logging


class JSONFormatter(logging.Formatter):
    """Custom JSON formatter — the standard logging module has no built-in JSONFormatter."""
    def format(self, record: logging.LogRecord) -> str:
        base = logging.LogRecord("", 0, "", 0, "", (), None).__dict__
        log_obj = {
            "timestamp": self.formatTime(record),
            "level": record.levelname,
            "message": record.getMessage(),
        }
        log_obj.update({k: v for k, v in record.__dict__.items() if k not in base})
        return json.dumps(log_obj, sort_keys=True)


class AuditChain:
    """Append-only, hash-chained authorization ledger. Each record commits to
    the previous record's digest, so a post-hoc edit is detectable."""
    GENESIS = "0" * 64

    def __init__(self, logger: logging.Logger) -> None:
        self._logger = logger
        self._prev = self.GENESIS

    def record(self, *, decision: str, principal: Principal, action: str, detail: str) -> str:
        entry = {
            "prev": self._prev,
            "decision": decision,             # "granted" | "denied"
            "subject": principal.subject,
            "tenant_id": principal.tenant_id,
            "action": action,
            "detail": detail,
        }
        digest = hashlib.sha256(
            json.dumps(entry, sort_keys=True, separators=(",", ":")).encode()
        ).hexdigest()
        self._logger.info("authz_decision", extra={**entry, "digest": digest})
        self._prev = digest
        return digest


_handler = logging.StreamHandler()
_handler.setFormatter(JSONFormatter())
audit_logger = logging.getLogger("access_boundary")
audit_logger.addHandler(_handler)
audit_logger.setLevel(logging.INFO)

Route geometry is intentionally excluded from the audit record — only the action label and the principal are stored, never raw coordinate arrays — so that the compliance log itself does not become a channel for exfiltrating operational topology. Sensitive geometry that must be logged elsewhere is masked by the same discipline applied to raw fixes in the Schema Validation Pipelines reference before it reaches any archival sink.

Regulatory Mapping

Each control maps to a specific federal rule; the citation is what lets an auditor tie an access decision back to the authority that requires it.

• hos_exp token expiry enforces the Hours-of-Service driving and on-duty ceilings under 49 CFR 395 — the token becomes void at the same instant the driver’s duty window closes, so the credential cannot outlive the legal authority to drive. Supporting records must be retained for six months under 49 CFR 395.8(k), which the hash-chained log satisfies.
• Scope grants for CDL- and hazmat-gated actions map to commercial driver licensing standards under 49 CFR 383 and to the hazardous-materials endorsement security threat assessment under 49 CFR 1572 (TSA). A principal lacking the endorsement is issued a token without the corresponding scope, so the gate rejects the dispatch structurally.
• The hash-chained AuditChain provides the tamper-evident record required to support EPA hazardous-waste e-Manifest submissions under 40 CFR 262 Subpart B (EPA Form 8700-22); manifest records must be retained for three years under 40 CFR 262.40. The field-level manifest mappings are specified in the JSON Schema for Municipal Disposal Tracking reference.
• Tenant isolation at token verification implements the access-control separation expected under the NIST SP 800-53 Access Control (AC) family for systems processing multiple municipal jurisdictions, preventing one fleet’s principal from reading or mutating another’s routing state.
• Hazardous-materials routing restrictions governed by 49 CFR 397 are enforced upstream in the route contract; the access boundary’s role is to guarantee that only a principal holding route:dispatch for the correct tenant can commit such a route at all.

Validation & Verification

Confirm the boundary fires correctly with tests that assert both the granted and the denied paths, and check the authz_decision fields the chain emits. A missing scope, a tenant mismatch, and a passed HOS deadline must each raise AccessDenied; a fully-authorized principal must reach the solver.

import pytest
from datetime import timedelta


def _principal(**over) -> Principal:
    now = datetime.now(timezone.utc)
    base = dict(
        subject="drv-118",
        tenant_id="city-sac",
        scopes=frozenset({Scope.GRAPH_MODIFY, Scope.ROUTE_DISPATCH}),
        hos_deadline=now + timedelta(hours=2),
        issued_at=now,
    )
    base.update(over)
    return Principal(**base)


def test_authorized_call_reaches_solver():
    result = apply_route_plan(_principal(), {"stops": [1, 2, 3]})
    assert result["committed"] is True
    assert result["stops"] == 3


def test_missing_scope_is_denied():
    weak = _principal(scopes=frozenset({Scope.TELEMETRY_READ}))
    with pytest.raises(AccessDenied, match="missing_scope:graph:modify"):
        apply_route_plan(weak, {"stops": [1]})


def test_passed_hos_deadline_is_denied():
    stale = _principal(hos_deadline=datetime.now(timezone.utc) - timedelta(minutes=1))
    with pytest.raises(AccessDenied, match="hos_deadline_passed"):
        apply_route_plan(stale, {"stops": [1]})


def test_audit_chain_is_tamper_evident():
    chain = AuditChain(audit_logger)
    d1 = chain.record(decision="granted", principal=_principal(),
                      action="apply_route_plan", detail="stops=3")
    d2 = chain.record(decision="denied", principal=_principal(),
                      action="apply_route_plan", detail="missing_scope")
    assert d1 != d2 and chain._prev == d2   # each record commits to the previous digest

In production, alert on authz_decision events where decision="denied" grouped by subject and action: a burst of denials on one subject is an early signal of a misconfigured role grant or a credential being probed, and grouping by tenant_id surfaces cross-tenant access attempts immediately.

Failure Modes & Edge Cases

The boundary fails closed: when it cannot make a positive authorization decision, it denies. The two operational failure modes that need explicit handling are an identity provider that becomes unreachable and a token-verification path that starts throwing under load.

When the identity provider degrades, the gate must not fall open. It degrades to a read-only posture — telemetry ingestion continues so live sensor data is not lost, but no principal can mutate topology until identities can be verified again. That controlled degradation is the credential-side counterpart to the deterministic route degradation defined in the Fallback Routing Logic reference.

from tenacity import retry, stop_after_attempt, wait_exponential_jitter


class IdentityProviderDown(Exception):
    pass


@retry(
    stop=stop_after_attempt(4),
    wait=wait_exponential_jitter(initial=0.2, max=5.0),
    reraise=True,
)
def fetch_public_key(kid: str, client) -> str:
    """Fetch and cache the IdP signing key. Exponential backoff with jitter
    prevents a thundering herd when the IdP recovers."""
    resp = client.get_jwk(kid)
    if resp is None:
        raise IdentityProviderDown(f"no key for kid={kid}")
    return resp["public_key"]


def authorize_or_degrade(token: str, kid: str, tenant_id: str, client) -> Principal | None:
    """Return a verified Principal, or None to signal read-only degraded mode.
    Never returns a principal that could not be cryptographically verified."""
    try:
        key = fetch_public_key(kid, client)
    except IdentityProviderDown:
        audit_logger.warning("idp_unreachable", extra={"kid": kid, "mode": "read_only"})
        return None                      # fail closed: mutation is refused, not permitted
    return verify_token(token, key, tenant_id=tenant_id)

Two edge cases deserve explicit handling. First, clock skew between the issuer and the workers can make a just-issued token appear not-yet-valid or a just-valid token appear expired; keep worker clocks NTP-synchronized and treat the hos_deadline as authoritative to the second rather than adding a lenient grace window that would let an out-of-hours driver slip through. Second, a token that is cryptographically valid but carries a scope the current Scope enum no longer recognizes must be rejected at construction — the Scope(s) coercion in verify_token raises ValueError, which becomes an AccessDenied, so a retired capability cannot linger in a long-lived token.

Integration Checklist

• PyJWT pinned to 2.8.0; jwt.decode restricts algorithms to RS256 and never accepts none or client-supplied HS*.
• verify_token enforces tenant_id on every request so a token for one fleet cannot act on another.
• Every solver-facing operation is wrapped with @requires(<scope>); no path constructs a route mutation bypassing the gate.
• hos_exp is populated from the driver’s live Hours-of-Service ceiling and enforced by _assert_token_fresh.
• No hardcoded service accounts exist; machine callers arrive as a Principal through the same gate as humans.
• AuditChain writes a hash-chained record for every granted and denied decision; digests are archived to append-only storage.
• Audit records exclude raw coordinate geometry; any geometry logged elsewhere is masked before archival.
• authz_decision denials are alerted on, grouped by subject, action, and tenant_id.
• IdP outage degrades to read-only (mutation refused), verified by a test that asserts authorize_or_degrade returns None when the key fetch fails.
• Worker clocks are NTP-synchronized; retention meets 49 CFR 395.8(k) (6 months) and 40 CFR 262.40 (3 years).

FAQ

Why bind token expiry to Hours of Service instead of a fixed TTL?

A fixed TTL either expires too early (forcing needless re-auth mid-shift) or too late (leaving a valid credential after a driver’s duty window has legally closed). Embedding the driver’s HOS ceiling as an hos_exp claim makes the credential void at exactly the moment the authority to drive ends, enforced independently by every worker without a shared blocklist. The duty-time rules themselves come from the DOT/FMCSA Rule Mapping reference.

Should authorization be checked inside the solver or around it?

Around it. The @requires decorator runs the freshness and scope checks before the wrapped body executes, so there is no code path in which the OR-Tools Implementation mutates topology for an unauthorized principal. Checks scattered inside solver logic are easy to miss on one branch and impossible to audit as a single boundary.

What stops one municipality's session from reaching another's routes?

Tenant isolation is enforced at verification: verify_token rejects any token whose tenant claim does not match the fleet being addressed, so a valid credential for fleet A raises AccessDenied the moment it targets fleet B — even when both fleets are served by the same worker pool. The tenant_id is also written into every audit record, so cross-tenant attempts are visible in the log.

How does the hash-chained audit log detect tampering?

Each record includes the SHA-256 digest of the previous record, and its own digest is computed over that. Editing or deleting any past record changes its digest, which breaks the prev reference of every record after it. An auditor recomputes the chain from genesis; the first mismatch pinpoints exactly where the ledger was altered, which is what makes the record admissible as a compliance trail.

What happens to dispatch when the identity provider goes down?

The boundary fails closed. authorize_or_degrade returns None rather than a principal, which puts the platform in read-only mode: telemetry keeps flowing but no topology mutation is permitted until identities can be verified again. This is the credential-side mirror of the deterministic degradation in the Fallback Routing Logic layer — never fail open.

Up: Core Architecture & Compliance Mapping

Related

• RBAC Setup for Waste Ops Dashboards — role-to-scope registry and least-privilege enforcement for operator UIs.
• Route Schema Design — the immutable data contract every authorized route must satisfy.
• DOT/FMCSA Rule Mapping — translating federal HOS and licensing mandates into solver and credential rules.
• Fallback Routing Logic — deterministic degradation when a route or identity dependency fails.
• Schema Validation Pipelines — masking and validating raw telemetry before it is archived or promoted.