def parse()

in cfn_policy_validator/parsers/resource/lambda_aws.py [0:0]

• 36 lines of code
• 6 McCabe index (conditional complexity)

    def parse(self, _, resource):
        evaluated_resource = resource.eval(lambda_permission_schema)

        properties = evaluated_resource['Properties']
        action = properties['Action']
        function_name = properties['FunctionName']
        principal_name = properties['Principal']

        source_account = properties.get('SourceAccount')
        source_arn = properties.get('SourceArn')

        policy_statement = {
            'Effect': 'Allow',
            'Action': action
        }

        if 'amazonaws' in principal_name:
            # this is a permission to a service
            policy_statement['Principal'] = {'Service': principal_name}
        else:
            # otherwise this is a permission to an account
            policy_statement['Principal'] = {'AWS': principal_name}

        # function name can be a full ARN, a partial ARN (a full suffix of an ARN) or the actual function name
        if function_name.startswith('arn:'):  # full ARN
            policy_statement['Resource'] = function_name
        elif 'function:' in function_name:  # partial ARN
            # partial arn must include at least function: and ":" is not a valid function name character
            arn_parts = function_name.split(":")
            # grab all parts of the partial ARN up until "function:"
            parts_before_function = list(itertools.takewhile(lambda part: part.lower() != 'function', arn_parts))

            # a list of a valid ARN before "function:"
            valid_arn_parts_before_function = ["arn", self.partition, "lambda", self.region, self.account_id]

            # construct the base_arn based on what's missing in the partial ARN
            number_of_missing_arn_parts = len(valid_arn_parts_before_function) - len(parts_before_function)
            base_arn = ":".join(valid_arn_parts_before_function[:number_of_missing_arn_parts])

            policy_statement['Resource'] = f'{base_arn}:{function_name}'
        else:  # just the function name
            policy_statement['Resource'] = f'arn:{self.partition}:lambda:{self.region}:{self.account_id}:function:{function_name}'

        if source_account is not None:
            policy_statement['Condition'] = {'StringEquals': {'AWS:SourceAccount': source_account}}

        if source_arn is not None:
            condition = policy_statement.get('Condition', {})
            condition['ArnLike'] = {'AWS:SourceArn': source_arn}
            policy_statement['Condition'] = condition

        # The permissions policies dictionary holds the statements per lambda function.  The statements are built
        # into an actual policy when they are retrieved.
        function_arn = policy_statement['Resource']
        statements = self.permissions_policies[function_arn]
        statements.append(policy_statement)