安全公告流监控（专业版）

> 企业级安全公告监控。公告关联+可利用性评分+状态追踪+Webhook告警，全功能覆盖。
将安全公告监控提升到企业级标准。专业版在免费版基础监控之上，新增公告关联分析、可利用性评分、状态追踪管理、速率限制、完整性校验、Webhook通知六大高级能力，帮助安全团队深度分析缺陷情报并自动响应.

架构总览

输入格式

┌──────────────────────────────────────────────────────────────┐
│              安全公告流监控 (专业版 PRO)                      │
├──────────────────────────────────────────────────────────────┤
│                                                                │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐         │
│  │ 订阅源管理   │  │ 解析与分级   │  │ 关键词过滤   │         │
│  │ (基础+扩展)  │  │ (基础+高级)  │  │ (基础+扩展)  │         │
│  └──────────────┘  └──────────────┘  └──────────────┘         │
│         │                │                │                    │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐         │
│  │ 公告关联分析  │  │ 可利用性评分 │  │ 状态追踪     │         │
│  │  ✅PRO       │  │  ✅PRO       │  │  ✅PRO       │         │
│  │ 产品/版本    │  │ EPSS-like    │  │ 生命周期     │         │
│  └──────────────┘  └──────────────┘  └──────────────┘         │
│         │                │                │                    │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐         │
│  │ 速率限制     │  │ 完整性校验   │  │ Webhook通知  │         │
│  │  ✅PRO       │  │  ✅PRO       │  │  ✅PRO       │         │
│  │ 礼貌抓取     │  │ SHA-256      │  │ 告警集成     │         │
│  └──────────────┘  └──────────────┘  └──────────────┘         │
│         │                │                │                    │
│         └────────────────┼────────────────┘                   │
│                          ▼                                    │
│  ┌────────────────────────────────────────────────────┐       │
│  │         分析与报表层                                │       │
│  │  多维统计 │ 优先级排序 │ 影响评估 │ 趋势分析       │       │
│  └────────────────────────────────────────────────────┘       │
└──────────────────────────────────────────────────────────────┘

快速开始

1. 阅读## 核心能力章节了解skill功能
2. 按## 依赖说明配置环境
3. 执行所需能力对应的命令
4. 参考## 错误处理章节处理异常
5. 查看## FAQ解答常见疑问

30秒上手（公告关联分析）

import json
import re
import hashlib
from pathlib import Path
from datetime import datetime
class ProFeedMonitor:
    """专业版安全公告流监控"""
    def __init__(self):
        self.store = Path.home() / "workspace" / "feedstream" / "pro"
        self.store.mkdir(parents=True, exist_ok=True)
        self.advisories_file = self.store / "advisories.json"
        if not self.advisories_file.exists():
            self.advisories_file.write_text("[]", encoding="utf-8")
    def extract_公告_ids(self, text):
        """提取公告编号"""
        pattern = r'公告-\d{4}-\d{4,7}'
        return list(set(re.findall(pattern, text, re.IGNORECASE)))
    def correlate_products(self, advisory, product_inventory):
        """关联受影响的产品"""
        text = (advisory["title"] + " " + advisory.get("description", "")).lower()
        affected = []
        for product in product_inventory:
            product_name = product["name"].lower()
            if product_name in text:
                affected.append({
                    "product": product["name"],
                    "installed_version": product.get("version", "unknown"),
                    "is_affected": True,
                    "公告_ids": self.extract_公告_ids(text)
                })
        advisory["affected_products"] = affected
        return affected
    def calculate_exploitability(self, advisory):
        """可利用性评分（EPSS-like简化模型）"""
        score = 0.0
        severity_scores = {"critical": 0.9, "high": 0.7, "medium": 0.4, "low": 0.2}
        score = severity_scores.get(advisory.get("severity", "info"), 0.1)
        exploit_indicators = ["exploit", "poc", "proof of concept", "weaponized",
                              "active exploitation", "in the wild", "patch available"]
        for indicator in exploit_indicators:
            if indicator in text:
                score = min(score + 0.05, 0.99)
        mitigators = ["mitigation available", "workaround", "not exploitable",
                      "requires authentication", "local access required"]
        for mitigator in mitigators:
            if mitigator in text:
                score = max(score - 0.1, 0.01)
        advisory["exploitability_score"] = round(score, 2)
        advisory["exploitability_level"] = (
            "critical" if score >= 0.8 else
            "high" if score >= 0.6 else
            "medium" if score >= 0.3 else
            "low"
        )
        return advisory["exploitability_score"]
