diff --git a/dahua/client.py b/dahua/client.py index 41f73fb..baf3daa 100644 --- a/dahua/client.py +++ b/dahua/client.py @@ -458,19 +458,27 @@ def factory_reset(self, names: list | None = None) -> dict: # -- firmware ----------------------------------------------------------- def firmware_state(self) -> Any: - """Current upgrader state (safe, read-only).""" - return self.call("upgrader.getState") + """Current upgrader state (safe, read-only): ``upgrader.getState``.""" + return self.call(const.UPGRADER_STATE) def upgrade_firmware(self, path: str, *, confirm: bool = False, fw_type: str = "System", chunk_size: int = 0x8000, progress: Callable[[int, int], None] | None = None) -> Any: - """Flash a firmware image: ``upgrader.start`` → chunked send → ``execute``. + """Flash a firmware image over DHIP: + ``upgrader.prepare`` → chunked ``upgrader.appendData`` → ``upgrader.execute``. + + These are the `hunter` daemon's RPC upgrade handlers as reversed on a + Zenointel GK7205 camera (the same ones the web ``upgrader.cgi`` bridges + to). *path* is the vendor upgrade package (a Dahua "zzip" — see + ``tools/zzip.py`` in the zenointel project), not a raw partition image. .. danger:: - This can permanently **brick** the device and is *reconstructed* from - the documented Dahua upgrade flow — its orchestration is validated - against a mock but it was deliberately never run on hardware. You - must pass ``confirm=True`` to proceed. + This can permanently **brick** the device. The orchestration is + validated against a mock and the method names are reversed from + firmware, but the JSON param names are not byte-proven and this has + deliberately never been run on hardware. Probe :meth:`firmware_state` + first, keep a UART/backup recovery path ready, and pass + ``confirm=True`` to proceed. """ if not confirm: raise ValueError( @@ -478,21 +486,21 @@ def upgrade_firmware(self, path: str, *, confirm: bool = False, "hardware; pass confirm=True to proceed") import os total = os.path.getsize(path) - self.call("upgrader.start", {"Type": fw_type}) + self.call(const.UPGRADER_PREPARE, {"Type": fw_type}) sent = 0 with open(path, "rb") as fh: while True: chunk = fh.read(chunk_size) if not chunk: break - resp, _ = self.request("upgrader.send", + resp, _ = self.request(const.UPGRADER_APPEND, {"Offset": sent, "Length": len(chunk)}, data=chunk) - self._check(resp, "upgrader.send") + self._check(resp, const.UPGRADER_APPEND) sent += len(chunk) if progress: progress(sent, total) - return self.call("upgrader.execute") + return self.call(const.UPGRADER_EXECUTE) # -- snapshot ----------------------------------------------------------- def snapshot(self, channel: int = 0, http_port: int = 80) -> bytes: diff --git a/dahua/const.py b/dahua/const.py index fbd0000..cb34e03 100644 --- a/dahua/const.py +++ b/dahua/const.py @@ -66,6 +66,18 @@ REBOOT = "magicBox.reboot" SHUTDOWN = "magicBox.shutdown" +# -- firmware upgrade ------------------------------------------------------- +# The `hunter` daemon's RPC upgrade handlers (reversed on a Zenointel GK7205 +# camera): prepare -> appendData(chunk) -> execute; getState is read-only. These +# are the same handlers the web /cgi-bin/upgrader.cgi bridges to. The JSON param +# names below (Type / Offset+Length) match the reversed "append upgrade data" +# stream but are not byte-proven — verify against upgrader.getState / a web +# capture before trusting a real flash. +UPGRADER_STATE = "upgrader.getState" +UPGRADER_PREPARE = "upgrader.prepare" +UPGRADER_APPEND = "upgrader.appendData" +UPGRADER_EXECUTE = "upgrader.execute" + # -- PTZ operation codes ---------------------------------------------------- # Dahua PTZ verbs accepted by ptz.start / ptz.stop (the legacy code-based API, # the most widely-supported PTZ interface across firmwares). diff --git a/tests/test_dahua.py b/tests/test_dahua.py index 92c6da9..9f88d21 100644 --- a/tests/test_dahua.py +++ b/tests/test_dahua.py @@ -322,8 +322,8 @@ def send(req): return {"result": True} handlers = { - "upgrader.start": lambda r: {"result": True}, - "upgrader.send": send, + "upgrader.prepare": lambda r: {"result": True}, + "upgrader.appendData": send, "upgrader.execute": lambda r: {"result": True}, } # The fake server doesn't expose binary bodies to handlers, so assert @@ -340,8 +340,8 @@ def send(req): progress=lambda s, t: seen.append((s, t))) finally: os.remove(blob) - self.assertEqual(srv.received.count("upgrader.send"), 3) # 4096*3 covers 10000 - self.assertIn("upgrader.start", srv.received) + self.assertEqual(srv.received.count("upgrader.appendData"), 3) # 4096*3 covers 10000 + self.assertIn("upgrader.prepare", srv.received) self.assertIn("upgrader.execute", srv.received) self.assertEqual(seen[-1], (10000, 10000))