Setting the file. One moment.
Validate Terraform Policy · Tf Best Practices · aws/agent-toolkit-for-aws · Skills Docs
Back to the top of the page 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 _default_action_type
— line 379
This file
Number 33.4
Position 4 of 38
Type Python
Size 44 KB
Lines 1,029 scripts/ validate-terraform-policy.py
Python · 1,029 lines · 44 KB
15 publicly_accessible = true (fail-open when variable-driven or absent).
16 - db_sg_no_public_ingress: an inline aws_security_group ingress covering a
17 database port (5432 / 3306) must not allow 0.0.0.0/0 (IPv4) or ::/0 (IPv6)
18 (fail-open on separate aws_security_group_rule /
19 aws_vpc_security_group_ingress_rule resources, which this static reader
20 cannot correlate).
21 - sg_no_public_admin_ingress: an inline ingress must not open a well-known
22 admin/datastore port (SSH, RDP, Redis, Memcached, Mongo, Elasticsearch,
23 Kibana) to 0.0.0.0/0 or ::/0. Scoped to a fixed never-public port list — web
24 ports and app/game ports are not flagged. Same inline-only fail-open scope.
25 Both SG rules read cidr_blocks and ipv6_cidr_blocks independently: an
26 IPv6-only or dual-stack VPC is exposed by ::/0 exactly as IPv4 is by
27 0.0.0.0/0.
28 - no_wildcard_iam: a literal IAM policy document with Effect "Allow" must not
29 use Action "*" or Resource "*", except an isolated
30 elasticbeanstalk:CreateStorageLocation statement because AWS provides no
31 resource-level permission for it (fail-open on aws_iam_policy_document data
32 sources, whose statements are not visible as literal JSON here).
33 - rds_encryption_at_rest: aws_db_instance / aws_rds_cluster must set
34 storage_encrypted = true (RDS defaults to UNENCRYPTED). Fail-open when
35 variable-driven. S3 is intentionally NOT checked: buckets have default
36 SSE-S3 since Jan 2023, so a missing SSE block is not an unencrypted bucket.
37 - elasticache_encryption_at_rest: aws_elasticache_replication_group must set
38 at_rest_encryption_enabled = true; a standalone Redis aws_elasticache_cluster
39 must set transit_encryption_enabled = true (at_rest_encryption_enabled is not
40 a valid argument on that resource — at-rest requires a replication group).
41 Fail-open when variable-driven.
42
43 Usage:
44 python3 validate-terraform-policy.py /path/to/terraform [--json report.json]
45
46 Exit 0 on POLICY_OK, 1 on POLICY_FAIL, 2 on usage/IO error.
47 """
48
49 from __future__ import annotations
50
51 import argparse
52 import ipaddress
53 import json
54 import re
55 import sys
56 from dataclasses import asdict, dataclass
57 from pathlib import Path
58
59 RESOURCE_OPEN = re.compile(
60 r 'resource \s + " ( ?P<type> [ a-zA-Z0-9_ ] + ) " \s + " ( ?P<name> [ ^" ] + ) " \s * \{ ' ,
61 re. MULTILINE ,
62 )
63
64
65 @dataclass ( frozen = True )
66 class Violation :
67 check: str # "policy"
68 rule: str # "alb_https_listener" | "alb_http_redirect" | "no_tf_files"
69 # | "rds_not_public" | "db_sg_no_public_ingress"
70 # | "no_wildcard_iam" | "rds_encryption_at_rest"
71 file : str
72 line: int # 1-based; 0 if unknown
73 severity: str # "error" | "warning"
74 summary: str
75 fix_hint: str
76
77
78 @dataclass ( frozen = True )
79 class ListenerSpec :
80 file : str
81 name: str
82 line: int
83 port: int | None
84 protocol: str | None
85 action_type: str | None
86 has_certificate_arn: bool
87
88
89 def _read_tf_files (terraform_dir: Path) -> list[tuple[ str , str ]]:
90 files: list[tuple[ str , str ]] = []
91 for path in sorted (terraform_dir.rglob( "*.tf" )):
92 files.append(( str (path.relative_to(terraform_dir)), path.read_text( encoding = "utf-8" )))
93 return files
94
95
96 _HEREDOC_OPEN = re.compile( r "<< ( - ? )\s * ([ A-Za-z_ ][ A-Za-z0-9_ ] * ) " )
97
98
99 def _lex_hcl (block: str ) -> tuple[list[ int ], list[ bool ], list[ bool ]]:
100 """Scan HCL once, returning per-index (brace depth, is-real-code, is-comment)
101 arrays.
102
103 Only braces that are actual block delimiters count toward depth. HCL permits
104 `{` and `}` inside `#` / `//` line comments, `/* */` block comments, quoted
105 strings, and heredoc bodies, and those are NOT delimiters — counting them
106 would let unrelated text shift the apparent nesting of a real attribute.
107 `is_code` additionally lets callers ignore an attribute that only appears
108 commented out. `is_comment` marks exactly the comment spans (delimiters
109 included, terminating newline excluded), distinguishing them from the other
110 non-code states — a reader that must ignore comments but still read string
111 contents (list entries are quoted strings) needs that distinction.
112
113 Interpolations are treated as opaque string content: `{` and `}` inside a
114 string are both ignored, so the running depth is unaffected either way. This
115 requires tracking interpolation nesting, because `${...}` may contain its own
116 quoted strings (`"${lookup(var.m, "a{b")}"`). Without that, the inner string's
117 closing quote would end the OUTER string and everything after it would be
118 scanned as code — putting textual braces back into the depth count, which is
119 the failure mode this lexer exists to prevent.
120 """
121 n = len (block)
122 depths = [ 0 ] * n
123 codes = [ False ] * n
124 comments = [ False ] * n
125 depth = 0
126 state = "code"
127 tag = ""
128 tag_indent_ok = False
129 interp = 0 # open `${` levels inside the current string
130 inner_str = False # inside a quoted string nested in an interpolation
131 i = 0
132 while i < n:
133 ch = block[i]
134 nxt = block[i + 1 ] if i + 1 < n else ""
135 if state == "code" :
136 if ch == "#" or (ch == "/" and nxt == "/" ):
137 depths[i], codes[i], comments[i] = depth, False , True
138 state = "line_comment"
139 i += 1
140 continue
141 if ch == "/" and nxt == "*" :
142 depths[i], depths[i + 1 ] = depth, depth
143 comments[i], comments[i + 1 ] = True , True
144 state = "block_comment"
145 i += 2
146 continue
147 if ch == '"' :
148 depths[i], codes[i] = depth, False
149 state = "string"
150 interp = 0
151 inner_str = False
152 i += 1
153 continue
154 if ch == "<" and nxt == "<" :
155 heredoc = _HEREDOC_OPEN .match(block, i)
156 if heredoc:
157 for j in range (i, heredoc.end()):
158 depths[j] = depth
159 tag = heredoc.group( 2 )
160 # Only `<<-TAG` permits an indented terminator; plain `<<TAG`
161 # requires it at column 0, so an indented tag-looking line is
162 # ordinary body text.
163 tag_indent_ok = heredoc.group( 1 ) == "-"
164 state = "heredoc"
165 i = heredoc.end()
166 continue
167 depths[i], codes[i] = depth, True
168 if ch == "{" :
169 depth += 1
170 elif ch == "}" :
171 depth -= 1
172 i += 1
173 continue
174
175 depths[i] = depth
176 if state == "line_comment" :
177 if ch == " \n " :
178 state = "code"
179 else :
180 comments[i] = True
181 i += 1
182 elif state == "block_comment" :
183 comments[i] = True
184 if ch == "*" and nxt == "/" :
185 depths[i + 1 ] = depth
186 comments[i + 1 ] = True
187 state = "code"
188 i += 2
189 else :
190 i += 1
191 elif state == "string" :
192 if ch == " \\ " and i + 1 < n:
193 depths[i + 1 ] = depth
194 i += 2
195 elif ch == "$" and nxt == "{" and not inner_str:
196 depths[i + 1 ] = depth
197 interp += 1
198 i += 2
199 elif interp > 0 and ch == '"' :
200 # A quote inside `${...}` opens/closes a NESTED string; it must not
201 # be mistaken for the end of the interpolated string itself.
202 inner_str = not inner_str
203 i += 1
204 elif interp > 0 and ch == "}" and not inner_str:
205 interp -= 1
206 i += 1
207 else :
208 if ch == '"' and interp == 0 :
209 state = "code"
210 i += 1
211 else : # heredoc — body is literal until a line holding only the tag
212 end = block.find( " \n " , i)
213 line_end = n if end == - 1 else end + 1
214 for j in range (i, line_end):
215 depths[j] = depth
216 line = block[i:line_end]
217 terminator = line.strip() if tag_indent_ok else line.rstrip()
218 if terminator == tag:
219 state = "code"
220 i = line_end
221 return depths, codes, comments
222
223
224 def _extract_braced_block (
225 content: str ,
226 open_brace: int ,
227 lexed: tuple[list[ int ], list[ bool ], list[ bool ]] | None = None ,
228 ) -> tuple[ str , int ]:
229 """Return (block_text_including_braces, index_after_close).
230
231 Uses _lex_hcl, not raw brace counting: a `}` inside a comment, string, or
232 heredoc body is not a delimiter, and treating it as one truncates the block
233 early. That silently drops every attribute after the stray brace — a resource
234 body cut before `publicly_accessible = true` reports no violation at all. Pass
235 `lexed` to reuse a scan already done for this exact `content`.
236 """
237 depths, codes, _ = lexed if lexed is not None else _lex_hcl(content)
238 if open_brace >= len (codes) or not codes[open_brace]:
239 return content[open_brace:], len (content)
240 target = depths[open_brace] + 1
241 for idx in range (open_brace + 1 , len (content)):
242 if codes[idx] and content[idx] == "}" and depths[idx] == target:
243 return content[open_brace : idx + 1 ], idx + 1
244 return content[open_brace:], len (content)
245
246
247 def _extract_blocks (content: str , resource_type: str ) -> list[tuple[ str , str , int ]]:
248 """Return (name, body, 1-based line) for each resource of resource_type.
249
250 The file is lexed once and the scan is shared with every extraction, so
251 extraction, attribute presence, and attribute values all agree on what counts
252 as code. A resource declaration that is itself commented out is skipped.
253 """
254 lexed = _lex_hcl(content)
255 _, codes, _ = lexed
256 blocks: list[tuple[ str , str , int ]] = []
257 for match in RESOURCE_OPEN .finditer(content):
258 if match.group( "type" ) != resource_type:
259 continue
260 if not codes[match.start()]:
261 continue
262 name = match.group( "name" )
263 brace_start = match.end() - 1
264 body, _ = _extract_braced_block(content, brace_start, lexed)
265 line = content.count( " \n " , 0 , match.start()) + 1
266 blocks.append((name, body, line))
267 return blocks
268
269
270 def _own_blocks (body: str , block_type: str ) -> list[ str ]:
271 """Bodies of `block_type { ... }` blocks declared directly in `body`.
272
273 Shares the lexer so a commented-out `ingress {` is not treated as a real rule.
274 """
275 lexed = _lex_hcl(body)
276 _, codes, _ = lexed
277 found: list[ str ] = []
278 for match in re.finditer( rf " { re.escape(block_type) } \s*\ {{ " , body):
279 if not codes[match.start()]:
280 continue
281 inner, _ = _extract_braced_block(body, match.end() - 1 , lexed)
282 found.append(inner)
283 return found
284
285
286 def _has_own_attr (block: str , attr: str ) -> bool :
287 """True if `attr` is assigned among the block's OWN attributes.
288
289 Presence must use the same lexical rules as value reading. A probe that counts
290 a commented-out assignment while the reader correctly ignores it makes the two
291 disagree, and the rules that branch on "attribute present but unreadable →
292 variable-driven → fail open" then skip a resource whose attribute is genuinely
293 absent.
294 """
295 return (
296 _first_own_attr(block, rf "^\s* { re.escape(attr) } \s*=" , re. MULTILINE ) is not None
297 )
298
299
300 def _first_own_attr (block: str , pattern: str , flags: int = 0 ) -> re.Match[ str ] | None :
301 """First match of `pattern` that sits among the block's OWN attributes.
302
303 `_extract_blocks` / `_extract_braced_block` return a resource body INCLUDING
304 its nested blocks, and a plain `re.search` takes the first match anywhere —
305 so an attribute of a nested block can be read as if it belonged to the
306 resource. `aws_lb_listener` is the live example: `default_action { redirect {
307 port = "443", protocol = "HTTPS" } }` carries both a `port` and a `protocol`,
308 and HCL does not require the listener's own `port` to precede its
309 `default_action` (`terraform fmt` will not reorder attributes relative to
310 blocks). Read whole-body, such a listener reports port 443 / protocol HTTPS
311 instead of 80 / HTTP.
312
313 Candidates are therefore filtered by brace depth — only those at the body's
314 own depth are eligible (1 for a braced body as returned by
315 _extract_braced_block, 0 for a bare attribute fragment) — and a candidate that
316 is merely commented out is skipped.
317
318 Depth comes from _lex_hcl, NOT from raw `{`/`}` counting. Raw counting is
319 unsafe here: a brace in an unrelated comment or string literal would shift the
320 depth of a real top-level attribute and hide it, and for these rules a hidden
321 attribute means a MISSED violation — `publicly_accessible = true` reported as
322 compliant. That is the failure class this reader exists to prevent, so
323 lexically-irrelevant braces must not participate.
324 """
325 base = 1 if block.lstrip().startswith( "{" ) else 0
326 depths, codes, _ = _lex_hcl(block)
327 for match in re.finditer(pattern, block, flags):
328 start = match.start()
329 if start < len (codes) and codes[start] and depths[start] == base:
330 return match
331 return None
332
333
334 def _attr_string (block: str , attr: str ) -> str | None :
335 match = _first_own_attr(
336 block, rf '^\s* { re.escape(attr) } \s*=\s*"([^"]*)"' , re. MULTILINE
337 )
338 if match:
339 return match.group( 1 )
340 bool_match = _first_own_attr(
341 block,
342 rf "^\s* { re.escape(attr) } \s*=\s*(true|false)\b" ,
343 re. MULTILINE | re. IGNORECASE ,
344 )
345 return bool_match.group( 1 ).lower() if bool_match else None
346
347
348 def _attr_int (block: str , attr: str ) -> int | None :
349 """Read an integer attribute, accepting BARE (443) and QUOTED ("443") forms.
350
351 Terraform accepts a quoted integer for number-typed arguments (`port = "443"`
352 is valid, idiomatic, `terraform fmt`-clean HCL and converts to 443), so a
353 bare-only reader silently degrades every port-based rule: a correct HTTPS
354 listener written as port = "443" was reported as a missing HTTPS listener
355 (false positive), and — worse — an ingress with from_port = "5432" /
356 cidr_blocks = ["0.0.0.0/0"] produced no covered ports, so the public-ingress
357 rules never fired on a genuinely world-open database (fail-CLOSED miss).
358
359 The quoted arm requires the value to be ENTIRELY digits (the closing quote is
360 anchored against the digits), so "443x", "443-444" and "${var.port}" do NOT
361 read as 443 — they return None and keep the existing fail-open behaviour for
362 non-literal / non-integer values.
363
364 Accepting the quoted form also makes a nested block's `port` a candidate where
365 a bare-only pattern could never match one, so the search is scoped to the
366 block's own attributes via _first_own_attr — see its docstring.
367 """
368 match = _first_own_attr(
369 block,
370 rf '^\s* { re.escape(attr) } \s*=\s*(?:(\d+)|"(\d+)")' ,
371 re. MULTILINE ,
372 )
373 if not match:
374 return None
375 bare, quoted = match.group( 1 ), match.group( 2 )
376 return int (bare if bare is not None else quoted)
377
378
379 def _default_action_type (block: str ) -> str | None :
380 """Extract default_action { ... type = "X" ... } via BRACE-DEPTH matching.
381
382 NOTE : the naive r'default_action \\ s* \\ {[^}]*?type' approach breaks when the
383 default_action contains a nested block (redirect {} / forward {}) placed
384 BEFORE the type attribute — it stops at the first '}'. We isolate the full
385 default_action body by brace matching, then read `type` from it.
386 """
387 m = re.search( r "default_action \s * \{ " , block)
388 if not m:
389 return None
390 body, _ = _extract_braced_block(block, m.end() - 1 )
391 tmatch = re.search( r ' ^\s * type \s * = \s * " ([ ^" ] + ) "' , body, re. MULTILINE )
392 return tmatch.group( 1 ) if tmatch else None
393
394
395 def _has_internet_facing_alb (tf_files: list[tuple[ str , str ]]) -> bool :
396 """True if any aws_lb is an internet-facing APPLICATION load balancer.
397
398 The HTTPS-listener posture applies only to Application Load Balancers (L7).
399 Network (L4 TCP/UDP) and Gateway (L3) load balancers legitimately have no
400 HTTPS:443 listener, so an internet-facing NLB/GWLB must NOT be flagged.
401
402 - load_balancer_type == "network" | "gateway" (literal) => skip (not an ALB).
403 - load_balancer_type absent (Terraform default is "application"),
404 "application", or variable-driven => treat as an ALB (fail-safe).
405 - internal absent, "false", or variable-driven => internet-facing (fail-safe:
406 demand HTTPS unless explicitly internal=true).
407 """
408 for _, content in tf_files:
409 for _, body, _line in _extract_blocks(content, "aws_lb" ):
410 lb_type = _attr_string(body, "load_balancer_type" )
411 if lb_type in ( "network" , "gateway" ):
412 continue # L4/L3 balancer — HTTPS-listener rule does not apply
413 internal = _attr_string(body, "internal" )
414 if internal is None or internal == "false" :
415 return True
416 return False
417
418
419 def _parse_listeners (tf_files: list[tuple[ str , str ]]) -> list[ListenerSpec]:
420 listeners: list[ListenerSpec] = []
421 for rel_path, content in tf_files:
422 for name, body, line in _extract_blocks(content, "aws_lb_listener" ):
423 listeners.append(
424 ListenerSpec(
425 file = rel_path,
426 name = name,
427 line = line,
428 port = _attr_int(body, "port" ),
429 protocol = _attr_string(body, "protocol" ),
430 action_type = _default_action_type(body),
431 has_certificate_arn = "certificate_arn" in body,
432 )
433 )
434 return listeners
435
436
437 def check_alb_https_policy (terraform_dir: Path) -> list[Violation]:
438 tf_files = _read_tf_files(terraform_dir)
439 if not tf_files:
440 return [
441 Violation(
442 check = "policy" ,
443 rule = "no_tf_files" ,
444 file = "." ,
445 line = 0 ,
446 severity = "error" ,
447 summary = "No .tf files found in terraform directory" ,
448 fix_hint = "Ensure the generate step wrote terraform/ before validation" ,
449 )
450 ]
451
452 if not _has_internet_facing_alb(tf_files):
453 return []
454
455 listeners = _parse_listeners(tf_files)
456 violations: list[Violation] = []
457
458 https_ok = [
459 l
460 for l in listeners
461 if l.port == 443
462 and (l.protocol or "" ).upper() == "HTTPS"
463 and l.has_certificate_arn
464 and l.action_type == "forward"
465 ]
466
467 if not https_ok:
468 # Point at an aws_lb file when we can, else the first tf file.
469 lb_file = next (
470 (rel for rel, c in tf_files if _extract_blocks(c, "aws_lb" )),
471 tf_files[ 0 ][ 0 ],
472 )
473 violations.append(
474 Violation(
475 check = "policy" ,
476 rule = "alb_https_listener" ,
477 file = lb_file,
478 line = 0 ,
479 severity = "error" ,
480 summary = (
481 "Internet-facing ALB requires an HTTPS listener on port 443 "
482 "with certificate_arn and a forward action"
483 ),
484 fix_hint = (
485 'Add an aws_lb_listener on port 443, protocol "HTTPS", with '
486 "ssl_policy, certificate_arn, and a forward default_action"
487 ),
488 )
489 )
490
491 for l in listeners:
492 if l.port != 80 or (l.protocol or "" ).upper() != "HTTP" :
493 continue
494 if l.action_type == "forward" :
495 violations.append(
496 Violation(
497 check = "policy" ,
498 rule = "alb_http_redirect" ,
499 file = l.file,
500 line = l.line,
501 severity = "error" ,
502 summary = (
503 f "ALB HTTP listener ' { l.name } ' on port 80 forwards to targets; "
504 "it must redirect to HTTPS"
505 ),
506 fix_hint = (
507 "Replace the forward default_action with a redirect block: "
508 'type = "redirect", redirect { port = "443", protocol = "HTTPS", '
509 'status_code = "HTTP_301" }'
510 ),
511 )
512 )
513
514 return violations
515
516
517 _DB_PORTS = ( 5432 , 3306 )
518
519 # Well-known admin / datastore ports that should never be open to the whole
520 # internet in either address family (0.0.0.0/0 or ::/0).
521 # Deliberately EXCLUDES 5432/3306 (covered by db_sg_no_public_ingress, so no
522 # double-reporting) and web ports 80/443 (legitimately public). Kept tight to
523 # unambiguous "never public" ports so valid designs (e.g. game servers on high
524 # ports) are not falsely flagged.
525 _SENSITIVE_NONDB_PORTS = (
526 22 , # SSH
527 3389 , # RDP
528 6379 , # Redis
529 11211 , # Memcached
530 27017 , # MongoDB
531 9200 , # Elasticsearch HTTP
532 9300 , # Elasticsearch transport
533 5601 , # Kibana
534 )
535
536
537 def check_rds_not_public (tf_files: list[tuple[ str , str ]]) -> list[Violation]:
538 """Flag RDS resources that explicitly set publicly_accessible = true.
539
540 Fail-open: absent (RDS default is false) or variable-driven values do not
541 fire — only a literal `true` is a violation.
542 """
543 violations: list[Violation] = []
544 for rel_path, content in tf_files:
545 for res_type in ( "aws_db_instance" , "aws_rds_cluster" ):
546 for name, body, line in _extract_blocks(content, res_type):
547 if _attr_string(body, "publicly_accessible" ) == "true" :
548 violations.append(
549 Violation(
550 check = "policy" ,
551 rule = "rds_not_public" ,
552 file = rel_path,
553 line = line,
554 severity = "error" ,
555 summary = (
556 f " { res_type } ' { name } ' sets publicly_accessible = true — "
557 "the database is reachable from the internet"
558 ),
559 fix_hint = (
560 "Set publicly_accessible = false and place the database in "
561 "private subnets; reach it from application security groups only"
562 ),
563 )
564 )
565 return violations
566
567
568 def _ingress_covered_ports (ingress_body: str , ports: tuple[ int , ... ]) -> list[ int ]:
569 """Return the subset of `ports` whose value falls in this ingress rule's
570 [from_port, to_port] range. Empty when the range is missing/non-literal."""
571 from_p = _attr_int(ingress_body, "from_port" )
572 to_p = _attr_int(ingress_body, "to_port" )
573 if from_p is None or to_p is None :
574 return []
575 return [port for port in ports if from_p <= port <= to_p]
576
577
578 def _ingress_covers_db_port (ingress_body: str ) -> bool :
579 return bool (_ingress_covered_ports(ingress_body, _DB_PORTS ))
580
581
582 def _strip_hcl_comments (text: str ) -> str :
583 """Blank out comment spans, using the lexer's classification of them.
584
585 This must share _lex_hcl's single definition of "what counts as code" rather
586 than keep a parallel comment regex: a regex cannot know that a `/*` inside a
587 quoted string does not open a comment, and stripping from such a false opener
588 swallows every real attribute up to the next `*/` — including a live
589 `cidr_blocks = ["0.0.0.0/0"]`, which then reports POLICY_OK. Comment
590 characters are replaced with spaces rather than deleted, so surviving tokens
591 never fuse and line anchors keep working.
592 """
593 _, _, comments = _lex_hcl(text)
594 return "" .join( " " if comments[i] else ch for i, ch in enumerate (text))
595
596
597 def _attr_list_inner (block: str , attr: str ) -> str | None :
598 """Return the raw text inside a literal `attr = [ ... ]`, else None.
599
600 The name must start a line OR directly follow a `{`. The line-start arm keeps
601 `cidr_blocks` from matching the tail of `ipv6_cidr_blocks` (reading one list
602 while believing it is the other silently inverts the verdict); the brace arm
603 keeps an attribute that shares its block's opening line visible, e.g.
604 `ingress { cidr_blocks = [...]`, which is valid HCL that a line-anchored
605 pattern alone would miss.
606 """
607 m = re.search(
608 rf "(?:^|\ {{ )\s* { re.escape(attr) } \s*=\s*\[(.*?)\]" ,
609 block,
610 re. DOTALL | re. MULTILINE ,
611 )
612 return m.group( 1 ) if m else None
613
614
615 def _is_entire_internet (cidr: str ) -> bool :
616 """True if `cidr` is a literal range covering the whole address space.
617
618 Canonicalises through `ipaddress` instead of comparing text, because IPv6 has
619 many legal spellings of the zero address (`::/0`, `::0/0`, `0:0:0:0:0:0:0:0/0`)
620 and nothing in Terraform normalises them — `terraform fmt` treats a CIDR as an
621 opaque string. Anything that is not a parseable literal (`var.x`, `${...}`,
622 malformed text) returns False, preserving the module's fail-open posture.
623 """
624 try :
625 return ipaddress.ip_network(cidr, strict = False ).prefixlen == 0
626 except ValueError :
627 return False
628
629
630 def _ingress_public_cidrs (ingress_body: str ) -> list[ str ]:
631 """Return the "entire internet" CIDRs this ingress rule allows, both families.
632
633 IPv4 `0.0.0.0/0` and IPv6 `::/0` are equally public: on a dual-stack or
634 IPv6-only VPC, `::/0` on an admin port is a live exposure. Each family is read
635 under its own attribute name so the two are never confused, comments are
636 stripped first (by the shared lexer, so a comment marker inside a quoted
637 string is not a comment), and only quoted string literals count as list
638 entries — so a range mentioned in a comment inside the list is not an allowed
639 range. Values are returned as spelled in the file, so the violation names
640 what is written. Empty list => no unambiguous public exposure.
641 """
642 body = _strip_hcl_comments(ingress_body)
643 found: list[ str ] = []
644 for attr in ( "cidr_blocks" , "ipv6_cidr_blocks" ):
645 inner = _attr_list_inner(body, attr)
646 if inner is None :
647 continue
648 for cidr in re.findall( r '" ([ ^" ] * ) "' , inner):
649 if _is_entire_internet(cidr) and cidr not in found:
650 found.append(cidr)
651 return found
652
653
654 def check_db_sg_no_public_ingress (tf_files: list[tuple[ str , str ]]) -> list[Violation]:
655 """Flag inline security-group ingress that opens a DB port to 0.0.0.0/0 or ::/0.
656
657 Fail-open: only INLINE `ingress { ... }` blocks inside aws_security_group are
658 inspected. Separate aws_security_group_rule / aws_vpc_security_group_ingress_rule
659 resources are not correlated here (this static reader cannot resolve the
660 referenced security_group_id), so they never trigger a false positive.
661 """
662 violations: list[Violation] = []
663 for rel_path, content in tf_files:
664 for name, body, line in _extract_blocks(content, "aws_security_group" ):
665 # Walk each inline ingress block via brace matching.
666 for ingress_body in _own_blocks(body, "ingress" ):
667 public = _ingress_public_cidrs(ingress_body)
668 if _ingress_covers_db_port(ingress_body) and public:
669 violations.append(
670 Violation(
671 check = "policy" ,
672 rule = "db_sg_no_public_ingress" ,
673 file = rel_path,
674 line = line,
675 severity = "error" ,
676 summary = (
677 f "aws_security_group ' { name } ' has an ingress rule that opens a "
678 f "database port (5432/3306) to { ' and ' .join(public) } "
679 ),
680 fix_hint = (
681 "Restrict the ingress to the application security group "
682 "(security_groups = [aws_security_group.app.id]) or a private "
683 "CIDR — never 0.0.0.0/0 or ::/0 for a database port"
684 ),
685 )
686 )
687 return violations
688
689
690 def check_sg_no_public_admin_ingress (tf_files: list[tuple[ str , str ]]) -> list[Violation]:
691 """Flag inline security-group ingress that opens a well-known admin/datastore
692 port (SSH, RDP, Redis, Memcached, Mongo, Elasticsearch, Kibana) to 0.0.0.0/0
693 or ::/0.
694
695 Deliberately scoped to a fixed list of ports that are ~never legitimately
696 public — NOT "any public ingress" — so valid public workloads (web on
697 80/443, game servers on high ports, etc.) are not falsely flagged. DB ports
698 (5432/3306) are handled by db_sg_no_public_ingress and excluded here to avoid
699 double-reporting.
700
701 Fail-open (same scope as db_sg_no_public_ingress): only INLINE ingress blocks
702 inside aws_security_group are inspected; separate rule resources are not
703 correlated.
704 """
705 port_names = {
706 22 : "SSH" , 3389 : "RDP" , 6379 : "Redis" , 11211 : "Memcached" ,
707 27017 : "MongoDB" , 9200 : "Elasticsearch" , 9300 : "Elasticsearch" ,
708 5601 : "Kibana" ,
709 }
710 violations: list[Violation] = []
711 for rel_path, content in tf_files:
712 for name, body, line in _extract_blocks(content, "aws_security_group" ):
713 for ingress_body in _own_blocks(body, "ingress" ):
714 public = _ingress_public_cidrs(ingress_body)
715 if not public:
716 continue
717 hit = _ingress_covered_ports(ingress_body, _SENSITIVE_NONDB_PORTS )
718 if not hit:
719 continue
720 labels = ", " .join( f " { p } ( { port_names[p] } )" for p in sorted ( set (hit)))
721 violations.append(
722 Violation(
723 check = "policy" ,
724 rule = "sg_no_public_admin_ingress" ,
725 file = rel_path,
726 line = line,
727 severity = "error" ,
728 summary = (
729 f "aws_security_group ' { name } ' opens sensitive port(s) { labels } "
730 f "to { ' and ' .join(public) } "
731 ),
732 fix_hint = (
733 "Restrict this ingress to a bastion/app security group or a private "
734 "CIDR; never expose admin or datastore ports to the internet"
735 ),
736 )
737 )
738 return violations
739
740
741 def _iam_key_is_wildcard (body: str , key: str ) -> bool :
742 """True if an IAM policy `key` (Action/Resource) is a sole "*" — string OR
743 list form. Matches:
744 "Action" : "*" (heredoc JSON, string)
745 Action = "*" (jsonencode HCL, string)
746 "Resource" : ["*"] (heredoc JSON, single-element list)
747 Resource = ["*"] (jsonencode HCL, single-element list)
748 A list is only treated as wildcard when "*" is its ONLY element — a list that
749 also contains scoped ARNs/actions is not a blanket wildcard.
750 """
751 # String form: key = "*"
752 if re.search( rf '"? { key } "?\s*[:=]\s*"\*"' , body):
753 return True
754 # List form: key = [ "*" ] with nothing else inside the brackets.
755 m = re.search( rf '"? { key } "?\s*[:=]\s*\[(.*?)\]' , body, re. DOTALL )
756 if m:
757 inner = m.group( 1 ).strip()
758 if inner == '"*"' :
759 return True
760 return False
761
762
763 def _remove_allowed_resource_wildcard_statements (body: str ) -> str :
764 """Remove narrowly-approved IAM statements whose actions require Resource "*".
765
766 Keep this allowlist statement-shaped rather than action-token-shaped so an
767 approved action cannot hide a second, over-broad action in the same statement.
768 """
769 return re.sub(
770 r """
771 \{ \s *
772 " ? Effect" ? \s * [ := ]\s * "Allow" \s * , ? \s *
773 " ? Action" ? \s * [ := ]\s * "elasticbeanstalk:CreateStorageLocation" \s * , ? \s *
774 " ? Resource" ? \s * [ := ]\s * " \* " \s * , ? \s *
775 \}
776 """ ,
777 "" ,
778 body,
779 flags = re. DOTALL | re. VERBOSE ,
780 )
781
782
783 def check_no_wildcard_iam (tf_files: list[tuple[ str , str ]]) -> list[Violation]:
784 """Flag literal IAM policy JSON with an unsafe Allow Action/Resource "*".
785
786 Fail-open: only literal `policy = jsonencode({...})` / heredoc JSON inside
787 aws_iam_policy, aws_iam_role_policy, or *_inline_policy blocks is scanned.
788 aws_iam_policy_document DATA sources are not inspected (their statements are
789 HCL blocks, not literal JSON here) — avoids false positives on the common,
790 reviewable data-source pattern. Assume-role trust policies are excluded
791 (a Service/AWS principal trust with Action sts:AssumeRole is not a wildcard).
792 """
793 violations: list[Violation] = []
794 policy_res_types = ( "aws_iam_policy" , "aws_iam_role_policy" , "aws_iam_group_policy" ,
795 "aws_iam_user_policy" )
796 for rel_path, content in tf_files:
797 for res_type in policy_res_types:
798 for name, body, line in _extract_blocks(content, res_type):
799 # Accept both heredoc JSON ("Effect": "Allow") and HCL jsonencode
800 # ({...}) (Effect = "Allow") forms — the separator is : or =.
801 if not re.search( r '" ? Effect" ? \s * [ := ]\s * "Allow"' , body):
802 continue
803 body_without_allowed = _remove_allowed_resource_wildcard_statements(body)
804 if _iam_key_is_wildcard(body, "Action" ) or _iam_key_is_wildcard(
805 body_without_allowed, "Resource"
806 ):
807 violations.append(
808 Violation(
809 check = "policy" ,
810 rule = "no_wildcard_iam" ,
811 file = rel_path,
812 line = line,
813 severity = "error" ,
814 summary = (
815 f " { res_type } ' { name } ' grants Action \" * \" or Resource \" * \" "
816 "in an Allow statement — over-broad permissions"
817 ),
818 fix_hint = (
819 "Scope the policy to specific actions and resource ARNs; "
820 "isolate any documented action that requires Resource \" * \" "
821 ),
822 )
823 )
824 return violations
825
826
827 def check_rds_encryption_at_rest (tf_files: list[tuple[ str , str ]]) -> list[Violation]:
828 """Flag RDS resources that do not enable storage_encrypted (RDS defaults to
829 UNENCRYPTED).
830
831 Fail-open: fires only when storage_encrypted is literally `false` OR the
832 attribute is absent. A variable-driven value (storage_encrypted = var.x) is
833 NOT flagged. S3 is intentionally excluded — buckets have default SSE-S3 since
834 Jan 2023, so a missing SSE block is not an unencrypted bucket.
835 """
836 violations: list[Violation] = []
837 for rel_path, content in tf_files:
838 for res_type in ( "aws_db_instance" , "aws_rds_cluster" ):
839 for name, body, line in _extract_blocks(content, res_type):
840 val = _attr_string(body, "storage_encrypted" )
841 # Variable-driven / non-literal → attribute present but not "true"/"false".
842 has_attr = _has_own_attr(body, "storage_encrypted" )
843 if has_attr and val is None :
844 continue # variable-driven — fail open
845 if val == "true" :
846 continue
847 violations.append(
848 Violation(
849 check = "policy" ,
850 rule = "rds_encryption_at_rest" ,
851 file = rel_path,
852 line = line,
853 severity = "error" ,
854 summary = (
855 f " { res_type } ' { name } ' does not set storage_encrypted = true — "
856 "RDS storage defaults to unencrypted"
857 ),
858 fix_hint = "Set storage_encrypted = true (optionally with a kms_key_id)" ,
859 )
860 )
861 return violations
862
863
864 def check_elasticache_encryption_at_rest (tf_files: list[tuple[ str , str ]]) -> list[Violation]:
865 """Flag ElastiCache replication groups without at-rest encryption enabled.
866
867 Applies to aws_elasticache_replication_group (the resource that supports
868 at_rest_encryption_enabled). Fires when the attribute is literally `false` or
869 absent; a variable-driven value fails open. aws_elasticache_cluster is NOT
870 checked — standalone Memcached clusters don't support this attribute and
871 Redis-in-cluster is configured via the replication group.
872 """
873 violations: list[Violation] = []
874 for rel_path, content in tf_files:
875 for name, body, line in _extract_blocks(content, "aws_elasticache_replication_group" ):
876 val = _attr_string(body, "at_rest_encryption_enabled" )
877 has_attr = _has_own_attr(body, "at_rest_encryption_enabled" )
878 if has_attr and val is None :
879 continue # variable-driven — fail open
880 if val == "true" :
881 continue
882 violations.append(
883 Violation(
884 check = "policy" ,
885 rule = "elasticache_encryption_at_rest" ,
886 file = rel_path,
887 line = line,
888 severity = "error" ,
889 summary = (
890 f "aws_elasticache_replication_group ' { name } ' does not set "
891 "at_rest_encryption_enabled = true"
892 ),
893 fix_hint = (
894 "Set at_rest_encryption_enabled = true (and consider "
895 "transit_encryption_enabled = true for in-transit protection)"
896 ),
897 )
898 )
899 return violations
900
901
902 def check_elasticache_cluster_encryption (tf_files: list[tuple[ str , str ]]) -> list[Violation]:
903 """Flag single-node Redis aws_elasticache_cluster resources without encryption.
904
905 aws_elasticache_cluster is the standalone (non-replication-group) cache form.
906 Per the AWS provider schema, this resource exposes ONLY
907 transit_encryption_enabled — at_rest_encryption_enabled is NOT a valid
908 argument on aws_elasticache_cluster (it lives on
909 aws_elasticache_replication_group). So for a Redis-engine cluster
910 (engine = "redis") that is NOT a member of a replication group (no
911 replication_group_id) we require transit_encryption_enabled = true, and the
912 fix for at-rest is to move the cache into an encrypted
913 aws_elasticache_replication_group — never to add a nonexistent cluster
914 argument (which would produce Terraform that fails to apply).
915
916 Exempt (fail open), matching the replication-group check's style:
917 - engine == "memcached" (does not support these attributes),
918 - a variable-driven engine or an absent engine (cannot evaluate literally),
919 - a cluster that carries replication_group_id (encryption is enforced on the
920 owning aws_elasticache_replication_group).
921 A variable-driven encryption attribute also fails open.
922 """
923 violations: list[Violation] = []
924 for rel_path, content in tf_files:
925 for name, body, line in _extract_blocks(content, "aws_elasticache_cluster" ):
926 engine = _attr_string(body, "engine" )
927 if engine != "redis" :
928 continue # memcached / variable-driven / absent engine → exempt
929 if _has_own_attr(body, "replication_group_id" ):
930 continue # node of a replication group — encryption enforced there
931 # Only transit_encryption_enabled is a valid argument on this resource.
932 val = _attr_string(body, "transit_encryption_enabled" )
933 has_attr = _has_own_attr(body, "transit_encryption_enabled" )
934 if has_attr and val is None :
935 continue # variable-driven — fail open
936 if val == "true" :
937 continue
938 violations.append(
939 Violation(
940 check = "policy" ,
941 rule = "elasticache_encryption_at_rest" ,
942 file = rel_path,
943 line = line,
944 severity = "error" ,
945 summary = (
946 f "aws_elasticache_cluster ' { name } ' (engine = redis) does "
947 "not set transit_encryption_enabled = true"
948 ),
949 fix_hint = (
950 "Set transit_encryption_enabled = true on the Redis "
951 "aws_elasticache_cluster. For at-rest encryption, move the "
952 "cache into an aws_elasticache_replication_group with "
953 "at_rest_encryption_enabled = true — at_rest_encryption_enabled "
954 "is not a valid argument on aws_elasticache_cluster."
955 ),
956 )
957 )
958 return violations
959
960
961 def validate (terraform_dir: Path) -> tuple[ bool , list[Violation]]:
962 tf_files = _read_tf_files(terraform_dir)
963 if not tf_files:
964 # Preserve the existing no_tf_files verdict path.
965 return False , check_alb_https_policy(terraform_dir)
966
967 violations: list[Violation] = []
968 violations.extend(check_alb_https_policy(terraform_dir))
969 violations.extend(check_rds_not_public(tf_files))
970 violations.extend(check_db_sg_no_public_ingress(tf_files))
971 violations.extend(check_sg_no_public_admin_ingress(tf_files))
972 violations.extend(check_no_wildcard_iam(tf_files))
973 violations.extend(check_rds_encryption_at_rest(tf_files))
974 violations.extend(check_elasticache_encryption_at_rest(tf_files))
975 violations.extend(check_elasticache_cluster_encryption(tf_files))
976 return len (violations) == 0 , violations
977
978
979 def main () -> int :
980 parser = argparse.ArgumentParser( description = "Validate generated Terraform policy rules" )
981 parser.add_argument( "terraform_dir" , type = Path, help = "Path to terraform/ directory" )
982 parser.add_argument(
983 "--json" ,
984 type = Path,
985 default = None ,
986 help = "Optional path to write a machine-readable JSON verdict" ,
987 )
988 args = parser.parse_args()
989
990 terraform_dir = args.terraform_dir.resolve()
991 if not terraform_dir.is_dir():
992 print ( f "POLICY_FAIL | path= { terraform_dir } | reason=not_a_directory" , file = sys.stderr)
993 return 2
994
995 ok, violations = validate(terraform_dir)
996
997 if args.json is not None :
998 report = {
999 "check" : "policy" ,
1000 "policy_status" : "POLICY_OK" if ok else "POLICY_FAIL" ,
1001 "violations" : [asdict(v) for v in violations],
1002 }
1003 try :
1004 args.json.write_text(json.dumps(report, indent = 2 ) + " \n " , encoding = "utf-8" )
1005 except OSError as exc:
1006 print ( f "POLICY_FAIL | reason=json_write_failed | detail= { exc } " , file = sys.stderr)
1007 return 2
1008
1009 checks = (
1010 "alb_https,rds_not_public,db_sg_no_public_ingress,sg_no_public_admin_ingress,"
1011 "no_wildcard_iam,rds_encryption,elasticache_encryption,"
1012 "elasticache_cluster_encryption"
1013 )
1014
1015 if ok:
1016 print ( f "POLICY_OK | checks= { checks } " )
1017 return 0
1018
1019 print ( f "POLICY_FAIL | checks= { checks } " , file = sys.stderr)
1020 for v in violations:
1021 print (
1022 f "POLICY_FAIL | file= { v.file } | line= { v.line } | rule= { v.rule } | reason= { v.summary } " ,
1023 file = sys.stderr,
1024 )
1025 return 1
1026
1027
1028 if __name__ == "__main__" :
1029 sys.exit(main())