简介：这是一套基于Python Django开发的企业内部安全巡检工具源码，面向依托阿里云构建生产服务的中小企业运维与安全人员，用于排查ECS、SLB上是否存在未授权开放的外网端口。工具主体分为三部分：SLB外网开放Web端口扫描并配合浏览器模拟截图、ECS外网开放端口扫描，以及Web界面集中查阅结果，覆盖从扫描到可视化呈现的完整链路。资源包共42个文件，约885KB，包含11个py业务脚本、6个html页面模板、12个pyc编译文件，以及sqlite3数据库、ini配置、Dockerfile、sh部署脚本和masscan可执行文件等，另附jpg截图与md说明，便于快速理解项目结构。运行环境涉及Python、Redis、Selenium、Django、nmap与masscan，基于Python3开发。目前已有96人学习下载，适合具备一定Django与Linux基础、希望自建轻量级云资产巡检能力的开发者参考复用。

1. 巡检这件事，为什么用 Django 自己写比买现成的更划算

很多中小企业的运维现状是这样的：阿里云控制台上跑着十几台 ECS，前面挂着两三个 SLB，安全组规则是半年前配的，谁改过、改了什么，没人说得清。等到被通知某个端口暴露在外、或者某台机器被扫出弱口令，才回头一台台翻配置。买商业巡检产品当然可以，但一年几万块的授权费，对只有一两个运维的团队来说，性价比并不高。

用 Python + Django 自己写一套内部巡检工具，核心价值不在于省那笔钱，而在于你能把巡检逻辑和公司实际的资源命名规范、标签体系、告警渠道完全对齐。Django 自带 ORM、Admin 后台和迁移机制，天然适合做这种「定时拉取云资源状态、落库、比对、出报告」的工具。这篇文章讲的就是这样一套方案：怎么用 Django 把 ECS 和 SLB 的巡检跑起来，数据库怎么设计，哪些参数必须调，以及我在实际部署中踩过的坑。适合有基本 Python 基础、手上有阿里云账号、想给团队做一套轻量巡检系统的运维或后端同学。

2. 巡检工具的技术选型：Django 凭什么比脚本堆更合适

2.1 为什么不是一堆 crontab 脚本

最朴素的做法是写几个 Python 脚本，用 crontab 定时跑，结果直接打到日志里。这个方案在资源少于 5 台时能用，但很快会遇到三个问题。第一，脚本之间没有共享状态，A 脚本查过的 ECS 列表，B 脚本要重新调一次 API，阿里云 OpenAPI 是有 QPS 限制的，重复调用会触发限流。第二，结果没有结构化存储，想查「上周这台机器安全组是不是开放的 22 端口」只能翻日志。第三，没有界面，运维以外的人（比如技术负责人）想看巡检结果，你得截图发过去。

Django 解决的正是这三点：ORM 把资源清单和巡检结果落库，Admin 后台零成本给你一个可筛选、可导出的界面，管理命令（management command）配合 crontab 或 Celery 做定时调度。它不是最轻的方案，但在「要长期维护、要给别人看」这个前提下，是最省事的。

2.2 阿里云 SDK 的接入方式

阿里云官方提供了alibabacloud_ecs20140526和alibabacloud_slb20140515这两个 Python SDK 包，注意包名带版本号后缀，这是阿里云 SDK 的命名习惯，和早期的aliyun-python-sdk-ecs是两套东西。新项目建议直接用新版 SDK，接口更规范，返回的是对象而不是裸字典。

安装命令如下：

pip install alibabacloud_ecs20140526 alibabacloud_slb20140515 alibabacloud_tea_openapi

alibabacloud_tea_openapi是公共依赖，负责配置 AK/SK 和 endpoint，不能漏装。装完之后在 Django 的 settings 里不要把 AccessKey 硬编码，用环境变量读取：

