Skip to content

ec2

Sample rules

A few rules that use objects from this package:

public_access_security_groups_ssh_port_rule
import logging
from abc import abstractmethod
from typing import List, Dict, Set

from cloudrail.knowledge.context.aliases_dict import AliasesDict
from cloudrail.knowledge.context.aws.aws_environment_context import AwsEnvironmentContext
from cloudrail.knowledge.context.aws.resources.dms.dms_replication_instance import DmsReplicationInstance
from cloudrail.knowledge.context.aws.resources.ec2.network_interface import NetworkInterface
from cloudrail.knowledge.context.aws.resources.ec2.security_group import SecurityGroup
from cloudrail.knowledge.context.aws.resources.ec2.security_group_rule import SecurityGroupRulePropertyType
from cloudrail.knowledge.context.aws.resources.eks.eks_cluster import EksCluster
from cloudrail.knowledge.context.aws.resources.es.elastic_search_domain import ElasticSearchDomain
from cloudrail.knowledge.context.aws.resources.neptune.neptune_cluster import NeptuneCluster
from cloudrail.knowledge.context.aws.resources.neptune.neptune_instance import NeptuneInstance
from cloudrail.knowledge.context.aws.resources.rds.rds_cluster import RdsCluster
from cloudrail.knowledge.context.aws.resources.rds.rds_instance import RdsInstance
from cloudrail.knowledge.context.aws.resources.redshift.redshift import RedshiftCluster
from cloudrail.knowledge.context.connection import ConnectionType, PortConnectionProperty, ConnectionDetail
from cloudrail.knowledge.rules.aws.aws_base_rule import AwsBaseRule
from cloudrail.knowledge.rules.base_rule import Issue
from cloudrail.knowledge.rules.constants.known_ports import KnownPorts
from cloudrail.knowledge.rules.rule_parameters.base_paramerter import ParameterType
from cloudrail.knowledge.utils.port_utils import is_all_ports
from cloudrail.knowledge.utils.utils import is_port_in_range, is_all_ips


