aws2-3 Exercise 3: Rotation, Sizes and Costs, Secret Leaks, and Evidence in CloudTrail ========================================================================================= 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. The CloudTrail events are the documentation's printed examples plus INVENTED events (marked synthetic); costs are arithmetic on prices read on 2026-10-12 (no Region stated; free plans not applied), not a quote. --- rotation_model.py --- """rotation_model.py - the Secrets Manager rotation protocol (staging labels and the four Lambda steps) as a small MODEL. NOT AWS. Rules taken from the Secrets Manager pages read 2026-10-12 ("Lambda rotation functions", "Quotas"): * versions carry staging labels; AWSCURRENT is the version applications read * createSecret stores the new value under the label AWSPENDING (with the ClientRequestToken as the version id); it first checks whether a version with that token already exists, which "helps ensure idempotency" * setSecret changes the credential in the database to match AWSPENDING; before that it must check that AWSCURRENT is valid, that both versions are for the same user and the same host (the rotation function is a privileged deputy: confused deputy guard) * testSecret uses AWSPENDING against the database * finishSecret moves AWSCURRENT to the new version; Secrets Manager adds AWSPREVIOUS to the old one * if AWSPENDING exists on a version other than AWSCURRENT, a later rotation "assumes that a previous rotation request is still in progress and returns an error" * quotas: secret value 65,536 bytes; 100 versions per secret; unlabelled versions are removed when there are more than 100, but not those created less than 24 hours ago My reading, not a documented sentence: a retry with the SAME token resumes the failed rotation (that is what the idempotency remark is for), and a put that cannot make room under the 100-version quota fails. """ import secrets as pysecrets MAX_BYTES, MAX_VERSIONS, DAY = 65536, 100, 24 * 3600 class SmError(Exception): pass class Secret: def __init__(self, value): self.versions, self.labels, self.clock = {}, {}, 0.0 self.put(value, "v1", ["AWSCURRENT"]) def put(self, value, vid, labels=()): if len(value.encode()) > MAX_BYTES: raise SmError(f"secret value is {len(value.encode())} bytes; the limit is {MAX_BYTES}") if len(self.versions) >= MAX_VERSIONS: old = [v for v, d in self.versions.items() if v not in self.labels.values() and self.clock - d["created"] >= DAY] if not old: raise SmError(f"LimitExceeded: {MAX_VERSIONS} versions and none can be removed (unlabelled and older than 24 h)") for v in old[:len(self.versions) - MAX_VERSIONS + 1]: del self.versions[v] self.versions[vid] = {"value": value, "created": self.clock} for l in labels: self.labels[l] = vid def get(self, label="AWSCURRENT"): return self.versions[self.labels[label]]["value"] if label in self.labels else None def stages(self): out = {} for l, v in sorted(self.labels.items()): out.setdefault(v, []).append(l) return out def move_current(self, to_vid): old = self.labels.get("AWSCURRENT") if old: self.labels["AWSPREVIOUS"] = old # Secrets Manager adds AWSPREVIOUS to the old version self.labels["AWSCURRENT"] = to_vid class FakeDb: def __init__(self, user, password, host): self.user, self.password, self.host = user, password, host def connect(self, user, password, host): return (user, password, host) == (self.user, self.password, self.host) def parse(v): u, p, h = v.split(":") return {"user": u, "password": p, "host": h} def rotate(sec, db, token, fail_at=None, tamper_pending=None): """one rotation attempt. fail_at: step name that raises. returns the list of steps completed.""" cur_v, pend_v = sec.labels.get("AWSCURRENT"), sec.labels.get("AWSPENDING") if pend_v is not None and pend_v != cur_v and pend_v != token: raise SmError("a previous rotation is still in progress (AWSPENDING is on another version)") done = [] def step(name, fn): if fail_at == name: raise SmError("simulated failure in " + name) fn(); done.append(name) def createSecret(): if token not in sec.versions: # idempotent: reuse the pending version if this token already made one c = parse(sec.get()) sec.put(f"{c['user']}:{pysecrets.token_urlsafe(8)}:{c['host']}", token, ["AWSPENDING"]) def setSecret(): c, p = parse(sec.get()), parse(sec.versions[token]["value"]) if not db.connect(c["user"], c["password"], c["host"]): raise SmError("abandoned: the AWSCURRENT credential is not valid") if (c["user"], c["host"]) != (p["user"], p["host"]): raise SmError(f"abandoned: AWSCURRENT and AWSPENDING are for different resources ({c['user']}@{c['host']} vs {p['user']}@{p['host']})") db.password = p["password"] def testSecret(): p = parse(sec.versions[token]["value"]) if not db.connect(p["user"], p["password"], p["host"]): raise SmError("the AWSPENDING credential does not work") def finishSecret(): sec.move_current(token) for name, fn in (("createSecret", createSecret), ("setSecret", setSecret), ("testSecret", testSecret), ("finishSecret", finishSecret)): if name == "setSecret" and tamper_pending: sec.versions[token]["value"] = tamper_pending step(name, fn) return done def show(sec, db, note=""): st = ", ".join(f"{v}={'/'.join(l)}" for v, l in sec.stages().items()) reads = sec.get() works = db.connect(**parse(reads)) if reads else None print(f" labels: {st:<48} app reading AWSCURRENT can connect: {works} {note}") def attempt(label, fn): try: r = fn() print(f" {label:<70} -> ok {r if r else ''}") return True except SmError as e: print(f" {label:<70} -> SmError: {e}") return False if __name__ == "__main__": print("1. a normal rotation, step by step (the model prints the state after each step)") db = FakeDb("app", "p-initial", "db1"); sec = Secret("app:p-initial:db1") show(sec, db, "(start)") for upto in ("createSecret", "setSecret", "testSecret", "finishSecret"): db = FakeDb("app", "p-initial", "db1"); sec = Secret("app:p-initial:db1") steps = ("createSecret", "setSecret", "testSecret", "finishSecret") try: rotate(sec, db, "v2", fail_at=(steps[steps.index(upto) + 1] if upto != "finishSecret" else None)) # stop just after `upto` except SmError: pass show(sec, db, f"(after {upto})") print("\n2. failure inside setSecret, then a retry with the same token") db = FakeDb("app", "p-initial", "db1"); sec = Secret("app:p-initial:db1") attempt("rotate(token v2), failing in setSecret", lambda: rotate(sec, db, "v2", fail_at="setSecret")) show(sec, db, "(application unaffected)") pending_before = sec.get("AWSPENDING") attempt("retry rotate(token v2)", lambda: rotate(sec, db, "v2")) print(f" the retry reused the pending value: {sec.versions['v2']['value'] == pending_before}") show(sec, db, "(after retry)") print("\n3. a rotation left in progress blocks a rotation with a different token") db = FakeDb("app", "p-initial", "db1"); sec = Secret("app:p-initial:db1") attempt("first rotation fails in testSecret", lambda: rotate(sec, db, "v2", fail_at="testSecret")) attempt("a new rotation with token v3", lambda: rotate(sec, db, "v3")) attempt("the same token v2 again", lambda: rotate(sec, db, "v2")) print("\n4. the confused-deputy checks in setSecret") db = FakeDb("app", "p-initial", "db1"); sec = Secret("app:p-initial:db1") attempt("an attacker edits the AWSPENDING version to user 'admin' on the same host", lambda: rotate(sec, db, "v2", tamper_pending="admin:evil:db1")) print(f" database password unchanged: {db.password == 'p-initial'}") db = FakeDb("app", "p-initial", "db1"); sec = Secret("app:p-initial:db1") attempt("an attacker points AWSPENDING at another host", lambda: rotate(sec, db, "v2", tamper_pending="app:evil:db-other")) db = FakeDb("app", "changed-behind-our-back", "db1"); sec = Secret("app:p-initial:db1") attempt("the AWSCURRENT credential no longer works", lambda: rotate(sec, db, "v2")) print("\n5. quotas (value size, and the 100-version limit with the 24-hour rule)") sec = Secret("x") attempt("put a value of 65,536 bytes", lambda: sec.put("a" * 65536, "big1")) attempt("put a value of 65,537 bytes", lambda: sec.put("a" * 65537, "big2")) for every_min in (5, 10, 15): sec = Secret("start"); n = 0 try: for i in range(2, 3 * 24 * 60 // every_min): sec.clock = (i - 1) * every_min * 60 sec.put(f"value {i}", f"v{i}"); n += 1 print(f" an update every {every_min:>2} min for 3 days: {n} updates, {len(sec.versions)} versions kept, never hit the limit") except SmError as e: print(f" an update every {every_min:>2} min: failed at update {n + 2} (after {sec.clock / 3600:.1f} h): {str(e)[:60]}") --- py -X utf8 rotation_model.py --- 1. a normal rotation, step by step (the model prints the state after each step) labels: v1=AWSCURRENT app reading AWSCURRENT can connect: True (start) labels: v1=AWSCURRENT, v2=AWSPENDING app reading AWSCURRENT can connect: True (after createSecret) labels: v1=AWSCURRENT, v2=AWSPENDING app reading AWSCURRENT can connect: False (after setSecret) labels: v1=AWSCURRENT, v2=AWSPENDING app reading AWSCURRENT can connect: False (after testSecret) labels: v2=AWSCURRENT/AWSPENDING, v1=AWSPREVIOUS app reading AWSCURRENT can connect: True (after finishSecret) 2. failure inside setSecret, then a retry with the same token rotate(token v2), failing in setSecret -> SmError: simulated failure in setSecret labels: v1=AWSCURRENT, v2=AWSPENDING app reading AWSCURRENT can connect: True (application unaffected) retry rotate(token v2) -> ok ['createSecret', 'setSecret', 'testSecret', 'finishSecret'] the retry reused the pending value: True labels: v2=AWSCURRENT/AWSPENDING, v1=AWSPREVIOUS app reading AWSCURRENT can connect: True (after retry) 3. a rotation left in progress blocks a rotation with a different token first rotation fails in testSecret -> SmError: simulated failure in testSecret a new rotation with token v3 -> SmError: a previous rotation is still in progress (AWSPENDING is on another version) the same token v2 again -> SmError: abandoned: the AWSCURRENT credential is not valid 4. the confused-deputy checks in setSecret an attacker edits the AWSPENDING version to user 'admin' on the same host -> SmError: abandoned: AWSCURRENT and AWSPENDING are for different resources (app@db1 vs admin@db1) database password unchanged: True an attacker points AWSPENDING at another host -> SmError: abandoned: AWSCURRENT and AWSPENDING are for different resources (app@db1 vs app@db-other) the AWSCURRENT credential no longer works -> SmError: abandoned: the AWSCURRENT credential is not valid 5. quotas (value size, and the 100-version limit with the 24-hour rule) put a value of 65,536 bytes -> ok put a value of 65,537 bytes -> SmError: secret value is 65537 bytes; the limit is 65536 an update every 5 min: failed at update 101 (after 8.3 h): LimitExceeded: 100 versions and none can be removed (unlabel an update every 10 min: failed at update 101 (after 16.7 h): LimitExceeded: 100 versions and none can be removed (unlabel an update every 15 min for 3 days: 286 updates, 100 versions kept, never hit the limit --- secrets_compare.py --- """secrets_compare.py - will real key material fit in Parameter Store or Secrets Manager, and what does reading secrets cost? Real part: key and certificate sizes measured here with the 'cryptography' package (PEM text, as you would store it). Limits (AWS pages read 2026-10-12): Secrets Manager secret value 65,536 bytes; Parameter Store standard 4 KB, advanced 8 KB (I take KB as 1,024 bytes; the page says 'KB'). Prices (AWS pricing pages read 2026-10-12, text as returned by a page-fetch tool; no Region was named): Secrets Manager $0.40 per secret per month and $0.05 per 10,000 API calls; KMS customer managed key $1 per month and $0.03 per 10,000 requests with a free tier of 20,000 requests per month. Parameter Store standard: 'no additional charge' (advanced: charges apply, price not read). The cost part is ARITHMETIC on those figures, not a quote and not a measurement. """ import datetime, json from cryptography import x509 from cryptography.hazmat.primitives import hashes, serialization from cryptography.hazmat.primitives.asymmetric import ec, rsa from cryptography.x509.oid import NameOID LIMITS = {"Parameter Store standard": 4 * 1024, "Parameter Store advanced": 8 * 1024, "Secrets Manager": 65536} def pem_key(k): return k.private_bytes(serialization.Encoding.PEM, serialization.PrivateFormat.PKCS8, serialization.NoEncryption()) def cert(key, cn, issuer_key=None, issuer_cn=None): name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, cn)]) iname = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, issuer_cn or cn)]) now = datetime.datetime(2026, 10, 12, tzinfo=datetime.timezone.utc) c = (x509.CertificateBuilder().subject_name(name).issuer_name(iname).public_key(key.public_key()).serial_number(x509.random_serial_number()) .not_valid_before(now).not_valid_after(now + datetime.timedelta(days=365)).sign(issuer_key or key, hashes.SHA256())) return c.public_bytes(serialization.Encoding.PEM) def items(): r2, r4 = rsa.generate_private_key(65537, 2048), rsa.generate_private_key(65537, 4096) ec256 = ec.generate_private_key(ec.SECP256R1()) root, inter = rsa.generate_private_key(65537, 2048), rsa.generate_private_key(65537, 2048) chain = cert(r2, "leaf.example.test", inter, "inter.example.test") + cert(inter, "inter.example.test", root, "root.example.test") + cert(root, "root.example.test") out = { "database credentials as JSON": json.dumps({"username": "app", "password": "x" * 32, "host": "db1.example.internal", "port": 5432}).encode(), "EC P-256 private key (PEM)": pem_key(ec256), "RSA-2048 private key (PEM)": pem_key(r2), "RSA-4096 private key (PEM)": pem_key(r4), "RSA-2048 key + 3-certificate chain": pem_key(r2) + chain, "RSA-4096 key + 3-certificate chain": pem_key(r4) + chain, } return out if __name__ == "__main__": print("1. does it fit? (bytes of the PEM / JSON text)") print(f" {'item':<38}{'bytes':>7} " + " ".join(f"{k:<26}" for k in LIMITS)) for name, data in items().items(): print(f" {name:<38}{len(data):>7} " + " ".join(f"{'fits' if len(data) <= v else 'TOO BIG':<26}" for v in LIMITS.values())) print("\n2. what reading a secret costs: arithmetic on the prices above (Secrets Manager only; Parameter Store standard has no additional charge)") secs = 25 for label, per_month in (("each of 200 container starts a day reads every secret once", 200 * 30 * secs), ("each request reads every secret: 10 requests/s", int(10 * 3600 * 24 * 30) * secs), ("the same 10 requests/s, with a 5-minute cache in each of 20 instances", 20 * (30 * 24 * 12) * secs)): storage, calls = secs * 0.40, per_month / 10_000 * 0.05 print(f" {label:<76} {per_month:>13,} calls storage ${storage:>6.2f} + calls ${calls:>9.2f} = ${storage + calls:>9.2f} per month") print(" (25 secrets; the free-trial rules and Regional price differences were not stated on the page and are not applied)") --- py -X utf8 secrets_compare.py --- 1. does it fit? (bytes of the PEM / JSON text) item bytes Parameter Store standard Parameter Store advanced Secrets Manager database credentials as JSON 113 fits fits fits EC P-256 private key (PEM) 241 fits fits fits RSA-2048 private key (PEM) 1704 fits fits fits RSA-4096 private key (PEM) 3272 fits fits fits RSA-2048 key + 3-certificate chain 4767 TOO BIG fits fits RSA-4096 key + 3-certificate chain 6335 TOO BIG fits fits 2. what reading a secret costs: arithmetic on the prices above (Secrets Manager only; Parameter Store standard has no additional charge) each of 200 container starts a day reads every secret once 150,000 calls storage $ 10.00 + calls $ 0.75 = $ 10.75 per month each request reads every secret: 10 requests/s 648,000,000 calls storage $ 10.00 + calls $ 3240.00 = $ 3250.00 per month the same 10 requests/s, with a 5-minute cache in each of 20 instances 4,320,000 calls storage $ 10.00 + calls $ 21.60 = $ 31.60 per month (25 secrets; the free-trial rules and Regional price differences were not stated on the page and are not applied) --- secret_leaks.py --- """secret_leaks.py - where a secret passed as an environment variable or a command-line argument ends up. REAL experiments on this Windows machine with child processes; nothing touches AWS. 'hunter2' is a made-up password. Not run: docker inspect / docker exec env (Docker is installed but its engine was not running), Lambda or ECS environment settings. """ import os, re, subprocess, sys, time PY = sys.executable SECRET = "hunter2" def run_child(code, env=None): r = subprocess.run([PY, "-c", code], env=env, capture_output=True, text=True, stdin=subprocess.DEVNULL, timeout=60) return r.stdout.strip() if __name__ == "__main__": env = dict(os.environ, DB_PASSWORD=SECRET) print("1. an environment variable is inherited by every child process, and by their children") child = "import os,subprocess,sys; print('child sees:', os.environ.get('DB_PASSWORD')); print('grandchild sees:', subprocess.run([sys.executable,'-c','import os;print(os.environ.get(\"DB_PASSWORD\"))'],capture_output=True,text=True).stdout.strip())" for line in run_child(child, env).splitlines(): print(" ", line) print(" child started with the variable removed:", run_child("import os;print(os.environ.get('DB_PASSWORD'))", {k: v for k, v in env.items() if k != 'DB_PASSWORD'})) print("\n2. a 'debug dump' of the environment leaks it; redacting by key name misses a password inside a URL") envx = dict(env, DATABASE_URL=f"postgres://app:{SECRET}@db1.example.internal:5432/shop", LOG_LEVEL="debug") dump = run_child("import os,json;print(json.dumps(dict(os.environ)))", envx) print(" raw dump contains the secret:", SECRET in dump) SENSITIVE = re.compile(r"(PASS|SECRET|TOKEN|KEY)", re.I) import json redacted = {k: ("***" if SENSITIVE.search(k) else v) for k, v in json.loads(dump).items() if k in ("DB_PASSWORD", "DATABASE_URL", "LOG_LEVEL")} print(" after redacting variable NAMES that look sensitive:", json.dumps(redacted)) print(" the secret still appears:", SECRET in json.dumps(redacted), "(inside DATABASE_URL: the name has no PASS/SECRET/TOKEN/KEY in it)") print("\n3. a command-line argument is readable by any other process of the same user, while the program runs") p = subprocess.Popen([PY, "-c", "import time; time.sleep(25)", f"--password={SECRET}"], stdin=subprocess.DEVNULL) time.sleep(1.5) r = subprocess.run(["powershell", "-NoProfile", "-Command", f"(Get-CimInstance Win32_Process -Filter 'ProcessId={p.pid}').CommandLine"], capture_output=True, text=True, timeout=60) print(" another process reads its command line:", r.stdout.strip().replace(PY, "")) p.kill() print("\n4. an alternative: pass a FILE PATH (not the secret), or send the secret over a pipe at start-up") path = os.path.join(os.environ.get("TEMP", "."), "demo_secret.txt") open(path, "w").write(SECRET) child_env = {k: v for k, v in env.items() if k != "DB_PASSWORD"} p = subprocess.Popen([PY, "-c", "import sys,time; print('read from file:', open(sys.argv[1]).read()); time.sleep(25)", path], env=child_env, stdout=subprocess.PIPE, stdin=subprocess.DEVNULL, text=True) time.sleep(1.5) r = subprocess.run(["powershell", "-NoProfile", "-Command", f"(Get-CimInstance Win32_Process -Filter 'ProcessId={p.pid}').CommandLine"], capture_output=True, text=True, timeout=60) print(" another process reads its command line:", r.stdout.strip().replace(PY, "").replace(path, "/demo_secret.txt")) print(" the secret appears in that command line:", SECRET in r.stdout) p.kill(); os.remove(path) code = "import sys,time; s=sys.stdin.readline().strip(); print('child got the secret over a pipe, length', len(s)); sys.stdout.flush()" p = subprocess.Popen([PY, "-c", code], env=child_env, stdin=subprocess.PIPE, stdout=subprocess.PIPE, text=True) out, _ = p.communicate(SECRET + "\n", timeout=30) print(" ", out.strip(), "| environment copy in the child:", run_child("import os;print(os.environ.get('DB_PASSWORD'))", child_env)) print(" (a file or a pipe still has to be protected: the file needs permissions, and anything the process prints can still leak it)") --- py -X utf8 secret_leaks.py --- 1. an environment variable is inherited by every child process, and by their children child sees: hunter2 grandchild sees: hunter2 child started with the variable removed: None 2. a 'debug dump' of the environment leaks it; redacting by key name misses a password inside a URL raw dump contains the secret: True after redacting variable NAMES that look sensitive: {"DATABASE_URL": "postgres://app:hunter2@db1.example.internal:5432/shop", "DB_PASSWORD": "***", "LOG_LEVEL": "debug"} the secret still appears: True (inside DATABASE_URL: the name has no PASS/SECRET/TOKEN/KEY in it) 3. a command-line argument is readable by any other process of the same user, while the program runs another process reads its command line: -c "import time; time.sleep(25)" --password=hunter2 4. an alternative: pass a FILE PATH (not the secret), or send the secret over a pipe at start-up another process reads its command line: -c "import sys,time; print('read from file:', open(sys.argv[1]).read()); time.sleep(25)" /demo_secret.txt the secret appears in that command line: False child got the secret over a pipe, length 7 | environment copy in the child: None (a file or a pipe still has to be protected: the file needs permissions, and anything the process prints can still leak it) --- ct/decrypt_ok.json (copied from the KMS developer guide: valid) --- { "eventVersion": "1.11", "userIdentity": { "type": "IAMUser", "principalId": "EX_PRINCIPAL_ID", "arn": "arn:aws:iam::111122223333:user/Alice", "accountId": "111122223333", "accessKeyId": "EXAMPLE_KEY_ID", "userName": "Alice" }, "eventTime": "2025-05-20T20:45:00Z", "eventSource": "kms.amazonaws.com", "eventName": "Decrypt", "awsRegion": "us-west-2", "sourceIPAddress": "192.0.2.0", "userAgent": "AWS Internal", "requestParameters": { "encryptionAlgorithm": "SYMMETRIC_DEFAULT", "keyId": "arn:aws:kms:us-west-2:111122223333:key/1234abcd-12ab-34cd-56ef-1234567890ab", "encryptionContext": { "Department": "Engineering", "Project": "Alpha" } }, "responseElements": null, "additionalEventData": { "keyMaterialId": "123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef0" }, "requestID": "12345126-30d5-4b28-98b9-9153da559963", "eventID": "abcde202-ba1a-467c-b4ba-f729d45ae521", "readOnly": true, "resources": [ { "accountId": "111122223333", "type": "AWS::KMS::Key", "ARN": "arn:aws:kms:us-west-2:111122223333:key/1234abcd-12ab-34cd-56ef-1234567890ab" } ], "eventType": "AwsApiCall", "managementEvent": true, "recipientAccountId": "111122223333", "eventCategory": "Management", "tlsDetails": { "tlsVersion": "TLSv1.2", "cipherSuite": "ECDHE-RSA-AES256-GCM-SHA384", "clientProvidedHostHeader": "kms.us-west-2.amazonaws.com" } } --- ct/decrypt_fail_as_printed.json (copied exactly as printed: a comma is missing after errorCode) --- { "eventVersion": "1.08", "userIdentity": { "type": "IAMUser", "principalId": "EX_PRINCIPAL_ID", "arn": "arn:aws:iam::111122223333:user/Alice", "accountId": "111122223333", "accessKeyId": "EXAMPLE_KEY_ID", "userName": "Alice" }, "eventTime": "2022-11-24T18:57:43Z", "eventSource": "kms.amazonaws.com", "eventName": "Decrypt", "awsRegion": "us-west-2", "sourceIPAddress": "192.0.2.0", "userAgent": "AWS Internal", "errorCode": "IncorrectKeyException" "errorMessage": "The key ID in the request does not identify a CMK that can perform this operation.", "requestParameters": { "encryptionAlgorithm": "SYMMETRIC_DEFAULT", "keyId": "arn:aws:kms:us-west-2:111122223333:key/1234abcd-12ab-34cd-56ef-1234567890ab", "encryptionContext": { "Department": "Engineering", "Project": "Alpha" } }, "responseElements": null, "requestID": "22345126-30d5-4b28-98b9-9153da559963", "eventID": "abcde202-ba1a-467c-b4ba-f729d45ae521", "readOnly": true, "resources": [ { "accountId": "111122223333", "type": "AWS::KMS::Key", "ARN": "arn:aws:kms:us-west-2:111122223333:key/1234abcd-12ab-34cd-56ef-1234567890ab" } ], "eventType": "AwsApiCall", "recipientAccountId": "111122223333" } --- ct/generatedatakey_as_printed.json (copied exactly as printed: a comma is missing after recipientAccountId) --- { "eventVersion": "1.11", "userIdentity": { "type": "IAMUser", "principalId": "EX_PRINCIPAL_ID", "arn": "arn:aws:iam::111122223333:user/Alice", "accountId": "111122223333", "accessKeyId": "EXAMPLE_KEY_ID", "userName": "Alice" }, "eventTime": "2025-05-20T20:46:16Z", "eventSource": "kms.amazonaws.com", "eventName": "GenerateDataKey", "awsRegion": "us-east-1", "sourceIPAddress": "192.0.2.0", "userAgent": "AWS Internal", "requestParameters": { "keyId": "1234abcd-12ab-34cd-56ef-1234567890ab", "keySpec": "AES_256", "encryptionContext": { "Department": "Engineering", "Project": "Alpha" } }, "responseElements": null, "additionalEventData": { "keyMaterialId": "123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef0" }, "requestID": "e0eb83e3-63bc-11e4-bc2b-4198b6150d5c", "eventID": "a9dea4f9-8395-46c0-942c-f509c02c2b71", "readOnly": true, "resources": [{ "ARN": "arn:aws:kms:us-west-2:111122223333:key/1234abcd-12ab-34cd-56ef-1234567890ab", "accountId": "111122223333" }], "eventType": "AwsApiCall", "managementEvent": true, "recipientAccountId": "111122223333" "eventCategory": "Management", "tlsDetails": { "tlsVersion": "TLSv1.3", "cipherSuite": "TLS_AES_256_GCM_SHA384", "clientProvidedHostHeader": "kms.us-east-1.amazonaws.com" } } --- ct_analyze.py --- """ct_analyze.py - answer audit questions from CloudTrail KMS events. Three events are the AWS KMS developer guide's own example log entries (files in ct/, copied from the pages read 2026-10-12; two of them are saved exactly 'as printed'). The other events are INVENTED by gen_synthetic() to give the analysis something to find; they are marked synthetic and follow the same field layout. No real CloudTrail data was available. """ import collections, json, pathlib, re D = pathlib.Path(r"C:\lwaws\c3\ct") def try_parse(name): text = (D / name).read_text(encoding="utf-8") try: return json.loads(text), None, text except json.JSONDecodeError as e: return None, e, text def repair(text, err): """insert the missing comma: the parser stopped at the start of the next key; put the comma after the previous non-space character""" i = err.pos j = i - 1 while text[j] in " \t\r\n": j -= 1 return text[:j + 1] + "," + text[j + 1:] def gen_synthetic(): key = "arn:aws:kms:us-west-2:111122223333:key/1234abcd-12ab-34cd-56ef-1234567890ab" key2 = "arn:aws:kms:us-west-2:111122223333:key/9999aaaa-99aa-99aa-99aa-999999999999" def ev(t, name, who, k, ctx=None, err=None, region="us-west-2"): e = {"eventTime": t, "eventSource": "kms.amazonaws.com", "eventName": name, "awsRegion": region, "userIdentity": who, "requestParameters": {"keyId": k, **({"encryptionContext": ctx} if ctx else {})}, "resources": [{"ARN": k, "type": "AWS::KMS::Key"}], "_synthetic": True} if err: e["errorCode"] = err return e app = {"type": "AssumedRole", "arn": "arn:aws:sts::111122223333:assumed-role/App/i-0abc"} batch = {"type": "AssumedRole", "arn": "arn:aws:sts::111122223333:assumed-role/Batch/job-7"} admin = {"type": "AssumedRole", "arn": "arn:aws:sts::111122223333:assumed-role/KeyAdmin/carol"} ctx = {"app": "invoices", "tenant": "t-17"} out = [] for h in range(8, 14): out.append(ev(f"2026-10-11T{h:02d}:00:00Z", "GenerateDataKey", app, key, ctx)) out.append(ev(f"2026-10-11T{h:02d}:05:00Z", "Decrypt", app, key, ctx)) out.append(ev("2026-10-11T14:00:00Z", "Decrypt", batch, key, {"app": "invoices", "tenant": "t-99"}, err="InvalidCiphertextException")) out.append(ev("2026-10-11T14:30:00Z", "Decrypt", batch, key2, {"app": "payroll"})) out.append(ev("2026-10-11T15:00:00Z", "ScheduleKeyDeletion", admin, key2)) out.append(ev("2026-10-11T17:10:00Z", "Decrypt", batch, key2, {"app": "payroll"}, err="KMSInvalidStateException")) out.append(ev("2026-10-11T17:40:00Z", "Decrypt", batch, key2, {"app": "payroll"}, err="KMSInvalidStateException")) out.append(ev("2026-10-11T18:00:00Z", "CancelKeyDeletion", admin, key2)) return out def key_of(e): for r in e.get("resources", []): if "ARN" in r: return r["ARN"] return e["requestParameters"].get("keyId") if __name__ == "__main__": print("1. the documentation's own example events: do they parse?") events = [] for name in ("decrypt_ok.json", "decrypt_fail_as_printed.json", "generatedatakey_as_printed.json"): obj, err, text = try_parse(name) if obj is not None: print(f" {name:<34} valid JSON as printed") else: line = text.count("\n", 0, err.pos) + 1 print(f" {name:<34} NOT valid JSON as printed: {err.msg} (line {err.lineno} column {err.colno}); the line before is {text.splitlines()[line - 2].strip()!r}") fixed = repair(text, err) obj = json.loads(fixed) print(f" {'':<34} after inserting one comma: valid, {len(obj)} top-level fields") events.append(obj) events += gen_synthetic() print(f"\n {len(events)} events in the analysis: 3 from the documentation, {len(events) - 3} synthetic") print("\n2. data-quality check on the documentation's events: does the key ARN's Region match the event's awsRegion?") for e in events[:3]: k = key_of(e) region = k.split(":")[3] if k and k.startswith("arn:") else "(bare key id)" flag = "" if region == e["awsRegion"] else " <-- differs" print(f" {e['eventName']:<16} awsRegion {e['awsRegion']:<10} key {region:<14}{flag}") print("\n3. who used which key (events with a principal and a key; counts per principal)") by = collections.Counter((e["userIdentity"]["arn"].replace("arn:aws:sts::111122223333:assumed-role/", "role ").replace("arn:aws:iam::111122223333:user/", "user "), key_of(e)[-12:], e["eventName"]) for e in events) for (who, k, name), n in sorted(by.items()): print(f" {who:<22} key ...{k} {name:<20} {n:>3}") print("\n4. failures (errorCode) and what the request said") for e in events: if "errorCode" in e: ctx = e["requestParameters"].get("encryptionContext") print(f" {e['eventTime']} {e['eventName']:<10} {e['userIdentity']['arn'].split('/')[-2] if 'assumed' in e['userIdentity']['arn'] else e['userIdentity']['arn'].split('/')[-1]:<8} {e['errorCode']:<28} context {ctx}") print("\n5. the encryption context is plain text in the log: values that appear") vals = collections.defaultdict(set) for e in events: for k, v in (e["requestParameters"].get("encryptionContext") or {}).items(): vals[k].add(v) for k, v in sorted(vals.items()): print(f" {k}: {sorted(v)}") print("\n6. the documented alarm idea: any attempt to use a key that is pending deletion") sched = {key_of(e): e["eventTime"] for e in events if e["eventName"] == "ScheduleKeyDeletion"} cancel = {key_of(e): e["eventTime"] for e in events if e["eventName"] == "CancelKeyDeletion"} for k, t0 in sched.items(): t1 = cancel.get(k, "9999") hits = [e for e in events if key_of(e) == k and t0 < e["eventTime"] < t1 and e["eventName"] not in ("ScheduleKeyDeletion", "CancelKeyDeletion")] print(f" key ...{k[-12:]} scheduled for deletion {t0}, cancelled {cancel.get(k, '(not cancelled)')}: {len(hits)} attempted uses in between") for e in hits: print(f" {e['eventTime']} {e['eventName']} by {e['userIdentity']['arn'].split('/')[-2]} -> {e.get('errorCode', 'ok')}") --- py -X utf8 ct_analyze.py --- 1. the documentation's own example events: do they parse? decrypt_ok.json valid JSON as printed decrypt_fail_as_printed.json NOT valid JSON as printed: Expecting ',' delimiter (line 18 column 5); the line before is '"errorCode": "IncorrectKeyException"' after inserting one comma: valid, 18 top-level fields generatedatakey_as_printed.json NOT valid JSON as printed: Expecting ',' delimiter (line 39 column 5); the line before is '"recipientAccountId": "111122223333"' after inserting one comma: valid, 20 top-level fields 21 events in the analysis: 3 from the documentation, 18 synthetic 2. data-quality check on the documentation's events: does the key ARN's Region match the event's awsRegion? Decrypt awsRegion us-west-2 key us-west-2 Decrypt awsRegion us-west-2 key us-west-2 GenerateDataKey awsRegion us-east-1 key us-west-2 <-- differs 3. who used which key (events with a principal and a key; counts per principal) role App/i-0abc key ...1234567890ab Decrypt 6 role App/i-0abc key ...1234567890ab GenerateDataKey 6 role Batch/job-7 key ...1234567890ab Decrypt 1 role Batch/job-7 key ...999999999999 Decrypt 3 role KeyAdmin/carol key ...999999999999 CancelKeyDeletion 1 role KeyAdmin/carol key ...999999999999 ScheduleKeyDeletion 1 user Alice key ...1234567890ab Decrypt 2 user Alice key ...1234567890ab GenerateDataKey 1 4. failures (errorCode) and what the request said 2022-11-24T18:57:43Z Decrypt Alice IncorrectKeyException context {'Department': 'Engineering', 'Project': 'Alpha'} 2026-10-11T14:00:00Z Decrypt Batch InvalidCiphertextException context {'app': 'invoices', 'tenant': 't-99'} 2026-10-11T17:10:00Z Decrypt Batch KMSInvalidStateException context {'app': 'payroll'} 2026-10-11T17:40:00Z Decrypt Batch KMSInvalidStateException context {'app': 'payroll'} 5. the encryption context is plain text in the log: values that appear Department: ['Engineering'] Project: ['Alpha'] app: ['invoices', 'payroll'] tenant: ['t-17', 't-99'] 6. the documented alarm idea: any attempt to use a key that is pending deletion key ...999999999999 scheduled for deletion 2026-10-11T15:00:00Z, cancelled 2026-10-11T18:00:00Z: 2 attempted uses in between 2026-10-11T17:10:00Z Decrypt by Batch -> KMSInvalidStateException 2026-10-11T17:40:00Z Decrypt by Batch -> KMSInvalidStateException WHY THIS WORKS AS AN ANSWER --------------------------- The rotation model follows the four documented steps and labels (AWSCURRENT, AWSPENDING, AWSPREVIOUS), and each behaviour is tied to a sentence on the Lambda rotation functions or Quotas page: the idempotent createSecret, the confused-deputy checks in setSecret (same user, same host, AWSCURRENT valid), AWSPREVIOUS added by finishSecret, a stale AWSPENDING blocking the next rotation, the 65,536-byte value and 100-version limits. Two items are marked 'my reading': a same-token retry resumes the rotation, and a put that cannot make room under the version quota fails. The window between setSecret and finishSecret in which AWSCURRENT is wrong is derived from the steps, not quoted. The size table is a real measurement of PEM text: a 4,096-bit RSA key (3,272 bytes) fits Parameter Store standard but the same key with a three-certificate chain (6,335 bytes) does not. The cost arithmetic shows why you cache: the same 25 secrets cost about $3,250 a month read on every request at 10 requests per second and about $32 with a five-minute cache in 20 instances. The leak experiments are real child processes: an environment variable reaches every child and grandchild; a diagnostic dump leaks it, and redaction by variable name misses a password inside DATABASE_URL; a command-line argument is readable by another process (Get-CimInstance Win32_Process); a file path or a pipe keeps the secret out of both (but the file still needs protecting). CloudTrail: two of the three examples printed in the documentation are not valid JSON as printed (a missing comma each); the analysis repairs them minimally and then answers who used which key, which calls failed, which encryption-context values appear in clear text, and which attempts hit a key pending deletion. The GenerateDataKey example also names us-east-1 with a us-west-2 key ARN. Not run: Secrets Manager, Parameter Store, a rotation Lambda, real CloudTrail data, docker inspect (engine not running), Lambda/ECS environment settings. Not read: alternating-users and managed rotation, SecureString details, ACM validity and renewal.