mirror of
https://github.com/wiltodelta/remove-ai-watermarks.git
synced 2026-08-19 12:07:13 +02:00
Validate C2PA credentials before attribution
This commit is contained in:
+4
-1
@@ -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:
|
||||
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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:
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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"),
|
||||
)
|
||||
}
|
||||
|
||||
|
||||
@@ -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]:
|
||||
|
||||
@@ -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(
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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]))
|
||||
|
||||
+22
-4
@@ -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
|
||||
|
||||
@@ -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."""
|
||||
|
||||
@@ -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"])
|
||||
|
||||
+111
-6
@@ -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):
|
||||
|
||||
@@ -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(
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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."""
|
||||
|
||||
Reference in New Issue
Block a user