Setting the file. One moment. 31
Prompt Library For Startups
81
Creating Amazon Aurora Db Cluster With Instances
104
Routing Traffic With Route53 And CloudFront
Resilience Program Design
Creating API Gateway Stage
def test_heroku_eb_bad_elasticache_and_http_forward_fails
— line 159
This file
- Number
- 33.3
- Position
- 3 of 38
- Type
- Python
- Size
- 34 KB
- Lines
- 844
scripts/test_validate_terraform_policy.py
Python·844 lines·34 KB
__file__
).resolve().parents[
1
]
13SCRIPT = PLUGIN_SKILL_ROOT / "scripts" / "validate-terraform-policy.py"
14FIXTURES = PLUGIN_SKILL_ROOT / "fixtures" / "terraform-policy"
15GOOD_FIXTURE = FIXTURES / "good-https-redirect"
16BAD_HTTP_FORWARD = FIXTURES / "bad-http-forward"
17INTERNAL_ALB = FIXTURES / "internal-alb-only"
18BAD_RDS_PUBLIC_UNENCRYPTED = FIXTURES / "bad-rds-public-unencrypted"
19GOOD_RDS_PRIVATE_ENCRYPTED = FIXTURES / "good-rds-private-encrypted"
20BAD_DB_SG_PUBLIC = FIXTURES / "bad-db-sg-public"
21GOOD_DB_SG_SCOPED = FIXTURES / "good-db-sg-scoped"
22BAD_WILDCARD_IAM = FIXTURES / "bad-wildcard-iam"
23GOOD_SCOPED_IAM = FIXTURES / "good-scoped-iam"
24BAD_WILDCARD_IAM_LISTFORM = FIXTURES / "bad-wildcard-iam-listform"
25GOOD_IAM_SCOPED_LIST = FIXTURES / "good-iam-scoped-list"
26GOOD_INTERNET_NLB = FIXTURES / "good-internet-nlb"
27BAD_SG_PUBLIC_SSH = FIXTURES / "bad-sg-public-ssh"
28GOOD_SG_PUBLIC_WEBAPP = FIXTURES / "good-sg-public-webapp"
29BAD_SG_PUBLIC_IPV6 = FIXTURES / "bad-sg-public-ipv6"
30GOOD_SG_IPV6_SCOPED = FIXTURES / "good-sg-ipv6-scoped"
31BAD_ELASTICACHE_UNENCRYPTED = FIXTURES / "bad-elasticache-unencrypted"
32GOOD_ELASTICACHE_ENCRYPTED = FIXTURES / "good-elasticache-encrypted"
33BAD_ELASTICACHE_CLUSTER_REDIS_UNENCRYPTED = (
34 FIXTURES / "bad-elasticache-cluster-redis-unencrypted"
35)
36GOOD_ELASTICACHE_CLUSTER_REDIS_ENCRYPTED = (
37 FIXTURES / "good-elasticache-cluster-redis-encrypted"
38)
39GOOD_ELASTICACHE_CLUSTER_MEMCACHED = FIXTURES / "good-elasticache-cluster-memcached"
40GOOD_QUOTED_PORT_HTTPS = FIXTURES / "good-quoted-port-https"
41BAD_DB_SG_PUBLIC_QUOTED_PORT = FIXTURES / "bad-db-sg-public-quoted-port"
42GOOD_HEROKU_EB_LB = FIXTURES / "good-heroku-eb-loadbalanced"
43GOOD_HEROKU_EB_ONLY = FIXTURES / "good-heroku-eb-only"
44GOOD_HEROKU_EB_SINGLEINSTANCE = FIXTURES / "good-heroku-eb-singleinstance"
45GOOD_HEROKU_EB_CODEPIPELINE = FIXTURES / "good-heroku-eb-codepipeline"
46BAD_HEROKU_EB_ELASTICACHE = FIXTURES / "bad-heroku-eb-elasticache"
47
48
49def _load_validator_module():
50 """Import the validator for unit-level tests (its filename is hyphenated, so
51 it is not importable as a normal module name)."""
52 spec = importlib.util.spec_from_file_location("validate_terraform_policy", SCRIPT)
53 assert spec is not None and spec.loader is not None
54 module = importlib.util.module_from_spec(spec)
55 # Register before exec: @dataclass resolves annotations via sys.modules.
56 sys.modules[spec.name] = module
57 spec.loader.exec_module(module)
58 return module
59
60
61VALIDATOR = _load_validator_module()
62
63
64def run_policy_validator(terraform_dir: Path, json_out: Path | None = None) -> tuple[int, str]:
65 cmd = [sys.executable, str(SCRIPT), str(terraform_dir)]
66 if json_out is not None:
67 cmd += ["--json", str(json_out)]
68 result = subprocess.run(cmd, capture_output=True, text=True) # nosec B603
69 return result.returncode, result.stdout + result.stderr
70
71
72def test_good_https_redirect_passes() -> None:
73 code, out = run_policy_validator(GOOD_FIXTURE)
74 assert code == 0, out
75 assert "POLICY_OK" in out
76
77
78def test_bad_http_forward_fails() -> None:
79 code, out = run_policy_validator(BAD_HTTP_FORWARD)
80 assert code == 1, out
81 assert "POLICY_FAIL" in out
82 assert "redirect" in out.lower()
83
84
85def test_heroku_eb_loadbalanced_passes() -> None:
86 # EB compute fronted by a STANDALONE ALB (aws_lb_listener present) + RDS. Web
87 # 80/443 open to 0.0.0.0/0 on the ALB is a legitimate web pattern and must NOT
88 # be flagged; only admin/datastore ports are.
89 code, out = run_policy_validator(GOOD_HEROKU_EB_LB)
90 assert code == 0, out
91 assert "POLICY_OK" in out
92
93
94def test_heroku_eb_only_passes_vacuously() -> None:
95 # Pure EB LoadBalanced env, no standalone aws_lb — EB provisions the ALB from
96 # setting blocks the static checker cannot read, so the ALB rules pass with no
97 # listener to inspect. Documents that EB-managed ALB posture is authoring-only.
98 code, out = run_policy_validator(GOOD_HEROKU_EB_ONLY)
99 assert code == 0, out
100 assert "POLICY_OK" in out
101
102
103def test_heroku_eb_singleinstance_passes() -> None:
104 # EB SingleInstance: public-subnet instance, 80/443 open to the world, no ALB.
105 # Web ports are not flagged (only admin/datastore ports are) → POLICY_OK.
106 code, out = run_policy_validator(GOOD_HEROKU_EB_SINGLEINSTANCE)
107 assert code == 0, out
108 assert "POLICY_OK" in out
109
110
111def test_heroku_eb_codepipeline_required_wildcard_passes() -> None:
112 # AWS does not support resource-level permissions for CreateStorageLocation;
113 # its isolated Resource "*" statement is required, not an over-broad grant.
114 code, out = run_policy_validator(GOOD_HEROKU_EB_CODEPIPELINE)
115 assert code == 0, out
116 assert "POLICY_OK" in out
117
118
119def test_required_wildcard_action_cannot_hide_other_actions() -> None:
120 with tempfile.TemporaryDirectory() as tmp:
121 terraform_dir = Path(tmp)
122 (terraform_dir / "policy.tf").write_text(
123 '''
124resource "aws_iam_role_policy" "mixed" {
125 role = "example-role"
126 policy = jsonencode({
127 Statement = [{
128 Effect = "Allow"
129 Action = ["elasticbeanstalk:CreateStorageLocation", "s3:*"]
130 Resource = "*"
131 }]
132 })
133}
134'''
135 )
136 code, out = run_policy_validator(terraform_dir)
137 assert code == 1, out
138 assert "no_wildcard_iam" in out
139
140
141def test_heroku_policy_check_runs_after_eks_fragment() -> None:
142 phase_dir = (
143 PLUGIN_SKILL_ROOT.parent
144 / "heroku-to-aws"
145 / "references"
146 / "phases"
147 / "generate"
148 )
149 phase = (phase_dir / "generate.md").read_text()
150 terraform_fragment = (phase_dir / "generate-terraform.md").read_text()
151 assembler = (phase_dir / "generate-assemble.md").read_text()
152
153 assert phase.index("_id: eks-generate") < phase.index("_assemble:")
154 assert "Defer the authoritative Terraform policy check to the assembler" in terraform_fragment
155 assert "Authoritative Terraform policy check (after all Terraform producers)" in assembler
156 assert "after `eks-generate`" in assembler
157
158
159def test_heroku_eb_bad_elasticache_and_http_forward_fails() -> None:
160 # BOTH the unencrypted-ElastiCache and the HTTP-forward violations must fire —
161 # assert both rule IDs from --json so an alb_http_redirect regression can't hide
162 # behind the elasticache failure.
163 with tempfile.TemporaryDirectory() as tmp:
164 out_path = Path(tmp) / "verdict.json"
165 code, _ = run_policy_validator(BAD_HEROKU_EB_ELASTICACHE, json_out=out_path)
166 report = json.loads(out_path.read_text())
167 assert code == 1
168 assert report["policy_status"] == "POLICY_FAIL"
169 rules = {v["rule"] for v in report["violations"]}
170 assert "elasticache_encryption_at_rest" in rules, rules
171 assert "alb_http_redirect" in rules, rules
172
173
174def test_internal_alb_skips_https_requirement() -> None:
175 code, out = run_policy_validator(INTERNAL_ALB)
176 assert code == 0, out
177 assert "POLICY_OK" in out
178
179
180def test_json_report_written_on_pass() -> None:
181 with tempfile.TemporaryDirectory() as tmp:
182 out_path = Path(tmp) / "verdict.json"
183 code, _ = run_policy_validator(GOOD_FIXTURE, json_out=out_path)
184 assert code == 0
185 report = json.loads(out_path.read_text())
186 assert report["policy_status"] == "POLICY_OK"
187 assert report["violations"] == []
188
189
190def test_json_report_written_on_fail() -> None:
191 with tempfile.TemporaryDirectory() as tmp:
192 out_path = Path(tmp) / "verdict.json"
193 code, _ = run_policy_validator(BAD_HTTP_FORWARD, json_out=out_path)
194 assert code == 1
195 report = json.loads(out_path.read_text())
196 assert report["policy_status"] == "POLICY_FAIL"
197 rules = {v["rule"] for v in report["violations"]}
198 assert "alb_http_redirect" in rules
199
200
201def test_missing_directory_is_usage_error() -> None:
202 code, out = run_policy_validator(Path("/nonexistent/terraform/dir"))
203 assert code == 2, out
204 assert "not_a_directory" in out
205
206
207def test_no_tf_files_flagged() -> None:
208 with tempfile.TemporaryDirectory() as tmp:
209 code, out = run_policy_validator(Path(tmp))
210 assert code == 1, out
211 assert "No .tf files" in out
212
213
214def test_rds_public_and_unencrypted_fails() -> None:
215 code, out = run_policy_validator(BAD_RDS_PUBLIC_UNENCRYPTED)
216 assert code == 1, out
217 assert "POLICY_FAIL" in out
218
219
220def test_rds_public_unencrypted_reports_both_rules() -> None:
221 with tempfile.TemporaryDirectory() as tmp:
222 out_path = Path(tmp) / "verdict.json"
223 run_policy_validator(BAD_RDS_PUBLIC_UNENCRYPTED, json_out=out_path)
224 rules = {v["rule"] for v in json.loads(out_path.read_text())["violations"]}
225 assert "rds_not_public" in rules
226 assert "rds_encryption_at_rest" in rules
227
228
229def test_rds_private_encrypted_passes() -> None:
230 # Also proves variable-driven storage_encrypted fails open (not flagged).
231 code, out = run_policy_validator(GOOD_RDS_PRIVATE_ENCRYPTED)
232 assert code == 0, out
233 assert "POLICY_OK" in out
234
235
236def test_db_sg_public_ingress_fails() -> None:
237 code, out = run_policy_validator(BAD_DB_SG_PUBLIC)
238 assert code == 1, out
239 assert "db_sg_no_public_ingress" in out
240
241
242def test_db_sg_scoped_passes() -> None:
243 # Proves: app-SG-scoped DB ingress OK; public 443 not flagged as a DB port;
244 # separate aws_vpc_security_group_ingress_rule fails open (not correlated).
245 code, out = run_policy_validator(GOOD_DB_SG_SCOPED)
246 assert code == 0, out
247 assert "POLICY_OK" in out
248
249
250def test_wildcard_iam_fails() -> None:
251 code, out = run_policy_validator(BAD_WILDCARD_IAM)
252 assert code == 1, out
253 assert "no_wildcard_iam" in out
254
255
256def test_scoped_iam_passes() -> None:
257 # Proves scoped policy OK and assume-role trust policy is not a wildcard hit.
258 code, out = run_policy_validator(GOOD_SCOPED_IAM)
259 assert code == 0, out
260 assert "POLICY_OK" in out
261
262
263def test_wildcard_iam_listform_fails() -> None:
264 # Regression: Resource = ["*"] (single-element list) must fail, like "*".
265 code, out = run_policy_validator(BAD_WILDCARD_IAM_LISTFORM)
266 assert code == 1, out
267 assert "no_wildcard_iam" in out
268
269
270def test_iam_scoped_list_passes() -> None:
271 # A list of scoped actions / resource ARNs is not a wildcard — must pass.
272 code, out = run_policy_validator(GOOD_IAM_SCOPED_LIST)
273 assert code == 0, out
274 assert "POLICY_OK" in out
275
276
277def test_internet_facing_nlb_not_flagged() -> None:
278 # Regression: an internet-facing Network LB (L4) has no HTTPS:443 listener
279 # by design and must NOT trip the ALB HTTPS rule.
280 code, out = run_policy_validator(GOOD_INTERNET_NLB)
281 assert code == 0, out
282 assert "POLICY_OK" in out
283
284
285def test_sg_public_admin_ingress_fails() -> None:
286 code, out = run_policy_validator(BAD_SG_PUBLIC_SSH)
287 assert code == 1, out
288 assert "sg_no_public_admin_ingress" in out
289
290
291def test_sg_public_admin_ingress_reports_ssh_and_redis() -> None:
292 with tempfile.TemporaryDirectory() as tmp:
293 out_path = Path(tmp) / "verdict.json"
294 run_policy_validator(BAD_SG_PUBLIC_SSH, json_out=out_path)
295 summaries = " ".join(
296 v["summary"] for v in json.loads(out_path.read_text())["violations"]
297 )
298 assert "22 (SSH)" in summaries
299 assert "6379 (Redis)" in summaries
300
301
302def test_sg_public_webapp_passes() -> None:
303 # Public web ports (80/443), a high game-port range, and a privately-scoped
304 # SSH rule must all pass — no false positive from the sensitive-port rule.
305 code, out = run_policy_validator(GOOD_SG_PUBLIC_WEBAPP)
306 assert code == 0, out
307 assert "POLICY_OK" in out
308
309
310def test_sg_public_ipv6_ingress_fails() -> None:
311 """Regression: ipv6_cidr_blocks = ["::/0"] is public exposure and must fail.
312
313 The old checker searched r'cidr_blocks\\s*=\\s*\\[' which also matches the tail
314 of `ipv6_cidr_blocks`, and never looked for ::/0 at all — so an SG opening
315 SSH/Postgres to the entire IPv6 internet was reported POLICY_OK.
316 """
317 code, out = run_policy_validator(BAD_SG_PUBLIC_IPV6)
318 assert code == 1, out
319 assert "sg_no_public_admin_ingress" in out
320 assert "db_sg_no_public_ingress" in out
321
322
323def test_sg_public_ipv6_ingress_names_the_ipv6_cidr() -> None:
324 # The report must say ::/0, not 0.0.0.0/0 — the IPv4 list here is benign.
325 with tempfile.TemporaryDirectory() as tmp:
326 out_path = Path(tmp) / "verdict.json"
327 run_policy_validator(BAD_SG_PUBLIC_IPV6, json_out=out_path)
328 summaries = " ".join(
329 v["summary"] for v in json.loads(out_path.read_text())["violations"]
330 )
331 assert "::/0" in summaries
332 assert "22 (SSH)" in summaries
333 assert "0.0.0.0/0" not in summaries
334
335
336def test_ipv4_public_ingress_caught_when_ipv6_list_declared_first() -> None:
337 """Regression: the pre-fix regex made the IPv4 check ORDER-DEPENDENT.
338
339 r'cidr_blocks\\s*=\\s*\\[' matched the tail of `ipv6_cidr_blocks`, so when the
340 IPv6 attribute came first the checker read the IPv6 list and never looked at
341 cidr_blocks at all. A plain `0.0.0.0/0` on port 22 — the exact case this rule
342 exists to catch, with no IPv6 exposure involved — silently returned
343 POLICY_OK. Flipping the two lines was enough to make it fail correctly.
344 """
345 sg = """
346resource "aws_security_group" "ipv4_ssh_open" {
347 name = "test-sg"
348 vpc_id = "vpc-123"
349
350 ingress {
351 from_port = 22
352 to_port = 22
353 protocol = "tcp"
354 %s
355 }
356}
357"""
358 ipv6_first = 'ipv6_cidr_blocks = ["2001:db8:1234::/48"]\n cidr_blocks = ["0.0.0.0/0"]'
359 ipv4_first = 'cidr_blocks = ["0.0.0.0/0"]\n ipv6_cidr_blocks = ["2001:db8:1234::/48"]'
360 for label, attrs in (("ipv6-first", ipv6_first), ("ipv4-first", ipv4_first)):
361 with tempfile.TemporaryDirectory() as tmp:
362 (Path(tmp) / "vpc.tf").write_text(sg % attrs, encoding="utf-8")
363 code, out = run_policy_validator(Path(tmp))
364 assert code == 1, f"{label}: 0.0.0.0/0 on SSH must fail, got: {out}"
365 assert "22 (SSH)" in out, f"{label}: {out}"
366 assert "0.0.0.0/0" in out, f"{label}: must name the IPv4 range, got: {out}"
367
368
369def test_sg_ipv6_scoped_passes() -> None:
370 # Public web over ::/0 is fine; SSH/Postgres over a scoped IPv6 prefix must
371 # not fire. Guards against over-matching any ipv6_cidr_blocks value.
372 code, out = run_policy_validator(GOOD_SG_IPV6_SCOPED)
373 assert code == 0, out
374 assert "POLICY_OK" in out
375
376
377def _sg_fixture(attrs: str, port: int = 22) -> str:
378 return f"""
379resource "aws_security_group" "under_test" {{
380 name = "sg"
381 vpc_id = "vpc-123"
382
383 ingress {{
384 from_port = {port}
385 to_port = {port}
386 protocol = "tcp"
387 {attrs}
388 }}
389}}
390"""
391
392
393def _verdict(tf_body: str) -> tuple[int, str]:
394 with tempfile.TemporaryDirectory() as tmp:
395 (Path(tmp) / "vpc.tf").write_text(tf_body, encoding="utf-8")
396 return run_policy_validator(Path(tmp))
397
398
399def test_cidr_attribute_sharing_line_with_opening_brace_is_read() -> None:
400 """Regression: a line-anchored pattern alone hides an attribute that shares
401 its block's opening line, which is valid HCL:
402
403 ingress { cidr_blocks = ["0.0.0.0/0"]
404
405 An earlier revision of this fix used a bare `^\\s*` anchor and traded the
406 ipv6 tail-match bug for this false negative (main: POLICY_FAIL -> POLICY_OK).
407 The `(?:^|\\{)` alternation keeps both properties.
408 """
409 body = """
410resource "aws_security_group" "brace_same_line" {
411 name = "sg"
412 vpc_id = "vpc-123"
413
414 ingress { cidr_blocks = ["0.0.0.0/0"]
415 from_port = 22
416 to_port = 22
417 protocol = "tcp"
418 }
419}
420"""
421 code, out = _verdict(body)
422 assert code == 1, f"0.0.0.0/0 on the brace line must fail, got: {out}"
423 assert "22 (SSH)" in out, out
424
425
426def test_noncanonical_ipv6_zero_prefix_spellings_fire() -> None:
427 """`::/0` has many legal spellings and Terraform normalises none of them, so
428 the check canonicalises via ipaddress rather than substring-matching text."""
429 for spelling in ("::/0", "::0/0", "0:0:0:0:0:0:0:0/0", "0000:0000:0000:0000:0000:0000:0000:0000/0"):
430 code, out = _verdict(_sg_fixture(f'ipv6_cidr_blocks = ["{spelling}"]'))
431 assert code == 1, f"{spelling} is the whole IPv6 internet, must fail: {out}"
432 assert "sg_no_public_admin_ingress" in out, f"{spelling}: {out}"
433
434
435def test_scoped_and_nonliteral_cidrs_still_fail_open() -> None:
436 """The canonicalising check must not widen what counts as public: scoped
437 prefixes and non-literal values stay unflagged (documented fail-open)."""
438 for spelling in ("10.0.0.0/8", "2001:db8:1234::/48", "0.0.0.0/1", "128.0.0.0/1"):
439 attr = "ipv6_cidr_blocks" if ":" in spelling else "cidr_blocks"
440 code, out = _verdict(_sg_fixture(f'{attr} = ["{spelling}"]'))
441 assert code == 0, f"{spelling} is not the whole internet, must pass: {out}"
442 for nonliteral in ("var.admin_cidrs", "[var.cidr]", "local.allowed"):
443 inner = nonliteral if nonliteral.startswith("[") else f"[{nonliteral}]"
444 code, out = _verdict(_sg_fixture(f"cidr_blocks = {inner}"))
445 assert code == 0, f"non-literal {nonliteral} must fail open: {out}"
446
447
448def test_cidr_named_only_in_a_comment_is_not_an_allowed_range() -> None:
449 """False positive: a correctly-scoped SG that merely MENTIONS 0.0.0.0/0 in a
450 comment inside the list was reported POLICY_FAIL, blocking a valid migration.
451
452 Also pins the opposite direction — stripping comments must not swallow a real
453 entry that follows one, which a naive split-on-comma would have done.
454 """
455 scoped_with_comment = """
456 cidr_blocks = [
457 # TODO: was 0.0.0.0/0
458 "10.0.0.0/8",
459 ]"""
460 code, out = _verdict(_sg_fixture(scoped_with_comment, port=3389))
461 assert code == 0, f"a commented CIDR is not an allowed range: {out}"
462
463 comment_then_real = """
464 cidr_blocks = [
465 # temporary, remove before prod
466 "0.0.0.0/0",
467 ]"""
468 code, out = _verdict(_sg_fixture(comment_then_real, port=3389))
469 assert code == 1, f"a real 0.0.0.0/0 after a comment must still fire: {out}"
470 assert "3389 (RDP)" in out, out
471
472
473def test_commented_out_cidr_attribute_is_not_read() -> None:
474 # A commented-out attribute must not be read as set; the live value here is
475 # an SG reference, so this SG is correctly scoped and must pass.
476 body = _sg_fixture('# cidr_blocks = ["0.0.0.0/0"]\n security_groups = ["sg-abc123"]')
477 code, out = _verdict(body)
478 assert code == 0, f"commented-out cidr_blocks must not fire: {out}"
479
480
481def test_comment_marker_inside_string_does_not_swallow_real_cidr() -> None:
482 """False negative: comment stripping must share the lexer, not re-scan with a
483 regex. A regex sees the `/*` inside the description STRING as a comment
484 opener and deletes everything up to the `*/` in the later line comment —
485 including the live `cidr_blocks = ["0.0.0.0/0"]` between them, reporting a
486 public SSH ingress as POLICY_OK. The lexer knows string interiors are not
487 comment openers, so the public range survives stripping and fires.
488 """
489 body = _sg_fixture(
490 'description = "temp /* migration window"\n'
491 ' cidr_blocks = ["0.0.0.0/0"]\n'
492 " # closes */ in runbook"
493 )
494 code, out = _verdict(body)
495 assert code == 1, f"0.0.0.0/0 after a string containing /* must still fire: {out}"
496 assert "22 (SSH)" in out, out
497
498
499def test_elasticache_unencrypted_fails() -> None:
500 code, out = run_policy_validator(BAD_ELASTICACHE_UNENCRYPTED)
501 assert code == 1, out
502 assert "elasticache_encryption_at_rest" in out
503
504
505def test_elasticache_encrypted_passes() -> None:
506 # Encrypted RG passes; variable-driven RG fails open (not flagged).
507 code, out = run_policy_validator(GOOD_ELASTICACHE_ENCRYPTED)
508 assert code == 0, out
509 assert "POLICY_OK" in out
510
511
512def test_elasticache_cluster_redis_unencrypted_fails() -> None:
513 # Single-node Redis aws_elasticache_cluster with no encryption must fail.
514 code, out = run_policy_validator(BAD_ELASTICACHE_CLUSTER_REDIS_UNENCRYPTED)
515 assert code == 1, out
516 assert "elasticache_encryption_at_rest" in out
517
518
519def test_elasticache_cluster_redis_encrypted_passes() -> None:
520 # Redis aws_elasticache_cluster with transit_encryption_enabled on must pass
521 # (at_rest_encryption_enabled is not valid on this resource).
522 code, out = run_policy_validator(GOOD_ELASTICACHE_CLUSTER_REDIS_ENCRYPTED)
523 assert code == 0, out
524 assert "POLICY_OK" in out
525
526
527def test_elasticache_cluster_memcached_exempt_passes() -> None:
528 # A Memcached cluster stays exempt (fail open) and must not be flagged.
529 code, out = run_policy_validator(GOOD_ELASTICACHE_CLUSTER_MEMCACHED)
530 assert code == 0, out
531 assert "POLICY_OK" in out
532
533
534def test_block_form_forward_https_listener_does_not_false_fail() -> None:
535 """Regression: a valid HTTPS listener whose default_action uses a NESTED
536 forward { ... } block before `type` must NOT be misparsed as missing/wrong.
537
538 The old r'default_action{[^}]*?type' regex stopped at the first '}' (the end
539 of the nested block) and failed to read `type`, producing a false POLICY_FAIL.
540 """
541 with tempfile.TemporaryDirectory() as tmp:
542 (Path(tmp) / "compute.tf").write_text(
543 """
544resource "aws_lb" "app" {
545 name = "app-alb"
546 internal = false
547}
548
549resource "aws_lb_listener" "https" {
550 load_balancer_arn = aws_lb.app.arn
551 port = 443
552 protocol = "HTTPS"
553 ssl_policy = "ELBSecurityPolicy-TLS13-1-2-2021-06"
554 # Placeholder — the checker only tests for the presence of `certificate_arn`,
555 # not its value (see has_certificate_arn). A literal ACM ARN here trips secret
556 # scanners for no benefit, so use a variable reference.
557 certificate_arn = var.acm_certificate_arn
558
559 default_action {
560 forward {
561 target_group {
562 arn = aws_lb_target_group.app.arn
563 }
564 }
565 type = "forward"
566 }
567}
568
569resource "aws_lb_listener" "http_redirect" {
570 load_balancer_arn = aws_lb.app.arn
571 port = 80
572 protocol = "HTTP"
573
574 default_action {
575 type = "redirect"
576 redirect {
577 port = "443"
578 protocol = "HTTPS"
579 status_code = "HTTP_301"
580 }
581 }
582}
583""",
584 encoding="utf-8",
585 )
586 code, out = run_policy_validator(Path(tmp))
587 assert code == 0, f"block-form forward should pass, got: {out}"
588 assert "POLICY_OK" in out
589
590
591def test_quoted_port_https_listener_passes() -> None:
592 """Regression: Terraform accepts a quoted integer for number arguments, so a
593 correct HTTPS listener written as port = "443" must NOT be falsely flagged.
594
595 Reading only bare integers left port=None, so https_ok was empty and a valid
596 stack failed with rule=alb_https_listener.
597 """
598 code, out = run_policy_validator(GOOD_QUOTED_PORT_HTTPS)
599 assert code == 0, out
600 assert "POLICY_OK" in out
601
602
603def test_quoted_port_db_sg_public_ingress_fails() -> None:
604 """Regression: from_port / to_port = "5432" with cidr_blocks = ["0.0.0.0/0"]
605 is a world-open database and must fire, not silently pass."""
606 code, out = run_policy_validator(BAD_DB_SG_PUBLIC_QUOTED_PORT)
607 assert code == 1, out
608 assert "db_sg_no_public_ingress" in out
609
610
611def test_quoted_port_sg_public_admin_ingress_fails() -> None:
612 """Same quoted-port regression for the sensitive-port rule (SSH on 22)."""
613 with tempfile.TemporaryDirectory() as tmp:
614 (Path(tmp) / "vpc.tf").write_text(
615 """
616resource "aws_security_group" "bad" {
617 name = "bad-ssh-sg"
618
619 ingress {
620 from_port = "22"
621 to_port = "22"
622 protocol = "tcp"
623 cidr_blocks = ["0.0.0.0/0"]
624 }
625}
626""",
627 encoding="utf-8",
628 )
629 code, out = run_policy_validator(Path(tmp))
630 assert code == 1, out
631 assert "sg_no_public_admin_ingress" in out
632 assert "22 (SSH)" in out
633
634
635def test_attr_int_reads_bare_and_quoted_integers() -> None:
636 assert VALIDATOR._attr_int(" port = 443\n", "port") == 443
637 assert VALIDATOR._attr_int(' port = "443"\n', "port") == 443
638 assert VALIDATOR._attr_int(' from_port = "0"\n', "from_port") == 0
639
640
641def test_attr_int_rejects_non_literal_quoted_values() -> None:
642 """Precision guard: a quoted value must be ENTIRELY digits. Anything else is
643 non-literal and must return None so the rules keep failing open."""
644 for body in (
645 ' port = "443x"\n',
646 ' port = "x443"\n',
647 ' port = "443-444"\n',
648 ' port = "${var.port}"\n',
649 ' port = " 443"\n',
650 ' port = ""\n',
651 " port = var.port\n",
652 " port = local.ports[0]\n",
653 ):
654 assert VALIDATOR._attr_int(body, "port") is None, body
655
656
657def test_attr_int_absent_attribute_is_none() -> None:
658 assert VALIDATOR._attr_int(' protocol = "HTTPS"\n', "port") is None
659
660
661# A listener whose own `port`/`protocol` are written AFTER its default_action.
662# Valid HCL — `terraform fmt` does not reorder attributes relative to blocks — and
663# the nested redirect carries both a `port` and a `protocol`, so a whole-body read
664# returns the redirect's 443/HTTPS instead of the listener's own 80/HTTP.
665_LISTENER_ATTRS_AFTER_NESTED_BLOCK = """{
666 load_balancer_arn = aws_lb.app.arn
667
668 default_action {
669 type = "redirect"
670 redirect {
671 port = "443"
672 protocol = "HTTPS"
673 }
674 }
675
676 port = 80
677 protocol = "HTTP"
678}"""
679
680
681def test_attrs_read_from_own_block_not_nested_block() -> None:
682 """Accepting quoted integers makes a nested `port = "443"` a match candidate
683 where a bare-only pattern could never match one, so reads are scoped to the
684 block's own attributes. Pinned for both readers: _attr_string had the same
685 whole-body weakness before this change and shares the scoping now."""
686 assert VALIDATOR._attr_int(_LISTENER_ATTRS_AFTER_NESTED_BLOCK, "port") == 80
687 assert (
688 VALIDATOR._attr_string(_LISTENER_ATTRS_AFTER_NESTED_BLOCK, "protocol") == "HTTP"
689 )
690
691
692def test_attr_nested_only_value_is_not_promoted() -> None:
693 """An attribute that exists ONLY inside a nested block is absent from the
694 enclosing block — returning the nested value would invent an attribute.
695
696 Written as `terraform fmt` emits it (one attribute per line), which is the form
697 that actually reaches the line-anchored pattern; a collapsed one-line
698 `redirect { port = "443" }` never matches and so would not exercise scoping."""
699 body = '{\n default_action {\n redirect {\n port = "443"\n }\n }\n}'
700 assert VALIDATOR._attr_int(body, "port") is None
701
702
703def test_attr_interpolation_does_not_hide_later_attributes() -> None:
704 """`${...}` braces are balanced, so an interpolated value earlier in the body
705 must not make a following top-level attribute unreadable."""
706 body = '{\n identifier = "${var.project}-db"\n storage_encrypted = true\n}'
707 assert VALIDATOR._attr_string(body, "storage_encrypted") == "true"
708
709
710def test_quote_nested_in_interpolation_does_not_end_the_string() -> None:
711 """A `"` inside `${...}` opens a NESTED string; it must not end the outer one.
712
713 Regression guard: the first lexer set `state = "code"` on any `"` seen in a
714 string, so an inner string's closing quote ended the outer string and the rest
715 of the line was scanned as code — putting textual braces back into the depth
716 count. That is the exact failure class the lexer was added to close, reached by
717 a different route, and it is reachable from ordinary HCL (`replace(v, "{", "")`,
718 `templatefile("t.tpl", {"a"="{"})`).
719
720 Both directions matter: an opening brace skews depth (the attribute is judged
721 nested and skipped), a closing brace truncates the body outright.
722 """
723 for label, body in (
724 ("open brace in nested quote", '{\n identifier = "${lookup(var.m, "a{b")}"\n port = 80\n}'),
725 ("close brace in nested quote", '{\n n = "${replace(v, "}", "")}"\n port = 80\n}'),
726 ("both braces in nested quote", '{\n n = "${replace(v, "{}", "")}"\n port = 80\n}'),
727 ("nested quote, no brace", '{\n a = "${lookup(var.tags, "Name")}"\n port = 80\n}'),
728 ("two interpolations", '{\n a = "${f("x{")}-${g("y}")}"\n port = 80\n}'),
729 ):
730 assert VALIDATOR._attr_int(body, "port") == 80, label
731
732
733def test_lexically_irrelevant_braces_do_not_hide_attributes() -> None:
734 """A brace in a comment, string, or heredoc is not a block delimiter, so it must
735 not shift a real top-level attribute out of scope.
736
737 Regression guard: an earlier revision counted raw `{`/`}` and DID hide the
738 attribute here. For these rules a hidden attribute means a MISSED violation
739 (`publicly_accessible = true` read as absent, hence compliant), so "fail open"
740 is the wrong default — the value is a plain top-level literal, not ambiguous."""
741 for label, body in (
742 ("hash comment", '{\n # primary { instance\n port = 80\n}'),
743 ("slash comment", '{\n // primary { instance\n port = 80\n}'),
744 ("block comment", '{\n /* primary { instance */\n port = 80\n}'),
745 ("string", '{\n identifier = "primary { db"\n port = 80\n}'),
746 ("unbalanced string", '{\n name = "a { b"\n port = 80\n}'),
747 ("heredoc", '{\n policy = <<EOT\n { not code }\nEOT\n port = 80\n}'),
748 ):
749 assert VALIDATOR._attr_int(body, "port") == 80, label
750
751
752def test_extraction_not_truncated_by_lexically_irrelevant_closing_brace() -> None:
753 """A `}` in a comment, string, or heredoc body is not a delimiter, so it must
754 not end the resource body early.
755
756 Regression guard: raw brace counting in `_extract_braced_block` truncated the
757 body before `publicly_accessible = true`, so the resource reached the rules
758 with the offending attribute already discarded — zero violations."""
759 for label, filler in (
760 ("hash comment", " # capacity review } pending"),
761 ("slash comment", " // capacity review } pending"),
762 ("block comment", " /* capacity review } pending */"),
763 ("string", ' identifier = "primary } db"'),
764 ("heredoc", " description = <<EOT\n a closing } brace\nEOT"),
765 ("closing brace in a quote nested in an interpolation", ' n = "${replace(v, "}", "")}"'),
766 ):
767 content = (
768 'resource "aws_db_instance" "pg" {\n'
769 f"{filler}\n"
770 " publicly_accessible = true\n"
771 "}\n"
772 )
773 blocks = VALIDATOR._extract_blocks(content, "aws_db_instance")
774 assert len(blocks) == 1, label
775 assert (
776 VALIDATOR._attr_string(blocks[0][1], "publicly_accessible") == "true"
777 ), label
778
779
780def test_regular_heredoc_terminator_must_be_unindented() -> None:
781 """`<<TAG` requires the terminator at column 0; only `<<-TAG` tolerates
782 indentation. Treating both as tolerant exits a regular heredoc early, so its
783 body text is read as real code — here an in-heredoc `port = 443` was returned
784 instead of the listener's own `port = 80`."""
785 regular = '{\n policy = <<EOT\n EOT\n port = 443\nEOT\n port = 80\n}'
786 assert VALIDATOR._attr_int(regular, "port") == 80
787 indented = '{\n policy = <<-EOT\n { not code }\n EOT\n port = 80\n}'
788 assert VALIDATOR._attr_int(indented, "port") == 80
789
790
791def test_attribute_presence_probe_matches_value_scoping() -> None:
792 """`_has_own_attr` must agree with `_attr_string` about what counts as set.
793
794 The encryption rules read "attribute present but unreadable" as
795 variable-driven and skip. If presence counted a commented-out assignment the
796 reader ignores, a resource whose attribute is genuinely ABSENT — hence
797 unencrypted by default — would be skipped instead of flagged."""
798 only_commented = '{\n identifier = "pg"\n /*\n storage_encrypted = false\n */\n}'
799 assert VALIDATOR._has_own_attr(only_commented, "storage_encrypted") is False
800 assert VALIDATOR._attr_string(only_commented, "storage_encrypted") is None
801 real = '{\n storage_encrypted = var.encrypt\n}'
802 assert VALIDATOR._has_own_attr(real, "storage_encrypted") is True
803 assert VALIDATOR._attr_string(real, "storage_encrypted") is None
804
805
806def test_commented_out_resource_is_not_extracted() -> None:
807 """A commented-out resource declaration is not a resource."""
808 content = (
809 '# resource "aws_db_instance" "ghost" {\n'
810 "# publicly_accessible = true\n"
811 "# }\n"
812 )
813 assert VALIDATOR._extract_blocks(content, "aws_db_instance") == []
814
815
816def test_commented_out_attribute_is_not_read() -> None:
817 """An attribute that only appears inside a comment is not set."""
818 assert VALIDATOR._attr_int("{\n # port = 443\n}", "port") is None
819 assert VALIDATOR._attr_int("{\n // port = 443\n}", "port") is None
820 assert VALIDATOR._attr_int("{\n /*\n port = 443\n */\n}", "port") is None
821 assert (
822 VALIDATOR._attr_string('{\n /*\n protocol = "HTTPS"\n */\n}', "protocol")
823 is None
824 )
825
826
827def test_every_fixture_matches_its_good_bad_prefix() -> None:
828 """Enforce the naming invariant the fixture set relies on: `good-*` (and the
829 internal-ALB case) must be POLICY_OK, `bad-*` must be POLICY_FAIL. Without
830 this, each fixture is wired up by hand and a new one added without its own
831 test would silently get no coverage."""
832 fixture_dirs = sorted(d for d in FIXTURES.iterdir() if d.is_dir())
833 # Exact, not `>=`: a floor cannot detect fixtures being deleted down to it,
834 # which is the removal this assertion exists to catch. Update deliberately
835 # when adding or removing a fixture.
836 assert len(fixture_dirs) == 31, f"fixture count changed: {[d.name for d in fixture_dirs]}"
837 for fixture in fixture_dirs:
838 code, out = run_policy_validator(fixture)
839 if fixture.name.startswith("bad-"):
840 assert code == 1, f"{fixture.name} must FAIL: {out}"
841 assert "POLICY_FAIL" in out, f"{fixture.name}: {out}"
842 else:
843 assert code == 0, f"{fixture.name} must pass: {out}"
844 assert "POLICY_OK" in out, f"{fixture.name}: {out}"