class PublicAccessSecurityGroupsPortRule(AwsBaseRule):

    def __init__(self, port: KnownPorts) -> None:
        self.port = port

    @abstractmethod
    def get_id(self) -> str:
        pass

    def should_run_rule(self, environment_context: AwsEnvironmentContext) -> bool:
        return bool(environment_context.get_used_network_interfaces())

    def execute(self, env_context: AwsEnvironmentContext, parameters: Dict[ParameterType, any]) -> List[Issue]:
        eni_list: AliasesDict[NetworkInterface] = env_context.get_used_network_interfaces()
        logging.error(f'len of eni_list {len(eni_list)}')
        self.remove_from_eni_list(eni_list, parameters)
        if self.port.value == KnownPorts.ALL:
            eni_to_sg_map: Dict[NetworkInterface, Set[SecurityGroup]] = self.find_sg_issues(eni_list)
            message: str = ("~Internet~. {0} `{1}` has internet gateway. "
                            "Instance `{2}` is on `{1}`. {0} routes traffic from instance to internet gateway. "
                            "{0} uses Network ACL's `{3}` which allows all ports range. Instance uses security group `{4}`. "
                            "`{4}` allows all ports range. ~Instance~")
            return [
                Issue(
                    message.format(
                        eni.subnet.get_type(),
                        eni.subnet.get_friendly_name(),
                        eni.owner.get_friendly_name(),
                        eni.subnet.network_acl.get_friendly_name(),
                        sg.get_friendly_name()),
                    eni.owner,
                    sg)
                for eni in eni_to_sg_map for sg in eni_to_sg_map[eni]]

        else:
            eni_to_sg_map: Dict[NetworkInterface, Set[SecurityGroup]] = self.find_sg_issues(eni_list)
            message: str = ("~Internet~. {0} `{1}` has internet gateway. "
                            "Instance `{2}` is on `{1}`. {0} routes traffic from instance to internet gateway. "
                            "{0} uses Network ACL's `{3}` which allows port `{4}`. Instance uses security group `{5}`. "
                            "`{5}` allows port `{4}`. ~Instance~")

            return [
                Issue(
                    message.format(
                        eni.subnet.get_type(),
                        eni.subnet.get_friendly_name(),
                        eni.owner.get_friendly_name(),
                        eni.subnet.network_acl.get_friendly_name(),
                        self.port.value,
                        sg.get_friendly_name()),
                    eni.owner,
                    sg)
                for eni in eni_to_sg_map for sg in eni_to_sg_map[eni]]

    @staticmethod
    def remove_from_eni_list(eni_list: AliasesDict[NetworkInterface], parameters: Dict[ParameterType, any]):
        eni_exclude_list = {
            NeptuneCluster,
            NeptuneInstance,
            RedshiftCluster,
            RdsCluster,
            RdsInstance,
            ElasticSearchDomain,
            EksCluster,
            DmsReplicationInstance,
        }
        for ec2 in parameters.get(ParameterType.FIREWALL_EC2, []):
            for eni in ec2.network_resource.network_interfaces:
                eni_list.remove(eni)
        enis_to_delete = [eni for eni in eni_list if type(eni.owner) in eni_exclude_list]
        for eni in enis_to_delete:
            eni_list.remove(eni)

    def find_sg_issues(self, eni_list: AliasesDict[NetworkInterface]) -> Dict[NetworkInterface, Set[SecurityGroup]]:
        eni_to_sg_rules_map: Dict[NetworkInterface, Set[SecurityGroup]] = {}
        for eni in eni_list:
            is_eni_accessible_from_all_ips_on_port = any(
                self._is_public_connection(x) and
                self._is_all_ips_connection(x.connection_property) and
                self._is_connection_open_on_port(x.connection_property)
                for x in eni.inbound_connections
            )

            # Get security group that allows public access (any)
            if is_eni_accessible_from_all_ips_on_port:
                if self.port == KnownPorts.ALL:
                    eni_to_sg_rules_map[eni] = self._get_all_allow_all_port_range_sg(eni)
                else:
                    eni_to_sg_rules_map[eni] = self._get_all_allow_in_bound_port_sg(eni)

        return eni_to_sg_rules_map

    def _get_all_allow_in_bound_port_sg(self, eni: NetworkInterface) -> Set[SecurityGroup]:
        return {sg for sg in eni.security_groups for permission in sg.inbound_permissions
                if permission.is_in_range(self.port.value)
                and permission.property_type == SecurityGroupRulePropertyType.IP_RANGES and
                is_all_ips(permission.property_value)}

    @staticmethod
    def _get_all_allow_all_port_range_sg(eni: NetworkInterface) -> Set[SecurityGroup]:
        return {sg for sg in eni.security_groups for permission in sg.inbound_permissions
                if is_all_ports((permission.from_port, permission.to_port))
                and permission.property_type == SecurityGroupRulePropertyType.IP_RANGES and
                is_all_ips(permission.property_value)}

    @staticmethod
    def _is_public_connection(con_detail: ConnectionDetail):
        return con_detail.connection_type == ConnectionType.PUBLIC

    @staticmethod
    def _is_all_ips_connection(connection_property: PortConnectionProperty):
        return is_all_ips(connection_property.cidr_block)

    def _is_connection_open_on_port(self, connection_property: PortConnectionProperty):
        return any(port_range for port_range in connection_property.ports
                   if (self.port != KnownPorts.ALL and not is_all_ports(port_range) and is_port_in_range(port_range, self.port.value))
                   or (self.port == KnownPorts.ALL and is_all_ports(port_range)))


class PublicAccessSecurityGroupsSshPortRule(PublicAccessSecurityGroupsPortRule):

    def get_id(self) -> str:
        return 'public_access_security_groups_ssh_port_rule'

    def __init__(self):
        super().__init__(KnownPorts.SSH)


class PublicAccessSecurityGroupsRdpPortRule(PublicAccessSecurityGroupsPortRule):

    def get_id(self) -> str:
        return 'public_access_security_groups_rdp_port_rule'

    def __init__(self):
        super().__init__(KnownPorts.RDP)


class PublicAccessSecurityGroupsOracleDbDefaultPortRule(PublicAccessSecurityGroupsPortRule):

    def get_id(self) -> str:
        return 'public_access_security_groups_oracle_db_default_port_rule'

    def __init__(self):
        super().__init__(KnownPorts.ORACLE_DB_DEFAULT)


class PublicAccessSecurityGroupsOracleDbPortRule(PublicAccessSecurityGroupsPortRule):

    def get_id(self) -> str:
        return 'public_access_security_groups_oracle_db_port_rule'

    def __init__(self):
        super().__init__(KnownPorts.ORACLE_DB)


class PublicAccessSecurityGroupsOracleDbSslPortRule(PublicAccessSecurityGroupsPortRule):

    def get_id(self) -> str:
        return 'public_access_security_groups_oracle_db_ssl_port_rule'

    def __init__(self):
        super().__init__(KnownPorts.ORACLE_DB_SSL)


