janware/jw-pkg
1
mirror of ssh://git.janware.com/janware/proj/jw-pkg synced 2026-04-24 09:13:37 +02:00
Code Activity Actions

lib.SSHClient: Log more details around exceptions

Browse source 
Add a wrapper around urlparse() and Paramiko's connect() function, in
order to log some more info in case an exception is thrown.

Signed-off-by: Jan Lindemann <jan@janware.com>
This commit is contained in:
Jan Lindemann 2026-03-18 07:27:59 +01:00
commit 49daa86696
1 changed files with 15 additions and 3 deletions
Download patch file Download diff file Expand all files Collapse all files

18
src/python/jw/pkg/lib/SSHClient.py
View file

@ -13,7 +13,11 @@ class SSHClient(ExecContext):
def __init__(self, uri: str, *args, **kwargs) -> None:
super().__init__(uri=uri, *args, **kwargs)
parsed = urlparse(uri)
try:
parsed = urlparse(uri)
except Exception as e:
log(ERR, f'Failed to parse SSH URI "{uri}"')
raise
self.__hostname = parsed.hostname
self.__password: parsed.password
self.__username = parsed.username
@ -103,7 +107,11 @@ class SSHClientInternal(SSHClient): # export
ret = paramiko.SSHClient()
ret.set_missing_host_key_policy(paramiko.AutoAddPolicy())
path_to_key=os.path.join(os.environ['HOME'], '.ssh', 'id_rsa')
ret.connect(self.hostname, key_filename=path_to_key, allow_agent=True)
try:
ret.connect(self.hostname, key_filename=path_to_key, allow_agent=True)
except Exception as e:
log(ERR, f'Failed to connect to {self.hostname} with key file {path_to_key} ({str(e)})')
raise
s = ret.get_transport().open_session()
# set up the agent request handler to handle agent requests from the server
paramiko.agent.AgentRequestHandler(s)
@ -120,7 +128,11 @@ class SSHClientInternal(SSHClient): # export
return SCPClient(self.__ssh.get_transport())
async def _run_ssh(self, cmd: list[str], cmd_input: str|None) -> Result:
stdin, stdout, stderr = self.__ssh.exec_command(shlex.join(cmd), timeout=self.__timeout)
try:
stdin, stdout, stderr = self.__ssh.exec_command(shlex.join(cmd), timeout=self.__timeout)
except Exception as e:
log(ERR, f'Command failed for {self.uri}: "{shlex.join(cmd)}"')
raise
if cmd_input is not None:
stdin.write(cmd_input)
exit_status = stdout.channel.recv_exit_status()