chengma/cmbuyer
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(client): sanitize T-103 device evidence

Browse Source
This commit is contained in:
QiuSW
2026-08-04 11:16:11 +08:00
parent 17e295e99d
commit 9b2eb74478
3 changed files with 808 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
client/tests/device/test_sku_evidence_sanitizer.py
+341
View File
@@ -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 (
"<hierarchy rotation='0'>"
f"<node bounds='[0,0][1080,540]' text='{TEST_ADDRESS}' content-desc='{MASKED_PHONE} {FULL_PHONE}' />"
f"<node bounds='[0,540][1080,2400]' text='{SAFE_TEXT}' />"
"</hierarchy>"
)
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 = (
"<hierarchy>"
f"<node bounds='[0,0][1080,2400]' text='{TEST_ADDRESS}' content-desc='{MASKED_PHONE}'>"
f"<node bounds='[0,0][1080,540]' text='{FULL_PHONE}' />"
f"<node bounds='[0,540][1080,2400]' text='{SAFE_TEXT}' />"
"</node></hierarchy>"
)
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", "<hierarchy><node text='missing bounds' /></hierarchy>"),
(
"private-parent-with-lower-child",
"<hierarchy><node bounds='[0,0][1080,540]'><node bounds='[0,540][1080,2400]' text='x' /></node></hierarchy>",
),
("full-phone-below", f"<hierarchy><node bounds='[0,540][1080,2400]' text='{FULL_PHONE}' /></hierarchy>"),
("masked-phone-below", f"<hierarchy><node bounds='[0,540][1080,2400]' text='{MASKED_PHONE}' /></hierarchy>"),
)
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("<hierarchy>", 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 = (
"<hierarchy>"
"<node bounds='[0,0][1080,540]' text='private' />"
f"<node bounds='[0,540][1080,2400]' text='{value}' />"
"</hierarchy>"
)
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 = (
"<hierarchy>"
"<node bounds='[0,0][1080,540]' text='private' />"
"<node bounds='[0,540][1080,1000]' text='138' content-desc='0013' />"
"<node bounds='[0,1000][1080,2400]' text='8000' />"
"</hierarchy>"
)
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"<hierarchy><node bounds='[0,540][1080,2400]' text='{SAFE_TEXT}' /></hierarchy>"),
("no-below", "<hierarchy><node bounds='[0,0][1080,540]' text='private' /></hierarchy>"),
)
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")