2026年安徽省网络信息产业
网络与信息安全管理员
职业技能竞赛官方 WP

#!/usr/bin/env python3
"""第1题 台账大抢救 · 官方复现脚本

用法：
    python3 solve_q1.py 选手包/附件/asset_inventory.csv \
        classification_result.csv 选手包/附件/canonicalize.py
"""
import csv, hashlib, subprocess, sys

MAPPING = {
    "办事指南": ("公共服务", "办事指南", "非个人信息", "L1", "PUBLIC|OPEN"),
    "服务目录": ("公共服务", "服务目录", "非个人信息", "L1", "PUBLIC|CATALOG"),
    "设备状态": ("运行管理", "设备状态", "非个人信息", "L2", "INTERNAL|OPS"),
    "安全审计": ("运行管理", "安全审计", "一般个人信息", "L3", "PI|AUDIT"),
    "身份标识": ("人口服务", "身份标识", "敏感个人信息", "L4", "SPI|IDENTITY"),
    "联系方式": ("人口服务", "联系方式", "一般个人信息", "L3", "PI|CONTACT"),
    "待遇信息": ("社会保障", "待遇信息", "敏感个人信息", "L4", "SPI|FINANCE"),
    "诊疗信息": ("卫生健康", "诊疗信息", "敏感个人信息", "L4", "SPI|HEALTH"),
    "企业登记": ("法人服务", "企业登记", "非个人信息", "L2", "INTERNAL|CORP"),
    "法定代表人": ("法人服务", "法定代表人", "一般个人信息", "L3", "PI|LEGAL_REP"),
    "共享交换": ("政务协同", "共享交换", "非个人信息", "L3", "IMPORTANT|EXCHANGE"),
    "人脸特征": ("生物识别", "人脸特征", "敏感个人信息", "L4", "SPI|BIOMETRIC"),
}
FIELDS = ["asset_id", "category", "subcategory", "personal_type", "level", "tags"]


def classify(asset_path, output_path):
    with open(asset_path, encoding="utf-8-sig", newline="") as f:
        assets = list(csv.DictReader(f))
    rows = []
    for r in assets:
        hits = [v for k, v in MAPPING.items() if k in r["purpose"]]
        hit = max(hits, key=lambda v: int(v[3][1:]))
        rows.append(dict(zip(FIELDS, [r["asset_id"], *hit])))
    rows.sort(key=lambda r: r["asset_id"])
    with open(output_path, "w", encoding="utf-8", newline="") as f:
        w = csv.DictWriter(f, fieldnames=FIELDS, lineterminator="\n")
        w.writeheader()
        w.writerows(rows)
    return rows


if __name__ == "__main__":
    if len(sys.argv) < 2:
        raise SystemExit("用法: python3 solve_q1.py asset_inventory.csv [out.csv] [canonicalize.py]")
    out = sys.argv[2] if len(sys.argv) > 2 else "classification_result.csv"
    rows = classify(sys.argv[1], out)
    print(f"{len(rows)} 条已写出 -> {out}")
    if len(sys.argv) > 3:
        subprocess.run([sys.executable, sys.argv[3], out], check=True)

#!/usr/bin/env python3
"""第3题 真假回执 · 官方复现脚本

依赖：本机 OpenSSL 需支持 SM2 / SM3 / SM4。
用法：
    python3 solve_q3.py 选手包/附件
"""
from __future__ import annotations
import hashlib, json, re, struct, subprocess, sys, tempfile
from pathlib import Path


def run(cmd, check=True):
    return subprocess.run(cmd, capture_output=True, check=check)


def openssl_verify(pub, sig, data):
    r = run(["openssl", "dgst", "-sm3", "-verify", str(pub), "-signature", str(sig), str(data)], check=False)
    return b"Verified OK" in (r.stdout + r.stderr)


def openssl_sm2_decrypt(priv, blob):
    return run(["openssl", "pkeyutl", "-decrypt", "-inkey", str(priv), "-in", str(blob)]).stdout


def openssl_sm4_cbc_decrypt(key, iv, ciphertext):
    with tempfile.TemporaryDirectory() as td:
        inp = Path(td) / "in.bin"
        inp.write_bytes(ciphertext)
        return run(["openssl", "enc", "-d", "-sm4-cbc", "-K", key.hex(), "-iv", iv.hex(), "-in", str(inp)]).stdout


