1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
| # policies/network-security.sentinel
# 確保安全群組不允許來自 0.0.0.0/0 的危險連接埠存取
import "tfplan/v2" as tfplan
import "strings"
# 禁止對外開放的危險連接埠
param restricted_ports default [22, 3389, 3306, 5432, 27017, 6379]
# 取得安全群組規則
get_security_group_rules = func() {
return filter tfplan.resource_changes as _, rc {
rc.type is "aws_security_group_rule" and
rc.mode is "managed" and
(rc.change.actions contains "create" or rc.change.actions contains "update")
}
}
# 取得內嵌規則的安全群組
get_security_groups = func() {
return filter tfplan.resource_changes as _, rc {
rc.type is "aws_security_group" and
rc.mode is "managed" and
(rc.change.actions contains "create" or rc.change.actions contains "update")
}
}
# 檢查 CIDR 是否為開放存取
is_open_cidr = func(cidr) {
return cidr is "0.0.0.0/0" or cidr is "::/0"
}
# 檢查連接埠範圍是否包含受限連接埠
includes_restricted_port = func(from_port, to_port) {
for restricted_ports as port {
if from_port <= port and port <= to_port {
return true
}
}
return false
}
# 驗證安全群組規則
validate_sg_rule = func(rule) {
# 只檢查入站規則
if rule.change.after.type is not "ingress" {
return true
}
cidr_blocks = rule.change.after.cidr_blocks else []
from_port = rule.change.after.from_port else 0
to_port = rule.change.after.to_port else 0
for cidr_blocks as cidr {
if is_open_cidr(cidr) and includes_restricted_port(from_port, to_port) {
print("Security group rule", rule.address,
"allows restricted port range", from_port, "-", to_port,
"from", cidr)
return false
}
}
return true
}
# 驗證內嵌安全群組規則
validate_sg_inline_rules = func(sg) {
ingress_rules = sg.change.after.ingress else []
for ingress_rules as rule {
cidr_blocks = rule.cidr_blocks else []
from_port = rule.from_port else 0
to_port = rule.to_port else 0
for cidr_blocks as cidr {
if is_open_cidr(cidr) and includes_restricted_port(from_port, to_port) {
print("Security group", sg.address,
"has inline rule allowing restricted port range",
from_port, "-", to_port, "from", cidr)
return false
}
}
}
return true
}
# 主要規則
main = rule {
all get_security_group_rules() as _, rule {
validate_sg_rule(rule)
} and
all get_security_groups() as _, sg {
validate_sg_inline_rules(sg)
}
}
|