audit-labs/audit-tools

 1"""
 2Collect security-group ingress rules open to the internet, across all regions.
 3
 4Only rules allowing 0.0.0.0/0 or ::/0 are reported — one row per open rule.
 5"""
 6
 7import sys
 8
 9from botocore.exceptions import ClientError
10
11from .api import enabled_regions
12
13OPEN_V4 = "0.0.0.0/0"
14OPEN_V6 = "::/0"
15
16
17def security_groups(cfg):
18    session = cfg["session"]
19    rows = []
20    for region in enabled_regions(cfg):
21        try:
22            ec2 = session.client("ec2", region_name=region)
23            groups = _all_groups(ec2)
24        except ClientError as e:
25            print(f"  Skipping {region}: ec2 returned {e}", file=sys.stderr)
26            continue
27        for sg in groups:
28            rows.extend(_open_rules(region, sg))
29    return rows
30
31
32def _all_groups(ec2):
33    groups = []
34    for page in ec2.get_paginator("describe_security_groups").paginate():
35        groups.extend(page.get("SecurityGroups", []))
36    return groups
37
38
39def _open_rules(region, sg):
40    rows = []
41    for perm in sg.get("IpPermissions", []):
42        open_to = [
43            r["CidrIp"] for r in perm.get("IpRanges", []) if r.get("CidrIp") == OPEN_V4
44        ]
45        open_to += [
46            r["CidrIpv6"]
47            for r in perm.get("Ipv6Ranges", [])
48            if r.get("CidrIpv6") == OPEN_V6
49        ]
50        if not open_to:
51            continue
52        rows.append(
53            {
54                "region": region,
55                "group_id": sg.get("GroupId", ""),
56                "group_name": sg.get("GroupName", ""),
57                "protocol": perm.get("IpProtocol", ""),
58                "from_port": perm.get("FromPort", "all"),
59                "to_port": perm.get("ToPort", "all"),
60                "open_to": ", ".join(open_to),
61            }
62        )
63    return rows