Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
if not isinstance(fd, int) and hasattr(fd, 'fileno'):
fd = fd.fileno()
if not isinstance(fd, int):
raise ExceptionPexpect(
"The fd argument is not an int. If this is a command string then maybe you want to use pexpect.spawn.")
try: # make sure fd is a valid file descriptor
os.fstat(fd)
except OSError:
raise ExceptionPexpect("The fd argument is not a valid file descriptor.")
self.args = None
self.command = None
spawn.__init__(self, None, args, timeout, maxread, searchwindowsize, logfile)
self.child_fd = fd
self.own_fd = False
self.closed = False
self.name = '' % fd
def __init__(self, command, lava_timeout, logger=None, cwd=None):
if not lava_timeout or not isinstance(lava_timeout, Timeout):
raise LAVABug("ShellCommand needs a timeout set by the calling Action")
if not logger:
raise LAVABug("ShellCommand needs a logger")
if sys.version_info[0] == 2:
pexpect.spawn.__init__(
self, command,
timeout=lava_timeout.duration,
cwd=cwd,
logfile=ShellLogger(logger),
)
elif sys.version_info[0] == 3:
pexpect.spawn.__init__(
self, command,
timeout=lava_timeout.duration,
cwd=cwd,
logfile=ShellLogger(logger),
encoding='utf-8',
)
self.name = "ShellCommand"
self.logger = logger
# set a default newline character, but allow actions to override as neccessary
def __init__ (self, timeout=30, maxread=2000, searchwindowsize=None,
logfile=None, cwd=None, env=None, ignore_sighup=True, echo=True,
options={}, encoding=None, codec_errors='strict',
debug_command_string=False):
spawn.__init__(self, None, timeout=timeout, maxread=maxread,
searchwindowsize=searchwindowsize, logfile=logfile,
cwd=cwd, env=env, ignore_sighup=ignore_sighup, echo=echo,
encoding=encoding, codec_errors=codec_errors)
self.name = ''
#SUBTLE HACK ALERT! Note that the command that SETS the prompt uses a
#slightly different string than the regular expression to match it. This
#is because when you set the prompt the command will echo back, but we
#don't want to match the echoed command. So if we make the set command
#slightly different than the regex we eliminate the problem. To make the
#set command different we add a backslash in front of $. The $ doesn't
#need to be escaped, but it doesn't hurt and serves to make the set
#prompt command different than the regex.
# used to match the command-line prompt
def __init__(self, channel, *args, **kwargs):
pexpect.spawn.__init__(self, None, *args, **kwargs)
self.channel = channel
self.child_fd = None
self.closed = False
self.name = '' % channel.get_id()
def __init__ (self, timeout=30, maxread=2000, searchwindowsize=None,
logfile=None, cwd=None, env=None, ignore_sighup=True, echo=True,
options={}, encoding=None, codec_errors='strict'):
spawn.__init__(self, None, timeout=timeout, maxread=maxread,
searchwindowsize=searchwindowsize, logfile=logfile,
cwd=cwd, env=env, ignore_sighup=ignore_sighup, echo=echo,
encoding=encoding, codec_errors=codec_errors)
self.name = ''
#SUBTLE HACK ALERT! Note that the command that SETS the prompt uses a
#slightly different string than the regular expression to match it. This
#is because when you set the prompt the command will echo back, but we
#don't want to match the echoed command. So if we make the set command
#slightly different than the regex we eliminate the problem. To make the
#set command different we add a backslash in front of $. The $ doesn't
#need to be escaped, but it doesn't hurt and serves to make the set
#prompt command different than the regex.
# used to match the command-line prompt
try:
import xmltodict
HAS_XMLTODICT = True
except ImportError as e:
HAS_XMLTODICT = False
XMLTODICT_IMPORT_ERR = e
HAS_PEXPECT = False
try:
import pexpect
# echo was added in pexpect 3.3+ which is newer than the RHEL package
# we can only use pexpect for kerb auth if echo is a valid kwarg
# https://github.com/ansible/ansible/issues/43462
if hasattr(pexpect, 'spawn'):
argspec = getargspec(pexpect.spawn.__init__)
if 'echo' in argspec.args:
HAS_PEXPECT = True
except ImportError as e:
pass
# used to try and parse the hostname and detect if IPv6 is being used
try:
import ipaddress
HAS_IPADDRESS = True
except ImportError:
HAS_IPADDRESS = False
display = Display()
class Connection(ConnectionBase):
def __init__(self, sock, *args, **kwargs):
pexpect.spawn.__init__(self, None, *args, **kwargs)
self.child_fd = sock.fileno()
self.closed = False
self.name = '' % self.child_fd
def __init__(self, command, lava_timeout, logger=None, cwd=None):
if not lava_timeout or not isinstance(lava_timeout, Timeout):
raise LAVABug("ShellCommand needs a timeout set by the calling Action")
if not logger:
raise LAVABug("ShellCommand needs a logger")
if sys.version_info[0] == 2:
pexpect.spawn.__init__(
self, command,
timeout=lava_timeout.duration,
cwd=cwd,
logfile=ShellLogger(logger),
)
elif sys.version_info[0] == 3:
pexpect.spawn.__init__(
self, command,
timeout=lava_timeout.duration,
cwd=cwd,
logfile=ShellLogger(logger),
encoding='utf-8',
)
self.name = "ShellCommand"
self.logger = logger
# set a default newline character, but allow actions to override as neccessary
self.linesep = LINE_SEPARATOR
self.lava_timeout = lava_timeout