def iter_pcapng_payloads(pcap: Path):
    data, payloads, off, n = pcap.read_bytes(), [], 0, 0
    n = len(data)
    while off + 8 <= n:
        block_type, total = struct.unpack_from("<II", data, off)
        if total < 12 or off + total > n:
            break
        body = data[off + 8 : off + total - 4]
        if block_type == 6 and len(body) >= 20:
            captlen = struct.unpack_from("<I", body, 12)[0]
            packet = body[20 : 20 + captlen]
            if len(packet) >= 54 and packet[12:14] == b"\x08\x00":
                ihl = (packet[14] & 0x0F) * 4
                tcp_off = 14 + ihl
                if len(packet) >= tcp_off + 13:
                    doff = (packet[tcp_off + 12] >> 4) * 4
                    payload = packet[tcp_off + doff :]
                    if payload:
                        payloads.append(payload)
        off += total
    return payloads


def analyze_traffic(pcap: Path):
    states = {}
    for raw in iter_pcapng_payloads(pcap):
        text = raw.decode("utf-8", errors="ignore")
        m = re.search(r"TLS-ATTEMPT-(\d+)\|.*?result=([A-Z_]+)(?:\|state=([A-Z]+))?", text)
        if m:
            states[int(m.group(1))] = f"{m.group(2)}|{m.group(3) or ''}".rstrip("|")
    return states


def main():
    if len(sys.argv) != 2:
        raise SystemExit("用法: python3 solve_q3.py <附件目录>")
    att = Path(sys.argv[1]).resolve()
    pub, priv = att / "trust/sender_sign_public.pem", att / "trust/receiver_sm2_private.pem"
    pcap, cand_root = att / "evidence/transfer.pcapng", att / "candidates"
    print("=== 1. 流量取证 ===")
    states = analyze_traffic(pcap)
    for a in sorted(states):
        print(f"  attempt={a}: {states[a]}")
    current = {a for a, s in states.items() if "ACCEPT" in s and "CURRENT" in s}
    print("  CURRENT =", sorted(current))

    print("=== 2. 批量验签 + 摘要 ===")
    verified = []
    for d in sorted(p for p in cand_root.iterdir() if p.is_dir() and p.name.startswith("xfer_")):
        meta = json.loads((d / f"{d.name}.json").read_text(encoding="utf-8"))
        ok = openssl_verify(pub, d / f"{d.name}.sig", d / f"{d.name}.json")
        digest_ok = hashlib.sha256((d / meta["encrypted_file"]).read_bytes()).hexdigest() == meta["encrypted_sha256"]
        print(f"  {d.name}: sig={'OK' if ok else 'FAIL'} sha256={'OK' if digest_ok else 'FAIL'} attempt={meta.get('transfer_attempt')}")
        if ok and digest_ok:
            verified.append((d, meta))

    matched = [(d, m) for d, m in verified if int(m.get("transfer_attempt", -1)) in current]
    if len(matched) != 1:
        raise SystemExit(f"未能唯一锁定: { [d.name for d, _ in matched] }")
    d, meta = matched[0]
    print(f"  => {d.name}")

    print("=== 3. 解密 ===")
    key = openssl_sm2_decrypt(priv, d / meta["wrapped_key_file"])
    plain = openssl_sm4_cbc_decrypt(key, bytes.fromhex(meta["iv_hex"]), (d / meta["encrypted_file"]).read_bytes())
    obj = json.loads(plain.decode("utf-8").strip())
    code = obj["receipt_code"]
    print(code)
    print("flag{" + hashlib.md5(code.encode("utf-8")).hexdigest() + "}")


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""第5题 上线前的评估底稿 · 官方复现脚本

用法：
    python3 solve_q5.py 选手包/附件