monitor = ProFeedMonitor()
inventory = [
    {"name": "Apache", "version": "2.4.49"},
    {"name": "nginx", "version": "1.21.0"},
    {"name": "数据库", "version": "13.3"},
]
advisory = {
    "title": "公告-2021-41773 Apache HTTP Server Path Traversal",
    "description": "A flaw was found in Apache HTTP Server 2.4.49. An attacker can use path traversal to access files. Exploit available.",
    "severity": "critical"
}
公告s = monitor.extract_公告_ids(advisory["title"] + " " + advisory["description"])
print(f"公告编号：{公告s}")
affected = monitor.correlate_products(advisory, inventory)
print(f"受影响产品：{[p['product'] for p in affected]}")
score = monitor.calculate_exploitability(advisory)
print(f"可利用性评分：{score} ({advisory['exploitability_level']})")

120秒标准搭建

配置状态追踪与完整性校验：

import json
import hashlib
from pathlib import Path
from datetime import datetime
class EnterpriseFeedMonitor(ProFeedMonitor):
    """企业级安全公告监控"""
    def __init__(self):
        super().__init__()
        self.states_file = self.store / "states.json"
        self.integrity_file = self.store / "integrity.json"
        for f in [self.states_file, self.integrity_file]:
            if not f.exists():
                f.write_text("{}", encoding="utf-8")
    def update_state(self, advisory_id, new_state, note=""):
        """更新公告状态"""
        states = json.loads(self.states_file.read_text(encoding="utf-8"))
        old_state = states.get(advisory_id, {}).get("state", "new")
        states[advisory_id] = {
            "state": new_state,  # new / read / processing / resolved / ignored
            "previous_state": old_state,
            "updated_at": datetime.now().isoformat(),
            "note": note,
            "history": states.get("history", []) + [{
                "state": old_state,
                "timestamp": datetime.now().isoformat(),
                "note": note
            }]
        }
        self.states_file.write_text(json.dumps(states, ensure_ascii=False, indent=2), encoding="utf-8")
        print(f"✓ 状态更新：{advisory_id} {old_state} → {new_state}")
    def get_state(self, advisory_id):
        """获取公告状态"""
loads(self.states_file.read_text(encoding="utf-8"))
        return states.get(advisory_id, {"state": "new"})
    def list_by_state(self, state):
        """按状态筛选公告"""
loads(self.states_file.read_text(encoding="utf-8"))
        advisories = json.loads(self.advisories_file.read_text(encoding="utf-8"))
        matched = []
        for adv in advisories:
            adv_state = states.get(adv.get("id", ""), {}).get("state", "new")
            if adv_state == state:
                adv["current_state"] = adv_state
                matched.append(adv)
        print(f"\n状态 '{state}' 的公告：{len(matched)} 条")
        for a in matched[:10]:
            print(f"  [{a.get('severity','info')}] {a['title'][:60]}")
        return matched
    def verify_integrity(self):
        """数据完整性校验"""
