Encrypting DVIR Data at Rest and in Transit

A DVIR carries personally identifiable information a carrier is obligated to protect: the driver’s license number, the driver’s signature, and an identity linkage to hours-of-service records. If that data is exposed — read off an unencrypted disk, sniffed off an internal hop, or exfiltrated from a backup — the carrier faces breach-notification and privacy liability entirely separate from any FMCSA finding. This guide implements the encryption boundary for a DVIR pipeline: AES-256 at rest via envelope encryption with KMS-managed data keys, TLS 1.2 or higher for every hop in transit, field-level encryption of the specific PII fields, and key rotation that keeps old records readable. It is one of the controls behind Compliance Boundary Enforcement in Cloud Workflows, part of the Core DVIR Architecture & FMCSA Compliance Mapping pipeline.

Encryption interacts with the audit trail in a way that is easy to get wrong. The immutable record sealed in WORM Retention and SHA-256 Hashing for DVIR Records must remain verifiable across a key rotation, which forces a deliberate decision about whether you hash the plaintext or the ciphertext — the wrong choice makes every historical audit_hash unverifiable the day you rotate a key.

• Python 3.10+.
• cryptography 42+ — AES-GCM authenticated encryption (AESGCM) for field-level encryption.
• boto3 1.34+ with AWS KMS — generate_data_key and decrypt for envelope encryption; a customer-managed KMS key (CMK) with rotation enabled.
• TLS 1.2+ terminated at every service boundary — enforced in the load balancer and the application, not assumed.
• The sealed-record contract and the audit_hash (SHA-256 over canonical JSON) convention from the Immutable Audit Trail and WORM Retention section.

Do not send DVIR fields to KMS to encrypt directly — KMS has a 4 KB limit and per-call cost. Instead ask KMS for a data key: it returns a plaintext AES-256 key and a KMS-wrapped copy. Encrypt the fields locally with the plaintext key, store the wrapped copy next to the ciphertext, and discard the plaintext key from memory immediately.

import boto3

kms = boto3.client("kms")


def new_data_key(cmk_id: str) -> tuple[bytes, bytes]:
    """Return (plaintext_key, wrapped_key). AES_256 => 32-byte key.
    Wipe plaintext_key as soon as encryption is done."""
    resp = kms.generate_data_key(KeyId=cmk_id, KeySpec="AES_256")
    return resp["Plaintext"], resp["CiphertextBlob"]

Use AES-GCM — it authenticates as well as encrypts, so a tampered ciphertext fails to decrypt rather than returning garbage. Generate a fresh 96-bit nonce per field per encryption; never reuse a nonce with the same key. Bind the ciphertext to the record with additional authenticated data (AAD) set to the record_id, so a ciphertext cannot be lifted from one record into another.

import os
from cryptography.hazmat.primitives.ciphers.aead import AESGCM

PII_FIELDS = ("driver_license_number", "driver_signature")


def encrypt_fields(record: dict, plaintext_key: bytes, record_id: str) -> dict:
    aes = AESGCM(plaintext_key)
    aad = record_id.encode("utf-8")
    out = dict(record)
    for field in PII_FIELDS:
        value = record[field].encode("utf-8")
        nonce = os.urandom(12)            # 96-bit, unique per field per encryption
        ct = aes.encrypt(nonce, value, aad)
        out[field] = {"nonce": nonce.hex(), "ct": ct.hex()}
    return out

Persist the wrapped data key and a key_version with the record so a reader knows which CMK generation to unwrap with. The plaintext key never touches disk. Overwrite it in memory as soon as the fields are encrypted.

def seal_encrypted(record: dict, cmk_id: str, record_id: str) -> dict:
    plaintext_key, wrapped_key = new_data_key(cmk_id)
    try:
        encrypted = encrypt_fields(record, plaintext_key, record_id)
    finally:
        # best-effort scrub of the plaintext key from memory
        plaintext_key = bytes(len(plaintext_key))
    encrypted["_wrapped_key"] = wrapped_key.hex()
    encrypted["_key_version"] = cmk_id  # CMK ARN pins the generation used
    return encrypted

To read, unwrap the stored data key with KMS and decrypt each field. Because the wrapped key is stored with the record, a record written under an older CMK rotation decrypts correctly after rotation — KMS retains prior key material and selects it by the wrapped blob. This is what makes rotation safe.

