尧图网页设计 
简介这是一套面向基于阿里云构建生产服务的中小企业运维与安全人员的内部安全巡检工具源码采用Python Django开发用于巡扫ECS与SLB是否存在未授权访问的开放端口。主体功能分为三部分SLB外网开放Web端口扫描并配合浏览器模拟截图、ECS外网开放端口扫描以及通过Web界面集中查阅扫描结果适合具备一定Python与Linux基础的运维开发人员二次部署使用。资源包共42个文件约885KB包含11个py业务脚本、12个pyc编译文件、6个html页面模板、5张jpg截图以及sqlite3数据库、ini配置、Dockerfile、sh部署脚本和masscan可执行文件等覆盖扫描逻辑、Web展示与容器化部署各环节。运行环境需window或linux搭配python、redis、selenium、django、nmap与masscan基于python3开发。目前已有96人学习可帮助读者快速获得一套可落地的云资产端口巡检方案理解SLB与ECS扫描的代码组织方式与结果呈现思路。1. 巡检这件事为什么值得用 Django 自己写一套很多基于阿里云跑生产服务的中小团队安全巡检的现状是ECS 安全组开了哪些端口、SLB 后端挂了哪些机器、证书还有几天到期全靠人肉登录控制台一页页翻。机器少的时候还能忍一旦上了十几台 ECS 加几个 SLB靠记忆和截图管理就是灾难。我见过最典型的一次翻车是某台 ECS 的 22 端口对 0.0.0.0/0 开放了三个月没人发现直到被扫出异常登录才后知后觉。这个标题讲的就是用 Python 的 Django 框架搭一套企业内部用的安全巡检工具把 ECS 和 SLB 的配置定期拉下来做规则比对把不合规项落库、出报表。它解决的不是「防黑客」这种大命题而是「我到底开了哪些口子、哪些机器配置漂移了」这种看得见摸得着的问题。适合谁适合手里有几台到几十台阿里云资源、有 Python 基础、想自己掌控巡检逻辑而不是买现成 SaaS 的中小团队运维或后端。整套东西的核心是 Django 做 Web 层和任务调度阿里云 SDK 做数据采集数据库存历史快照做对比。下面从选型到落地一步步拆。2. 巡检工具的技术选型Django 为什么比脚本堆更合适2.1 用 Django 而不是纯脚本的三个理由一开始我也觉得巡检嘛写几个 Python 脚本挂 crontab 不就完了。真做起来才发现脚本方案有三个绕不过去的坎。第一是结果存储和查询脚本跑完输出一堆 JSON 文件想查「上周三那台 ECS 的安全组是什么样」得翻文件Django 自带 ORM 和数据库迁移巡检结果直接进表按时间、按资源、按规则维度查询都是几行代码的事。第二是权限和展示脚本没有界面运维想看结果得 SSH 上去 catDjango 的 admin 后台开箱即用稍微定制就能给不同角色看不同视图。第三是任务编排脚本的定时靠 crontab失败了没有重试和告警记录Django 配合 Celery 或简单的 management command 能把每次执行的状态、耗时、异常都记下来。选 Django 还有个现实原因团队里会 Python 的人多Django 的文档和生态成熟新人接手成本低。热词里「django项目实战新手」「django一站式教程」搜索量一直不低说明这套栈的群众基础在。用 Django 做巡检工具本质是把它当成一个「带 ORM 和后台的定时任务框架」来用Web 展示只是顺带。2.2 阿里云 SDK 的接入方式与鉴权设计采集 ECS 和 SLB 数据靠的是阿里云官方 Python SDK。ECS 用alibabacloud_ecs20140526SLB 用alibabacloud_slb20140515这两个包在 PyPI 上都能直接装。鉴权用 RAM 子账号的 AccessKey千万别用主账号 AK权限按最小化给只读 ECS 和 SLB 的 Describe 类接口就够了。pip install alibabacloud_ecs20140526 alibabacloud_slb20140515 alibabacloud_tea_openapi装完之后鉴权客户端这样初始化把 region 和 AK 抽到配置里不要硬编码from alibabacloud_tea_openapi import models as open_api_models from alibabacloud_ecs20140526.client import Client as EcsClient def build_ecs_client(region: str, ak: str, sk: str) - EcsClient: # 用 RAM 子账号 AK只授予只读权限 config open_api_models.Config( access_key_idak, access_key_secretsk, endpointfecs.{region}.aliyuncs.com, ) return EcsClient(config)这里的endpoint必须带 region跨 region 采集要建多个 client。AK 建议放环境变量或配置中心代码里只读不写。常见做法是把多个 region 的 AK 统一管理巡检任务按 region 循环。参数上endpoint写错会直接抛连接异常access_key_secret泄露是最大的安全风险所以这套工具自己的配置安全也得当回事。2.3 数据库表结构怎么设计才扛得住历史对比巡检工具的价值一半在「当前状态」一半在「历史对比」。表结构设计不好后面想做「这台机器上周和这周的安全组差异」就得哭。我一般会分三张核心表资源表存 ECS/SLB 的基本信息快照表存每次采集的完整配置 JSON规则结果表存每条规则的判定结果。# models.py 核心表结构 from django.db import models class CloudResource(models.Model): resource_type models.CharField(max_length16) # ecs / slb resource_id models.CharField(max_length64, uniqueTrue) region models.CharField(max_length32) name models.CharField(max_length128, blankTrue) created_at models.DateTimeField(auto_now_addTrue) class ResourceSnapshot(models.Model): resource models.ForeignKey(CloudResource, on_deletemodels.CASCADE) # 存原始配置 JSON方便回溯任意字段 raw_config models.JSONField() collected_at models.DateTimeField(db_indexTrue) class RuleResult(models.Model): snapshot models.ForeignKey(ResourceSnapshot, on_deletemodels.CASCADE) rule_id models.CharField(max_length64) passed models.BooleanField() detail models.TextField(blankTrue) checked_at models.DateTimeField(auto_now_addTrue)raw_config用 JSONField 是关键阿里云返回的字段几十个不可能每个都建列存原始 JSON 后续想加规则不用改表。collected_at和checked_at加索引因为查询几乎都带时间范围。热词里「django执行查询-删除对象」这类操作在清理过期快照时会用到比如保留 90 天数据用ResourceSnapshot.objects.filter(collected_at__ltcutoff).delete()批量删注意外键级联会连带删 RuleResult删之前想清楚。3. ECS 与 SLB 巡检规则怎么写从采集到判定3.1 拉取 ECS 安全组和实例列表的最小实现ECS 巡检的核心是安全组规则。先拉实例列表再逐个拉实例绑定的安全组最后拉安全组的入方向和出方向规则。分页是个坑DescribeInstances默认一页 10 条必须显式传PageSize和PageNumber循环拉全。from alibabacloud_ecs20140526 import models as ecs_models def fetch_all_instances(client, region: str): instances [] page 1 while True: req ecs_models.DescribeInstancesRequest( region_idregion, page_size100, # 最大 100别用默认 10 page_numberpage, ) resp client.describe_instances(req) batch resp.body.instances.instance or [] instances.extend(batch) if len(batch) 100: break page 1 return instancesPageSize最大 100设小了请求次数翻倍设大了会被服务端截断。循环终止条件用「返回数量小于 PageSize」比用 TotalCount 更稳因为采集过程中资源可能增减。拿到实例后SecurityGroupIds字段里是安全组 ID 列表再调DescribeSecurityGroupAttribute拉具体规则。这一步要注意安全组规则分入方向ingress和出方向egress巡检主要看入方向。3.2 SLB 监听配置和证书到期检查SLB 巡检关注两块监听器配置和 HTTPS 证书到期时间。监听器要检查是否开了不必要的端口、后端服务器健康状态证书要算剩余天数提前 30 天告警。from alibabacloud_slb20140515 import models as slb_models from datetime import datetime, timezone def check_slb_certificates(client, region: str): alerts [] req slb_models.DescribeLoadBalancersRequest(region_idregion) resp client.describe_load_balancers(req) for lb in resp.body.load_balancers.load_balancer or []: # 拉该 SLB 下所有 HTTPS 监听 listener_req slb_models.DescribeLoadBalancerHTTPSListenerAttributeRequest( region_idregion, load_balancer_idlb.load_balancer_id, listener_port443, ) try: listener client.describe_load_balancer_https_listener_attribute(listener_req) cert_id listener.body.server_certificate_id # 证书到期时间需另调 DescribeServerCertificates days_left get_cert_days_left(client, region, cert_id) if days_left 30: alerts.append({ lb_id: lb.load_balancer_id, cert_id: cert_id, days_left: days_left, }) except Exception as e: # 没有 HTTPS 监听会抛异常属正常情况 continue return alerts这里有个血泪经验没配 HTTPS 监听的 SLB 调DescribeLoadBalancerHTTPSListenerAttribute会直接抛异常必须 try 包住否则整个巡检任务中断。days_left的计算要用 UTC 时间阿里云返回的到期时间是 UTC 格式本地时区直接减会差 8 小时边界情况下会误报。3.3 规则引擎把判定逻辑和采集解耦规则不要写死在采集代码里否则加一条规则就得改采集逻辑。我一般把规则抽成独立的判定函数输入是快照 JSON输出是 pass/fail 加详情。这样规则可以配置化甚至存数据库动态加载。# rules/ecs_rules.py def rule_ssh_not_public(snapshot: dict) - tuple: 检查 22 端口是否对 0.0.0.0/0 开放 for rule in snapshot.get(ingress_rules, []): if rule.get(port_range) 22/22 and rule.get(source_cidr_ip) 0.0.0.0/0: return False, f安全组 {rule.get(security_group_id)} 的 22 端口对全网开放 return True, def rule_rdp_not_public(snapshot: dict) - tuple: 检查 3389 端口 for rule in snapshot.get(ingress_rules, []): if rule.get(port_range) 3389/3389 and rule.get(source_cidr_ip) 0.0.0.0/0: return False, 3389 端口对全网开放 return True, # 规则注册表新增规则只加一行 ECS_RULES { ecs_ssh_public: rule_ssh_not_public, ecs_rdp_public: rule_rdp_not_public, }每个规则函数签名统一为「输入快照 dict返回 (bool, str)」这样遍历注册表就能跑完所有规则。port_range的格式是起始/结束单端口就是22/22范围端口是1/65535判定时要考虑范围覆盖的情况不能只匹配精确值。规则 ID 存进 RuleResult 表方便按规则维度统计通过率。4. 巡检任务调度与结果落库的工程细节4.1 用 management command 还是 Celery定时巡检有两种做法。简单场景用 Django 的 management command 配合系统 crontab一个命令跑一次全量巡检好处是依赖少、调试直观。复杂场景用 Celery beat 做调度好处是任务状态可追踪、支持重试和并发。# management/commands/run_inspection.py from django.core.management.base import BaseCommand from inspection.tasks import inspect_all_regions class Command(BaseCommand): help 执行一次全量 ECS/SLB 巡检 def add_arguments(self, parser): parser.add_argument(--region, typestr, defaultNone, help只巡检指定 region不传则全量) def handle(self, *args, **options): region options[region] result inspect_all_regions(region) self.stdout.write( f巡检完成: 资源 {result[resources]} 个, f不合规 {result[failed]} 项 )add_arguments让命令支持按 region 过滤调试时只跑一个 region 快很多。crontab 里配0 2 * * * cd /path python manage.py run_inspection凌晨 2 点跑避开业务高峰。如果资源多、单次跑超过几分钟就该上 Celery 拆任务按 region 或按资源类型并发。4.2 采集失败的重试与幂等处理阿里云 API 偶尔会限流或超时采集失败不能直接让整个任务挂掉。我一般给每个 API 调用包一层重试用指数退避最多重试 3 次。import time from functools import wraps def retry_on_throttle(max_retries3, base_delay1): def decorator(func): wraps(func) def wrapper(*args, **kwargs): for attempt in range(max_retries): try: return func(*args, **kwargs) except Exception as e: # 只对限流和超时重试权限错误重试没意义 if Throttling in str(e) or Timeout in str(e): if attempt max_retries - 1: raise time.sleep(base_delay * (2 ** attempt)) else: raise return wrapper return decorator幂等性靠resource_id唯一约束保证同一资源重复采集时用update_or_create更新快照不会产生重复记录。base_delay * (2 ** attempt)是 1、2、4 秒的退避别设太短否则限流更严重。权限错误如 AccessDenied不重试因为重试一百次还是失败直接抛出让上层记录。4.3 结果落库的批量写入优化一次全量巡检可能产生几千条 RuleResult逐条save()会慢得离谱。用bulk_create批量写几百条一批。def save_rule_results(snapshot, results: list): objs [ RuleResult( snapshotsnapshot, rule_idrule_id, passedpassed, detaildetail, ) for rule_id, (passed, detail) in results.items() ] # batch_size 按数据库调整MySQL 一般 500-1000 RuleResult.objects.bulk_create(objs, batch_size500)batch_size不是越大越好MySQL 的max_allowed_packet限制了单条 SQL 大小500 到 1000 是稳妥区间。bulk_create不会触发save()信号如果有依赖信号的逻辑要手动处理。另外注意bulk_create返回的对象在某些数据库后端不带主键后续要用主键得重新查。5. 这套巡检工具最容易翻车的几个地方5.1 现象巡检结果里安全组规则总是空的原因DescribeSecurityGroupAttribute返回的规则在Permissions.Permission里但不同 SDK 版本字段名大小写有差异有的返回permissions有的返回Permissions。直接按一个名字取会拿到 None。解决取值时做兼容或者打印一次原始响应确认字段名。我一般写个safe_get辅助函数同时尝试大小写两种 key取到就返回。5.2 现象SLB 证书到期时间算出来差一天原因阿里云返回的NotAfter是 UTC 时间字符串用本地时区的datetime.now()去减时区差导致天数计算偏差临界值时会误报或漏报。解决统一用datetime.now(timezone.utc)做基准或者把返回时间转成带时区的 datetime 再算。别用datetime.utcnow()它返回的是 naive datetimePython 3.12 之后已标记废弃。5.3 现象巡检任务跑一半卡死日志停在某个 API 调用原因阿里云 SDK 默认没有超时设置网络抖动时请求会一直挂着。批量采集几十台机器卡一个就拖垮整个任务。解决在 client 配置里显式设read_timeout和connect_timeout一般 10 到 30 秒。配合前面的重试装饰器超时后重试而不是无限等。5.4 现象数据库快照表几个月就涨到几个 G原因每次巡检都存完整 JSON 快照一天一次几十台机器几个月下来数据量很可观。JSONField 不做压缩的话占用更大。解决定期清理过期快照保留 90 天足够回溯。清理用 management command 挂 crontab删之前先归档到冷存储或导出文件。另外可以在快照表加个is_changed字段配置没变时不存新快照只更新collected_at能省一大半空间。5.5 现象RAM 子账号 AK 权限给多了巡检工具能改配置原因图省事直接给了AliyunECSFullAccess结果工具理论上能删实例。巡检工具只需要读权限给多了是安全隐患。解决自定义 RAM 策略只授予ecs:Describe*和slb:Describe*这类只读 Action。AK 定期轮换泄露了影响可控。这套工具本身是查安全的自己权限给大了就本末倒置了。6. 让巡检结果真正被用起来告警收敛与差异对比工具跑起来只是第一步结果没人看等于白做。我踩过的坑是一开始每次巡检都发全量报告几十条不合规项运维看两天就麻木了后来直接忽略。真正有用的是「变化」而不是「状态」。差异对比的做法是每次巡检后把当前快照和上一次快照做 diff只对新增的不合规项告警。比如某台 ECS 昨天 22 端口还是关的今天开了这才值得推。实现上在 RuleResult 表里加个is_new标记落库时对比上一次同规则的passed值从 True 变 False 就是新增问题。def mark_new_failures(snapshot, results: dict): 对比上次结果标记新增的不合规项 for rule_id, (passed, detail) in results.items(): if passed: continue # 查该资源该规则上一次的结果 last RuleResult.objects.filter( snapshot__resourcesnapshot.resource, rule_idrule_id, ).order_by(-checked_at).first() is_new last is None or last.passed # 上次通过或没记录算新增 RuleResult.objects.create( snapshotsnapshot, rule_idrule_id, passedFalse, detaildetail, is_newis_new, )告警只推is_newTrue的项收敛效果立竿见影。告警渠道用阿里云短信 API 或钉钉机器人短信 API 发不出去是热词里高频问题多半是签名和模板没审核通过或者手机号格式不对调试时先用控制台的测试功能确认签名模板可用。验证这套工具是否靠谱我有个习惯手动在控制台改一条安全组规则加个 22 端口全网开放然后跑一次巡检看能不能在 5 分钟内告警出来。能说明采集、判定、落库、告警整条链路通了不能就顺着链路一段段查。这个「故意制造问题再验证」的习惯比看多少文档都管用。工具的价值不在于写得多漂亮而在于它报出来的每一条你都敢信。希望帮到你。本文还有配套的精品资源点击获取 Django实战：自建阿里云ECS与SLB安全巡检工具
简介这是一套面向基于阿里云构建生产服务的中小企业运维与安全人员的内部安全巡检工具源码采用Python Django开发用于巡扫ECS与SLB是否存在未授权访问的开放端口。主体功能分为三部分SLB外网开放Web端口扫描并配合浏览器模拟截图、ECS外网开放端口扫描以及通过Web界面集中查阅扫描结果适合具备一定Python与Linux基础的运维开发人员二次部署使用。资源包共42个文件约885KB包含11个py业务脚本、12个pyc编译文件、6个html页面模板、5张jpg截图以及sqlite3数据库、ini配置、Dockerfile、sh部署脚本和masscan可执行文件等覆盖扫描逻辑、Web展示与容器化部署各环节。运行环境需window或linux搭配python、redis、selenium、django、nmap与masscan基于python3开发。目前已有96人学习可帮助读者快速获得一套可落地的云资产端口巡检方案理解SLB与ECS扫描的代码组织方式与结果呈现思路。1. 巡检这件事为什么值得用 Django 自己写一套很多基于阿里云跑生产服务的中小团队安全巡检的现状是ECS 安全组开了哪些端口、SLB 后端挂了哪些机器、证书还有几天到期全靠人肉登录控制台一页页翻。机器少的时候还能忍一旦上了十几台 ECS 加几个 SLB靠记忆和截图管理就是灾难。我见过最典型的一次翻车是某台 ECS 的 22 端口对 0.0.0.0/0 开放了三个月没人发现直到被扫出异常登录才后知后觉。这个标题讲的就是用 Python 的 Django 框架搭一套企业内部用的安全巡检工具把 ECS 和 SLB 的配置定期拉下来做规则比对把不合规项落库、出报表。它解决的不是「防黑客」这种大命题而是「我到底开了哪些口子、哪些机器配置漂移了」这种看得见摸得着的问题。适合谁适合手里有几台到几十台阿里云资源、有 Python 基础、想自己掌控巡检逻辑而不是买现成 SaaS 的中小团队运维或后端。整套东西的核心是 Django 做 Web 层和任务调度阿里云 SDK 做数据采集数据库存历史快照做对比。下面从选型到落地一步步拆。2. 巡检工具的技术选型Django 为什么比脚本堆更合适2.1 用 Django 而不是纯脚本的三个理由一开始我也觉得巡检嘛写几个 Python 脚本挂 crontab 不就完了。真做起来才发现脚本方案有三个绕不过去的坎。第一是结果存储和查询脚本跑完输出一堆 JSON 文件想查「上周三那台 ECS 的安全组是什么样」得翻文件Django 自带 ORM 和数据库迁移巡检结果直接进表按时间、按资源、按规则维度查询都是几行代码的事。第二是权限和展示脚本没有界面运维想看结果得 SSH 上去 catDjango 的 admin 后台开箱即用稍微定制就能给不同角色看不同视图。第三是任务编排脚本的定时靠 crontab失败了没有重试和告警记录Django 配合 Celery 或简单的 management command 能把每次执行的状态、耗时、异常都记下来。选 Django 还有个现实原因团队里会 Python 的人多Django 的文档和生态成熟新人接手成本低。热词里「django项目实战新手」「django一站式教程」搜索量一直不低说明这套栈的群众基础在。用 Django 做巡检工具本质是把它当成一个「带 ORM 和后台的定时任务框架」来用Web 展示只是顺带。2.2 阿里云 SDK 的接入方式与鉴权设计采集 ECS 和 SLB 数据靠的是阿里云官方 Python SDK。ECS 用alibabacloud_ecs20140526SLB 用alibabacloud_slb20140515这两个包在 PyPI 上都能直接装。鉴权用 RAM 子账号的 AccessKey千万别用主账号 AK权限按最小化给只读 ECS 和 SLB 的 Describe 类接口就够了。pip install alibabacloud_ecs20140526 alibabacloud_slb20140515 alibabacloud_tea_openapi装完之后鉴权客户端这样初始化把 region 和 AK 抽到配置里不要硬编码from alibabacloud_tea_openapi import models as open_api_models from alibabacloud_ecs20140526.client import Client as EcsClient def build_ecs_client(region: str, ak: str, sk: str) - EcsClient: # 用 RAM 子账号 AK只授予只读权限 config open_api_models.Config( access_key_idak, access_key_secretsk, endpointfecs.{region}.aliyuncs.com, ) return EcsClient(config)这里的endpoint必须带 region跨 region 采集要建多个 client。AK 建议放环境变量或配置中心代码里只读不写。常见做法是把多个 region 的 AK 统一管理巡检任务按 region 循环。参数上endpoint写错会直接抛连接异常access_key_secret泄露是最大的安全风险所以这套工具自己的配置安全也得当回事。2.3 数据库表结构怎么设计才扛得住历史对比巡检工具的价值一半在「当前状态」一半在「历史对比」。表结构设计不好后面想做「这台机器上周和这周的安全组差异」就得哭。我一般会分三张核心表资源表存 ECS/SLB 的基本信息快照表存每次采集的完整配置 JSON规则结果表存每条规则的判定结果。# models.py 核心表结构 from django.db import models class CloudResource(models.Model): resource_type models.CharField(max_length16) # ecs / slb resource_id models.CharField(max_length64, uniqueTrue) region models.CharField(max_length32) name models.CharField(max_length128, blankTrue) created_at models.DateTimeField(auto_now_addTrue) class ResourceSnapshot(models.Model): resource models.ForeignKey(CloudResource, on_deletemodels.CASCADE) # 存原始配置 JSON方便回溯任意字段 raw_config models.JSONField() collected_at models.DateTimeField(db_indexTrue) class RuleResult(models.Model): snapshot models.ForeignKey(ResourceSnapshot, on_deletemodels.CASCADE) rule_id models.CharField(max_length64) passed models.BooleanField() detail models.TextField(blankTrue) checked_at models.DateTimeField(auto_now_addTrue)raw_config用 JSONField 是关键阿里云返回的字段几十个不可能每个都建列存原始 JSON 后续想加规则不用改表。collected_at和checked_at加索引因为查询几乎都带时间范围。热词里「django执行查询-删除对象」这类操作在清理过期快照时会用到比如保留 90 天数据用ResourceSnapshot.objects.filter(collected_at__ltcutoff).delete()批量删注意外键级联会连带删 RuleResult删之前想清楚。3. ECS 与 SLB 巡检规则怎么写从采集到判定3.1 拉取 ECS 安全组和实例列表的最小实现ECS 巡检的核心是安全组规则。先拉实例列表再逐个拉实例绑定的安全组最后拉安全组的入方向和出方向规则。分页是个坑DescribeInstances默认一页 10 条必须显式传PageSize和PageNumber循环拉全。from alibabacloud_ecs20140526 import models as ecs_models def fetch_all_instances(client, region: str): instances [] page 1 while True: req ecs_models.DescribeInstancesRequest( region_idregion, page_size100, # 最大 100别用默认 10 page_numberpage, ) resp client.describe_instances(req) batch resp.body.instances.instance or [] instances.extend(batch) if len(batch) 100: break page 1 return instancesPageSize最大 100设小了请求次数翻倍设大了会被服务端截断。循环终止条件用「返回数量小于 PageSize」比用 TotalCount 更稳因为采集过程中资源可能增减。拿到实例后SecurityGroupIds字段里是安全组 ID 列表再调DescribeSecurityGroupAttribute拉具体规则。这一步要注意安全组规则分入方向ingress和出方向egress巡检主要看入方向。3.2 SLB 监听配置和证书到期检查SLB 巡检关注两块监听器配置和 HTTPS 证书到期时间。监听器要检查是否开了不必要的端口、后端服务器健康状态证书要算剩余天数提前 30 天告警。from alibabacloud_slb20140515 import models as slb_models from datetime import datetime, timezone def check_slb_certificates(client, region: str): alerts [] req slb_models.DescribeLoadBalancersRequest(region_idregion) resp client.describe_load_balancers(req) for lb in resp.body.load_balancers.load_balancer or []: # 拉该 SLB 下所有 HTTPS 监听 listener_req slb_models.DescribeLoadBalancerHTTPSListenerAttributeRequest( region_idregion, load_balancer_idlb.load_balancer_id, listener_port443, ) try: listener client.describe_load_balancer_https_listener_attribute(listener_req) cert_id listener.body.server_certificate_id # 证书到期时间需另调 DescribeServerCertificates days_left get_cert_days_left(client, region, cert_id) if days_left 30: alerts.append({ lb_id: lb.load_balancer_id, cert_id: cert_id, days_left: days_left, }) except Exception as e: # 没有 HTTPS 监听会抛异常属正常情况 continue return alerts这里有个血泪经验没配 HTTPS 监听的 SLB 调DescribeLoadBalancerHTTPSListenerAttribute会直接抛异常必须 try 包住否则整个巡检任务中断。days_left的计算要用 UTC 时间阿里云返回的到期时间是 UTC 格式本地时区直接减会差 8 小时边界情况下会误报。3.3 规则引擎把判定逻辑和采集解耦规则不要写死在采集代码里否则加一条规则就得改采集逻辑。我一般把规则抽成独立的判定函数输入是快照 JSON输出是 pass/fail 加详情。这样规则可以配置化甚至存数据库动态加载。# rules/ecs_rules.py def rule_ssh_not_public(snapshot: dict) - tuple: 检查 22 端口是否对 0.0.0.0/0 开放 for rule in snapshot.get(ingress_rules, []): if rule.get(port_range) 22/22 and rule.get(source_cidr_ip) 0.0.0.0/0: return False, f安全组 {rule.get(security_group_id)} 的 22 端口对全网开放 return True, def rule_rdp_not_public(snapshot: dict) - tuple: 检查 3389 端口 for rule in snapshot.get(ingress_rules, []): if rule.get(port_range) 3389/3389 and rule.get(source_cidr_ip) 0.0.0.0/0: return False, 3389 端口对全网开放 return True, # 规则注册表新增规则只加一行 ECS_RULES { ecs_ssh_public: rule_ssh_not_public, ecs_rdp_public: rule_rdp_not_public, }每个规则函数签名统一为「输入快照 dict返回 (bool, str)」这样遍历注册表就能跑完所有规则。port_range的格式是起始/结束单端口就是22/22范围端口是1/65535判定时要考虑范围覆盖的情况不能只匹配精确值。规则 ID 存进 RuleResult 表方便按规则维度统计通过率。4. 巡检任务调度与结果落库的工程细节4.1 用 management command 还是 Celery定时巡检有两种做法。简单场景用 Django 的 management command 配合系统 crontab一个命令跑一次全量巡检好处是依赖少、调试直观。复杂场景用 Celery beat 做调度好处是任务状态可追踪、支持重试和并发。# management/commands/run_inspection.py from django.core.management.base import BaseCommand from inspection.tasks import inspect_all_regions class Command(BaseCommand): help 执行一次全量 ECS/SLB 巡检 def add_arguments(self, parser): parser.add_argument(--region, typestr, defaultNone, help只巡检指定 region不传则全量) def handle(self, *args, **options): region options[region] result inspect_all_regions(region) self.stdout.write( f巡检完成: 资源 {result[resources]} 个, f不合规 {result[failed]} 项 )add_arguments让命令支持按 region 过滤调试时只跑一个 region 快很多。crontab 里配0 2 * * * cd /path python manage.py run_inspection凌晨 2 点跑避开业务高峰。如果资源多、单次跑超过几分钟就该上 Celery 拆任务按 region 或按资源类型并发。4.2 采集失败的重试与幂等处理阿里云 API 偶尔会限流或超时采集失败不能直接让整个任务挂掉。我一般给每个 API 调用包一层重试用指数退避最多重试 3 次。import time from functools import wraps def retry_on_throttle(max_retries3, base_delay1): def decorator(func): wraps(func) def wrapper(*args, **kwargs): for attempt in range(max_retries): try: return func(*args, **kwargs) except Exception as e: # 只对限流和超时重试权限错误重试没意义 if Throttling in str(e) or Timeout in str(e): if attempt max_retries - 1: raise time.sleep(base_delay * (2 ** attempt)) else: raise return wrapper return decorator幂等性靠resource_id唯一约束保证同一资源重复采集时用update_or_create更新快照不会产生重复记录。base_delay * (2 ** attempt)是 1、2、4 秒的退避别设太短否则限流更严重。权限错误如 AccessDenied不重试因为重试一百次还是失败直接抛出让上层记录。4.3 结果落库的批量写入优化一次全量巡检可能产生几千条 RuleResult逐条save()会慢得离谱。用bulk_create批量写几百条一批。def save_rule_results(snapshot, results: list): objs [ RuleResult( snapshotsnapshot, rule_idrule_id, passedpassed, detaildetail, ) for rule_id, (passed, detail) in results.items() ] # batch_size 按数据库调整MySQL 一般 500-1000 RuleResult.objects.bulk_create(objs, batch_size500)batch_size不是越大越好MySQL 的max_allowed_packet限制了单条 SQL 大小500 到 1000 是稳妥区间。bulk_create不会触发save()信号如果有依赖信号的逻辑要手动处理。另外注意bulk_create返回的对象在某些数据库后端不带主键后续要用主键得重新查。5. 这套巡检工具最容易翻车的几个地方5.1 现象巡检结果里安全组规则总是空的原因DescribeSecurityGroupAttribute返回的规则在Permissions.Permission里但不同 SDK 版本字段名大小写有差异有的返回permissions有的返回Permissions。直接按一个名字取会拿到 None。解决取值时做兼容或者打印一次原始响应确认字段名。我一般写个safe_get辅助函数同时尝试大小写两种 key取到就返回。5.2 现象SLB 证书到期时间算出来差一天原因阿里云返回的NotAfter是 UTC 时间字符串用本地时区的datetime.now()去减时区差导致天数计算偏差临界值时会误报或漏报。解决统一用datetime.now(timezone.utc)做基准或者把返回时间转成带时区的 datetime 再算。别用datetime.utcnow()它返回的是 naive datetimePython 3.12 之后已标记废弃。5.3 现象巡检任务跑一半卡死日志停在某个 API 调用原因阿里云 SDK 默认没有超时设置网络抖动时请求会一直挂着。批量采集几十台机器卡一个就拖垮整个任务。解决在 client 配置里显式设read_timeout和connect_timeout一般 10 到 30 秒。配合前面的重试装饰器超时后重试而不是无限等。5.4 现象数据库快照表几个月就涨到几个 G原因每次巡检都存完整 JSON 快照一天一次几十台机器几个月下来数据量很可观。JSONField 不做压缩的话占用更大。解决定期清理过期快照保留 90 天足够回溯。清理用 management command 挂 crontab删之前先归档到冷存储或导出文件。另外可以在快照表加个is_changed字段配置没变时不存新快照只更新collected_at能省一大半空间。5.5 现象RAM 子账号 AK 权限给多了巡检工具能改配置原因图省事直接给了AliyunECSFullAccess结果工具理论上能删实例。巡检工具只需要读权限给多了是安全隐患。解决自定义 RAM 策略只授予ecs:Describe*和slb:Describe*这类只读 Action。AK 定期轮换泄露了影响可控。这套工具本身是查安全的自己权限给大了就本末倒置了。6. 让巡检结果真正被用起来告警收敛与差异对比工具跑起来只是第一步结果没人看等于白做。我踩过的坑是一开始每次巡检都发全量报告几十条不合规项运维看两天就麻木了后来直接忽略。真正有用的是「变化」而不是「状态」。差异对比的做法是每次巡检后把当前快照和上一次快照做 diff只对新增的不合规项告警。比如某台 ECS 昨天 22 端口还是关的今天开了这才值得推。实现上在 RuleResult 表里加个is_new标记落库时对比上一次同规则的passed值从 True 变 False 就是新增问题。def mark_new_failures(snapshot, results: dict): 对比上次结果标记新增的不合规项 for rule_id, (passed, detail) in results.items(): if passed: continue # 查该资源该规则上一次的结果 last RuleResult.objects.filter( snapshot__resourcesnapshot.resource, rule_idrule_id, ).order_by(-checked_at).first() is_new last is None or last.passed # 上次通过或没记录算新增 RuleResult.objects.create( snapshotsnapshot, rule_idrule_id, passedFalse, detaildetail, is_newis_new, )告警只推is_newTrue的项收敛效果立竿见影。告警渠道用阿里云短信 API 或钉钉机器人短信 API 发不出去是热词里高频问题多半是签名和模板没审核通过或者手机号格式不对调试时先用控制台的测试功能确认签名模板可用。验证这套工具是否靠谱我有个习惯手动在控制台改一条安全组规则加个 22 端口全网开放然后跑一次巡检看能不能在 5 分钟内告警出来。能说明采集、判定、落库、告警整条链路通了不能就顺着链路一段段查。这个「故意制造问题再验证」的习惯比看多少文档都管用。工具的价值不在于写得多漂亮而在于它报出来的每一条你都敢信。希望帮到你。本文还有配套的精品资源点击获取