freeipa/ipatests/pytest_ipa/integration/transport.py
Sergey Orlov ae533e2998 ipatests: allocate pseudo-terminal only for specific command
While "ktutil" does require a pseudo-terminal on particular systems to
operate, majority of programs do not need it.
At the same time invoking `ssh` with forced pseudo-terminal allocation
interferes with sessions multiplexing feature and increases connection
time. The increase can be as large as 10 seconds in certain cases which
leads to unexpected EOFs of pexpect utility.

Reviewed-By: Florence Blanc-Renaud <frenaud@redhat.com>
2021-03-29 14:55:23 +03:00

60 lines
2.0 KiB
Python

#
# Copyright (C) 2020 FreeIPA Contributors see COPYING for license
#
"""Enhanced SSH transport for pytest multihost
Provides SSH password login for OpenSSH transport
"""
import os
from .expect import IpaTestExpect
from pytest_multihost.transport import OpenSSHTransport
class IPAOpenSSHTransport(OpenSSHTransport):
def _get_ssh_argv(self):
"""Return the path to SSH and options needed for every call"""
control_file = os.path.join(self.control_dir.path, "control")
known_hosts_file = os.path.join(self.control_dir.path, "known_hosts")
argv = [
"ssh",
"-l",
self.host.ssh_username,
"-o",
"ControlPath=%s" % control_file,
"-o",
"StrictHostKeyChecking=no",
"-o",
"UserKnownHostsFile=%s" % known_hosts_file,
]
if self.host.ssh_key_filename:
key_filename = os.path.expanduser(self.host.ssh_key_filename)
argv.extend(["-i", key_filename])
elif self.host.ssh_password:
password_file = os.path.join(self.control_dir.path, "password")
with open(password_file, "w") as f:
os.fchmod(f.fileno(), 0o600)
f.write(self.host.ssh_password)
f.write("\n")
argv = ["sshpass", f"-f{password_file}"] + argv
else:
self.log.critical("No SSH credentials configured")
raise RuntimeError("No SSH credentials configured")
argv.append(self.host.external_hostname)
self.log.debug("SSH invocation: %s", argv)
return argv
def spawn_expect(self, argv, default_timeout, encoding, extra_ssh_options):
self.log.debug('Starting pexpect ssh session')
if isinstance(argv, str):
argv = [argv]
if extra_ssh_options is None:
extra_ssh_options = []
argv = self._get_ssh_argv() + ['-q'] + extra_ssh_options + argv
return IpaTestExpect(argv, default_timeout, encoding)