企业官网建设流程全解析

简介：这是一套基于Python的威胁情报自动播报与聚合工具，面向安全运维人员、威胁情报分析者及对漏洞动态有持续跟踪需求的开发者。项目整合360、奇安信、红后、绿盟、安全客、斗象等公开情报源，自动爬取CVE等威胁信息，并通过邮件推送、页面TOP10展示与本地归档三种方式输出情报。资源共70个文件，以Python脚本（25个）、XML配置、DAT缓存数据、HTML页面模板为主，压缩包仅738KB；代码结构清晰，包含爬虫、数据处理、消息通知、SQL建库脚本及GitHub Actions工作流配置，便于二次开发和自定义部署。已有191人学习下载。借助本项目可快速搭建一套免服务器运维的情报订阅系统，理解多源情报采集、去重归档与多渠道播报的完整实现思路，适合个人或小团队持续跟踪外部威胁动态。

1. 威胁情报不是爬虫项目：先想清楚你要汇总什么

做威胁情报（threat-intelligence）踩过最大的坑，是把它当成爬虫项目来启动。真正跑起来才发现，80% 的工作量不在抓取，而在清洗、去重、验证和关联。公开渠道的安全通告、漏洞库、恶意 IP 列表每天产生上千条数据，但其中大量是重复的，还有不少是误报和过时信息。这套资源的核心不是给你一堆采集脚本，而是把「采集 → 解析 → 存储 → 查询 → 验证」的完整链路串起来，用 Python 实现一个能落地的 IOC（入侵指标）汇总分析流程。适合安全工程师、运维人员，以及需要做内部威胁情报平台选型但又不想一开始就上重型商业方案的人。你跟着做出来的东西，可以对接已有的告警系统，也可以当成一个独立的情报查询工具用。

2. 整体架构与数据模型：SQLite 加事件表，不上一堆重组件

2.1 为什么不用 Elasticsearch 或关系型大库

很多人一上来就考虑 Elasticsearch，理由是情报数据要全文检索。但实际场景里，一个中小团队每天处理的情报量在几千到几万条级别，SQLite 完全扛得住，而且部署成本几乎为零。Elasticsearch 的运维负担、内存占用、数据导入导出的麻烦，在这个量级上不划算。我一般建议先跑通 SQLite 版本，等真的出现性能瓶颈再迁移，到时候表结构和查询逻辑大体不用改。

数据模型的设计上，我见过不少反例：有人把 IOC 塞进一个超大宽表，结果查一次关联要 JOIN 五六个字段，效率极低。核心思路应该是一张事件主表（incidents）加一张指标明细表（iocs），两表之间通过事件 ID 关联。主表存事件元信息，明细表存 IP、域名、哈希、URL 等具体指标。这样设计的好处是，一个事件可能包含多个 IOC 类型，而你查询某个 IP 时只需要在明细表里做索引扫描，速度快，逻辑也清晰。

2.2 表结构设计：字段别贪多，够用就行

给出一份可以直接照抄的建表 SQL，这是整套资源里所有脚本的基础。

CREATE TABLE incidents ( id INTEGER PRIMARY KEY AUTOINCREMENT, source TEXT NOT NULL, title TEXT, description TEXT, published_at TEXT, fetched_at TEXT DEFAULT CURRENT_TIMESTAMP, severity INTEGER DEFAULT 2, status TEXT DEFAULT 'new' ); CREATE TABLE iocs ( id INTEGER PRIMARY KEY AUTOINCREMENT, incident_id INTEGER NOT NULL, ioc_type TEXT NOT NULL, value TEXT NOT NULL, confidence REAL DEFAULT 0.5, first_seen TEXT, last_seen TEXT, FOREIGN KEY (incident_id) REFERENCES incidents(id) ); CREATE INDEX idx_iocs_type_value ON iocs(ioc_type, value); CREATE INDEX idx_iocs_value ON iocs(value);

这里几个字段需要解释一下。severity用整数不用字符串，0 到 3 分别对应低危到危急，排序和过滤都方便；confidence是浮点数，0 到 1 之间，表示这条 IOC 的可信度，来自厂商标注或者你自己的验证逻辑；status字段标记处理状态，new、investigating、confirmed、false_positive四种。索引方面，iocs表必须建value的单列索引，因为这是查询频率最高的路径。

2.3 模块划分：采集、解析、入库三个脚本各自独立

这套资源的代码包按功能拆成了四个模块，collector.py负责抓取，parser.py负责解析，storage.py负责入库，query.py负责查询导出。分开写比一个大文件实用得多——采集源出问题只需要改 collector，解析规则变了只动 parser，互不干扰。

3. 采集层实现：多源抓取与频率控制

