Install
openclaw skills install @thcjp/feedstream-monitor-freeopenclaw skills install @thcjp/feedstream-monitor-free聚合多源安全公告,自动分级去重。让安全威胁第一时间被感知。
将分散在各处的安全公告聚合到统一平台。本技能提供多源RSS/Atom订阅管理、严重性分级、关键词过滤、本地去重能力,帮助安全团队高效监控漏洞情报。
┌─────────────────────────────────────────────────┐
│ 安全公告流监控 (免费版) │
├─────────────────────────────────────────────────┤
│ │
│ ┌──────────────────────────────────────┐ │
│ │ 订阅源管理层 │ │
│ │ CVE/NVD │ 厂商公告 │ 安全博客 │ │
│ └──────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌──────────────────────────────────────┐ │
│ │ 解析与处理层 │ │
│ │ RSS/Atom解析 → 严重性分级 → 去重 │ │
│ └──────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌──────────────────────────────────────┐ │
│ │ 过滤与查询层 │ │
│ │ 关键词过滤 │ 严重性筛选 │ 产品匹配 │ │
│ └──────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌──────────────────────────────────────┐ │
│ │ 本地存储层 │ │
│ │ feeds.json │ advisories.json │ │
│ └──────────────────────────────────────┘ │
└─────────────────────────────────────────────────┘
import json
import hashlib
from pathlib import Path
from datetime import datetime
import urllib.request
import xml.etree.ElementTree as ET
STORE_DIR = Path.home() / "workspace" / "feedstream" / "monitor"
STORE_DIR.mkdir(parents=True, exist_ok=True)
FEEDS_FILE = STORE_DIR / "feeds.json"
ADVISORIES_FILE = STORE_DIR / "advisories.json"
for f in [FEEDS_FILE, ADVISORIES_FILE]:
if not f.exists():
f.write_text("[]", encoding="utf-8")
def add_feed(name, url, feed_type="rss", category="general"):
"""添加订阅源"""
feeds = json.loads(FEEDS_FILE.read_text(encoding="utf-8"))
feed = {
"id": f"feed_{len(feeds)+1:03d}",
"name": name,
"url": url,
"type": feed_type, # rss / atom
"category": category, # cve / vendor / blog / general
"status": "active",
"added_at": datetime.now().isoformat(),
"last_fetched": None,
"fetch_count": 0
}
feeds.append(feed)
FEEDS_FILE.write_text(json.dumps(feeds, ensure_ascii=False, indent=2), encoding="utf-8")
print(f"✓ 订阅源已添加:{feed['id']} - {name}")
return feed
add_feed("NVD漏洞公告", "https://nvd.nist.gov/feeds/xml/cve/misc/nvd-rss-analyzed.xml", "rss", "cve")
add_feed("CERT公告", "https://www.kb.cert.org/vulfeed", "rss", "vendor")
add_feed("安全客", "https://api.anquanke.com/data/v1/rss", "rss", "blog")
配置完整的抓取、解析与分级系统:
详细代码示例已移至
references/detail.md
配置关键词过滤与定期检查:
class FilteredMonitor(FeedStreamMonitor):
"""带关键词过滤的监控器"""
def __init__(self):
super().__init__()
self.filters_file = self.store / "filters.json"
if not self.filters_file.exists():
self.filters_file.write_text("[]", encoding="utf-8")
def add_filter(self, name, keywords, action="highlight"):
"""添加关键词过滤器"""
filters = json.loads(self.filters_file.read_text(encoding="utf-8"))
f = {
"id": f"filter_{len(filters)+1:03d}",
"name": name,
"keywords": keywords,
"action": action, # highlight / hide / alert
"created_at": datetime.now().isoformat()
}
filters.append(f)
self.filters_file.write_text(json.dumps(filters, ensure_ascii=False, indent=2), encoding="utf-8")
print(f"✓ 过滤器已添加:{f['id']} - {name}")
return f
def apply_filters(self):
"""应用过滤器"""
advisories = json.loads(self.advisories_file.read_text(encoding="utf-8"))
filters = json.loads(self.filters_file.read_text(encoding="utf-8"))
for advisory in advisories:
text = (advisory["title"] + " " + advisory.get("description", "")).lower()
matched_filters = []
for f in filters:
if any(kw.lower() in text for kw in f["keywords"]):
matched_filters.append(f["name"])
advisory["matched_filters"] = matched_filters
self.advisories_file.write_text(
json.dumps(advisories, ensure_ascii=False, indent=2), encoding="utf-8")
highlighted = [a for a in advisories if a.get("matched_filters")]
print(f"过滤完成:{len(highlighted)} 条公告匹配过滤器")
return highlighted
def fetch_all(self):
"""抓取所有活跃订阅源"""
feeds = json.loads(self.feeds_file.read_text(encoding="utf-8"))
active = [f for f in feeds if f["status"] == "active"]
print(f"开始抓取 {len(active)} 个订阅源...")
for feed in active:
self.fetch_feed(feed["id"])
self.apply_filters()
self.stats()
fm = FilteredMonitor()
fm.add_filter("Apache相关", ["apache", "tomcat", "struts"], "highlight")
fm.add_filter("数据库漏洞", ["mysql", "postgresql", "redis", "mongodb"], "highlight")
fm.add_filter("远程代码执行", ["rce", "remote code execution"], "alert")
fm.fetch_all()
| 类别 | 说明 | 典型来源 |
|---|---|---|
| cve | CVE漏洞公告 | NVD、MITRE CVE |
| vendor | 厂商安全公告 | Microsoft、Apache、Oracle |
| blog | 安全博客 | 安全客、FreeBuf、Seebug |
| cert | 应急响应中心 | US-CERT、CNVD、CNNVD |
| general | 通用安全信息 | 安全新闻、技术文章 |
输入: 用户提供订阅源管理所需的指令和必要参数。 处理: 按照skill规范执行订阅源管理操作,遵循单一意图原则。 输出: 返回订阅源管理的执行结果,包含操作状态和输出数据。
| 级别 | 关键词 | 说明 |
|---|---|---|
| critical | critical, rce, zero-day, 0day | 紧急:远程代码执行、零日漏洞 |
| high | high, sql injection, xss, privilege escalation | 严重:高危漏洞 |
| medium | medium, dos, bypass | 中等:中危漏洞 |
| low | low, information disclosure, minor | 低危:信息泄露等 |
| info | 无匹配关键词 | 信息:一般性公告 |
输入: 用户提供严重性分级所需的指令和必要参数。 处理: 按照skill规范执行严重性分级操作,遵循单一意图原则。 输出: 返回严重性分级的执行结果,包含操作状态和输出数据。
| 过滤动作 | 说明 |
|---|---|
| highlight | 高亮显示匹配的公告 |
| hide | 隐藏匹配的公告 |
| alert | 匹配时发送告警 |
输入: 用户提供关键词过滤所需的指令和必要参数。 处理: 按照skill规范执行关键词过滤操作,遵循单一意图原则。 输出: 返回关键词过滤的执行结果,包含操作状态和输出数据。
基于标题+链接的MD5哈希去重,确保同一公告不会被重复存储。
输入: 用户提供去重机制所需的指令和必要参数。 处理: 按照skill规范执行去重机制操作,遵循单一意图原则。 输出: 返回去重机制的执行结果,包含操作状态和输出数据。
input_params参数,支持创建/查询/导出操作
能力覆盖范围:本skill的核心能力覆盖以下场景关键词:安全公告订阅流监、控免费版、支持多源、RSS、Atom、严重性分级、基础关键词过滤与、本地去重、安全公告流监控免、费版是面向安全运、维团队的漏洞与公、告订阅监控工具、聚合多个安全信息、订阅流、自动解析、帮助团队第一时间、感知安全威胁、Use、when、需要系统监控、日志分析、运维告警、部署管理时使用、不适用于物理硬件、适用于独立开发者、企业团队和自动化、工作流场景等。角色:安全运维工程师
场景描述:监控NVD的CVE公告,关注critical和high级别漏洞。
monitor = FeedStreamMonitor()
monitor.add_feed("NVD", "https://nvd.nist.gov/feeds/xml/cve/misc/nvd-rss-analyzed.xml", "rss", "cve")
monitor.fetch_feed("feed_001")
monitor.search(severity="critical")
角色:系统管理员
场景描述:追踪使用产品的厂商安全公告,及时了解补丁信息。
fm = FilteredMonitor()
fm.add_filter("关注产品", ["apache", "nginx", "postgresql", "redis"], "highlight")
fm.fetch_all()
角色:安全分析师
场景描述:聚合多个安全博客和新闻源,收集安全情报。
monitor = FeedStreamMonitor()
monitor.add_feed("安全客", "https://api.anquanke.com/data/v1/rss", "rss", "blog")
monitor.fetch_all()
monitor.stats()
免费版支持RSS 2.0和Atom 1.0两种主流格式。RSS使用 <item> 标签解析,Atom使用 <entry> 标签解析。大多数安全信息源(NVD、CERT、安全博客)都提供RSS或Atom订阅。
基于关键词匹配:扫描公告标题和描述中的关键词,匹配到"critical/rce/zero-day"等标记为critical级别,"high/sql injection"等标记为high级别。这是一种启发式方法,不如CVSS评分精确,但能快速识别高优先级公告。
每条公告基于标题和链接生成MD5哈希作为唯一ID。抓取新公告时,与已存储的哈希比对,跳过已存在的。这确保同一公告即使被多个源转发也只存储一次。
过滤器帮助从大量公告中筛选关注的内容。例如设置"Apache相关"过滤器,关键词为["apache", "tomcat", "struts"],所有匹配的公告会被高亮标记。支持highlight(高亮)、hide(隐藏)、alert(告警)三种动作。
所有数据存储在本地 ~/workspace/feedstream/monitor/ 目录下:feeds.json(订阅源配置)、advisories.json(公告数据)、filters.json(过滤器配置)。不涉及云端同步,确保数据隐私。可通过修改 STORE_DIR 自定义路径。
| 依赖项 | 类型 | 是否必需 | 获取方式 |
|---|---|---|---|
| Python标准库 | 内置 | 必需 | Python自带(urllib/xml/hashlib/json) |
| feedparser | Python库 | 可选 | pip install feedparser(更强大的RSS解析) |
| LLM API | API | 必需 | 由Agent平台内置LLM提供 |
本技能基于原始开源安全公告监控作品改进,保留原始版权声明:
本改进作品在原始作品基础上进行了深度差异化改造,包括但不限于:
原始MIT license允许使用、复制、修改和分发,需保留版权声明。本改进作品在保留原始版权声明的基础上添加自有署名,完全符合MIT license要求。
本免费体验版限制以下高级功能:
解锁全部功能请使用专业版:feedstream-monitor-pro
| 错误场景 | 原因 | 处理方式 |
|---|---|---|
| 配置错误 | 参数缺失或格式错误 | 检查依赖说明中的配置要求 |
| 运行时错误 | 运行环境不满足 | 确认运行环境符合依赖说明 |
| 网络错误 | 连接超时或不可达 | 执行ping命令测试网络连通性,检查防火墙和代理设置连接后执行ping命令测试网络连通性,检查防火墙和代理设置连接后重新执行命令,参考国内替代方案 |
输入:用户提供操作指令和必要参数
输出:返回执行结果,包含操作状态和输出数据
用户: 执行核心功能
Skill: 正在执行核心功能...
Skill: 执行完成,结果如下: 操作成功