class PublicAccessSecurityGroupsMySqlPortRule(PublicAccessSecurityGroupsPortRule):

    def get_id(self) -> str:
        return 'public_access_security_groups_mysql_port_rule'

    def __init__(self):
        super().__init__(KnownPorts.MYSQL)


class PublicAccessSecurityGroupsPostgresPortRule(PublicAccessSecurityGroupsPortRule):

    def get_id(self) -> str:
        return 'public_access_security_groups_postgres_port_rule'

    def __init__(self):
        super().__init__(KnownPorts.POSTGRES)


class PublicAccessSecurityGroupsRedisPortRule(PublicAccessSecurityGroupsPortRule):

    def get_id(self) -> str:
        return 'public_access_security_groups_redis_port_rule'

    def __init__(self):
        super().__init__(KnownPorts.REDIS)


class PublicAccessSecurityGroupsMongodbPortRule(PublicAccessSecurityGroupsPortRule):

    def get_id(self) -> str:
        return 'public_access_security_groups_mongodb_port_rule'

    def __init__(self):
        super().__init__(KnownPorts.MONGODB)


class PublicAccessSecurityGroupsMongodbShardClusterPortRule(PublicAccessSecurityGroupsPortRule):

    def get_id(self) -> str:
        return 'public_access_security_groups_mongodb_shard_cluster_port_rule'

    def __init__(self):
        super().__init__(KnownPorts.MONGODB_SHARD_CLUSTER)


class PublicAccessSecurityGroupsCassandraPortRule(PublicAccessSecurityGroupsPortRule):

    def get_id(self) -> str:
        return 'public_access_security_groups_cassandra_port_rule'

    def __init__(self):
        super().__init__(KnownPorts.CASSANDRA)


class PublicAccessSecurityGroupsCassandraThriftPortRule(PublicAccessSecurityGroupsPortRule):

    def get_id(self) -> str:
        return 'public_access_security_groups_cassandra_thrift_port_rule'

    def __init__(self):
        super().__init__(KnownPorts.CASSANDRA_THRIFT)


class PublicAccessSecurityGroupsCassandraMngPortRule(PublicAccessSecurityGroupsPortRule):

    def get_id(self) -> str:
        return 'public_access_security_groups_cassandra_mng_port_rule'

    def __init__(self):
        super().__init__(KnownPorts.CASSANDRA_MNG)


class PublicAccessSecurityGroupsMemcachedPortRule(PublicAccessSecurityGroupsPortRule):

    def get_id(self) -> str:
        return 'public_access_security_groups_memcached_port_rule'

    def __init__(self):
        super().__init__(KnownPorts.MEMCACHED)


class PublicAccessSecurityGroupsElasticsearchNodesPortRule(PublicAccessSecurityGroupsPortRule):

    def get_id(self) -> str:
        return 'public_access_security_groups_elasticsearch_nodes_port_rule'

    def __init__(self):
        super().__init__(KnownPorts.ELASTICSEARCH_NODES)


class PublicAccessSecurityGroupsElasticsearchPortRule(PublicAccessSecurityGroupsPortRule):

    def get_id(self) -> str:
        return 'public_access_security_groups_elasticsearch_port_rule'

    def __init__(self):
        super().__init__(KnownPorts.ELASTICSEARCH)


class PublicAccessSecurityGroupsKibanaPortRule(PublicAccessSecurityGroupsPortRule):

    def get_id(self) -> str:
        return 'public_access_security_groups_kibana_port_rule'

    def __init__(self):
        super().__init__(KnownPorts.KIBANA)


class PublicAccessSecurityGroupsAllPortsRule(PublicAccessSecurityGroupsPortRule):

    def get_id(self) -> str:
        return 'public_access_security_groups_all_ports_rule'

    def __init__(self):
        super().__init__(KnownPorts.ALL)
vpc_endpoint_s3_exposure
from abc import abstractmethod
from typing import List, Dict
from cloudrail.knowledge.context.mergeable import Mergeable
from cloudrail.knowledge.context.aws.resources.ec2.network_interface import NetworkInterface
from cloudrail.knowledge.context.aws.resources.prefix_lists import PrefixLists, PrefixList
from cloudrail.knowledge.context.aws.resources.ec2.route import Route
from cloudrail.knowledge.context.aws.aws_environment_context import AwsEnvironmentContext
from cloudrail.knowledge.context.aws.resources.service_name import AwsServiceType
from cloudrail.knowledge.rules.aws.context_aware.vpc_endpoints.abstract_vpc_endpoint_gateway_rule import AbstractVpcEndpointGatewayRule
from cloudrail.knowledge.rules.base_rule import Issue
from cloudrail.knowledge.rules.rule_parameters.base_paramerter import ParameterType



