Security Group Audit — Find Unrestricted Inbound Rules (0.0.0.0/0)

import boto3
import json
from botocore.exceptions import ClientError


# Ports that should NEVER be open to 0.0.0.0/0
SENSITIVE_PORTS = {
    22:    "SSH",
    3389:  "RDP",
    3306:  "MySQL",
    5432:  "PostgreSQL",
    27017: "MongoDB",
    6379:  "Redis",
    9200:  "Elasticsearch",
    8080:  "HTTP-Alt",
    443:   "HTTPS",
    80:    "HTTP",
}

# These ports are always CRITICAL — direct admin or database access
HIGH_RISK_PORTS = {22, 3389, 3306, 5432, 27017, 6379}


def audit_security_groups(
    region: str = "us-east-1",
    auto_remediate: bool = False,
) -> list[dict]:
    """
    Iterates all Security Groups in the region and checks each
    inbound rule (IpPermissions) for unrestricted CIDR ranges.

    IpPermissions structure:
    [
      {
        "IpProtocol": "tcp",
        "FromPort": 22,
        "ToPort": 22,
        "IpRanges": [{"CidrIp": "0.0.0.0/0", "Description": ""}],
        "Ipv6Ranges": [{"CidrIpv6": "::/0"}],
        "UserIdGroupPairs": [],   # Cross-account SG references
        "PrefixListIds": []
      },
      ...
    ]

    IpProtocol = "-1" means ALL traffic (any protocol, any port).
    """
    ec2 = boto3.client("ec2", region_name=region)
    findings: list[dict] = []

    paginator = ec2.get_paginator("describe_security_groups")
    for page in paginator.paginate():
        for sg in page["SecurityGroups"]:
            sg_id   = sg["GroupId"]
            sg_name = sg["GroupName"]
            vpc_id  = sg.get("VpcId", "EC2-Classic")

            # ── Check each inbound rule ────────────────────────────
            for rule in sg.get("IpPermissions", []):
                from_port = rule.get("FromPort", 0)
                to_port   = rule.get("ToPort",   65535)
                protocol  = rule.get("IpProtocol", "-1")

                # Collect all unrestricted CIDR sources (IPv4 + IPv6)
                unrestricted_cidrs = []
                for cidr_range in rule.get("IpRanges", []):
                    if cidr_range.get("CidrIp") == "0.0.0.0/0":
                        unrestricted_cidrs.append("0.0.0.0/0")
                for ipv6_range in rule.get("Ipv6Ranges", []):
                    if ipv6_range.get("CidrIpv6") == "::/0":
                        unrestricted_cidrs.append("::/0")

                if not unrestricted_cidrs:
                    continue   # This rule is restricted to specific CIDRs — safe

                # ── Check each sensitive port against this rule ────
                for port, service in SENSITIVE_PORTS.items():
                    # protocol == "-1" means ALL traffic (matches every port)
                    # Otherwise, check if the port falls within [from_port, to_port]
                    port_exposed = (
                        protocol == "-1"
                        or (from_port <= port <= to_port)
                    )

                    if not port_exposed:
                        continue

                    severity = "CRITICAL" if port in HIGH_RISK_PORTS else "HIGH"

                    for cidr in unrestricted_cidrs:
                        finding = {
                            "sg_id":    sg_id,
                            "sg_name":  sg_name,
                            "vpc_id":   vpc_id,
                            "port":     port,
                            "service":  service,
                            "protocol": protocol,
                            "cidr":     cidr,
                            "severity": severity,
                        }
                        findings.append(finding)

                        icon = "🔴" if severity == "CRITICAL" else "🟠"
                        print(
                            f"{icon} [{severity}] {sg_id} ({sg_name}) in {vpc_id} "
                            f"allows {service} port {port} from {cidr}"
                        )

                        # ── Auto-remediate critical findings ───────
                        if auto_remediate and port in HIGH_RISK_PORTS:
                            revoke_rule(ec2, sg_id, rule, cidr)

    # ── Summary ────────────────────────────────────────────────────
    critical = [f for f in findings if f["severity"] == "CRITICAL"]
    high     = [f for f in findings if f["severity"] == "HIGH"]

    print(f"\n{'='*60}")
    print(f"Total findings:  {len(findings)}")
    print(f"Critical:        {len(critical)}")
    print(f"High:            {len(high)}")

    if critical:
        print("\nCRITICAL — Immediate action required:")
        seen = set()
        for f in critical:
            key = f"{f['sg_id']}:{f['port']}"
            if key not in seen:
                seen.add(key)
                print(f"  {f['sg_id']} ({f['sg_name']}) → {f['service']} from {f['cidr']}")

    return findings