# settings.py import os ALIYUN_ACCESS_KEY_ID = os.environ.get("ALIYUN_AK_ID") ALIYUN_ACCESS_KEY_SECRET = os.environ.get("ALIYUN_AK_SECRET") ALIYUN_REGION_ID = os.environ.get("ALIYUN_REGION", "cn-hangzhou")

这里有个参数要特别注意：ALIYUN_REGION_ID决定了你巡检哪个地域的资源。如果你的 ECS 分布在多个地域，不能只配一个，需要在数据库里维护一张地域表，巡检时遍历。我见过有人只配了默认地域，结果华东 2 的机器一直没被扫到，这种漏检比误报更危险。

2.3 数据库表结构怎么设计

巡检工具的数据模型不用复杂，但两张核心表必须分清：资源快照表和巡检结果表。资源快照表存「当前拉到的资源状态」，巡检结果表存「某次巡检发现的某条问题」。分开的好处是，你可以对比两次快照，发现「这台机器的安全组昨天还是关的，今天开了」这种变更。

# inspector/models.py from django.db import models class EcsInstance(models.Model): instance_id = models.CharField(max_length=64, unique=True) instance_name = models.CharField(max_length=128, blank=True) region_id = models.CharField(max_length=32) public_ip = models.GenericIPAddressField(null=True, blank=True) private_ip = models.GenericIPAddressField(null=True, blank=True) status = models.CharField(max_length=32) security_group_ids = models.JSONField(default=list) tags = models.JSONField(default=list) last_synced = models.DateTimeField(auto_now=True) class Meta: indexes = [models.Index(fields=["region_id", "status"])] class InspectionFinding(models.Model): SEVERITY_CHOICES = [("high", "高"), ("medium", "中"), ("low", "低")] resource_type = models.CharField(max_length=16) # ecs / slb resource_id = models.CharField(max_length=64) rule_code = models.CharField(max_length=32) severity = models.CharField(max_length=8, choices=SEVERITY_CHOICES) detail = models.TextField() found_at = models.DateTimeField(auto_now_add=True) resolved = models.BooleanField(default=False)

security_group_ids和tags用JSONField存，是因为这两个字段结构不固定，安全组可能挂多个，标签是键值对列表，用关系表反而增加查询复杂度。InspectionFinding里的rule_code是规则编号，比如ECS_SG_OPEN_22，这样后续做规则开关、统计某类问题的数量都方便。resolved字段让你能标记「已处理」，避免每次巡检都重复报同一个问题。

3. 把 ECS 和 SLB 拉进数据库：同步命令的完整实现

3.1 封装阿里云客户端

不要在业务代码里到处new客户端，统一封装一个工厂函数。这样 AK 轮换、超时调整只改一处。

# inspector/aliyun_client.py from alibabacloud_tea_openapi import models as open_api_models from alibabacloud_ecs20140526.client import Client as EcsClient from alibabacloud_slb20140515.client import Client as SlbClient from django.conf import settings def _base_config(endpoint: str) -> open_api_models.Config: config = open_api_models.Config( access_key_id=settings.ALIYUN_ACCESS_KEY_ID, access_key_secret=settings.ALIYUN_ACCESS_KEY_SECRET, ) config.endpoint = endpoint config.read_timeout = 10000 # 毫秒 config.connect_timeout = 5000 return config def get_ecs_client(region: str) -> EcsClient: return EcsClient(_base_config(f"ecs.{region}.aliyuncs.com")) def get_slb_client(region: str) -> SlbClient: return SlbClient(_base_config(f"slb.{region}.aliyuncs.com"))

read_timeout设 10 秒是有原因的：默认值偏短，在资源多、分页请求时容易超时抛异常，巡检任务直接中断。设太长又会在网络故障时卡住整个任务。10 秒是我在几十台规模下比较稳的值，资源上百台可以适当加到 15 秒。

3.2 分页拉取 ECS 列表

阿里云 ECS 的DescribeInstances接口单次最多返回 100 条，必须循环翻页，否则资源一多就漏。

