def clusters_disable_public_access()

in source/eksfedctl/create_script.py [0:0]

• 31 lines of code
• 1 McCabe index (conditional complexity)

def clusters_disable_public_access(config, clusters):
    metadata = config.yaml["metadata"]
    [region1, region2] = metadata["regions"]

    sg1id = clusters[0]["resourcesVpcConfig"]["clusterSecurityGroupId"]
    sg2id = clusters[1]["resourcesVpcConfig"]["clusterSecurityGroupId"]

    ec2 = boto3.client("ec2", region_name=region1)
    ec2.authorize_security_group_ingress(GroupId=sg1id, IpPermissions=[
        {"IpProtocol": "tcp",
         "FromPort": 443,
         "ToPort": 443,
         "IpRanges": [{"CidrIp": get_cidr_for_vpc(0)},
                      {"CidrIp": get_cidr_for_vpc(1)},
                      {"CidrIp": get_cidr_for_vpc(2)}]}
    ])

    ec2 = boto3.client("ec2", region_name=region2)
    ec2.authorize_security_group_ingress(GroupId=sg2id, IpPermissions=[
        {"IpProtocol": "tcp",
         "FromPort": 443,
         "ToPort": 443,
         "IpRanges": [{"CidrIp": get_cidr_for_vpc(0)},
                      {"CidrIp": get_cidr_for_vpc(1)},
                      {"CidrIp": get_cidr_for_vpc(2)}]}
    ])

    exec_command(["eksctl", "utils", "update-cluster-endpoints",
                  f"--cluster={clusters[0]['name']}", f"--region={region1}",
                  "--public-access=false", "--private-access=true",
                  "--approve"])

    exec_command(["eksctl", "utils", "update-cluster-endpoints",
                  f"--cluster={clusters[1]['name']}", f"--region={region2}",
                  "--public-access=false", "--private-access=true",
                  "--approve"])