From f3752bfdc88c674328c1ec940b129ebd6c9ed2ed Mon Sep 17 00:00:00 2001 From: John Date: Thu, 18 Jun 2026 14:38:42 +0300 Subject: [PATCH] add set_telnet/telnet_enabled for InnerServer.Telnet config Verified on hardware (SD-2N-4G, firmware 30.13...R): - set_telnet(True) opens port 23, set_telnet(False) closes it - Read-modify-write preserves other InnerServer subkeys (SSH, FTP, etc.) - Sync + async surfaces, CLI subcommand, unit test --- dahua/aio.py | 28 ++++++++++++++++++++++++++++ dahua/cli.py | 15 ++++++++++++++- dahua/client.py | 28 ++++++++++++++++++++++++++++ tests/test_dahua.py | 24 ++++++++++++++++++++++++ 4 files changed, 94 insertions(+), 1 deletion(-) diff --git a/dahua/aio.py b/dahua/aio.py index e1f3172..82a5a9f 100644 --- a/dahua/aio.py +++ b/dahua/aio.py @@ -220,6 +220,34 @@ async def get_config(self, name: str) -> Any: async def set_config(self, name: str, table: Any) -> dict: return await self.call(const.SET_CONFIG, {"name": name, "table": table}) + async def get_inner_server_config(self) -> Any: + """Return the ``InnerServer`` config table (Telnet, SSH, FTP, ...).""" + return await self.get_config("InnerServer") + + async def set_telnet(self, enable: bool = True) -> dict: + """Enable or disable the built-in telnet server (``InnerServer.Telnet``). + + Read-modify-write so other ``InnerServer`` subkeys (SSH, FTP, …) are + preserved. + """ + table = await self.get_inner_server_config() + if isinstance(table, dict): + sub = table.get("Telnet") + if not isinstance(sub, dict): + sub = {} + table["Telnet"] = sub + sub["Enable"] = enable + else: + table = {"Telnet": {"Enable": enable}} + return await self.set_config("InnerServer", table) + + async def telnet_enabled(self) -> bool: + """Return ``True`` if the built-in telnet server is enabled.""" + table = await self.get_inner_server_config() + if isinstance(table, dict): + return bool(table.get("Telnet", {}).get("Enable", False)) + return False + # -- users -------------------------------------------------------------- async def get_users(self) -> list: params = await self.call(const.GET_USERS) diff --git a/dahua/cli.py b/dahua/cli.py index ec399d6..d525638 100644 --- a/dahua/cli.py +++ b/dahua/cli.py @@ -101,6 +101,10 @@ def build_parser() -> argparse.ArgumentParser: p_call = sub.add_parser("call", help="raw RPC2 method") p_call.add_argument("method") p_call.add_argument("--params") + + p_tel = sub.add_parser("telnet", help="enable/disable the built-in telnet server") + p_tel.add_argument("state", nargs="?", choices=("on", "off"), + help="omit to read the current state") return ap @@ -204,6 +208,15 @@ def main(argv: list[str] | None = None) -> int: file=sys.stderr) elif cmd == "files": _print(cam.find_files(args.start, args.end, channel=args.channel)) + elif cmd == "telnet": + if args.state is None: + print("on" if cam.telnet_enabled() else "off") + elif args.state == "on": + cam.set_telnet(True) + print("[+] telnet enabled", file=sys.stderr) + else: + cam.set_telnet(False) + print("[+] telnet disabled", file=sys.stderr) elif cmd == "events": listener = cam.events(codes=args.codes) print(f"[+] listening for events {args.codes} (Ctrl-C to stop)", @@ -214,7 +227,7 @@ def main(argv: list[str] | None = None) -> int: listener.stop() else: print("nothing to do; pass -m METHOD or a subcommand " - "(info/config/users/ptz/snapshot/events)", file=sys.stderr) + "(info/config/users/ptz/snapshot/events/telnet)", file=sys.stderr) return 2 return 0 diff --git a/dahua/client.py b/dahua/client.py index 17907b3..41f73fb 100644 --- a/dahua/client.py +++ b/dahua/client.py @@ -155,6 +155,34 @@ def get_network_config(self) -> Any: def get_snap_config(self) -> Any: return self.get_config("Snap") + def get_inner_server_config(self) -> Any: + """Return the ``InnerServer`` config table (Telnet, SSH, FTP, ...).""" + return self.get_config("InnerServer") + + def set_telnet(self, enable: bool = True) -> dict: + """Enable or disable the built-in telnet server (``InnerServer.Telnet``). + + Read-modify-write so other ``InnerServer`` subkeys (SSH, FTP, …) are + preserved. + """ + table = self.get_inner_server_config() + if isinstance(table, dict): + sub = table.get("Telnet") + if not isinstance(sub, dict): + sub = {} + table["Telnet"] = sub + sub["Enable"] = enable + else: + table = {"Telnet": {"Enable": enable}} + return self.set_config("InnerServer", table) + + def telnet_enabled(self) -> bool: + """Return ``True`` if the built-in telnet server is enabled.""" + table = self.get_inner_server_config() + if isinstance(table, dict): + return bool(table.get("Telnet", {}).get("Enable", False)) + return False + # -- OSD / channel title ----------------------------------------------- def get_channel_titles(self) -> list: """The channel title overlay strings, one per channel.""" diff --git a/tests/test_dahua.py b/tests/test_dahua.py index 47efc77..92c6da9 100644 --- a/tests/test_dahua.py +++ b/tests/test_dahua.py @@ -233,6 +233,30 @@ def test_absolute_and_presets(self): self.assertEqual(presets[0]["arg2"], 3) +class TestTelnet(unittest.TestCase): + def test_set_telnet_read_modify_write(self): + store = {"table": {"Telnet": {"Enable": False}, "SSH": {"Enable": True}}} + + def get_cfg(req): + return {"result": True, "params": {"table": store["table"]}} + + def set_cfg(req): + store["table"] = req["params"]["table"] + return {"result": True, "params": {"options": None}} + + with FakeDHIPServer({"configManager.getConfig": get_cfg, + "configManager.setConfig": set_cfg}) as srv: + with DahuaClient("127.0.0.1", srv.port) as cam: + cam.login(USER, PASS, keep_alive=False) + self.assertFalse(cam.telnet_enabled()) + cam.set_telnet(True) + self.assertTrue(store["table"]["Telnet"]["Enable"]) + self.assertTrue(store["table"]["SSH"]["Enable"]) # preserved + cam.set_telnet(False) + self.assertFalse(store["table"]["Telnet"]["Enable"]) + self.assertTrue(store["table"]["SSH"]["Enable"]) + + class TestOSD(unittest.TestCase): def test_channel_title_round_trip(self): store = {"table": [{"Name": "cam0"}]}