# inspector/sync.py from alibabacloud_ecs20140526 import models as ecs_models from .aliyun_client import get_ecs_client from .models import EcsInstance def sync_ecs(region: str) -> int: client = get_ecs_client(region) page = 1 total = 0 while True: req = ecs_models.DescribeInstancesRequest( region_id=region, page_size=100, page_number=page, ) resp = client.describe_instances(req) instances = resp.body.instances.instance or [] if not instances: break for inst in instances: EcsInstance.objects.update_or_create( instance_id=inst.instance_id, defaults={ "instance_name": inst.instance_name or "", "region_id": region, "public_ip": _first_ip(inst.public_ip_address), "private_ip": _first_ip(inst.private_ip_address), "status": inst.status, "security_group_ids": inst.security_group_ids.security_group_id or [], "tags": [{"key": t.tag_key, "value": t.tag_value} for t in (inst.tags.tag or [])], }, ) total += 1 page += 1 return total def _first_ip(ip_obj): if ip_obj and ip_obj.ip_address: return ip_obj.ip_address[0] return None

update_or_create是关键，它保证同一台机器重复同步不会产生重复记录，同时更新状态。_first_ip处理的是 SDK 返回的 IP 是列表结构，一台机器可能有多个公网 IP，这里只取第一个，如果你的场景需要全部保留，把字段改成 JSONField 存列表。

3.3 SLB 同步与监听端口检查

SLB 的巡检重点和 ECS 不同，ECS 看安全组，SLB 看监听配置——特别是有没有监听 0.0.0.0 且后端是 HTTP 的 80 端口。

# inspector/sync.py（续） from alibabacloud_slb20140515 import models as slb_models from .aliyun_client import get_slb_client from .models import SlbLoadBalancer, InspectionFinding def sync_slb(region: str) -> int: client = get_slb_client(region) req = slb_models.DescribeLoadBalancersRequest(region_id=region) resp = client.describe_load_balancers(req) lbs = resp.body.load_balancers.load_balancer or [] for lb in lbs: SlbLoadBalancer.objects.update_or_create( load_balancer_id=lb.load_balancer_id, defaults={ "name": lb.load_balancer_name or "", "region_id": region, "address": lb.address, "status": lb.load_balancer_status, }, ) _check_slb_listeners(client, lb.load_balancer_id) return len(lbs) def _check_slb_listeners(client, lb_id: str): req = slb_models.DescribeLoadBalancerAttributeRequest( load_balancer_id=lb_id ) resp = client.describe_load_balancer_attribute(req) listeners = resp.body.listener_ports.listener_port or [] for port in listeners: if port == 80: InspectionFinding.objects.create( resource_type="slb", resource_id=lb_id, rule_code="SLB_HTTP_80", severity="medium", detail=f"SLB {lb_id} 开放了 80 端口，建议确认是否已配置 HTTPS 跳转", )

这里DescribeLoadBalancerAttribute是逐个 SLB 查详情，SLB 数量一般不多，可以接受。如果 SLB 超过 50 个，建议改用DescribeLoadBalancerListeners批量接口，减少调用次数。

3.4 用 management command 串起来

把同步逻辑包成 Django 管理命令，这样既能手动跑，也能挂 crontab。

# inspector/management/commands/run_inspection.py from django.core.management.base import BaseCommand from inspector.sync import sync_ecs, sync_slb from inspector.rules import run_all_rules class Command(BaseCommand): help = "执行一次全量巡检" def add_arguments(self, parser): parser.add_argument("--region", default="cn-hangzhou") def handle(self, *args, **options): region = options["region"] ecs_count = sync_ecs(region) slb_count = sync_slb(region) findings = run_all_rules(region) self.stdout.write( f"同步 ECS {ecs_count} 台，SLB {slb_count} 个，" f"发现 {findings} 条问题" )

跑起来就是一行命令：

python manage.py run_inspection --region cn-hangzhou