class AbstractVpcEndpointGatewayIsNotUsedRule(AbstractVpcEndpointGatewayRule):

    @abstractmethod
    def get_id(self) -> str:
        pass

    def execute(self, env_context: AwsEnvironmentContext, parameters: Dict[ParameterType, any]) -> List[Issue]:
        vpc_list, region_to_service_map, vpc_to_eni_map = self._init_maps(env_context)
        issues_list: List[Issue] = []
        region_to_prefix_lists_map: Dict[str, PrefixLists] = self._create_prefix_list_by_region_map(env_context)
        for vpc in vpc_list:
            for eni in vpc_to_eni_map.get(vpc, []):
                if self._is_service_eni_match(eni):
                    prefix_list: PrefixLists = region_to_prefix_lists_map[vpc.region]
                    aws_service_pl: PrefixList = prefix_list.get_prefix_lists_by_service(self.aws_service_type.value)
                    if self._add_new_issue_from_outbound(eni, region_to_service_map, issues_list, aws_service_pl):
                        break

        return issues_list

    def _add_new_issue_from_outbound(self, eni: NetworkInterface, region_to_service_map: Dict[str, List[Mergeable]],
                                     issues_list: List[Issue], service_pl: PrefixList) -> bool:
        if self._is_public_connection_exist(eni):
            most_specific_service_pl_route: Route = self._get_most_specific_service_pl_route(eni.subnet.route_table, service_pl)
            if not self._is_valid_vpc_endpoint_route(most_specific_service_pl_route, service_pl, eni.vpc.endpoints):
                for service in region_to_service_map[eni.vpc.region]:
                    issues_list.append(Issue(f"~The {eni.vpc.get_type()}~. `{eni.vpc.get_friendly_name()}` in region `{eni.vpc.region}`"
                                             f" is in use but not leveraging {self.aws_service_type.name} Endpoint Gateway", service, eni.vpc))
                return True
        return False


class S3VpcEndpointGatewayNotUsedRule(AbstractVpcEndpointGatewayIsNotUsedRule):

    def __init__(self) -> None:
        super().__init__(AwsServiceType.S3, (443, 80), self.S3_SERVICES_EXCLUDE_LIST, False)

    def get_id(self) -> str:
        return "vpc_endpoint_s3_exposure"


class DynamoDbVpcEndpointGatewayNotUsedRule(AbstractVpcEndpointGatewayIsNotUsedRule):

    def __init__(self) -> None:
        super().__init__(AwsServiceType.DYNAMODB, (443,), self.DYNAMODB_SERVICES_INCLUDE_LIST, True)

    def get_id(self) -> str:
        return "vpc_endpoint_dynamodb_exposure"
ec2_role_share_rule
from typing import List, Dict

from cloudrail.knowledge.rules.aws.aws_base_rule import AwsBaseRule
from cloudrail.knowledge.rules.base_rule import Issue
from cloudrail.knowledge.rules.rule_parameters.base_paramerter import ParameterType
from cloudrail.knowledge.context.aws.resources.ec2.ec2_instance import Ec2Instance
from cloudrail.knowledge.context.aws.aws_environment_context import AwsEnvironmentContext


class Ec2RoleShareRule(AwsBaseRule):

    def get_id(self) -> str:
        return 'ec2_role_share_rule'

    def execute(self, env_context: AwsEnvironmentContext, parameters: Dict[ParameterType, any]) -> List[Issue]:
        issues: List[Issue] = []

        ec2s: List[Ec2Instance] = env_context.ec2s
        profile_to_public_ec2 = {}
        for public_ec2 in (x for x in ec2s if x.network_resource.is_inbound_public and x.iam_profile_name):
            profile_to_public_ec2[public_ec2.iam_profile_name] = public_ec2
        for private_ec2 in (x for x in ec2s if not x.network_resource.is_inbound_public and x.iam_profile_name):
            public_ec2 = profile_to_public_ec2.get(private_ec2.iam_profile_name)
            profile = private_ec2.iam_role.get_friendly_name() \
                if private_ec2.iam_role  \
                else private_ec2.iam_profile_name
            if public_ec2:
                issues.append(
                    Issue(
                        f"~Instance `{public_ec2.get_friendly_name()}`~. Instance is publicly exposed. "
                        f"Instance uses IAM role `{profile}`. "
                        f"Private EC2 instance shares IAM role `{profile}` as well. "
                        f"~Instance `{private_ec2.get_friendly_name()}`~",
                        private_ec2,
                        private_ec2.iam_role))
        return issues

    def should_run_rule(self, environment_context: AwsEnvironmentContext) -> bool:
        return bool(environment_context.ec2s)
