#!/usr/bin/env python3 """Validate MLPS level-2/level-3 intake and material checklist CSV files.""" from __future__ import annotations import argparse import csv import json import re from collections import defaultdict from pathlib import Path INTAKE_COLUMNS = { "field_id", "section", "field_name", "applicable_level", "requirement", "value", "status", "evidence_level", "evidence_source", "owner", "confidentiality", "consistency_key", "notes", } MATERIAL_COLUMNS = { "material_id", "material_name", "applicable_level", "requirement", "status", "file_path", "issuer_or_owner", "seal_required", "review_required", "source_basis", "notes", } STATUS_VALUES = {"confirmed", "pending", "not_applicable", "conflict"} LEVEL_VALUES = {"common", "2", "3", "local"} REQUIREMENT_VALUES = {"required", "conditional", "optional"} CONFIDENTIALITY_VALUES = {"公开", "内部", "受限"} REQUIRED_FIELD_IDS = { "base.project_name", "base.system_name", "report.system_name", "filing.system_name", "base.level", "base.region", "base.template_version", "responsibility.entity", "filing.entity", "responsibility.security_dept", "responsibility.operation_unit", "scope.object_boundary", "scope.included_modules", "scope.domain_entry", "scope.identity", "scope.database", "business.services", "business.users", "business.account_stats", "business.impact", "service.impact", "data.categories", "data.volume", "data.personal_info", "data.important_core", "data.lifecycle", "interfaces.external", "network.topology", "network.zones", "network.external_links", "facility.rooms", "assets.servers", "assets.software", "assets.network", "security.identity", "security.audit", "security.vulnerability", "security.backup", "security.incident", "filing.review_conclusion", } LEVEL3_REQUIRED_FIELD_IDS = {"assets.security", "security.policies", "security.products"} REQUIRED_MATERIAL_IDS = {"M01", "M02", "M04", "M10", "M11", "M12", "M13"} LEVEL3_REQUIRED_MATERIAL_IDS = {"M14", "M15"} PLACEHOLDERS = {"", "待确认", "待补充", "待填写", "pending", "todo", "n/a", "na", "未知"} ABSOLUTE_TERMS = ("已启用", "已部署", "统一", "全部", "不存在", "不涉及", "无外联", "无数据流转") SECRET_RE = re.compile(r"(?i)(password|passwd|pwd|secret|token|access[_-]?key|private[_ -]?key|BEGIN [A-Z ]*PRIVATE KEY)") CHINESE_SECRET_RE = re.compile(r"(?:密码|口令|密钥|私钥)\s*[:=：]\s*\S+") MOBILE_RE = re.compile(r"(? list[dict[str, str]]: with path.open("r", encoding="utf-8-sig", newline="") as handle: reader = csv.DictReader(handle) columns = set(reader.fieldnames or []) missing = required_columns - columns if missing: raise ValueError(f"{path}: missing columns: {', '.join(sorted(missing))}") return [{key: (value or "").strip() for key, value in row.items()} for row in reader] def applies(row: dict[str, str], level: str) -> bool: return row.get("applicable_level") in {"common", level, "local"} def issue(items: list[dict[str, str]], severity: str, rule: str, item_id: str, message: str) -> None: items.append({"severity": severity, "rule": rule, "item_id": item_id, "message": message}) def validate_intake(rows: list[dict[str, str]], level: str, items: list[dict[str, str]]) -> None: seen_ids: set[str] = set() consistency: dict[str, list[tuple[str, str]]] = defaultdict(list) for row in rows: item_id = row["field_id"] or "" if item_id in seen_ids: issue(items, "HIGH", "DUPLICATE_ID", item_id, "字段 ID 重复") seen_ids.add(item_id) if row["applicable_level"] not in LEVEL_VALUES: issue(items, "HIGH", "INVALID_ENUM", item_id, f"非法 applicable_level: {row['applicable_level']}") if row["requirement"] not in REQUIREMENT_VALUES: issue(items, "HIGH", "INVALID_ENUM", item_id, f"非法 requirement: {row['requirement']}") if row["status"] not in STATUS_VALUES: issue(items, "HIGH", "INVALID_ENUM", item_id, f"非法 status: {row['status']}") if row["evidence_level"] and row["evidence_level"] not in {"A", "B", "C", "D"}: issue(items, "MEDIUM", "INVALID_ENUM", item_id, f"非法 evidence_level: {row['evidence_level']}") if row["confidentiality"] not in CONFIDENTIALITY_VALUES: issue(items, "MEDIUM", "INVALID_ENUM", item_id, f"非法 confidentiality: {row['confidentiality']}") if not applies(row, level): continue value_normalized = row["value"].strip().lower() missing_value = value_normalized in PLACEHOLDERS if row["requirement"] == "required" and (missing_value or row["status"] in {"pending", "conflict"}): issue(items, "HIGH", "REQUIRED_VALUE", item_id, f"必填字段未确认：{row['field_name']}") if row["status"] == "confirmed": if not row["evidence_level"] or not row["evidence_source"]: issue(items, "HIGH", "CONFIRMED_WITHOUT_EVIDENCE", item_id, "confirmed 缺少证据等级或来源") elif row["evidence_level"] == "D": issue(items, "HIGH", "CONFIRMED_BY_INFERENCE", item_id, "confirmed 不能仅由 D 级推测支撑") if row["status"] == "not_applicable" and not (row["notes"] or row["evidence_source"]): issue(items, "MEDIUM", "UNEXPLAINED_NA", item_id, "不适用缺少依据") if any(term in row["value"] for term in ABSOLUTE_TERMS) and not row["evidence_source"]: issue(items, "HIGH", "ABSOLUTE_CLAIM", item_id, "绝对表述缺少证据来源") sensitive = row["value"] + " " + row["notes"] if SECRET_RE.search(sensitive) or CHINESE_SECRET_RE.search(sensitive): issue(items, "BLOCKER", "SENSITIVE_VALUE", item_id, "疑似口令、令牌、密钥或私钥，不得进入通用资料") if ID_RE.search(sensitive) or MOBILE_RE.search(sensitive): severity = "MEDIUM" if row["confidentiality"] == "受限" else "BLOCKER" issue(items, severity, "SENSITIVE_VALUE", item_id, "疑似身份证号或手机号，应只保存在受控附件") if IP_RE.search(sensitive) and row["confidentiality"] != "受限": issue(items, "HIGH", "SENSITIVE_VALUE", item_id, "出现 IP 但密级不是受限") if row["consistency_key"] and row["status"] == "confirmed" and not missing_value: consistency[row["consistency_key"]].append((item_id, row["value"])) expected_ids = REQUIRED_FIELD_IDS | (LEVEL3_REQUIRED_FIELD_IDS if level == "3" else set()) for missing_id in sorted(expected_ids - seen_ids): issue(items, "BLOCKER", "MISSING_TEMPLATE_ROW", missing_id, "必需采集行被删除") for key, values in consistency.items(): distinct = {value for _, value in values} if len(distinct) > 1: detail = "; ".join(f"{item_id}={value}" for item_id, value in values) issue(items, "BLOCKER", "CONSISTENCY_CONFLICT", key, f"同一口径出现多个确认值：{detail}") declared = [row for row in rows if row["field_id"] == "base.level" and row["status"] == "confirmed"] for row in declared: accepted = {level, f"第{level}级", "第二级" if level == "2" else "第三级"} if row["value"] not in accepted: issue(items, "BLOCKER", "LEVEL_MISMATCH", "base.level", f"表内等级 {row['value']} 与校验等级 {level} 不一致") def validate_materials(rows: list[dict[str, str]], level: str, base_dir: Path, items: list[dict[str, str]]) -> None: seen_ids: set[str] = set() for row in rows: item_id = row["material_id"] or "" if item_id in seen_ids: issue(items, "HIGH", "DUPLICATE_ID", item_id, "材料 ID 重复") seen_ids.add(item_id) if row["applicable_level"] not in LEVEL_VALUES or row["requirement"] not in REQUIREMENT_VALUES or row["status"] not in STATUS_VALUES: issue(items, "HIGH", "INVALID_ENUM", item_id, "材料行存在非法适用等级、必要性或状态") if not applies(row, level): continue if row["requirement"] == "required" and row["status"] != "confirmed": issue(items, "HIGH", "REQUIRED_MATERIAL", item_id, f"必要材料未确认：{row['material_name']}") if row["status"] == "confirmed": if not row["file_path"]: issue(items, "HIGH", "REQUIRED_MATERIAL", item_id, "confirmed 材料缺少文件路径") else: material_path = Path(row["file_path"]) if not material_path.is_absolute(): material_path = base_dir / material_path if not material_path.exists(): issue(items, "HIGH", "MATERIAL_PATH_MISSING", item_id, f"材料路径不存在：{material_path}") if row["status"] == "not_applicable" and not (row["notes"] or row["source_basis"]): issue(items, "MEDIUM", "UNEXPLAINED_NA", item_id, "不适用缺少受理或模板依据") expected_ids = REQUIRED_MATERIAL_IDS | (LEVEL3_REQUIRED_MATERIAL_IDS if level == "3" else set()) for missing_id in sorted(expected_ids - seen_ids): issue(items, "BLOCKER", "MISSING_TEMPLATE_ROW", missing_id, "必要材料行被删除") def render_markdown(level: str, items: list[dict[str, str]]) -> str: order = {"BLOCKER": 0, "HIGH": 1, "MEDIUM": 2, "INFO": 3} sorted_items = sorted(items, key=lambda item: (order[item["severity"]], item["rule"], item["item_id"])) counts = {severity: sum(item["severity"] == severity for item in items) for severity in order} lines = [ "# 二级/三级等保资料缺项校验报告", "", f"- 校验等级：第 {level} 级", f"- 结果：BLOCKER {counts['BLOCKER']} / HIGH {counts['HIGH']} / MEDIUM {counts['MEDIUM']} / INFO {counts['INFO']}", "- 边界：本报告只做资料完整性、证据、矛盾和敏感信息检查，不代替专家、测评机构或受理机关结论。", "", "## 问题清单", "", ] if not sorted_items: lines.append("未发现脚本可识别的问题；仍需完成人工真实性与地方受理口径复核。") else: lines.extend(["| 严重度 | 规则 | 对象 | 问题 |", "| --- | --- | --- | --- |"]) for item in sorted_items: message = item["message"].replace("|", "\\|") lines.append(f"| {item['severity']} | {item['rule']} | {item['item_id']} | {message} |") lines.extend([ "", "## 人工复核", "", "- [ ] 定级对象、等级、责任主体和受理模板已确认", "- [ ] 拓扑、资产、网络外联、安全设备和报告正文一致", "- [ ] 业务/数据/账号统计带日期和来源", "- [ ] 三级危害论证由真实事实支撑", "- [ ] 受限附件未进入通用或公开仓库", "- [ ] 认证方、专家或受理机关差异意见已留书面记录", "", ]) return "\n".join(lines) def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--level", choices=("2", "3"), required=True) parser.add_argument("--intake", type=Path, required=True) parser.add_argument("--materials", type=Path, required=True) parser.add_argument("--report", type=Path, required=True) parser.add_argument("--json", dest="json_path", type=Path) args = parser.parse_args() issues: list[dict[str, str]] = [] intake_rows = read_csv(args.intake, INTAKE_COLUMNS) material_rows = read_csv(args.materials, MATERIAL_COLUMNS) validate_intake(intake_rows, args.level, issues) validate_materials(material_rows, args.level, args.materials.parent, issues) args.report.parent.mkdir(parents=True, exist_ok=True) args.report.write_text(render_markdown(args.level, issues), encoding="utf-8") if args.json_path: args.json_path.parent.mkdir(parents=True, exist_ok=True) args.json_path.write_text(json.dumps({"level": args.level, "issues": issues}, ensure_ascii=False, indent=2), encoding="utf-8") blockers = sum(item["severity"] == "BLOCKER" for item in issues) high = sum(item["severity"] == "HIGH" for item in issues) print(f"validated level={args.level} issues={len(issues)} blockers={blockers} high={high}") return 2 if blockers else (1 if high else 0) if __name__ == "__main__": raise SystemExit(main())