脚本会筛出入表风险、定级，写出 数据安全风险清单.csv，再调用官方 结果校验.py。
"""
from __future__ import annotations
import csv, subprocess, sys
from pathlib import Path

风险清单字段 = [
    "风险编号", "核查记录编号", "风险源编号", "标准评估主题",
    "风险危害程度", "风险发生可能性", "风险等级",
    "涉及数据编号", "涉及活动编号", "备注",
]
危害顺序 = ["低", "中", "较高", "高", "很高"]
可能性顺序 = ["低", "中", "高"]
矩阵 = {
    ("很高", "高"): "重大安全风险", ("很高", "中"): "重大安全风险", ("很高", "低"): "中安全风险",
    ("高", "高"): "重大安全风险", ("高", "中"): "高安全风险", ("高", "低"): "中安全风险",
    ("较高", "高"): "中安全风险", ("较高", "中"): "低安全风险", ("较高", "低"): "轻微安全风险",
    ("中", "高"): "低安全风险", ("中", "中"): "低安全风险", ("中", "低"): "轻微安全风险",
    ("低", "高"): "轻微安全风险", ("低", "中"): "轻微安全风险", ("低", "低"): "轻微安全风险",
}
阶段路径 = {
    "scope": "01_信息调研/01-01_评估委托与边界.csv",
    "assets": "01_信息调研/01-02_数据资产台账.csv",
    "findings": "02_风险识别/02-01_现场核查记录.csv",
    "impacts": "03_风险分析/03-01_危害分析底稿.csv",
    "controls": "03_风险分析/03-02_控制运行与事件证据.csv",
}


def 读表(att, key):
    with (att / 阶段路径[key]).open(encoding="utf-8-sig", newline="") as f:
        return list(csv.DictReader(f))


def 最高(values, order):
    return max(values, key=order.index)


def 频次等级(count):
    return "高" if count >= 6 else ("中" if count >= 1 else "低")


def 控制弱点等级(design, operation):
    if design in {"缺失", "无效"} or operation in {"缺失", "无效"}:
        return "高"
    if "部分有效" in {design, operation}:
        return "中"
    return "低"


def 关联等级(value):
    return {"易串联": "高", "可关联": "中", "独立": "低"}[value]


def main(att: Path):
    scope = {r["项目"]: r["内容"] for r in 读表(att, "scope")}
    assets = {r["数据编号"]: r for r in 读表(att, "assets")}
    findings = 读表(att, "findings")
    controls = {r["风险源编号"]: r for r in 读表(att, "controls")}
    impacts = 读表(att, "impacts")
    in_scope = set(scope["纳入范围处理活动"].split("；"))

    chosen = []
    for finding in findings:
        data_ids = finding["涉及数据编号"].split("+")
        activity_ids = finding["涉及活动编号"].split("+")
        valid = (
            finding["记录状态"] == "已确认"
            and all(assets.get(i, {}).get("是否纳入范围") == "是" for i in data_ids)
            and all(i in in_scope for i in activity_ids)
        )
        print(("保留 " if valid else "剔除 ") + finding["核查记录编号"])
        if valid:
            chosen.append(finding)
    chosen.sort(key=lambda x: x["核查记录编号"])

    register = []
    for index, finding in enumerate(chosen, 1):
        source = finding["风险源编号"]
        hazard = 最高([r["建议危害程度"] for r in impacts if r["风险源编号"] == source and r["复核状态"] == "有效"], 危害顺序)
        control = controls[source]
        likelihood = 最高([
            频次等级(int(control["六个月内同类事件或问题次数"])),
            控制弱点等级(control["控制设计有效性"], control["控制运行有效性"]),
            关联等级(control["风险源关联性"]),
        ], 可能性顺序)
        risk_id = f"R-{index:02d}"
        level = 矩阵[(hazard, likelihood)]
        register.append({
            "风险编号": risk_id, "核查记录编号": finding["核查记录编号"], "风险源编号": source,
            "标准评估主题": finding["标准评估主题"], "风险危害程度": hazard,
            "风险发生可能性": likelihood, "风险等级": level,
            "涉及数据编号": finding["涉及数据编号"], "涉及活动编号": finding["涉及活动编号"], "备注": "",
        })
        print(f"{risk_id}: 危害={hazard} 可能性={likelihood} -> {level}")

    out = att / "05_评估交付" / "数据安全风险清单.csv"
    with out.open("w", encoding="utf-8-sig", newline="") as f:
        w = csv.DictWriter(f, fieldnames=风险清单字段, lineterminator="\n")
        w.writeheader()
        w.writerows(register)
    result = subprocess.run(
        [sys.executable, str(att / "05_评估交付" / "结果校验.py"), str(out)],
        check=True, capture_output=True, text=True,
    )
    print(result.stdout, end="")


if __name__ == "__main__":
    if len(sys.argv) != 2:
        raise SystemExit("用法：python3 solve_q5.py <附件目录>")
    main(Path(sys.argv[1]).resolve())

export BASE_URL=http://题目地址
export Q6_USER='analyst01'
export Q6_PASS='Data@2026'
python3 solve_q6.py

#!/usr/bin/env python3
"""第6题 数据共享交换平台 · 公开复现脚本