car_unused_security_group
from typing import Dict, List

from cloudrail.knowledge.context.aws.aws_environment_context import AwsEnvironmentContext
from cloudrail.knowledge.rules.base_rule import BaseRule, Issue
from cloudrail.knowledge.rules.rule_parameters.base_paramerter import ParameterType


class EnsureNoUnusedSecurityGroups(BaseRule):

    def get_id(self) -> str:
        return 'car_unused_security_group'

    def execute(self, env_context: AwsEnvironmentContext, parameters: Dict[ParameterType, any]) -> List[Issue]:
        issues: List[Issue] = []
        # Currently, we catch only security groups which are exists, and not ones which are being created.
        # This is in order to avoid scenario in which a security group created, and will be associated using a different infra than TF.
        # In the future, we will add history track for resources, and this condition will not be needed.
        for security_group in [sg for sg in env_context.security_groups if
                               not sg.is_used
                               and not sg.is_new_resource()
                               and not sg.is_pseudo]:
            issues.append(
                Issue(
                    f'The {security_group.get_type()} `{security_group.get_friendly_name()}` is not used by any AWS resource'
                    , security_group, security_group))
        return issues

    def get_needed_parameters(self) -> List[ParameterType]:
        return []

    def should_run_rule(self, environment_context: AwsEnvironmentContext) -> bool:
        return bool(environment_context.security_groups)
no_vpc_peering_allowed_rule
from typing import List, Dict

from cloudrail.knowledge.rules.aws.aws_base_rule import AwsBaseRule
from cloudrail.knowledge.rules.base_rule import Issue
from cloudrail.knowledge.rules.rule_parameters.base_paramerter import ParameterType
from cloudrail.knowledge.context.aws.resources.ec2.route_table import RouteTargetType
from cloudrail.knowledge.context.aws.resources.ec2.subnet import Subnet
from cloudrail.knowledge.context.aws.aws_environment_context import AwsEnvironmentContext


class NoVpcPeeringAllowedRule(AwsBaseRule):

    def get_id(self) -> str:
        return 'no_vpc_peering_allowed_rule'

    def execute(self, env_context: AwsEnvironmentContext, parameters: Dict[ParameterType, any]) -> List[Issue]:
        peering_connections = self._check_for_peering_connections(env_context)
        issues = []
        for subnet, peerings in peering_connections.items():
            issues.append(Issue(f'{subnet.get_type()}: {subnet.get_friendly_name()} from {subnet.vpc.get_type()}: '
                                f'{subnet.vpc.get_friendly_name()} is using the following {subnet.vpc.get_type()}'
                                f'Peering connections: {peerings}', None, None))
        return issues

    @classmethod
    def _check_for_peering_connections(cls, env_context: AwsEnvironmentContext) -> Dict[Subnet, str]:
        peering_map = {}
        for subnet in env_context.subnets:
            peering_connection_routes = [x for x in subnet.route_table.routes if
                                         x.target_type == RouteTargetType.VPC_PEERING_ID]
            if peering_connection_routes:
                peering_map[subnet] = ', '.join(x.destination for x in peering_connection_routes)

        return peering_map

    def should_run_rule(self, environment_context: AwsEnvironmentContext) -> bool:
        return bool(environment_context.peering_connections)

Ec2Image (AwsResource)

Attributes:

Name Type Description
image_id str

The ID of the EC2 image (AMI ID).

is_public bool

True if the image is publicly shared.

custom_invalidation(self) inherited

A list of manual reasons why this resource should be invalidated

exclude_from_invalidation(self) inherited

A list of attributes that should be excluded from the invalidation process

AssociatePublicIpAddress (str, Enum)

An enumeration.

Ec2Instance (NetworkEntity, AwsClient)

Attributes:

Name Type Description
instance_id str

The ID of the instance.

name

The name of the EC2 instance, if set.

network_interfaces_ids List[str]

The network interfaces attached to the intance.

state str

The state of the instance.

image_id str

The ID of the AMI used for EC2.

image_data Optional[Ec2Image]

A pointer to the Ec2Image if found.

iam_profile_arn

The IAM profile assigned to this image, if one is assigned.

iam_profile_id

The ID of the IAM profile.

http_tokens str

The HTTP tokens setting - optional or required.

availability_zone Optional[str]

The availability zone the EC2 is in, if configured.

instance_type str

The Instance type (i.e. 'm5.8xlarge').

ebs_optimized bool

