Envelope Encryption and Key Rotation Without Re-encrypting Data

Lessons 2 to 7 all assumed the key was simply there. In production, the hard part is the key itself: where it lives, who can use it, and how to replace it without decrypting and re-encrypting terabytes of data. This lesson builds the pattern that cloud KMS products use, envelope encryption, on your own machine. Start a container with docker run --rm -it -v "$PWD":/work -w /work python:3.12-slim bash, then run pip install cryptography==50.0.2. The examples were run with Python 3.12.15 and cryptography 50.0.2. Every key is generated fresh on each run, and none of them is real.

Two keys: one for the data, one for the keys

A data-encryption key (DEK) encrypts the record, and a key-encryption key (KEK) encrypts only the DEK. The wrapped DEK is stored next to the data, and the KEK is kept somewhere else.

# envelope.py
import os
from cryptography.hazmat.primitives.ciphers.aead import AESGCM

kek = AESGCM.generate_key(bit_length=256)    # key-encryption key: lives in a KMS/HSM
record = b"name=Ada Lovelace; card=4111 1111 1111 1111;" * 1000

dek = AESGCM.generate_key(bit_length=256)    # data-encryption key: one per record
data_nonce = os.urandom(12)
ciphertext = AESGCM(dek).encrypt(data_nonce, record, None)

wrap_nonce = os.urandom(12)
wrapped_dek = AESGCM(kek).encrypt(wrap_nonce, dek, b"kek-v1")
del dek                                       # store only the wrapped copy

print("record bytes:     ", len(record))
print("ciphertext bytes: ", len(ciphertext))
print("wrapped DEK bytes:", len(wrapped_dek))

# to read: unwrap the DEK with the KEK, then decrypt the data
dek = AESGCM(kek).decrypt(wrap_nonce, wrapped_dek, b"kek-v1")
plain = AESGCM(dek).decrypt(data_nonce, ciphertext, None)
print("round trip ok:    ", plain == record)

$ python envelope.py
record bytes:      44000
ciphertext bytes:  44016
wrapped DEK bytes: 48
round trip ok:     True

The data grows by 16 bytes, the GCM tag from lesson 2. The wrapped DEK is 48 bytes: 32 bytes of key plus its own 16-byte tag. That size doesn’t change whether the record is 44 KB or 44 GB, and it’s why this pattern exists. A KMS never sees your data. It sees only 32-byte keys to wrap and unwrap, so it can sit behind an API with strict access control, audit every call, and keep the KEK inside hardware it never leaves. A stolen database dump holds ciphertext and wrapped DEKs, and neither is useful without a call to the KMS.

Note the b"kek-v1" passed as associated data. It labels which KEK wrapped this DEK, and lesson 2’s AES-GCM check means a wrong label fails to unwrap. The next example uses that label.

Rotate the KEK by re-wrapping 32 bytes

Rotating a key means new encryptions use a new key, and old data is moved off the old one. With envelope encryption, “moved” means unwrapping each DEK with the old KEK and wrapping it with the new one.

# rotate_kek.py
import hashlib, os
from cryptography.exceptions import InvalidTag
from cryptography.hazmat.primitives.ciphers.aead import AESGCM

def wrap(kek, kek_id, dek):
    nonce = os.urandom(12)
    return {"kek_id": kek_id, "nonce": nonce,
            "wrapped": AESGCM(kek).encrypt(nonce, dek, kek_id.encode())}

def unwrap(kek, w):
    return AESGCM(kek).decrypt(w["nonce"], w["wrapped"], w["kek_id"].encode())

kek_v1, kek_v2 = AESGCM.generate_key(256), AESGCM.generate_key(256)
dek = AESGCM.generate_key(256)
nonce = os.urandom(12)
ciphertext = AESGCM(dek).encrypt(nonce, b"x" * 10_000_000, None)  # 10 MB
w1 = wrap(kek_v1, "kek-v1", dek)
before = hashlib.sha256(ciphertext).hexdigest()

w2 = wrap(kek_v2, "kek-v2", unwrap(kek_v1, w1))   # rotation: re-wrap 32 bytes
print("bytes re-encrypted:", len(unwrap(kek_v2, w2)))
print("data unchanged:    ", before == hashlib.sha256(ciphertext).hexdigest())
print("now wrapped under: ", w2["kek_id"])
try:
    unwrap(kek_v1, w2)
except InvalidTag:
    print("kek-v1 on new wrap: InvalidTag")

$ python rotate_kek.py
bytes re-encrypted: 32
data unchanged:     True
now wrapped under:  kek-v2
kek-v1 on new wrap: InvalidTag

