diff --git a/client/scripts/sanitize_sku_panel_evidence.py b/client/scripts/sanitize_sku_panel_evidence.py new file mode 100644 index 0000000..c87f61f --- /dev/null +++ b/client/scripts/sanitize_sku_panel_evidence.py @@ -0,0 +1,41 @@ +"""本机脱敏 T-103 raw 证据到同级 derived;不连接设备或解析页面语义。""" + +from __future__ import annotations + +import argparse +from pathlib import Path +import sys + + +CLIENT_ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(CLIENT_ROOT / "src")) + +from cmbuyer_client.device.sku_evidence_sanitizer import ( + SkuEvidenceSanitizationError, + sanitize_sku_panel_evidence, +) + + +def parse_arguments(argv: list[str] | None = None) -> argparse.Namespace: + parser = argparse.ArgumentParser(description="将本机 raw 规格面板证据确定性脱敏到同级 derived。") + parser.add_argument("--raw-dir", required=True, type=Path, help="仅允许名为 raw 的本机原始证据目录。") + parser.add_argument("--output-dir", required=True, type=Path, help="仅允许 raw 同级且名为 derived 的新目录。") + return parser.parse_args(argv) + + +def main(argv: list[str] | None = None) -> int: + arguments = parse_arguments(argv) + try: + result = sanitize_sku_panel_evidence(arguments.raw_dir, arguments.output_dir) + except SkuEvidenceSanitizationError as error: + # 错误不回显 raw 路径、manifest/XML、地址、手机号或 serial。 + print(f"证据脱敏失败:{error}", file=sys.stderr) + return 1 + + print(f"派生证据脱敏完成:{result.output_directory}") + print(f"manifest:{result.manifest_path}") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/client/src/cmbuyer_client/device/sku_evidence_sanitizer.py b/client/src/cmbuyer_client/device/sku_evidence_sanitizer.py new file mode 100644 index 0000000..5ee06df --- /dev/null +++ b/client/src/cmbuyer_client/device/sku_evidence_sanitizer.py @@ -0,0 +1,426 @@ +"""T-103 原始规格面板证据的本机确定性隐私脱敏。 + +此模块只处理人工采集的本地文件:不连接设备、不理解拼多多页面,也不识别规格或价格。 +""" + +from __future__ import annotations + +from dataclasses import dataclass +from hashlib import sha256 +import json +from pathlib import Path +import re +import shutil +from typing import Any +from uuid import uuid4 +from xml.etree import ElementTree + +from PIL import Image, ImageDraw, UnidentifiedImageError + +from ..pdd.product_url import ProductUrl, ProductUrlError, parse_product_url + + +SANITIZER_VERSION = "t103-privacy-v1" +EXPECTED_GOODS_ID = "937122477375" +EXPECTED_PDD_VERSION = "8.17.0" +EXPECTED_DEVICE_MODEL = "PKG110" +EXPECTED_ANDROID_VERSION = "16" +EXPECTED_WIDTH = 1080 +EXPECTED_HEIGHT = 2400 +HUMAN_DECLARED_STATES = frozenset( + {"initial", "one-dimension-selected", "all-dimensions-selected"} +) +_ARTIFACT_FILES = ("screenshot.png", "hierarchy.xml") +_SHA256_RE = re.compile(r"[0-9a-f]{64}\Z") +_BOUNDS_RE = re.compile(r"\[(-?\d+),(-?\d+)\]\[(-?\d+),(-?\d+)\]\Z") +_FULL_PHONE_RE = re.compile(r"(?:\+?86)?1[3-9]\d{9}") +_MASKED_PHONE_RE = re.compile(r"1[3-9]\d\*{4}\d{4}") +_MASK_TRANSLATION = str.maketrans({"*": "*", "•": "*", "·": "*", "×": "*", "x": "*", "X": "*"}) +_SEPARATOR_RE = re.compile(r"[\s\-‐‑‒–—―()()]+") + + +class SkuEvidenceSanitizationError(RuntimeError): + """原始证据不能被安全地发布为派生证据。""" + + +@dataclass +class _CleanupStats: + """仅记录节点数量,供派生 manifest 审计;不记录任何页面文本。""" + + removed_nodes: int = 0 + cleared_crossing_nodes: int = 0 + retained_below_nodes: int = 0 + + +@dataclass(frozen=True) +class PrivacyMaskConfig: + """仅描述已人工确认的隐私几何区域,绝不承担页面或规格判据。""" + + version: str + width: int + height: int + privacy_top: int + + +PRIVACY_MASK_CONFIG = PrivacyMaskConfig( + version=SANITIZER_VERSION, + width=EXPECTED_WIDTH, + height=EXPECTED_HEIGHT, + # 主审在原始截图确认 y < 540 为收货/手机号区域;整宽遮罩优先保护隐私而非保留版面。 + privacy_top=540, +) + + +@dataclass(frozen=True) +class SkuEvidenceSanitizationResult: + """已经原子发布的派生证据位置。""" + + output_directory: Path + manifest_path: Path + screenshot_path: Path + hierarchy_path: Path + + +def sanitize_sku_panel_evidence(raw_directory: Path, output_directory: Path) -> SkuEvidenceSanitizationResult: + """校验 raw 三文件,并发布同级 ``derived`` 的脱敏副本。 + + 目标已存在时在读取任何输入前拒绝,避免混入旧派生物或覆盖人工保留文件。 + """ + + raw = Path(raw_directory) + target = Path(output_directory) + _validate_directories(raw, target) + if target.exists(): + raise SkuEvidenceSanitizationError("派生证据目录已存在,拒绝覆盖。") + + staging: Path | None = None + try: + source_manifest_path = _required_file(raw, "manifest.json") + source_screenshot_path = _required_file(raw, "screenshot.png") + source_hierarchy_path = _required_file(raw, "hierarchy.xml") + manifest = _read_source_manifest(source_manifest_path) + link, state, source_hashes = _validate_source_manifest(manifest) + _verify_source_hashes(source_screenshot_path, source_hierarchy_path, source_hashes) + + staging = raw.parent / f".derived.staging-{uuid4().hex}" + staging.mkdir() + derived_screenshot_path = staging / "screenshot.png" + _sanitize_screenshot(source_screenshot_path, derived_screenshot_path) + derived_hierarchy_path = staging / "hierarchy.xml" + cleanup_stats = _sanitize_hierarchy(source_hierarchy_path, derived_hierarchy_path) + + derived_manifest_path = staging / "manifest.json" + derived_manifest_path.write_text( + json.dumps( + _derived_manifest( + manifest, + link, + state, + source_manifest_path, + source_screenshot_path, + source_hierarchy_path, + derived_screenshot_path, + derived_hierarchy_path, + cleanup_stats, + ), + ensure_ascii=False, + indent=2, + sort_keys=True, + ) + + "\n", + encoding="utf-8", + ) + _publish_staging(staging, target) + except SkuEvidenceSanitizationError: + _clean_staging(staging) + raise + except (OSError, ValueError, ElementTree.ParseError, UnidentifiedImageError) as error: + _clean_staging(staging) + # 原始异常可能含文件路径、JSON/XML 文本或其他敏感内容,不能向 CLI/日志传播。 + raise SkuEvidenceSanitizationError("原始证据无法安全脱敏,未发布任何派生产物。") from error + except Exception as error: + _clean_staging(staging) + raise SkuEvidenceSanitizationError("原始证据脱敏未完成,未发布任何派生产物。") from error + + return SkuEvidenceSanitizationResult( + output_directory=target, + manifest_path=target / "manifest.json", + screenshot_path=target / "screenshot.png", + hierarchy_path=target / "hierarchy.xml", + ) + + +def _validate_directories(raw: Path, target: Path) -> None: + if raw.name != "raw" or not raw.is_dir(): + raise SkuEvidenceSanitizationError("原始证据目录必须是存在的 raw 目录。") + if target.name != "derived" or target.parent != raw.parent: + raise SkuEvidenceSanitizationError("派生证据目录必须是 raw 同级的 derived 目录。") + + +def _required_file(raw: Path, filename: str) -> Path: + candidate = raw / filename + if not candidate.is_file(): + raise SkuEvidenceSanitizationError("原始证据文件集合不完整。") + return candidate + + +def _read_source_manifest(path: Path) -> dict[str, Any]: + try: + value = json.loads(path.read_text(encoding="utf-8")) + except (OSError, UnicodeDecodeError, json.JSONDecodeError) as error: + raise SkuEvidenceSanitizationError("原始证据 manifest 无效。") from error + if not isinstance(value, dict): + raise SkuEvidenceSanitizationError("原始证据 manifest 结构无效。") + return value + + +def _validate_source_manifest(manifest: dict[str, Any]) -> tuple[ProductUrl, str, dict[str, str]]: + product = manifest.get("product") + device = manifest.get("device") + state = manifest.get("human_declared_state") + if manifest.get("schema_version") != 1 or not isinstance(product, dict) or not isinstance(device, dict): + raise SkuEvidenceSanitizationError("原始证据 manifest 缺少必要元数据。") + canonical_url = product.get("canonical_url") + goods_id = product.get("goods_id") + try: + link = parse_product_url(canonical_url) + except ProductUrlError as error: + raise SkuEvidenceSanitizationError("原始证据商品元数据不匹配。") from error + if ( + link.goods_id != EXPECTED_GOODS_ID + or goods_id != EXPECTED_GOODS_ID + or device.get("model") != EXPECTED_DEVICE_MODEL + or device.get("android_version") != EXPECTED_ANDROID_VERSION + or device.get("pdd_version") != EXPECTED_PDD_VERSION + or device.get("pdd_package") != "com.xunmeng.pinduoduo" + or not isinstance(state, str) + or state not in HUMAN_DECLARED_STATES + ): + raise SkuEvidenceSanitizationError("原始证据元数据与脱敏配置不匹配。") + return link, state, _artifact_hashes(manifest) + + +def _artifact_hashes(manifest: dict[str, Any]) -> dict[str, str]: + artifacts = manifest.get("artifacts") + if not isinstance(artifacts, list): + raise SkuEvidenceSanitizationError("原始证据 manifest 缺少文件校验信息。") + hashes: dict[str, str] = {} + for artifact in artifacts: + if not isinstance(artifact, dict): + raise SkuEvidenceSanitizationError("原始证据 manifest 文件校验信息无效。") + path = artifact.get("path") + digest = artifact.get("sha256") + if path not in _ARTIFACT_FILES or path in hashes or not isinstance(digest, str) or not _SHA256_RE.fullmatch(digest): + raise SkuEvidenceSanitizationError("原始证据 manifest 文件校验信息无效。") + hashes[path] = digest + if set(hashes) != set(_ARTIFACT_FILES): + raise SkuEvidenceSanitizationError("原始证据 manifest 文件校验信息不完整。") + return hashes + + +def _verify_source_hashes(screenshot_path: Path, hierarchy_path: Path, expected: dict[str, str]) -> None: + if ( + _sha256_file(screenshot_path) != expected["screenshot.png"] + or _sha256_file(hierarchy_path) != expected["hierarchy.xml"] + ): + raise SkuEvidenceSanitizationError("原始证据文件校验失败。") + + +def _sanitize_screenshot(source: Path, target: Path) -> None: + try: + with Image.open(source) as image: + image.load() + if image.format != "PNG" or image.size != (PRIVACY_MASK_CONFIG.width, PRIVACY_MASK_CONFIG.height): + raise SkuEvidenceSanitizationError("原始截图分辨率或格式与脱敏配置不匹配。") + sanitized = image.convert("RGBA") + except SkuEvidenceSanitizationError: + raise + except (OSError, UnidentifiedImageError) as error: + raise SkuEvidenceSanitizationError("原始截图无效。") from error + + # 用不透明黑色覆盖 y < 540,保证截图与 XML 使用相同的隐私几何边界。 + ImageDraw.Draw(sanitized).rectangle( + (0, 0, PRIVACY_MASK_CONFIG.width - 1, PRIVACY_MASK_CONFIG.privacy_top - 1), + fill=(0, 0, 0, 255), + ) + sanitized.save(target, format="PNG", optimize=False, compress_level=9) + + +def _sanitize_hierarchy(source: Path, target: Path) -> _CleanupStats: + try: + root = ElementTree.parse(source).getroot() + except (OSError, ElementTree.ParseError) as error: + raise SkuEvidenceSanitizationError("原始节点树无效。") from error + if root.tag != "hierarchy": + raise SkuEvidenceSanitizationError("原始节点树结构不匹配。") + if not list(root): + raise SkuEvidenceSanitizationError("原始节点树结构不匹配。") + stats = _CleanupStats() + _clear_node_text(root) + for child in list(root): + _sanitize_node(root, child, stats) + if stats.removed_nodes < 1 or stats.retained_below_nodes < 1: + raise SkuEvidenceSanitizationError("原始节点树未满足隐私几何结构。") + if _contains_phone(root): + raise SkuEvidenceSanitizationError("派生节点树仍包含手机号,拒绝发布。") + ElementTree.ElementTree(root).write(target, encoding="utf-8", xml_declaration=True) + return stats + + +def _sanitize_node(parent: ElementTree.Element, node: ElementTree.Element, stats: _CleanupStats) -> None: + if node.tag != "node": + raise SkuEvidenceSanitizationError("原始节点树结构不匹配。") + bounds = _parse_bounds(node.get("bounds")) + position = _vertical_position(bounds) + if position == "private": + # 私有带内的父节点不可以悄然包含下方子节点,否则会把仍需审计的下方内容一起丢失。 + if any(_vertical_position(_parse_bounds(descendant.get("bounds"))) != "private" for descendant in node.iter("node")): + raise SkuEvidenceSanitizationError("原始节点树 bounds 结构不匹配。") + stats.removed_nodes += sum(1 for _ in node.iter("node")) + parent.remove(node) + return + if position == "crossing": + # 全屏/跨界容器可保留其下方子节点,但自身所有属性和文本都可能含地址或手机号。 + _clear_node_text(node) + stats.cleared_crossing_nodes += 1 + else: + stats.retained_below_nodes += 1 + for child in list(node): + _sanitize_node(node, child, stats) + + +def _parse_bounds(value: object) -> tuple[int, int, int, int]: + if not isinstance(value, str): + raise SkuEvidenceSanitizationError("原始节点树 bounds 缺失或无效。") + match = _BOUNDS_RE.fullmatch(value) + if match is None: + raise SkuEvidenceSanitizationError("原始节点树 bounds 缺失或无效。") + left, top, right, bottom = (int(group) for group in match.groups()) + if not (0 <= left < right <= PRIVACY_MASK_CONFIG.width and 0 <= top < bottom <= PRIVACY_MASK_CONFIG.height): + raise SkuEvidenceSanitizationError("原始节点树 bounds 超出脱敏配置。") + return left, top, right, bottom + + +def _vertical_position(bounds: tuple[int, int, int, int]) -> str: + _, top, _, bottom = bounds + if bottom <= PRIVACY_MASK_CONFIG.privacy_top: + return "private" + if top >= PRIVACY_MASK_CONFIG.privacy_top: + return "below" + return "crossing" + + +def _clear_node_text(node: ElementTree.Element) -> None: + node.attrib = {"bounds": node.attrib["bounds"]} if "bounds" in node.attrib else {} + node.text = None + node.tail = None + + +def _contains_phone(root: ElementTree.Element) -> bool: + """逐项与跨节点复检电话,避免分隔符、遮罩字符或节点切分绕过。""" + + all_values: list[str] = [] + content_values: list[str] = [] + for element in root.iter(): + if element.text: + all_values.append(element.text) + content_values.append(element.text) + for attribute, value in element.attrib.items(): + all_values.append(value) + if attribute != "bounds": + content_values.append(value) + if element.tail: + all_values.append(element.tail) + content_values.append(element.tail) + normalized_values = [_normalize_phone_value(value) for value in all_values] + normalized_all_document = "".join(normalized_values) + normalized_document = "".join(_normalize_phone_value(value) for value in content_values) + return ( + any(_matches_phone(value) for value in normalized_values) + or _matches_phone(normalized_all_document) + or _matches_phone(normalized_document) + ) + + +def _normalize_phone_value(value: str) -> str: + return _SEPARATOR_RE.sub("", value.translate(_MASK_TRANSLATION)) + + +def _matches_phone(value: str) -> bool: + return _FULL_PHONE_RE.search(value) is not None or _MASKED_PHONE_RE.search(value) is not None + + +def _derived_manifest( + source_manifest: dict[str, Any], + link: ProductUrl, + state: str, + source_manifest_path: Path, + source_screenshot_path: Path, + source_hierarchy_path: Path, + derived_screenshot_path: Path, + derived_hierarchy_path: Path, + cleanup_stats: _CleanupStats, +) -> dict[str, Any]: + device = source_manifest["device"] + return { + "schema_version": 1, + "privacy_tier": "SANITIZED", + "sanitizer_version": PRIVACY_MASK_CONFIG.version, + "privacy_mask": { + "width": PRIVACY_MASK_CONFIG.width, + "height": PRIVACY_MASK_CONFIG.height, + "rectangle": [0, 0, PRIVACY_MASK_CONFIG.width, PRIVACY_MASK_CONFIG.privacy_top], + }, + "privacy_cleanup": { + "removed_nodes": cleanup_stats.removed_nodes, + "cleared_crossing_nodes": cleanup_stats.cleared_crossing_nodes, + "retained_below_nodes": cleanup_stats.retained_below_nodes, + }, + "product": {"goods_id": link.goods_id}, + "human_declared_state": state, + "device": { + "model": device["model"], + "android_version": device.get("android_version"), + "pdd_package": device["pdd_package"], + "pdd_version": device["pdd_version"], + }, + # source hashes stay only in the local derived manifest; no raw path, serial or body is retained. + "source": { + "manifest_sha256": _sha256_file(source_manifest_path), + "artifacts": [ + {"path": "screenshot.png", "sha256": _sha256_file(source_screenshot_path)}, + {"path": "hierarchy.xml", "sha256": _sha256_file(source_hierarchy_path)}, + ], + }, + "derived": { + "artifacts": [ + {"path": "screenshot.png", "sha256": _sha256_file(derived_screenshot_path)}, + {"path": "hierarchy.xml", "sha256": _sha256_file(derived_hierarchy_path)}, + ] + }, + } + + +def _sha256_file(path: Path) -> str: + digest = sha256() + with path.open("rb") as source: + for chunk in iter(lambda: source.read(1024 * 1024), b""): + digest.update(chunk) + return digest.hexdigest() + + +def _clean_staging(staging: Path | None) -> None: + if staging is not None and staging.exists(): + shutil.rmtree(staging) + + +def _publish_staging(staging: Path, target: Path) -> None: + """发布前二次检查,并使用目录 rename 而不是会覆盖目标的 replace。""" + + if target.exists(): + raise SkuEvidenceSanitizationError("派生证据目录已存在,拒绝覆盖。") + try: + staging.rename(target) + except OSError as error: + # 竞态中新目标出现或文件系统拒绝 rename 时一律不尝试覆盖或重试。 + raise SkuEvidenceSanitizationError("派生证据目录发布失败,未覆盖已有目录。") from error diff --git a/client/tests/device/test_sku_evidence_sanitizer.py b/client/tests/device/test_sku_evidence_sanitizer.py new file mode 100644 index 0000000..b38f0fb --- /dev/null +++ b/client/tests/device/test_sku_evidence_sanitizer.py @@ -0,0 +1,341 @@ +"""T-103 脱敏器测试:全部证据为合成数据,绝不读取真实 raw 目录。""" + +from __future__ import annotations + +from hashlib import sha256 +import json +from pathlib import Path +import sys +from tempfile import TemporaryDirectory +import unittest +from unittest.mock import patch +from xml.etree import ElementTree + +from PIL import Image + + +CLIENT_ROOT = Path(__file__).resolve().parents[2] +sys.path.insert(0, str(CLIENT_ROOT / "src")) + +from cmbuyer_client.device.sku_evidence_sanitizer import ( + EXPECTED_GOODS_ID, + EXPECTED_HEIGHT, + EXPECTED_WIDTH, + HUMAN_DECLARED_STATES, + SkuEvidenceSanitizationError, + sanitize_sku_panel_evidence, +) + + +TEST_SERIAL = "synthetic-serial-never-publish" +TEST_ADDRESS = "SYNTHETIC_ADDRESS_NEVER_PUBLISH" +FULL_PHONE = "13800138000" +MASKED_PHONE = "138****0000" +SAFE_TEXT = "synthetic-safe-lower-content" + + +def _hash(path: Path) -> str: + digest = sha256() + with path.open("rb") as source: + for chunk in iter(lambda: source.read(1024 * 1024), b""): + digest.update(chunk) + return digest.hexdigest() + + +def _default_xml() -> str: + return ( + "" + f"" + f"" + "" + ) + + +def _write_raw( + root: Path, + *, + state: str = "initial", + goods_id: str = EXPECTED_GOODS_ID, + model: str = "PKG110", + android_version: str = "16", + pdd_version: str = "8.17.0", + size: tuple[int, int] = (EXPECTED_WIDTH, EXPECTED_HEIGHT), + xml: str | None = None, +) -> Path: + raw = root / "raw" + raw.mkdir(parents=True) + screenshot = raw / "screenshot.png" + image = Image.new("RGB", size, color=(0, 180, 0)) + if size == (EXPECTED_WIDTH, EXPECTED_HEIGHT): + for y in range(540): + for x in range(8): + image.putpixel((x, y), (255, 0, 0)) + image.save(screenshot, format="PNG") + hierarchy = raw / "hierarchy.xml" + hierarchy.write_text(_default_xml() if xml is None else xml, encoding="utf-8") + manifest = { + "schema_version": 1, + "product": { + "goods_id": goods_id, + "canonical_url": f"https://mobile.yangkeduo.com/goods.html?goods_id={goods_id}", + }, + "human_declared_state": state, + "serial": TEST_SERIAL, + "channel": "wifi", + "device": { + "model": model, + "android_version": android_version, + "pdd_package": "com.xunmeng.pinduoduo", + "pdd_version": pdd_version, + }, + "artifacts": [ + {"path": "screenshot.png", "sha256": _hash(screenshot)}, + {"path": "hierarchy.xml", "sha256": _hash(hierarchy)}, + ], + } + (raw / "manifest.json").write_text(json.dumps(manifest, sort_keys=True), encoding="utf-8") + return raw + + +class SkuEvidenceSanitizerTests(unittest.TestCase): + def test_all_declared_states_mask_screenshot_and_xml_without_raw_metadata(self) -> None: + for state in sorted(HUMAN_DECLARED_STATES): + with self.subTest(state=state), TemporaryDirectory() as temporary: + raw = _write_raw(Path(temporary), state=state) + result = sanitize_sku_panel_evidence(raw, raw.parent / "derived") + with Image.open(result.screenshot_path) as image: + self.assertEqual(image.getpixel((0, 0)), (0, 0, 0, 255)) + self.assertEqual(image.getpixel((100, 600)), (0, 180, 0, 255)) + derived_xml = result.hierarchy_path.read_text(encoding="utf-8") + manifest = result.manifest_path.read_text(encoding="utf-8") + + self.assertIn(SAFE_TEXT, derived_xml) + self.assertNotIn(TEST_ADDRESS, derived_xml) + self.assertNotIn(FULL_PHONE, derived_xml) + self.assertNotIn(MASKED_PHONE, derived_xml) + self.assertIn(f'"human_declared_state": "{state}"', manifest) + self.assertIn('"privacy_tier": "SANITIZED"', manifest) + self.assertIn('"sanitizer_version": "t103-privacy-v1"', manifest) + self.assertIn('"rectangle": [', manifest) + self.assertIn('"removed_nodes": 1', manifest) + self.assertIn('"cleared_crossing_nodes": 0', manifest) + self.assertIn('"retained_below_nodes": 1', manifest) + self.assertNotIn("canonical_url", manifest) + self.assertNotIn(TEST_SERIAL, manifest) + self.assertNotIn("serial", manifest) + self.assertNotIn("channel", manifest) + self.assertNotIn(TEST_ADDRESS, manifest) + self.assertNotIn(FULL_PHONE, manifest) + + def test_crossing_container_keeps_lower_children_but_clears_its_text(self) -> None: + xml = ( + "" + f"" + f"" + f"" + "" + ) + with TemporaryDirectory() as temporary: + raw = _write_raw(Path(temporary), xml=xml) + result = sanitize_sku_panel_evidence(raw, raw.parent / "derived") + root = ElementTree.parse(result.hierarchy_path).getroot() + crossing = root.find("node") + + self.assertIsNotNone(crossing) + assert crossing is not None + self.assertEqual(crossing.attrib, {"bounds": "[0,0][1080,2400]"}) + self.assertEqual(len(list(crossing)), 1) + self.assertEqual(list(crossing)[0].get("text"), SAFE_TEXT) + manifest = json.loads(result.manifest_path.read_text(encoding="utf-8")) + self.assertEqual( + manifest["privacy_cleanup"], + {"removed_nodes": 1, "cleared_crossing_nodes": 1, "retained_below_nodes": 1}, + ) + + def test_same_raw_and_config_produce_identical_derived_files(self) -> None: + with TemporaryDirectory() as temporary: + root = Path(temporary) + left_raw = _write_raw(root / "left") + right_raw = _write_raw(root / "right") + left = sanitize_sku_panel_evidence(left_raw, left_raw.parent / "derived") + right = sanitize_sku_panel_evidence(right_raw, right_raw.parent / "derived") + + for left_path, right_path in ( + (left.screenshot_path, right.screenshot_path), + (left.hierarchy_path, right.hierarchy_path), + (left.manifest_path, right.manifest_path), + ): + self.assertEqual(left_path.read_bytes(), right_path.read_bytes()) + + def test_hash_and_metadata_mismatch_fail_closed_without_leak(self) -> None: + scenarios = ( + ("hash", {}, "screenshot"), + ("model", {"model": "other"}, None), + ("android", {"android_version": "15"}, None), + ("version", {"pdd_version": "8.17.1"}, None), + ("goods", {"goods_id": "123"}, None), + ("state", {"state": "guessed"}, None), + ("resolution", {"size": (100, 100)}, None), + ) + for name, kwargs, corrupt_file in scenarios: + with self.subTest(name=name), TemporaryDirectory() as temporary: + raw = _write_raw(Path(temporary), **kwargs) + if corrupt_file is not None: + (raw / f"{corrupt_file}.png").write_bytes(b"changed") + with self.assertRaises(SkuEvidenceSanitizationError) as raised: + sanitize_sku_panel_evidence(raw, raw.parent / "derived") + + message = str(raised.exception) + self.assertNotIn(TEST_SERIAL, message) + self.assertNotIn(TEST_ADDRESS, message) + self.assertNotIn(FULL_PHONE, message) + self.assertFalse((raw.parent / "derived").exists()) + self.assertEqual(list(raw.parent.glob(".derived.staging-*")), []) + + def test_malformed_inputs_and_bounds_or_phone_residue_fail_closed(self) -> None: + malformed = ( + ("manifest", None), + ("png", None), + ("xml", None), + ("bounds", ""), + ( + "private-parent-with-lower-child", + "", + ), + ("full-phone-below", f""), + ("masked-phone-below", f""), + ) + for kind, xml in malformed: + with self.subTest(kind=kind), TemporaryDirectory() as temporary: + raw = _write_raw(Path(temporary), xml=xml) + if kind == "manifest": + (raw / "manifest.json").write_text("{invalid", encoding="utf-8") + elif kind == "png": + (raw / "screenshot.png").write_bytes(b"not a png") + manifest = json.loads((raw / "manifest.json").read_text(encoding="utf-8")) + manifest["artifacts"][0]["sha256"] = _hash(raw / "screenshot.png") + (raw / "manifest.json").write_text(json.dumps(manifest), encoding="utf-8") + elif kind == "xml": + (raw / "hierarchy.xml").write_text("", encoding="utf-8") + manifest = json.loads((raw / "manifest.json").read_text(encoding="utf-8")) + manifest["artifacts"][1]["sha256"] = _hash(raw / "hierarchy.xml") + (raw / "manifest.json").write_text(json.dumps(manifest), encoding="utf-8") + with self.assertRaises(SkuEvidenceSanitizationError): + sanitize_sku_panel_evidence(raw, raw.parent / "derived") + + self.assertFalse((raw.parent / "derived").exists()) + self.assertEqual(list(raw.parent.glob(".derived.staging-*")), []) + + def test_phone_recheck_rejects_separator_mask_and_cross_node_bypasses(self) -> None: + variants = ( + "138 0013-8000", + "138****0000", + "138••••0000", + "138xxxx0000", + "138XXXX0000", + ) + for value in variants: + xml = ( + "" + "" + f"" + "" + ) + with self.subTest(value=value), TemporaryDirectory() as temporary: + raw = _write_raw(Path(temporary), xml=xml) + with self.assertRaises(SkuEvidenceSanitizationError): + sanitize_sku_panel_evidence(raw, raw.parent / "derived") + self.assertFalse((raw.parent / "derived").exists()) + + split_xml = ( + "" + "" + "" + "" + "" + ) + with TemporaryDirectory() as temporary: + raw = _write_raw(Path(temporary), xml=split_xml) + with self.assertRaises(SkuEvidenceSanitizationError): + sanitize_sku_panel_evidence(raw, raw.parent / "derived") + self.assertFalse((raw.parent / "derived").exists()) + + def test_privacy_geometry_requires_removed_and_retained_nodes(self) -> None: + scenarios = ( + ("no-private", f""), + ("no-below", ""), + ) + for name, xml in scenarios: + with self.subTest(name=name), TemporaryDirectory() as temporary: + raw = _write_raw(Path(temporary), xml=xml) + with self.assertRaises(SkuEvidenceSanitizationError): + sanitize_sku_panel_evidence(raw, raw.parent / "derived") + self.assertFalse((raw.parent / "derived").exists()) + self.assertEqual(list(raw.parent.glob(".derived.staging-*")), []) + + def test_manifest_schema_package_and_artifact_structure_are_required(self) -> None: + def mutate(manifest: dict[str, object], kind: str) -> None: + if kind == "schema": + manifest["schema_version"] = 2 + elif kind == "package": + manifest["device"]["pdd_package"] = "com.example.other" # type: ignore[index] + elif kind == "missing": + manifest.pop("artifacts") + elif kind == "duplicate": + manifest["artifacts"].append(manifest["artifacts"][0]) # type: ignore[index] + elif kind == "bad-hash": + manifest["artifacts"][0]["sha256"] = "g" * 64 # type: ignore[index] + + for kind in ("schema", "package", "missing", "duplicate", "bad-hash"): + with self.subTest(kind=kind), TemporaryDirectory() as temporary: + raw = _write_raw(Path(temporary)) + manifest_path = raw / "manifest.json" + manifest = json.loads(manifest_path.read_text(encoding="utf-8")) + mutate(manifest, kind) + manifest_path.write_text(json.dumps(manifest), encoding="utf-8") + with self.assertRaises(SkuEvidenceSanitizationError) as raised: + sanitize_sku_panel_evidence(raw, raw.parent / "derived") + + self.assertNotIn(TEST_SERIAL, str(raised.exception)) + self.assertNotIn(TEST_ADDRESS, str(raised.exception)) + self.assertFalse((raw.parent / "derived").exists()) + + def test_target_created_during_publish_is_preserved_and_staging_is_removed(self) -> None: + with TemporaryDirectory() as temporary: + raw = _write_raw(Path(temporary)) + target = raw.parent / "derived" + + def race_rename(destination: Path) -> None: + destination.mkdir() + (destination / "sentinel.txt").write_text("keep", encoding="utf-8") + raise FileExistsError("simulated publish race") + + with patch("cmbuyer_client.device.sku_evidence_sanitizer.Path.rename", side_effect=race_rename): + with self.assertRaises(SkuEvidenceSanitizationError): + sanitize_sku_panel_evidence(raw, target) + + self.assertEqual((target / "sentinel.txt").read_text(encoding="utf-8"), "keep") + self.assertEqual(list(raw.parent.glob(".derived.staging-*")), []) + + def test_existing_derived_is_preserved_without_reading_or_writing_raw(self) -> None: + with TemporaryDirectory() as temporary: + raw = _write_raw(Path(temporary)) + target = raw.parent / "derived" + target.mkdir() + sentinel = target / "sentinel.txt" + sentinel.write_text("keep", encoding="utf-8") + + with self.assertRaises(SkuEvidenceSanitizationError): + sanitize_sku_panel_evidence(raw, target) + + self.assertEqual(sentinel.read_text(encoding="utf-8"), "keep") + self.assertEqual(list(raw.parent.glob(".derived.staging-*")), []) + + def test_directory_contract_rejects_non_sibling_paths(self) -> None: + with TemporaryDirectory() as temporary: + root = Path(temporary) + raw = _write_raw(root) + with self.assertRaises(SkuEvidenceSanitizationError): + sanitize_sku_panel_evidence(raw, root / "not-derived") + with self.assertRaises(SkuEvidenceSanitizationError): + sanitize_sku_panel_evidence(root / "not-raw", root / "derived")