Skip to content


Sample rules

A few rules that use objects from this package:

from abc import abstractmethod
from typing import List, Dict
from import PrefixLists, PrefixList
from import Route
from import AwsEnvironmentContext
from import AwsServiceType
from import AbstractVpcEndpointGatewayRule
from cloudrail.knowledge.rules.base_rule import Issue
from cloudrail.knowledge.rules.rule_parameters.base_paramerter import ParameterType

class AbstractVpcEndpointRouteTableExposureRule(AbstractVpcEndpointGatewayRule):

    def get_id(self) -> str:

    def execute(self, env_context: AwsEnvironmentContext, parameters: Dict[ParameterType, any]) -> List[Issue]:
        vpc_list, _, vpc_to_eni_map = self._init_maps(env_context)
        region_to_prefix_lists_map: Dict[str, PrefixLists] = self._create_prefix_list_by_region_map(env_context)
        issues_list: List[Issue] = []

        for vpc in vpc_list:
            for eni in vpc_to_eni_map.get(vpc, []):
                if self._is_service_eni_match(eni):
                    for subnet in vpc.subnets:
                        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)
                        most_specific_service_pl_route: Route = self._get_most_specific_service_pl_route(subnet.route_table, aws_service_pl)
                        if vpc.endpoints and \
                                not (most_specific_service_pl_route and
                                     self._is_valid_vpc_endpoint_route(most_specific_service_pl_route, aws_service_pl, subnet.vpc.endpoints)):
                            issues_list.append(Issue(f"~The {vpc.get_type()}~. `{vpc.get_friendly_name()}` "
                                                     f"has a {aws_service_pl.service_name.upper()} Endpoint gateway "
                                                     f"but `{subnet.get_friendly_name()}` uses `{subnet.route_table.get_friendly_name()}`"
                                                     f", which does not have a route to the Endpoint gateway.", subnet.vpc, subnet.route_table))

        return issues_list

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

class S3VpcEndpointRouteTableExposureRule(AbstractVpcEndpointRouteTableExposureRule):

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

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

class DynamoDbVpcEndpointRouteTableExposureRule(AbstractVpcEndpointRouteTableExposureRule):

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

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

BillingMode (str, Enum)

An enumeration.

DynamoDbTable (AwsResource)


Name Type Description
table_name str

The name of the table.

table_arn str

The ARN of the table.

billing_mode BillingMode


partition_key str

The partition key used.

sort_key str

The sort key used.

write_capacity int

The write capacity configured.

read_capacity int

The read capacity configured.

fields_attributes List[TableField]

The list of table field attributes (may be empty).

kms_key_id Optional[str]

The KMS key ID to use to encrypt this table, if one is used.

kms_data Optional[KmsKey]

The actual KmsKey object referenced by the KMS ID.

server_side_encryption bool

True if SSE is 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

TableField dataclass


Name Type Description
name str

The field's name.

type TableFieldType

The field type (one of B for Byte, N for Number, S for String).

TableFieldType (str, Enum)

An enumeration.