用法：
    export BASE_URL=http://题目地址
    export Q6_USER='analyst01'
    export Q6_PASS='Data@2026'
    python3 solve_q6.py
"""
import hashlib, json, os, re, sys, urllib.parse, urllib.request
from http.cookiejar import CookieJar

BASE = os.environ.get("BASE_URL", "http://127.0.0.1:8000").rstrip("/")
USER = os.environ.get("Q6_USER", "analyst01")
PASS = os.environ.get("Q6_PASS", "Data@2026")


def extract_salt(source_code: str) -> str:
    match = re.search(r'RAW_EXPORT_INTEGRITY_SALT\s*=\s*"([^"]+)"', source_code)
    if not match:
        raise RuntimeError("Source Map 里没有找到完整性盐值变量名")
    return match.group(1)


def digest(app_id, dataset_id, mask_rule_id, token, salt):
    material = "|".join([str(app_id), str(dataset_id), str(mask_rule_id), token, "raw", salt])
    return hashlib.sha256(material.encode("utf-8")).hexdigest()[:32]


def main():
    if not USER or not PASS:
        raise SystemExit("缺少登录账号或口令")
    jar = CookieJar()
    opener = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(jar))
    body = urllib.parse.urlencode({"username": USER, "password": PASS}).encode()
    login = urllib.request.Request(
        BASE + "/login",
        data=body,
        headers={"Content-Type": "application/x-www-form-urlencoded"},
        method="POST",
    )
    opener.open(login)

    target = None
    for app_id in range(1000, 1101):
        try:
            with opener.open(BASE + f"/api/applications/{app_id}") as resp:
                payload = json.loads(resp.read().decode())
        except Exception:
            continue
        app = payload.get("application") or payload
        if app.get("dataset_id") == 7 and str(app.get("status", "")).lower() in {"approved", "已通过"}:
            target = app
            print(f"[+] 读到他人申请 {app_id}")
            break
    if not target:
        raise SystemExit("没有读到可导出的目标申请，请确认已登录且接口可达")

    with opener.open(BASE + "/static/js/export.min.js.map") as resp:
        source = json.loads(resp.read().decode())["sourcesContent"][0]
    salt = extract_salt(source)
    integ = digest(target["id"], target["dataset_id"], target["mask_rule_id"], target["approval_token"], salt)
    print("[+] 已按前端泄漏算法重算完整性摘要")

    req = urllib.request.Request(
        BASE + "/api/exports/create",
        data=json.dumps({
            "application_id": target["id"],
            "dataset_id": target["dataset_id"],
            "mask_rule_id": target["mask_rule_id"],
            "approval_token": target["approval_token"],
            "data_version": "raw",
            "format": "csv",
            "integrity_digest": integ,
        }).encode(),
        headers={"Content-Type": "application/json"},
        method="POST",
    )
    with opener.open(req) as resp:
        created = json.loads(resp.read().decode())
    with opener.open(BASE + created["download_url"]) as resp:
        csv_body = resp.read().decode()
    match = re.search(r"flag\{[^}]+\}", csv_body)
    if not match:
        raise SystemExit("CSV 里没有找到 flag 线索")
    inner = match.group(0)[5:-1]
    print("[+] 已从原始导出中取出线索，正在计算提交值")
    print("[+] 提交 Flag:", "flag{" + hashlib.md5(inner.encode("utf-8")).hexdigest() + "}")


if __name__ == "__main__":
    raise SystemExit(main())