def revoke_rule(ec2_client, sg_id: str, rule: dict, cidr: str) -> None:
    """
    revoke_security_group_ingress() removes specific CIDR ranges from a rule.

    We reconstruct the IpPermissions entry from the original rule dict,
    replacing IpRanges with ONLY the CIDR we want to remove.
    Other CIDRs in the same rule are untouched.

    This is non-destructive — if the rule allows both 0.0.0.0/0 and
    10.0.0.0/8, only 0.0.0.0/0 is revoked; the private range stays.
    """
    try:
        ip_perm = {
            "IpProtocol": rule["IpProtocol"],
            "IpRanges":   [{"CidrIp": cidr}] if cidr != "::/0" else [],
            "Ipv6Ranges":  [{"CidrIpv6": cidr}] if cidr == "::/0" else [],
        }
        # Only include port range if the protocol is not "-1" (all traffic)
        if rule.get("IpProtocol") != "-1":
            ip_perm["FromPort"] = rule.get("FromPort", 0)
            ip_perm["ToPort"]   = rule.get("ToPort",   65535)

        ec2_client.revoke_security_group_ingress(
            GroupId=sg_id,
            IpPermissions=[ip_perm],
        )
        print(f"    ✅ Revoked {cidr} from {sg_id}")
    except ClientError as e:
        print(f"    ❌ Failed to revoke {cidr} from {sg_id}: {e}")


def save_report(findings: list[dict], output: str = "sg_audit_report.json") -> None:
    with open(output, "w") as f:
        json.dump(findings, f, indent=2)
    print(f"\nFull report saved to {output}")


if __name__ == "__main__":
    # Audit-only mode first (auto_remediate=False)
    findings = audit_security_groups(region="ap-south-1", auto_remediate=False)
    save_report(findings)

    # To auto-fix critical findings, change to:
    # audit_security_groups(region="ap-south-1", auto_remediate=True)

# Enable the AWS managed Config rule that detects this automatically
config = boto3.client("config")
config.put_config_rule(
    ConfigRule={
        "ConfigRuleName": "restricted-ssh",
        "Source": {
            "Owner": "AWS",
            "SourceIdentifier": "INCOMING_SSH_DISABLED",
        },
    }
)

SENSITIVE_PORTS = {22: "SSH", 3389: "RDP", 3306: "MySQL", ...}
HIGH_RISK_PORTS = {22, 3389, 3306, 5432, 27017, 6379}

ec2 = boto3.client("ec2", region_name=region)
paginator = ec2.get_paginator("describe_security_groups")
for page in paginator.paginate():
    for sg in page["SecurityGroups"]:

for rule in sg.get("IpPermissions", []):
    from_port = rule.get("FromPort", 0)
    to_port   = rule.get("ToPort",   65535)
    protocol  = rule.get("IpProtocol", "-1")

unrestricted_cidrs = []
for cidr_range in rule.get("IpRanges", []):
    if cidr_range.get("CidrIp") == "0.0.0.0/0":
        unrestricted_cidrs.append("0.0.0.0/0")
for ipv6_range in rule.get("Ipv6Ranges", []):
    if ipv6_range.get("CidrIpv6") == "::/0":
        unrestricted_cidrs.append("::/0")

for port, service in SENSITIVE_PORTS.items():
    port_exposed = (
        protocol == "-1"
        or (from_port <= port <= to_port)
    )
    if not port_exposed:
        continue
    severity = "CRITICAL" if port in HIGH_RISK_PORTS else "HIGH"

ip_perm = {
    "IpProtocol": rule["IpProtocol"],
    "IpRanges":   [{"CidrIp": cidr}] if cidr != "::/0" else [],
    "Ipv6Ranges":  [{"CidrIpv6": cidr}] if cidr == "::/0" else [],
}
if rule.get("IpProtocol") != "-1":
    ip_perm["FromPort"] = rule.get("FromPort", 0)
    ip_perm["ToPort"]   = rule.get("ToPort",   65535)

ec2_client.revoke_security_group_ingress(
    GroupId=sg_id,
    IpPermissions=[ip_perm],
)