Adding support for new resource types in check-no-public-access api.

README.md

@@ -185,6 +185,13 @@ Parses IAM identity-based and resource-based policies from AWS CloudFormation te
 | AWS::SNS::TopicPolicy        | x                  | | x |
 | AWS::SecretsManager::ResourcePolicy | x           | | x |
 | AWS::IAM::Role (trust policy) | x | x | x |
+| AWS::S3Tables::TableBucket | | | x |
+| AWS::ApiGateway::RestApi | | | x |
+| AWS::CodeArtifact::Domain | | | x |
+| AWS::Backup::BackupVault | | | x |
+| AWS::CloudTrail::Dashboard | | | x |
+| AWS::CloudTrail::EventDataStore | | | x |
+| AWS::S3Express::AccessPoint | | | x |
 
 ### Intrinsic function and Pseudo parameter support
 

cfn_policy_validator/canonical_user_id.py

@@ -11,7 +11,7 @@
 
 # Resolution of the canonical user in an account which is a possible principal value for a policy and also
 # used when evaluating S3 bucket ACLs.
-def get_canonical_user(region):
+def get_canonical_user(region, logical_name_of_resource=None, resource=None):
     global canonical_user_id
     if canonical_user_id is not None:
         return canonical_user_id

cfn_policy_validator/cloud_trail_attributes.py

@@ -0,0 +1,79 @@
+"""
+Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+SPDX-License-Identifier: MIT-0
+"""
+
+from cfn_policy_validator import client
+from cfn_policy_validator.application_error import ApplicationError
+
+def get_dashboard_created_time(region, resource_name, resource=None):
+    return get_dashboard_attribute(region, resource_name, 'CreatedTimestamp')
+
+def get_dashboard_status(region, resource_name, resource=None):
+    return get_dashboard_attribute(region, resource_name, 'Status')
+
+def get_dashboard_type(region, resource_name, resource=None):
+    return get_dashboard_attribute(region, resource_name, 'Type')
+
+def get_dashboard_updated_time(region, resource_name, resource=None):
+    return get_dashboard_attribute(region, resource_name, 'UpdatedTimestamp')
+
+def get_eventdatastore_arn(arn_pattern, resource_name, resource, visited_nodes, region):
+    return get_eventdatastore_arn_from_client(region, resource_name)
+
+def get_eventdatastore_created_time(region, resource_name, resource=None):
+    return get_eventdatastore_attribute(region, resource_name, 'CreatedTimestamp')
+
+def get_eventdatastore_status(region, resource_name, resource=None):
+    return get_eventdatastore_attribute(region, resource_name, 'Status')
+
+def get_eventdatastore_updated_time(region, resource_name, resource=None):
+    return get_eventdatastore_attribute(region, resource_name, 'UpdatedTimestamp')
+
+
+def get_dashboard_attribute(region, resource_name, attribute):
+    supported_attributes = ['Type', 'CreatedTimestamp', 'Status', 'UpdatedTimestamp']
+    cloudtrail_client = client.build('cloudtrail', region)
+    try:
+        if attribute not in supported_attributes:
+            raise ApplicationError(f"Attribute {attribute} is not supported. Supported attributes are {supported_attributes}")
+        response = cloudtrail_client.get_dashboard(
+            DashboardId=resource_name
+        )
+        return response[attribute]
+    except Exception as e:
+        raise ApplicationError(f"Error: {e}")
+
+def get_eventdatastore_arn_from_client(region, resource_name):
+    cloudtrail_client = client.build('cloudtrail', region)
+    next_token = None
+    client_config = {
+        'MaxResults': 25
+    }
+
+    while True:
+        if next_token:
+            client_config['NextToken'] = nextToken
+        response = cloudtrail_client.list_event_data_stores(**client_config)
+        for eventdatastore in response['EventDataStores']:
+            if eventdatastore['Name'] == resource_name:
+                return eventdatastore['EventDataStoreArn']
+        nextToken = response.get('NextToken')
+        if not nextToken:
+            break
+    raise ApplicationError(f"CloudTrail Event Data Store {resource_name} not found")
+
+
+def get_eventdatastore_attribute(region, resource_name, attribute):
+    cloudtrail_client = client.build('cloudtrail', region)
+    supported_attributes = ['CreatedTimestamp', 'Status', 'UpdatedTimestamp']
+
+    if attribute not in supported_attributes:
+        raise ApplicationError(f"Attribute {attribute} is not supported. Supported attributes are {supported_attributes}")
+
+    arn=get_eventdatastore_arn_from_client(region, resource_name)
+    event_data_store_response = cloudtrail_client.get_event_data_store(EventDataStore=arn)
+    ret = event_data_store_response[attribute]
+    return ret
+
+

cfn_policy_validator/parsers/resource/api_gateway.py

