Skip to content
146 changes: 99 additions & 47 deletions src/c2pa/c2pa.py
Original file line number Diff line number Diff line change
Expand Up @@ -1196,16 +1196,21 @@ def load_settings(settings: Union[str, dict], format: str = "json") -> None:


def _get_mime_type_from_path(path: Union[str, Path]) -> str:
Comment thread
tmathern marked this conversation as resolved.
"""Attempt to guess the MIME type from a file path (with extension).
"""Attempt to guess the MIME type from a file path's extension.
When the extension is missing or unrecognized, this returns an empty
string so the caller hands it to the lib for auto-detection.
A recognized-but-wrong extension still returns a mimetype:
the native layer will attempt to correct it from the bytes when
reading (the real type still needs to be supported by the lib),
and if it can't, an error will happen then.

Args:
path: File path as string or Path object

Returns:
MIME type string

Raises:
C2paError.NotSupported: If MIME type cannot be determined
MIME type string, or an empty string
(when it cannot be determined from the extension).
An empty string here means the native lib should attempt auto-detect.
"""
path_obj = Path(path)
file_extension = path_obj.suffix.lower() if path_obj.suffix else ""
Expand All @@ -1215,11 +1220,9 @@ def _get_mime_type_from_path(path: Union[str, Path]) -> str:
# so we bypass it and set the correct type
return "image/dng"
else:
mime_type = mimetypes.guess_type(str(path))[0]
if not mime_type:
raise C2paError.NotSupported(
f"Could not determine MIME type for file: {path}")
return mime_type
# Fall back to an empty string for extensionless or unknown files.
# Empty string flags this as a guess-type attempt.
return mimetypes.guess_type(str(path))[0] or ""


class ContextProvider(ABC):
Expand Down Expand Up @@ -1949,7 +1952,8 @@ def _get_supported_mime_types(ffi_func, cache):
try:
if arr[i] is None:
continue
mime_type = arr[i].decode("utf-8", errors='replace')
mime_type = arr[i].decode(
"utf-8", errors='replace').strip().lower()
if mime_type:
result.append(mime_type)
except Exception:
Expand All @@ -1969,27 +1973,51 @@ def _get_supported_mime_types(ffi_func, cache):


def _validate_and_encode_format(
format_str: str, supported_types: list[str], class_name: str
format_str: Optional[str],
supported_types: list[str],
class_name: str,
allow_autodetect: bool = True,
) -> bytes:
"""Validate a MIME type / format string and encode it to UTF-8 bytes.
"""Validate a MIME type/format string and encode it to UTF-8 bytes.

None, an empty string, or whitespace all mean the same thing: no format
was given. When allow_autodetect is True (default), that requests
auto-detection and the native library guesses the format from the bytes.
Auto-detection is best effort, so pass the format when you know it.
When allow_autodetect is False, a missing format is rejected instead.
A given format is matched case- and whitespace-insensitively against the
supported list.

Args:
format_str: The MIME type or format string to validate
supported_types: List of supported MIME types
class_name: Name of the calling class (for error messages)
format_str: The MIME type or format to validate, or None. Pass None,
an empty string, or whitespace to request auto-detection from the
asset's bytes (only when ``allow_autodetect`` is True).
supported_types: List of supported MIME types.
class_name: Name of the calling class (for error messages).
allow_autodetect: When True (default), a missing format requests
auto-detection. When False, a missing format raises
C2paError.NotSupported.

Returns:
UTF-8 encoded format bytes
UTF-8 encoded format bytes (empty bytes for auto-detection).

Raises:
C2paError.NotSupported: If the format is not supported
C2paError.Encoding: If the string contains invalid UTF-8 characters
C2paError.NotSupported: If a given format is unsupported, or if the
format is missing and ``allow_autodetect`` is False.
C2paError.Encoding: If the format contains invalid UTF-8 characters.
"""
if format_str.lower() not in supported_types:
key = (format_str or "").strip()
Comment thread
tmathern marked this conversation as resolved.
if not key:
if allow_autodetect:
return b""
Comment thread
tmathern marked this conversation as resolved.
raise C2paError.NotSupported(
f"{class_name} requires an explicit format (MIME type)"
)
if key.lower() not in supported_types:
raise C2paError.NotSupported(
f"{class_name} does not support {format_str}")
try:
return format_str.encode('utf-8')
return key.encode('utf-8')
except UnicodeError as e:
raise C2paError.Encoding(
f"Invalid UTF-8 characters in input: {e}")
Expand Down Expand Up @@ -2057,7 +2085,7 @@ def _is_mime_type_supported(cls, mime_type: str) -> bool:
@overload
def try_create(
cls,
format_or_path: Union[str, Path],
format_or_path: Union[str, Path, None],
stream: Optional[Any] = None,
manifest_data: Optional[Any] = None,
) -> Optional["Reader"]: ...
Expand All @@ -2066,7 +2094,7 @@ def try_create(
@overload
def try_create(
cls,
format_or_path: Union[str, Path],
format_or_path: Union[str, Path, None],
stream: Optional[Any],
manifest_data: Optional[Any],
context: 'ContextProvider',
Expand All @@ -2075,7 +2103,7 @@ def try_create(
@classmethod
def try_create(
cls,
format_or_path: Union[str, Path],
format_or_path: Union[str, Path, None],
stream: Optional[Any] = None,
manifest_data: Optional[Any] = None,
context: Optional['ContextProvider'] = None,
Expand All @@ -2090,7 +2118,8 @@ def try_create(
exceptions for the expected case of no manifest.

Args:
format_or_path: The format or path to read from
format_or_path: The format or path to read from.
None or an empty string requests auto-detection (best effort).
stream: Optional stream to read from (Python stream-like object)
manifest_data: Optional manifest data in bytes
context: Optional ContextProvider for settings
Expand All @@ -2113,37 +2142,46 @@ def try_create(
@overload
def __init__(
self,
format_or_path: Union[str, Path],
format_or_path: Union[str, Path, None],
stream: Optional[Any] = None,
manifest_data: Optional[Any] = None,
) -> None: ...

@overload
def __init__(
self,
format_or_path: Union[str, Path],
format_or_path: Union[str, Path, None],
stream: Optional[Any],
manifest_data: Optional[Any],
context: 'ContextProvider',
) -> None: ...

def __init__(
self,
format_or_path: Union[str, Path],
format_or_path: Union[str, Path, None],
stream: Optional[Any] = None,
manifest_data: Optional[Any] = None,
context: Optional['ContextProvider'] = None,
):
"""Create a new Reader.

The format is optional: pass None, an empty string, or read an
extensionless file by path to let the native library detect the
format. Detection is best effort, so pass the format when you know it.
A recognized but wrong format/extension is also corrected from the bytes
when reading (so reading doesn't fail on wrong file extension).
If the bytes are not a recognized asset, a C2paError is raised.

Args:
format_or_path: The format or path to read from
format_or_path: The format (MIME type) or path to read from.
None or an empty string requests auto-detection.
stream: Optional stream to read from (Python stream-like object)
manifest_data: Optional manifest data in bytes
context: Optional context implementing ContextProvider with settings

Raises:
C2paError: If there was an error creating the reader
C2paError: If there was an error creating the reader, including
when the format cannot be detected from an unrecognized asset
C2paError.Encoding: If any of the string inputs
contain invalid UTF-8 characters
"""
Expand Down Expand Up @@ -2175,29 +2213,28 @@ def __init__(
supported = Reader.get_supported_mime_types()

if stream is None:
# Create a stream from the file path in format_or_path
# Create a stream from the file path in format_or_path.
# Empty format = native lib will try to guess format.
path = str(format_or_path)
mime_type = _get_mime_type_from_path(path)

if not mime_type:
raise C2paError.NotSupported(
f"Could not determine MIME type for file: {path}")

format_bytes = _validate_and_encode_format(
mime_type, supported, "Reader")
self._init_from_file(path, format_bytes)

elif isinstance(stream, str):
# stream is a file path, format_or_path is the format
format_bytes = _validate_and_encode_format(
str(format_or_path), supported, "Reader")
None if format_or_path is None else str(format_or_path),
supported, "Reader")
self._init_from_file(
stream, format_bytes, manifest_data)

else:
# format_or_path is a format string, stream is a stream object
format_bytes = _validate_and_encode_format(
str(format_or_path), supported, "Reader")
None if format_or_path is None else str(format_or_path),
supported, "Reader")

with Stream(stream) as stream_obj:
self._create_reader(
Expand Down Expand Up @@ -2272,23 +2309,23 @@ def _init_from_context(self, context, format_or_path,
supported = Reader.get_supported_mime_types()

if stream is None:
# Empty format = native lib will try to guess format.
path = str(format_or_path)
mime_type = _get_mime_type_from_path(path)
if not mime_type:
raise C2paError.NotSupported(
f"Could not determine MIME type for file: {path}")
format_bytes = _validate_and_encode_format(
mime_type, supported, "Reader")
self._backing_file = open(path, 'rb')
self._own_stream = Stream(self._backing_file)
elif isinstance(stream, str):
format_bytes = _validate_and_encode_format(
str(format_or_path), supported, "Reader")
None if format_or_path is None else str(format_or_path),
supported, "Reader")
self._backing_file = open(stream, 'rb')
self._own_stream = Stream(self._backing_file)
else:
format_bytes = _validate_and_encode_format(
str(format_or_path), supported, "Reader")
None if format_or_path is None else str(format_or_path),
supported, "Reader")
self._own_stream = Stream(stream)

try:
Expand Down Expand Up @@ -2394,15 +2431,16 @@ def _get_cached_manifest_data(self) -> Optional[dict]:

return self._manifest_data_cache

def with_fragment(self, format: str, stream,
def with_fragment(self, format: Optional[str], stream,
fragment_stream) -> "Reader":
"""Process a BMFF fragment stream with this reader.

Used for fragmented BMFF media (DASH/HLS streaming) where
content is split into init segments and fragment files.

Args:
format: MIME type of the media (e.g., "video/mp4")
format: MIME type of the media (e.g., "video/mp4").
None or an empty string requests auto-detection (best effort).
stream: Stream-like object with the main/init segment data
fragment_stream: Stream-like object with the fragment data

Expand Down Expand Up @@ -3338,7 +3376,8 @@ def write_ingredient_archive(self, ingredient_id: str, stream: Any) -> None:
result = _lib.c2pa_builder_write_ingredient_archive(
self._handle, ingredient_id_str, stream_obj._stream)

_check_ffi_operation_result(result,
_check_ffi_operation_result(
result,
Builder._ERROR_MESSAGES["archive_error"].format(
"Unknown error"
),
Expand All @@ -3361,7 +3400,8 @@ def add_ingredient_from_archive(self, stream: Any) -> None:
result = _lib.c2pa_builder_add_ingredient_from_archive(
self._handle, stream_obj._stream)

_check_ffi_operation_result(result,
_check_ffi_operation_result(
result,
Builder._ERROR_MESSAGES["archive_read_error"].format(
"Unknown error"
),
Expand Down Expand Up @@ -3434,7 +3474,8 @@ def _sign_internal(
raise C2paError("Invalid or closed signer")

format_bytes = _validate_and_encode_format(
format, Builder.get_supported_mime_types(), "Builder")
format, Builder.get_supported_mime_types(), "Builder",
allow_autodetect=False)
manifest_bytes_ptr = ctypes.POINTER(ctypes.c_ubyte)()

try:
Expand Down Expand Up @@ -3647,9 +3688,17 @@ def sign_file(
Manifest bytes

Raises:
C2paError.NotSupported: If the format cannot be determined from
the source path (extensionless or unrecognized extension),
since signing requires an explicit, resolvable format.
C2paError: If there was an error during signing
"""
mime_type = _get_mime_type_from_path(source_path)
if not mime_type:
raise C2paError.NotSupported(
"Could not determine the format (MIME type) from "
f"'{source_path}'. Sign a stream with an explicit format "
"instead, or use a recognized file extension.")

try:
with (
Expand All @@ -3660,6 +3709,9 @@ def sign_file(
return self.sign(signer, mime_type, source_file, dest_file)
# else:
return self.sign(mime_type, source_file, dest_file)
except C2paError:
# Preserve C2paError and its subtypes
raise
except Exception as e:
raise C2paError(f"Error signing file: {str(e)}") from e

Expand Down
Loading
Loading