Encrypting PII Fields at Rest with Envelope Encryption

This page is a focused implementation guide under Data Boundary Enforcement: it shows how to encrypt claimant PII at the field level using envelope encryption — a per-record data key wrapped by a key-encryption key — so a database compromise yields ciphertext, and key rotation never means re-encrypting the whole table.

A claims database holds exactly the fields a breach notification statute cares about: claimant name, Social Security number, date of birth, bank account for settlement disbursement, and often protected health information from injury claims. Encrypting the whole disk protects nothing once the application is authenticated — the process reads plaintext, so does anyone who compromises it. Encrypting every field under a single application-wide key is barely better: that one key is now the entire blast radius, it lives in memory next to the data it protects, and rotating it means decrypting and re-encrypting every row in the table under a lock.

Envelope encryption resolves this. Each record gets its own data key used once to encrypt that record’s PII with AES-GCM; the data key itself is then encrypted (“wrapped”) by a long-lived key-encryption key (the KEK) that never leaves a key-management boundary. What you store next to the ciphertext is the wrapped data key, not the key that could read it. Rotating the KEK re-wraps data keys — a cheap operation on small blobs — without touching the encrypted payloads, so a rotation that would otherwise be a table-wide rewrite becomes a metadata update. This guide implements field-level envelope encryption with the cryptography library’s AES-GCM, abstracts the KEK behind a KMS-style client interface (so the same code runs against a cloud KMS or a local test double — never against real cloud keys in this example), handles rotation by KEK version, and addresses the deterministic-encryption tradeoff needed to search on an encrypted field. It is the enforcement mechanism behind the boundaries described in Data Boundary Enforcement, and every wrap, unwrap, and rotation emits an event for the Audit Log Schema Design trail.

This pattern targets Python 3.10+ and the cryptography library for AES-GCM, plus structlog for the audit lines. Pin both so the AEAD construction and its exception types are stable:

Copy
python -m venv .venv && source .venv/bin/activate
pip install "cryptography==42.*" "structlog==24.*"

The KEK lives behind a key-management service in production — AWS KMS, GCP KMS, HashiCorp Vault Transit, or an HSM — reached through a narrow wrap/unwrap interface. This guide defines that interface and a local test double so the code is runnable and testable without any cloud credentials. Do not point the example at a real cloud KEK; wire the production client behind the same interface once the pattern is verified. All PII values are handled as bytes at the crypto boundary and never logged.

The KEK is reached only through a wrap/unwrap contract. Depending on a Protocol rather than a concrete SDK keeps the encryption code testable and cloud-agnostic — production swaps in a real client, tests use the in-memory double below.

Copy
from __future__ import annotations

from dataclasses import dataclass
from typing import Protocol

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

log = structlog.get_logger("pii.envelope")


class KmsClient(Protocol):
    """Narrow key-management contract. Production binds this to a cloud KMS."""

    def wrap(self, data_key: bytes) -> tuple[bytes, str]:
        """Encrypt a data key under the current KEK; return (blob, kek_version)."""

    def unwrap(self, wrapped: bytes, kek_version: str) -> bytes:
        """Decrypt a wrapped data key using the named KEK version."""

    def current_version(self) -> str: ...


class LocalKmsDouble:
    """TEST ONLY. Simulates a KEK per version with AES-GCM. Never use real keys here."""

    def __init__(self) -> None:
        # Map version -> KEK. A real KMS keeps these inside its boundary.
        self._keks: dict[str, bytes] = {"v1": AESGCM.generate_key(bit_length=256)}
        self._current = "v1"

    def rotate(self) -> str:
        n = f"v{len(self._keks) + 1}"
        self._keks[n] = AESGCM.generate_key(bit_length=256)
        self._current = n
        return n

    def current_version(self) -> str:
        return self._current

    def wrap(self, data_key: bytes) -> tuple[bytes, str]:
        nonce = _nonce()
        blob = AESGCM(self._keks[self._current]).encrypt(nonce, data_key, b"kek-wrap")
        return nonce + blob, self._current

    def unwrap(self, wrapped: bytes, kek_version: str) -> bytes:
        nonce, blob = wrapped[:12], wrapped[12:]
        return AESGCM(self._keks[kek_version]).decrypt(nonce, blob, b"kek-wrap")

