diff --git a/scripts/intel/main.py b/scripts/intel/main.py index 8b90ca37..605372dd 100644 --- a/scripts/intel/main.py +++ b/scripts/intel/main.py @@ -31,6 +31,7 @@ from intel.route import route_advisories # noqa: E402 from intel.sources.runner import build_failure, collect_candidates, failure_summary, find_source, probe_source, probe_sources # noqa: E402 from intel.utils import isoformat, load_all_json, now_utc, parse_since, read_json, write_json # noqa: E402 from intel.validators import validate # noqa: E402 +from intel.versioning import discover_entities, sync_versions, write_entity_registry, write_version_registry # noqa: E402 def _load_existing_advisories() -> List[AdvisoryRecord]: @@ -146,6 +147,35 @@ def _summarize_changes(advisories: List[AdvisoryRecord]) -> Dict[str, Any]: } +def _selected_system_ids(source_map: Dict[str, Any]) -> set[str]: + return {system["system_id"] for system in source_map.get("systems", []) or []} + + +def _apply_discovery_and_version_sync( + source_map: Dict[str, Any], + advisories: List[AdvisoryRecord], + *, + deep: bool = False, + enqueue_lab: bool = False, +) -> tuple[List[AdvisoryRecord], Dict[str, Any], Dict[str, Any]]: + selected_system_ids = _selected_system_ids(source_map) + advisory_rows = [item.to_dict() for item in advisories] + discovery = discover_entities(source_map, advisory_rows, write_registry=False) + write_entity_registry(discovery["entities"], selected_system_ids=selected_system_ids) + version_state = sync_versions( + source_map, + advisory_rows, + entity_records=discovery["entities"], + deep=deep, + enqueue_lab=enqueue_lab, + write_registry=False, + ) + write_entity_registry(version_state["entities"], selected_system_ids=selected_system_ids) + write_version_registry(version_state["versions"], selected_system_ids=selected_system_ids) + synced = route_advisories(source_map, [AdvisoryRecord(**item) for item in version_state["advisories"]]) + return synced, discovery["summary"], version_state["summary"] + + def _select_hotlane( advisories: List[AdvisoryRecord], triage: List[Dict[str, Any]], @@ -239,6 +269,8 @@ def pipeline( tier: str | None, include_undated: bool, hotlane_only: bool = False, + deep_version_sync: bool = False, + enqueue_lab: bool = False, ) -> tuple[list[AdvisoryRecord], list[Dict[str, Any]], list[str], Dict[str, Any]]: if tier == "history-full": since_dt = None @@ -252,7 +284,16 @@ def pipeline( if hotlane_only: advisories, triage = _select_hotlane(advisories, triage) advisories, triage = _merge_existing_registry(advisories, triage) + advisories = route_advisories(source_map, advisories) + advisories, discovery_summary, version_summary = _apply_discovery_and_version_sync( + source_map, + advisories, + deep=deep_version_sync, + enqueue_lab=enqueue_lab, + ) change_summary = _summarize_changes(advisories) + change_summary["auto_promoted_entity_count"] = discovery_summary.get("auto_promoted_count", 0) + change_summary["version_sync"] = version_summary render_map = source_map selected_system_ids = None if len(source_map["systems"]) != len(full_source_map["systems"]): @@ -332,7 +373,14 @@ def cmd_ingest(args) -> int: if since == "last-success": state = read_json(STATE_PATH, default={}) or {} since = state.get("last_success", "30d") - advisories, triage, failures, summary = pipeline(full_source_map, source_map, since, None, include_undated=False) + advisories, triage, failures, summary = pipeline( + full_source_map, + source_map, + since, + None, + include_undated=False, + enqueue_lab=True, + ) _write_state("success" if not failures else "degraded", record_success=not failures) print( f"Ingested {len(advisories)} advisories, new {summary['new_count']}, updated {summary['updated_count']}, triage {len(triage)}, failures {len(failures)}" @@ -343,7 +391,15 @@ def cmd_ingest(args) -> int: def cmd_hotlane(args) -> int: full_source_map = load_source_map() source_map = _filter_source_map(full_source_map, args.system) - advisories, triage, failures, summary = pipeline(full_source_map, source_map, "1d", None, include_undated=False, hotlane_only=True) + advisories, triage, failures, summary = pipeline( + full_source_map, + source_map, + "1d", + None, + include_undated=False, + hotlane_only=True, + enqueue_lab=True, + ) _write_state("success" if not failures else "degraded", record_success=not failures) print( f"Hotlane synced {len(advisories)} advisories, new {summary['new_count']}, updated {summary['updated_count']}, triage {len(triage)}, failures {len(failures)}" @@ -354,7 +410,15 @@ def cmd_hotlane(args) -> int: def cmd_reconcile(args) -> int: full_source_map = load_source_map() source_map = _filter_source_map(full_source_map, args.system) - advisories, triage, failures, summary = pipeline(full_source_map, source_map, "30d", None, include_undated=False) + advisories, triage, failures, summary = pipeline( + full_source_map, + source_map, + "30d", + None, + include_undated=False, + deep_version_sync=True, + enqueue_lab=True, + ) _write_state("success" if not failures else "degraded", record_success=not failures) print( f"Reconciled {len(advisories)} advisories, new {summary['new_count']}, updated {summary['updated_count']}, triage {len(triage)}, failures {len(failures)}" @@ -382,6 +446,8 @@ def cmd_backfill(args) -> int: args.tier, include_undated=True, hotlane_only=args.hotlane_only, + deep_version_sync=args.tier == "history-full", + enqueue_lab=True, ) print( f"Backfilled {len(advisories)} advisories, new {summary['new_count']}, updated {summary['updated_count']}, triage {len(triage)}, failures {len(failures)}" @@ -389,6 +455,45 @@ def cmd_backfill(args) -> int: return 0 if not failures else 1 +def cmd_discover_entities(args) -> int: + full_source_map = load_source_map() + source_map = _filter_source_map(full_source_map, args.system) + advisories = [item for item in _load_existing_advisories() if item.system_id in _selected_system_ids(source_map)] + discovery = discover_entities(source_map, advisories, write_registry=False) + write_entity_registry(discovery["entities"], selected_system_ids=_selected_system_ids(source_map)) + _refresh_render_state(full_source_map, source_map) + print( + f"Discovered cataloged_entities={discovery['summary'].get('cataloged_entity_total', 0)} " + f"candidate_backlog={discovery['summary'].get('candidate_entity_total', 0)} " + f"auto_promoted={discovery['summary'].get('auto_promoted_count', 0)}" + ) + return 0 + + +def cmd_sync_versions(args) -> int: + full_source_map = load_source_map() + source_map = _filter_source_map(full_source_map, args.system) + selected_ids = _selected_system_ids(source_map) + advisories = [item for item in _load_existing_advisories() if item.system_id in selected_ids] + synced, discovery_summary, version_summary = _apply_discovery_and_version_sync( + source_map, + route_advisories(source_map, advisories), + deep=args.deep, + enqueue_lab=True, + ) + _refresh_render_state(full_source_map, source_map) + print( + "Version sync completed: " + f"cataloged_entities={discovery_summary.get('cataloged_entity_total', 0)} " + f"auto_promoted={discovery_summary.get('auto_promoted_count', 0)} " + f"latest_synced={version_summary.get('latest_version_synced_count', 0)} " + f"source_gap={version_summary.get('source_gap_count', 0)} " + f"security_versions={version_summary.get('security_version_total', 0)} " + f"lab_enqueued={version_summary.get('lab_enqueued_count', 0)}" + ) + return 0 + + def cmd_monitor(args) -> int: full_source_map = load_source_map() source_map = _filter_source_map(full_source_map, args.system) @@ -399,6 +504,9 @@ def cmd_monitor(args) -> int: audit = write_source_catalog_audit(source_map) + existing_advisories = [item for item in _load_existing_advisories() if item.system_id in _selected_system_ids(source_map)] + _apply_discovery_and_version_sync(source_map, route_advisories(source_map, existing_advisories), deep=False, enqueue_lab=False) + probes, failures = probe_sources(source_map) retried_probes, remaining_failures, retries_performed = _retry_degraded_sources(source_map, failures) if retried_probes: @@ -419,7 +527,14 @@ def cmd_monitor(args) -> int: state = read_json(STATE_PATH, default={}) or {} since = state.get("last_success", "30d") - advisories, triage, ingest_failures, summary = pipeline(full_source_map, source_map, since, None, include_undated=False) + advisories, triage, ingest_failures, summary = pipeline( + full_source_map, + source_map, + since, + None, + include_undated=False, + enqueue_lab=True, + ) alerts = build_alerts( source_health.get("failures", []), previous_alerts=previous_alerts, @@ -495,6 +610,15 @@ def main() -> int: render.add_argument("--system", action="append") render.set_defaults(func=cmd_render) + discover = subparsers.add_parser("discover-entities", help="Discover and auto-catalog stable security-related entities") + discover.add_argument("--system", action="append") + discover.set_defaults(func=cmd_discover_entities) + + sync_versions_parser = subparsers.add_parser("sync-versions", help="Refresh latest versions and security-related version history") + sync_versions_parser.add_argument("--system", action="append") + sync_versions_parser.add_argument("--deep", action="store_true") + sync_versions_parser.set_defaults(func=cmd_sync_versions) + source_health = subparsers.add_parser("source-health", help="Check source adapter health without mutating registry advisories") source_health.add_argument("--tier", choices=["history-full", "rolling-24m"]) source_health.add_argument("--system", action="append") diff --git a/scripts/intel/monitoring.py b/scripts/intel/monitoring.py index 482f93f4..b2327ef6 100644 --- a/scripts/intel/monitoring.py +++ b/scripts/intel/monitoring.py @@ -7,6 +7,7 @@ from typing import Any, Dict, List from intel.config import ( ALERTS_PATH, ENTITY_COMPLETENESS_PATH, + LAB_ENQUEUE_SUMMARY_PATH, MACHINE_READABLE_SOURCE_KINDS, MONITORING_DIR, MONITOR_SUMMARY_PATH, @@ -14,6 +15,8 @@ from intel.config import ( SOURCE_CATALOG_AUDIT_MD_PATH, SOURCE_CATALOG_AUDIT_PATH, SOURCE_HEALTH_PATH, + VERSION_BACKLOG_PATH, + VERSION_COMPLETENESS_PATH, iter_all_sources, ) from intel.utils import ensure_dir, isoformat, now_utc, parse_dt, read_json, write_json, write_text @@ -365,6 +368,9 @@ def write_monitoring_state( open_alerts = [item for item in alerts if item.get("status") == "open"] generated_at = source_health.get("generated_at") or isoformat(now_utc()) entity_completeness = read_json(ENTITY_COMPLETENESS_PATH, default={}) or {} + version_completeness = read_json(VERSION_COMPLETENESS_PATH, default={}) or {} + version_backlog = read_json(VERSION_BACKLOG_PATH, default={}) or {} + lab_enqueue_summary = read_json(LAB_ENQUEUE_SUMMARY_PATH, default={}) or {} summary = { "generated_at": generated_at, "active_source_count": source_health.get("active_source_count", 0), @@ -397,12 +403,27 @@ def write_monitoring_state( "version_mapped_count": entity_completeness.get("version_mapped_count", 0), "official_source_covered_count": entity_completeness.get("official_source_covered_count", 0), }, + "version_coverage": { + "cataloged_entity_total": version_completeness.get("cataloged_entity_total", 0), + "latest_version_synced_count": version_completeness.get("latest_version_synced_count", 0), + "source_gap_count": version_completeness.get("source_gap_count", 0), + "security_version_total": version_completeness.get("security_version_total", 0), + "security_version_entity_count": version_completeness.get("security_version_entity_count", 0), + "auto_promoted_entity_count": version_completeness.get("auto_promoted_entity_count", 0), + "lab_enqueued_count": version_completeness.get("lab_enqueued_count", 0), + }, + "lab_enqueue": { + "enqueued": lab_enqueue_summary.get("enqueued", 0), + "queue_total": lab_enqueue_summary.get("queue_total", 0), + "pending_count": len(lab_enqueue_summary.get("pending", []) or []), + }, } snapshot = { "generated_at": generated_at, "source_catalog_audit": audit, "source_health": source_health, "alerts": alerts, + "version_backlog": version_backlog, "monitor_summary": summary, } write_json(MONITOR_SUMMARY_PATH, summary) diff --git a/scripts/intel/render.py b/scripts/intel/render.py index 250b404e..979f6f73 100644 --- a/scripts/intel/render.py +++ b/scripts/intel/render.py @@ -23,6 +23,7 @@ from intel.config import ( from intel.entities import build_entity_views from intel.models import AdvisoryRecord from intel.utils import ensure_dir, isoformat, now_utc, write_json, write_text +from intel.versioning import build_version_views, write_version_views from lab.render import render_dashboard as render_lab_dashboard from lab.repro import annotate_with_latest_run, latest_runs_by_advisory @@ -677,6 +678,12 @@ def render_generated( write_json(ENTITY_QUEUES_PATH, entity_views["queues"]) write_text(ENTITY_CATALOG_REPORT_MD_PATH, entity_views["catalog_report_markdown"]) write_text(ENTITY_BACKLOG_REPORT_MD_PATH, entity_views["backlog_report_markdown"]) + version_views = build_version_views( + source_map, + advisories, + entity_records=entity_views["entities"], + ) + write_version_views(version_views) render_lab_dashboard( advisory_records=[item.to_dict() for item in advisories], source_map_data=source_map, diff --git a/scripts/lab/render.py b/scripts/lab/render.py index d67fc814..894bd0a7 100644 --- a/scripts/lab/render.py +++ b/scripts/lab/render.py @@ -554,6 +554,7 @@ def _build_architecture_data(summary: Dict[str, Any], source_map: Dict[str, Any] _link("retired sources", "/docs/retired-sources.html", "退役源、退役原因与 replacement map。"), _link("entity catalog report", "/docs/entity-catalog-report.html", "分层实体覆盖、history-full 完整度与 workflow 指标。"), _link("entity discovery backlog", "/docs/entity-discovery-backlog.html", "待编目 repo / 插件 / 包 backlog 与等待原因。"), + _link("version sync report", "/docs/version-sync-report.html", "安全相关版本同步、source-gap 与版本驱动 lab enqueue 摘要。"), _link("repro-map 真值", "/docs/repro-map.html", "复现族路由、浏览器要求和日志策略。"), _link("覆盖矩阵", "/docs/coverage-matrix.html", "自动生成覆盖摘要的本地镜像。"), _link("设计来源清单", "/docs/design-source.html", "Lovart 模板本地 vendor manifest。"), @@ -569,6 +570,9 @@ def _build_architecture_data(summary: Dict[str, Any], source_map: Dict[str, Any] _link("entity-completeness.json", "/data/entity-completeness.json", "实体级 catalog 完整度、版本映射与 workflow 覆盖。"), _link("entity-discovery-backlog.json", "/data/entity-discovery-backlog.json", "发现但尚未正式编目的 repo / 插件 / 包 backlog。"), _link("entity-queues.json", "/data/entity-queues.json", "discovery/history/latest/workflow 四类队列摘要。"), + _link("version-completeness.json", "/data/version-completeness.json", "最新版本同步覆盖、安全相关版本历史与 auto-promoted 统计。"), + _link("version-backlog.json", "/data/version-backlog.json", "source-gap、未解决版本缺口与 lab pending 队列。"), + _link("release-index.json", "/data/release-index.json", "安全相关版本记录索引真值。"), _link("runs.json", "/runs.json", "最近 run 的结构化详情。"), _link("systems.json", "/systems.json", "系统级覆盖与浏览器证据摘要。"), _link("entities.json", "/entities.json", "分层实体索引、实体状态和系统归属。"), @@ -838,6 +842,9 @@ def _build_architecture_data(summary: Dict[str, Any], source_map: Dict[str, Any] _field("实体完整度", "/data/entity-completeness.json"), _field("发现 backlog", "/data/entity-discovery-backlog.json"), _field("实体队列", "/data/entity-queues.json"), + _field("版本完整度", "/data/version-completeness.json"), + _field("版本 backlog", "/data/version-backlog.json"), + _field("版本索引", "/data/release-index.json"), _field("默认入口", "/index.html"), _field("总览入口", "/overview/index.html"), _field("运行入口", "/runs/index.html"), @@ -1365,6 +1372,10 @@ def render_dashboard( entity_completeness = read_json(ROOT / "08-threat-intel" / "generated" / "entity-completeness.json", default={}) or {} entity_backlog = read_json(ROOT / "08-threat-intel" / "generated" / "entity-discovery-backlog.json", default=[]) or [] entity_queues = read_json(ROOT / "08-threat-intel" / "generated" / "entity-queues.json", default={}) or {} + version_completeness = read_json(ROOT / "08-threat-intel" / "generated" / "version-completeness.json", default={}) or {} + version_backlog = read_json(ROOT / "08-threat-intel" / "generated" / "version-backlog.json", default={}) or {} + release_index = read_json(ROOT / "08-threat-intel" / "generated" / "release-index.json", default={}) or {} + lab_enqueue_summary = read_json(ROOT / "08-threat-intel" / "generated" / "lab-enqueue-summary.json", default={}) or {} entity_records = load_json_dir(ROOT / "08-threat-intel" / "registry" / "entities") source_map = source_map_data if source_map_data is not None else (read_yaml(SOURCE_MAP_PATH, default={}) or {}) repro_map = repro_map_data if repro_map_data is not None else (read_yaml(REPRO_MAP_PATH, default={}) or {}) @@ -1373,6 +1384,7 @@ def render_dashboard( advisory_map = {item["canonical_id"]: item for item in merged_advisories if item.get("canonical_id")} profile_map = load_profiles() entity_summary_map = {item.get("system_id"): item for item in (entity_completeness.get("systems") or []) if item.get("system_id")} + version_summary_map = {item.get("system_id"): item for item in (version_completeness.get("systems") or []) if item.get("system_id")} entities_by_system: Dict[str, List[Dict[str, Any]]] = {} for item in sorted(entity_records, key=lambda value: (value.get("root_system_id") or "", value.get("entity_type") or "", value.get("display_name") or "")): entities_by_system.setdefault(item.get("root_system_id") or "", []).append(item)