diff --git a/README.md b/README.md index dff43af..892c6e6 100644 --- a/README.md +++ b/README.md @@ -21,12 +21,31 @@ dump to your device. ## First identify your camera -The known failure affects the `EF-S7-V1.0.30B` camera family. A newer -`EF-S7-V1.0.30D` revision uses different hardware and software and is not -affected by this particular problem. +The known failure affects one 8 MiB flash-layout variant of the +`EF-S7-V1.0.30B` camera family. An earlier 16 MiB 30B variant uses materially +different hardware and firmware. A newer `EF-S7-V1.0.30D` revision also uses +different hardware and software and is not affected by this particular +problem. Do this identification before building a USB cable, buying a programmer, or -running the tools. +running a repair operation. + +If the camera feed still works, start with the read-only stream check. Install +[`cc2camera`](docs/INSTALLATION.md), find the printer's IP address or hostname, +and run: + +```sh +cc2camera identify-camera PRINTER +``` + +For example: + +```sh +cc2camera identify-camera 192.168.1.50 +``` + +If the stream is unavailable, the signature is unknown, or you want authoritative +visual confirmation, inspect the hardware: 1. Power the printer off and unplug it from mains power. 2. Remove the camera module from the printer by undoing its single mounting @@ -39,18 +58,25 @@ unclear, remove the two housing screws and check the complete PCB revision. ![TX5110 processor on a newer CC2 camera](docs/images/ef-s7-v1.0.30d-tx5110.jpg) -| Identification | What is known | -|---|---| -| `EF-S7-V1.0.30B` / Ingenic T23 | This is the family on which the failure has been observed. Continue below. | -| `EF-S7-V1.0.30D` / likely TX5110 | This revision is not affected by the known failure. This guide does not apply. | -| Any other revision or processor | It has not been investigated. We do not know whether it is affected, and this guide does not apply. | +The four-digit PCB manufacturing code appears to use `WWYY` week/year order. +It is useful supporting evidence, but is not a safe cutoff on its own: component +or firmware changes may not align exactly with calendar weeks. + +| Identification | Observed manufacturing codes | What is known | +|---|---|---| +| `EF-S7-V1.0.30B` / Ingenic T23 / 8 MiB `ZB25VQ64` family | `0226`, `0526` (two units), `1526` | This is the supported layout on which the failure has been observed. Continue below. | +| `EF-S7-V1.0.30B` / Ingenic T23 / 16 MiB `P25Q128H` family | `4025` | This early layout is materially different. The known erase defect has not been established on it, and the current fix does not apply. | +| `EF-S7-V1.0.30D` / likely TX5110 | Not established | This revision is not affected by the known failure. This guide does not apply. | +| Any other revision, processor or flash layout | Not established | It has not been investigated. We do not know whether it is affected, and this guide does not apply. | ![EF-S7-V1.0.30D PCB marking](docs/images/ef-s7-v1.0.30d-revision.png) Photographs of the `EF-S7-V1.0.30B` board are available in the [OpenCentauri camera documentation](https://docs.opencentauri.cc/hardware/CC2/camera/). -The processor marking is a quick screening aid; the full PCB marking is the -authoritative visual identification. +The stream fingerprint is a convenient read-only family-identification aid. +The complete hardware markings and the tool's firmware and flash-layout checks +determine whether a 30B camera is supported. Never bypass a refusal because a +date code or stream signature appears to match. ## Choose the path that matches your camera diff --git a/cc2camera/cli.py b/cc2camera/cli.py index 90b0fcd..75a28cc 100644 --- a/cc2camera/cli.py +++ b/cc2camera/cli.py @@ -12,6 +12,7 @@ from . import __version__ from . import image as image_tools +from . import stream_identify from .display import invocation from .startup import install_startup from .restore_prepare import prepare_restore, validate_preparation_image @@ -125,6 +126,48 @@ def command_info(args) -> int: return 0 +def command_identify_camera(args) -> int: + """Identify a known camera family from the printer's read-only MJPEG stream.""" + + try: + identification = stream_identify.identify_camera_stream( + args.printer, timeout=args.timeout + ) + except stream_identify.StreamIdentificationError as exc: + raise ProtocolError(str(exc)) from exc + if identification.family is None: + raise ProtocolError( + "camera stream signature is not recognized; camera revision remains unknown" + ) + + print(f"Camera stream matches: {identification.family}") + print(f"Frames checked: {identification.frames_checked} consistent JPEG frames") + if identification.family == "EF-S7-V1.0.30B": + print( + "Evidence: this encoder signature has been observed on two independent " + "known 30B cameras." + ) + print( + "30B includes more than one flash-layout variant. The affected 8 MiB " + "variant has been observed with PCB date codes 0226, 0526 and 1526; " + "an earlier 4025 unit uses a different, unsupported 16 MiB layout." + ) + print( + "The stream cannot distinguish those variants. Date codes are supporting " + "evidence only; do not bypass the tool's firmware and flash-layout checks." + ) + else: + print( + "Evidence: this encoder signature has been observed on one known 30D camera." + ) + print("The known 30B camera failure has not been observed on the 30D family.") + print( + "Stream fingerprinting is a read-only family-identification aid; hardware " + "markings and the validated firmware layout determine whether the fix applies." + ) + return 0 + + def command_backup(args) -> int: output = Path(args.output) # Reject a non-atomic destination form before spending time on any physical @@ -372,8 +415,8 @@ def progress(done: int, total: int) -> None: if remaining <= 0: raise ProtocolError( "post-restore ADB availability timeout expired before the " - "temporary start. No HID ADB-start command was sent, and the " - "restore is not post-verified" + "temporary start. No HID ADB-start command was sent, and " + "the restore is not post-verified" ) try: start_adb_through_upload_command() @@ -442,6 +485,19 @@ def parser() -> argparse.ArgumentParser: info = commands.add_parser("device-info", parents=[common], help="validate root ADB and show flash layout; read-only") info.set_defaults(func=command_info) + identify = commands.add_parser( + "identify-camera", + help="identify a known 30B/30D camera from the printer MJPEG stream; read-only", + ) + identify.add_argument("printer", help="printer hostname or IP address") + identify.add_argument( + "--timeout", + type=_positive_finite_duration, + default=stream_identify.DEFAULT_TIMEOUT, + help="HTTP stream timeout in seconds (default: 5)", + ) + identify.set_defaults(func=command_identify_camera) + backup = commands.add_parser("backup", parents=[common], help="save three consecutive identical flash reads; read-only") backup.add_argument("output", help="new backup archive ending in .zip") backup.add_argument("--accept-bootloader-sha256", type=_sha256_argument, diff --git a/cc2camera/stream_identify.py b/cc2camera/stream_identify.py new file mode 100644 index 0000000..5ee3cc0 --- /dev/null +++ b/cc2camera/stream_identify.py @@ -0,0 +1,310 @@ +"""Read-only identification of known CC2 camera families from their MJPEG stream.""" + +from __future__ import annotations + +from dataclasses import dataclass +import hashlib +import ipaddress +from typing import BinaryIO, Callable +from urllib.request import urlopen + + +DEFAULT_STREAM_PORT = 8080 +DEFAULT_TIMEOUT = 5.0 +IDENTIFICATION_FRAMES = 3 +MAX_FRAME_BYTES = 2 * 1024 * 1024 +MAX_MULTIPART_LINE = 8192 +MAX_MULTIPART_HEADERS = 32 * 1024 +MAX_PREAMBLE_BYTES = 32 * 1024 + +# Observed on two independent EF-S7-V1.0.30B cameras. +_DQT_30B_SHA256 = "13660d69eacf5a054bdf3d88d8aead55e5857f704b6f3d9fa098ca93ecf9c44b" +# Observed on one EF-S7-V1.0.30D camera. +_DQT_30D_SHA256 = "2d13c72678b9293bb85e06153666f0b383896859603f8dc6f0cee57e816efec1" + +_MARKERS_30B = ("DQT", "DQT", "SOF0", "DHT", "DHT", "DHT", "DHT", "SOS") +_MARKERS_30D = ( + "APP0", "DQT", "DQT", "SOF0", "DRI", "DHT", "DHT", "DHT", "DHT", "SOS" +) + + +class StreamIdentificationError(ValueError): + """The camera stream could not be safely parsed for identification.""" + + +@dataclass(frozen=True) +class JpegFingerprint: + width: int + height: int + sampling: tuple[tuple[int, int, int], ...] + marker_sequence: tuple[str, ...] + dqt_sha256: str + jfif: tuple[int, int, int, int, int] | None + restart_interval: int | None + restart_markers: int + + +@dataclass(frozen=True) +class CameraIdentification: + family: str | None + fingerprint: JpegFingerprint + frames_checked: int + + +def _marker_name(marker: int) -> str: + names = { + 0xC0: "SOF0", + 0xC2: "SOF2", + 0xC4: "DHT", + 0xD8: "SOI", + 0xD9: "EOI", + 0xDA: "SOS", + 0xDB: "DQT", + 0xDD: "DRI", + 0xE0: "APP0", + } + if 0xE1 <= marker <= 0xEF: + return f"APP{marker - 0xE0}" + return names.get(marker, f"0xFF{marker:02X}") + + +def fingerprint_jpeg(frame: bytes) -> JpegFingerprint: + if len(frame) < 4 or not frame.startswith(b"\xff\xd8") or not frame.endswith(b"\xff\xd9"): + raise StreamIdentificationError("stream frame is not a complete JPEG") + + pos = 2 + marker_sequence: list[str] = [] + dqt_payloads: list[bytes] = [] + width = height = 0 + sampling: tuple[tuple[int, int, int], ...] = () + jfif: tuple[int, int, int, int, int] | None = None + restart_interval: int | None = None + scan_start: int | None = None + + while pos < len(frame) - 2: + if frame[pos] != 0xFF: + raise StreamIdentificationError("invalid JPEG marker structure before scan data") + while pos < len(frame) and frame[pos] == 0xFF: + pos += 1 + if pos >= len(frame): + raise StreamIdentificationError("truncated JPEG marker") + marker = frame[pos] + pos += 1 + if marker in range(0xD0, 0xD8) or marker == 0x01: + marker_sequence.append(_marker_name(marker)) + continue + if marker == 0xD9: + raise StreamIdentificationError("JPEG ended before a scan") + if pos + 2 > len(frame): + raise StreamIdentificationError("truncated JPEG segment length") + length = int.from_bytes(frame[pos : pos + 2], "big") + if length < 2 or pos + length > len(frame): + raise StreamIdentificationError("invalid JPEG segment length") + payload = frame[pos + 2 : pos + length] + name = _marker_name(marker) + marker_sequence.append(name) + pos += length + + if marker == 0xDB: + dqt_payloads.append(payload) + elif marker == 0xC0: + if len(payload) < 6: + raise StreamIdentificationError("truncated baseline JPEG frame header") + height = int.from_bytes(payload[1:3], "big") + width = int.from_bytes(payload[3:5], "big") + components = payload[5] + if len(payload) != 6 + components * 3: + raise StreamIdentificationError("invalid baseline JPEG component table") + sampling = tuple( + (payload[6 + i * 3], payload[7 + i * 3] >> 4, payload[7 + i * 3] & 0x0F) + for i in range(components) + ) + elif marker == 0xE0 and payload.startswith(b"JFIF\0") and len(payload) >= 12: + jfif = ( + payload[5], + payload[6], + payload[7], + int.from_bytes(payload[8:10], "big"), + int.from_bytes(payload[10:12], "big"), + ) + elif marker == 0xDD: + if len(payload) != 2: + raise StreamIdentificationError("invalid JPEG restart interval") + restart_interval = int.from_bytes(payload, "big") + elif marker == 0xDA: + scan_start = pos + break + + if scan_start is None or not width or not height or not dqt_payloads: + raise StreamIdentificationError("JPEG lacks required baseline encoder metadata") + scan = frame[scan_start:-2] + restart_markers = sum( + scan.count(bytes((0xFF, marker))) for marker in range(0xD0, 0xD8) + ) + return JpegFingerprint( + width=width, + height=height, + sampling=sampling, + marker_sequence=tuple(marker_sequence), + dqt_sha256=hashlib.sha256(b"".join(dqt_payloads)).hexdigest(), + jfif=jfif, + restart_interval=restart_interval, + restart_markers=restart_markers, + ) + + +def classify_fingerprint(fingerprint: JpegFingerprint) -> str | None: + common = ( + fingerprint.width == 640 + and fingerprint.height == 360 + and fingerprint.sampling == ((1, 2, 2), (2, 1, 1), (3, 1, 1)) + ) + if not common: + return None + if ( + fingerprint.marker_sequence == _MARKERS_30B + and fingerprint.dqt_sha256 == _DQT_30B_SHA256 + and fingerprint.jfif is None + and fingerprint.restart_interval is None + and fingerprint.restart_markers == 0 + ): + return "EF-S7-V1.0.30B" + if ( + fingerprint.marker_sequence == _MARKERS_30D + and fingerprint.dqt_sha256 == _DQT_30D_SHA256 + and fingerprint.jfif == (1, 2, 1, 72, 72) + and fingerprint.restart_interval == 40 + and fingerprint.restart_markers == 22 + ): + return "EF-S7-V1.0.30D" + return None + + +def identify_frames(frames: list[bytes]) -> CameraIdentification: + if not frames: + raise StreamIdentificationError("camera stream contained no JPEG frames") + fingerprints = [fingerprint_jpeg(frame) for frame in frames] + first = fingerprints[0] + if any(item != first for item in fingerprints[1:]): + return CameraIdentification(None, first, len(fingerprints)) + return CameraIdentification(classify_fingerprint(first), first, len(fingerprints)) + + +def _readline(stream: BinaryIO) -> bytes: + line = stream.readline(MAX_MULTIPART_LINE + 1) + if len(line) > MAX_MULTIPART_LINE: + raise StreamIdentificationError("MJPEG multipart line is too long") + return line + + +def _read_exact(stream: BinaryIO, length: int) -> bytes: + chunks: list[bytes] = [] + remaining = length + while remaining: + chunk = stream.read(remaining) + if not chunk: + raise StreamIdentificationError("MJPEG stream ended inside a JPEG frame") + chunks.append(chunk) + remaining -= len(chunk) + return b"".join(chunks) + + +def read_mjpeg_frames(stream: BinaryIO, boundary: bytes, count: int) -> list[bytes]: + delimiter = b"--" + boundary + closing = delimiter + b"--" + frames: list[bytes] = [] + preamble = 0 + + while len(frames) < count: + line = _readline(stream) + if not line: + break + stripped = line.rstrip(b"\r\n") + if stripped == closing: + break + if stripped != delimiter: + preamble += len(line) + if preamble > MAX_PREAMBLE_BYTES: + raise StreamIdentificationError("MJPEG multipart boundary was not found") + continue + + header_bytes = 0 + headers: dict[bytes, bytes] = {} + while True: + line = _readline(stream) + if not line: + raise StreamIdentificationError("MJPEG stream ended inside multipart headers") + header_bytes += len(line) + if header_bytes > MAX_MULTIPART_HEADERS: + raise StreamIdentificationError("MJPEG multipart headers are too large") + if line in (b"\r\n", b"\n"): + break + if b":" not in line: + raise StreamIdentificationError("invalid MJPEG multipart header") + name, value = line.split(b":", 1) + headers[name.strip().lower()] = value.strip() + + content_type = headers.get(b"content-type", b"").split(b";", 1)[0].strip().lower() + if content_type != b"image/jpeg": + raise StreamIdentificationError("MJPEG part is not image/jpeg") + raw_length = headers.get(b"content-length") + if raw_length is None: + raise StreamIdentificationError("MJPEG part has no Content-Length") + try: + length = int(raw_length) + except ValueError as exc: + raise StreamIdentificationError("invalid MJPEG Content-Length") from exc + if not 0 < length <= MAX_FRAME_BYTES: + raise StreamIdentificationError("MJPEG frame length is outside the accepted bound") + frames.append(_read_exact(stream, length)) + + if len(frames) < count: + raise StreamIdentificationError( + f"MJPEG stream ended after {len(frames)} frame(s); {count} are required" + ) + return frames + + +def _stream_url(host: str, port: int = DEFAULT_STREAM_PORT) -> str: + value = host.strip() + if not value or any(ch.isspace() for ch in value) or any(ch in value for ch in "/?#@"): + raise StreamIdentificationError( + "printer address must be a hostname or IP address, without a URL path" + ) + if ":" in value: + try: + ipaddress.IPv6Address(value) + except ValueError as exc: + raise StreamIdentificationError( + "printer address must not include a port; port 8080 is used" + ) from exc + value = f"[{value}]" + return f"http://{value}:{port}/" + + +def identify_camera_stream( + host: str, + *, + timeout: float = DEFAULT_TIMEOUT, + opener: Callable[..., object] = urlopen, +) -> CameraIdentification: + url = _stream_url(host) + try: + response = opener(url, timeout=timeout) + except OSError as exc: + raise StreamIdentificationError( + f"could not open camera stream at {url}: {exc}" + ) from exc + with response: + content_type = response.headers.get_content_type() + boundary = response.headers.get_param("boundary") + if content_type != "multipart/x-mixed-replace" or not boundary: + raise StreamIdentificationError( + "camera endpoint is not an MJPEG multipart stream" + ) + if isinstance(boundary, str): + boundary_bytes = boundary.encode("ascii", "strict") + else: + boundary_bytes = bytes(boundary) + frames = read_mjpeg_frames(response, boundary_bytes, IDENTIFICATION_FRAMES) + return identify_frames(frames) diff --git a/docs/CLI.md b/docs/CLI.md index 6eb0c3a..d557722 100644 --- a/docs/CLI.md +++ b/docs/CLI.md @@ -11,6 +11,7 @@ current folder. Python installations also support `python -m cc2camera`. ```text cc2camera devices [--adb EXECUTABLE] cc2camera device-info [--adb EXECUTABLE] [--serial SERIAL] +cc2camera identify-camera PRINTER [--timeout SECONDS] cc2camera start-adb [--adb EXECUTABLE] [--serial SERIAL] [--timeout SECONDS] cc2camera install-adb-startup --backup BACKUP.zip [--adb EXECUTABLE] [--serial SERIAL] [--yes] cc2camera install-erase-fix --backup BACKUP.zip [--adb EXECUTABLE] [--serial SERIAL] [--yes] @@ -21,6 +22,20 @@ cc2camera backup OUTPUT.zip [--adb EXECUTABLE] [--serial SERIAL] - `devices`: best-effort, read-only USB/HID and ADB listing. - `device-info`: validate root ADB and the supported MTD partition layout. This is not a complete firmware-image inspection. +- `identify-camera`: connect only to the printer's HTTP MJPEG stream on port + 8080 and compare three consecutive JPEG encoder fingerprints with the known + `EF-S7-V1.0.30B` and `EF-S7-V1.0.30D` signatures. Pass a printer hostname or + IP address, for example `cc2camera identify-camera 192.168.1.50`. The command + does not use USB, HID or ADB and does not modify the printer or camera. + An unrecognized or changing signature is refused as unknown rather than + guessed. The 30B signature has been observed on two independent known 30B + cameras; the 30D signature has been observed on one known 30D camera. The + stream cannot distinguish the early 16 MiB and affected 8 MiB 30B variants. + Affected 8 MiB cameras have been observed with PCB manufacturing codes + `0226`, `0526` and `1526`; a supplied early 16 MiB camera is marked `4025`. + These codes appear to use `WWYY` order, but are supporting evidence rather + than a safe compatibility cutoff. The tool's firmware and flash-layout + checks determine whether the fix applies. Default HTTP timeout: 5 seconds. - `start-adb`: temporarily start the existing daemon; no persistent startup file. Default timeout: 30 seconds. - `install-adb-startup`: install `enabled/90-adb.sh` and the shared @@ -140,6 +155,7 @@ The tool does not operate an external programmer. ## Exit status 0 means the requested operation succeeded. 2 means a CLI, validation, or -operating-system error. A successful dry run is not a successful physical -restore. A failed post-write check means a write may already have occurred; -read the error and keep the original backup. +operating-system error. An unknown `identify-camera` signature is a validation +failure and exits with 2 rather than guessing a camera revision. A successful +dry run is not a successful physical restore. A failed post-write check means a +write may already have occurred; read the error and keep the original backup. diff --git a/docs/STARTUP-HOOKS-VALIDATION.md b/docs/STARTUP-HOOKS-VALIDATION.md index adede09..57efc04 100644 --- a/docs/STARTUP-HOOKS-VALIDATION.md +++ b/docs/STARTUP-HOOKS-VALIDATION.md @@ -2,9 +2,9 @@ These findings cover the supported Ingenic T23 camera with ZB25VQ64 flash (JEDEC ID `5e4017`), Linux `3.10.14__isvp_pike_1.0__`, build #32 dated -2025-12-03. Tests were performed by the developer on one camera; private -same-camera backup archives were compared offline. No dumps or identity-bearing -file contents are included here. +2025-12-03, and PCB manufacturing code `1526`. Tests were performed by the +developer on one camera; private same-camera backup archives were compared +offline. No dumps or identity-bearing file contents are included here. The tested camera already had the image-based preventive startup change that copies defaults only when missing. Hook operation and the deliberate pressure diff --git a/hardware-recovery/TECHNICAL_DETAILS.md b/hardware-recovery/TECHNICAL_DETAILS.md index d9bedbe..67a9e9b 100644 --- a/hardware-recovery/TECHNICAL_DETAILS.md +++ b/hardware-recovery/TECHNICAL_DETAILS.md @@ -105,6 +105,21 @@ compact layout. Zlib decoding caps output at the declared size plus one byte. ## Supported image family +The `EF-S7-V1.0.30B` PCB label covers at least two materially different +hardware and firmware layouts. The four-digit PCB manufacturing code appears +to use `WWYY` week/year order. It is useful evidence, but is not treated as a +compatibility boundary: + +| Observed layout | Manufacturing-code observations | Evidence | +|---|---|---| +| Supported 8 MiB `ZB25VQ64` family | `0226`, `0526` (two units), `1526` | The failure was reported on all four cameras. The `1526` unit and one recovered bricked unit have been physically recovered or protected with this fix. | +| Early 16 MiB `P25Q128H` family | `4025` | Two supplied full dumps share the 16 MiB partition map. Photographs show the T23, 30B PCB label, manufacturing code and 128-Mbit flash. Their JFFS2 cleanmarkers occur at `0x8000` intervals, matching an erase size handled by their kernel. The known `0x4000` erase mismatch has not been established on this layout. | + +The early dumps were analyzed privately and remain outside the repository. +The 16 MiB family is not accepted by the builder or installers. A manufacturing +code, PCB revision or stream signature never overrides firmware, capacity, +partition-map or erase-geometry validation. + This release intentionally accepts only the exact firmware family already verified in three independent camera dumps with different unit identities: - 8 MiB `ZB25VQ64` SPI NOR diff --git a/tests/test_cli.py b/tests/test_cli.py index a3d1c02..acb44e1 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -28,10 +28,42 @@ def test_exact_command_surface(self): import argparse action = next(a for a in cli.parser()._actions if isinstance(a, argparse._SubParsersAction)) self.assertEqual(set(action.choices), { - "devices", "device-info", "start-adb", "install-adb-startup", + "devices", "device-info", "identify-camera", "start-adb", "install-adb-startup", "backup", "inspect-image", "build-image", "restore", "install-erase-fix", }) + def test_identify_camera_is_read_only_and_reports_known_family(self): + fingerprint = cli.stream_identify.JpegFingerprint( + 640, 360, ((1, 2, 2), (2, 1, 1), (3, 1, 1)), + ("DQT", "DQT", "SOF0", "DHT", "DHT", "DHT", "DHT", "SOS"), + "13660d69eacf5a054bdf3d88d8aead55e5857f704b6f3d9fa098ca93ecf9c44b", + None, None, 0, + ) + result = cli.stream_identify.CameraIdentification( + "EF-S7-V1.0.30B", fingerprint, 3 + ) + with mock.patch.object(cli.stream_identify, "identify_camera_stream", return_value=result) as identify, mock.patch.object(cli, "_adb") as adb, mock.patch.object(cli, "expected_devices") as hid, redirect_stdout(io.StringIO()) as output: + self.assertEqual(cli.main(["identify-camera", "printer.local"]), 0) + identify.assert_called_once_with("printer.local", timeout=5.0) + adb.assert_not_called() + hid.assert_not_called() + report = output.getvalue() + self.assertIn("EF-S7-V1.0.30B", report) + self.assertIn("0226, 0526 and 1526", report) + self.assertIn("4025", report) + self.assertIn("stream cannot distinguish", report) + self.assertIn("read-only family-identification aid", report) + self.assertNotIn("failure applies to the 30B family", report) + + def test_identify_camera_refuses_unknown_signature(self): + fingerprint = cli.stream_identify.JpegFingerprint( + 1, 1, (), (), "0" * 64, None, None, 0 + ) + result = cli.stream_identify.CameraIdentification(None, fingerprint, 3) + with mock.patch.object(cli.stream_identify, "identify_camera_stream", return_value=result), redirect_stderr(io.StringIO()) as error: + self.assertEqual(cli.main(["identify-camera", "printer.local"]), 2) + self.assertIn("not recognized", error.getvalue()) + def test_image_defaults_and_repeatable_reads(self): args = cli.parser().parse_args(["build-image", "one.bin", "--confirm-read", "two.bin", "--confirm-read", "three.bin"]) self.assertEqual(args.config_mode, "serial-only") diff --git a/tests/test_stream_identify.py b/tests/test_stream_identify.py new file mode 100644 index 0000000..445fe97 --- /dev/null +++ b/tests/test_stream_identify.py @@ -0,0 +1,108 @@ +"""Read-only MJPEG camera identification contracts using synthetic JPEG data.""" + +from __future__ import annotations + +import io +import unittest +from unittest import mock + +from cc2camera import stream_identify + + +class StreamIdentifyTests(unittest.TestCase): + def test_known_30b_fingerprint_requires_complete_signature(self): + known = stream_identify.JpegFingerprint( + width=640, + height=360, + sampling=((1, 2, 2), (2, 1, 1), (3, 1, 1)), + marker_sequence=("DQT", "DQT", "SOF0", "DHT", "DHT", "DHT", "DHT", "SOS"), + dqt_sha256="13660d69eacf5a054bdf3d88d8aead55e5857f704b6f3d9fa098ca93ecf9c44b", + jfif=None, + restart_interval=None, + restart_markers=0, + ) + self.assertEqual(stream_identify.classify_fingerprint(known), "EF-S7-V1.0.30B") + changed = stream_identify.JpegFingerprint( + **{**known.__dict__, "restart_markers": 1} + ) + self.assertIsNone(stream_identify.classify_fingerprint(changed)) + + def test_known_30d_fingerprint_requires_complete_signature(self): + known = stream_identify.JpegFingerprint( + width=640, + height=360, + sampling=((1, 2, 2), (2, 1, 1), (3, 1, 1)), + marker_sequence=( + "APP0", "DQT", "DQT", "SOF0", "DRI", + "DHT", "DHT", "DHT", "DHT", "SOS", + ), + dqt_sha256="2d13c72678b9293bb85e06153666f0b383896859603f8dc6f0cee57e816efec1", + jfif=(1, 2, 1, 72, 72), + restart_interval=40, + restart_markers=22, + ) + self.assertEqual(stream_identify.classify_fingerprint(known), "EF-S7-V1.0.30D") + changed = stream_identify.JpegFingerprint( + **{**known.__dict__, "restart_interval": 41} + ) + self.assertIsNone(stream_identify.classify_fingerprint(changed)) + + def test_multipart_reader_is_bounded_and_requires_jpeg_parts(self): + jpeg = b"\xff\xd8synthetic\xff\xd9" + payload = ( + b"--frame\r\nContent-Type: image/jpeg\r\nContent-Length: " + + str(len(jpeg)).encode() + + b"\r\n\r\n" + + jpeg + + b"\r\n" + ) * 3 + self.assertEqual( + stream_identify.read_mjpeg_frames(io.BytesIO(payload), b"frame", 3), + [jpeg, jpeg, jpeg], + ) + + bad = b"--frame\r\nContent-Type: text/plain\r\nContent-Length: 1\r\n\r\nx" + with self.assertRaisesRegex(stream_identify.StreamIdentificationError, "image/jpeg"): + stream_identify.read_mjpeg_frames(io.BytesIO(bad), b"frame", 1) + + def test_multipart_reader_rejects_short_or_oversized_frames(self): + short = b"--frame\r\nContent-Type: image/jpeg\r\nContent-Length: 5\r\n\r\nabc" + with self.assertRaisesRegex(stream_identify.StreamIdentificationError, "inside a JPEG"): + stream_identify.read_mjpeg_frames(io.BytesIO(short), b"frame", 1) + + large = ( + b"--frame\r\nContent-Type: image/jpeg\r\nContent-Length: " + + str(stream_identify.MAX_FRAME_BYTES + 1).encode() + + b"\r\n\r\n" + ) + with self.assertRaisesRegex(stream_identify.StreamIdentificationError, "accepted bound"): + stream_identify.read_mjpeg_frames(io.BytesIO(large), b"frame", 1) + + def test_printer_address_does_not_accept_paths_or_ports(self): + self.assertEqual(stream_identify._stream_url("192.0.2.10"), "http://192.0.2.10:8080/") + self.assertEqual(stream_identify._stream_url("camera.local"), "http://camera.local:8080/") + self.assertEqual(stream_identify._stream_url("2001:db8::1"), "http://[2001:db8::1]:8080/") + for bad in ("http://camera.local", "camera.local/path", "camera.local:9000", ""): + with self.subTest(bad=bad), self.assertRaises(stream_identify.StreamIdentificationError): + stream_identify._stream_url(bad) + + def test_inconsistent_frames_fail_closed_as_unknown(self): + first = stream_identify.JpegFingerprint( + 640, 360, ((1, 2, 2), (2, 1, 1), (3, 1, 1)), + ("DQT", "DQT", "SOF0", "DHT", "DHT", "DHT", "DHT", "SOS"), + "13660d69eacf5a054bdf3d88d8aead55e5857f704b6f3d9fa098ca93ecf9c44b", + None, None, 0, + ) + second = stream_identify.JpegFingerprint( + **{**first.__dict__, "dqt_sha256": "0" * 64} + ) + with mock.patch.object( + stream_identify, "fingerprint_jpeg", side_effect=[first, second] + ): + result = stream_identify.identify_frames([b"one", b"two"]) + self.assertIsNone(result.family) + self.assertEqual(result.frames_checked, 2) + + +if __name__ == "__main__": + unittest.main()