@@ -0,0 +1,51 @@
+"""
+Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+SPDX-License-Identifier: MIT-0
+"""
+
+from cfn_policy_validator import ApplicationError
+from cfn_policy_validator.parsers.output import Policy, Resource
+
+class ApiGatewayRestApiPolicyParser:
+    """ AWS::ApiGateway::RestApi
+    """
+
+    def __init__(self):
+        self.rest_api_policies = []
+
+    def parse(self, _, resource):
+        evaluated_resource = resource.eval(rest_api_policy_schema)
+        properties = evaluated_resource['Properties']
+
+        policy_document = properties.get('Policy')
+        if policy_document is None:
+            # we don't need to parse resources that don't have policies and policy is optional
+            return
+        name = properties['Name']
+
+        policy = Policy('Policy', policy_document)
+        resource = Resource(name, 'AWS::ApiGateway::RestApi', policy)
+
+        self.rest_api_policies.append(resource)
+
+    def get_policies(self):
+        return self.rest_api_policies
+
+rest_api_policy_schema = {
+    'type': 'object',
+    'properties': {
+        'Properties': {
+            'type': 'object',
+            'properties': {
+                'Policy': {
+                    'type': 'object'
+                },
+                'Name': {
+                    'type': 'string'
+                }
+            },
+            'required': ['Name']
+        }
+    },
+    'required': ['Properties']
+}

cfn_policy_validator/parsers/resource/backup.py

@@ -0,0 +1,50 @@
+"""
+Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+SPDX-License-Identifier: MIT-0
+"""
+
+from cfn_policy_validator import ApplicationError
+from cfn_policy_validator.parsers.output import Policy, Resource
+
+class BackupBackupVaultPolicyParser:
+    """ AWS::Backup::BackupVault
+    """
+
+    def __init__(self):
+        self.backup_vault_policies = []
+
+    def parse(self, _, resource):
+        evaluated_resource = resource.eval(backup_vault_policy_schema)
+        properties = evaluated_resource['Properties']
+
+        policy_document = properties.get('AccessPolicy')
+        if policy_document is None:
+            # we don't need to parse resources that don't have policies and policy is optional
+            return
+        name = properties['BackupVaultName']
+
+        policy = Policy('AccessPolicy', policy_document)
+        resource = Resource(name, 'AWS::Backup::BackupVault', policy)
+        self.backup_vault_policies.append(resource)
+
+    def get_policies(self):
+        return self.backup_vault_policies
+
+backup_vault_policy_schema = {
+    'type': 'object',
+    'properties': {
+        'Properties': {
+            'type': 'object',
+            'properties': {
+                'AccessPolicy': {
+                    'type': 'object'
+                },
+                'BackupVaultName': {
+                    'type': 'string'
+                }
+            },
+            'required': ['BackupVaultName']
+        }
+    },
+    'required': ['Properties']
+}

cfn_policy_validator/parsers/resource/cloud_trail.py

@@ -0,0 +1,76 @@
+"""
+Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+SPDX-License-Identifier: MIT-0
+"""
+
+from cfn_policy_validator import ApplicationError
+from cfn_policy_validator.parsers.output import Policy, Resource
+import re
+from typing import Tuple, Optional
+
+class CloudTrailResourcePolicyParser:
+    """ AWS::CloudTrail::ResourcePolicy
+    """
+
+    def __init__(self):
+        self.resource_policies = []
+
+    @staticmethod
+    def extract_cloudtrail_resource_info(arn) -> Optional[Tuple[str, str]]:
+        """
+        Extract both the resource type and resource name from a CloudTrail ARN.
+
+        Args:
+            arn (str): The CloudTrail ARN to parse
+
+        Returns:
+            Tuple[str, str] or None: A tuple containing (resource_type, resource_name) if match found,
+                                    or None if no match
+        """
+        # Pattern captures both resource type and resource name
+        pattern = r'arn:aws:cloudtrail:[^:]*:[^:]*:([^/]+)/([^/]+)'
+        match = re.match(pattern, arn)
+
+        if match:
+            resource_type = match.group(1)  # Extract resource type
+            resource_name = match.group(2)  # Extract resource name
+            return resource_type, resource_name
+
+        return None
+
+    def parse(self, _, resource):
+        evaluated_resource = resource.eval(resource_policy_schema)
+        properties = evaluated_resource['Properties']
+
+        policy_document = properties['ResourcePolicy']
+        resource, name = self.extract_cloudtrail_resource_info(properties['ResourceArn'])
+        supported_resource_types = {'dashboard': 'AWS::CloudTrail::Dashboard', 'eventdatastore':'AWS::CloudTrail::EventDataStore'}
+        resource_type = supported_resource_types.get(resource)
+        if resource_type is None:
+            raise ApplicationError(f"Unsupported resource type {resource}")
+        policy = Policy('ResourcePolicy', policy_document)
+        resource = Resource(name, resource_type, policy)
+
+        self.resource_policies.append(resource)
+
+    def get_policies(self):
+        return self.resource_policies
+
+resource_policy_schema = {
+    'type': 'object',
+    'properties': {
+        'Properties': {
+            'type': 'object',
+            'properties': {
+                'ResourcePolicy': {
+                    'type': 'object'
+                },
+                'ResourceArn': {
+                    'type': 'string'
+                }
+            },
+            'required': ['ResourcePolicy', 'ResourceArn']
+        }
+    },
+    'required': ['Properties']
+}

