Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand All @@ -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
"""
```

Expand Down
17 changes: 12 additions & 5 deletions mfd_connect/ssh.py
Original file line number Diff line number Diff line change
Expand Up @@ -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:
"""
Expand All @@ -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 = {
Expand Down Expand Up @@ -562,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)
Expand Down Expand Up @@ -904,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,
Expand Down