aws2-4 Exercise 3: Change Set Approximation, Failure Simulation, Capabilities, Quotas and Drift ================================================================================================== Run with Python 3.14.0 on Windows 11 in a scratch venv: cfn-lint 1.57.2, cfn-flip 1.3.0, PyYAML 6.0.3. Setup: py -m venv .venv ; .venv/Scripts/python.exe -m pip install cfn-lint cfn-flip pyyaml. VERIFICATION LEVEL: 1 (run here; no AWS account, no AWS CLI, no stack created). cfn-resolve, the change set approximation, the failure simulator, the capability and quota checkers are MODELS of CloudFormation documentation pages read on 2026-10-12 and 2026-10-13 (through a page-fetch tool that summarises pages): they are not AWS, and AWS may change its rules and limits. cfn_resolve.py is the file listed in Exercise 2 and mutate_lint.py the file in Exercise 1; copy them next to these scripts. --- changeset_model.py --- """changeset_model.py - an APPROXIMATION of what a CloudFormation change set would show, computed locally. NOT AWS and not a substitute for one. What is real: the two templates are rendered by cfn_resolve.py (a model) and the 'create-only properties' come from the resource schemas bundled inside cfn-lint (installed version printed), which are the CloudFormation resource-provider schemas. Rules, from the 'Update CloudFormation stacks using change sets' and 'View a change set' pages read 2026-10-12: * Add: a resource that is in the new template and not the old. Remove: in the old and not the new. Modify: its properties changed * the change set shows property-level changes (path, before, after) * a modification "can cause the resource to be interrupted or replaced (recreated)" My approximation of the Replacement field: True when a changed property is a create-only property in the schema, Conditional when it is a conditional create-only property, False otherwise. A False here can still be an INTERRUPTION (the schema does not record those), and a change that depends on a value only AWS knows (a Ref or GetAtt to another resource) is reported as 'unknown value'. Not modelled: Dynamic changes in nested stacks, resource imports, Transform, drift-aware change sets, UpdateReplacePolicy and DeletionPolicy effects. """ import json, pathlib import cfnlint from cfn_resolve import Template, parse_template, Unresolved SCHEMAS = pathlib.Path(cfnlint.__file__).parent / "data" / "schemas" _PROV = json.loads((SCHEMAS / "providers" / "us-east-1.json").read_text(encoding="utf-8")) def schema(rtype): h = _PROV.get(rtype) return json.loads((SCHEMAS / "resources" / f"{h}.json").read_text(encoding="utf-8")) if h else None def top(paths): return {p.split("/")[2] for p in (paths or []) if p.startswith("/properties/")} def walk(a, b, path=""): """property-level differences: list of (path, kind, before, after)""" out = [] if isinstance(a, dict) and isinstance(b, dict): for k in sorted(set(a) | set(b)): if k not in a: out.append((f"{path}/{k}", "Add", None, b[k])) elif k not in b: out.append((f"{path}/{k}", "Remove", a[k], None)) else: out += walk(a[k], b[k], f"{path}/{k}") elif isinstance(a, list) and isinstance(b, list) and len(a) == len(b): for i, (x, y) in enumerate(zip(a, b)): out += walk(x, y, f"{path}/{i}") elif a != b: out.append((path, "Modify", a, b)) return out def has_unknown(v): if isinstance(v, Unresolved): return True if isinstance(v, dict): return any(has_unknown(x) for x in v.values()) if isinstance(v, list): return any(has_unknown(x) for x in v) return isinstance(v, str) and (" {new[name]['Type']}", "Replacement": "(not modelled)", "Details": []}) continue d = walk(old[name]["Properties"], new[name]["Properties"], "/Properties") if not d: continue s = schema(old[name]["Type"]) or {} co, cco = top(s.get("createOnlyProperties")), top(s.get("conditionalCreateOnlyProperties")) repl = "False" for path, kind, b, a in d: prop = path.split("/")[2] if prop in co: repl = "True" elif prop in cco and repl != "True": repl = "Conditional" rows.append({"Action": "Modify", "Id": name, "Type": old[name]["Type"], "Replacement": repl, "Details": d}) return rows def show(title, rows): print() print(f"-- {title.strip()}") if not rows: print(" (no changes)") for r in rows: print(f" {r['Action']:<7}{r['Id']:<22}{r['Type']:<26}Replacement: {r['Replacement']}") for path, kind, b, a in r["Details"]: fmt = lambda v: "" if has_unknown(v) else json.dumps(v)[:40] print(f" {kind:<7}{path:<62}{fmt(b)} -> {fmt(a)}") def render(text, params): return Template(parse_template(text), params).render() if __name__ == "__main__": base = pathlib.Path(r"C:/lwaws/c4/app_stack.yaml").read_text(encoding="utf-8") P = {"Environment": "dev", "RetentionDays": "30", "KeyAdminRoleArn": "arn:aws:iam::111122223333:role/KeyAdmin"} import importlib.metadata as md; cfn_ver = md.version("cfn-lint") print(f"schemas from cfn-lint {cfn_ver}\n") old = render(base, P)["Resources"] show("1. parameter change: Environment dev -> prod", changes(old, render(base, dict(P, Environment="prod"))["Resources"])) show("\n2. parameter change: RetentionDays 30 -> 90", changes(old, render(base, dict(P, RetentionDays="90"))["Resources"])) show("\n3. template edit: give the bucket a fixed BucketName", changes(old, render(base.replace(" BucketEncryption:", " BucketName: acme-reports-prod\n BucketEncryption:", 1), P)["Resources"])) show("\n4. template edit: give the work queue a fixed QueueName", changes(old, render(base.replace(" WorkQueue:\n Type: AWS::SQS::Queue\n Properties:\n", " WorkQueue:\n Type: AWS::SQS::Queue\n Properties:\n QueueName: work\n", 1), P)["Resources"])) import re only_bucket = re.sub(r"(? prod Add BacklogAlarm AWS::CloudWatch::Alarm Replacement: - Modify DataKey AWS::KMS::Key Replacement: False Modify /Properties/Description "Data key for demo-stack (dev)" -> "Data key for demo-stack (prod)" Modify ReportBucket AWS::S3::Bucket Replacement: False Modify /Properties/Tags/0/Value "dev" -> "prod" Modify WorkQueue AWS::SQS::Queue Replacement: False Modify /Properties/RedrivePolicy/maxReceiveCount 3 -> 5 Modify /Properties/VisibilityTimeout 30 -> 120 -- 2. parameter change: RetentionDays 30 -> 90 Modify ReportBucket AWS::S3::Bucket Replacement: False Modify /Properties/LifecycleConfiguration/Rules/0/ExpirationInDays "30" -> "90" -- 3. template edit: give the bucket a fixed BucketName Modify ReportBucket AWS::S3::Bucket Replacement: True Add /Properties/BucketName null -> "acme-reports-prod" -- 4. template edit: give the work queue a fixed QueueName Modify WorkQueue AWS::SQS::Queue Replacement: True Add /Properties/QueueName null -> "work" -- 5. template edit: rename only the bucket's logical ID, ReportBucket to Reports (the bucket policy still points at it) Remove ReportBucket AWS::S3::Bucket Replacement: - Modify ReportBucketPolicy AWS::S3::BucketPolicy Replacement: True Modify /Properties/Bucket -> Modify /Properties/PolicyDocument/Statement/0/Resource/0 -> Modify /Properties/PolicyDocument/Statement/0/Resource/1 -> Add Reports AWS::S3::Bucket Replacement: - Modify WorkerRole AWS::IAM::Role Replacement: False Modify /Properties/Policies/0/PolicyDocument/Statement/1/Resource -> -- 6. parameter change: a different key administrator role Modify DataKey AWS::KMS::Key Replacement: False Modify /Properties/KeyPolicy/Statement/1/Principal/AWS "arn:aws:iam::111122223333:role/KeyAdmin -> "arn:aws:iam::111122223333:role/OtherAdm create-only properties in the bundled schemas (a change to one of these means Replacement: True in this model) AWS::S3::Bucket ['BucketName', 'BucketNamePrefix', 'BucketNamespace'] AWS::SQS::Queue ['FifoQueue', 'QueueName'] AWS::KMS::Key (none listed) AWS::KMS::Alias ['AliasName'] AWS::IAM::Role ['Path', 'RoleName'] AWS::CloudWatch::Alarm ['AlarmName'] AWS::S3::BucketPolicy ['Bucket'] --- stack_graph.py --- """stack_graph.py - three local checks on a template: (1) the resource dependency graph and a MODEL of how a failed deployment stops; (2) which IAM capability a deployment needs; (3) the documented size and count quotas. NOT AWS. Rules from the CloudFormation pages read 2026-10-12: * Stack failure options: "CloudFormation identifies dependencies between resources to parallelize independent provisioning actions", "stops at the first failure in each independent provisioning path", "a failure in one path doesn't affect other provisioning paths"; with the option to preserve successfully provisioned resources the successful ones stay (the default, rolling everything back, is not modelled in detail) * aws cloudformation deploy: --capabilities takes CAPABILITY_IAM and CAPABILITY_NAMED_IAM; "If you have IAM resources, you can specify either capability. If you have IAM resources with custom names, you must specify CAPABILITY_NAMED_IAM"; without it: an InsufficientCapabilities error * quotas page: 500 resources, 200 parameters, 200 outputs, 200 mappings, template description 1,024 bytes, template body 51,200 bytes in a request and 1 MB from an S3 object, parameter value 4,096 bytes My own list (the docs say only 'IAM resources with custom names'): the name properties checked are RoleName, UserName, GroupName, InstanceProfileName, ManagedPolicyName. """ import json, pathlib, re from cfn_resolve import Template, parse_template NAME_PROPS = {"AWS::IAM::Role": "RoleName", "AWS::IAM::User": "UserName", "AWS::IAM::Group": "GroupName", "AWS::IAM::InstanceProfile": "InstanceProfileName", "AWS::IAM::ManagedPolicy": "ManagedPolicyName"} def refs_in(node, resources): """logical IDs of resources referenced by Ref, GetAtt and Sub variables inside a node""" found = set() def sub_vars(s): for m in re.finditer(r"\$\{([^}!][^}]*)\}", s): name = m.group(1).split(".")[0] if name in resources: found.add(name) def rec(n): if isinstance(n, dict): for k, v in n.items(): if k == "Ref" and isinstance(v, str) and v in resources: found.add(v) elif k == "Fn::GetAtt": found.add((v[0] if isinstance(v, list) else v.split(".")[0])) if (v[0] if isinstance(v, list) else v.split(".")[0]) in resources else None elif k == "Fn::Sub": sub_vars(v if isinstance(v, str) else v[0]) if isinstance(v, list): rec(v[1]) else: rec(v) elif isinstance(n, list): for x in n: rec(x) rec(node) return found def graph(data, active=None): res = data["Resources"] names = set(res) if active is None else set(active) g = {} for n in names: deps = refs_in(res[n].get("Properties", {}), names) | refs_in(res[n].get("Metadata", {}), names) d = res[n].get("DependsOn", []) deps |= {x for x in ([d] if isinstance(d, str) else d) if x in names} deps.discard(n) g[n] = deps return g def cycle(g): state = {} def visit(n, path): if state.get(n) == 1: return path[path.index(n):] + [n] if state.get(n) == 2: return None state[n] = 1 for m in sorted(g[n]): c = visit(m, path + [n]) if c: return c state[n] = 2 return None for n in sorted(g): c = visit(n, []) if c: return c return None def waves(g): done, out = set(), [] while len(done) < len(g): w = sorted(n for n in g if n not in done and g[n] <= done) if not w: raise ValueError("cycle") out.append(w) done |= set(w) return out def deploy(g, failing): """preserve-successful-resources model: created / FAILED / not attempted""" status = {} for w in waves(g): for n in w: if any(status[d] != "created" for d in g[n]): status[n] = "not attempted" else: status[n] = "FAILED" if n in failing else "created" return status def capabilities(data): iam = [n for n, r in data["Resources"].items() if r["Type"].startswith("AWS::IAM::")] named = [n for n in iam if NAME_PROPS.get(data["Resources"][n]["Type"]) in data["Resources"][n].get("Properties", {})] if named: return "CAPABILITY_NAMED_IAM", iam return ("CAPABILITY_IAM" if iam else "(none)"), iam def limits(text, data): n = lambda k: len(data.get(k, {})) checks = [("resources", n("Resources"), 500), ("parameters", n("Parameters"), 200), ("outputs", n("Outputs"), 200), ("mappings", n("Mappings"), 200), ("description bytes", len(str(data.get("Description", "")).encode()), 1024), ("body bytes (direct request)", len(text.encode()), 51200), ("body bytes (S3 object, 1 MB)", len(text.encode()), 1024 * 1024)] return [(name, v, mx, v > mx) for name, v, mx in checks] if __name__ == "__main__": B = pathlib.Path(r"C:/lwaws/c4") text = (B / "app_stack.yaml").read_text(encoding="utf-8") P = {"Environment": "prod", "KeyAdminRoleArn": "arn:aws:iam::111122223333:role/KeyAdmin"} data = parse_template(text) active = Template(data, P).render()["Resources"].keys() g = graph(data, active) print("1. dependency graph of app_stack.yaml (Environment=prod: BacklogAlarm exists)") for n in sorted(g): print(f" {n:<22} depends on {sorted(g[n]) or '-'}") print(" waves (each wave can be provisioned in parallel once the earlier waves are done):") for i, w in enumerate(waves(g), 1): print(f" wave {i}: {', '.join(w)}") print(f" cycle in the real template: {cycle(g)}") bad = parse_template(text.replace('AWS: !Sub "arn:${AWS::Partition}:iam::${AWS::AccountId}:root"', "AWS: !GetAtt WorkerRole.Arn")) gb = graph(bad, active) print(f" cycle after letting the key policy name the role: {cycle(gb)} (cfn-lint said E3004: Circular dependency with [WorkerRole])") print("\n2. a failed deployment, preserve-successful-resources model (which resources get created, fail, or are never attempted)") for failing in ({"DataKey"}, {"WorkDeadLetterQueue"}, {"ReportBucket"}, {"WorkerRole"}, {"BacklogAlarm"}): s = deploy(g, failing) print(f" {next(iter(failing)):<20} fails -> created: {sorted(k for k, v in s.items() if v == 'created')}") print(f" {'':<20} not attempted: {sorted(k for k, v in s.items() if v == 'not attempted')}") print("\n3. which capability a deployment needs") cap, iam = capabilities(data) print(f" app_stack.yaml: {cap} (IAM resources: {iam})") named = parse_template(text.replace(" WorkerRole:\n Type: AWS::IAM::Role\n Properties:\n", " WorkerRole:\n Type: AWS::IAM::Role\n Properties:\n RoleName: work-role\n")) print(f" same template with RoleName: work-role: {capabilities(named)[0]}") scp = parse_template((B / "doc" / "ch1_scp_stack.yaml").read_text(encoding="utf-8")) print(f" Chapter 1's SCP template (AWS::Organizations::Policy): {capabilities(scp)[0]}") print("\n4. the documented quotas, applied to the template and to the cfn-lint experiment's generated templates") import mutate_lint as ml for label, t in (("app_stack.yaml", text), ) + tuple((l, tx) for l, tx in ml.LIMITS): d = parse_template(t) over = [f"{name} {v:,} > {mx:,}" for name, v, mx, bad_ in limits(t, d) if bad_] print(f" {label[:78]:<80} {'; '.join(over) or 'within every limit'}") --- py -X utf8 stack_graph.py --- 1. dependency graph of app_stack.yaml (Environment=prod: BacklogAlarm exists) BacklogAlarm depends on ['WorkQueue'] DataKey depends on - DataKeyAlias depends on ['DataKey'] ReportBucket depends on ['DataKey'] ReportBucketPolicy depends on ['ReportBucket'] WorkDeadLetterQueue depends on ['DataKey'] WorkQueue depends on ['DataKey', 'WorkDeadLetterQueue'] WorkerRole depends on ['DataKey', 'ReportBucket', 'WorkQueue'] waves (each wave can be provisioned in parallel once the earlier waves are done): wave 1: DataKey wave 2: DataKeyAlias, ReportBucket, WorkDeadLetterQueue wave 3: ReportBucketPolicy, WorkQueue wave 4: BacklogAlarm, WorkerRole cycle in the real template: None cycle after letting the key policy name the role: ['DataKey', 'WorkerRole', 'DataKey'] (cfn-lint said E3004: Circular dependency with [WorkerRole]) 2. a failed deployment, preserve-successful-resources model (which resources get created, fail, or are never attempted) DataKey fails -> created: [] not attempted: ['BacklogAlarm', 'DataKeyAlias', 'ReportBucket', 'ReportBucketPolicy', 'WorkDeadLetterQueue', 'WorkQueue', 'WorkerRole'] WorkDeadLetterQueue fails -> created: ['DataKey', 'DataKeyAlias', 'ReportBucket', 'ReportBucketPolicy'] not attempted: ['BacklogAlarm', 'WorkQueue', 'WorkerRole'] ReportBucket fails -> created: ['BacklogAlarm', 'DataKey', 'DataKeyAlias', 'WorkDeadLetterQueue', 'WorkQueue'] not attempted: ['ReportBucketPolicy', 'WorkerRole'] WorkerRole fails -> created: ['BacklogAlarm', 'DataKey', 'DataKeyAlias', 'ReportBucket', 'ReportBucketPolicy', 'WorkDeadLetterQueue', 'WorkQueue'] not attempted: [] BacklogAlarm fails -> created: ['DataKey', 'DataKeyAlias', 'ReportBucket', 'ReportBucketPolicy', 'WorkDeadLetterQueue', 'WorkQueue', 'WorkerRole'] not attempted: [] 3. which capability a deployment needs app_stack.yaml: CAPABILITY_IAM (IAM resources: ['WorkerRole']) same template with RoleName: work-role: CAPABILITY_NAMED_IAM Chapter 1's SCP template (AWS::Organizations::Policy): (none) 4. the documented quotas, applied to the template and to the cfn-lint experiment's generated templates app_stack.yaml within every limit 501 resources (documented maximum is 500) resources 501 > 500 500 resources (at the maximum) within every limit 201 parameters (documented maximum is 200) parameters 201 > 200 201 outputs (documented maximum is 200) outputs 201 > 200 description of 1,100 bytes (documented maximum is 1,024) description bytes 1,238 > 1,024 template body of about 60,000 bytes (documented direct-request maximum is 51,2 body bytes (direct request) 62,191 > 51,200 --- drift_demo.py --- """drift_demo.py - reproduce the property differences in the documentation's own drift-detection example, and show how little a comparison can see. The example is on the 'Detect drift on an entire CloudFormation stack' page (read 2026-10-12): an AWS::SQS::Queue whose ExpectedProperties and ActualProperties are printed as JSON strings, with the documented result: two PropertyDifferences, /DelaySeconds (actual 120, expected 20) and /RedrivePolicy/maxReceiveCount (actual 12, expected 10), difference type NOT_EQUAL. This script only COMPARES two JSON documents the way the page describes. It cannot detect drift in a real account: detect-stack-drift reads the real resources. """ import json EXPECTED = '{"ReceiveMessageWaitTimeSeconds":0,"DelaySeconds":20,"RedrivePolicy":{"deadLetterTargetArn":"arn:aws:sqs:us-east-1:099908667365:my-stack-with-resource-drift-DLQ-1BCY7HHD5QIM3","maxReceiveCount":10},"MessageRetentionPeriod":345600,"MaximumMessageSize":262144,"VisibilityTimeout":60,"QueueName":"my-stack-with-resource-drift-Queue-494PBHCO76H4"}' ACTUAL = '{"ReceiveMessageWaitTimeSeconds":0,"DelaySeconds":120,"RedrivePolicy":{"deadLetterTargetArn":"arn:aws:sqs:us-east-1:099908667365:my-stack-with-resource-drift-DLQ-1BCY7HHD5QIM3","maxReceiveCount":12},"MessageRetentionPeriod":345600,"MaximumMessageSize":262144,"VisibilityTimeout":60,"QueueName":"my-stack-with-resource-drift-Queue-494PBHCO76H4"}' DOCUMENTED = [("/DelaySeconds", "120", "20", "NOT_EQUAL"), ("/RedrivePolicy/maxReceiveCount", "12", "10", "NOT_EQUAL")] def differences(expected, actual, path=""): out = [] if isinstance(expected, dict) and isinstance(actual, dict): for k in sorted(set(expected) | set(actual)): if k not in actual: out.append((f"{path}/{k}", None, str(expected[k]), "REMOVE")) # my extension: the page's example shows only NOT_EQUAL elif k not in expected: out.append((f"{path}/{k}", str(actual[k]), None, "ADD")) else: out += differences(expected[k], actual[k], f"{path}/{k}") elif expected != actual: out.append((path, str(actual), str(expected), "NOT_EQUAL")) return out if __name__ == "__main__": got = differences(json.loads(EXPECTED), json.loads(ACTUAL)) print("computed from the page's ExpectedProperties and ActualProperties:") for path, a, e, kind in got: print(f" {path:<34} actual {a:<6} expected {e:<6} {kind}") print("matches the PropertyDifferences printed on the page:", got == DOCUMENTED) print("") print("what else a comparison like this reports (on top of the two documented differences)") exp, act = json.loads(EXPECTED), json.loads(ACTUAL) base = set(differences(exp, act)) extra = lambda a: [(p, k) for p, x, e, k in differences(exp, a) if (p, x, e, k) not in base] print(" a property added by hand to the live resource :", extra(dict(act, Tags=[{"Key": "added-by-hand", "Value": "1"}]))) print(" a property missing from the live resource :", extra({k: v for k, v in act.items() if k != "VisibilityTimeout"})) print(" the expected document compared with itself :", differences(exp, exp), "(no drift)") print("") print("ADD and REMOVE are my extension; the page's example prints only NOT_EQUAL. The page says drift detection reports on 'each resource in the stack that supports drift detection'.") --- py -X utf8 drift_demo.py --- computed from the page's ExpectedProperties and ActualProperties: /DelaySeconds actual 120 expected 20 NOT_EQUAL /RedrivePolicy/maxReceiveCount actual 12 expected 10 NOT_EQUAL matches the PropertyDifferences printed on the page: True what else a comparison like this reports (on top of the two documented differences) a property added by hand to the live resource : [('/Tags', 'ADD')] a property missing from the live resource : [('/VisibilityTimeout', 'REMOVE')] the expected document compared with itself : [] (no drift) ADD and REMOVE are my extension; the page's example prints only NOT_EQUAL. The page says drift detection reports on 'each resource in the stack that supports drift detection'. WHY THIS WORKS AS AN ANSWER --------------------------- The change set approximation does what a change set is documented to do (Add, Remove, Modify, with property-level before and after) and adds the one judgement a reviewer needs most, whether an edit replaces a resource, using the create-only properties from the resource schemas bundled in cfn-lint. The six edits show the lessons: a parameter change modifies in place; adding BucketName or QueueName to an existing resource is a replacement; renaming a logical ID is a remove plus an add and replaces the bucket policy that pointed at it. Its limits are stated: it cannot see interruptions, the meanings of Replacement True and Conditional are mine, and the schema lists no create-only properties for AWS::KMS::Key. The failure simulator applies the documented rules (independent paths continue, dependents of a failure are not attempted, preserve-successful-resources keeps what worked) to the dependency graph extracted from the template, and its cycle finder agrees with cfn-lint's E3004. The capability checker implements the documented IAM rules with my own list of name properties. The quota checker uses the documented numbers and agrees with the cfn-lint experiment, including the two quotas cfn-lint does not check. The drift comparison reproduces the page's own PropertyDifferences exactly; ADD and REMOVE are my extension. Not run: a real change set, rollback, rollback triggers, drift detection, nested stacks, StackSets, UpdateReplacePolicy behaviour (page not read).