cfn_policy_validator/parsers/resource/code_artifact.py

@@ -0,0 +1,51 @@
+"""
+Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+SPDX-License-Identifier: MIT-0
+"""
+
+from cfn_policy_validator import ApplicationError
+from cfn_policy_validator.parsers.output import Policy, Resource
+
+class CodeArtifactDomainPolicyParser:
+    """ AWS::CodeArtifact::Domain
+    """
+
+    def __init__(self):
+        self.code_artifact_policies = []
+
+    def parse(self, _, resource):
+        evaluated_resource = resource.eval(code_artifact_policy_schema)
+        properties = evaluated_resource['Properties']
+
+        policy_document = properties.get('PermissionsPolicyDocument')
+        if policy_document is None:
+            # we don't need to parse resources that don't have policies and policy is optional
+            return
+        name = properties['DomainName']
+
+        policy = Policy('PermissionsPolicyDocument', policy_document)
+        resource = Resource(name, 'AWS::CodeArtifact::Domain', policy)
+
+        self.code_artifact_policies.append(resource)
+
+    def get_policies(self):
+        return self.code_artifact_policies
+
+code_artifact_policy_schema = {
+    'type': 'object',
+    'properties': {
+        'Properties': {
+            'type': 'object',
+            'properties': {
+                'PermissionsPolicyDocument': {
+                    'type': 'object'
+                },
+                'DomainName': {
+                    'type': 'string'
+                }
+            },
+            'required': ['DomainName']
+        }
+    },
+    'required': ['Properties']
+}

cfn_policy_validator/parsers/resource/parser.py

@@ -6,11 +6,16 @@
 
 from cfn_policy_validator.parsers.resource.kms import KmsKeyPolicyParser
 from cfn_policy_validator.parsers.resource.s3 import S3BucketPolicyParser, S3AccessPointPolicyParser, \
-    S3MultiRegionAccessPointPolicyParser, S3BucketAclParser
+    S3MultiRegionAccessPointPolicyParser, S3BucketAclParser, S3TableBucketPolicyParser
 from cfn_policy_validator.parsers.resource.sns import SnsTopicPolicyParser
 from cfn_policy_validator.parsers.resource.sqs import SqsQueuePolicyParser
 from cfn_policy_validator.parsers.resource.lambda_aws import LambdaPermissionPolicyParser, LambdaLayerVersionPermissionParser
 from cfn_policy_validator.parsers.resource.secrets_manager import SecretsManagerPolicyParser
+from cfn_policy_validator.parsers.resource.api_gateway import ApiGatewayRestApiPolicyParser
+from cfn_policy_validator.parsers.resource.code_artifact import CodeArtifactDomainPolicyParser
+from cfn_policy_validator.parsers.resource.cloud_trail import CloudTrailResourcePolicyParser
+from cfn_policy_validator.parsers.resource.s3_express import S3ExpressAccessPointPolicyParser
+from cfn_policy_validator.parsers.resource.backup import BackupBackupVaultPolicyParser
 
 from cfn_policy_validator.parsers.utils.topological_sorter import TopologicalSorter
 
@@ -35,12 +40,18 @@ def parse(cls, template, account_config, excluded_resource_types={}):
             'AWS::S3::MultiRegionAccessPointPolicy': S3MultiRegionAccessPointPolicyParser(),
             'AWS::S3::Bucket': S3BucketAclParser(),
             'AWS::S3::BucketPolicy': S3BucketPolicyParser(),
+            'AWS::S3Tables::TableBucketPolicy': S3TableBucketPolicyParser(),
             'AWS::SQS::QueuePolicy': SqsQueuePolicyParser(),
             'AWS::SNS::TopicPolicy': SnsTopicPolicyParser(),
             'AWS::KMS::Key': KmsKeyPolicyParser(),
             'AWS::Lambda::Permission': LambdaPermissionPolicyParser(account_config),
             'AWS::Lambda::LayerVersionPermission': LambdaLayerVersionPermissionParser(account_config.partition),
-            'AWS::SecretsManager::ResourcePolicy': SecretsManagerPolicyParser()
+            'AWS::SecretsManager::ResourcePolicy': SecretsManagerPolicyParser(),
+            'AWS::ApiGateway::RestApi': ApiGatewayRestApiPolicyParser(), 
+            'AWS::CodeArtifact::Domain' : CodeArtifactDomainPolicyParser(),
+            'AWS::CloudTrail::ResourcePolicy' : CloudTrailResourcePolicyParser(),
+            'AWS::S3Express::AccessPoint' : S3ExpressAccessPointPolicyParser(),
+            'AWS::Backup::BackupVault' : BackupBackupVaultPolicyParser()
         }
 
         invoked_parsers = set()