From 80e585caf594e2939686dc7b7802dcfe755ba76b Mon Sep 17 00:00:00 2001 From: "Lasota, Adrian" Date: Mon, 6 Oct 2025 11:37:03 +0200 Subject: [PATCH 1/2] feat: Add with_privileges parameter to SSHConnection for elevated command execution Signed-off-by: Lasota, Adrian --- README.md | 3 ++- mfd_connect/ssh.py | 4 +++- 2 files changed, 5 insertions(+), 2 deletions(-) diff --git a/README.md b/README.md index b1efbb5..74b2805 100644 --- a/README.md +++ b/README.md @@ -286,7 +286,7 @@ It's a Connection done by paramiko library. Constructor: -`SSHConnection(ip: str, port: int = 22, username: str, password: Optional[str], key_path: Optional[Union[List[Union[str, "Path"]], str, "Path"]] = None, skip_key_verification: bool = False, model: "BaseModel | None" = None, default_timeout: int | None = None,)` +`SSHConnection(ip: str, port: int = 22, username: str, password: Optional[str], key_path: Optional[Union[List[Union[str, "Path"]], str, "Path"]] = None, skip_key_verification: bool = False, model: "BaseModel | None" = None, default_timeout: int | None = None, with_privileges: bool = False)` ```python """ Initialise SSHConnection. @@ -300,6 +300,7 @@ Initialise SSHConnection. :param skip_key_verification: To skip checking of host's key, equivalent of StrictHostKeyChecking=no :param model: pydantic model of connection :param default_timeout: Set default_timeout property. +:param with_privileges: Whether to use sudo or runas for commands that require elevated privileges """ ``` diff --git a/mfd_connect/ssh.py b/mfd_connect/ssh.py index 8bee6e6..016370d 100644 --- a/mfd_connect/ssh.py +++ b/mfd_connect/ssh.py @@ -73,6 +73,7 @@ def __init__( model: "BaseModel | None" = None, default_timeout: int | None = None, cache_system_data: bool = True, + with_privileges: bool = False, **kwargs, ) -> None: """ @@ -91,9 +92,10 @@ def __init__( :param model: pydantic model of connection :param default_timeout: Timeout value for executing timeout for entire class. :param cache_system_data: Flag to cache system data like self._os_type, OS name, OS bitness and CPU architecture + :param with_privileges: Whether to use sudo or runas for commands that require elevated privileges """ super().__init__(ip, model, default_timeout, cache_system_data) - self.__use_sudo = False + self.__use_sudo = with_privileges self._ip = IPAddress(ip) self._connection = SSHClient() self._connection_details = { From 9bebac9ccd4816d7abba9df54cbad7612c0b0d32 Mon Sep 17 00:00:00 2001 From: Mateusz Chrominski Date: Mon, 20 Oct 2025 12:26:04 +0200 Subject: [PATCH 2/2] fix: Escape '"' when sudo enabled and 'echo' called. Signed-off-by: Mateusz Chrominski --- mfd_connect/ssh.py | 13 +++++++++---- 1 file changed, 9 insertions(+), 4 deletions(-) diff --git a/mfd_connect/ssh.py b/mfd_connect/ssh.py index 016370d..8d9de95 100644 --- a/mfd_connect/ssh.py +++ b/mfd_connect/ssh.py @@ -564,7 +564,7 @@ def execute_command( logger.log( level=log_levels.MFD_INFO, msg="[Warning] A pseudo-terminal was requested, " - "but please be aware that this is not recommended and may lead to unexpected behavior.", + "but please be aware that this is not recommended and may lead to unexpected behavior.", ) self._verify_command_correctness(command) @@ -906,9 +906,14 @@ def _adjust_command(self, command: str) -> str: :param command: command to adjust :return: command """ - if self.__use_sudo: - return f'sudo sh -c "{command}"' if "echo" in command else f"sudo {command}" - return command + if not self.__use_sudo: + return command + + if "echo" in command: + _command = command.replace('"', '\\"') + return f'sudo sh -c "{_command}"' + + return f"sudo {command}" def start_process_by_start_tool( self,