3.1 公开安全通告源的通用采集模板

常见的威胁情报源大致分两类：一类是结构化 API，返回 JSON 或 CSV；另一类是安全公告页面，返回 HTML 或 RSS。第一类写起来简单，第二类才是真正麻烦的地方。给一个处理 RSS 通告源的模板，适配大多数安全厂商的公告系统。

import requests import xml.etree.ElementTree as ET from datetime import datetime, timezone FEED_URLS = { "example_cert": "https://example-cert.org/feeds/advisories", "example_vendor": "https://example-vendor.com/security/bulletin.rss", } def fetch_feed(name, url, timeout=15): headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) threat-intel-collector/1.0" } try: resp = requests.get(url, headers=headers, timeout=timeout) resp.raise_for_status() return resp.content except requests.exceptions.Timeout: print(f"[{name}] timeout after {timeout}s, skip this round") return None except requests.exceptions.RequestException as e: print(f"[{name}] request failed: {e}") return None def parse_rss(content): root = ET.fromstring(content) items = [] for item in root.iter("item"): entry = { "title": item.findtext("title", "").strip(), "link": item.findtext("link", "").strip(), "description": item.findtext("description", "").strip(), "pub_date": item.findtext("pubDate", "").strip(), } items.append(entry) return items

两个参数值得注意。timeout=15是请求超时阈值，设短了网络抖动容易失败，设长了采集任务会被卡住，15 秒是我试下来比较平衡的值。User-Agent必须设置成真实浏览器样式，很多情报源会拒绝裸的 Python 请求。RSS 解析用xml.etree.ElementTree而不是正则，因为 XML 实体的转义规则用正则容易漏，尤其 description 字段里经常带 HTML 标签。

3.2 增量采集与去重：记住上次的游标位置

全量抓取每次都从头开始，既浪费带宽又会产生大量重复数据。正确做法是记录每个源上次成功获取到的最新条目时间或 ID，下次增量拉取。下面这段代码展示了如何利用fetched_at做增量判断：

import sqlite3 from datetime import datetime, timedelta def get_last_fetch_time(conn, source): row = conn.execute( "SELECT MAX(fetched_at) FROM incidents WHERE source = ?", (source,) ).fetchone() if row and row[0]: return datetime.fromisoformat(row[0]) return datetime.now(timezone.utc) - timedelta(days=7) def is_duplicate(conn, source, title): row = conn.execute( "SELECT id FROM incidents WHERE source = ? AND title = ? LIMIT 1", (source, title), ).fetchone() return row is not None

get_last_fetch_time的作用是给采集器一个时间窗口，只处理这个时间之后发布的条目。is_duplicate用source + title作为自然键判断是否已经入库。这里有个细节：不要用link字段做唯一性判断，因为有些源的链接会带跟踪参数，同一个通告可能生成多个不同 URL，导致重复入库。

3.3 抓取失败的补偿策略

采集任务跑起来之后你会发现，情报源不稳定是常态，尤其是境外源，经常超时或者返回 502。常见的做法是连续失败三次才告警，单次失败只记录日志，下一轮继续重试。这个逻辑必须写进主循环，否则任何一次抓取异常都会白屏崩溃。

import time FAIL_THRESHOLD = 3 def run_collector(conn, source, url, fail_count=0): content = fetch_feed(source, url) if content is None: fail_count += 1 if fail_count >= FAIL_THRESHOLD: print(f"[{source}] failed {FAIL_THRESHOLD} times, manual check required") fail_count = 0 return fail_count items = parse_rss(content) for item in items: if not is_duplicate(conn, source, item["title"]): insert_incident(conn, source, item) return 0

这段代码的核心是fail_count的累积与重置机制。连续失败三次说明源可能挂了或者改版了，需要人工介入；单次失败不处理，等下次调度。采集频率控制在每 30 到 60 分钟一轮比较合适，低于 30 分钟容易被封 IP，高于 60 分钟情报时效性会打折。

4. 情报解析与 IOC 提取：正则之外还要有启发式规则

4.1 从通告正文里抠出四类 IOC

通告的正文通常是一段自然语言描述，夹杂着 IP 地址、域名、哈希值、URL。第一版我直接用正则提取，效果很差：URL 里包含的域名会被误识别为独立域名，文件哈希和软件版本号混在一起也分不清。后来改成「正则提取候选值 + 启发式规则分类」的两步走思路。

