From 2eab24a2e1225bbcca95825adb04da2e91c34186 Mon Sep 17 00:00:00 2001 From: Victor Kuznetsov Date: Sat, 15 Aug 2026 11:31:59 -0700 Subject: [PATCH] Validate C2PA credentials before attribution --- docs/cli.md | 5 +- docs/module-internals.md | 24 ++ docs/python-api.md | 9 + docs/supported-signals.md | 6 +- docs/verification-plan.md | 9 + src/remove_ai_watermarks/_internal/c2pa.py | 316 +++++++++++++++++- .../_internal/constants.py | 4 + src/remove_ai_watermarks/api.py | 3 +- src/remove_ai_watermarks/cli.py | 7 + src/remove_ai_watermarks/identify.py | 180 ++++++++-- src/remove_ai_watermarks/metadata.py | 21 +- tests/conftest.py | 26 +- tests/test_api.py | 9 + tests/test_cli.py | 2 + tests/test_identify.py | 117 ++++++- tests/test_metadata.py | 4 + tests/test_metadata_internals.py | 123 +++++++ tests/test_metadata_record.py | 8 + 18 files changed, 826 insertions(+), 47 deletions(-) diff --git a/docs/cli.md b/docs/cli.md index 1d3ad1e..e296578 100644 --- a/docs/cli.md +++ b/docs/cli.md @@ -40,7 +40,10 @@ remove-ai-watermarks identify image.png `identify` always inspects supported metadata. When pixel extras are installed, it also evaluates supported visible and invisible pixel signals. When no signal is found, it reports the origin as unknown. It does not claim the image is -clean. +clean. For C2PA files, the text report shows asset integrity, claim-signature, +signer-trust, and signer-validity results separately. An intact claim from an +untrusted or expired signer is reported at medium confidence; a failed asset +binding or signature does not confirm the claimed origin. Machine readable output: diff --git a/docs/module-internals.md b/docs/module-internals.md index 89bce5a..c3a4061 100644 --- a/docs/module-internals.md +++ b/docs/module-internals.md @@ -364,6 +364,30 @@ Regression coverage: official `c2pa-python` reader first. Its byte-level PNG parser remains a fallback for partial and synthetic fixtures that the official reader rejects. +Structured extraction is limited to the active manifest and the ingredient +manifests reachable from it. Validation is preserved as separate dimensions: +asset binding integrity, claim signature, signer trust, and signer certificate +validity. A matching asset hash and valid claim signature do not make an +untrusted or expired signer trusted. `identify` therefore assigns high confidence +only when all four dimensions validate, medium confidence to an intact but +untrusted or unvalidated claim, and no origin verdict to a hash/signature failure. +The failed claim remains in the marker inventory and keeps the removal gate +fail-safe because a post-signing container edit can invalidate C2PA without +removing a declared pixel watermark. + +The structured walk also treats an exact known AI product in a reachable +`claim_generator` as an AI assertion. This covers update chains where the active +manifest names only `c2pa-tool` while a validated ingredient names Dreamina, and +Firefly chains that identify `Adobe_Firefly` without repeating a digital source +type. Unreachable manifests remain excluded. + +The SDK default enables trust verification but supplies no production trust +anchors. Consequently, an installation without an explicitly maintained C2PA +trust bundle reports otherwise valid signer chains as untrusted and keeps their +attribution at medium confidence. Shipping or fetching an official trust bundle +requires a separate update, provenance, and availability policy; do not silently +convert `signingCredential.untrusted` into trusted based on a vendor-name match. + Vendor attribution comes from the registry in [`_internal/constants.py`](../src/remove_ai_watermarks/_internal/constants.py). Derived issuer and platform maps should not be maintained separately. diff --git a/docs/python-api.md b/docs/python-api.md index 8839c7d..36efb46 100644 --- a/docs/python-api.md +++ b/docs/python-api.md @@ -155,8 +155,17 @@ from remove_ai_watermarks.identify import identify report = identify(Path("input.png")) print(report.platform) print(report.signals) +print(report.c2pa_validation) ``` +`c2pa_validation`, when present, reports `integrity`, `signature`, +`signer_trust`, and `signer_validity` independently, plus the reader status +codes. A valid hash and signature with an untrusted or expired signer is a +medium-confidence signed claim. A hash or signature failure does not confirm the +claimed platform or AI origin. Fallback parsing reports unknown validation +dimensions, while a raw marker in an unsupported or malformed container can +leave `c2pa_validation` as `None`. + Use `check_visible=False` and `check_invisible=False` for metadata-only inspection through the compatible path-based API: diff --git a/docs/supported-signals.md b/docs/supported-signals.md index dd0a85a..e77398a 100644 --- a/docs/supported-signals.md +++ b/docs/supported-signals.md @@ -86,7 +86,11 @@ The inspection and stripping code handles signals in these groups: `identify` combines detected signals into a `ProvenanceReport`. It reports unknown when evidence is absent. It never treats missing metadata as proof that -an image is human made. +an image is human made. C2PA presence alone is not a verified identity: the +report distinguishes asset binding, claim signature, signer trust, and signer +validity. High-confidence C2PA attribution requires all four; intact but +untrusted or fallback claims are medium-confidence, while a failed binding or +signature contributes no origin verdict. ## File and container formats diff --git a/docs/verification-plan.md b/docs/verification-plan.md index 99e19b5..f249757 100644 --- a/docs/verification-plan.md +++ b/docs/verification-plan.md @@ -39,6 +39,15 @@ confidence), reviewed once, then re-baselined. Lost detections are the alarm. Implemented as `scripts/sidecar_regression.py` (resumable, ~1.5 h at 8 workers). +A 2026-08-15 metadata-only C2PA regression audit over the local historical +corpus caught two structured-parser gaps before release: an exact Firefly claim +generator without a repeated source type, and a Dreamina generator carried by a +reachable ingredient under a generic update manifest. Both now have focused +tests. The same audit confirmed that invalid hash/signature claims lose origin +attribution without losing the C2PA inventory, and exposed the absence of +production trust anchors in the SDK defaults. Dataset-derived counts and +identifiers remain in the gitignored audit output. + #### Local run protocol Re-run `identify` against locally recorded sidecars, classify losses separately from intended new detections, and keep generated reports under `.local-eval/`. Do not commit dataset-derived counts or identifiers. diff --git a/src/remove_ai_watermarks/_internal/c2pa.py b/src/remove_ai_watermarks/_internal/c2pa.py index 3fe975c..0cb3918 100644 --- a/src/remove_ai_watermarks/_internal/c2pa.py +++ b/src/remove_ai_watermarks/_internal/c2pa.py @@ -8,6 +8,7 @@ import json import logging import re import struct +from collections import deque from dataclasses import dataclass from pathlib import Path from typing import TYPE_CHECKING, Any, cast @@ -37,6 +38,31 @@ with contextlib.suppress(Exception): _C2PA_READER_AVAILABLE = _C2paReader is not None _PNG_HEADER = struct.Struct(">I4s") +_CONTENT_BINDING_MATCHES = ( + "assertion.dataHash.match", + "assertion.bmffHash.match", + "assertion.boxesHash.match", + "assertion.collectionHash.match", + "assertion.multiAssetHash.match", +) +_CONTENT_BINDING_FAILURE_MARKERS = ( + "Hash.incorrectFileCount", + "Hash.malformed", + "Hash.mismatch", + "Hash.missingPart", + "assertion.hashedURI.mismatch", + "claim.hardBindings.missing", + "hashedUri.mismatch", + "ingredient.manifest.mismatch", +) +_SIGNATURE_FAILURES = frozenset( + { + "claimSignature.mismatch", + "claimSignature.missing", + "claimSignature.outsideValidity", + } +) + @dataclass(frozen=True) class _PngChunk: @@ -185,6 +211,159 @@ def _active_manifest(store: dict[str, Any]) -> dict[str, Any]: return cast("dict[str, Any]", active) if isinstance(active, dict) else {} +def _manifest_chain(store: dict[str, Any]) -> list[dict[str, Any]]: + """Return the active manifest followed by credential-usable reachable ingredients.""" + manifests = store.get("manifests") + active_label = store.get("active_manifest") + if not isinstance(manifests, dict) or not isinstance(active_label, str): + return [] + typed_manifests = cast("dict[object, object]", manifests) + pending = deque([active_label]) + seen: set[str] = set() + chain: list[dict[str, Any]] = [] + while pending: + label = pending.popleft() + if label in seen: + continue + seen.add(label) + manifest_value = typed_manifests.get(label) + if not isinstance(manifest_value, dict): + continue + manifest = cast("dict[str, Any]", manifest_value) + chain.append(manifest) + ingredients = manifest.get("ingredients") + if not isinstance(ingredients, list): + continue + for ingredient_value in cast("list[object]", ingredients): + if not isinstance(ingredient_value, dict): + continue + ingredient = cast("dict[object, object]", ingredient_value) + child = ingredient.get("active_manifest") + if ( + isinstance(child, str) + and child not in seen + and not _store_has_invalid_credential(cast("dict[str, Any]", ingredient)) + ): + pending.append(child) + return chain + + +def _status_codes(store: dict[str, Any]) -> tuple[list[str], list[str], list[str]]: + """Return ordered status codes for this store's active credential.""" + successes: list[str] = [] + informationals: list[str] = [] + failures: list[str] = [] + + def extend(status_set: object) -> None: + if not isinstance(status_set, dict): + return + mapping = cast("dict[object, object]", status_set) + for key, target in ( + ("success", successes), + ("informational", informationals), + ("failure", failures), + ): + values = mapping.get(key) + if not isinstance(values, list): + continue + for value in cast("list[object]", values): + if not isinstance(value, dict): + continue + code = cast("dict[object, object]", value).get("code") + if isinstance(code, str) and code: + target.append(code) + + validation_results = store.get("validation_results") + if isinstance(validation_results, dict): + results = cast("dict[object, object]", validation_results) + extend(results.get("activeManifest")) + + if not (successes or informationals or failures): + # Older readers expose only the flat non-success list. It still carries the + # exact codes needed to reject a mismatched binding or invalid credential. + values = store.get("validation_status") + if isinstance(values, list): + for value in cast("list[object]", values): + if not isinstance(value, dict): + continue + code = cast("dict[object, object]", value).get("code") + if isinstance(code, str) and code: + failures.append(code) + return successes, informationals, failures + + +def _store_has_invalid_credential(store: dict[str, Any]) -> bool: + """Return whether this store's own active credential failed validation.""" + _, _, failures = _status_codes(store) + return any(marker in code for code in failures for marker in _CONTENT_BINDING_FAILURE_MARKERS) or any( + code in _SIGNATURE_FAILURES for code in failures + ) + + +def c2pa_info_has_invalid_credential(info: dict[str, Any]) -> bool: + """Return whether parsed C2PA info reports a failed binding or signature.""" + return info.get("c2pa_integrity") == "invalid" or info.get("c2pa_signature") == "invalid" + + +def c2pa_info_has_removal_hint(info: dict[str, Any]) -> bool: + """Return whether a C2PA AI or watermark claim should keep removal fail-safe.""" + return bool(info.get("ai_source_kind") or info.get("synthid_watermark")) + + +def _has_suffix(codes: list[str], suffixes: tuple[str, ...]) -> bool: + return any(code.endswith(suffix) for code in codes for suffix in suffixes) + + +def _validation_fields(store: dict[str, Any]) -> dict[str, Any]: + successes, informationals, failures = _status_codes(store) + all_codes = list(dict.fromkeys([*successes, *informationals, *failures])) + + binding_failed = any(marker in code for code in failures for marker in _CONTENT_BINDING_FAILURE_MARKERS) + binding_matched = _has_suffix(successes, _CONTENT_BINDING_MATCHES) + signature_failed = any(code in _SIGNATURE_FAILURES for code in failures) + signature_validated = "claimSignature.validated" in successes + + if binding_failed: + integrity = "invalid" + elif binding_matched: + integrity = "valid" + else: + integrity = "unknown" + if signature_failed: + signature = "invalid" + elif signature_validated: + signature = "valid" + else: + signature = "unknown" + + if "signingCredential.trusted" in successes: + signer_trust = "trusted" + elif "signingCredential.untrusted" in failures: + signer_trust = "untrusted" + else: + signer_trust = "unknown" + + if any(code in failures for code in ("signingCredential.invalid", "signingCredential.ocsp.revoked")): + signer_validity = "invalid" + elif "signingCredential.expired" in failures: + signer_validity = "expired" + elif "claimSignature.insideValidity" in successes: + signer_validity = "valid" + else: + signer_validity = "unknown" + + state = store.get("validation_state") + return { + "c2pa_validation_source": "reader", + "c2pa_validation_state": state if isinstance(state, str) and state else "unknown", + "c2pa_integrity": integrity, + "c2pa_signature": signature, + "c2pa_signer_trust": signer_trust, + "c2pa_signer_validity": signer_validity, + "c2pa_validation_codes": all_codes, + } + + def _claim_generator_from_store(store: dict[str, Any]) -> str | None: active = _active_manifest(store) direct = active.get("claim_generator") @@ -199,6 +378,127 @@ def _claim_generator_from_store(store: dict[str, Any]) -> str | None: return None +def _structured_manifest_fields(store: dict[str, Any]) -> dict[str, Any]: + """Extract provenance only from the active manifest and its ingredient graph.""" + chain = _manifest_chain(store) + if not chain: + return {} + + info: dict[str, Any] = {} + issuers: list[str] = [] + tools: list[str] = [] + actions: list[str] = [] + source_types: list[str] = [] + soft_binding_algorithms: list[str] = [] + claim_generator_asserts_ai = False + + def add_tool_matches(value: str, *, asserts_ai: bool = False) -> None: + nonlocal claim_generator_asserts_ai + matches = _ordered_matches(value.encode(), C2PA_AI_TOOLS) + tools.extend(matches) + if asserts_ai and matches: + claim_generator_asserts_ai = True + + for manifest in chain: + signature_value = manifest.get("signature_info") + if isinstance(signature_value, dict): + signature = cast("dict[object, object]", signature_value) + for key in ("issuer", "common_name", "certificate_issuer"): + value = signature.get(key) + if isinstance(value, str): + issuers.extend(_ordered_matches(value.encode(), C2PA_ISSUERS)) + + direct_generator = manifest.get("claim_generator") + if isinstance(direct_generator, str): + add_tool_matches(direct_generator, asserts_ai=True) + + candidates = manifest.get("claim_generator_info") + if isinstance(candidates, list): + for candidate_value in cast("list[object]", candidates): + if not isinstance(candidate_value, dict): + continue + name = cast("dict[object, object]", candidate_value).get("name") + if isinstance(name, str): + add_tool_matches(name, asserts_ai=True) + + assertions = manifest.get("assertions") + if not isinstance(assertions, list): + continue + for assertion_value in cast("list[object]", assertions): + if not isinstance(assertion_value, dict): + continue + assertion = cast("dict[object, object]", assertion_value) + label = assertion.get("label") + data = assertion.get("data") + if isinstance(label, str) and label.startswith("c2pa.soft-binding") and isinstance(data, dict): + algorithm = cast("dict[object, object]", data).get("alg") + if isinstance(algorithm, str): + soft_binding_algorithms.append(algorithm) + if not (isinstance(label, str) and label.startswith("c2pa.actions") and isinstance(data, dict)): + continue + action_values = cast("dict[object, object]", data).get("actions") + if not isinstance(action_values, list): + continue + for action_value in cast("list[object]", action_values): + if not isinstance(action_value, dict): + continue + action = cast("dict[object, object]", action_value) + action_name = action.get("action") + if isinstance(action_name, str): + actions.append(C2PA_ACTIONS.get(action_name.encode(), action_name.removeprefix("c2pa."))) + source_type = action.get("digitalSourceType") + if isinstance(source_type, str): + source_types.append(source_type) + software_agent = action.get("softwareAgent") + if isinstance(software_agent, dict): + name = cast("dict[object, object]", software_agent).get("name") + if isinstance(name, str): + add_tool_matches(name) + + issuers = list(dict.fromkeys(issuers)) + tools = list(dict.fromkeys(tools)) + actions = list(dict.fromkeys(actions)) + if issuers: + info["issuer"] = ", ".join(issuers) + if tools: + info["ai_tool"] = ", ".join(tools) + if actions: + info["actions"] = ", ".join(actions) + if claim_generator_asserts_ai: + info["c2pa_identity_ai"] = True + + generated = any( + "trainedAlgorithmicMedia" in value and "compositeWithTrainedAlgorithmicMedia" not in value + for value in source_types + ) + enhanced = any("compositeWithTrainedAlgorithmicMedia" in value for value in source_types) + procedural = any("algorithmicMedia" in value for value in source_types) + if generated: + info.update(source_type="trainedAlgorithmicMedia (AI-generated)", ai_source_kind="generated") + elif enhanced: + info.update(source_type="compositeWithTrainedAlgorithmicMedia (AI-enhanced)", ai_source_kind="enhanced") + elif procedural: + info["source_type"] = "algorithmicMedia" + + has_watermark_action = any(action.startswith("watermarked") for action in actions) + if has_watermark_action: + info["watermarked"] = True + if info.get("ai_source_kind"): + selected_bytes = json.dumps(chain, ensure_ascii=False).encode() + synthid = synthid_evidence_vendors_in(selected_bytes, has_watermark_action=has_watermark_action) + if synthid: + info["synthid_vendors"] = synthid + info["synthid_watermark"] = synthid_verdict(", ".join(synthid)) + + if soft_binding_algorithms: + algorithms = "\n".join(soft_binding_algorithms).encode() + soft_bindings = soft_binding_vendors_in(algorithms) + if soft_bindings: + info["soft_binding_vendors"] = soft_bindings + info["soft_binding"] = ", ".join(soft_bindings) + return info + + def synthid_verdict(vendors: str) -> str: """Describe why supported provenance establishes a SynthID watermark.""" return f"present according to {vendors} provenance" @@ -276,16 +576,28 @@ def _populate_registry_fields(buffer: bytes, info: dict[str, Any]) -> bool: def _base_info(byte_count: int, *, fallback: bool = False) -> dict[str, Any]: container = "C2PA manifest" if fallback else "C2PA manifest store" - return { + info: dict[str, Any] = { "has_c2pa": True, "type": "C2PA (Coalition for Content Provenance and Authenticity)", "c2pa_manifest": f"{container} ({byte_count} bytes)", } + if fallback: + info.update( + c2pa_validation_source="fallback", + c2pa_validation_state="unknown", + c2pa_integrity="unknown", + c2pa_signature="unknown", + c2pa_signer_trust="unknown", + c2pa_signer_validity="unknown", + c2pa_validation_codes=[], + ) + return info def _info_from_store(store: dict[str, Any], encoded: bytes) -> dict[str, Any]: info = _base_info(len(encoded)) - _populate_registry_fields(encoded, info) + info.update(_structured_manifest_fields(store)) + info.update(_validation_fields(store)) generator = _claim_generator_from_store(store) if generator is not None: info["claim_generator"] = generator diff --git a/src/remove_ai_watermarks/_internal/constants.py b/src/remove_ai_watermarks/_internal/constants.py index 01c3482..e0f70b0 100644 --- a/src/remove_ai_watermarks/_internal/constants.py +++ b/src/remove_ai_watermarks/_internal/constants.py @@ -107,6 +107,9 @@ C2PA_IDENTITY_AI_ORGS = frozenset(vendor.org for vendor in C2PA_AI_VENDORS if ve # Keep this attribution beside the issuer registry so every C2PA consumer has one # canonical source rather than maintaining a derived product map in identify.py. C2PA_CLAIM_GENERATOR_PLATFORMS: tuple[tuple[str, str], ...] = ( + ("adobe_firefly", "Adobe Firefly"), + ("firefly", "Adobe Firefly"), + ("dreamina", "ByteDance (Doubao / Jimeng / Volcano Engine)"), ("higgsfield ai", "Higgsfield AI"), ("topaz labs image api", "Topaz Labs"), ("tiktok ad creative toolbox", "TikTok Ad Creative Toolbox"), @@ -122,6 +125,7 @@ C2PA_AI_TOOLS = { ("DALL", "DALL-E"), ("Imagen", "Imagen"), ("Firefly", "Firefly"), + ("Dreamina", "Dreamina"), ) } diff --git a/src/remove_ai_watermarks/api.py b/src/remove_ai_watermarks/api.py index 8d27eaf..9a8d178 100644 --- a/src/remove_ai_watermarks/api.py +++ b/src/remove_ai_watermarks/api.py @@ -329,12 +329,13 @@ class _SourceEvidence: if evidence is None: return True try: + from remove_ai_watermarks._internal.c2pa import c2pa_info_has_removal_hint from remove_ai_watermarks.identify import identify_from_evidence report = identify_from_evidence(evidence, image_path=self._path, check_invisible=True) except Exception: return True - return bool(report.ai_from_metadata) + return report.ai_from_metadata or c2pa_info_has_removal_hint(evidence.c2pa_info) def _provenance_from_report(report: Any, path: Path) -> frozenset[str]: diff --git a/src/remove_ai_watermarks/cli.py b/src/remove_ai_watermarks/cli.py index 5adecb6..21313de 100644 --- a/src/remove_ai_watermarks/cli.py +++ b/src/remove_ai_watermarks/cli.py @@ -1356,6 +1356,13 @@ def cmd_identify(ctx: click.Context, source: Path, no_visible: bool, as_json: bo verdict = "AI-generated (fully synthetic)" console.print(f"\n Verdict: {verdict} (confidence: {report.confidence})") console.print(f" Platform: {report.platform or 'undetermined'}") + if report.c2pa_validation is not None: + validation = report.c2pa_validation + console.print( + " C2PA validation: " + f"integrity={validation['integrity']}, signature={validation['signature']}, " + f"signer trust={validation['signer_trust']}, signer validity={validation['signer_validity']}" + ) if report.is_ai_generated is None: console.print( diff --git a/src/remove_ai_watermarks/identify.py b/src/remove_ai_watermarks/identify.py index d59ced1..ea54f0a 100644 --- a/src/remove_ai_watermarks/identify.py +++ b/src/remove_ai_watermarks/identify.py @@ -28,6 +28,8 @@ from typing import TYPE_CHECKING, Any, cast from remove_ai_watermarks._internal.c2pa import ( c2pa_info_from_manifest_store, + c2pa_info_has_invalid_credential, + c2pa_info_has_removal_hint, cbor_text_after, extract_c2pa_info, soft_binding_vendors_in, @@ -114,6 +116,18 @@ _SYNTHID_CAVEAT = ( "SynthID presence comes from supported provenance here; the pixel watermark is not locally " "decoded (proprietary decoder). Confirm via the Gemini app or openai.com/verify." ) +_C2PA_UNTRUSTED_CAVEAT = ( + "The C2PA claim signature and asset binding validate, but the signer identity is not anchored " + "to a trusted credential here; treat the named platform as a signed claim, not a verified identity." +) +_C2PA_UNVALIDATED_CAVEAT = ( + "The C2PA marker was parsed without cryptographic validation; treat its origin and watermark " + "assertions as unverified claims." +) +_C2PA_INVALID_CAVEAT = ( + "The embedded C2PA claim no longer validates against this asset. Its origin and watermark assertions " + "are retained only as removal hints, not as verified provenance." +) _IPTC_ONLY_CAVEAT = "The IPTC 'Made with AI' tag flags AI provenance but does not identify the specific platform." _INVISIBLE_WM_CAVEAT = ( "The open invisible watermark is fragile: it does not survive JPEG re-encoding " @@ -452,10 +466,10 @@ class ProvenanceReport: # (C2PA AI issuer / SynthID provenance, IPTC, AIGC, local gen params, EXIF/xAI, or # an open DWT-DCT decode) -- as opposed to a visible mark, provenance-only # TrustMark, or a weak medium-confidence hint (hf-job, Samsung genAIType). This - # is exactly the set of signals an invisible/diffusion scrub targets: a - # visible-only or no-signal image has it False. Equivalent to - # ``confidence == "high"``; - # surfaced as a field so callers gate on intent, not on the string. + # is the set of signals an invisible/diffusion scrub targets: a visible-only + # or no-signal image has it False. An intact C2PA AI claim from an untrusted + # signer is deliberately medium-confidence while this remains True; callers + # should gate on intent, not on the confidence string. ai_from_metadata: bool = False watermarks: list[str] = field(default_factory=list[str]) signals: list[Signal] = field(default_factory=list["Signal"]) @@ -465,6 +479,11 @@ class ProvenanceReport: # AI-generation markers). Non-empty means the provenance is internally # inconsistent -- a strong tell of spoofed, transplanted, or laundered metadata. integrity_clashes: list[str] = field(default_factory=list[str]) + # Orthogonal C2PA checks. A valid content binding does not make an untrusted + # signer identity trusted, and an expired credential does not by itself mean the + # signed bytes changed. Fallback parsing reports each dimension as unknown; + # None means no C2PA result exists. + c2pa_validation: dict[str, Any] | None = None def to_dict( self, @@ -500,9 +519,40 @@ class ProvenanceReport: ], "caveats": list(self.caveats), "integrity_clashes": list(self.integrity_clashes), + "c2pa_validation": self.c2pa_validation, } +def _c2pa_validation(info: dict[str, Any]) -> dict[str, Any] | None: + source = info.get("c2pa_validation_source") + if not isinstance(source, str): + return None + codes = info.get("c2pa_validation_codes") + return { + "source": source, + "state": str(info.get("c2pa_validation_state", "unknown")), + "integrity": str(info.get("c2pa_integrity", "unknown")), + "signature": str(info.get("c2pa_signature", "unknown")), + "signer_trust": str(info.get("c2pa_signer_trust", "unknown")), + "signer_validity": str(info.get("c2pa_signer_validity", "unknown")), + "codes": list(cast("list[object]", codes)) if isinstance(codes, list) else [], + } + + +def _c2pa_credential_level(info: dict[str, Any]) -> str: + """Return invalid, verified, or unverified for provenance attribution.""" + if c2pa_info_has_invalid_credential(info): + return "invalid" + if ( + info.get("c2pa_integrity") == "valid" + and info.get("c2pa_signature") == "valid" + and info.get("c2pa_signer_trust") == "trusted" + and info.get("c2pa_signer_validity") == "valid" + ): + return "verified" + return "unverified" + + def extract_provenance_evidence(image_path: Path) -> ProvenanceEvidence: """Read all file-backed metadata needed by provenance verdict logic once.""" return ProvenanceEvidence( @@ -1085,10 +1135,18 @@ def _identify_from_evidence( # ── C2PA Content Credentials ──────────────────────────────────── has_c2pa = bool(info) or c2pa_marker_in(head) + c2pa_validation = _c2pa_validation(info) + c2pa_level = _c2pa_credential_level(info) + c2pa_usable = c2pa_level != "invalid" + failed_c2pa_codes = [ + str(code) + for code in cast("list[object]", info.get("c2pa_validation_codes", [])) + if "mismatch" in str(code) or "invalid" in str(code) + ] issuers = [info["issuer"]] if info.get("issuer") else _issuers_in(region) # Full AI generation (trainedAlgorithmicMedia) vs an AI-enhanced real photo # (compositeWithTrainedAlgorithmicMedia). The structured kind is parsed once in - # _internal.c2pa._populate_registry_fields (covers PNG + any container the c2pa-python + # _internal.c2pa._structured_manifest_fields (covers PNG + any container the c2pa-python # reader handles); fall back to a raw head scan for the non-PNG raw-blob path # where extract_c2pa_info returns {}. Full generation wins when both appear. source_kind = _metadata_source_kind(info, head) @@ -1098,8 +1156,11 @@ def _identify_from_evidence( # Restricted to the ``asserts_ai`` vendors (distinctive brand strings), so it # does not reopen the incidental-mention problem the common-word issuers have. issuer_blob = " ".join(issuers) - c2pa_identity_ai = has_c2pa and any(org in issuer_blob for org in C2PA_IDENTITY_AI_ORGS) - c2pa_is_ai = source_kind is not None or c2pa_identity_ai + c2pa_identity_ai = has_c2pa and ( + bool(info.get("c2pa_identity_ai")) or any(org in issuer_blob for org in C2PA_IDENTITY_AI_ORGS) + ) + c2pa_claims_ai = source_kind is not None or c2pa_identity_ai + c2pa_is_ai = c2pa_usable and c2pa_claims_ai # Generator string (for the signal detail): structured for PNG, CBOR-scanned # for other containers. Best-effort -- some manifests key it as # `claim_generator_info` (Pixel), so this can be None even when a device is @@ -1119,20 +1180,41 @@ def _identify_from_evidence( camera_label or signer_label or (_claim_generator_platform(generator) if c2pa_is_ai else None) + or (_claim_generator_platform(str(info.get("ai_tool"))) if c2pa_is_ai and info.get("ai_tool") else None) or _attribute_platform(issuers, is_ai=c2pa_is_ai) ) - if has_c2pa + if has_c2pa and c2pa_usable else None ) if has_c2pa: detail = ", ".join(filter(None, [", ".join(issuers), generator, info.get("source_type")])) - signals.append(Signal("c2pa", detail or "C2PA manifest present", "high")) - watermarks.append(f"C2PA Content Credentials ({', '.join(issuers) or 'unknown signer'})") + if c2pa_level == "invalid": + suffix = f"; {', '.join(failed_c2pa_codes)}" if failed_c2pa_codes else "" + signals.append(Signal("c2pa", f"C2PA manifest present, credential integrity invalid{suffix}", "medium")) + watermarks.append("C2PA Content Credentials (invalid asset binding or signature)") + caveats.append(_C2PA_INVALID_CAVEAT) + else: + signals.append( + Signal("c2pa", detail or "C2PA manifest present", "high" if c2pa_level == "verified" else "medium") + ) + watermarks.append(f"C2PA Content Credentials ({', '.join(issuers) or 'unknown signer'})") + if c2pa_level == "unverified": + caveats.append( + _C2PA_UNTRUSTED_CAVEAT + if info.get("c2pa_integrity") == "valid" and info.get("c2pa_signature") == "valid" + else _C2PA_UNVALIDATED_CAVEAT + ) # Record the AI-origin vendor for clash detection only when the source is # actually AI -- classify the issuer attribution / generator, NOT the # resolved `platform` (which may be a camera device token whose label, # e.g. "Google Pixel", would mis-normalize to an AI vendor). - if c2pa_is_ai and (v := (_vendor_of(_attribute_platform(issuers, is_ai=True)) or _vendor_of(generator))): + if c2pa_is_ai and ( + v := ( + _vendor_of(_attribute_platform(issuers, is_ai=True)) + or _vendor_of(generator) + or _vendor_of(str(info.get("ai_tool", ""))) + ) + ): ai_vendor_claims["c2pa"] = v # ── C2PA cloud-manifest reference (Durable Content Credentials) ─ @@ -1170,9 +1252,13 @@ def _identify_from_evidence( if not synthid and trained_source and c2pa_marker_in(head) and (vendors := synthid_evidence_vendors_in(region)): synthid = synthid_verdict(", ".join(vendors)) if synthid: - watermarks.append(f"SynthID watermark ({synthid})") + watermarks.append( + f"SynthID watermark ({synthid})" + if c2pa_usable + else f"SynthID watermark claimed by invalid C2PA credentials ({synthid})" + ) caveats.append(_SYNTHID_CAVEAT) - if v := _vendor_of(synthid): + if c2pa_usable and (v := _vendor_of(synthid)): ai_vendor_claims["synthid"] = v # ── C2PA soft-binding: a named forensic/third-party watermark vendor ─ @@ -1180,12 +1266,17 @@ def _identify_from_evidence( # the watermark itself can't be decoded; names whose watermark stamped the pixels. soft_binding = meta.get("soft_binding") or (", ".join(v) if (v := soft_binding_vendors_in(region)) else None) if soft_binding: - signals.append(Signal("soft_binding", f"C2PA soft binding: {soft_binding}", "high")) + signals.append( + Signal( + "soft_binding", f"C2PA soft binding: {soft_binding}", "high" if c2pa_level == "verified" else "medium" + ) + ) watermarks.append(f"Forensic watermark soft binding ({soft_binding})") # ── IPTC "Made with AI" (Meta etc.), only meaningful without C2PA ─ iptc = any(m in head for m in IPTC_AI_MARKERS) - if iptc and not has_c2pa: + standalone_iptc = iptc and not has_c2pa + if standalone_iptc: signals.append(Signal("iptc", "digitalSourceType (Made with AI)", "high")) watermarks.append("IPTC digitalSourceType (Made with AI)") caveats.append(_IPTC_ONLY_CAVEAT) @@ -1311,8 +1402,18 @@ def _identify_from_evidence( exif_gen = any(s.name == "exif_generator" for s in signals) xai_sig = any(s.name == "xai_signature" for s in signals) ai_from_metadata = bool( - (has_c2pa and (c2pa_is_ai or synthid)) - or iptc + (has_c2pa and c2pa_usable and (c2pa_is_ai or synthid)) + or standalone_iptc + or iptc_ai + or aigc + or local_keys + or invisible_wm + or exif_gen + or xai_sig + ) + high_ai_from_metadata = bool( + (has_c2pa and c2pa_level == "verified" and (c2pa_is_ai or synthid)) + or standalone_iptc or iptc_ai or aigc or local_keys @@ -1330,7 +1431,7 @@ def _identify_from_evidence( if ai_from_metadata: is_ai: bool | None = True - confidence = "high" + confidence = "high" if high_ai_from_metadata else "medium" elif visible_only or hf_only or samsung_only: is_ai = True confidence = "medium" @@ -1339,7 +1440,17 @@ def _identify_from_evidence( confidence = "none" # ── Integrity clashes: contradictions between independent signals ─ - clashes = _integrity_clashes(ai_vendor_claims, camera_label, camera_has_ai_marker=bool(ai_vendor_claims)) + clashes = _integrity_clashes( + ai_vendor_claims, + camera_label if c2pa_usable else None, + camera_has_ai_marker=bool(ai_vendor_claims), + ) + if c2pa_level == "invalid": + clashes.insert( + 0, + "C2PA credentials failed integrity validation" + + (f": {', '.join(failed_c2pa_codes)}" if failed_c2pa_codes else "."), + ) caveats.append(_STRIP_CAVEAT) # De-duplicate while preserving order. @@ -1352,12 +1463,13 @@ def _identify_from_evidence( confidence=confidence, # Meaningful for the same digitalSourceType whether carried by C2PA or a # standalone IPTC/XMP label. Other AI signals leave it None. - ai_source_kind=source_kind if (is_ai and (has_c2pa or iptc)) else None, + ai_source_kind=(source_kind if (is_ai and ((has_c2pa and c2pa_usable) or standalone_iptc)) else None), ai_from_metadata=ai_from_metadata, watermarks=watermarks, signals=signals, caveats=caveats, integrity_clashes=clashes, + c2pa_validation=c2pa_validation, ) @@ -1434,13 +1546,14 @@ def has_invisible_target(image_path: Path) -> bool: The decision gate for the diffusion scrub (``invisible`` / ``all`` / ``batch``): regenerating pixels removes an AI-specific invisible watermark (SynthID, open DWT-DCT) but degrades a real photo, so it must not run when there is - nothing to remove. Runs :func:`identify` with ``check_visible=False`` -- a - visible mark is handled by the separate visible pass and is NOT a diffusion - target -- and ``check_invisible=True`` so an open watermark counts. Returns - ``report.ai_from_metadata`` (C2PA AI issuer / SynthID provenance, IPTC, AIGC, local - gen params, EXIF/xAI, or open DWT-DCT). TrustMark alone does not trigger the - scrub because it also protects human-authored work and therefore is not an AI - signal by itself. + nothing to remove. It runs the same evidence pipeline as :func:`identify` + with visible checks disabled and invisible checks enabled, so a visible mark + is handled by the separate visible pass and is NOT a diffusion target. Returns + True for ``report.ai_from_metadata`` (C2PA AI issuer / SynthID provenance, + IPTC, AIGC, local gen params, EXIF/xAI, or open DWT-DCT), and also when an + invalid C2PA claim retains an AI or watermark removal hint. TrustMark alone + does not trigger the scrub because it also protects human-authored work and + therefore is not an AI signal by itself. IMPORTANT -- this cannot prove a pixel SynthID is absent: SynthID is detectable only through its C2PA proxy, so a metadata-stripped AI image reads as no signal @@ -1451,8 +1564,17 @@ def has_invisible_target(image_path: Path) -> bool: watermark on a paid removal is worse than over-regenerating a clean image. """ try: - report = identify(image_path, check_visible=False, check_invisible=True) + evidence = extract_provenance_evidence(image_path) + report = _identify_from_evidence( + evidence, + image_path=image_path, + check_visible=False, + check_invisible=True, + ) except Exception: # unreadable / detector error -> do not skip the removal logger.debug("has_invisible_target: identify failed, defaulting to run", exc_info=True) return True - return report.ai_from_metadata + # An asset edit can invalidate the C2PA binding while leaving the declared pixel + # watermark intact. Keep the removal gate fail-safe without promoting that broken + # claim back into the provenance verdict. + return report.ai_from_metadata or c2pa_info_has_removal_hint(evidence.c2pa_info) diff --git a/src/remove_ai_watermarks/metadata.py b/src/remove_ai_watermarks/metadata.py index 50f146e..8de83a0 100644 --- a/src/remove_ai_watermarks/metadata.py +++ b/src/remove_ai_watermarks/metadata.py @@ -812,10 +812,19 @@ def synthid_source(image_path: Path) -> str | None: Returns: Comma-joined vendor name(s) (e.g. ``"OpenAI"``) or None. """ - from remove_ai_watermarks._internal.c2pa import extract_c2pa_info, synthid_evidence_vendors_in + from remove_ai_watermarks._internal.c2pa import ( + c2pa_info_has_invalid_credential, + extract_c2pa_info, + synthid_evidence_vendors_in, + ) - # PNG: the caBX chunk parser gives a clean, structured issuer. - vendors = extract_c2pa_info(image_path).get("synthid_vendors") + # Prefer the official reader's structured result. A failed asset binding or + # signature cannot establish the claim, and the raw fallback below must not + # resurrect it from the same damaged manifest bytes. + c2pa = extract_c2pa_info(image_path) + if c2pa_info_has_invalid_credential(c2pa): + return None + vendors = c2pa.get("synthid_vendors") if vendors: return ", ".join(vendors) @@ -1151,6 +1160,12 @@ def get_ai_metadata(image_path: Path) -> dict[str, str]: "actions", "synthid_watermark", "soft_binding", + "c2pa_validation_source", + "c2pa_validation_state", + "c2pa_integrity", + "c2pa_signature", + "c2pa_signer_trust", + "c2pa_signer_validity", ): if key in c2pa: result.setdefault(key, str(c2pa[key])) diff --git a/tests/conftest.py b/tests/conftest.py index be01cd4..e6541ff 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -2,7 +2,9 @@ from __future__ import annotations -from typing import TYPE_CHECKING +import struct +import zlib +from pathlib import Path import cv2 import numpy as np @@ -10,9 +12,6 @@ import pytest from PIL import Image from PIL.PngImagePlugin import PngInfo -if TYPE_CHECKING: - from pathlib import Path - @pytest.fixture def clean_photo(tmp_path: Path) -> Path: @@ -76,3 +75,22 @@ def tmp_clean_png(tmp_path: Path) -> Path: path = tmp_path / "clean.png" img.save(path, pnginfo=pnginfo) return path + + +@pytest.fixture +def tampered_chatgpt_png(tmp_path: Path) -> Path: + """Add valid PNG metadata after signing so the C2PA asset hash no longer matches.""" + source = Path(__file__).resolve().parents[1] / "data" / "fixtures" / "provenance" / "chatgpt-1.png" + data = source.read_bytes() + iend = data.rfind(b"\x00\x00\x00\x00IEND") + assert iend >= 0 + + kind = b"tEXt" + payload = b"c2pa-test\x00benign post-signing metadata mutation" + chunk = ( + struct.pack(">I", len(payload)) + kind + payload + struct.pack(">I", zlib.crc32(kind + payload) & 0xFFFFFFFF) + ) + target = tmp_path / "tampered-chatgpt.png" + target.write_bytes(data[:iend] + chunk + data[iend:]) + assert target.read_bytes() != data + return target diff --git a/tests/test_api.py b/tests/test_api.py index 2b0f35f..b5c6753 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -511,6 +511,15 @@ class TestSourceEvidenceHolder: assert holder.visible_provenance() == api.visible_provenance(DOUBAO) assert holder.has_invisible_target() == identify.has_invisible_target(DOUBAO) + def test_holder_preserves_invalid_c2pa_removal_hint(self, tampered_chatgpt_png): + from remove_ai_watermarks import api, identify + + holder = api._SourceEvidence(tampered_chatgpt_png) + + assert identify.identify(tampered_chatgpt_png, check_visible=False).ai_from_metadata is False + assert holder.has_invisible_target() is True + assert holder.has_invisible_target() == identify.has_invisible_target(tampered_chatgpt_png) + def test_extraction_failure_fails_safe_in_both_directions(self, monkeypatch, tmp_path): """No provenance means no relaxation; an unknown invisible target means SCRUB. Leaving a watermark on a paid removal is worse than over-regenerating.""" diff --git a/tests/test_cli.py b/tests/test_cli.py index cf482b2..bb6a259 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -723,6 +723,8 @@ class TestIdentifyCommand: result = runner.invoke(main, ["identify", str(sample), "--no-visible"]) assert result.exit_code == 0 assert "AI-generated (fully synthetic)" in result.output + assert "C2PA validation: integrity=valid, signature=valid" in result.output + assert "signer trust=untrusted, signer validity=expired" in result.output def test_identify_json_is_valid(self, runner, tmp_png_with_ai_metadata): result = runner.invoke(main, ["identify", str(tmp_png_with_ai_metadata), "--no-visible", "--json"]) diff --git a/tests/test_identify.py b/tests/test_identify.py index 02b894f..25f89cf 100644 --- a/tests/test_identify.py +++ b/tests/test_identify.py @@ -39,6 +39,79 @@ SAMPLES_DIR = Path(__file__).resolve().parent.parent / "data" / "fixtures" / "pr class TestProvenanceEvidence: + def test_exact_ai_claim_generator_can_assert_ai_without_source_type(self, tmp_path: Path): + path = tmp_path / "firefly.png" + info = { + "has_c2pa": True, + "issuer": "Adobe", + "claim_generator": "Adobe_Firefly", + "ai_tool": "Firefly", + "c2pa_identity_ai": True, + "c2pa_validation_source": "reader", + "c2pa_validation_state": "Valid", + "c2pa_integrity": "valid", + "c2pa_signature": "valid", + "c2pa_signer_trust": "untrusted", + "c2pa_signer_validity": "valid", + "c2pa_validation_codes": ["assertion.dataHash.match", "claimSignature.validated"], + } + evidence = ProvenanceEvidence( + path=path, + c2pa_info=info, + ai_metadata={}, + scan=b"jumb c2pa Adobe_Firefly", + iptc_ai_system=None, + aigc_label=None, + exif_generator=None, + xai_signature=False, + huggingface_job=None, + samsung_genai=None, + ) + + report = identify_from_evidence(evidence) + + assert report.is_ai_generated is True + assert report.confidence == "medium" + assert report.platform == "Adobe Firefly" + + def test_fully_validated_c2pa_claim_is_high_confidence(self, tmp_path: Path): + path = tmp_path / "validated.png" + info = { + "has_c2pa": True, + "issuer": "OpenAI", + "source_type": "trainedAlgorithmicMedia (AI-generated)", + "ai_source_kind": "generated", + "c2pa_validation_source": "reader", + "c2pa_validation_state": "Valid", + "c2pa_integrity": "valid", + "c2pa_signature": "valid", + "c2pa_signer_trust": "trusted", + "c2pa_signer_validity": "valid", + "c2pa_validation_codes": [ + "assertion.dataHash.match", + "claimSignature.validated", + "signingCredential.trusted", + ], + } + evidence = ProvenanceEvidence( + path=path, + c2pa_info=info, + ai_metadata={}, + scan=b"jumb c2pa OpenAI trainedAlgorithmicMedia", + iptc_ai_system=None, + aigc_label=None, + exif_generator=None, + xai_signature=False, + huggingface_job=None, + samsung_genai=None, + ) + + report = identify_from_evidence(evidence) + + assert report.is_ai_generated is True + assert report.confidence == "high" + assert report.platform == "OpenAI (ChatGPT / gpt-image / DALL-E / Sora)" + def test_external_metadata_record_builds_equivalent_evidence(self, tmp_path: Path): path = tmp_path / "external.jpg" signature = "A" * 64 @@ -428,12 +501,14 @@ class TestIdentifySamsungGalaxy: path.write_bytes(b"\xff\xd8\xff\xe1jumbc2pa" + blob + b"\xff\xd9") return path - def test_galaxy_trained_source_is_high_ai(self, tmp_path: Path): + def test_galaxy_trained_source_is_unverified_ai(self, tmp_path: Path): path = self._jpeg(tmp_path, "s25.jpg", b"Samsung Galaxy Galaxy S25 c2pa-rs trainedAlgorithmicMedia") r = identify(path, check_visible=False, check_invisible=False) assert r.is_ai_generated is True - assert r.confidence == "high" + assert r.confidence == "medium" assert r.platform == "Samsung Galaxy (C2PA)" + assert r.c2pa_validation is None + assert any("without cryptographic validation" in caveat for caveat in r.caveats) assert r.integrity_clashes == [] # device cert + AI source-type is legitimate, not a clash def test_galaxy_genai_only_is_medium_ai(self, tmp_path: Path): @@ -477,7 +552,7 @@ class TestIdentifyRealSamples: def test_openai_chatgpt(self): r = identify(SAMPLES_DIR / "chatgpt-1.png", check_visible=False) assert r.is_ai_generated is True - assert r.confidence == "high" + assert r.confidence == "medium" assert r.platform assert "OpenAI" in r.platform assert any("C2PA" in w for w in r.watermarks) @@ -546,9 +621,39 @@ class TestIdentifyRealSamples: # both invisible/metadata targets, so the diffusion scrub should run. assert has_invisible_target(SAMPLES_DIR / "chatgpt-1.png") is True assert has_invisible_target(SAMPLES_DIR / "mj-1.png") is True - # ai_from_metadata mirrors confidence == "high" and backs the helper. + # ai_from_metadata records scrub intent even when an untrusted signer + # makes the provenance verdict medium-confidence. assert identify(SAMPLES_DIR / "chatgpt-1.png", check_visible=False).ai_from_metadata is True + def test_untrusted_but_intact_c2pa_is_medium_confidence(self): + report = identify(SAMPLES_DIR / "chatgpt-1.png", check_visible=False, check_invisible=False) + + assert report.is_ai_generated is True + assert report.confidence == "medium" + assert report.ai_from_metadata is True + assert report.c2pa_validation is not None + assert report.c2pa_validation["source"] == "reader" + assert report.c2pa_validation["state"] == "Invalid" + assert report.c2pa_validation["integrity"] == "valid" + assert report.c2pa_validation["signature"] == "valid" + assert report.c2pa_validation["signer_trust"] == "untrusted" + assert report.c2pa_validation["signer_validity"] == "expired" + assert "assertion.dataHash.match" in report.c2pa_validation["codes"] + assert any("not anchored" in caveat for caveat in report.caveats) + + def test_hash_mismatch_does_not_confirm_origin_but_keeps_scrub_fail_safe(self, tampered_chatgpt_png: Path): + report = identify(tampered_chatgpt_png, check_visible=False, check_invisible=False) + + assert report.is_ai_generated is None + assert report.platform is None + assert report.confidence == "none" + assert report.ai_source_kind is None + assert report.ai_from_metadata is False + assert report.c2pa_validation is not None + assert report.c2pa_validation["integrity"] == "invalid" + assert any("dataHash.mismatch" in clash for clash in report.integrity_clashes) + assert has_invisible_target(tampered_chatgpt_png) is True + def test_has_invisible_target_false_on_clean_photo(self, clean_photo: Path): # No detectable invisible signal -> skip the scrub (do not degrade a clean image). assert has_invisible_target(clean_photo) is False @@ -566,13 +671,13 @@ class TestHasInvisibleTargetFailSafe: """The scrub gate fails SAFE: when a detector errors, it runs the removal.""" def test_detector_error_defaults_to_run(self, tmp_path: Path): - # If identify raises (a detector crash), the gate must return True so the + # If evidence evaluation raises (a detector crash), the gate must return True so the # caller still attempts removal -- leaving a watermark on a paid removal is # worse than over-regenerating. (Garbage bytes do NOT raise; identify returns # a clean None verdict there, so that path correctly skips -- see below.) bad = tmp_path / "x.png" bad.write_bytes(b"not image bytes") - with patch("remove_ai_watermarks.identify.identify", side_effect=RuntimeError("boom")): + with patch("remove_ai_watermarks.identify._identify_from_evidence", side_effect=RuntimeError("boom")): assert has_invisible_target(bad) is True def test_unreadable_bytes_are_not_a_target(self, tmp_path: Path): diff --git a/tests/test_metadata.py b/tests/test_metadata.py index bd369ff..20349e9 100644 --- a/tests/test_metadata.py +++ b/tests/test_metadata.py @@ -500,6 +500,10 @@ class TestGetAiMetadataRealSample: assert "OpenAI" in meta["issuer"] assert "synthid_watermark" not in meta assert "trainedAlgorithmicMedia" in meta["source_type"] + assert meta["c2pa_integrity"] == "valid" + assert meta["c2pa_signature"] == "valid" + assert meta["c2pa_signer_trust"] == "untrusted" + assert meta["c2pa_signer_validity"] == "expired" @pytest.mark.parametrize( diff --git a/tests/test_metadata_internals.py b/tests/test_metadata_internals.py index db51115..5a4e6ef 100644 --- a/tests/test_metadata_internals.py +++ b/tests/test_metadata_internals.py @@ -12,6 +12,7 @@ from PIL import Image from remove_ai_watermarks._internal.c2pa import ( _parse_c2pa_chunk, + c2pa_info_from_manifest_store, cbor_text_after, extract_c2pa_chunk, extract_c2pa_info, @@ -153,6 +154,110 @@ class TestC2PA: def test_c2pa_returns_false_for_non_png(self, tmp_jpeg_path): assert not has_c2pa_metadata(tmp_jpeg_path) + def test_structured_extraction_ignores_unreachable_manifests(self): + store = { + "active_manifest": "active", + "manifests": { + "active": { + "signature_info": {"issuer": "Adobe"}, + "assertions": [], + }, + "unreachable": { + "signature_info": {"issuer": "OpenAI"}, + "assertions": [ + { + "label": "c2pa.actions.v2", + "data": { + "actions": [ + { + "action": "c2pa.created", + "digitalSourceType": "trainedAlgorithmicMedia", + } + ] + }, + } + ], + }, + }, + } + + info = c2pa_info_from_manifest_store(store) + + assert info["issuer"] == "Adobe" + assert "source_type" not in info + assert "ai_source_kind" not in info + assert "c2pa_identity_ai" not in info + + def test_reachable_ingredient_claim_generator_can_assert_ai(self): + store = { + "active_manifest": "update", + "manifests": { + "update": { + "claim_generator": "c2pa-tool/0.1.0", + "ingredients": [{"active_manifest": "created"}], + "assertions": [], + }, + "created": { + "claim_generator": "Dreamina/7.5.0", + "assertions": [], + }, + }, + } + + info = c2pa_info_from_manifest_store(store) + + assert info["ai_tool"] == "Dreamina" + assert info["c2pa_identity_ai"] is True + + def test_invalid_ingredient_does_not_taint_active_validation_or_supply_claims(self): + store = { + "active_manifest": "update", + "validation_results": { + "activeManifest": { + "success": [ + {"code": "assertion.dataHash.match"}, + {"code": "claimSignature.validated"}, + ], + "failure": [{"code": "signingCredential.untrusted"}], + }, + "ingredientDeltas": [ + { + "validationDeltas": { + "failure": [{"code": "assertion.dataHash.mismatch"}], + } + } + ], + }, + "manifests": { + "update": { + "claim_generator": "c2pa-tool/0.1.0", + "ingredients": [ + { + "active_manifest": "created", + "validation_results": { + "activeManifest": { + "failure": [{"code": "assertion.dataHash.mismatch"}], + } + }, + } + ], + "assertions": [], + }, + "created": { + "claim_generator": "Dreamina/7.5.0", + "assertions": [], + }, + }, + } + + info = c2pa_info_from_manifest_store(store) + + assert info["c2pa_integrity"] == "valid" + assert info["c2pa_signature"] == "valid" + assert info["c2pa_signer_trust"] == "untrusted" + assert "ai_tool" not in info + assert "c2pa_identity_ai" not in info + SAMPLES_DIR = Path(__file__).resolve().parent.parent / "data" / "fixtures" / "provenance" CURRENT_OPENAI_SAMPLE = ( @@ -227,6 +332,22 @@ class TestC2PARealSamples: # Structured claim generator is exact, not a CBOR-scanned best-effort. assert info["claim_generator"] == "ChatGPT" + def test_reader_reports_intact_but_untrusted_credentials(self): + info = extract_c2pa_info(SAMPLES_DIR / "chatgpt-1.png") + + assert info["c2pa_integrity"] == "valid" + assert info["c2pa_signature"] == "valid" + assert info["c2pa_signer_trust"] == "untrusted" + assert info["c2pa_signer_validity"] == "expired" + assert "assertion.dataHash.match" in info["c2pa_validation_codes"] + + def test_reader_reports_post_signing_container_mutation(self, tampered_chatgpt_png): + info = extract_c2pa_info(tampered_chatgpt_png) + + assert info["c2pa_integrity"] == "invalid" + assert info["c2pa_signature"] == "valid" + assert "assertion.dataHash.mismatch" in info["c2pa_validation_codes"] + def test_fallback_to_png_parser_when_reader_unavailable(self, monkeypatch): """With the reader disabled, the hand-rolled PNG parser still works.""" from remove_ai_watermarks._internal import c2pa @@ -237,6 +358,8 @@ class TestC2PARealSamples: assert "OpenAI" in info["issuer"] assert "trainedAlgorithmicMedia" in info["source_type"] assert "synthid_watermark" not in info + assert info["c2pa_integrity"] == "unknown" + assert info["c2pa_validation_source"] == "fallback" class TestC2PAInjectValidation: diff --git a/tests/test_metadata_record.py b/tests/test_metadata_record.py index 88819b8..130be6f 100644 --- a/tests/test_metadata_record.py +++ b/tests/test_metadata_record.py @@ -293,6 +293,14 @@ class TestReportTransport: assert convenience == explicit + def test_c2pa_validation_survives_the_portable_record(self, tampered_chatgpt_png: Path): + direct = identify(tampered_chatgpt_png, check_visible=False, check_invisible=False) + portable = identify_metadata_record(collect_metadata_record(tampered_chatgpt_png), path=tampered_chatgpt_png) + + assert portable == direct + assert portable.c2pa_validation is not None + assert portable.c2pa_validation["integrity"] == "invalid" + def test_a_webp_record_matches(tmp_path: Path): """RIFF has its own walk; a chunk kept or dropped wrongly shows up here."""