Ten megabytes of ciphertext stayed byte-for-byte the same, and its SHA-256 matches. Only the 32-byte DEK was re-encrypted. After the re-wrap, the old KEK can’t open the new wrapping, so once every row reports kek-v2 you can retire kek-v1. That limits future exposure only. The DEK itself didn’t change, so anyone holding an old kek-v1 wrapping, from a database dump, a backup or a replica, plus a leaked kek-v1 can still recover it and decrypt the data. Routine rotation is re-wrapping. A KEK compromise means treating every DEK it wrapped as exposed, and re-encrypting that data under new DEKs.

Storing kek_id with each wrapped DEK matters as much as the crypto. Without it, a reader with three KEKs has to try them all, and a rotation job can’t tell which rows are finished. AWS KMS, Google Cloud KMS and Azure Key Vault each keep a key identifier or version with their wrapped output for the same reason.

The same holds when a single DEK leaks: re-wrapping it changes nothing, because the attacker already holds the key that encrypts the data. A leaked DEK means re-encrypting that record with a new DEK. KEK rotation is cheap; DEK rotation costs a full re-encryption, which is why DEKs are kept per record or per file, never shared across a whole database.

MultiFernet: rotation for Fernet tokens

Fernet is the library’s high-level recipe (AES-128-CBC plus HMAC-SHA256, in one token), often used with no DEK/KEK split. MultiFernet holds several Fernet keys, so you can rotate them.

# multifernet.py
from cryptography.fernet import Fernet, MultiFernet, InvalidToken

old_key, new_key = Fernet.generate_key(), Fernet.generate_key()
old_token = Fernet(old_key).encrypt(b"api-secret=sk_test_FAKE_123")

ring = MultiFernet([Fernet(new_key), Fernet(old_key)])  # first key encrypts
print("old token readable:", ring.decrypt(old_token))

rotated = ring.rotate(old_token)                         # decrypt, re-encrypt with new
print("rotated readable:  ", ring.decrypt(rotated))

retired = Fernet(old_key)                                # old key, on its own
for name, token in [("old token", old_token), ("rotated", rotated)]:
    try:
        retired.decrypt(token)
        print(f"retired key, {name:9}: ok")
    except InvalidToken:
        print(f"retired key, {name:9}: InvalidToken")

$ python multifernet.py
old token readable: b'api-secret=sk_test_FAKE_123'
rotated readable:   b'api-secret=sk_test_FAKE_123'
retired key, old token: ok
retired key, rotated  : InvalidToken

MultiFernet tries each key in turn to decrypt, but always encrypts with the first one. So the rotation order is: deploy the ring with the new key first and the old key second, so everything still reads. Then run rotate over every stored token, and finally remove the old key. The last two lines show why the order matters. The old key alone still reads the token nobody rotated, and fails on the rotated one. If you drop the old key before every token is rotated, those unrotated tokens become unreadable for good.

Unlike KEK rotation, rotate re-encrypts the whole token, because Fernet has no separate data key. That’s fine for short secrets such as API tokens and session data. For large files, use the envelope pattern above.

Bind each wrapped DEK to the row it belongs to

A wrapped DEK copied into another user’s row goes unnoticed in the first example. Putting the record ID into the associated data catches it.

# bind_context.py
import os
from cryptography.exceptions import InvalidTag
from cryptography.hazmat.primitives.ciphers.aead import AESGCM

kek = AESGCM.generate_key(256)

def wrap_for(record_id, dek):
    nonce = os.urandom(12)
    aad = f"kek-v1|{record_id}".encode()     # who this DEK belongs to
    return nonce, AESGCM(kek).encrypt(nonce, dek, aad)

def unwrap_for(record_id, nonce, wrapped):
    aad = f"kek-v1|{record_id}".encode()
    return AESGCM(kek).decrypt(nonce, wrapped, aad)

alice = wrap_for("user:1001", AESGCM.generate_key(256))

print("alice's DEK for user:1001:", len(unwrap_for("user:1001", *alice)), "bytes")
try:
    unwrap_for("user:1002", *alice)          # copied onto another user's row
except InvalidTag:
    print("alice's DEK for user:1002: InvalidTag")

$ python bind_context.py
alice's DEK for user:1001: 32 bytes
alice's DEK for user:1002: InvalidTag

The associated data isn’t secret and isn’t stored inside the ciphertext, but it is authenticated. The reader rebuilds it from the row being read, so a DEK moved to a different row fails to unwrap. Cloud KMS calls this the encryption context (AWS) or additional authenticated data (Google Cloud), and the same rule applies there. Put the tenant or record ID in it. On AWS, CloudTrail also records the encryption context with every KMS call, so the audit log shows whose key was unwrapped.

How this maps to a cloud KMS

Everything above ran locally, with the KEK held in a Python variable. In production the only change is where the KEK lives:

Terraform sensitive data and Ansible Vault show the same idea applied to secrets in configuration. GitHub Actions secrets shows it for CI.

Practice

1. Wrap with RFC 3394 AES Key Wrap instead. Many KMS and HSM APIs wrap keys with AES Key Wrap, not AES-GCM. Wrap a DEK with aes_key_wrap, compare its size with the 48 bytes above, and unwrap it with the wrong KEK.