import re IP_RE = re.compile(r"(?<!\d)(?:\d{1,3}\.){3}\d{1,3}(?!\d)") HASH_RE = re.compile(r"\b(?:[a-fA-F0-9]{32}|[a-fA-F0-9]{40}|[a-fA-F0-9]{64})\b") DOMAIN_RE = re.compile(r"\b(?:[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?\.)+[a-z]{2,}\b", re.IGNORECASE) URL_RE = re.compile(r"https?://[^\s<>\"']+") def extract_iocs(text): iocs = [] for m in URL_RE.finditer(text): iocs.append(("url", m.group(0))) for m in HASH_RE.finditer(text): h = m.group(0).lower() if len(h) == 32: iocs.append(("md5", h)) elif len(h) == 40: iocs.append(("sha1", h)) else: iocs.append(("sha256", h)) for m in DOMAIN_RE.finditer(text): domain = m.group(0).lower() if not any(domain in u for u, _ in iocs if _ == "url"): iocs.append(("domain", domain)) for m in IP_RE.finditer(text): ip = m.group(0) if is_valid_ip(ip): iocs.append(("ip", ip)) return iocs

这里有个顺序问题：必须先提取 URL，再从结果里排除域名，否则 URL 里的域名会被当成独立域名重复计入。HASH_RE里三个长度分别对应 MD5、SHA1、SHA256，提取后统一转成小写，因为后面做关联分析时大小写不一致会导致 JOIN 失败。is_valid_ip是一个校验函数，过滤掉类似999.1.1.1这种正则能匹配但不合法的值。

4.2 启发式过滤：去掉明显误报

正则提取只是第一步，误报依然很多。比如正文里提到的示例 IP192.0.2.1（RFC 文档保留地址段）、内部 IP10.0.0.1、以及作者举例用的假域名example.com。这些必须在解析阶段干掉，否则存进去之后每次查询都会被干扰。

RESERVED_IP_RANGES = ["10.", "192.168.", "127.", "0.", "172.16.", "172.17.", "172.18.", "172.19.", "172.2", "172.3"] RESERVED_DOMAINS = ["example.com", "example.org", "example.net", "test.com", "localhost"] def filter_false_positives(iocs): filtered = [] for ioc_type, value in iocs: if ioc_type == "ip": if any(value.startswith(prefix) for prefix in RESERVED_IP_RANGES): continue elif ioc_type == "domain": if any(value.endswith(domain) for domain in RESERVED_DOMAINS): continue filtered.append((ioc_type, value)) return filtered

这段过滤逻辑看着简单，但实际很有用。地址段前缀的判断用了startswith而不是精确匹配，是因为172.16.0.1到172.31.255.255这些私网地址段范围较宽，用前缀匹配是业内默认做法。注意172.2和172.3是为了覆盖172.20.x.x到172.30.x.x的部分地址段，不算精确，但够用。

4.3 置信度打分：不是所有情报可信度都一样

厂商通告里的 IOC 可信度高，第三方汇总源的可信度低。这套资源里把置信度拆成两个维度：来源权重和指标类型权重。厂商官方源给 0.9，第三方聚合源给 0.6；IP 类指标因为动态变化频繁，打 0.1 的降权系数，域名类降权 0.2。

SOURCE_WEIGHT = {"vendor_official": 0.9, "aggregator": 0.6, "community": 0.4} TYPE_WEIGHT = {"sha256": 1.0, "sha1": 0.9, "md5": 0.8, "domain": 0.7, "url": 0.6, "ip": 0.4} def compute_confidence(source_type, ioc_type): return round(SOURCE_WEIGHT.get(source_type, 0.5) * TYPE_WEIGHT.get(ioc_type, 0.5), 2)

哈希值的权重比 IP 高是有原因的：攻击者换 IP 成本极低，换一个文件哈希成本高得多，所以哈希的稳定性更强。置信度最终存到iocs表的confidence字段里，查询的时候可以按这个字段排序或过滤。

5. 踩坑与排查：存活性验证、编码脏数据、增量逻辑失效

5.1 存活性验证：情报库里 70% 的 IP 可能已经失效

现象：从情报库里随机抽出 100 条 IP 数据做检测，能连上的不到三成，大量情报来源已经是死数据。

原因：IP 和域名的存活性周期很短，攻击基础设施经常变，国内 IP 更是说封就封。采集入库时不验证，结果只能当历史档案用。

解决：入库后 24 小时内必须做一轮存活性探测。IP 用 TCP 连接测试常见端口，域名做 DNS 解析确认是否仍然指向有效地址。注意事项是探测频率不能太高，否则情报源会把你封掉。

5.2 编码与脏数据：公告页面编码不一致导致解析挂掉

现象：某个情报源昨天还能正常解析，今天突然提取不到任何 IOC，检查日志发现解析器报了一堆乱码错误。

原因：不同源用的编码不一样，有的 GBK，有的 UTF-8，还有的页面声称是 UTF-8 实际是 Latin-1。requests 的resp.text默认使用响应头声称的编码，遇到错误的声明就会解出乱码。