Generate a new 256-bit data key per record, encrypt the PII with AES-GCM (which authenticates as well as encrypts), then hand the data key to the KMS to be wrapped. Bind the record id as AEAD associated data so a ciphertext cannot be silently transplanted onto a different record.

Copy
import os


def _nonce() -> bytes:
    return os.urandom(12)   # 96-bit nonce, the AES-GCM standard


@dataclass(frozen=True, slots=True)
class EncryptedField:
    ciphertext: bytes       # includes the GCM tag
    nonce: bytes
    wrapped_data_key: bytes
    kek_version: str


def encrypt_field(plaintext: bytes, *, record_id: str, kms: KmsClient) -> EncryptedField:
    if not isinstance(plaintext, bytes):
        raise TypeError("plaintext must be bytes; encode at the boundary")
    data_key = AESGCM.generate_key(bit_length=256)
    nonce = _nonce()
    aad = record_id.encode("utf-8")
    ciphertext = AESGCM(data_key).encrypt(nonce, plaintext, aad)
    wrapped, version = kms.wrap(data_key)
    # data_key goes out of scope here; only the wrapped form is retained.
    log.info("pii.encrypted", record_id=record_id, kek_version=version,
             bytes=len(ciphertext))
    return EncryptedField(ciphertext, nonce, wrapped, version)

Decryption reverses the envelope: unwrap the data key via the KMS using the stored KEK version, then let AES-GCM verify the authentication tag before returning plaintext. A tampered ciphertext or a mismatched record id fails the tag check and raises InvalidTag — decryption never returns unauthenticated bytes.

Copy
from cryptography.exceptions import InvalidTag


def decrypt_field(field: EncryptedField, *, record_id: str, kms: KmsClient) -> bytes:
    data_key = kms.unwrap(field.wrapped_data_key, field.kek_version)
    aad = record_id.encode("utf-8")
    try:
        plaintext = AESGCM(data_key).decrypt(field.nonce, field.ciphertext, aad)
    except InvalidTag as exc:
        log.warning("pii.tamper_detected", record_id=record_id,
                    kek_version=field.kek_version)
        raise PiiIntegrityError(record_id) from exc
    log.info("pii.decrypted", record_id=record_id, kek_version=field.kek_version)
    return plaintext


class PiiIntegrityError(Exception):
    """AES-GCM tag verification failed: ciphertext, nonce, AAD or key mismatch."""

    def __init__(self, record_id: str) -> None:
        super().__init__(f"integrity check failed for record {record_id}")
        self.record_id = record_id

KEK rotation only re-wraps data keys: unwrap under the old version, wrap under the new, leave the ciphertext untouched. Separately, because GCM ciphertext is intentionally non-deterministic you cannot equality-search it; derive a deterministic, keyed HMAC token for the narrow set of fields you must look up, accepting that determinism leaks equality.

Copy
import hashlib
import hmac


def rewrap_to_current(field: EncryptedField, *, kms: KmsClient) -> EncryptedField:
    if field.kek_version == kms.current_version():
        return field
    data_key = kms.unwrap(field.wrapped_data_key, field.kek_version)
    wrapped, new_version = kms.wrap(data_key)
    log.info("pii.rewrapped", from_version=field.kek_version, to_version=new_version)
    from dataclasses import replace
    return replace(field, wrapped_data_key=wrapped, kek_version=new_version)


def search_token(plaintext: bytes, *, index_key: bytes) -> str:
    """Deterministic keyed token for exact-match lookup on an encrypted field.

    Tradeoff: identical plaintext yields identical tokens, so this leaks equality.
    Use only for fields that must be searched (e.g. SSN dedupe), never as the payload.
    """
    return hmac.new(index_key, plaintext, hashlib.sha256).hexdigest()

Prove the round trip, the rotation invariance, and — most importantly — that tampering is rejected rather than silently decrypted. The tamper test is the one that demonstrates the AEAD guarantee is real.

Copy
def test_round_trip_recovers_plaintext() -> None:
    kms = LocalKmsDouble()
    field = encrypt_field(b"123-45-6789", record_id="clm-001", kms=kms)
    assert decrypt_field(field, record_id="clm-001", kms=kms) == b"123-45-6789"


def test_ciphertext_is_not_plaintext() -> None:
    kms = LocalKmsDouble()
    field = encrypt_field(b"123-45-6789", record_id="clm-001", kms=kms)
    assert b"123-45-6789" not in field.ciphertext