loads(self.advisories_file.read_text(encoding="utf-8"))
        integrity = json.loads(self.integrity_file.read_text(encoding="utf-8"))
        verified = 0
        corrupted = 0
        new_records = []
        for adv in advisories:
            adv_id = adv.get("id", "")
            content_hash = hashlib.sha256(
                json.dumps(adv, sort_keys=True, ensure_ascii=False).encode()
            ).hexdigest()
            if adv_id in integrity:
                if integrity[adv_id]["hash"] == content_hash:
                    verified += 1
                else:
                    corrupted += 1
                    print(f"⚠ 完整性校验失败：{adv_id}")
                    integrity[adv_id]["hash"] = content_hash
                    integrity[adv_id]["last_verified"] = datetime.now().isoformat()
            else:
                new_records.append(adv_id)
                integrity[adv_id] = {
                    "hash": content_hash,
                    "first_seen": datetime.now().isoformat(),
                    "last_verified": datetime.now().isoformat()
                }
        self.integrity_file.write_text(
            json.dumps(integrity, ensure_ascii=False, indent=2), encoding="utf-8")
        print(f"\n=== 完整性校验报告 ===")
        print(f"已验证：{verified} 条")
        print(f"新记录：{len(new_records)} 条")
        print(f"已损坏：{corrupted} 条")
        return {"verified": verified, "new": len(new_records), "corrupted": corrupted}
monitor = EnterpriseFeedMonitor()
monitor.update_state("adv_001", "processing", "开始分析影响范围")
monitor.update_state("adv_002", "resolved", "已应用补丁修复")
monitor.list_by_state("processing")
monitor.list_by_state("resolved")
monitor.verify_integrity()

300秒完整配置

配置速率限制与Webhook通知：

import time
import json
import urllib.request
from datetime import datetime, timedelta
class FullEnterpriseMonitor(EnterpriseFeedMonitor):
    """完整企业级监控（含速率限制与Webhook）"""
    def __init__(self):
        super().__init__()
        self.rate_limit_file = self.store / "rate_limits.json"
        self.webhook_file = self.store / "webhooks.json"
        if not self.rate_limit_file.exists():
            self.rate_limit_file.write_text("{}", encoding="utf-8")
        if not self.webhook_file.exists():
            self.webhook_file.write_text("[]", encoding="utf-8")
    def fetch_with_rate_limit(self, feed_url, min_interval=300):
        """带速率限制的抓取"""
        limits = json.loads(self.rate_limit_file.read_text(encoding="utf-8"))
        last_fetch = limits.get(feed_url, {}).get("last_fetch")
        if last_fetch:
            last_time = datetime.fromisoformat(last_fetch)
            elapsed = (datetime.now() - last_time).total_seconds()
            if elapsed < min_interval:
                wait = min_interval - elapsed
                print(f"速率限制：需等待 {wait:.0f} 秒")
                return None
        try:
            req = urllib.request.Request(feed_url, headers={"User-Agent": "FeedStreamMonitor-Pro/1.0"})
            with urllib.request.urlopen(req, timeout=30) as resp:
                content = resp.read().decode("utf-8", errors="ignore")
            limits[feed_url] = {
                "last_fetch": datetime.now().isoformat(),
                "fetch_count": limits.get(feed_url, {}).get("fetch_count", 0) + 1
            }
            self.rate_limit_file.write_text(
dumps(limits, ensure_ascii=False, indent=2), encoding="utf-8")
            print(f"✓ 抓取成功（速率限制：{min_interval}秒间隔）")
            return content
        except Exception as e:
            print(f"✗ 抓取失败：{e}")
            return None
    def add_webhook(self, name, url, events=None, severity_filter=None):
        """添加Webhook通知"""
        webhooks = json.loads(self.webhook_file.read_text(encoding="utf-8"))
        webhook = {
            "id": f"hook_{len(webhooks)+1:03d}",
            "name": name,
            "url": url,
            "events": events or ["new_critical", "new_high"],
            "severity_filter": severity_filter or ["critical", "high"],
            "status": "active",
            "created_at": datetime.now().isoformat(),
            "sent_count": 0,
            "failed_count": 0
        }
        webhooks.append(webhook)
        self.webhook_file.write_text(
            json.dumps(webhooks, ensure_ascii=False, indent=2), encoding="utf-8")
        print(f"✓ Webhook已添加：{webhook['id']} - {name}")
        return webhook
    def send_webhook(self, advisory):
        """发送Webhook通知"""
