create_securitygroup.py 文件源码

def add_cidr_rules(self, rules):
        """Add cidr rules to security group via boto.

        Args:
            rules (list): Allowed Security Group ports and protocols.

        Returns:
            True: Upon successful completion.

        Raises:
            SpinnakerSecurityGroupError: boto3 call failed to add CIDR block to
                Security Group.
        """
        session = boto3.session.Session(profile_name=self.env, region_name=self.region)
        client = session.client('ec2')

        group_id = get_security_group_id(self.app_name, self.env, self.region)

        for rule in rules:
            data = {
                'DryRun':
                False,
                'GroupId':
                group_id,
                'IpPermissions': [{
                    'IpProtocol': rule['protocol'],
                    'FromPort': rule['start_port'],
                    'ToPort': rule['end_port'],
                    'IpRanges': [{
                        'CidrIp': rule['app']
                    }]
                }]
            }
            self.log.debug('Security Group rule: %s', data)

            try:
                client.authorize_security_group_ingress(**data)
            except botocore.exceptions.ClientError as error:
                if 'InvalidPermission.Duplicate' in str(error):
                    self.log.debug('Duplicate rule exist, that is OK.')
                else:
                    msg = 'Unable to add cidr rules to {}'.format(rule.get('app'))
                    self.log.error(msg)
                    raise SpinnakerSecurityGroupError(msg)

        return True