423 lines
20 KiB
Python
423 lines
20 KiB
Python
"""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_SCREENSHOT_HEIGHT,
|
|
EXPECTED_SCREENSHOT_WIDTH,
|
|
EXPECTED_XML_HEIGHT,
|
|
EXPECTED_XML_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,2376]' text='{SAFE_TEXT}' />"
|
|
"</hierarchy>"
|
|
)
|
|
|
|
|
|
def _write_raw(
|
|
root: Path,
|
|
*,
|
|
state: str = "panel-opened-target-preselected",
|
|
goods_id: str = EXPECTED_GOODS_ID,
|
|
model: str = "PKG110",
|
|
android_version: str = "16",
|
|
pdd_version: str = "8.17.0",
|
|
size: tuple[int, int] = (EXPECTED_SCREENSHOT_WIDTH, EXPECTED_SCREENSHOT_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_SCREENSHOT_WIDTH, EXPECTED_SCREENSHOT_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-v3"', manifest)
|
|
self.assertIn('"screenshot_space": {', manifest)
|
|
self.assertIn('"xml_coordinate_space": {', manifest)
|
|
self.assertIn('"height": 2376', manifest)
|
|
self.assertIn('"height": 2376', manifest)
|
|
self.assertIn('"privacy_mask_rectangle": [', manifest)
|
|
self.assertIn('"removed_nodes": 1', manifest)
|
|
self.assertIn('"cleared_crossing_nodes": 0', manifest)
|
|
self.assertIn('"retained_below_nodes": 1', manifest)
|
|
self.assertIn('"max_right": 1080', manifest)
|
|
self.assertIn('"max_bottom": 2376', 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,2376]' text='{TEST_ADDRESS}' content-desc='{MASKED_PHONE}'>"
|
|
f"<node bounds='[0,0][1080,540]' text='{FULL_PHONE}' />"
|
|
f"<node bounds='[0,540][1080,2376]' 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,2376]"})
|
|
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,
|
|
"max_right": 1080,
|
|
"max_bottom": 2376,
|
|
},
|
|
)
|
|
|
|
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_manifest_records_distinct_screenshot_and_xml_coordinate_spaces(self) -> None:
|
|
with TemporaryDirectory() as temporary:
|
|
raw = _write_raw(Path(temporary))
|
|
result = sanitize_sku_panel_evidence(raw, raw.parent / "derived")
|
|
manifest = json.loads(result.manifest_path.read_text(encoding="utf-8"))
|
|
|
|
self.assertEqual(
|
|
manifest["screenshot_space"],
|
|
{"width": 1080, "height": 2376, "privacy_mask_rectangle": [0, 0, 1080, 540]},
|
|
)
|
|
self.assertEqual(
|
|
manifest["xml_coordinate_space"],
|
|
{
|
|
"width": 1080,
|
|
"height": 2376,
|
|
"privacy_mask_rectangle": [0, 0, 1080, 540],
|
|
"observed_max": {"right": 1080, "bottom": 2376},
|
|
},
|
|
)
|
|
|
|
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),
|
|
("old-screenshot-space", {"size": (1080, 2400)}, None),
|
|
("other-screenshot-space", {"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_old_and_unknown_human_states_are_rejected(self) -> None:
|
|
for state in ("initial", "one-dimension-selected", "all-dimensions-selected", "guessed"):
|
|
with self.subTest(state=state), TemporaryDirectory() as temporary:
|
|
raw = _write_raw(Path(temporary), state=state)
|
|
with self.assertRaises(SkuEvidenceSanitizationError):
|
|
sanitize_sku_panel_evidence(raw, raw.parent / "derived")
|
|
self.assertFalse((raw.parent / "derived").exists())
|
|
|
|
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,2376]' text='x' /></node></hierarchy>",
|
|
),
|
|
("full-phone-below", f"<hierarchy><node bounds='[0,540][1080,2376]' text='{FULL_PHONE}' /></hierarchy>"),
|
|
("masked-phone-below", f"<hierarchy><node bounds='[0,540][1080,2376]' 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,2376]' 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,2376]' 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,2376]' 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_xml_coordinate_space_must_have_exact_configured_maximums(self) -> None:
|
|
scenarios = (
|
|
(
|
|
"short-width",
|
|
"<hierarchy><node bounds='[0,0][1079,540]' text='private' />"
|
|
"<node bounds='[0,540][1079,2376]' text='safe' /></hierarchy>",
|
|
"1079x2376",
|
|
),
|
|
(
|
|
"short-height",
|
|
"<hierarchy><node bounds='[0,0][1080,540]' text='private' />"
|
|
"<node bounds='[0,540][1080,2375]' text='safe' /></hierarchy>",
|
|
"1080x2375",
|
|
),
|
|
(
|
|
"wide-width",
|
|
"<hierarchy><node bounds='[0,0][1081,540]' text='private' />"
|
|
"<node bounds='[0,540][1081,2376]' text='safe' /></hierarchy>",
|
|
"1081x2376",
|
|
),
|
|
(
|
|
"old-v2-xml-height",
|
|
"<hierarchy><node bounds='[0,0][1080,540]' text='private' />"
|
|
"<node bounds='[0,540][1080,2400]' text='safe' /></hierarchy>",
|
|
"1080x2400",
|
|
),
|
|
)
|
|
for name, xml, observed in scenarios:
|
|
with self.subTest(name=name), TemporaryDirectory() as temporary:
|
|
raw = _write_raw(Path(temporary), xml=xml)
|
|
with self.assertRaises(SkuEvidenceSanitizationError) as raised:
|
|
sanitize_sku_panel_evidence(raw, raw.parent / "derived")
|
|
|
|
self.assertIn(observed, str(raised.exception))
|
|
self.assertNotIn(TEST_SERIAL, str(raised.exception))
|
|
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")
|