aws2-3 Exercise 1: Envelope Encryption by Hand, With a Stand-In Key Service, Then With the AWS Encryption SDK ============================================================================================================= Run with Python 3.14.0 on Windows 11 in a scratch venv: cryptography 50.0.2, aws-encryption-sdk 4.0.7, boto3/botocore 1.43.112 (offline, through botocore's Stubber), parliament 1.6.4 with setuptools 80.10.2 (parliament imports pkg_resources). Setup: py -m venv .venv ; .venv/Scripts/python.exe -m pip install cryptography aws-encryption-sdk boto3 parliament "setuptools<81". VERIFICATION LEVEL: 1 (run here; no AWS account, no AWS CLI, no AWS call). LocalKms, the policy evaluator and the rotation protocol are MODELS of AWS KMS / IAM / Secrets Manager documentation pages read on 2026-10-12 (through a page-fetch tool that summarises pages): they are not AWS, and AWS may change its rules, limits and prices. Docker is installed but its engine was not running, so no container was used. --- envelope.py (LocalKms is NOT AWS KMS; the header lists which documented behaviours it copies) --- """envelope.py - envelope encryption, built by hand with the 'cryptography' package and a LOCAL STAND-IN for AWS KMS. *** LocalKms is NOT AWS KMS. *** It holds its keys in memory and copies a few behaviours that the AWS KMS developer guide documents (read 2026-10-12): * direct Encrypt accepts at most 4,096 bytes (symmetric) -> Encrypt API page * the ciphertext blob records which key and key version made it -> Encrypt API page ("AWS KMS stores this information in the ciphertext blob") * the encryption context is additional authenticated data: decrypt must give the exact same one -> Encrypt API page * rotation adds new key material; old material stays and still decrypts; the key ID does not change -> Rotate keys page * rotation does NOT re-encrypt data or rotate data keys -> Rotate keys page * GenerateDataKey returns a plaintext data key and a copy encrypted under the KMS key -> Encrypt API page / concepts * a KMS key that is not enabled cannot be used for cryptographic operations -> Deleting keys page (pending deletion) What is real here: the AES-256-GCM encryption itself, the sizes, the timings, and the failures on a wrong context or a tampered byte. """ import hashlib, json, os, struct, time from cryptography.exceptions import InvalidTag from cryptography.hazmat.primitives.ciphers.aead import AESGCM MAGIC = b"LK1" MAX_DIRECT = 4096 class KmsError(Exception): pass def canon(ctx): return json.dumps(ctx or {}, sort_keys=True, separators=(",", ":")).encode() class LocalKms: def __init__(self): self.keys = {} # key_id -> {"versions": [bytes,...], "enabled": bool} def create_key(self): kid = os.urandom(8).hex() self.keys[kid] = {"versions": [AESGCM.generate_key(256)], "enabled": True} return kid def rotate(self, kid): self.keys[kid]["versions"].append(AESGCM.generate_key(256)) return len(self.keys[kid]["versions"]) def set_enabled(self, kid, flag): self.keys[kid]["enabled"] = flag def _wrap(self, kid, version, data, ctx): nonce = os.urandom(12) body = AESGCM(self.keys[kid]["versions"][version - 1]).encrypt(nonce, data, canon(ctx)) return MAGIC + bytes.fromhex(kid) + bytes([version]) + nonce + body def _unwrap(self, blob, ctx): if blob[:3] != MAGIC: raise KmsError("InvalidCiphertextException: not a LocalKms blob") kid, version, nonce, body = blob[3:11].hex(), blob[11], blob[12:24], blob[24:] k = self.keys.get(kid) if k is None: raise KmsError("NotFoundException") if not k["enabled"]: raise KmsError("DisabledException: the key is not enabled") if version > len(k["versions"]): raise KmsError("InvalidCiphertextException: unknown key material version") try: return kid, version, AESGCM(k["versions"][version - 1]).decrypt(nonce, body, canon(ctx)) except InvalidTag: raise KmsError("InvalidCiphertextException: wrong encryption context or damaged blob") def _current(self, kid): k = self.keys[kid] if not k["enabled"]: raise KmsError("DisabledException: the key is not enabled") return len(k["versions"]) def encrypt(self, kid, plaintext, ctx=None): if len(plaintext) > MAX_DIRECT: raise KmsError(f"ValidationException: Plaintext is {len(plaintext)} bytes; the limit is {MAX_DIRECT}") return self._wrap(kid, self._current(kid), plaintext, ctx) def decrypt(self, blob, ctx=None): return self._unwrap(blob, ctx)[2] def generate_data_key(self, kid, ctx=None): dk = AESGCM.generate_key(256) return dk, self._wrap(kid, self._current(kid), dk, ctx) def reencrypt(self, blob, ctx=None): kid, _, pt = self._unwrap(blob, ctx) return self._wrap(kid, self._current(kid), pt, ctx) def envelope_encrypt(kms, kid, data, ctx): dk, wrapped = kms.generate_data_key(kid, ctx) nonce = os.urandom(12) ct = AESGCM(dk).encrypt(nonce, data, canon(ctx)) return {"wrapped_key": wrapped, "nonce": nonce, "ciphertext": ct, "ctx": ctx} def envelope_decrypt(kms, obj, ctx): dk = kms.decrypt(obj["wrapped_key"], ctx) return AESGCM(dk).decrypt(obj["nonce"], obj["ciphertext"], canon(ctx)) def attempt(label, fn): try: r = fn() print(f" {label:<78} -> ok{(' (' + str(r) + ')') if r is not None else ''}") return True except Exception as e: name = type(e).__name__ print(f" {label:<78} -> {name}: {str(e)[:70]}") return False if __name__ == "__main__": kms = LocalKms() kid = kms.create_key() ctx = {"app": "invoices", "tenant": "t-17"} print("1. direct encryption has a small limit (4,096 bytes, as documented for KMS)") attempt("encrypt 4,096 bytes directly", lambda: len(kms.encrypt(kid, os.urandom(4096), ctx))) attempt("encrypt 4,097 bytes directly", lambda: kms.encrypt(kid, os.urandom(4097), ctx)) print("\n2. envelope encryption of 50 MB: only the 32-byte data key goes through the key service") data = os.urandom(50 * 1024 * 1024) t = time.perf_counter(); obj = envelope_encrypt(kms, kid, data, ctx); te = time.perf_counter() - t t = time.perf_counter(); back = envelope_decrypt(kms, obj, ctx); td = time.perf_counter() - t print(f" plaintext {len(data):,} bytes; ciphertext {len(obj['ciphertext']):,} bytes (+{len(obj['ciphertext']) - len(data)}); nonce {len(obj['nonce'])}; wrapped data key {len(obj['wrapped_key'])} bytes") print(f" round trip identical: {back == data}; encrypt {te:.2f} s, decrypt {td:.2f} s on this machine (single run)") print("\n3. the encryption context is bound to the data") small = envelope_encrypt(kms, kid, b"invoice 4411: total 1,250.00", ctx) attempt("decrypt with the same context", lambda: envelope_decrypt(kms, small, ctx).decode()) attempt("decrypt with tenant t-18 instead of t-17", lambda: envelope_decrypt(kms, small, {"app": "invoices", "tenant": "t-18"})) attempt("decrypt with no context at all", lambda: envelope_decrypt(kms, small, {})) bad = dict(small, ciphertext=bytes([small["ciphertext"][0] ^ 1]) + small["ciphertext"][1:]) attempt("decrypt after flipping ONE bit of the ciphertext", lambda: envelope_decrypt(kms, bad, ctx)) other = envelope_encrypt(kms, kid, b"invoice 9000: total 99.00", ctx) swapped = dict(small, wrapped_key=other["wrapped_key"]) attempt("decrypt after swapping in another object's wrapped data key", lambda: envelope_decrypt(kms, swapped, ctx)) print("\n4. rotation changes the KEY MATERIAL, not your data (the Rotate keys page)") before = hashlib.sha256(small["ciphertext"]).hexdigest()[:12] v_old = small["wrapped_key"][11] print(f" wrapped key made with key version {v_old}; rotating the key ...") print(f" key now has {kms.rotate(kid)} versions") attempt("decrypt the OLD object after rotation", lambda: envelope_decrypt(kms, small, ctx).decode()) new = envelope_encrypt(kms, kid, b"invoice 5000", ctx) print(f" a NEW object is wrapped with key version {new['wrapped_key'][11]}") moved = dict(small, wrapped_key=kms.reencrypt(small["wrapped_key"], ctx)) after = hashlib.sha256(moved["ciphertext"]).hexdigest()[:12] print(f" re-encrypting only the wrapped data key moves it to version {moved['wrapped_key'][11]}; bulk ciphertext hash before {before} after {after} (unchanged)") attempt("decrypt the object after the wrapped key was moved", lambda: envelope_decrypt(kms, moved, ctx).decode()) kms.keys[kid]["versions"][0] = os.urandom(32) # simulate deleting version 1's material attempt("decrypt an object still wrapped under the DESTROYED version 1", lambda: envelope_decrypt(kms, small, ctx)) attempt("decrypt the object that had been moved to version 2", lambda: envelope_decrypt(kms, moved, ctx).decode()) print("\n5. a key that is not enabled cannot unwrap data keys (the Deleting keys page: pending deletion cannot be used)") kid2 = kms.create_key(); obj2 = envelope_encrypt(kms, kid2, b"payroll", {"app": "hr"}) dk_in_memory = kms.decrypt(obj2["wrapped_key"], {"app": "hr"}) # a running app still holds its plaintext data key kms.set_enabled(kid2, False) attempt("a fresh process asks the key service to unwrap", lambda: envelope_decrypt(kms, obj2, {"app": "hr"})) attempt("the running app that cached the plaintext data key", lambda: AESGCM(dk_in_memory).decrypt(obj2["nonce"], obj2["ciphertext"], canon({"app": "hr"})).decode()) attempt("generate a NEW data key under the disabled key", lambda: kms.generate_data_key(kid2, {"app": "hr"})) kms.set_enabled(kid2, True) attempt("after re-enabling the key", lambda: envelope_decrypt(kms, obj2, {"app": "hr"}).decode()) --- py -X utf8 envelope.py (timings are one run on this machine) --- 1. direct encryption has a small limit (4,096 bytes, as documented for KMS) encrypt 4,096 bytes directly -> ok (4136) encrypt 4,097 bytes directly -> KmsError: ValidationException: Plaintext is 4097 bytes; the limit is 4096 2. envelope encryption of 50 MB: only the 32-byte data key goes through the key service plaintext 52,428,800 bytes; ciphertext 52,428,816 bytes (+16); nonce 12; wrapped data key 72 bytes round trip identical: True; encrypt 0.02 s, decrypt 0.01 s on this machine (single run) 3. the encryption context is bound to the data decrypt with the same context -> ok (invoice 4411: total 1,250.00) decrypt with tenant t-18 instead of t-17 -> KmsError: InvalidCiphertextException: wrong encryption context or damaged blob decrypt with no context at all -> KmsError: InvalidCiphertextException: wrong encryption context or damaged blob decrypt after flipping ONE bit of the ciphertext -> InvalidTag: decrypt after swapping in another object's wrapped data key -> InvalidTag: 4. rotation changes the KEY MATERIAL, not your data (the Rotate keys page) wrapped key made with key version 1; rotating the key ... key now has 2 versions decrypt the OLD object after rotation -> ok (invoice 4411: total 1,250.00) a NEW object is wrapped with key version 2 re-encrypting only the wrapped data key moves it to version 2; bulk ciphertext hash before 25f1d45e58ed after 25f1d45e58ed (unchanged) decrypt the object after the wrapped key was moved -> ok (invoice 4411: total 1,250.00) decrypt an object still wrapped under the DESTROYED version 1 -> KmsError: InvalidCiphertextException: wrong encryption context or damaged blob decrypt the object that had been moved to version 2 -> ok (invoice 4411: total 1,250.00) 5. a key that is not enabled cannot unwrap data keys (the Deleting keys page: pending deletion cannot be used) a fresh process asks the key service to unwrap -> KmsError: DisabledException: the key is not enabled the running app that cached the plaintext data key -> ok (payroll) generate a NEW data key under the disabled key -> KmsError: DisabledException: the key is not enabled after re-enabling the key -> ok (payroll) --- esdk_demo.py (the real AWS Encryption SDK with a RAW AES key; no KMS) --- """esdk_demo.py - the AWS Encryption SDK for Python (real library, version printed) doing envelope encryption with a RAW AES key that lives on this machine, instead of a KMS key. No call to AWS is made, and nothing here proves how a KMS keyring behaves; it shows what the SDK itself does: message format, encryption context, commitment policy and failure behaviour. """ import logging, os, sys, warnings warnings.filterwarnings("ignore") logging.getLogger("aws_encryption_sdk").setLevel(logging.CRITICAL) # the SDK logs a traceback for every failed decrypt sys.unraisablehook = lambda *a: None # and a 'Footer not read' message from a destructor import aws_encryption_sdk from aws_encryption_sdk import CommitmentPolicy from aws_encryption_sdk.identifiers import EncryptionKeyType, WrappingAlgorithm from aws_encryption_sdk.internal.crypto.wrapping_keys import WrappingKey from aws_encryption_sdk.key_providers.raw import RawMasterKeyProvider class LocalProvider(RawMasterKeyProvider): provider_id = "local-demo" _keys = {} def _get_raw_key(self, key_id): return WrappingKey(wrapping_algorithm=WrappingAlgorithm.AES_256_GCM_IV12_TAG16_NO_PADDING, wrapping_key=self._keys[key_id.decode() if isinstance(key_id, bytes) else key_id], wrapping_key_type=EncryptionKeyType.SYMMETRIC) def provider(keys, ids): p = LocalProvider() p._keys = keys for i in ids: p.add_master_key(i) return p def attempt(label, fn): try: r = fn() print(f" {label:<76} -> ok{(' (' + str(r) + ')') if r is not None else ''}") except Exception as e: print(f" {label:<76} -> {type(e).__name__}: {str(e)[:70]}") if __name__ == "__main__": print("aws-encryption-sdk", aws_encryption_sdk.__version__) client = aws_encryption_sdk.EncryptionSDKClient(commitment_policy=CommitmentPolicy.REQUIRE_ENCRYPT_REQUIRE_DECRYPT) keys = {"key-A": os.urandom(32), "key-B": os.urandom(32)} ctx = {"app": "invoices", "tenant": "t-17"} data = b"invoice 4411: total 1,250.00 " * 1000 ct, header = client.encrypt(source=data, key_provider=provider(keys, ["key-A"]), encryption_context=ctx) print(f"\nplaintext {len(data):,} bytes -> message {len(ct):,} bytes (+{len(ct) - len(data)})") print("message format version:", header.version, "| algorithm:", header.algorithm.name) print("encrypted data keys in the header:", len(header.encrypted_data_keys), "| commitment policy: REQUIRE_ENCRYPT_REQUIRE_DECRYPT") print("encryption context in the header:", {k: (v[:20] + "...") if len(v) > 24 else v for k, v in header.encryption_context.items()}) for needle in (b"invoices", b"t-17", b"1,250.00"): print(f" the bytes {needle!r} appear in the ciphertext message: {needle in ct}") print("\ndecrypting") attempt("with the right key", lambda: len(client.decrypt(source=ct, key_provider=provider(keys, ["key-A"]))[0])) attempt("with the wrong key (key-B only)", lambda: client.decrypt(source=ct, key_provider=provider(keys, ["key-B"]))) pt, hdr2 = client.decrypt(source=ct, key_provider=provider(keys, ["key-A"])) print(" the decrypt result also returns the context:", hdr2.encryption_context == header.encryption_context) tampered = bytearray(ct); tampered[-20] ^= 1 attempt("after flipping one bit near the end of the message", lambda: client.decrypt(source=bytes(tampered), key_provider=provider(keys, ["key-A"]))) tampered = bytearray(ct); tampered[60] ^= 1 attempt("after flipping one bit in the header", lambda: client.decrypt(source=bytes(tampered), key_provider=provider(keys, ["key-A"]))) print("\nrequiring the context on decrypt (the application's job: the SDK returns it, you must check it)") def require(expected): _, h = client.decrypt(source=ct, key_provider=provider(keys, ["key-A"])) missing = {k: v for k, v in expected.items() if h.encryption_context.get(k) != v} if missing: raise ValueError("context mismatch: " + str(missing)) return "context matches" attempt("expect tenant t-17", lambda: require({"tenant": "t-17"})) attempt("expect tenant t-99", lambda: require({"tenant": "t-99"})) print("\ntwo master keys: either one decrypts (a message can be readable by two keys, e.g. two Regions)") ct2, h2 = client.encrypt(source=b"shared secret", key_provider=provider(keys, ["key-A", "key-B"]), encryption_context=ctx) print(" encrypted data keys:", len(h2.encrypted_data_keys)) attempt("decrypt with key-A only", lambda: client.decrypt(source=ct2, key_provider=provider(keys, ["key-A"]))[0].decode()) attempt("decrypt with key-B only", lambda: client.decrypt(source=ct2, key_provider=provider(keys, ["key-B"]))[0].decode()) --- py -X utf8 esdk_demo.py (the base64 public key in the header differs on every run) --- aws-encryption-sdk 4.0.7 plaintext 29,000 bytes -> message 29,674 bytes (+674) message format version: SerializationVersion.V2 | algorithm: AES_256_GCM_HKDF_SHA512_COMMIT_KEY_ECDSA_P384 encrypted data keys in the header: 1 | commitment policy: REQUIRE_ENCRYPT_REQUIRE_DECRYPT encryption context in the header: {'app': 'invoices', 'tenant': 't-17', 'aws-crypto-public-key': 'A3iPVpBnrKHuE1pbUBLx...'} the bytes b'invoices' appear in the ciphertext message: True the bytes b't-17' appear in the ciphertext message: True the bytes b'1,250.00' appear in the ciphertext message: False decrypting with the right key -> ok (29000) with the wrong key (key-B only) -> DecryptKeyError: Unable to decrypt any data key the decrypt result also returns the context: True after flipping one bit near the end of the message -> InvalidSignature: after flipping one bit in the header -> InvalidTag: requiring the context on decrypt (the application's job: the SDK returns it, you must check it) expect tenant t-17 -> ok (context matches) expect tenant t-99 -> ValueError: context mismatch: {'tenant': 't-99'} two master keys: either one decrypts (a message can be readable by two keys, e.g. two Regions) encrypted data keys: 2 decrypt with key-A only -> ok (shared secret) decrypt with key-B only -> ok (shared secret) WHY THIS WORKS AS AN ANSWER --------------------------- The envelope pattern is built from real AES-256-GCM, so the sizes and failures are real: 4,097 bytes are refused by the stand-in (as the KMS Encrypt page says for KMS), 50 MB is encrypted with 16 bytes of overhead, a wrong or missing encryption context, one flipped bit and a swapped wrapped key all fail. The rotation part mirrors the Rotate keys page: new key material is added, old material still decrypts, the bulk ciphertext hash is unchanged when only the wrapped key is re-encrypted, and an object still wrapped under destroyed material is lost. The disabled-key part shows why a key change can seem harmless: a process that cached the plaintext data key keeps working, a new process cannot unwrap. The Encryption SDK part is a real library, but with a local key: it shows the message format and that the encryption context is stored in clear text in the message (not secret), that the SDK returns the context and the application must check it, and that damage in the header or at the end fails; it proves nothing about a KMS keyring. Not run: any real KMS call, latency or throttling; data key caching; a KMS keyring; other languages' SDKs.