Indication whether the EC2 instance has EBS optimization enabled of not.

monitoring_enabled bool

Indication if the launched EC2 instance will have detailed monitoring enabled.

custom_invalidation(self) inherited

A list of manual reasons why this resource should be invalidated

exclude_from_invalidation(self) inherited

A list of attributes that should be excluded from the invalidation process

Ec2RawData dataclass

Internal implementation detail, ignore.

ElasticIp (AwsResource)

Attributes:

Name Type Description
allocation_id str

The ID of the elastic IP's allocation.

public_ip str

The public IP of the elastic IP. May be "0.0.0.0" to denote that we do not know what it is (usually when the resource is still being built).

private_ip Optional[str]

The private IP of the elastic IP, may be None.

custom_invalidation(self) inherited

A list of manual reasons why this resource should be invalidated

exclude_from_invalidation(self) inherited

A list of attributes that should be excluded from the invalidation process

InternetGateway (AwsResource)

Attributes:

Name Type Description
vpc_id str

The ID of the VPC the IGW belongs to.

igw_id str

The ID of the IGW.

igw_type IgwType

The type of the IGW.

custom_invalidation(self) inherited

A list of manual reasons why this resource should be invalidated

exclude_from_invalidation(self) inherited

A list of attributes that should be excluded from the invalidation process

MainRouteTableAssociation (AwsResource)

Attributes:

Name Type Description
vpc_id

The VPC the route table is the main one for.

route_table_id

The ID of the route table that is to be the main one for the VPC.

custom_invalidation(self) inherited

A list of manual reasons why this resource should be invalidated

exclude_from_invalidation(self) inherited

A list of attributes that should be excluded from the invalidation process

NatGateways (NetworkEntity)

Attributes:

Name Type Description
nat_gateway_id str

The ID of this NAT gateway.

allocation_id str

The allocation ID used with this NAT gateway.

subnet_id str

The subnet the NAT is tired to.

eni_id str

The elastic network interface the NAT gateway is tied to.

private_ip str

The private IP of the NAT gateway.

public_ip str

The public IP of the NAT gateway.

custom_invalidation(self) inherited

A list of manual reasons why this resource should be invalidated

exclude_from_invalidation(self) inherited

A list of attributes that should be excluded from the invalidation process

NetworkAcl (AwsResource)

Attributes:

Name Type Description
network_acl_id str

The ID of the NACL.

vpc_id str

The ID of the VPC the NACL belongs to.

is_default bool

True if this is the default NACL for the subnets.

name str

The name of the NACL.

subnet_ids List[str]

List of IDs of subnets the NACL bleongs to.

inbound_rules List[NetworkAclRule]

The inbound/ingress rules defined in the NACL.

outbound_rules List[NetworkAclRule]

The outbound/egress rules defined in the NACL.

custom_invalidation(self) inherited

A list of manual reasons why this resource should be invalidated

exclude_from_invalidation(self) inherited

A list of attributes that should be excluded from the invalidation process

NetworkAclRule (AwsResource)

Attributes:

Name Type Description
network_acl_id

The ID of the NACL this rule belongs to.

cidr_block

The CIDR block the rule applies to.

from_port

The bottom of the port range applicable to the rule.

to_port

The top of the port range applicable to the rule.

rule_action

The action the rule takes (allow / deny).

rule_number

The number of the rule in the NACL table.

rule_type

The type of the rule - inbound or outbound.

ip_protocol_type IpProtocol

The IP protocol the rule applies to.

custom_invalidation(self) inherited

A list of manual reasons why this resource should be invalidated

exclude_from_invalidation(self) inherited

A list of attributes that should be excluded from the invalidation process

RuleAction (Enum)

An enumeration.

RuleType (Enum)

An enumeration.

NetworkInterface (ConnectionInstance, AwsResource)

Represents a network interface that can be assigned to a specific network-bound resource. Sometimes NetworkInterfaces may be auto-generated by Cloudrail to provide more data in the context.

Attributes:

Name Type Description
eni_id str

The ID of the elastic network interface.

subnet_id str

The ID of the subnet it's attached to.

subnet 'Subnet'

The actual Subnet object if found.

primary_ip_address str

The primary IP address attached to the interface.

secondary_ip_addresses List[str]

List of secondary IP addresses attached to the interface, if any exist.

public_ip_address Optional[str]

The public IP address of the interface, if it has one.

ipv6_ip_addresses List[str]

The IPv6 addresses of the interface, if they are configured.

security_groups_ids List[str]

The security groups attached to the interface

security_groups List['SecurityGroup']

