11.3Network Rule Verification

Security Group Testing with Terraform Plan

# tests/test_network_rules.py
import json
import pytest

@pytest.fixture
def tfplan():
    """Load Terraform plan JSON."""
    with open("tfplan.json") as f:
        return json.load(f)

def test_no_public_ingress_on_database_sg(tfplan):
    """Database security groups must not allow ingress from 0.0.0.0/0."""
    for change in tfplan["resource_changes"]:
        if change["type"] == "aws_security_group":
            if "database" in change["address"] or "rds" in change["address"]:
                ingress_rules = change["change"]["after"].get("ingress", [])
                for rule in ingress_rules:
                    cidrs = rule.get("cidr_blocks", [])
                    assert "0.0.0.0/0" not in cidrs, \
                        f"Database SG {change['address']} allows public ingress"
                    assert "::/0" not in cidrs, \
                        f"Database SG {change['address']} allows public IPv6 ingress"

def test_database_sg_only_allows_specific_ports(tfplan):
    """Database security groups should only allow database ports."""
    allowed_ports = {5432, 3306, 27017, 6379}  # Postgres, MySQL, Mongo, Redis

    for change in tfplan["resource_changes"]:
        if change["type"] == "aws_security_group":
            if "database" in change["address"] or "rds" in change["address"]:
                ingress_rules = change["change"]["after"].get("ingress", [])
                for rule in ingress_rules:
                    from_port = rule.get("from_port", 0)
                    to_port = rule.get("to_port", 0)
                    assert from_port in allowed_ports, \
                        f"Database SG allows unexpected port {from_port}"
                    assert from_port == to_port, \
                        f"Database SG has port range {from_port}-{to_port}"

def test_no_unrestricted_egress(tfplan):
    """Security groups should not have unrestricted egress to the internet."""
    for change in tfplan["resource_changes"]:
        if change["type"] == "aws_security_group":
            egress_rules = change["change"]["after"].get("egress", [])
            for rule in egress_rules:
                if "0.0.0.0/0" in rule.get("cidr_blocks", []):
                    # If egress is open, it must not be on all ports
                    assert rule.get("from_port", 0) != 0 or \
                           rule.get("to_port", 0) != 65535, \
                        f"SG {change['address']} has unrestricted egress on all ports"

def test_ssh_restricted_to_vpn(tfplan):
    """SSH access (port 22) must be restricted to VPN CIDR blocks only."""
    vpn_cidrs = ["10.0.0.0/8", "172.16.0.0/12"]

    for change in tfplan["resource_changes"]:
        if change["type"] == "aws_security_group_rule":
            after = change["change"]["after"]
            if after.get("from_port") == 22 and after.get("type") == "ingress":
                cidrs = after.get("cidr_blocks", [])
                for cidr in cidrs:
                    assert cidr in vpn_cidrs, \
                        f"SSH rule in {change['address']} allows access from {cidr}"

VPC Configuration Testing

def test_private_subnets_have_no_public_ips(tfplan):
    """Private subnets must not auto-assign public IPs."""
    for change in tfplan["resource_changes"]:
        if change["type"] == "aws_subnet":
            if "private" in change["address"]:
                map_public = change["change"]["after"].get(
                    "map_public_ip_on_launch", False
                )
                assert map_public is False, \
                    f"Private subnet {change['address']} assigns public IPs"

def test_nacl_denies_known_bad_ports(tfplan):
    """Network ACLs should deny traffic on known dangerous ports."""
    dangerous_ports = [23, 135, 139, 445, 1433, 3389]  # Telnet, RPC, SMB, MSSQL, RDP

    for change in tfplan["resource_changes"]:
        if change["type"] == "aws_network_acl_rule":
            after = change["change"]["after"]
            if after.get("rule_action") == "allow":
                from_port = after.get("from_port", 0)
                to_port = after.get("to_port", 0)
                for port in dangerous_ports:
                    if from_port <= port <= to_port:
                        pytest.fail(
                            f"NACL rule {change['address']} allows "
                            f"dangerous port {port}"
                        )