loads(self.webhook_file.read_text(encoding="utf-8"))
        severity = advisory.get("severity", "info")
        for hook in webhooks:
            if hook["status"] != "active":
                continue
            if severity not in hook.get("severity_filter", []):
                continue
            payload = json.dumps({
                "event": "new_advisory",
                "severity": severity,
                "title": advisory["title"],
                "description": advisory.get("description", ""),
                "link": advisory.get("link", ""),
                "公告_ids": advisory.get("公告_ids", []),
                "exploitability_score": advisory.get("exploitability_score", 0),
            }, ensure_ascii=False).encode()
            try:
                    hook["url"],
                    data=payload,
                    headers={"Content-Type": "application/json"},
                    method="POST"
                )
request.urlopen(req, timeout=10) as resp:
                    if resp.status == 200:
                        hook["sent_count"] += 1
                        print(f"✓ Webhook已发送：{hook['name']} → {advisory['title'][:40]}")
                    else:
                        hook["failed_count"] += 1
                        print(f"✗ Webhook失败（{resp.status}）：{hook['name']}")
            except Exception as e:
                hook["failed_count"] += 1
                print(f"✗ Webhook异常：{hook['name']} - {e}")
        self.webhook_file.write_text(
            json.dumps(webhooks, ensure_ascii=False, indent=2), encoding="utf-8")
    def generate_report(self):
        """生成分析报告"""
loads(self.advisories_file.read_text(encoding="utf-8"))
loads(self.states_file.read_text(encoding="utf-8"))
        report = {
            "generated_at": datetime.now().isoformat(),
            "total_advisories": len(advisories),
            "by_severity": {},
            "by_state": {},
            "top_exploitability": [],
            "pending_action": []
        }
        for adv in advisories:
            sev = adv.get("severity", "info")
            report["by_severity"][sev] = report["by_severity"].get(sev, 0) + 1
        for adv in advisories:
            state = states.get(adv.get("id", ""), {}).get("state", "new")
            report["by_state"][state] = report["by_state"].get(state, 0) + 1
        scored = [a for a in advisories if "exploitability_score" in a]
        scored.sort(key=lambda x: x["exploitability_score"], reverse=True)
        report["top_exploitability"] = [
            {"title": a["title"][:60], "score": a["exploitability_score"],
             "level": a.get("exploitability_level", "")}
            for a in scored[:10]
        ]
        for adv in advisories:
get(adv.get("id", ""), {}).get("state", "new")
            if state in ["new", "read"] and adv.get("severity") in ["critical", "high"]:
                report["pending_action"].append({
                    "title": adv["title"][:60],
                    "severity": adv.get("severity"),
                    "state": state
                })
        print(f"\n=== 安全公告分析报告 ===")
        print(f"生成时间：{report['generated_at']}")
        print(f"公告总数：{report['total_advisories']}")
        print(f"\n按严重性：")
        for sev, count in sorted(report["by_severity"].items()):
            print(f"  {sev:<10}: {count}")
        print(f"\n按状态：")
        for state, count in sorted(report["by_state"].items()):
            print(f"  {state:<15}: {count}")
        print(f"\n待处理（critical/high且未处理）：{len(report['pending_action'])} 条")
        return report
monitor = FullEnterpriseMonitor()
monitor.add_webhook(
    "安全告警群",
    "https://hooks.example.com/security-alert",
    events=["new_critical"],
    severity_filter=["critical"]
)
content = monitor.fetch_with_rate_limit("https://nvd.nist.gov/feeds/xml/公告/misc/nvd-rss-analyzed.xml",
                                        min_interval=300)
advisory = {
    "title": "公告-2021-44228 Apache Log4j Remote Code Execution",
    "severity": "critical",
    "description": "Apache Log4j2 JNDI features do not protect against attacker-controlled LDAP...",
    "公告_ids": ["公告-2021-44228"],
    "exploitability_score": 0.95
}
monitor.send_webhook(advisory)
monitor.generate_report()

核心能力

公告关联分析（专业版）

new → read → processing → resolved
  ↓       ↓        ↓
  └───────┴────────┴──→ ignored