Implementing Role-Based Access for DVIR Data

Driver Vehicle Inspection Reports carry legally binding driver attestations, defect classifications, and repair certifications that DOT auditors treat as primary evidence. This page answers one focused question: how do you enforce role-scoped access to DVIR records in a cloud pipeline so that no principal can read, mutate, or certify a report outside their authority? Get this wrong and the consequence is concrete — a mechanic overwriting a driver’s original defect notes, or a dispatcher clearing an out-of-service (OOS) condition, breaks the chain of custody required under 49 CFR § 396.11(a) and § 396.13, and a single unauthorized state transition can invalidate the record during a compliance review. Role-based access control (RBAC) here is not a convenience layer; it is the enforcement mechanism that keeps every inspection record attributable and defensible.

This is the access-control implementation under the broader Cloud Workflow Compliance Boundaries for DVIR framework — specifically the third boundary, data segregation and access control. Read that page first for how ingestion validation and deterministic compliance evaluation gate the record before it reaches the authorization layer described here.

Before implementing the authorization layer, you need the following in place:

• Python 3.10+ — the code below uses PEP 604 union syntax (str | None) and dict[...] generics.
• fastapi and a JWT verification dependency (pyjwt or python-jose) that populates request.state.claims with a verified role claim upstream of this middleware. Never trust an unverified claim.
• The canonical DVIR record contract — this page assumes the field names (dvir_id, status, driver_id, defect_codes, repair_signature) defined in the Standardized DVIR JSON Schema Design. Keep the schema as the single source of truth so IAM condition keys and application guards reference identical attribute names.
• The regulatory state model from the FMCSA DVIR Rule 396.11 Breakdown, which fixes who may sign off a defect and when a vehicle may return to service.
• An IAM-capable datastore. The examples target DynamoDB and AWS IAM, but the two-layer pattern (storage-layer policy + application-layer state guard) ports to any provider.

The DVIR lifecycle progresses through five discrete states, and each transition is owned by exactly one role:

The compliance_officer role is read-only across every state and never appears as a transition owner.

Enforce access at two layers. The storage-layer IAM policy is your coarse, provider-enforced boundary; the application-layer guard enforces the fine-grained state-transition invariants that IAM condition keys cannot express.

Translate the state model into an IAM policy that constrains both which records a principal can touch and which attributes it can project. Do not grant blanket dynamodb:Query or s3:GetObject. In AWS IAM, StringEquals and ForAllValues:StringEquals are separate condition blocks under Condition — nesting one inside the other is invalid and silently ignored by the evaluator:

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "DVIRRoleScopedAccess",
      "Effect": "Allow",
      "Action": ["dynamodb:GetItem", "dynamodb:Query"],
      "Resource": "arn:aws:dynamodb:us-east-1:123456789012:table/dvir-reports",
      "Condition": {
        "StringEquals": {
          "dynamodb:LeadingKeys": ["${aws:PrincipalTag/Role}"]
        },
        "ForAllValues:StringEquals": {
          "dynamodb:Attributes": [
            "dvir_id", "status", "driver_id", "defect_codes", "repair_signature"
          ]
        }
      }
    }
  ]
}

StringEquals on dynamodb:LeadingKeys restricts a principal to records whose partition key matches the role tag on their session, so a mechanic cannot query another role’s partition. ForAllValues:StringEquals on dynamodb:Attributes caps the projected attribute set and stops over-fetching. Issue the compliance_officer role a separate statement that omits the LeadingKeys condition to grant read-only visibility across all partitions, but never add a write action to it. For the exact evaluation order and how deny/allow precedence resolves, consult the AWS IAM policy evaluation logic reference.

IAM can say “this principal may write to this table”; it cannot say “a mechanic may move a record from pending_repair to certified_repair but nowhere else.” Encode that invariant as data, mapping the current state to the set of states each role may move it into:

