Merge pull request #1 from OpenIPC/add-telnet-control

Add telnet control (set_telnet/telnet_enabled + CLI)

Author: widgetii

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)

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
 

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."""

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"}]}