def test_tampered_ciphertext_raises() -> None:
    kms = LocalKmsDouble()
    field = encrypt_field(b"claimant-dob-1980-02-11", record_id="clm-002", kms=kms)
    flipped = bytearray(field.ciphertext)
    flipped[0] ^= 0x01                       # flip one bit
    from dataclasses import replace
    tampered = replace(field, ciphertext=bytes(flipped))
    import pytest
    with pytest.raises(PiiIntegrityError):
        decrypt_field(tampered, record_id="clm-002", kms=kms)


def test_wrong_record_id_fails_aad() -> None:
    kms = LocalKmsDouble()
    field = encrypt_field(b"acct-99887766", record_id="clm-003", kms=kms)
    import pytest
    with pytest.raises(PiiIntegrityError):
        decrypt_field(field, record_id="clm-OTHER", kms=kms)   # AAD mismatch


def test_rotation_preserves_plaintext_without_touching_payload() -> None:
    kms = LocalKmsDouble()
    field = encrypt_field(b"123-45-6789", record_id="clm-004", kms=kms)
    original_ct = field.ciphertext
    kms.rotate()
    rewrapped = rewrap_to_current(field, kms=kms)
    assert rewrapped.kek_version == kms.current_version()
    assert rewrapped.ciphertext == original_ct          # payload untouched
    assert decrypt_field(rewrapped, record_id="clm-004", kms=kms) == b"123-45-6789"

Run under python -m pytest -q. The tamper and AAD tests are the security-relevant assertions: they confirm that any modification to the ciphertext, or any attempt to read a record’s field under a different record id, is rejected at the tag check instead of returning corrupted plaintext.

Field-level envelope encryption is what lets a claims platform answer a breach-notification question honestly: if the datastore is exfiltrated, the attacker holds ciphertext and wrapped data keys, and the KEK needed to unwrap them never left the key-management boundary — so the exposure of regulated PII and PHI is materially different from a plaintext leak under state privacy statutes (CCPA/CPRA, the NYDFS cybersecurity regulation, and HIPAA where injury-claim health data is involved). Make every wrap, unwrap, rotation, and PiiIntegrityError an event written to the Audit Log Schema Design trail, recording the record id, the KEK version, and the operation — never the plaintext or the data key. Keying decryption through the KMS also means access to PII is access to a KEK, so revoking a KEK grant instantly and verifiably cuts off decryption across the whole table, which is the enforcement point the boundaries in Data Boundary Enforcement are meant to guarantee.

• Decryption fails after a deploy with InvalidTag. Symptom: previously readable records raise PiiIntegrityError. Cause: the AAD changed — usually the record id was reformatted — so the tag no longer verifies. Fix: treat the AAD as part of the ciphertext contract; migrate deliberately by decrypting under the old AAD and re-encrypting, never change it in place.
• Rotation appears to do nothing. Symptom: kek_version never advances. Cause: rewrap_to_current short-circuits when the field is already current, but the KEK was never actually rotated in the KMS. Fix: rotate the KEK in the key-management service first, then sweep records with rewrap_to_current; the sweep is safe to run repeatedly.
• Cannot search an encrypted SSN. Symptom: exact-match lookups return nothing. Cause: GCM ciphertext is non-deterministic by design, so equality never matches. Fix: index the narrow searchable field with the keyed search_token, accepting that deterministic tokens leak equality; keep the token separate from the authenticated payload and rotate the index key on its own schedule.
• Data key or plaintext shows up in logs. Symptom: a secret is visible in log aggregation. Cause: a debug line serialized an EncryptedField or a caught exception carried plaintext. Fix: log only record id, KEK version, and byte lengths as shown; never place PII or key material in a log call or an exception message.
• Nonce reuse warning under high volume. Symptom: a security scanner flags GCM nonce collision risk. Cause: nonces were derived from a counter that reset, not from a CSPRNG. Fix: generate every nonce with os.urandom(12) and never reuse a nonce with the same data key; because each record gets a fresh data key, per-record random nonces are safe.

• Data Boundary Enforcement — the parent guide whose boundaries this encryption enforces
• Audit Log Schema Design — where every wrap, unwrap, and rotation event is recorded
• Building Append-Only Hash-Chained Audit Logs — the tamper-evident trail these crypto events flow into
• How to Map ISO Policy Forms to JSON Schemas — the schema layer that marks which fields carry PII and require this treatment
• Core Architecture & Compliance Mapping — the control plane this data-protection stage belongs to