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

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

代码语言:sh
复制
# 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项目

结果:

代码语言:python
代码运行次数:0
复制
Cloud Studio 代码运行
#################################################

-- 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

sgid

nothing

sg-11111

n/a

sg-22222

n/a

目标表格:updated_sg_info.xlsx

sgid

SecurityGroupName

InnerAccessPolicy

IpProtocol

PortRange

SourceCidrIp

SecurityGroupRuleId

sg-11111

sgName1

Accept

TCP

3389/3389

0.0.0.0/0

sgr-11111

sg-11111

sgName2

Accept

TCP

8080/8083

0.0.0.0/0

sgr-11112

sg-22222

sgName3

Accept

UCP

51/51

0.0.0.0/0

sgr-11113