# keywrap_demo.py
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
from cryptography.hazmat.primitives.keywrap import (
    InvalidUnwrap, aes_key_unwrap, aes_key_wrap)

kek, other_kek = AESGCM.generate_key(256), AESGCM.generate_key(256)
dek = AESGCM.generate_key(256)

wrapped = aes_key_wrap(kek, dek)              # RFC 3394: no nonce to store
print("wrapped DEK bytes:", len(wrapped))
print("unwrap ok:        ", aes_key_unwrap(kek, wrapped) == dek)
try:
    aes_key_unwrap(other_kek, wrapped)
except InvalidUnwrap:
    print("wrong KEK:         InvalidUnwrap")

$ python keywrap_demo.py
wrapped DEK bytes: 40
unwrap ok:         True
wrong KEK:         InvalidUnwrap

2. Rotate a whole table. Keep a keyring of {kek_id: kek}, wrap 1,000 DEKs under kek-v1, add kek-v2 as the current key, and re-wrap only the rows not already on it. Prove no row is left on kek-v1 before deleting it from the keyring.

# keyring_rotate.py
import os
from cryptography.hazmat.primitives.ciphers.aead import AESGCM

keyring = {"kek-v1": AESGCM.generate_key(256)}
current = "kek-v1"

def wrap(dek):
    nonce = os.urandom(12)
    return {"kek_id": current, "nonce": nonce,
            "wrapped": AESGCM(keyring[current]).encrypt(nonce, dek, current.encode())}

def unwrap(w):
    return AESGCM(keyring[w["kek_id"]]).decrypt(w["nonce"], w["wrapped"], w["kek_id"].encode())

rows = [wrap(AESGCM.generate_key(256)) for _ in range(1000)]
keyring["kek-v2"] = AESGCM.generate_key(256)
current = "kek-v2"                            # new writes use v2 from now on

stale = [i for i, w in enumerate(rows) if w["kek_id"] != current]
for i in stale:
    rows[i] = wrap(unwrap(rows[i]))           # lazy or batch: same operation
print("rows re-wrapped:", len(stale))
print("still on kek-v1:", sum(w["kek_id"] == "kek-v1" for w in rows))
del keyring["kek-v1"]                         # safe only after the count is 0
print("all readable:   ", all(len(unwrap(w)) == 32 for w in rows))

$ python keyring_rotate.py
rows re-wrapped: 1000
still on kek-v1: 0
all readable:    True

3. Get the MultiFernet order wrong. Build the ring with the old key first, then rotate a token and encrypt a new one. Can the old key, on its own, still read both?

# wrong_order.py
from cryptography.fernet import Fernet, MultiFernet

old_key, new_key = Fernet.generate_key(), Fernet.generate_key()
token = Fernet(old_key).encrypt(b"api-secret=sk_test_FAKE_123")

ring = MultiFernet([Fernet(old_key), Fernet(new_key)])  # old key listed first
rotated = ring.rotate(token)
fresh = ring.encrypt(b"new secret")

retired = Fernet(old_key)
print("retired key reads rotated:", retired.decrypt(rotated))
print("retired key reads fresh:  ", retired.decrypt(fresh))

$ python wrong_order.py
retired key reads rotated: b'api-secret=sk_test_FAKE_123'
retired key reads fresh:   b'new secret'

4. Crypto-shredding. Store two users’ records, each under its own DEK. Then “forget” one user by deleting only their 48-byte wrapped DEK, and show that the ciphertext left behind can no longer be read.

# shred.py
import os
from cryptography.hazmat.primitives.ciphers.aead import AESGCM

kek = AESGCM.generate_key(256)
store = {}                                    # user_id -> (wrapped DEK, data)

for user in ["user:1001", "user:1002"]:
    dek = AESGCM.generate_key(256)
    wn, dn = os.urandom(12), os.urandom(12)
    store[user] = {"wn": wn, "wrapped": AESGCM(kek).encrypt(wn, dek, user.encode()),
                   "dn": dn, "data": AESGCM(dek).encrypt(dn, f"{user} history".encode(), None)}

def read(user):
    row = store[user]
    if row["wrapped"] is None:
        return "unrecoverable (DEK destroyed)"
    dek = AESGCM(kek).decrypt(row["wn"], row["wrapped"], user.encode())
    return AESGCM(dek).decrypt(row["dn"], row["data"], None).decode()

store["user:1001"]["wrapped"] = None          # "right to be forgotten": drop 48 bytes
for user in store:
    print(f"{user}: {read(user)}")
print("ciphertext still on disk:", len(store["user:1001"]["data"]), "bytes")

$ python shred.py
user:1001: unrecoverable (DEK destroyed)
user:1002: user:1002 history
ciphertext still on disk: 33 bytes