The actual SGs the interface uses.

description str

The description set for the interface, if any.

is_primary bool

True if it's the primary interface for the resource it is attached to.

availability_zone str

The AZ this interface is in.

owner Optional[AwsResource]

The resource that owns this interface.

custom_invalidation(self) inherited

A list of manual reasons why this resource should be invalidated

exclude_from_invalidation(self) inherited

A list of attributes that should be excluded from the invalidation process

PeeringConnection (AwsResource)

Attributes:

Name Type Description
peering_id str

The ID of the peering connection.

requester_vpc_info PeeringVpcInfo

The information of the VPC that initiated the peering.

accepter_vpc_info PeeringVpcInfo

The information of the VPC that received and accepted the peering.

status str

The status of the peering connection.

custom_invalidation(self) inherited

A list of manual reasons why this resource should be invalidated

exclude_from_invalidation(self) inherited

A list of attributes that should be excluded from the invalidation process

PeeringVpcInfo dataclass

Attributes:

Name Type Description
vpc_id str

The ID of the VPC (it may be requester or accepter, depending on what side this is on).

cidr_blocks List[str]

The CIDR bblocks exposed by the VPC to the peer.

Route (AwsResource)

Attributes:

Name Type Description
route_table_id

The ID of the routing table the route belongs to.

target

The target of the route (value depends on type).

target_type

The type of the route's target.

destination

The destination subnet defined for the route.

peering_connection Optional[PeeringConnection]

If hte target is a VPC_PEERING_ID, then this is a pointer to the matching connection.

custom_invalidation(self) inherited

A list of manual reasons why this resource should be invalidated

exclude_from_invalidation(self) inherited

A list of attributes that should be excluded from the invalidation process

RouteTargetType (str, Enum)

An enumeration.

RouteTable (AwsResource)

Attributes:

Name Type Description
route_table_id str

The ID of the routing table.

vpc_id str

The VPC the RT belongs to.

name str

The name of the RT.

routes List[Route]

A list of routes in this table.

is_main_route_table bool

A flag indicating this is the VPC's main route table.

custom_invalidation(self) inherited

A list of manual reasons why this resource should be invalidated

exclude_from_invalidation(self) inherited

A list of attributes that should be excluded from the invalidation process

RouteTableAssociation (AwsResource)

Attributes:

Name Type Description
subnet_id str

The ID of the subnet to associate the route table to.

route_table_id str

The route table to associate to the subnet.

custom_invalidation(self) inherited

A list of manual reasons why this resource should be invalidated

exclude_from_invalidation(self) inherited

A list of attributes that should be excluded from the invalidation process

SecurityGroup (AwsResource)

Attributes:

Name Type Description
security_group_id str

The ID of the security group.

name str

The name of the security group.

vpc_id str

The VPC the SG belongs to.

inbound_permissions List[SecurityGroupRule]

The inbound rules included in the security group.

outbound_permissions List[SecurityGroupRule]

The outbound rules included in the security group.

is_default

True if this is the default SG in the VPC.

has_description bool

True if this SG has a description configured that is not one of the pre-canned ones (like "Managed by Terraform").

_used_by Set[AwsResource]

A set of resources that use this security group.

is_used: bool property readonly

This property returns true if any resource is using this security group, including invalidated resources.

Note: Network interfaces that use this security group, but are not attached to a resource, will not count.

custom_invalidation(self) inherited

A list of manual reasons why this resource should be invalidated

exclude_from_invalidation(self)

A list of attributes that should be excluded from the invalidation process

get_rule_matches(sg_rules1, sg_rules2) staticmethod

Finds the overlapping rules (rules from one SG that match the other).

ConnectionType (str, Enum)

An enumeration.

SecurityGroupRule (AwsResource)

Attributes:

Name Type Description
from_port int

The bottom part of the port range the rule applies to.

to_port int

The top part of te port range the rule applies to.

ip_protocol IpProtocol

The IP protocol used in the rule.

property_type SecurityGroupRulePropertyType

The type of the rule, depending if it's targeting an IP destination, another security gruop, or a prefix list.

property_value str

If the type is SECURITY_GROUP_ID, then this is the GroupId. If the type is IP_RANGES, then this is the CIDR block. If the type is PREFIX_LIST_ID, then this is the Prefix List ID.

has_description bool

True if the rule has a description set that is not a canned one (like "Managed by Terraform").

connection_type ConnectionType

The type of the rule - inbound or outbound.

security_group_id str

The SG the rule belongs to.

custom_invalidation(self) inherited

A list of manual reasons why this resource should be invalidated

