利用python脚本实现自动阿里云rds添加ip白名单

公司购买了阿里云的rds云数据库，因为本地的ip是动态ip，非静态的，所以rds云数据库需要添加ip把名单后，本地才能直接访问。当然也可以用固定ip的服务器作为跳板机，通过跳板机访问。但是因为公司有一些本地项目，需要直连云数据库做数据分析，所以动态ip加入rds的ip白名单就成了一个必须要解决的问题了。

本文还涉及到了很多参数的获取，比如AccessKey、AccessSecret、RegionId。这此都是通过阿里云的管理后台获取的。

如果你想查看 利用python脚本实现自动阿里云域名DNS解析：点击这里查看

python的版本是3.12

安装阿里云rds相关sdk

pip install aliyun-python-sdk-rds 
pip install aliyun-python-sdk-ecs

如果安装报错可能需要安装 python -dev

rpm -ivh python-dev

脚本代码：

vim rds_dns.py

#!/usr/bin/env python
# coding=utf-8
from aliyunsdkcore import client
from aliyunsdkecs.request.v20140526 import DescribeSecurityGroupAttributeRequest
from aliyunsdkecs.request.v20140526 import AuthorizeSecurityGroupRequest
from aliyunsdkecs.request.v20140526 import AuthorizeSecurityGroupEgressRequest
from aliyunsdkecs.request.v20140526 import RevokeSecurityGroupRequest
from aliyunsdkecs.request.v20140526 import RevokeSecurityGroupEgressRequest
from aliyunsdkecs.request.v20140526 import DescribeSecurityGroupsRequest
from aliyunsdkrds.request.v20140815 import ModifySecurityIpsRequest
class Alis():
    def __init__(self, AccessKey, AccessSecret, RegionId):
       self.AccessKey = AccessKey
       self.AccessSecret = AccessSecret
       self.RegionId = RegionId
    def client(self):
       c = client.AcsClient(self.AccessKey, self.AccessSecret, self.RegionId)
       return c
    def describeSecurityGroupsRequest(self):
       request = DescribeSecurityGroupsRequest.DescribeSecurityGroupsRequest()
       request.set_accept_format('json')
       return request
    def describeSecurityGroupAttributeRequest(self,SecurityGroupId,NicType='internet',Direction='ingress'):
        """查询安全组规则
        """
        request = DescribeSecurityGroupAttributeRequest.DescribeSecurityGroupAttributeRequest()
        request.set_SecurityGroupId(SecurityGroupId)
        request.set_accept_format('json')
        request.set_NicType(NicType)
        request.set_Direction(Direction)
        return request
    def authorizeSecurityGroupRequest(self, SecurityGroupId, IpProtocol, Direction, PortRange, SourceCidrIp, DestCidrIp, Priority=1):
        """授权安全组规则
        """
        if Direction == 'ingress':
            request = AuthorizeSecurityGroupRequest.AuthorizeSecurityGroupRequest()
        elif Direction == 'egress':
            request = AuthorizeSecurityGroupEgressRequest.AuthorizeSecurityGroupEgressRequest()
        else:
            raise NameError("The specified parameter 'Direction' is not valid.")
        request.set_SecurityGroupId(SecurityGroupId)
        request.set_IpProtocol(IpProtocol)
        request.set_PortRange(PortRange)
        if SourceCidrIp:
            request.set_SourceCidrIp(SourceCidrIp)
        if DestCidrIp:
            request.set_DestCidrIp(DestCidrIp)
        request.set_Priority(Priority)
        request.set_Description("内部IP访问")
        request.set_Policy('accept')
        request.set_accept_format('json')
        return request
    def revokeSecurityGroupRequest(self, SecurityGroupId, IpProtocol, Direction, PortRange, SourceCidrIp, DestCidrIp, Priority=1):
        """删除安全组规则
        """
        if Direction == 'ingress':
            request = RevokeSecurityGroupRequest.RevokeSecurityGroupRequest()
        elif Direction == 'egress':
            request = RevokeSecurityGroupEgressRequest.RevokeSecurityGroupEgressRequest()
        else:
            raise NameError("The specified parameter 'Direction' is not valid.")
        request.set_SecurityGroupId(SecurityGroupId)
        request.set_IpProtocol(IpProtocol)
        request.set_PortRange(PortRange)
        if SourceCidrIp:
            request.set_SourceCidrIp(SourceCidrIp)
        if DestCidrIp:
            request.set_DestCidrIp(DestCidrIp)
        request.set_Priority(Priority)
        request.set_Policy('accept')
        request.set_accept_format('json')
        return request
    def modifySecurityIpsRequest(self, DBInstanceId, SecurityIps, DBInstanceIPArrayName, DBInstanceIPArrayAttribute):
        """修改数据库实例白名单
        """
        request = ModifySecurityIpsRequest.ModifySecurityIpsRequest()
        request.set_DBInstanceId(DBInstanceId)
        request.set_SecurityIps(SecurityIps)
        request.set_DBInstanceIPArrayName(DBInstanceIPArrayName)
        request.set_DBInstanceIPArrayAttribute(DBInstanceIPArrayAttribute)
        return request
if __name__ == '__main__':
     ali = Alis(AccessKey, AccessSecret, RegionId) # 阿里云后台创建
     clt = ali.client()
     req = ali.revokeSecurityGroupRequest(securityGroupId, ipProtocol, direction, 
         portRange, sourceCidrIp, destCidrIp, priority)
     res = clt.do_action_with_exception(req)
     print("删除安全组%s" % res)
     req = ali.authorizeSecurityGroupRequest(securityGroupId, ipProtocol, direction,
         portRange, sourceCidrIp, destCidrIp, priority)
     res = clt.do_action_with_exception(req)
     print("新增安全组%s" % res)
     req = ali.modifySecurityIpsRequest(dbInstanceId, securityIps,
         dbInstanceIPArrayName, dbInstanceIPArrayAttribute)
     res = clt.do_action_with_exception(req)
     print("修改RDS白名单%s" % res)

执行请参考：点击这里查看

文档参考：

操作参考：https://blog.csdn.net/qq_34453866/article/details/136056032

api接口参数：https://api.aliyun.com/document/Rds/2014-08-15/ModifySecurityIps