hao/websafe-kb
1
0
派生 0
代码 工单 合并请求 1 工作流 软件包 项目 发布 百科 活动

更新: 2531 个文件 - 2026-03-17 21:00:03

浏览代码
这个提交包含在：
hao
2026-03-17 21:00:04 -07:00
父节点 a3edc88834
当前提交 080e55a98c
共修改 2531 个文件，包含 135521 行新增和 3725 行删除
下载 Patch 文件 下载 Diff 文件 展开所有文件 折叠所有文件

221
scripts/lab/render.py
查看文件

@@ -8,8 +8,8 @@ from pathlib import Path
from typing import Any, Dict, List
from lab.config import ADVISORIES_DIR, CASE_RUNS_DIR, DASHBOARD_DIR, REPRO_MAP_PATH, ROOT, RUNS_DIR, SOURCE_MAP_PATH
from lab.repro import load_profiles
from lab.utils import ensure_dir, isoformat, load_json_dir, now_utc, read_yaml, unique, write_json, write_text
from lab.repro import annotate_with_latest_run, latest_runs_by_advisory, load_profiles
from lab.utils import ensure_dir, isoformat, load_json_dir, now_utc, read_json, read_yaml, unique, write_json, write_text
TEMPLATES_DIR = ROOT / "scripts" / "lab" / "dashboard_templates"
@@ -156,9 +156,12 @@ def _profile_meta(profile: Dict[str, Any]) -> Dict[str, Any]:
"cleanup_policy": profile.get("cleanup_policy"),
"artifact_source": profile.get("artifact_source", {}),
"success_criteria": profile.get("success_criteria", []),
"success_assertions": profile.get("success_assertions", []),
"seed_actions": profile.get("seed_actions", []),
"attack_actions": profile.get("attack_actions", []),
"browser_assertions": profile.get("browser_assertions", {}),
"runner_id": profile.get("runner_id"),
"fixture_path": profile.get("fixture_path"),
"allowed_target_types": profile.get("allowed_target_types", []),
"required_services": profile.get("required_services", []),
}
@@ -211,6 +214,157 @@ def _status_label(value: str | None) -> str:
return STATUS_LABELS.get(value or "", value or "-")
def _family_name(advisory: Dict[str, Any], profile_map: Dict[str, Dict[str, Any]]) -> str:
profile = profile_map.get(advisory.get("repro_profile_id") or "", {})
return profile.get("vuln_family") or advisory.get("repro_profile_id") or "unknown"
def _latest_advisories(advisories: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
run_map = latest_runs_by_advisory()
return [annotate_with_latest_run(item, run_map.get(item.get("canonical_id"))) for item in advisories]
def _build_completeness(
advisories: List[Dict[str, Any]],
runs: List[Dict[str, Any]],
profile_map: Dict[str, Dict[str, Any]],
run_summary: Dict[str, Any],
) -> Dict[str, Any]:
latest_statuses: Dict[str, int] = {}
historical_statuses: Dict[str, int] = {}
systems: Dict[str, Dict[str, Any]] = {}
for item in advisories:
status = item.get("verification_status", "triage-manual")
latest_statuses[status] = latest_statuses.get(status, 0) + 1
system = systems.setdefault(
item["system_id"],
{
"system_id": item["system_id"],
"display_name": item.get("display_name", item["system_id"]),
"total": 0,
"verified_real": 0,
"verified_synthetic": 0,
"blocked": 0,
"manual": 0,
"families": {},
},
)
system["total"] += 1
family = _family_name(item, profile_map)
family_entry = system["families"].setdefault(
family,
{"family": family, "total": 0, "verified_real": 0, "verified_synthetic": 0, "blocked": 0, "manual": 0},
)
family_entry["total"] += 1
if status == "verified-real":
system["verified_real"] += 1
family_entry["verified_real"] += 1
elif status == "verified-synthetic":
system["verified_synthetic"] += 1
family_entry["verified_synthetic"] += 1
elif status.startswith("blocked-"):
system["blocked"] += 1
family_entry["blocked"] += 1
else:
system["manual"] += 1
family_entry["manual"] += 1
for item in runs:
status = item.get("verification_status", "triage-manual")
historical_statuses[status] = historical_statuses.get(status, 0) + 1
systems_list = []
for entry in sorted(systems.values(), key=lambda value: value["system_id"]):
entry["families"] = sorted(entry["families"].values(), key=lambda value: value["family"])
systems_list.append(entry)
advisory_total = len(advisories)
verified_real = latest_statuses.get("verified-real", 0)
verified_synthetic = latest_statuses.get("verified-synthetic", 0)
blocked = sum(count for key, count in latest_statuses.items() if key.startswith("blocked-"))
manual = advisory_total - verified_real - verified_synthetic - blocked
complete = advisory_total > 0 and advisory_total == verified_real
return {
"generated_at": isoformat(now_utc()),
"advisory_total": advisory_total,
"latest_statuses": latest_statuses,
"historical_statuses": historical_statuses,
"verified_real": verified_real,
"verified_synthetic": verified_synthetic,
"blocked": blocked,
"manual": manual,
"verified_ratio": round((verified_real / advisory_total) * 100, 1) if advisory_total else 0.0,
"complete": complete,
"systems": systems_list,
"ingest_health": {
"failure_count": len(run_summary.get("failures", []) or []),
"failures": run_summary.get("failures", []) or [],
},
"historical_blockers": [
"Docker daemon unavailable caused provision-compose-environment blocked-artifact.",
"Family profiles previously used note-only attack runners and dry-run placeholders.",
"Baseline and browser steps were skipped when environment readiness was not enforced.",
"Latest completeness now uses one advisory -> latest run semantics instead of historical run piles.",
],
}
def _write_testing_completeness_report(completeness: Dict[str, Any]) -> None:
lines = [
"# 全库 Advisory 完整度报告",
"",
f"- 生成时间: `{completeness['generated_at']}`",
f"- 最新 advisory 完整度: `{completeness['verified_real']}/{completeness['advisory_total']}` `verified-real`",
f"- 合成验证数量: `{completeness['verified_synthetic']}`",
f"- 阻塞数量: `{completeness['blocked']}`",
f"- 人工/待补证据数量: `{completeness['manual']}`",
f"- 完整度百分比: `{completeness['verified_ratio']}%`",
"",
"## 系统覆盖矩阵",
"",
"| 系统 | 总数 | verified-real | verified-synthetic | blocked | manual | family 覆盖 |",
"| --- | ---: | ---: | ---: | ---: | ---: | --- |",
]
for system in completeness["systems"]:
family_text = ", ".join(
f"{item['family']}({item['verified_real']}/{item['total']})" for item in system["families"]
)
lines.append(
f"| {system['system_id']} | {system['total']} | {system['verified_real']} | {system['verified_synthetic']} | {system['blocked']} | {system['manual']} | {family_text} |"
)
lines.extend(
[
"",
"## 历史阻塞项修复纪要",
"",
]
)
for item in completeness.get("historical_blockers", []):
lines.append(f"- {item}")
lines.extend(
[
"",
"## Ingest / Source 健康度",
"",
f"- source failures: `{completeness['ingest_health']['failure_count']}`",
]
)
for item in completeness["ingest_health"].get("failures", []):
lines.append(f"- {item}")
lines.extend(
[
"",
"## 剩余风险说明",
"",
"- 本报告按 advisory 的最新 run 计算；历史失败 run 仅保留审计价值，不再污染完整度数字。",
"- `browser_required=true` 的案例必须同时存在基线与攻击后浏览器证据，缺失则不会进入 `verified-real`。",
"- source collector 健康度单独计数；只有当 failures 归零时，报告与 dashboard 才算真正全绿。",
]
)
write_text(ROOT / "docs" / "testing-completeness-report.md", "\n".join(lines))
def _build_architecture_data(summary: Dict[str, Any], source_map: Dict[str, Any], repro_map: Dict[str, Any]) -> Dict[str, Any]:
source_systems = source_map.get("systems", []) or []
repro_by_system = {item.get("system_id"): item for item in (repro_map.get("systems", []) or []) if item.get("system_id")}
@@ -225,6 +379,7 @@ def _build_architecture_data(summary: Dict[str, Any], source_map: Dict[str, Any]
_link("旧版工作台", "/legacy/index.html", "保留的 legacy 回退入口。"),
_link("项目功能文档", "/docs/project-features.html", "项目能力、目录结构与自动化链路总览。"),
_link("前端设计文档", "/docs/frontend-dashboard-design.html", "当前本地工作台的交互与视觉规范。"),
_link("完整度报告", "/docs/testing-completeness-report.html", "89 条 advisory 的最新完整度中文报告。"),
_link("安全编码索引", "/docs/secure-code-index.html", "secure-code 修复库本地镜像。"),
_link("仓库入口镜像", "/docs/root-readme.html", "仓库根 README 的本地镜像。"),
_link("授权模型", "/docs/authorization-model.html", "允许目标范围、全局原则与记录要求。"),
@@ -237,6 +392,7 @@ def _build_architecture_data(summary: Dict[str, Any], source_map: Dict[str, Any]
data_links = [
_link("summary.json", "/summary.json", "全局摘要、状态分布和最近失败。"),
_link("completeness.json", "/data/completeness.json", "最新 advisory 完整度、系统/family 进度与 ingest 健康度。"),
_link("runs.json", "/runs.json", "最近 run 的结构化详情。"),
_link("systems.json", "/systems.json", "系统级覆盖与浏览器证据摘要。"),
_link("advisories.json", "/advisories.json", "漏洞条目元数据与来源。"),
@@ -736,6 +892,12 @@ def _write_dashboard_docs(architecture: Dict[str, Any]) -> None:
(ROOT / "08-threat-intel" / "generated" / "coverage-matrix.md").read_text(encoding="utf-8"),
"工作台内置镜像页：当前覆盖矩阵生成结果。",
),
(
"testing-completeness-report.html",
"中文完整度报告",
(ROOT / "docs" / "testing-completeness-report.md").read_text(encoding="utf-8"),
"工作台内置镜像页：89 条 advisory 最新完整度、family 矩阵与 ingest 健康度。",
),
]
manifest_body = LOVART_VENDOR_MANIFEST.read_text(encoding="utf-8") if LOVART_VENDOR_MANIFEST.exists() else "{}"
@@ -981,16 +1143,18 @@ def render_dashboard() -> Dict[str, str]:
ensure_dir(DASHBOARD_DIR)
advisory_records = load_json_dir(ADVISORIES_DIR)
runs = load_json_dir(RUNS_DIR)
run_summary = read_json(ROOT / "08-threat-intel" / "generated" / "run-summary.json", default={}) or {}
source_map = read_yaml(SOURCE_MAP_PATH, default={}) or {}
repro_map = read_yaml(REPRO_MAP_PATH, default={}) or {}
source_system_map = {item["system_id"]: item for item in source_map.get("systems", []) if item.get("system_id")}
advisory_map = {item["canonical_id"]: item for item in advisory_records if item.get("canonical_id")}
merged_advisories = _latest_advisories(advisory_records)
advisory_map = {item["canonical_id"]: item for item in merged_advisories if item.get("canonical_id")}
profile_map = load_profiles()
_sync_run_bundles(runs)
systems: Dict[str, Dict[str, Any]] = {}
for advisory in advisory_records:
for advisory in merged_advisories:
system = systems.setdefault(
advisory["system_id"],
{
@@ -1007,6 +1171,7 @@ def render_dashboard() -> Dict[str, str]:
"category": source_system_map.get(advisory["system_id"], {}).get("category", advisory.get("category")),
"tier": source_system_map.get(advisory["system_id"], {}).get("tier"),
"output_dir": source_system_map.get(advisory["system_id"], {}).get("output_dir"),
"families": {},
},
)
system["total"] += 1
@@ -1019,7 +1184,7 @@ def render_dashboard() -> Dict[str, str]:
system["blocked"] += 1
else:
system["manual"] += 1
browser = advisory.get("browser_evidence", {})
browser = advisory.get("browser_evidence") or {}
if browser.get("required"):
system["browser_required"] += 1
if browser.get("present"):
@@ -1027,6 +1192,13 @@ def render_dashboard() -> Dict[str, str]:
latest = advisory.get("updated_at") or advisory.get("published_at") or ""
if latest > system["latest_update"]:
system["latest_update"] = latest
family = _family_name(advisory, profile_map)
family_entry = system["families"].setdefault(family, {"family": family, "total": 0, "verified_real": 0, "manual": 0})
family_entry["total"] += 1
if status == "verified-real":
family_entry["verified_real"] += 1
elif status not in {"verified-synthetic"}:
family_entry["manual"] += 1
recent_runs = sorted(runs, key=lambda item: item.get("finished_at") or "", reverse=True)[:100]
decorated_runs: List[Dict[str, Any]] = []
@@ -1079,32 +1251,57 @@ def render_dashboard() -> Dict[str, str]:
summary = {
"generated_at": isoformat(now_utc()),
"advisory_count": len(advisory_records),
"advisory_count": len(merged_advisories),
"run_count": len(runs),
"statuses": {},
"run_statuses": {},
"recent_failures": [],
}
for item in runs:
for item in merged_advisories:
status = item.get("verification_status", "triage-manual")
summary["statuses"][status] = summary["statuses"].get(status, 0) + 1
summary["systems"] = sorted(systems.values(), key=lambda entry: (-entry["total"], entry["system_id"]))
for item in runs:
status = item.get("verification_status", "triage-manual")
summary["run_statuses"][status] = summary["run_statuses"].get(status, 0) + 1
summary["systems"] = sorted(
[
{
**entry,
"families": sorted(entry["families"].values(), key=lambda value: value["family"]),
}
for entry in systems.values()
],
key=lambda entry: (-entry["total"], entry["system_id"]),
)
summary["recent_failures"] = [
{
"run_id": item["run_id"],
"advisory_id": item["advisory_id"],
"run_id": item.get("last_run_id"),
"advisory_id": item["canonical_id"],
"status": item.get("verification_status"),
"title": item.get("advisory_meta", {}).get("title"),
"title": item.get("title"),
"blocked_reason": item.get("blocked_reason"),
}
for item in decorated_runs
for item in sorted(merged_advisories, key=lambda value: value.get("updated_at") or value.get("published_at") or "", reverse=True)
if item.get("verification_status") in {"triage-manual", "blocked-artifact", "blocked-destructive"}
][:20]
completeness = _build_completeness(merged_advisories, runs, profile_map, run_summary)
summary["completeness"] = {
"advisory_total": completeness["advisory_total"],
"verified_real": completeness["verified_real"],
"verified_synthetic": completeness["verified_synthetic"],
"blocked": completeness["blocked"],
"manual": completeness["manual"],
"verified_ratio": completeness["verified_ratio"],
"complete": completeness["complete"],
}
write_json(DASHBOARD_DIR / "summary.json", summary)
write_json(DASHBOARD_DIR / "runs.json", decorated_runs)
write_json(DASHBOARD_DIR / "systems.json", summary["systems"])
write_json(DASHBOARD_DIR / "advisories.json", {key: _advisory_meta(value) for key, value in advisory_map.items()})
write_json(DASHBOARD_DIR / "profiles.json", {key: _profile_meta(value) for key, value in profile_map.items()})
write_json(DASHBOARD_DIR / "data" / "completeness.json", completeness)
_write_testing_completeness_report(completeness)
architecture = _build_architecture_data(summary, source_map, repro_map)
write_json(DASHBOARD_DIR / "architecture.json", architecture)