diff --git a/docs/module-internals.md b/docs/module-internals.md index 6f762e3..029751b 100644 --- a/docs/module-internals.md +++ b/docs/module-internals.md @@ -650,6 +650,24 @@ Key contracts: [`_internal/isobmff.py`](../src/remove_ai_watermarks/_internal/isobmff.py). - Native MP4/MOV TC260 `AIGC` entries are read from `moov.udta.meta.keys/ilst` and blanked without changing box sizes. +- TC260 `SecurityData` signatures are checked by + [`_internal/tc260_signature.py`](../src/remove_ai_watermarks/_internal/tc260_signature.py) + on top of a verification-only SM3/SM2 in + [`_internal/sm2.py`](../src/remove_ai_watermarks/_internal/sm2.py), because no + dependency provides the SM2 curve. The label message is the guide's appendix + A.1 string (`"Label":"1","ContentProducer":"...","ProduceID":"..."`) under the + default SM2 user ID, and MiniMax's `LabelMataData` spelling is accepted beside + the guide's `Md`. Only the label signature is checked: MiniMax's binding + signature also verifies, but no verdict reads it, and its content hash is not + recomputed because MiniMax names no `CntSel`. The key travels in the file, so + only a key pinned in `TC260_SIGNING_KEYS` names a signer, and a pinned signer + is checked with the pinned point whatever parity the file claims; an unpinned + key that verifies proves integrity only and adds no signal. `ReservedCode2`, + the propagator's slot, is not read because no sample fills it. Verifications + are memoized, since `get_ai_metadata` and the verdict check the same label. + `tests/test_sm2.py` pins the GB/T 32905 SM3 vectors, OpenSSL's SM2 verify + vectors, and the real MiniMax signatures. Over 411 local images and videos the + only verdict-record change was the added marker on the nine MiniMax videos. - Native MP4/MOV `workflow` and `prompt` entries use that keyed metadata-list reader and the same offset-preserving removal path. - Native MKV/WebM TC260 `AIGC` entries are read from diff --git a/docs/research-sweep-2026-09.md b/docs/research-sweep-2026-09.md index 8762d3e..cbc3871 100644 --- a/docs/research-sweep-2026-09.md +++ b/docs/research-sweep-2026-09.md @@ -53,8 +53,7 @@ This page is an index. Each finding lives in its canonical page. guide (TC260-PG-20258A) covers documents this image and video tool does not process. The security-protection guide (TC260-PG-202511A) records a `SecurityData` JSON object, a digital signature over the label and optionally - the content, in `ReservedCode1` and `ReservedCode2`. The library accepts those - fields as label evidence but does not verify the signature. MiniMax is the one + the content, in `ReservedCode1` and `ReservedCode2`. MiniMax is the one producer in the local corpus that writes a guide-shaped object (checked 2026-09-29): four MiniMax videos served by Higgsfield and Runway, two of them tracked (`higgsfield-hailuo-2-3.mp4`, `higgsfield-minimax-h3.mp4`), carry @@ -72,8 +71,8 @@ This page is an index. Each finding lives in its canonical page. fields with opaque vendor strings (Qwen a base64 digest, Vidu 32 hex digits, Wan and HappyHorse `L-`/`K-` prefixed values), not `SecurityData`. The key sits in the file it signs, so a check proves authenticity only against a pinned - MiniMax key; verifying in `identify` needs SM2/SM3, which no current dependency - provides. + MiniMax key. `identify` now verifies it with a built-in SM3/SM2 and a pinned + MiniMax key ([module internals](module-internals.md)). - **Experiments.** Chroma restoration after regeneration (arXiv:2508.21072) was measured on 2026-09-27 and is not shipped: transplanted source chroma drew one explicit "inconclusive" and two unclear Gemini answers where the untouched diff --git a/docs/supported-signals.md b/docs/supported-signals.md index b88a692..3fd261d 100644 --- a/docs/supported-signals.md +++ b/docs/supported-signals.md @@ -102,6 +102,11 @@ The inspection and stripping code handles signals in these groups: `script.onMetaData.AIGC` placements, plus the audio placements of TC260-PG-202510A: a RIFF `AIGC` chunk in WAV, an ID3v2 `TXXX` frame described `AIGC` in MP3, and an `AIGC=` Vorbis comment in OGG, Opus and FLAC; +- TC260-PG-202511A `SecurityData` label signatures (SM3withSM2) in a label's + `ReservedCode1`: a signature that verifies against a pinned producer key is a + high-confidence `aigc_signature` signal (MiniMax is the one pinned signer), a + signature that does not verify is a caveat, and every result appears as the + `aigc_signature` metadata marker; - xAI and Grok signature-plus-UUID pairs in ordinary EXIF, ImageMagick PNG raw EXIF profiles, XMP description/creator fields, PNG Description/Author text, and IPTC Caption-Abstract/By-line fields; diff --git a/src/remove_ai_watermarks/_internal/sm2.py b/src/remove_ai_watermarks/_internal/sm2.py new file mode 100644 index 0000000..b85c195 --- /dev/null +++ b/src/remove_ai_watermarks/_internal/sm2.py @@ -0,0 +1,137 @@ +"""SM3 hashing and SM2 signature verification (GB/T 32905, GB/T 32918). + +Verification only: no key generation, signing, or secret material, so none of the +constant-time concerns of a signing implementation apply. The TC260 security +guide (TC260-PG-202511A) names SM3withSM2 for AI-label signatures, and no +dependency of this package provides SM2 (``cryptography`` has SM3 but not the +curve). Correctness is pinned by the standard's SM3 vectors, OpenSSL's SM2 verify +vectors, and real MiniMax label signatures in ``tests/test_sm2.py``. +""" + +from __future__ import annotations + +import struct + +Point = tuple[int, int] + +# The SM2 recommended 256-bit curve (GB/T 32918.5). +P = 0xFFFFFFFEFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF00000000FFFFFFFFFFFFFFFF +A = 0xFFFFFFFEFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF00000000FFFFFFFFFFFFFFFC +B = 0x28E9FA9E9D9F5E344D5A9E4BCF6509A7F39789F515AB8F92DDBCBD414D940E93 +N = 0xFFFFFFFEFFFFFFFFFFFFFFFFFFFFFFFF7203DF6B21C6052B53BBF40939D54123 +G: Point = ( + 0x32C4AE2C1F1981195F9904466A39C9948FE30BBFF2660BE1715A4589334C74C7, + 0xBC3736A2F4F6779C59BDCEE36B692153D0A9877CC62A474002DF32E52139F0A0, +) + +# GB/T 32918.2's default signer identity, used when a signature names none. +DEFAULT_USER_ID = b"1234567812345678" + +_SM3_IV = (0x7380166F, 0x4914B2B9, 0x172442D7, 0xDA8A0600, 0xA96F30BC, 0x163138AA, 0xE38DEE4D, 0xB0FB0E4E) +_MASK = 0xFFFFFFFF + + +def _rotl(x: int, n: int) -> int: + n %= 32 + return ((x << n) | (x >> (32 - n))) & _MASK + + +def _p0(x: int) -> int: + return x ^ _rotl(x, 9) ^ _rotl(x, 17) + + +def _p1(x: int) -> int: + return x ^ _rotl(x, 15) ^ _rotl(x, 23) + + +def sm3(data: bytes) -> bytes: + """SM3 digest of ``data`` (32 bytes).""" + bit_length = len(data) * 8 + padded = data + b"\x80" + b"\x00" * ((55 - len(data)) % 64) + bit_length.to_bytes(8, "big") + v = list(_SM3_IV) + for offset in range(0, len(padded), 64): + w = list(struct.unpack(">16I", padded[offset : offset + 64])) + for j in range(16, 68): + w.append(_p1(w[j - 16] ^ w[j - 9] ^ _rotl(w[j - 3], 15)) ^ _rotl(w[j - 13], 7) ^ w[j - 6]) + w1 = [w[j] ^ w[j + 4] for j in range(64)] + a, b, c, d, e, f, g, h = v + for j in range(64): + t = 0x79CC4519 if j < 16 else 0x7A879D8A + ss1 = _rotl((_rotl(a, 12) + e + _rotl(t, j)) & _MASK, 7) + ss2 = ss1 ^ _rotl(a, 12) + if j < 16: + ff, gg = a ^ b ^ c, e ^ f ^ g + else: + ff, gg = (a & b) | (a & c) | (b & c), (e & f) | (~e & g) + tt1 = (ff + d + ss2 + w1[j]) & _MASK + tt2 = (gg + h + ss1 + w[j]) & _MASK + a, b, c, d, e, f, g, h = tt1, a, _rotl(b, 9), c, _p0(tt2), e, _rotl(f, 19), g + v = [x ^ y for x, y in zip(v, (a, b, c, d, e, f, g, h), strict=True)] + return struct.pack(">8I", *v) + + +def _add(p: Point | None, q: Point | None) -> Point | None: + if p is None: + return q + if q is None: + return p + if p[0] == q[0] and (p[1] + q[1]) % P == 0: + return None + numerator, denominator = (3 * p[0] * p[0] + A, 2 * p[1]) if p == q else (q[1] - p[1], q[0] - p[0]) + slope = numerator * pow(denominator, -1, P) % P + x = (slope * slope - p[0] - q[0]) % P + return x, (slope * (p[0] - x) - p[1]) % P + + +def _multiply(k: int, point: Point) -> Point | None: + result: Point | None = None + addend: Point | None = point + while k: + if k & 1: + result = _add(result, addend) + addend = _add(addend, addend) + k >>= 1 + return result + + +def on_curve(point: Point) -> bool: + x, y = point + return 0 <= x < P and 0 <= y < P and (y * y - (x * x * x + A * x + B)) % P == 0 + + +def lift_x(x: int, *, odd: bool) -> Point | None: + """The curve point with this x coordinate and the requested y parity, if any.""" + if not 0 <= x < P: + return None + y_squared = (x * x * x + A * x + B) % P + y = pow(y_squared, (P + 1) // 4, P) # P % 4 == 3 + if y * y % P != y_squared: + return None + return (x, y) if (y & 1) == odd else (x, P - y) + + +def signer_digest(public_key: Point, message: bytes, user_id: bytes = DEFAULT_USER_ID) -> bytes: + """``e = SM3(Z_A || M)``, the value an SM3withSM2 signature signs.""" + identity_bits = (len(user_id) * 8).to_bytes(2, "big") + fields = (A, B, G[0], G[1], public_key[0], public_key[1]) + z = sm3(identity_bits + user_id + b"".join(value.to_bytes(32, "big") for value in fields)) + return sm3(z + message) + + +def verify_digest(public_key: Point, digest: bytes, r: int, s: int) -> bool: + """Verify an SM2 signature ``(r, s)`` over an already computed digest ``e``.""" + if not (1 <= r < N and 1 <= s < N) or not on_curve(public_key): + return False + t = (r + s) % N + if t == 0: + return False + point = _add(_multiply(s, G), _multiply(t, public_key)) + return point is not None and (int.from_bytes(digest, "big") + point[0]) % N == r + + +def verify(public_key: Point, message: bytes, signature: bytes, user_id: bytes = DEFAULT_USER_ID) -> bool: + """Verify a raw 64-byte ``r || s`` SM3withSM2 signature over ``message``.""" + if len(signature) != 64: + return False + r, s = int.from_bytes(signature[:32], "big"), int.from_bytes(signature[32:], "big") + return verify_digest(public_key, signer_digest(public_key, message, user_id), r, s) diff --git a/src/remove_ai_watermarks/_internal/tc260_signature.py b/src/remove_ai_watermarks/_internal/tc260_signature.py new file mode 100644 index 0000000..0569269 --- /dev/null +++ b/src/remove_ai_watermarks/_internal/tc260_signature.py @@ -0,0 +1,157 @@ +"""Verify TC260 ``SecurityData`` label signatures (TC260-PG-202511A). + +The guide lets a label producer record a ``SecurityData`` JSON object in the +label's ``ReservedCode1``: SM3withSM2 signatures (``PubSD`` entries of type +``DS``), the public key (``PubKey``), and content bindings (``Bindings``). Only +the producer side is read here, because that is the only shape measured: MiniMax +writes it on every video in the local corpus, and ``ReservedCode2`` (the +propagator's slot) was empty on all of them. Other producers fill the reserved +fields with opaque vendor strings that are not ``SecurityData``. + +Only the label signature is checked. MiniMax spells its type ``LabelMataData`` +where the guide says ``Md``; both are accepted. MiniMax's second signature, over +its ``Bindings`` entry, also verifies (docs/research-sweep-2026-09.md), but no +verdict reads it, and the content hash inside that entry is not recomputed +because MiniMax names no content-selection method. The key travels inside the +file it signs, so a signature proves only integrity unless the key matches a +pinned signer in ``TC260_SIGNING_KEYS``. +""" + +from __future__ import annotations + +import functools +import json +from dataclasses import dataclass +from typing import TYPE_CHECKING, Any, Literal, cast + +from remove_ai_watermarks._internal import sm2 + +if TYPE_CHECKING: + from collections.abc import Mapping + +Status = Literal["verified", "failed", "absent"] + +SM3_WITH_SM2 = "1.2.156.10197.1.501" + + +def _pinned(x: int) -> sm2.Point: + point = sm2.lift_x(x, odd=False) + if point is None: + raise ValueError(f"pinned TC260 key x={x:#x} is not on the SM2 curve") + return point + + +# Label-signing keys pinned to the organization that holds them: the even-y point +# over the key's x coordinate. MiniMax's is the ``KeyValue`` carried by all four +# MiniMax videos in the local corpus (Hailuo 2.3, H3 and H3 Max, served by +# Higgsfield and Runway, 2026-09-29), with a leading ``00``. +TC260_SIGNING_KEYS: dict[str, sm2.Point] = { + "MiniMax": _pinned(0xA0B3B0B6A0C9B0C89CAB328342AF4E8221EC5B40799CBE835AB4251F7B47E4FD), +} + +_LABEL_TBS_TYPES = frozenset({"Md", "LabelMataData"}) +# The producer's signed fields, in GB 45438 appendix E order ("Label" to value3). +_PRODUCER_FIELDS = ("Label", "ContentProducer", "ProduceID") + + +@dataclass(frozen=True) +class Tc260Signature: + """Outcome of checking one label's ``SecurityData`` label signature.""" + + label: Status + signer: str | None + + def describe(self) -> str: + if self.label == "verified": + if self.signer: + return f"TC260 label signature verified (pinned {self.signer} key)" + return "TC260 label signature verified with its own embedded key (no pinned signer)" + if self.label == "failed": + return "TC260 label signature does not verify" + return "TC260 SecurityData without a label signature" + + +def _candidate_keys(key_hex: str) -> tuple[list[sm2.Point], str | None]: + """The points a ``KeyValue`` can denote, and the pinned signer when it matches one.""" + try: + raw = bytes.fromhex(key_hex) + except ValueError: + return [], None + if len(raw) == 65 and raw[0] == 4: + raw = raw[1:] + if len(raw) == 33 and raw[0] == 0: + raw = raw[1:] # MiniMax pads a bare x coordinate with 00 + if len(raw) == 64: + points = [(int.from_bytes(raw[:32], "big"), int.from_bytes(raw[32:], "big"))] + elif len(raw) == 33 and raw[0] in (2, 3): + points = [p for p in [sm2.lift_x(int.from_bytes(raw[1:], "big"), odd=raw[0] == 3)] if p] + elif len(raw) == 32: + x = int.from_bytes(raw, "big") # a bare x coordinate: either y may be meant + points = [p for p in (sm2.lift_x(x, odd=False), sm2.lift_x(x, odd=True)) if p] + else: + return [], None + for signer, pinned in TC260_SIGNING_KEYS.items(): + if any(p[0] == pinned[0] for p in points): + return [pinned], signer + return points, None + + +def _label_message(label: Mapping[str, str]) -> bytes: + """Appendix A.1: ``"Label":"1","ContentProducer":"...","ProduceID":"..."``.""" + return ",".join( + f"{json.dumps(field)}:{json.dumps(label.get(field, ''), ensure_ascii=False)}" for field in _PRODUCER_FIELDS + ).encode() + + +def _object(value: object) -> dict[str, Any]: + """``value`` as a JSON object, or empty when it is anything else.""" + return cast("dict[str, Any]", value) if isinstance(value, dict) else {} + + +def _objects(value: object) -> list[dict[str, Any]]: + """The JSON objects in a JSON array, or none when ``value`` is not an array.""" + items = cast("list[object]", value) if isinstance(value, list) else [] + return [cast("dict[str, Any]", item) for item in items if isinstance(item, dict)] + + +@functools.lru_cache(maxsize=256) +def _verified(point: sm2.Point, message: bytes, signature: bytes) -> bool: + # get_ai_metadata and the identify verdict both check the same label, and + # removal re-reads metadata; the pure-Python curve math runs once per triple. + return sm2.verify(point, message, signature) + + +def check_tc260_signature(label: Mapping[str, str]) -> Tc260Signature | None: + """Check the producer's ``SecurityData`` in ``ReservedCode1``; None when it has none.""" + try: + parsed: object = json.loads(label.get("ReservedCode1", "")) + except ValueError: + return None + security = _object(_object(parsed).get("SecurityData")) + if security.get("Type") != "TC260PG": + return None + public = _objects(security.get("PubSD")) + keys = { + str(entry.get("KeyID", 0)): _candidate_keys(str(entry.get("KeyValue", ""))) + for entry in reversed(public) # reversed: the first entry per KeyID wins + if entry.get("Type") == "PubKey" and entry.get("AlgID") == SM3_WITH_SM2 + } + message = _label_message(label) + results: list[bool] = [] + signer: str | None = None + for entry in public: + if entry.get("Type") != "DS" or entry.get("AlgID") != SM3_WITH_SM2: + continue + if _object(entry.get("TBSData")).get("Type") not in _LABEL_TBS_TYPES: + continue + points, key_signer = keys.get(str(entry.get("KeyID", 0)), ([], None)) + try: + signature = bytes.fromhex(str(entry.get("Signature", ""))) + except ValueError: + signature = b"" + verified = any(_verified(point, message, signature) for point in points) + results.append(verified) + if verified: + signer = signer or key_signer + status: Status = "absent" if not results else ("verified" if any(results) else "failed") + return Tc260Signature(label=status, signer=signer) diff --git a/src/remove_ai_watermarks/identify.py b/src/remove_ai_watermarks/identify.py index 6a5b695..0512b7f 100644 --- a/src/remove_ai_watermarks/identify.py +++ b/src/remove_ai_watermarks/identify.py @@ -49,6 +49,7 @@ from remove_ai_watermarks._internal.constants import ( C2PA_SIGNER_PLATFORMS, ) from remove_ai_watermarks._internal.schema import require_schema_version +from remove_ai_watermarks._internal.tc260_signature import check_tc260_signature from remove_ai_watermarks.metadata import ( AI_METADATA_KEYS, AIGC_MARKERS, @@ -159,6 +160,10 @@ _HF_JOB_CAVEAT = ( "medium-confidence signal. Higgsfield adds the tag after signing, which invalidates any " "C2PA manifest the upstream model embedded." ) +_TC260_SIGNATURE_FAILED_CAVEAT = ( + "The TC260 label carries a SecurityData signature that does not verify against its own " + "fields, so the label was altered after signing or copied from another file." +) _C2PA_CLOUD_CAVEAT = ( "The embedded C2PA manifest is absent but an XMP provenance pointer to the " "vendor's cloud manifest store survives, so the Content Credentials remain " @@ -1486,6 +1491,11 @@ def _identify_from_evidence( platform = "China AIGC-labeled content (TC260 standard)" if manufacturer := _tc260_manufacturer_of(producer): ai_vendor_claims["aigc"] = manufacturer + if aigc_data and (signature := check_tc260_signature(aigc_data)): + if signature.label == "verified" and signature.signer: + signals.append(Signal("aigc_signature", signature.describe(), "high")) + elif signature.label == "failed": + caveats.append(_TC260_SIGNATURE_FAILED_CAVEAT) # ── Local diffusion parameters (Stable Diffusion / ComfyUI) ────── local_keys = sorted(k for k in meta if k.lower() in _LOCAL_GEN_KEYS) diff --git a/src/remove_ai_watermarks/metadata.py b/src/remove_ai_watermarks/metadata.py index 0764fd5..9451050 100644 --- a/src/remove_ai_watermarks/metadata.py +++ b/src/remove_ai_watermarks/metadata.py @@ -1424,6 +1424,10 @@ def get_ai_metadata(image_path: Path) -> dict[str, str]: # sentence above, whose wording can change without notice. if producer: result["aigc_producer"] = producer + from remove_ai_watermarks._internal.tc260_signature import check_tc260_signature + + if signature := check_tc260_signature(aigc): + result["aigc_signature"] = signature.describe() app_scan = scan_head(image_path) app_provenance, app_generator = _app_metadata_evidence(app_scan) diff --git a/tests/test_sm2.py b/tests/test_sm2.py new file mode 100644 index 0000000..0fa6a20 --- /dev/null +++ b/tests/test_sm2.py @@ -0,0 +1,233 @@ +"""SM3/SM2 verification and TC260 SecurityData label signatures.""" + +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from remove_ai_watermarks._internal import sm2, tc260_signature +from remove_ai_watermarks._internal.tc260_signature import check_tc260_signature +from remove_ai_watermarks.metadata import aigc_label + +ROOT = Path(__file__).resolve().parents[1] +MINIMAX_VIDEOS = ( + ROOT / "data" / "fixtures" / "provenance" / "higgsfield-hailuo-2-3.mp4", + ROOT / "data" / "fixtures" / "provenance" / "higgsfield-minimax-h3.mp4", +) + +# OpenSSL test/recipes/30-test_evp_data/evppkey_sm2.txt, key SM2_key1: the public +# point of its PKCS#8 key and two "Verify" rows (digest e, DER signature r, s). +OPENSSL_KEY = ( + 0x393A2AD9E27E43ACAAAB85A19D3B85591E14406157175ADDD37BF77FF0CAF9ED, + 0x37FCF3A8B4F55C038523261FCA6636AFE6A7AFA2A1482A606F5CB3A18B2E2A1E, +) +OPENSSL_VERIFY = ( + ( + "D7AD397F6FFA5D4F7F11E7217F241607DC30618C236D2C09C1B9EA8FDADEE2E8", + 0xAB1DB64DE7C40EDBDE6651C9B8EBDB804673DB836E5D5C7FE15DCF9ED2725037, + 0xEBA714451FF69B0BB930B379E192E7CD5FA6E3C41C7FBD8303B799AB54A54621, + ), + ( + "B1139602C6ECC9E15E2F3F9C635A1AFE737058BC15387479C1EA0D0B3D90E9E5", + 0xE6E0414EBD3A656C35602AF14AB20287DBF30D57AF75C49A188ED4B42391F224, + 0x2F54F277C606F4605E1CE9514947FFDDF94C67A539804A4ED17F852288BDBE2E, + ), +) + + +def _minimax_label() -> dict[str, str]: + label = aigc_label(MINIMAX_VIDEOS[0]) + assert label is not None + return label + + +def _with_security(label: dict[str, str], edit: object) -> dict[str, str]: + security = json.loads(label["ReservedCode1"]) + edit(security["SecurityData"]) # type: ignore[operator] + return {**label, "ReservedCode1": json.dumps(security, separators=(",", ":"))} + + +class TestSm3: + @pytest.mark.parametrize( + ("message", "digest"), + [ + (b"abc", "66c7f0f462eeedd9d1f2d46bdc10e4e24167c4875cf2f7a2297da02b8f4ba8e0"), + (b"abcd" * 16, "debe9ff92275b8a138604889c18e5a4d6fdb70e5387e5765293dcba39c0c5732"), + ], + ) + def test_standard_vectors(self, message: bytes, digest: str) -> None: + # GB/T 32905-2016 appendix A examples 1 and 2. + assert sm2.sm3(message).hex() == digest + + +class TestSm2: + @pytest.mark.parametrize(("digest", "r", "s"), OPENSSL_VERIFY) + def test_openssl_vectors_verify(self, digest: str, r: int, s: int) -> None: + assert sm2.verify_digest(OPENSSL_KEY, bytes.fromhex(digest), r, s) + + @pytest.mark.parametrize(("digest", "r", "s"), OPENSSL_VERIFY) + def test_a_changed_digest_or_signature_fails(self, digest: str, r: int, s: int) -> None: + changed = bytes.fromhex(digest)[:-1] + b"\x00" + assert not sm2.verify_digest(OPENSSL_KEY, changed, r, s) + assert not sm2.verify_digest(OPENSSL_KEY, bytes.fromhex(digest), r, s ^ 1) + + def test_lift_x_returns_the_requested_parity(self) -> None: + for odd in (False, True): + point = sm2.lift_x(OPENSSL_KEY[0], odd=odd) + assert point is not None + assert sm2.on_curve(point) + assert (point[1] & 1) == odd + assert OPENSSL_KEY in {sm2.lift_x(OPENSSL_KEY[0], odd=False), sm2.lift_x(OPENSSL_KEY[0], odd=True)} + + def test_an_off_curve_key_is_rejected(self) -> None: + digest, r, s = OPENSSL_VERIFY[0] + assert not sm2.verify_digest((OPENSSL_KEY[0], OPENSSL_KEY[1] + 1), bytes.fromhex(digest), r, s) + # Congruent to the real key mod P, so the point arithmetic alone would accept it. + assert not sm2.verify_digest((OPENSSL_KEY[0] + sm2.P, OPENSSL_KEY[1]), bytes.fromhex(digest), r, s) + + +class TestTc260Signature: + @pytest.mark.parametrize("path", MINIMAX_VIDEOS, ids=lambda p: p.name) + def test_real_minimax_labels_verify_against_the_pinned_key(self, path: Path) -> None: + label = aigc_label(path) + assert label is not None + result = check_tc260_signature(label) + assert result == tc260_signature.Tc260Signature(label="verified", signer="MiniMax") + assert result.describe() == "TC260 label signature verified (pinned MiniMax key)" + + def test_a_changed_produce_id_fails_the_label_signature(self) -> None: + label = _minimax_label() + produce_id = label["ProduceID"] + altered = {**label, "ProduceID": produce_id[:-1] + ("0" if produce_id[-1] != "0" else "1")} + result = check_tc260_signature(altered) + assert result is not None + assert (result.label, result.signer) == ("failed", None) + + def test_a_repeated_check_verifies_once(self, monkeypatch: pytest.MonkeyPatch) -> None: + calls: list[bytes] = [] + real_verify = sm2.verify + + def counting(point: sm2.Point, message: bytes, signature: bytes) -> bool: + calls.append(message) + return real_verify(point, message, signature) + + tc260_signature._verified.cache_clear() + monkeypatch.setattr(sm2, "verify", counting) + label = _minimax_label() + first, second = check_tc260_signature(label), check_tc260_signature(dict(label)) + assert first == second + assert first is not None + assert first.label == "verified" + assert len(calls) == 1 + + def test_an_unpinned_key_proves_integrity_only(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(tc260_signature, "TC260_SIGNING_KEYS", {}) + result = check_tc260_signature(_minimax_label()) + assert result is not None + assert (result.label, result.signer) == ("verified", None) + assert "no pinned signer" in result.describe() + + def test_a_pinned_signer_is_checked_with_the_pinned_point(self) -> None: + # The file's own parity claim is untrusted: an odd-y spelling of MiniMax's + # x coordinate still verifies, because the pinned even-y point is used. + def edit(security: dict[str, object]) -> None: + for entry in security["PubSD"]: # type: ignore[union-attr] + if entry["Type"] == "PubKey": + entry["KeyValue"] = "03" + entry["KeyValue"][-64:] + + result = check_tc260_signature(_with_security(_minimax_label(), edit)) + assert result is not None + assert (result.label, result.signer) == ("verified", "MiniMax") + + def test_the_first_key_with_an_id_wins(self) -> None: + def edit(security: dict[str, object]) -> None: + foreign = { + "Type": "PubKey", + "AlgID": tc260_signature.SM3_WITH_SM2, + "KeyValue": "04" + f"{OPENSSL_KEY[0]:064x}{OPENSSL_KEY[1]:064x}", + } + security["PubSD"].append(foreign) # type: ignore[union-attr] + + result = check_tc260_signature(_with_security(_minimax_label(), edit)) + assert result is not None + assert (result.label, result.signer) == ("verified", "MiniMax") + + def test_a_foreign_key_does_not_verify_the_label(self) -> None: + def edit(security: dict[str, object]) -> None: + for entry in security["PubSD"]: # type: ignore[union-attr] + if entry["Type"] == "PubKey": + entry["KeyValue"] = "04" + f"{OPENSSL_KEY[0]:064x}{OPENSSL_KEY[1]:064x}" + + result = check_tc260_signature(_with_security(_minimax_label(), edit)) + assert result is not None + assert (result.label, result.signer) == ("failed", None) + + @pytest.mark.parametrize( + "reserved", + [ + "", + "S7Wr/Evf+Mq+T3SWymUoO7naAs0WPNLrahOqG9WzR+o=", + '{"SecurityData":{"Type":"OTHER","Version":1}}', + "[1, 2]", + ], + ) + def test_reserved_codes_without_security_data_are_not_signatures(self, reserved: str) -> None: + assert check_tc260_signature({"Label": "1", "ReservedCode1": reserved}) is None + + +def _labeled_png(path: Path, label: dict[str, str]) -> Path: + from PIL import Image + from PIL.PngImagePlugin import PngInfo + + info = PngInfo() + info.add_text("AIGC", json.dumps(label, separators=(",", ":"))) + Image.new("RGB", (64, 64), (90, 120, 150)).save(path, pnginfo=info) + return path + + +class TestIdentifyIntegration: + def test_a_verified_pinned_signature_is_a_high_confidence_signal(self, tmp_path: Path) -> None: + from remove_ai_watermarks.identify import identify + + report = identify( + _labeled_png(tmp_path / "signed.png", _minimax_label()), check_visible=False, check_invisible=False + ) + signatures = [signal for signal in report.signals if signal.name == "aigc_signature"] + assert [(signal.detail, signal.confidence) for signal in signatures] == [ + ("TC260 label signature verified (pinned MiniMax key)", "high") + ] + assert not any("does not verify" in caveat for caveat in report.caveats) + + def test_a_failed_signature_is_a_caveat_not_a_signal(self, tmp_path: Path) -> None: + from remove_ai_watermarks.identify import identify + + label = _minimax_label() + altered = {**label, "ProduceID": "0" * len(label["ProduceID"])} + report = identify(_labeled_png(tmp_path / "altered.png", altered), check_visible=False, check_invisible=False) + assert not any(signal.name == "aigc_signature" for signal in report.signals) + assert any("does not verify" in caveat for caveat in report.caveats) + + def test_the_metadata_record_path_matches_the_file_path(self, tmp_path: Path) -> None: + from remove_ai_watermarks.identify import identify, identify_metadata_record + from remove_ai_watermarks.metadata import get_ai_metadata + from remove_ai_watermarks.metadata_record import collect_metadata_record + + path = _labeled_png(tmp_path / "signed.png", _minimax_label()) + record = json.loads(json.dumps(collect_metadata_record(path))) + from_record = identify_metadata_record(record, path=path) + from_file = identify(path, check_visible=False, check_invisible=False) + assert from_record.signals == from_file.signals + assert from_record.caveats == from_file.caveats + assert get_ai_metadata(path)["aigc_signature"] == "TC260 label signature verified (pinned MiniMax key)" + + def test_a_minimax_video_carries_the_signature_marker(self) -> None: + from remove_ai_watermarks.video import identify_video + + report = identify_video(MINIMAX_VIDEOS[0], check_visible=False) + assert report.metadata_markers["aigc_signature"] == "TC260 label signature verified (pinned MiniMax key)" + + +if __name__ == "__main__": + raise SystemExit(pytest.main([__file__]))