解决：抓取后强制用resp.content配合chardet重新检测编码，不要直接用resp.text。代码库里已经内置了这个逻辑，你只需要在改源的时候注意别把这个步骤跳过去。

5.3 增量逻辑失效：漏抓两周数据

现象：某天发现incidents表里有个源的数据停留在两周前，但采集任务一直显示成功执行。

原因：get_last_fetch_time用的是MAX(fetched_at)，但fetched_at是入库时间，不是信息发布的时间。如果某次批量导入历史数据，fetched_at会被整体刷新，导致增量窗口错乱。

解决：改用published_at字段作为增量游标，这个字段保存的是情报源原始发布时间，不会因为入库操作而变化。如果某个源不提供published_at，就用link指纹去重，两套逻辑互相兜底。

5.4 业务误报：情报平台命中≠安全事件

现象：情报平台频繁告警，说内网 IP 访问了恶意域名，但核查后发现是某个业务系统的合法调用。

原因：很多云服务商的 IP 段被误报进威胁情报库，比如共享 IP 上有人跑过恶意程序，同 IP 的其他用户全部受害。这种情况只靠情报源本身无法区分。

解决：在查询层加白名单机制，把业务已知的合法域名和 IP 段维护成一张bypass_rules表，命中告警时先查白名单再决定是否升级。这不是让你忽略威胁，而是降低无效告警对研判精力的消耗。

6. 关联分析与内部情报平台对接：让数据真正用起来

6.1 交叉关联：同一攻击者留下的线索串起来

情报数据的价值在关联而不在单条。一个事件里提取出的恶意 IP 可能在其他事件的域名解析记录里出现过，把这个关联关系找出来，就能画出攻击基础设施的轮廓。

def correlate_by_ip(conn, target_ip): rows = conn.execute( """ SELECT DISTINCT i2.incident_id, i2.value, i2.ioc_type FROM iocs i1 JOIN iocs i2 ON i1.incident_id = i2.incident_id WHERE i1.value = ? AND i1.ioc_type = 'ip' AND i2.value != ? """, (target_ip, target_ip), ).fetchall() return rows

这段 SQL 做的事很直接：找到包含目标 IP 的所有事件，再把这些事件涉及的其他 IOC 全部拉出来。SELECT DISTINCT是为了去重，因为同一个事件里同一个 IOC 可能被重复提取多次。实际用的时候还可以加i1.confidence > 0.6这类条件，把低置信度的关联过滤掉，避免噪声干扰判断。

6.2 MRF 排序：怎么知道哪条情报最值得先处理

事件批量入库之后，你面临的就是「先看哪条」的问题。按时间倒序是最初级的方式，但容易漏掉重要性高的事件。我在这套资源里实现了 MRF（Most Relevant First）排序，综合事件严重级别、IOC 数量、来源可信度三个维度计算一个分数：

def compute_mrf_score(conn, incident_id): inc = conn.execute( "SELECT severity, source FROM incidents WHERE id = ?", (incident_id,) ).fetchone() if not inc: return 0 severity, source = inc ioc_count = conn.execute( "SELECT COUNT(*) FROM iocs WHERE incident_id = ?", (incident_id,) ).fetchone()[0] source_conf = SOURCE_WEIGHT.get(source, 0.5) score = severity * 10 + min(ioc_count, 10) * 2 + source_conf * 5 return score

这个打分公式是可调的：严重级别的权重最大，因为漏洞利用的紧急程度是首要维度；IOC 数量起到辅助作用，一个事件带 10 条指标比只带 1 条的研判价值更高；来源权重最小，只是兜底。你可以根据自己的业务调整系数，比如对挖矿事件更敏感就把severity系数加大。

6.3 导出对接：生成给防火墙和 EDR 用的 IOC 列表

情报平台的最终输出是给设备消费的。代码包里包含一个导出模块，支持两种格式：分别是纯文本 IP 列表和 CSV 格式。

python query.py --type ip --min-confidence 0.7 --format txt python query.py --type sha256 --since 2024-01-01 --format csv

导出结果可以直接导入防火墙的地址组、SIEM 的威胁情报插件，或者分发给内部威胁研判群。建议每天定时导出一次，而不是实时查询——实时查询会让数据库频繁被读，影响采集入库的性能。我第一次就踩了这个坑，导出任务和采集任务抢数据库锁，导致入库延迟。从那以后，我强制把导出改成每天凌晨跑一次，数据写入和读取彻底分开，再没出过问题。

情报系统是一步步长大的，先把采集和入库跑通，再叠加过滤和关联，最后再对接设备。希望这套资源帮你少走点弯路。

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