Merge branch 'develop' into moxa-support

ktbyers · web-flow · commit ee3da59c75f7 · 2025-11-15T16:24:06.000-08:00
diff --git a/.gitignore b/.gitignore
@@ -26,6 +26,7 @@ bin/.netmiko.cfg
 
 # Virtual Environment files
 .venv/
+.venv_old/
 bin/
 lib/
 pyvenv.cfg
diff --git a/PLATFORMS.md b/PLATFORMS.md
@@ -25,6 +25,7 @@
 - Calix B6
 - Casa Systems CMTS
 - Centec Networks
+- Check Point GAiA
 - Cisco AireOS (Wireless LAN Controllers)
 - Cisco ASA
 - Cisco S200
@@ -120,6 +121,7 @@
 - MRV LX
 - Nokia/Alcatel SR-OS
 - Nokia SR Linux
+- Perle IOLAN Console Server
 - Optilink EOLT 9702 (telnet only)
 - QuantaMesh
 - Rad ETX
diff --git a/netmiko/base_connection.py b/netmiko/base_connection.py
@@ -193,7 +193,7 @@ def __init__(
         serial_settings: Optional[Dict[str, Any]] = None,
         fast_cli: bool = True,
         _legacy_mode: bool = False,
-        session_log: Optional[SessionLog] = None,
+        session_log: Optional[Union[str, io.BufferedIOBase, SessionLog]] = None,
         session_log_record_writes: bool = False,
         session_log_file_mode: str = "write",
         allow_auto_change: bool = False,
@@ -2011,6 +2011,29 @@ def check_enable_mode(self, check_string: str = "") -> bool:
         output = self.read_until_prompt(read_entire_line=True)
         return check_string in output
 
+    def enable_secret_handler(
+        self,
+        pattern: str,
+        output: str,
+        re_flags: int = re.IGNORECASE,
+    ) -> str:
+        """
+        Some platforms require special handling when entering 'enable' mode.
+
+        Send the "secret" in response to password pattern
+
+        :param pattern: pattern to search for indicating device is waiting for password
+
+        :param output: Accumulated output from 'enable()' method.
+
+        :param re_flags: Regular expression flags used in conjunction with pattern
+
+        """
+        if re.search(pattern, output, flags=re_flags):
+            self.write_channel(self.normalize_cmd(self.secret))
+            new_output = self.read_until_prompt()
+        return new_output
+
     def enable(
         self,
         cmd: str = "",
@@ -2054,10 +2077,9 @@ def enable(
                 pattern=pattern, re_flags=re_flags, read_entire_line=True
             )
 
-            # Send the "secret" in response to password pattern
-            if re.search(pattern, output, flags=re_flags):
-                self.write_channel(self.normalize_cmd(self.secret))
-                output += self.read_until_prompt()
+            output += self.enable_secret_handler(
+                pattern=pattern, output=output, re_flags=re_flags
+            )
 
             # Search for terminating pattern if defined
             if enable_pattern and not re.search(enable_pattern, output):
diff --git a/netmiko/checkpoint/checkpoint_gaia_ssh.py b/netmiko/checkpoint/checkpoint_gaia_ssh.py
@@ -1,3 +1,7 @@
+import time
+import re
+from typing import Optional, Any
+
 from netmiko.no_config import NoConfig
 from netmiko.base_connection import BaseConnection
 
@@ -8,16 +12,87 @@ class CheckPointGaiaSSH(NoConfig, BaseConnection):
     firewalls.
     """
 
+    prompt_pattern = r"[>#]"
+
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
+        # Kept running into issues with command_echo and duplicate echoes of commands.
+        self.fast_cli = False
+        fast_cli = kwargs.get("fast_cli") or False
+        kwargs["fast_cli"] = fast_cli
+        return super().__init__(*args, **kwargs)
+
     def session_preparation(self) -> None:
         """
         Prepare the session after the connection has been established.
 
         Set the base prompt for interaction ('>').
         """
-        self._test_channel_read(pattern=r"[>#]")
+        self._test_channel_read(pattern=self.prompt_pattern)
         self.set_base_prompt()
         self.disable_paging(command="set clienv rows 0")
 
+        # Clear read buffer
+        time.sleep(0.3 * self.global_delay_factor)
+        self.clear_buffer()
+
+    def check_enable_mode(self, check_string: str = "#") -> bool:
+        """Check if in enable mode. Return boolean."""
+        return super().check_enable_mode(check_string=check_string)
+
+    def enable_secret_handler(
+        self,
+        pattern: str,
+        output: str,
+        re_flags: int = re.IGNORECASE,
+    ) -> str:
+        """
+        Check Point Gaia requires very particular timing for this 'expert'
+        password handling to work.
+
+        Send the "secret" in response to password pattern
+        """
+        if re.search(pattern, output, flags=re_flags):
+            self.write_channel(self.secret)
+            time.sleep(0.3 * self.global_delay_factor)
+            self.write_channel(self.RETURN)
+            time.sleep(0.3 * self.global_delay_factor)
+            new_output = self.read_until_pattern(pattern=self.prompt_pattern)
+        return new_output
+
+    def enable(
+        self,
+        cmd: str = "expert",
+        pattern: str = r"expert password",
+        enable_pattern: Optional[str] = r"\#",
+        check_state: bool = True,
+        re_flags: int = re.IGNORECASE,
+    ) -> str:
+        """
+        Enter expert mode.
+
+        Check Point Gaia is very finicky on the timing of sending this 'expert' password.
+        """
+        output = super().enable(
+            cmd=cmd,
+            pattern=pattern,
+            enable_pattern=enable_pattern,
+            check_state=check_state,
+            re_flags=re_flags,
+        )
+        self.set_base_prompt()
+        return output
+
+    def exit_enable_mode(self, exit_command: str = "exit") -> str:
+        """Exit expert mode."""
+        output = ""
+        if self.check_enable_mode():
+            self.write_channel(self.normalize_cmd(exit_command))
+            output += self.read_until_pattern(pattern=r">")
+            self.set_base_prompt()
+            if self.check_enable_mode():
+                raise ValueError("Failed to exit enable mode.")
+        return output
+
     def save_config(
         self, cmd: str = "", confirm: bool = False, confirm_response: str = ""
     ) -> str:
diff --git a/netmiko/mikrotik/mikrotik_ssh.py b/netmiko/mikrotik/mikrotik_ssh.py
@@ -222,8 +222,8 @@ def remote_space_available(self, search_pattern: str = "") -> int:
         sys_res = self.ssh_ctl_chan._send_command_timing_str(remote_cmd).splitlines()
         for res in sys_res:
             if "free-memory" in res:
-                spaceMib = res.strip().replace("free-memory: ", "").replace("MiB", "")
-                return int(float(spaceMib) * 1048576)
+                space_str = res.strip().replace("free-memory: ", "")
+                return self._format_to_bytes(space_str)
         raise ValueError("Unexpected output from remote_space_available")
 
     def remote_file_size(
diff --git a/netmiko/nokia/__init__.py b/netmiko/nokia/__init__.py
@@ -5,5 +5,12 @@
 )
 
 from netmiko.nokia.nokia_srl import NokiaSrlSSH
+from netmiko.nokia.nokia_isam import NokiaIsamSSH
 
-__all__ = ["NokiaSrosSSH", "NokiaSrosFileTransfer", "NokiaSrosTelnet", "NokiaSrlSSH"]
+__all__ = [
+    "NokiaSrosSSH",
+    "NokiaSrosFileTransfer",
+    "NokiaSrosTelnet",
+    "NokiaSrlSSH",
+    "NokiaIsamSSH",
+]
diff --git a/netmiko/nokia/nokia_isam.py b/netmiko/nokia/nokia_isam.py
@@ -0,0 +1,73 @@
+import time
+import re
+from typing import Any
+
+from netmiko.base_connection import BaseConnection
+from netmiko.no_enable import NoEnable
+
+
+class NokiaIsamSSH(BaseConnection, NoEnable):
+
+    def session_preparation(self) -> None:
+        self._test_channel_read()
+        self.set_base_prompt()
+        commands = [
+            "environment inhibit-alarms",
+            "environment screen-length 0",
+        ]
+        for command in commands:
+            self.disable_paging(command=command, cmd_verify=True, pattern=r"#")
+        time.sleep(0.3 * self.global_delay_factor)
+        self.clear_buffer()
+
+    def set_base_prompt(self, *args: Any, **kwargs: Any) -> str:
+        """Remove the > when navigating into the different config level."""
+        cur_base_prompt = super().set_base_prompt(*args, **kwargs)
+        match = re.search(r"\*?(.*?)(>.*)*#", cur_base_prompt)
+        if match:
+            # strip off >... from base_prompt; strip off leading *
+            self.base_prompt: str = match.group(1)
+
+        return self.base_prompt
+
+    def cleanup(self, command: str = "logout") -> None:
+        """Gracefully exit the SSH session."""
+        try:
+            if self.check_config_mode():
+                self.exit_config_mode()
+        except Exception:
+            pass
+        # Always try to send final command
+        if self.session_log:
+            self.session_log.fin = True
+        self.write_channel(command + self.RETURN)
+
+    def check_config_mode(
+        self,
+        check_string: str = ">configure",
+        pattern: str = "#",
+        force_regex: bool = False,
+    ) -> bool:
+        """Use equivalent enable method."""
+        return super().check_config_mode(
+            check_string=check_string, pattern=pattern, force_regex=force_regex
+        )
+
+    def config_mode(
+        self, config_command: str = "configure", pattern: str = "", re_flags: int = 0
+    ) -> str:
+        return super().config_mode(
+            config_command=config_command, pattern=pattern, re_flags=re_flags
+        )
+
+    def exit_config_mode(self, exit_config: str = "exit", pattern: str = "") -> str:
+        return super().exit_config_mode(exit_config=exit_config)
+
+    def save_config(
+        self, cmd: str = "admin save", confirm: bool = False, confirm_response: str = ""
+    ) -> str:
+        return self._send_command_str(
+            command_string=cmd,
+            strip_prompt=False,
+            strip_command=False,
+        )
diff --git a/netmiko/perle/__init__.py b/netmiko/perle/__init__.py
@@ -0,0 +1,3 @@
+from netmiko.perle.perle_ssh import PerleIolanSSH
+
+__all__ = ("PerleIolanSSH",)
diff --git a/netmiko/perle/perle_ssh.py b/netmiko/perle/perle_ssh.py
@@ -0,0 +1,94 @@
+import re
+from typing import Any, Dict, List, Optional, Union
+
+from netmiko.cisco_base_connection import CiscoBaseConnection
+from netmiko.no_config import NoConfig
+from netmiko.utilities import structured_data_converter
+
+
+class PerleIolanSSH(NoConfig, CiscoBaseConnection):
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
+        super().__init__(*args, **kwargs)
+        self.default_enter = kwargs.get("default_enter", "\r")
+
+    def session_preparation(self) -> None:
+        self._test_channel_read()
+        self.set_base_prompt(alt_prompt_terminator="$")
+
+    def enable(
+        self,
+        cmd: str = "admin",
+        pattern: str = "ssword:",
+        enable_pattern: Optional[str] = None,
+        check_state: bool = True,
+        re_flags: int = re.IGNORECASE,
+    ) -> str:
+        return super().enable(cmd, pattern, enable_pattern, check_state, re_flags)
+
+    def exit_enable_mode(self, *args: Any, **kwargs: Any) -> str:
+        # Perle does not have this concept
+        return ""
+
+    def save_config(
+        self, cmd: str = "save", confirm: bool = True, confirm_response: str = "y"
+    ) -> str:
+        return super().save_config(cmd, confirm, confirm_response)
+
+    def send_command_timing(
+        self,
+        command_string: str,
+        *args: Any,
+        **kwargs: Any,
+    ) -> Union[str, List[Any], Dict[str, Any]]:
+        real_command_string = command_string
+        real_strip_command = kwargs.get("strip_command", True)
+        real_strip_prompt = kwargs.get("strip_prompt", True)
+        command = command_string
+        more = r"< Hit any key >"
+        kwargs["strip_prompt"] = False
+        kwargs["strip_command"] = False
+
+        output = str(super().send_command_timing(command_string, *args, **kwargs))
+
+        command_string = " "
+        kwargs["normalize"] = False
+        kwargs["strip_command"] = True
+        while more in output:
+            output = re.sub(r"\n" + more, "", output)
+            output += str(
+                super().send_command_timing(
+                    command_string,
+                    *args,
+                    **kwargs,
+                )
+            )
+
+        output = self._sanitize_output(
+            output,
+            strip_command=real_strip_command,
+            command_string=command,
+            strip_prompt=real_strip_prompt,
+        )
+        return_data = structured_data_converter(
+            command=real_command_string,
+            raw_data=output,
+            platform=self.device_type,
+            use_textfsm=kwargs.get("use_textfsm", False),
+            use_ttp=kwargs.get("use_ttp", False),
+            use_genie=kwargs.get("use_genie", False),
+            textfsm_template=kwargs.get("textfsm_template", None),
+            ttp_template=kwargs.get("ttp_template", None),
+            raise_parsing_error=kwargs.get("raise_parsing_error", False),
+        )
+
+        return return_data
+
+    def strip_prompt(self, a_string: str) -> str:
+        # Delete repeated prompts
+        old_string = a_string
+        while (a_string := super().strip_prompt(a_string)) != old_string:
+            old_string = a_string
+        return a_string
+
+    def cleanup(self, command: str = "logout") -> None:
+        return super().cleanup(command)
diff --git a/netmiko/ssh_dispatcher.py b/netmiko/ssh_dispatcher.py
diff --git a/tests/test_netmiko_show.py b/tests/test_netmiko_show.py

Original file line number	Diff line number	Diff line change
`@@ -0,0 +1,3 @@`
	`1`	`+from netmiko.perle.perle_ssh import PerleIolanSSH`
	`2`	`+`
	`3`	`+__all__ = ("PerleIolanSSH",)`