--region做成参数而不是写死，是为了多地域时能用 crontab 配多条，或者在外面套一层 shell 循环。run_all_rules是规则引擎的入口，下一章展开。

4. 巡检规则怎么写：从安全组到端口暴露的判定逻辑

4.1 规则引擎的最小设计

规则不需要复杂框架，一个函数注册表就够了。每条规则接收资源对象，返回 finding 列表。

# inspector/rules.py from .models import EcsInstance, InspectionFinding RULES = [] def rule(code, severity): def wrapper(fn): RULES.append((code, severity, fn)) return fn return wrapper @rule("ECS_SG_OPEN_22", "high") def check_ssh_open(ecs: EcsInstance): findings = [] # 这里需要安全组详情，实际实现中会先同步安全组规则 if _sg_allows_all(ecs.security_group_ids): findings.append({ "resource_id": ecs.instance_id, "detail": f"ECS {ecs.instance_id} 安全组允许 0.0.0.0/0 访问 22 端口", }) return findings def run_all_rules(region: str) -> int: count = 0 for ecs in EcsInstance.objects.filter(region_id=region): for code, severity, fn in RULES: for f in fn(ecs): InspectionFinding.objects.create( resource_type="ecs", resource_id=f["resource_id"], rule_code=code, severity=severity, detail=f["detail"], ) count += 1 return count

_sg_allows_all需要你先同步安全组规则。安全组规则不在 ECS 实例信息里，要单独调DescribeSecurityGroupAttribute，把规则存到一张SecurityGroupRule表，判定时查库。这一步很多人会漏，导致规则永远返回空。

4.2 安全组规则的同步与判定

# inspector/sync.py（续） from alibabacloud_ecs20140526 import models as ecs_models def sync_security_group_rules(region: str, sg_id: str): client = get_ecs_client(region) req = ecs_models.DescribeSecurityGroupAttributeRequest( region_id=region, security_group_id=sg_id, ) resp = client.describe_security_group_attribute(req) for perm in (resp.body.permissions.permission or []): SecurityGroupRule.objects.update_or_create( security_group_id=sg_id, direction=perm.direction, ip_protocol=perm.ip_protocol, port_range=perm.port_range, source_cidr_ip=perm.source_cidr_ip or "", defaults={"policy": perm.policy}, )

判定「22 端口对全网开放」的逻辑就是查SecurityGroupRule里port_range包含 22、source_cidr_ip是0.0.0.0/0、policy是accept的记录。port_range的格式是22/22或1/65535，解析时要按/拆分再比较，不能直接字符串相等。

4.3 巡检频率与告警去重

巡检不是越频繁越好。ECS 状态变化不频繁，15 分钟一次足够；安全组规则变更少，可以 1 小时一次。关键是告警去重：同一条 finding 如果已经存在且未 resolved，不要重复创建。

def create_finding_if_new(resource_id, rule_code, severity, detail): exists = InspectionFinding.objects.filter( resource_id=resource_id, rule_code=rule_code, resolved=False, ).exists() if not exists: InspectionFinding.objects.create( resource_type="ecs", resource_id=resource_id, rule_code=rule_code, severity=severity, detail=detail, )

这个exists查询在数据量大时会慢，给(resource_id, rule_code, resolved)建联合索引。另外，如果问题被修复了，要有反向逻辑把对应的 finding 标记为 resolved，否则列表里会一直挂着已解决的问题。

5. 部署与避坑：那些让我加班到凌晨的细节

5.1 AK 权限最小化

巡检工具只需要读权限，千万不要用主账号 AK。在 RAM 里建一个子账号，只授予AliyunECSReadOnlyAccess和AliyunSLBReadOnlyAccess两个系统策略。我见过有人图省事直接用主账号 AK，结果工具被入侵后整个云账号沦陷。AK 存环境变量，不要提交到 Git，.env文件加进.gitignore。

5.2 常见问题排查