exclude_from_invalidation(self) inherited

A list of attributes that should be excluded from the invalidation process

SecurityGroupRulePropertyType (Enum)

An enumeration.

Subnet (AwsResource)

Attributes:

Name Type Description
subnet_id str

The ID of the subnet.

vpc_id str

The ID of the VPC the subnet belongs to.

vpc 'Vpc'

The VPC the subnet bleongs to.

cidr_block str

The subnet's CIDR block.

name str

The name of the subnet.

availability_zone str

The AZ the subnet is in.

map_public_ip_on_launch

True if resources should have a public IP assigned to them upon launch.

is_default bool

True if it's the default subnet of a VPC in an AZ.

route_table RouteTable

The main route table associated with this subnet.

network_acl NetworkAcl

The main NACL associated with this subnet.

custom_invalidation(self) inherited

A list of manual reasons why this resource should be invalidated

exclude_from_invalidation(self) inherited

A list of attributes that should be excluded from the invalidation process

TransitGateway (AwsResource)

Attributes:

Name Type Description
name str

The name of the Transit Gateway.

tgw_id str

The Transit Gateway's ID.

state str

The state of the TGW, one of available | deleted | deleting | modifying | pending.

route_tables List[TransitGatewayRouteTable]

The routing tables connected to this transit gateway.

custom_invalidation(self) inherited

A list of manual reasons why this resource should be invalidated

exclude_from_invalidation(self) inherited

A list of attributes that should be excluded from the invalidation process

TransitGatewayRoute (AwsResource)

Attributes:

Name Type Description
destination_cidr_block str

The destination CIDR block the route is targeting.

state TransitGatewayRouteState

The state of the route (active or blackhole).

route_type TransitGatewayRouteType

The type of the route (static or propagated).

route_table_id str

The routing table the route belongs to.

custom_invalidation(self) inherited

A list of manual reasons why this resource should be invalidated

exclude_from_invalidation(self) inherited

A list of attributes that should be excluded from the invalidation process

TransitGatewayRouteState (Enum)

An enumeration.

TransitGatewayRouteType (Enum)

An enumeration.

TransitGatewayRouteAttachment (AwsResource)

Attributes:

Name Type Description
tgw_id str

The Transit Gateway the route is to be attached to.

resource_type TransitGatewayResourceType

The type of the resource attached to the TGW.

resource_id str

The ID of the resource attached to the TGW.

custom_invalidation(self) inherited

A list of manual reasons why this resource should be invalidated

exclude_from_invalidation(self) inherited

A list of attributes that should be excluded from the invalidation process

TransitGatewayRouteTable (AwsResource)

Attributes:

Name Type Description
tgw_id str

The TGW the route table belongs to.

route_table_id str

The id of the route table.

associations List[TransitGatewayRouteTableAssociation]

A list of route table to TGW associations.

routes List[TransitGatewayRoute]

The routes included in this route table.

custom_invalidation(self) inherited

A list of manual reasons why this resource should be invalidated

exclude_from_invalidation(self) inherited

A list of attributes that should be excluded from the invalidation process

TransitGatewayRouteTableAssociation (AwsResource)

Attributes:

Name Type Description
tgw_attachment_id str

The ID of the TGW attachment.

tgw_route_table_id str

The route table to associate.

attachment TransitGatewayVpcAttachment

The actual TGW attachment object.

custom_invalidation(self) inherited

A list of manual reasons why this resource should be invalidated

exclude_from_invalidation(self) inherited

A list of attributes that should be excluded from the invalidation process

TransitGatewayRouteTablePropagation (AwsResource)

Attributes:

Name Type Description
tgw_attachment_id str

The TGW attachment ID.

tgw_route_table_id str

The ID of the route table attached.

attachment TransitGatewayVpcAttachment

The actual attachment.

custom_invalidation(self) inherited

A list of manual reasons why this resource should be invalidated

exclude_from_invalidation(self) inherited

A list of attributes that should be excluded from the invalidation process

TransitGatewayVpcAttachment (AwsResource)

Attributes:

Name Type Description
attachment_id str

The ID of the attachment.

state str

The state of the attachment, one of pending | failing | failed | available | deleting | deleted |modifying | rolling-back.

resource_type TransitGatewayResourceType

The type of the resource attached.

resource_id str

The ID of the resource attached.

name str

The name of the attachment.

subnet_ids List[str]

The IDs of the subnets attached to the transit gateway.

custom_invalidation(self) inherited

A list of manual reasons why this resource should be invalidated

exclude_from_invalidation(self) inherited

A list of attributes that should be excluded from the invalidation process