# Role-to-state transition matrix aligned with 49 CFR § 396.11(a) (driver
# report + certification) and § 396.13(b) (mechanic sign-off before return
# to service). Keys are current states; values are the states this role may
# transition to. An empty inner dict means the role may hold no write authority.
DVIR_ACCESS_MATRIX: dict[str, dict[str, list[str]]] = {
    "driver":             {"draft": ["pending_review"]},
    "fleet_manager":      {"pending_review": ["pending_repair", "archived"]},
    "mechanic":           {"pending_repair": ["certified_repair"]},
    "compliance_officer": {},  # Read-only; no write transitions permitted.
}

Intercept every mutating request before it reaches the data layer. Validate the verified role claim, look up the requested transition in the matrix, and reject the request with a deterministic status code when the transition is not authorized:

import functools
from typing import Callable

from fastapi import Request, HTTPException, status


def enforce_dvir_rbac(current_state: str, target_state: str) -> Callable:
    """
    ASGI decorator that validates the caller's verified role claim against
    DVIR_ACCESS_MATRIX before allowing a DVIR state transition. Reject, never
    hedge: an unauthorized transition returns 403 and writes nothing.
    """
    def decorator(func: Callable) -> Callable:
        @functools.wraps(func)
        async def wrapper(request: Request, *args, **kwargs):
            caller_role: str | None = request.state.claims.get("role")
            dvir_id: str | None = request.path_params.get("dvir_id")

            if not caller_role or caller_role not in DVIR_ACCESS_MATRIX:
                raise HTTPException(
                    status_code=status.HTTP_401_UNAUTHORIZED,
                    detail="Invalid or missing role claim",
                )

            role_permissions = DVIR_ACCESS_MATRIX[caller_role]
            allowed_targets = role_permissions.get(current_state, [])

            if target_state not in allowed_targets:
                raise HTTPException(
                    status_code=status.HTTP_403_FORBIDDEN,
                    detail=(
                        f"Role '{caller_role}' cannot transition DVIR "
                        f"{dvir_id} from '{current_state}' to '{target_state}'"
                    ),
                )

            return await func(request, *args, **kwargs)
        return wrapper
    return decorator

This guard decouples authorization from business routing and closes the gap that IAM alone leaves open: concurrent API calls that satisfy a coarse storage policy but violate the lifecycle order.

Once the guard passes, persist the new state with a conditional write so a concurrent caller cannot double-apply a transition, and emit a structured audit record for every decision — allow or deny:

import json
import logging

import boto3
from botocore.exceptions import ClientError

_table = boto3.resource("dynamodb").Table("dvir-reports")
_audit = logging.getLogger("dvir.audit")


def commit_transition(
    dvir_id: str, current_state: str, target_state: str,
    principal_arn: str, correlation_id: str,
) -> None:
    """Atomically advance the DVIR state; log the policy decision either way."""
    record = {
        "correlation_id": correlation_id,
        "principal_arn": principal_arn,
        "dvir_id": dvir_id,
        "previous_state": current_state,
        "target_state": target_state,
    }
    try:
        # Optimistic concurrency: only advance if the record is still in the
        # expected state, so a duplicate certification cannot be applied twice.
        _table.update_item(
            Key={"dvir_id": dvir_id},
            UpdateExpression="SET #s = :target",
            ConditionExpression="#s = :current",
            ExpressionAttributeNames={"#s": "status"},
            ExpressionAttributeValues={":target": target_state, ":current": current_state},
        )
        _audit.info(json.dumps({**record, "policy_decision": "ALLOW"}))
    except ClientError as exc:
        if exc.response["Error"]["Code"] == "ConditionalCheckFailedException":
            # State moved underneath us — reject rather than clobber.
            _audit.warning(json.dumps({**record, "policy_decision": "DENY_STALE_STATE"}))
            raise
        raise