def decrypt_fields(record: dict, record_id: str) -> dict:
    wrapped = bytes.fromhex(record["_wrapped_key"])
    plaintext_key = kms.decrypt(CiphertextBlob=wrapped)["Plaintext"]
    try:
        aes = AESGCM(plaintext_key)
        aad = record_id.encode("utf-8")
        out = dict(record)
        for field in PII_FIELDS:
            blob = record[field]
            pt = aes.decrypt(bytes.fromhex(blob["nonce"]),
                             bytes.fromhex(blob["ct"]), aad)
            out[field] = pt.decode("utf-8")
        return out
    finally:
        plaintext_key = bytes(len(plaintext_key))

The audit trail hashes the canonical plaintext record, not the ciphertext. Compute the audit_hash over the plaintext canonical JSON before field encryption, then encrypt for storage. If you hashed the ciphertext, a key rotation or re-encryption would change the ciphertext bytes and invalidate every historical digest — the record would appear tampered when it was not. Hash the meaning, encrypt the storage.

Prove three properties: ciphertext differs from plaintext and from run to run (nonce freshness), a round-trip recovers the original, and a record encrypted under an old key still decrypts after the CMK rotates.

import pytest


def test_ciphertext_differs_and_round_trips(cmk_id):
    rec = {"driver_license_number": "S1234567",
           "driver_signature": "a3f...", "record_id": "r1"}
    sealed = seal_encrypted(rec, cmk_id, "r1")
    assert sealed["driver_license_number"]["ct"] != rec["driver_license_number"].encode().hex()
    # a second encryption of the same value yields different ciphertext (fresh nonce)
    sealed2 = seal_encrypted(rec, cmk_id, "r1")
    assert sealed["driver_license_number"]["ct"] != sealed2["driver_license_number"]["ct"]
    assert decrypt_fields(sealed, "r1")["driver_license_number"] == "S1234567"


def test_rotated_key_still_decrypts_old_data(cmk_id):
    sealed = seal_encrypted(SAMPLE, cmk_id, "r2")
    kms.rotate_key_on_demand(KeyId=cmk_id)   # new key generation
    # record written before rotation must still read back correctly
    assert decrypt_fields(sealed, "r2") == SAMPLE_PLAINTEXT


def test_aad_binding_rejects_cross_record_ciphertext(cmk_id):
    sealed = seal_encrypted(SAMPLE, cmk_id, "r3")
    with pytest.raises(Exception):
        decrypt_fields(sealed, "r_other")   # wrong record_id as AAD => auth failure

• Nonce reuse. Reusing a nonce with the same AES-GCM key is catastrophic — it can leak the XOR of two plaintexts and break authentication. Always draw a fresh 96-bit nonce from os.urandom per field per encryption, and never derive it deterministically from the record. If you rotate the data key per record (as above), nonce collisions across records are already impossible.
• Encrypting before hashing vs after. Hash the canonical plaintext, then encrypt for storage. Hashing the ciphertext ties the audit_hash to a specific encryption, so any rotation or re-encryption invalidates the historical digest and makes an intact record look altered. Keep the digest a property of the record’s meaning.
• Key access in the audit path. An auditor’s read-only role must be able to decrypt for verification, but decryption is a KMS action that has to be granted explicitly — a role with s3:GetObject but no kms:Decrypt retrieves ciphertext it cannot read. Grant kms:Decrypt on the specific CMK to the audit role, and log every decrypt through CloudTrail so PII access is itself auditable.
• TLS downgrade on internal hops. Encrypting the edge and trusting the internal network is a common gap. Enforce TLS 1.2+ on every service-to-service call, reject plaintext HTTP at the application, and pin certificates for internal calls so a compromised hop cannot present a substitute cert. Data in transit between your own services is still data in transit.

• Compliance Boundary Enforcement in Cloud Workflows — the boundary these encryption controls help enforce.
• Implementing Role-Based Access for DVIR Data — who is granted the KMS decrypt permission in the audit path.
• WORM Retention and SHA-256 Hashing for DVIR Records — the audit hash that must survive a key rotation.
• Immutable Audit Trail and WORM Retention — the retention layer that stores the encrypted records.

Back to Compliance Boundary Enforcement in Cloud Workflows, part of Core DVIR Architecture & FMCSA Compliance Mapping.