现象：同步 ECS 时报Throttling限流错误。原因：分页循环调用太快，或者多个地域任务并发跑，超过 OpenAPI 的 QPS 限制。 解决：在每次分页请求之间time.sleep(0.2)，或者用 Django 的transaction把写库和请求分开，请求阶段只收集数据，最后批量写。并发任务用 Celery 时给每个地域的 task 加rate_limit。

现象：SLB 监听端口检查返回空列表。原因：DescribeLoadBalancerAttribute返回的listener_ports结构在不同 SDK 版本里字段名不一致，有的是listener_port，有的是listener_ports。 解决：打印resp.body的原始结构确认字段名，或者改用DescribeLoadBalancerListeners接口，它的返回结构更稳定。

现象：Admin 后台加载 ECS 列表特别慢。原因：EcsInstance表没有给region_id和status建索引，列表页默认按last_synced排序又没索引。 解决：在Meta.indexes里补上，或者用list_select_related减少关联查询。数据超过 5000 条时，Admin 的show_full_result_count关掉，它会对全表做 count。

现象：定时任务偶尔不执行。原因：crontab 的环境变量和登录 shell 不同，python路径找不到，或者 Django 的DJANGO_SETTINGS_MODULE没设。 解决：crontab 里用绝对路径的 python，命令写成cd /path/to/project && /usr/bin/python manage.py run_inspection，环境变量在脚本里source .env加载。

现象：巡检结果里同一台机器出现多条重复记录。原因：update_or_create的instance_id字段没有唯一约束，或者同步时 region 参数传错导致同一实例被当成不同地域的记录。 解决：确认instance_id在 model 里是unique=True，同步前先按 region 清空或标记旧数据。

6. 让巡检结果真正被用起来：报告导出与变更对比

工具跑起来只是第一步，真正产生价值的是让结果进入团队的日常流程。我一般会做两件事：一是每周自动生成一份巡检报告，二是做资源变更对比。

报告导出用 Django 的模板系统渲染 HTML，再转 PDF 或直接发邮件。核心是把InspectionFinding按 severity 分组统计：

# inspector/report.py from django.db.models import Count from .models import InspectionFinding def weekly_summary(): qs = InspectionFinding.objects.filter(resolved=False) by_severity = qs.values("severity").annotate(count=Count("id")) by_rule = (qs.values("rule_code") .annotate(count=Count("id")) .order_by("-count")[:10]) return { "total": qs.count(), "by_severity": {item["severity"]: item["count"] for item in by_severity}, "top_rules": list(by_rule), }

by_rule取前 10 条，是为了让报告聚焦在最高频的问题上，而不是把几百条 finding 全列出来没人看。这个统计结果可以直接塞进邮件正文，或者渲染成一个简单的 HTML 表格。

变更对比更实用。每次同步前，先把当前EcsInstance的关键字段（安全组、公网 IP、状态）快照到一张历史表，同步后对比差异，把「安全组从关闭变开放」这类变更单独告警。实现上可以用 Django 的pre_save信号，或者在同步函数里手动记录。我倾向于手动记录，因为信号在批量update_or_create时触发时机不好控制。

def record_change(instance_id, field, old_value, new_value): ResourceChange.objects.create( resource_id=instance_id, field_name=field, old_value=str(old_value), new_value=str(new_value), )

这张ResourceChange表不需要索引太多，按resource_id和created_at查就行。每周看一眼变更记录，比看静态的巡检列表更能发现异常——比如某台机器半夜安全组被改开了 3389，这种在静态列表里可能被淹没，但在变更记录里一眼就能看到。

最后说个我自己的习惯：巡检规则不要一次写太多，先上 5 条最关键的（SSH 开放、RDP 开放、SLB 无 HTTPS、ECS 无标签、安全组全放通），跑两周确认没有误报，再逐步加。规则一多，误报就会淹没真问题，团队很快就会忽略告警。这套工具的价值不在于规则多全，而在于每一条报出来的问题都值得处理。希望帮到你。

本文还有配套的精品资源，点击获取