Route every one of these structured payloads — correlation_id, principal_arn, dvir_id, previous_state, target_state, policy_decision — to a centralized SIEM. This is the evidence trail 49 CFR § 396.3© recordkeeping expects, and it feeds anomaly detection for repeated denials. Write the audit stream to WORM-compliant storage (S3 Object Lock in compliance mode) so a legal hold cannot be tampered with, and match retention to the minimums in 49 CFR § 396.11©(2) — three months for original reports, longer for records tied to an unsatisfied defect.

Prove the boundary holds before you ship it. Parametrize a test across every role and every state pair; the matrix is small enough to assert exhaustively.

import pytest


@pytest.mark.parametrize("role,current,target,allowed", [
    ("driver",             "draft",           "pending_review",   True),
    ("driver",             "pending_review",  "pending_repair",   False),  # driver cannot triage
    ("fleet_manager",      "pending_review",  "pending_repair",   True),
    ("mechanic",           "pending_repair",  "certified_repair", True),
    ("mechanic",           "draft",           "pending_review",   False),  # wrong state for role
    ("compliance_officer", "pending_review",  "pending_repair",   False),  # read-only, never writes
])
def test_transition_authorization(role, current, target, allowed):
    permitted = target in DVIR_ACCESS_MATRIX.get(role, {}).get(current, [])
    assert permitted is allowed

Confirm three properties explicitly:

1. Read-only means read-only. Assert that compliance_officer returns False for every non-empty transition — a single passing write case here is a compliance defect.
2. Idempotency holds under contention. Fire two concurrent certified_repair commits against the same dvir_id; exactly one must succeed and the other must raise ConditionalCheckFailedException. The audit log must show one ALLOW and one DENY_STALE_STATE.
3. Attribute projection is enforced. Use the IAM policy simulator against the Step 1 policy to verify a mechanic principal cannot project driver_id outside the allowed attribute set.

• Missing or drifted resource tags. IAM condition keys resolve against tags. If a DVIR record is created without its Role, FleetID, and ComplianceTier tags, dynamodb:LeadingKeys evaluates against an empty value and the policy either denies everything or, worse, matches too broadly. Propagate tags at record creation and add an AWS Config rule that flags untagged records.
• Nested condition operators that silently no-op. Placing ForAllValues:StringEquals inside StringEquals produces a policy that parses cleanly but ignores the inner constraint entirely, quietly granting full attribute access. Keep each operator as its own top-level block under Condition and verify with the policy simulator, not by inspection.
• Offline driver sync replaying stale transitions. A mobile client that queued a draft → pending_review submission offline may replay it after the fleet manager has already advanced the record. The Step 4 conditional write turns this into a clean DENY_STALE_STATE rather than a silent overwrite — never resolve these by “last write wins.” Reconcile offline timestamps against the recorded previous_state, not the wall clock.
• Trusting the JWT role claim without verification. The middleware reads request.state.claims, which is only safe if an upstream dependency has cryptographically verified the token’s signature and expiry. An unverified claim lets a caller assert any role. Verify before this guard runs, and reject unsigned or expired tokens outright.

Records that clear this authorization boundary but fail a downstream severity check are routed onward by the deterministic engine described in Critical vs Non-Critical Routing Logic; high-volume replays of offline submissions are absorbed by Async Batching for High-Volume Ingestion before they reach this layer.

• Cloud Workflow Compliance Boundaries for DVIR — the parent framework this access layer sits inside.
• Standardized DVIR JSON Schema Design — the field contract IAM and application guards both reference.
• JSON Schema Validation for Electronic DVIRs — the ingestion validation that runs before authorization.
• FMCSA DVIR Rule 396.11 Breakdown — the regulatory basis for the state model and sign-off rules.
• Critical vs Non-Critical Routing Logic — where authorized records go once their defects are classified.

This page is part of the Cloud Workflow Compliance Boundaries for DVIR topic. Back to the Core DVIR Architecture & FMCSA Compliance Mapping framework.