diff options
Diffstat (limited to 'src/VBox/ValidationKit/testdriver/txsclient.py')
| -rwxr-xr-x | src/VBox/ValidationKit/testdriver/txsclient.py | 2376 |
1 files changed, 2376 insertions, 0 deletions
diff --git a/src/VBox/ValidationKit/testdriver/txsclient.py b/src/VBox/ValidationKit/testdriver/txsclient.py new file mode 100755 index 00000000..343071d6 --- /dev/null +++ b/src/VBox/ValidationKit/testdriver/txsclient.py @@ -0,0 +1,2376 @@ +# -*- coding: utf-8 -*- +# $Id: txsclient.py $ +# pylint: disable=too-many-lines + +""" +Test eXecution Service Client. +""" +__copyright__ = \ +""" +Copyright (C) 2010-2023 Oracle and/or its affiliates. + +This file is part of VirtualBox base platform packages, as +available from https://www.virtualbox.org. + +This program is free software; you can redistribute it and/or +modify it under the terms of the GNU General Public License +as published by the Free Software Foundation, in version 3 of the +License. + +This program is distributed in the hope that it will be useful, but +WITHOUT ANY WARRANTY; without even the implied warranty of +MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU +General Public License for more details. + +You should have received a copy of the GNU General Public License +along with this program; if not, see <https://www.gnu.org/licenses>. + +The contents of this file may alternatively be used under the terms +of the Common Development and Distribution License Version 1.0 +(CDDL), a copy of it is provided in the "COPYING.CDDL" file included +in the VirtualBox distribution, in which case the provisions of the +CDDL are applicable instead of those of the GPL. + +You may elect to license modified versions of this file under the +terms and conditions of either the GPL or the CDDL or both. + +SPDX-License-Identifier: GPL-3.0-only OR CDDL-1.0 +""" +__version__ = "$Revision: 155244 $" + +# Standard Python imports. +import array; +import errno; +import os; +import select; +import socket; +import sys; +import threading; +import time; +import zlib; +import uuid; + +# Validation Kit imports. +from common import utils; +from testdriver import base; +from testdriver import reporter; +from testdriver.base import TdTaskBase; + +# Python 3 hacks: +if sys.version_info[0] >= 3: + long = int; # pylint: disable=redefined-builtin,invalid-name + +# +# Helpers for decoding data received from the TXS. +# These are used both the Session and Transport classes. +# + +def getU32(abData, off): + """Get a U32 field.""" + return abData[off] \ + + abData[off + 1] * 256 \ + + abData[off + 2] * 65536 \ + + abData[off + 3] * 16777216; + +def getSZ(abData, off, sDefault = None): + """ + Get a zero-terminated string field. + Returns sDefault if the string is invalid. + """ + cchStr = getSZLen(abData, off); + if cchStr >= 0: + abStr = abData[off:(off + cchStr)]; + try: + if sys.version_info < (3, 9, 0): + # Removed since Python 3.9. + sStr = abStr.tostring(); # pylint: disable=no-member + else: + sStr = abStr.tobytes(); + return sStr.decode('utf_8'); + except: + reporter.errorXcpt('getSZ(,%u)' % (off)); + return sDefault; + +def getSZLen(abData, off): + """ + Get the length of a zero-terminated string field, in bytes. + Returns -1 if off is beyond the data packet or not properly terminated. + """ + cbData = len(abData); + if off >= cbData: + return -1; + + offCur = off; + while abData[offCur] != 0: + offCur = offCur + 1; + if offCur >= cbData: + return -1; + + return offCur - off; + +def isValidOpcodeEncoding(sOpcode): + """ + Checks if the specified opcode is valid or not. + Returns True on success. + Returns False if it is invalid, details in the log. + """ + sSet1 = "ABCDEFGHIJKLMNOPQRSTUVWXYZ"; + sSet2 = "ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789-_ "; + if len(sOpcode) != 8: + reporter.error("invalid opcode length: %s" % (len(sOpcode))); + return False; + for i in range(0, 1): + if sSet1.find(sOpcode[i]) < 0: + reporter.error("invalid opcode char #%u: %s" % (i, sOpcode)); + return False; + for i in range(2, 7): + if sSet2.find(sOpcode[i]) < 0: + reporter.error("invalid opcode char #%u: %s" % (i, sOpcode)); + return False; + return True; + +# +# Helper for encoding data sent to the TXS. +# + +def u32ToByteArray(u32): + """Encodes the u32 value as a little endian byte (B) array.""" + return array.array('B', + ( u32 % 256, + (u32 // 256) % 256, + (u32 // 65536) % 256, + (u32 // 16777216) % 256) ); + +def escapeString(sString): + """ + Does $ escaping of the string so TXS doesn't try do variable expansion. + """ + return sString.replace('$', '$$'); + + + +class TransportBase(object): + """ + Base class for the transport layer. + """ + + def __init__(self, sCaller): + self.sDbgCreated = '%s: %s' % (utils.getTimePrefix(), sCaller); + self.fDummy = 0; + self.abReadAheadHdr = array.array('B'); + + def toString(self): + """ + Stringify the instance for logging and debugging. + """ + return '<%s: abReadAheadHdr=%s, sDbgCreated=%s>' % (type(self).__name__, self.abReadAheadHdr, self.sDbgCreated); + + def __str__(self): + return self.toString(); + + def cancelConnect(self): + """ + Cancels any pending connect() call. + Returns None; + """ + return None; + + def connect(self, cMsTimeout): + """ + Quietly attempts to connect to the TXS. + + Returns True on success. + Returns False on retryable errors (no logging). + Returns None on fatal errors with details in the log. + + Override this method, don't call super. + """ + _ = cMsTimeout; + return False; + + def disconnect(self, fQuiet = False): + """ + Disconnect from the TXS. + + Returns True. + + Override this method, don't call super. + """ + _ = fQuiet; + return True; + + def sendBytes(self, abBuf, cMsTimeout): + """ + Sends the bytes in the buffer abBuf to the TXS. + + Returns True on success. + Returns False on failure and error details in the log. + + Override this method, don't call super. + + Remarks: len(abBuf) is always a multiple of 16. + """ + _ = abBuf; _ = cMsTimeout; + return False; + + def recvBytes(self, cb, cMsTimeout, fNoDataOk): + """ + Receive cb number of bytes from the TXS. + + Returns the bytes (array('B')) on success. + Returns None on failure and error details in the log. + + Override this method, don't call super. + + Remarks: cb is always a multiple of 16. + """ + _ = cb; _ = cMsTimeout; _ = fNoDataOk; + return None; + + def isConnectionOk(self): + """ + Checks if the connection is OK. + + Returns True if it is. + Returns False if it isn't (caller should call diconnect). + + Override this method, don't call super. + """ + return True; + + def isRecvPending(self, cMsTimeout = 0): + """ + Checks if there is incoming bytes, optionally waiting cMsTimeout + milliseconds for something to arrive. + + Returns True if there is, False if there isn't. + + Override this method, don't call super. + """ + _ = cMsTimeout; + return False; + + def sendMsgInt(self, sOpcode, cMsTimeout, abPayload = array.array('B')): + """ + Sends a message (opcode + encoded payload). + + Returns True on success. + Returns False on failure and error details in the log. + """ + # Fix + check the opcode. + if len(sOpcode) < 2: + reporter.fatal('sendMsgInt: invalid opcode length: %d (\"%s\")' % (len(sOpcode), sOpcode)); + return False; + sOpcode = sOpcode.ljust(8); + if not isValidOpcodeEncoding(sOpcode): + reporter.fatal('sendMsgInt: invalid opcode encoding: \"%s\"' % (sOpcode)); + return False; + + # Start construct the message. + cbMsg = 16 + len(abPayload); + abMsg = array.array('B'); + abMsg.extend(u32ToByteArray(cbMsg)); + abMsg.extend((0, 0, 0, 0)); # uCrc32 + try: + abMsg.extend(array.array('B', \ + ( ord(sOpcode[0]), \ + ord(sOpcode[1]), \ + ord(sOpcode[2]), \ + ord(sOpcode[3]), \ + ord(sOpcode[4]), \ + ord(sOpcode[5]), \ + ord(sOpcode[6]), \ + ord(sOpcode[7]) ) ) ); + if abPayload: + abMsg.extend(abPayload); + except: + reporter.fatalXcpt('sendMsgInt: packing problem...'); + return False; + + # checksum it, padd it and send it off. + uCrc32 = zlib.crc32(abMsg[8:]); + abMsg[4:8] = u32ToByteArray(uCrc32); + + while len(abMsg) % 16: + abMsg.append(0); + + reporter.log2('sendMsgInt: op=%s len=%d timeout=%d' % (sOpcode, len(abMsg), cMsTimeout)); + return self.sendBytes(abMsg, cMsTimeout); + + def recvMsg(self, cMsTimeout, fNoDataOk = False): + """ + Receives a message from the TXS. + + Returns the message three-tuple: length, opcode, payload. + Returns (None, None, None) on failure and error details in the log. + """ + + # Read the header. + if self.abReadAheadHdr: + assert(len(self.abReadAheadHdr) == 16); + abHdr = self.abReadAheadHdr; + self.abReadAheadHdr = array.array('B'); + else: + abHdr = self.recvBytes(16, cMsTimeout, fNoDataOk); # (virtual method) # pylint: disable=assignment-from-none + if abHdr is None: + return (None, None, None); + if len(abHdr) != 16: + reporter.fatal('recvBytes(16) returns %d bytes!' % (len(abHdr))); + return (None, None, None); + + # Unpack and validate the header. + cbMsg = getU32(abHdr, 0); + uCrc32 = getU32(abHdr, 4); + + if sys.version_info < (3, 9, 0): + # Removed since Python 3.9. + sOpcode = abHdr[8:16].tostring(); # pylint: disable=no-member + else: + sOpcode = abHdr[8:16].tobytes(); + sOpcode = sOpcode.decode('ascii'); + + if cbMsg < 16: + reporter.fatal('recvMsg: message length is out of range: %s (min 16 bytes)' % (cbMsg)); + return (None, None, None); + if cbMsg > 1024*1024: + reporter.fatal('recvMsg: message length is out of range: %s (max 1MB)' % (cbMsg)); + return (None, None, None); + if not isValidOpcodeEncoding(sOpcode): + reporter.fatal('recvMsg: invalid opcode \"%s\"' % (sOpcode)); + return (None, None, None); + + # Get the payload (if any), dropping the padding. + abPayload = array.array('B'); + if cbMsg > 16: + if cbMsg % 16: + cbPadding = 16 - (cbMsg % 16); + else: + cbPadding = 0; + abPayload = self.recvBytes(cbMsg - 16 + cbPadding, cMsTimeout, False); # pylint: disable=assignment-from-none + if abPayload is None: + self.abReadAheadHdr = abHdr; + if not fNoDataOk : + reporter.log('recvMsg: failed to recv payload bytes!'); + return (None, None, None); + + while cbPadding > 0: + abPayload.pop(); + cbPadding = cbPadding - 1; + + # Check the CRC-32. + if uCrc32 != 0: + uActualCrc32 = zlib.crc32(abHdr[8:]); + if cbMsg > 16: + uActualCrc32 = zlib.crc32(abPayload, uActualCrc32); + uActualCrc32 = uActualCrc32 & 0xffffffff; + if uCrc32 != uActualCrc32: + reporter.fatal('recvMsg: crc error: expected %s, got %s' % (hex(uCrc32), hex(uActualCrc32))); + return (None, None, None); + + reporter.log2('recvMsg: op=%s len=%d' % (sOpcode, len(abPayload))); + return (cbMsg, sOpcode, abPayload); + + def sendMsg(self, sOpcode, cMsTimeout, aoPayload = ()): + """ + Sends a message (opcode + payload tuple). + + Returns True on success. + Returns False on failure and error details in the log. + Returns None if you pass the incorrectly typed parameters. + """ + # Encode the payload. + abPayload = array.array('B'); + for o in aoPayload: + try: + if utils.isString(o): + if sys.version_info[0] >= 3: + abPayload.extend(o.encode('utf_8')); + else: + # the primitive approach... + sUtf8 = o.encode('utf_8'); + for ch in sUtf8: + abPayload.append(ord(ch)) + abPayload.append(0); + elif isinstance(o, (long, int)): + if o < 0 or o > 0xffffffff: + reporter.fatal('sendMsg: uint32_t payload is out of range: %s' % (hex(o))); + return None; + abPayload.extend(u32ToByteArray(o)); + elif isinstance(o, array.array): + abPayload.extend(o); + else: + reporter.fatal('sendMsg: unexpected payload type: %s (%s) (aoPayload=%s)' % (type(o), o, aoPayload)); + return None; + except: + reporter.fatalXcpt('sendMsg: screwed up the encoding code...'); + return None; + return self.sendMsgInt(sOpcode, cMsTimeout, abPayload); + + +class Session(TdTaskBase): + """ + A Test eXecution Service (TXS) client session. + """ + + def __init__(self, oTransport, cMsTimeout, cMsIdleFudge, fTryConnect = False, fnProcessEvents = None): + """ + Construct a TXS session. + + This starts by connecting to the TXS and will enter the signalled state + when connected or the timeout has been reached. + """ + TdTaskBase.__init__(self, utils.getCallerName(), fnProcessEvents); + self.oTransport = oTransport; + self.sStatus = ""; + self.cMsTimeout = 0; + self.fErr = True; # Whether to report errors as error. + self.msStart = 0; + self.oThread = None; + self.fnTask = self.taskDummy; + self.aTaskArgs = None; + self.oTaskRc = None; + self.t3oReply = (None, None, None); + self.fScrewedUpMsgState = False; + self.fTryConnect = fTryConnect; + + if not self.startTask(cMsTimeout, False, "connecting", self.taskConnect, (cMsIdleFudge,)): + raise base.GenError("startTask failed"); + + def __del__(self): + """Make sure to cancel the task when deleted.""" + self.cancelTask(); + + def toString(self): + return '<%s fnTask=%s, aTaskArgs=%s, sStatus=%s, oTaskRc=%s, cMsTimeout=%s,' \ + ' msStart=%s, fTryConnect=%s, fErr=%s, fScrewedUpMsgState=%s, t3oReply=%s oTransport=%s, oThread=%s>' \ + % (TdTaskBase.toString(self), self.fnTask, self.aTaskArgs, self.sStatus, self.oTaskRc, self.cMsTimeout, + self.msStart, self.fTryConnect, self.fErr, self.fScrewedUpMsgState, self.t3oReply, self.oTransport, self.oThread); + + def taskDummy(self): + """Place holder to catch broken state handling.""" + raise Exception(); + + def startTask(self, cMsTimeout, fIgnoreErrors, sStatus, fnTask, aArgs = ()): + """ + Kicks of a new task. + + cMsTimeout: The task timeout in milliseconds. Values less than + 500 ms will be adjusted to 500 ms. This means it is + OK to use negative value. + sStatus: The task status. + fnTask: The method that'll execute the task. + aArgs: Arguments to pass to fnTask. + + Returns True on success, False + error in log on failure. + """ + if not self.cancelTask(): + reporter.maybeErr(not fIgnoreErrors, 'txsclient.Session.startTask: failed to cancel previous task.'); + return False; + + # Change status and make sure we're the + self.lockTask(); + if self.sStatus != "": + self.unlockTask(); + reporter.maybeErr(not fIgnoreErrors, 'txsclient.Session.startTask: race.'); + return False; + self.sStatus = "setup"; + self.oTaskRc = None; + self.t3oReply = (None, None, None); + self.resetTaskLocked(); + self.unlockTask(); + + self.cMsTimeout = max(cMsTimeout, 500); + self.fErr = not fIgnoreErrors; + self.fnTask = fnTask; + self.aTaskArgs = aArgs; + self.oThread = threading.Thread(target=self.taskThread, args=(), name=('TXS-%s' % (sStatus))); + self.oThread.setDaemon(True); # pylint: disable=deprecated-method + self.msStart = base.timestampMilli(); + + self.lockTask(); + self.sStatus = sStatus; + self.unlockTask(); + self.oThread.start(); + + return True; + + def cancelTask(self, fSync = True): + """ + Attempts to cancel any pending tasks. + Returns success indicator (True/False). + """ + self.lockTask(); + + if self.sStatus == "": + self.unlockTask(); + return True; + if self.sStatus == "setup": + self.unlockTask(); + return False; + if self.sStatus == "cancelled": + self.unlockTask(); + return False; + + reporter.log('txsclient: cancelling "%s"...' % (self.sStatus)); + if self.sStatus == 'connecting': + self.oTransport.cancelConnect(); + + self.sStatus = "cancelled"; + oThread = self.oThread; + self.unlockTask(); + + if not fSync: + return False; + + oThread.join(61.0); + + if sys.version_info < (3, 9, 0): + # Removed since Python 3.9. + return oThread.isAlive(); # pylint: disable=no-member + return oThread.is_alive(); + + def taskThread(self): + """ + The task thread function. + This does some housekeeping activities around the real task method call. + """ + if not self.isCancelled(): + try: + fnTask = self.fnTask; + oTaskRc = fnTask(*self.aTaskArgs); + except: + reporter.fatalXcpt('taskThread', 15); + oTaskRc = None; + else: + reporter.log('taskThread: cancelled already'); + + self.lockTask(); + + reporter.log('taskThread: signalling task with status "%s", oTaskRc=%s' % (self.sStatus, oTaskRc)); + self.oTaskRc = oTaskRc; + self.oThread = None; + self.sStatus = ''; + self.signalTaskLocked(); + + self.unlockTask(); + return None; + + def isCancelled(self): + """Internal method for checking if the task has been cancelled.""" + self.lockTask(); + sStatus = self.sStatus; + self.unlockTask(); + if sStatus == "cancelled": + return True; + return False; + + def hasTimedOut(self): + """Internal method for checking if the task has timed out or not.""" + cMsLeft = self.getMsLeft(); + if cMsLeft <= 0: + return True; + return False; + + def getMsLeft(self, cMsMin = 0, cMsMax = -1): + """Gets the time left until the timeout.""" + cMsElapsed = base.timestampMilli() - self.msStart; + if cMsElapsed < 0: + return cMsMin; + cMsLeft = self.cMsTimeout - cMsElapsed; + if cMsLeft <= cMsMin: + return cMsMin; + if cMsLeft > cMsMax > 0: + return cMsMax + return cMsLeft; + + def recvReply(self, cMsTimeout = None, fNoDataOk = False): + """ + Wrapper for TransportBase.recvMsg that stashes the response away + so the client can inspect it later on. + """ + if cMsTimeout is None: + cMsTimeout = self.getMsLeft(500); + cbMsg, sOpcode, abPayload = self.oTransport.recvMsg(cMsTimeout, fNoDataOk); + self.lockTask(); + self.t3oReply = (cbMsg, sOpcode, abPayload); + self.unlockTask(); + return (cbMsg, sOpcode, abPayload); + + def recvAck(self, fNoDataOk = False): + """ + Receives an ACK or error response from the TXS. + + Returns True on success. + Returns False on timeout or transport error. + Returns (sOpcode, sDetails) tuple on failure. The opcode is stripped + and there are always details of some sort or another. + """ + cbMsg, sOpcode, abPayload = self.recvReply(None, fNoDataOk); + if cbMsg is None: + return False; + sOpcode = sOpcode.strip() + if sOpcode == "ACK": + return True; + return (sOpcode, getSZ(abPayload, 0, sOpcode)); + + def recvAckLogged(self, sCommand, fNoDataOk = False): + """ + Wrapper for recvAck and logging. + Returns True on success (ACK). + Returns False on time, transport error and errors signalled by TXS. + """ + rc = self.recvAck(fNoDataOk); + if rc is not True and not fNoDataOk: + if rc is False: + reporter.maybeErr(self.fErr, 'recvAckLogged: %s transport error' % (sCommand)); + else: + reporter.maybeErr(self.fErr, 'recvAckLogged: %s response was %s: %s' % (sCommand, rc[0], rc[1])); + rc = False; + return rc; + + def recvTrueFalse(self, sCommand): + """ + Receives a TRUE/FALSE response from the TXS. + Returns True on TRUE, False on FALSE and None on error/other (logged). + """ + cbMsg, sOpcode, abPayload = self.recvReply(); + if cbMsg is None: + reporter.maybeErr(self.fErr, 'recvAckLogged: %s transport error' % (sCommand)); + return None; + + sOpcode = sOpcode.strip() + if sOpcode == "TRUE": + return True; + if sOpcode == "FALSE": + return False; + reporter.maybeErr(self.fErr, 'recvAckLogged: %s response was %s: %s' % (sCommand, sOpcode, getSZ(abPayload, 0, sOpcode))); + return None; + + def sendMsg(self, sOpcode, aoPayload = (), cMsTimeout = None): + """ + Wrapper for TransportBase.sendMsg that inserts the correct timeout. + """ + if cMsTimeout is None: + cMsTimeout = self.getMsLeft(500); + return self.oTransport.sendMsg(sOpcode, cMsTimeout, aoPayload); + + def asyncToSync(self, fnAsync, *aArgs): + """ + Wraps an asynchronous task into a synchronous operation. + + Returns False on failure, task return status on success. + """ + rc = fnAsync(*aArgs); + if rc is False: + reporter.log2('asyncToSync(%s): returns False (#1)' % (fnAsync)); + return rc; + + rc = self.waitForTask(self.cMsTimeout + 5000); + if rc is False: + reporter.maybeErr(self.fErr, 'asyncToSync: waitForTask (timeout %d) failed...' % (self.cMsTimeout,)); + self.cancelTask(); + #reporter.log2('asyncToSync(%s): returns False (#2)' % (fnAsync, rc)); + return False; + + rc = self.getResult(); + #reporter.log2('asyncToSync(%s): returns %s' % (fnAsync, rc)); + return rc; + + # + # Connection tasks. + # + + def taskConnect(self, cMsIdleFudge): + """Tries to connect to the TXS""" + while not self.isCancelled(): + reporter.log2('taskConnect: connecting ...'); + rc = self.oTransport.connect(self.getMsLeft(500)); + if rc is True: + reporter.log('taskConnect: succeeded'); + return self.taskGreet(cMsIdleFudge); + if rc is None: + reporter.log2('taskConnect: unable to connect'); + return None; + if self.hasTimedOut(): + reporter.log2('taskConnect: timed out'); + if not self.fTryConnect: + reporter.maybeErr(self.fErr, 'taskConnect: timed out'); + return False; + time.sleep(self.getMsLeft(1, 1000) / 1000.0); + if not self.fTryConnect: + reporter.maybeErr(self.fErr, 'taskConnect: cancelled'); + return False; + + def taskGreet(self, cMsIdleFudge): + """Greets the TXS""" + rc = self.sendMsg("HOWDY", ()); + if rc is True: + rc = self.recvAckLogged("HOWDY", self.fTryConnect); + if rc is True: + while cMsIdleFudge > 0: + cMsIdleFudge -= 1000; + time.sleep(1); + else: + self.oTransport.disconnect(self.fTryConnect); + return rc; + + def taskBye(self): + """Says goodbye to the TXS""" + rc = self.sendMsg("BYE"); + if rc is True: + rc = self.recvAckLogged("BYE"); + self.oTransport.disconnect(); + return rc; + + def taskVer(self): + """Requests version information from TXS""" + rc = self.sendMsg("VER"); + if rc is True: + rc = False; + cbMsg, sOpcode, abPayload = self.recvReply(); + if cbMsg is not None: + sOpcode = sOpcode.strip(); + if sOpcode == "ACK VER": + sVer = getSZ(abPayload, 0); + if sVer is not None: + rc = sVer; + else: + reporter.maybeErr(self.fErr, 'taskVer got a bad reply: %s' % (sOpcode,)); + else: + reporter.maybeErr(self.fErr, 'taskVer got 3xNone from recvReply.'); + return rc; + + def taskUuid(self): + """Gets the TXS UUID""" + rc = self.sendMsg("UUID"); + if rc is True: + rc = False; + cbMsg, sOpcode, abPayload = self.recvReply(); + if cbMsg is not None: + sOpcode = sOpcode.strip() + if sOpcode == "ACK UUID": + sUuid = getSZ(abPayload, 0); + if sUuid is not None: + sUuid = '{%s}' % (sUuid,) + try: + _ = uuid.UUID(sUuid); + rc = sUuid; + except: + reporter.errorXcpt('taskUuid got an invalid UUID string %s' % (sUuid,)); + else: + reporter.maybeErr(self.fErr, 'taskUuid did not get a UUID string.'); + else: + reporter.maybeErr(self.fErr, 'taskUuid got a bad reply: %s' % (sOpcode,)); + else: + reporter.maybeErr(self.fErr, 'taskUuid got 3xNone from recvReply.'); + return rc; + + # + # Process task + # pylint: disable=missing-docstring + # + + def taskExecEx(self, sExecName, fFlags, asArgs, asAddEnv, oStdIn, oStdOut, oStdErr, oTestPipe, sAsUser): # pylint: disable=too-many-arguments,too-many-locals,too-many-statements,line-too-long + # Construct the payload. + aoPayload = [long(fFlags), '%s' % (sExecName), long(len(asArgs))]; + for sArg in asArgs: + aoPayload.append('%s' % (sArg)); + aoPayload.append(long(len(asAddEnv))); + for sPutEnv in asAddEnv: + aoPayload.append('%s' % (sPutEnv)); + for o in (oStdIn, oStdOut, oStdErr, oTestPipe): + if utils.isString(o): + aoPayload.append(o); + elif o is not None: + aoPayload.append('|'); + o.uTxsClientCrc32 = zlib.crc32(b''); + else: + aoPayload.append(''); + aoPayload.append('%s' % (sAsUser)); + aoPayload.append(long(self.cMsTimeout)); + + # Kick of the EXEC command. + rc = self.sendMsg('EXEC', aoPayload) + if rc is True: + rc = self.recvAckLogged('EXEC'); + if rc is True: + # Loop till the process completes, feed input to the TXS and + # receive output from it. + sFailure = ""; + msPendingInputReply = None; + cbMsg, sOpcode, abPayload = (None, None, None); + while True: + # Pending input? + if msPendingInputReply is None \ + and oStdIn is not None \ + and not utils.isString(oStdIn): + try: + sInput = oStdIn.read(65536); + except: + reporter.errorXcpt('read standard in'); + sFailure = 'exception reading stdin'; + rc = None; + break; + if sInput: + # Convert to a byte array before handing it of to sendMsg or the string + # will get some zero termination added breaking the CRC (and injecting + # unwanted bytes). + abInput = array.array('B', sInput.encode('utf-8')); + oStdIn.uTxsClientCrc32 = zlib.crc32(abInput, oStdIn.uTxsClientCrc32); + rc = self.sendMsg('STDIN', (long(oStdIn.uTxsClientCrc32 & 0xffffffff), abInput)); + if rc is not True: + sFailure = 'sendMsg failure'; + break; + msPendingInputReply = base.timestampMilli(); + continue; + + rc = self.sendMsg('STDINEOS'); + oStdIn = None; + if rc is not True: + sFailure = 'sendMsg failure'; + break; + msPendingInputReply = base.timestampMilli(); + + # Wait for input (500 ms timeout). + if cbMsg is None: + cbMsg, sOpcode, abPayload = self.recvReply(cMsTimeout=500, fNoDataOk=True); + if cbMsg is None: + # Check for time out before restarting the loop. + # Note! Only doing timeout checking here does mean that + # the TXS may prevent us from timing out by + # flooding us with data. This is unlikely though. + if self.hasTimedOut() \ + and ( msPendingInputReply is None \ + or base.timestampMilli() - msPendingInputReply > 30000): + reporter.maybeErr(self.fErr, 'taskExecEx: timed out'); + sFailure = 'timeout'; + rc = None; + break; + # Check that the connection is OK. + if not self.oTransport.isConnectionOk(): + self.oTransport.disconnect(); + sFailure = 'disconnected'; + rc = False; + break; + continue; + + # Handle the response. + sOpcode = sOpcode.rstrip(); + if sOpcode == 'STDOUT': + oOut = oStdOut; + elif sOpcode == 'STDERR': + oOut = oStdErr; + elif sOpcode == 'TESTPIPE': + oOut = oTestPipe; + else: + oOut = None; + if oOut is not None: + # Output from the process. + if len(abPayload) < 4: + sFailure = 'malformed output packet (%s, %u bytes)' % (sOpcode, cbMsg); + reporter.maybeErr(self.fErr, 'taskExecEx: %s' % (sFailure)); + rc = None; + break; + uStreamCrc32 = getU32(abPayload, 0); + oOut.uTxsClientCrc32 = zlib.crc32(abPayload[4:], oOut.uTxsClientCrc32); + if uStreamCrc32 != (oOut.uTxsClientCrc32 & 0xffffffff): + sFailure = 'crc error - mine=%#x their=%#x (%s, %u bytes)' \ + % (oOut.uTxsClientCrc32 & 0xffffffff, uStreamCrc32, sOpcode, cbMsg); + reporter.maybeErr(self.fErr, 'taskExecEx: %s' % (sFailure)); + rc = None; + break; + try: + oOut.write(abPayload[4:]); + except: + sFailure = 'exception writing %s' % (sOpcode); + reporter.errorXcpt('taskExecEx: %s' % (sFailure)); + rc = None; + break; + elif sOpcode == 'STDINIGN' and msPendingInputReply is not None: + # Standard input is ignored. Ignore this condition for now. + msPendingInputReply = None; + reporter.log('taskExecEx: Standard input is ignored... why?'); + del oStdIn.uTxsClientCrc32; + oStdIn = '/dev/null'; + elif sOpcode in ('STDINMEM', 'STDINBAD', 'STDINCRC',)\ + and msPendingInputReply is not None: + # TXS STDIN error, abort. + # TODO: STDINMEM - consider undoing the previous stdin read and try resubmitt it. + msPendingInputReply = None; + sFailure = 'TXS is out of memory for std input buffering'; + reporter.maybeErr(self.fErr, 'taskExecEx: %s' % (sFailure)); + rc = None; + break; + elif sOpcode == 'ACK' and msPendingInputReply is not None: + msPendingInputReply = None; + elif sOpcode.startswith('PROC '): + # Process status message, handle it outside the loop. + rc = True; + break; + else: + sFailure = 'Unexpected opcode %s' % (sOpcode); + reporter.maybeErr(self.fErr, 'taskExecEx: %s' % (sFailure)); + rc = None; + break; + # Clear the message. + cbMsg, sOpcode, abPayload = (None, None, None); + + # If we sent an STDIN packet and didn't get a reply yet, we'll give + # TXS some 5 seconds to reply to this. If we don't wait here we'll + # get screwed later on if we mix it up with the reply to some other + # command. Hackish. + if msPendingInputReply is not None: + cbMsg2, sOpcode2, abPayload2 = self.oTransport.recvMsg(5000); + if cbMsg2 is not None: + reporter.log('taskExecEx: Out of order STDIN, got reply: %s, %s, %s [ignored]' + % (cbMsg2, sOpcode2, abPayload2)); + msPendingInputReply = None; + else: + reporter.maybeErr(self.fErr, 'taskExecEx: Pending STDIN, no reply after 5 secs!'); + self.fScrewedUpMsgState = True; + + # Parse the exit status (True), abort (None) or do nothing (False). + if rc is True: + if sOpcode == 'PROC OK': + pass; + else: + rc = False; + # Do proper parsing some other day if needed: + # PROC TOK, PROC TOA, PROC DWN, PROC DOO, + # PROC NOK + rc, PROC SIG + sig, PROC ABD, FAILED. + if sOpcode == 'PROC DOO': + reporter.log('taskExecEx: PROC DOO[FUS]: %s' % (abPayload,)); + elif sOpcode.startswith('PROC NOK'): + reporter.log('taskExecEx: PROC NOK: rcExit=%s' % (abPayload,)); + elif abPayload and sOpcode.startswith('PROC '): + reporter.log('taskExecEx: %s payload=%s' % (sOpcode, abPayload,)); + + else: + if rc is None: + # Abort it. + reporter.log('taskExecEx: sending ABORT...'); + rc = self.sendMsg('ABORT'); + while rc is True: + cbMsg, sOpcode, abPayload = self.oTransport.recvMsg(30000); + if cbMsg is None: + reporter.maybeErr(self.fErr, 'taskExecEx: Pending ABORT, no reply after 30 secs!') + self.fScrewedUpMsgState = True; + break; + if sOpcode.startswith('PROC '): + reporter.log('taskExecEx: ABORT reply: %s, %s, %s [ignored]' % (cbMsg, sOpcode, abPayload)); + break; + reporter.log('taskExecEx: ABORT in process, ignoring reply: %s, %s, %s' % (cbMsg, sOpcode, abPayload)); + # Check that the connection is OK before looping. + if not self.oTransport.isConnectionOk(): + self.oTransport.disconnect(); + break; + + # Fake response with the reason why we quit. + if sFailure is not None: + self.t3oReply = (0, 'EXECFAIL', sFailure); + rc = None; + else: + rc = None; + + # Cleanup. + for o in (oStdIn, oStdOut, oStdErr, oTestPipe): + if o is not None and not utils.isString(o): + del o.uTxsClientCrc32; # pylint: disable=maybe-no-member + # Make sure all files are closed + o.close(); # pylint: disable=maybe-no-member + reporter.log('taskExecEx: returns %s' % (rc)); + return rc; + + # + # Admin tasks + # + + def hlpRebootShutdownWaitForAck(self, sCmd): + """Wait for reboot/shutodwn ACK.""" + rc = self.recvAckLogged(sCmd); + if rc is True: + # poll a little while for server to disconnect. + uMsStart = base.timestampMilli(); + while self.oTransport.isConnectionOk() \ + and base.timestampMilli() - uMsStart >= 5000: + if self.oTransport.isRecvPending(min(500, self.getMsLeft())): + break; + self.oTransport.disconnect(); + return rc; + + def taskReboot(self): + rc = self.sendMsg('REBOOT'); + if rc is True: + rc = self.hlpRebootShutdownWaitForAck('REBOOT'); + return rc; + + def taskShutdown(self): + rc = self.sendMsg('SHUTDOWN'); + if rc is True: + rc = self.hlpRebootShutdownWaitForAck('SHUTDOWN'); + return rc; + + # + # CD/DVD control tasks. + # + + ## TODO + + # + # File system tasks + # + + def taskMkDir(self, sRemoteDir, fMode): + rc = self.sendMsg('MKDIR', (fMode, sRemoteDir)); + if rc is True: + rc = self.recvAckLogged('MKDIR'); + return rc; + + def taskMkDirPath(self, sRemoteDir, fMode): + rc = self.sendMsg('MKDRPATH', (fMode, sRemoteDir)); + if rc is True: + rc = self.recvAckLogged('MKDRPATH'); + return rc; + + def taskMkSymlink(self, sLinkTarget, sLink): + rc = self.sendMsg('MKSYMLNK', (sLinkTarget, sLink)); + if rc is True: + rc = self.recvAckLogged('MKSYMLNK'); + return rc; + + def taskRmDir(self, sRemoteDir): + rc = self.sendMsg('RMDIR', (sRemoteDir,)); + if rc is True: + rc = self.recvAckLogged('RMDIR'); + return rc; + + def taskRmFile(self, sRemoteFile): + rc = self.sendMsg('RMFILE', (sRemoteFile,)); + if rc is True: + rc = self.recvAckLogged('RMFILE'); + return rc; + + def taskRmSymlink(self, sRemoteSymlink): + rc = self.sendMsg('RMSYMLNK', (sRemoteSymlink,)); + if rc is True: + rc = self.recvAckLogged('RMSYMLNK'); + return rc; + + def taskRmTree(self, sRemoteTree): + rc = self.sendMsg('RMTREE', (sRemoteTree,)); + if rc is True: + rc = self.recvAckLogged('RMTREE'); + return rc; + + def taskChMod(self, sRemotePath, fMode): + rc = self.sendMsg('CHMOD', (int(fMode), sRemotePath,)); + if rc is True: + rc = self.recvAckLogged('CHMOD'); + return rc; + + def taskChOwn(self, sRemotePath, idUser, idGroup): + rc = self.sendMsg('CHOWN', (int(idUser), int(idGroup), sRemotePath,)); + if rc is True: + rc = self.recvAckLogged('CHOWN'); + return rc; + + def taskIsDir(self, sRemoteDir): + rc = self.sendMsg('ISDIR', (sRemoteDir,)); + if rc is True: + rc = self.recvTrueFalse('ISDIR'); + return rc; + + def taskIsFile(self, sRemoteFile): + rc = self.sendMsg('ISFILE', (sRemoteFile,)); + if rc is True: + rc = self.recvTrueFalse('ISFILE'); + return rc; + + def taskIsSymlink(self, sRemoteSymlink): + rc = self.sendMsg('ISSYMLNK', (sRemoteSymlink,)); + if rc is True: + rc = self.recvTrueFalse('ISSYMLNK'); + return rc; + + #def "STAT " + #def "LSTAT " + #def "LIST " + + def taskCopyFile(self, sSrcFile, sDstFile, fMode, fFallbackOkay): + """ Copies a file within the remote from source to destination. """ + _ = fFallbackOkay; # Not used yet. + # Note: If fMode is set to 0, it's up to the target OS' implementation with + # what a file mode the destination file gets created (i.e. via umask). + rc = self.sendMsg('CPFILE', (int(fMode), sSrcFile, sDstFile,)); + if rc is True: + rc = self.recvAckLogged('CPFILE'); + return rc; + + def taskUploadFile(self, sLocalFile, sRemoteFile, fMode, fFallbackOkay): + # + # Open the local file (make sure it exist before bothering TXS) and + # tell TXS that we want to upload a file. + # + try: + oLocalFile = utils.openNoInherit(sLocalFile, 'rb'); + except: + reporter.errorXcpt('taskUpload: failed to open "%s"' % (sLocalFile)); + return False; + + # Common cause with taskUploadStr + rc = self.taskUploadCommon(oLocalFile, sRemoteFile, fMode, fFallbackOkay); + + # Cleanup. + oLocalFile.close(); + return rc; + + def taskUploadString(self, sContent, sRemoteFile, fMode, fFallbackOkay): + # Wrap sContent in a file like class. + class InStringFile(object): # pylint: disable=too-few-public-methods + def __init__(self, sContent): + self.sContent = sContent; + self.off = 0; + + def read(self, cbMax): + cbLeft = len(self.sContent) - self.off; + if cbLeft == 0: + return ""; + if cbLeft <= cbMax: + sRet = self.sContent[self.off:(self.off + cbLeft)]; + else: + sRet = self.sContent[self.off:(self.off + cbMax)]; + self.off = self.off + len(sRet); + return sRet; + + oLocalString = InStringFile(sContent); + return self.taskUploadCommon(oLocalString, sRemoteFile, fMode, fFallbackOkay); + + def taskUploadCommon(self, oLocalFile, sRemoteFile, fMode, fFallbackOkay): + """Common worker used by taskUploadFile and taskUploadString.""" + # + # Command + ACK. + # + # Only used the new PUT2FILE command if we've got a non-zero mode mask. + # Fall back on the old command if the new one is not known by the TXS. + # + if fMode == 0: + rc = self.sendMsg('PUT FILE', (sRemoteFile,)); + if rc is True: + rc = self.recvAckLogged('PUT FILE'); + else: + rc = self.sendMsg('PUT2FILE', (fMode, sRemoteFile)); + if rc is True: + rc = self.recvAck(); + if rc is False: + reporter.maybeErr(self.fErr, 'recvAckLogged: PUT2FILE transport error'); + elif rc is not True: + if rc[0] == 'UNKNOWN' and fFallbackOkay: + # Fallback: + rc = self.sendMsg('PUT FILE', (sRemoteFile,)); + if rc is True: + rc = self.recvAckLogged('PUT FILE'); + else: + reporter.maybeErr(self.fErr, 'recvAckLogged: PUT2FILE response was %s: %s' % (rc[0], rc[1],)); + rc = False; + if rc is True: + # + # Push data packets until eof. + # + uMyCrc32 = zlib.crc32(b''); + while True: + # Read up to 64 KB of data. + try: + sRaw = oLocalFile.read(65536); + except: + rc = None; + break; + + # Convert to array - this is silly! + abBuf = array.array('B'); + if utils.isString(sRaw): + for i, _ in enumerate(sRaw): + abBuf.append(ord(sRaw[i])); + else: + abBuf.extend(sRaw); + sRaw = None; + + # Update the file stream CRC and send it off. + uMyCrc32 = zlib.crc32(abBuf, uMyCrc32); + if not abBuf: + rc = self.sendMsg('DATA EOF', (long(uMyCrc32 & 0xffffffff), )); + else: + rc = self.sendMsg('DATA ', (long(uMyCrc32 & 0xffffffff), abBuf)); + if rc is False: + break; + + # Wait for the reply. + rc = self.recvAck(); + if rc is not True: + if rc is False: + reporter.maybeErr(self.fErr, 'taskUpload: transport error waiting for ACK'); + else: + reporter.maybeErr(self.fErr, 'taskUpload: DATA response was %s: %s' % (rc[0], rc[1])); + rc = False; + break; + + # EOF? + if not abBuf: + break; + + # Send ABORT on ACK and I/O errors. + if rc is None: + rc = self.sendMsg('ABORT'); + if rc is True: + self.recvAckLogged('ABORT'); + rc = False; + return rc; + + def taskDownloadFile(self, sRemoteFile, sLocalFile): + try: + oLocalFile = utils.openNoInherit(sLocalFile, 'wb'); + except: + reporter.errorXcpt('taskDownload: failed to open "%s"' % (sLocalFile)); + return False; + + rc = self.taskDownloadCommon(sRemoteFile, oLocalFile); + + oLocalFile.close(); + if rc is False: + try: + os.remove(sLocalFile); + except: + reporter.errorXcpt(); + return rc; + + def taskDownloadString(self, sRemoteFile, sEncoding = 'utf-8', fIgnoreEncodingErrors = True): + # Wrap sContent in a file like class. + class OutStringFile(object): # pylint: disable=too-few-public-methods + def __init__(self): + self.asContent = []; + + def write(self, sBuf): + self.asContent.append(sBuf); + return None; + + oLocalString = OutStringFile(); + rc = self.taskDownloadCommon(sRemoteFile, oLocalString); + if rc is True: + rc = ''; + for sBuf in oLocalString.asContent: + if hasattr(sBuf, 'decode'): + rc += sBuf.decode(sEncoding, 'ignore' if fIgnoreEncodingErrors else 'strict'); + else: + rc += sBuf; + return rc; + + def taskDownloadCommon(self, sRemoteFile, oLocalFile): + """Common worker for taskDownloadFile and taskDownloadString.""" + rc = self.sendMsg('GET FILE', (sRemoteFile,)) + if rc is True: + # + # Process data packets until eof. + # + uMyCrc32 = zlib.crc32(b''); + while rc is True: + cbMsg, sOpcode, abPayload = self.recvReply(); + if cbMsg is None: + reporter.maybeErr(self.fErr, 'taskDownload got 3xNone from recvReply.'); + rc = None; + break; + + # Validate. + sOpcode = sOpcode.rstrip(); + if sOpcode not in ('DATA', 'DATA EOF',): + reporter.maybeErr(self.fErr, 'taskDownload got a error reply: opcode="%s" details="%s"' + % (sOpcode, getSZ(abPayload, 0, "None"))); + rc = False; + break; + if sOpcode == 'DATA' and len(abPayload) < 4: + reporter.maybeErr(self.fErr, 'taskDownload got a bad DATA packet: len=%u' % (len(abPayload))); + rc = None; + break; + if sOpcode == 'DATA EOF' and len(abPayload) != 4: + reporter.maybeErr(self.fErr, 'taskDownload got a bad EOF packet: len=%u' % (len(abPayload))); + rc = None; + break; + + # Check the CRC (common for both packets). + uCrc32 = getU32(abPayload, 0); + if sOpcode == 'DATA': + uMyCrc32 = zlib.crc32(abPayload[4:], uMyCrc32); + if uCrc32 != (uMyCrc32 & 0xffffffff): + reporter.maybeErr(self.fErr, 'taskDownload got a bad CRC: mycrc=%s remotecrc=%s' + % (hex(uMyCrc32), hex(uCrc32))); + rc = None; + break; + if sOpcode == 'DATA EOF': + rc = self.sendMsg('ACK'); + break; + + # Finally, push the data to the file. + try: + if sys.version_info < (3, 9, 0): + # Removed since Python 3.9. + abData = abPayload[4:].tostring(); + else: + abData = abPayload[4:].tobytes(); + oLocalFile.write(abData); + except: + reporter.errorXcpt('I/O error writing to "%s"' % (sRemoteFile)); + rc = None; + break; + rc = self.sendMsg('ACK'); + + # Send NACK on validation and I/O errors. + if rc is None: + rc = self.sendMsg('NACK'); + rc = False; + return rc; + + def taskPackFile(self, sRemoteFile, sRemoteSource): + rc = self.sendMsg('PKFILE', (sRemoteFile, sRemoteSource)); + if rc is True: + rc = self.recvAckLogged('PKFILE'); + return rc; + + def taskUnpackFile(self, sRemoteFile, sRemoteDir): + rc = self.sendMsg('UNPKFILE', (sRemoteFile, sRemoteDir)); + if rc is True: + rc = self.recvAckLogged('UNPKFILE'); + return rc; + + def taskExpandString(self, sString): + rc = self.sendMsg('EXP STR ', (sString,)); + if rc is True: + rc = False; + cbMsg, sOpcode, abPayload = self.recvReply(); + if cbMsg is not None: + sOpcode = sOpcode.strip(); + if sOpcode == "STRING": + sStringExp = getSZ(abPayload, 0); + if sStringExp is not None: + rc = sStringExp; + else: # Also handles SHORTSTR reply (not enough space to store result). + reporter.maybeErr(self.fErr, 'taskExpandString got a bad reply: %s' % (sOpcode,)); + else: + reporter.maybeErr(self.fErr, 'taskExpandString got 3xNone from recvReply.'); + return rc; + + # pylint: enable=missing-docstring + + + # + # Public methods - generic task queries + # + + def isSuccess(self): + """Returns True if the task completed successfully, otherwise False.""" + self.lockTask(); + sStatus = self.sStatus; + oTaskRc = self.oTaskRc; + self.unlockTask(); + if sStatus != "": + return False; + if oTaskRc is False or oTaskRc is None: + return False; + return True; + + def getResult(self): + """ + Returns the result of a completed task. + Returns None if not completed yet or no previous task. + """ + self.lockTask(); + sStatus = self.sStatus; + oTaskRc = self.oTaskRc; + self.unlockTask(); + if sStatus != "": + return None; + return oTaskRc; + + def getLastReply(self): + """ + Returns the last reply three-tuple: cbMsg, sOpcode, abPayload. + Returns a None, None, None three-tuple if there was no last reply. + """ + self.lockTask(); + t3oReply = self.t3oReply; + self.unlockTask(); + return t3oReply; + + # + # Public methods - connection. + # + + def asyncDisconnect(self, cMsTimeout = 30000, fIgnoreErrors = False): + """ + Initiates a disconnect task. + + Returns True on success, False on failure (logged). + + The task returns True on success and False on failure. + """ + return self.startTask(cMsTimeout, fIgnoreErrors, "bye", self.taskBye); + + def syncDisconnect(self, cMsTimeout = 30000, fIgnoreErrors = False): + """Synchronous version.""" + return self.asyncToSync(self.asyncDisconnect, cMsTimeout, fIgnoreErrors); + + def asyncVer(self, cMsTimeout = 30000, fIgnoreErrors = False): + """ + Initiates a task for getting the TXS version information. + + Returns True on success, False on failure (logged). + + The task returns the version string on success and False on failure. + """ + return self.startTask(cMsTimeout, fIgnoreErrors, "ver", self.taskVer); + + def syncVer(self, cMsTimeout = 30000, fIgnoreErrors = False): + """Synchronous version.""" + return self.asyncToSync(self.asyncVer, cMsTimeout, fIgnoreErrors); + + def asyncUuid(self, cMsTimeout = 30000, fIgnoreErrors = False): + """ + Initiates a task for getting the TXS UUID. + + Returns True on success, False on failure (logged). + + The task returns UUID string (in {}) on success and False on failure. + """ + return self.startTask(cMsTimeout, fIgnoreErrors, "uuid", self.taskUuid); + + def syncUuid(self, cMsTimeout = 30000, fIgnoreErrors = False): + """Synchronous version.""" + return self.asyncToSync(self.asyncUuid, cMsTimeout, fIgnoreErrors); + + # + # Public methods - execution. + # + + def asyncExecEx(self, sExecName, asArgs = (), asAddEnv = (), # pylint: disable=too-many-arguments + oStdIn = None, oStdOut = None, oStdErr = None, oTestPipe = None, + sAsUser = "", cMsTimeout = 3600000, fIgnoreErrors = False): + """ + Initiates a exec process task. + + Returns True on success, False on failure (logged). + + The task returns True if the process exited normally with status code 0. + The task returns None if on failure prior to executing the process, and + False if the process exited with a different status or in an abnormal + manner. Both None and False are logged of course and further info can + also be obtained by getLastReply(). + + The oStdIn, oStdOut, oStdErr and oTestPipe specifiy how to deal with + these streams. If None, no special action is taken and the output goes + to where ever the TXS sends its output, and ditto for input. + - To send to / read from the bitbucket, pass '/dev/null'. + - To redirect to/from a file, just specify the remote filename. + - To append to a file use '>>' followed by the remote filename. + - To pipe the stream to/from the TXS, specify a file like + object. For StdIn a non-blocking read() method is required. For + the other a write() method is required. Watch out for deadlock + conditions between StdIn and StdOut/StdErr/TestPipe piping. + """ + return self.startTask(cMsTimeout, fIgnoreErrors, "exec", self.taskExecEx, + (sExecName, long(0), asArgs, asAddEnv, oStdIn, + oStdOut, oStdErr, oTestPipe, sAsUser)); + + def syncExecEx(self, sExecName, asArgs = (), asAddEnv = (), # pylint: disable=too-many-arguments + oStdIn = '/dev/null', oStdOut = '/dev/null', + oStdErr = '/dev/null', oTestPipe = '/dev/null', + sAsUser = '', cMsTimeout = 3600000, fIgnoreErrors = False): + """Synchronous version.""" + return self.asyncToSync(self.asyncExecEx, sExecName, asArgs, asAddEnv, oStdIn, oStdOut, \ + oStdErr, oTestPipe, sAsUser, cMsTimeout, fIgnoreErrors); + + def asyncExec(self, sExecName, asArgs = (), asAddEnv = (), sAsUser = "", fWithTestPipe = True, sPrefix = '', \ + cMsTimeout = 3600000, fIgnoreErrors = False): + """ + Initiates a exec process test task. + + Returns True on success, False on failure (logged). + + The task returns True if the process exited normally with status code 0. + The task returns None if on failure prior to executing the process, and + False if the process exited with a different status or in an abnormal + manner. Both None and False are logged of course and further info can + also be obtained by getLastReply(). + + Standard in is taken from /dev/null. While both standard output and + standard error goes directly to reporter.log(). The testpipe is piped + to reporter.xxxx. + """ + + sStdIn = '/dev/null'; + oStdOut = reporter.FileWrapper('%sstdout' % sPrefix); + oStdErr = reporter.FileWrapper('%sstderr' % sPrefix); + if fWithTestPipe: oTestPipe = reporter.FileWrapperTestPipe(); + else: oTestPipe = '/dev/null'; # pylint: disable=redefined-variable-type + + return self.startTask(cMsTimeout, fIgnoreErrors, "exec", self.taskExecEx, + (sExecName, long(0), asArgs, asAddEnv, sStdIn, oStdOut, oStdErr, oTestPipe, sAsUser)); + + def syncExec(self, sExecName, asArgs = (), asAddEnv = (), sAsUser = '', fWithTestPipe = True, sPrefix = '', + cMsTimeout = 3600000, fIgnoreErrors = False): + """Synchronous version.""" + return self.asyncToSync(self.asyncExec, sExecName, asArgs, asAddEnv, sAsUser, fWithTestPipe, sPrefix, \ + cMsTimeout, fIgnoreErrors); + + # + # Public methods - system + # + + def asyncReboot(self, cMsTimeout = 30000, fIgnoreErrors = False): + """ + Initiates a reboot task. + + Returns True on success, False on failure (logged). + + The task returns True on success, False on failure (logged). The + session will be disconnected on successful task completion. + """ + return self.startTask(cMsTimeout, fIgnoreErrors, "reboot", self.taskReboot, ()); + + def syncReboot(self, cMsTimeout = 30000, fIgnoreErrors = False): + """Synchronous version.""" + return self.asyncToSync(self.asyncReboot, cMsTimeout, fIgnoreErrors); + + def asyncShutdown(self, cMsTimeout = 30000, fIgnoreErrors = False): + """ + Initiates a shutdown task. + + Returns True on success, False on failure (logged). + + The task returns True on success, False on failure (logged). + """ + return self.startTask(cMsTimeout, fIgnoreErrors, "shutdown", self.taskShutdown, ()); + + def syncShutdown(self, cMsTimeout = 30000, fIgnoreErrors = False): + """Synchronous version.""" + return self.asyncToSync(self.asyncShutdown, cMsTimeout, fIgnoreErrors); + + + # + # Public methods - file system + # + + def asyncMkDir(self, sRemoteDir, fMode = 0o700, cMsTimeout = 30000, fIgnoreErrors = False): + """ + Initiates a mkdir task. + + Returns True on success, False on failure (logged). + + The task returns True on success, False on failure (logged). + """ + return self.startTask(cMsTimeout, fIgnoreErrors, "mkDir", self.taskMkDir, (sRemoteDir, long(fMode))); + + def syncMkDir(self, sRemoteDir, fMode = 0o700, cMsTimeout = 30000, fIgnoreErrors = False): + """Synchronous version.""" + return self.asyncToSync(self.asyncMkDir, sRemoteDir, long(fMode), cMsTimeout, fIgnoreErrors); + + def asyncMkDirPath(self, sRemoteDir, fMode = 0o700, cMsTimeout = 30000, fIgnoreErrors = False): + """ + Initiates a mkdir -p task. + + Returns True on success, False on failure (logged). + + The task returns True on success, False on failure (logged). + """ + return self.startTask(cMsTimeout, fIgnoreErrors, "mkDirPath", self.taskMkDirPath, (sRemoteDir, long(fMode))); + + def syncMkDirPath(self, sRemoteDir, fMode = 0o700, cMsTimeout = 30000, fIgnoreErrors = False): + """Synchronous version.""" + return self.asyncToSync(self.asyncMkDirPath, sRemoteDir, long(fMode), cMsTimeout, fIgnoreErrors); + + def asyncMkSymlink(self, sLinkTarget, sLink, cMsTimeout = 30000, fIgnoreErrors = False): + """ + Initiates a symlink task. + + Returns True on success, False on failure (logged). + + The task returns True on success, False on failure (logged). + """ + return self.startTask(cMsTimeout, fIgnoreErrors, "mkSymlink", self.taskMkSymlink, (sLinkTarget, sLink)); + + def syncMkSymlink(self, sLinkTarget, sLink, cMsTimeout = 30000, fIgnoreErrors = False): + """Synchronous version.""" + return self.asyncToSync(self.asyncMkSymlink, sLinkTarget, sLink, cMsTimeout, fIgnoreErrors); + + def asyncRmDir(self, sRemoteDir, cMsTimeout = 30000, fIgnoreErrors = False): + """ + Initiates a rmdir task. + + Returns True on success, False on failure (logged). + + The task returns True on success, False on failure (logged). + """ + return self.startTask(cMsTimeout, fIgnoreErrors, "rmDir", self.taskRmDir, (sRemoteDir,)); + + def syncRmDir(self, sRemoteDir, cMsTimeout = 30000, fIgnoreErrors = False): + """Synchronous version.""" + return self.asyncToSync(self.asyncRmDir, sRemoteDir, cMsTimeout, fIgnoreErrors); + + def asyncRmFile(self, sRemoteFile, cMsTimeout = 30000, fIgnoreErrors = False): + """ + Initiates a rmfile task. + + Returns True on success, False on failure (logged). + + The task returns True on success, False on failure (logged). + """ + return self.startTask(cMsTimeout, fIgnoreErrors, "rmFile", self.taskRmFile, (sRemoteFile,)); + + def syncRmFile(self, sRemoteFile, cMsTimeout = 30000, fIgnoreErrors = False): + """Synchronous version.""" + return self.asyncToSync(self.asyncRmFile, sRemoteFile, cMsTimeout, fIgnoreErrors); + + def asyncRmSymlink(self, sRemoteSymlink, cMsTimeout = 30000, fIgnoreErrors = False): + """ + Initiates a rmsymlink task. + + Returns True on success, False on failure (logged). + + The task returns True on success, False on failure (logged). + """ + return self.startTask(cMsTimeout, fIgnoreErrors, "rmSymlink", self.taskRmSymlink, (sRemoteSymlink,)); + + def syncRmSymlink(self, sRemoteSymlink, cMsTimeout = 30000, fIgnoreErrors = False): + """Synchronous version.""" + return self.asyncToSync(self.asyncRmSymlink, sRemoteSymlink, cMsTimeout, fIgnoreErrors); + + def asyncRmTree(self, sRemoteTree, cMsTimeout = 30000, fIgnoreErrors = False): + """ + Initiates a rmtree task. + + Returns True on success, False on failure (logged). + + The task returns True on success, False on failure (logged). + """ + return self.startTask(cMsTimeout, fIgnoreErrors, "rmTree", self.taskRmTree, (sRemoteTree,)); + + def syncRmTree(self, sRemoteTree, cMsTimeout = 30000, fIgnoreErrors = False): + """Synchronous version.""" + return self.asyncToSync(self.asyncRmTree, sRemoteTree, cMsTimeout, fIgnoreErrors); + + def asyncChMod(self, sRemotePath, fMode, cMsTimeout = 30000, fIgnoreErrors = False): + """ + Initiates a chmod task. + + Returns True on success, False on failure (logged). + + The task returns True on success, False on failure (logged). + """ + return self.startTask(cMsTimeout, fIgnoreErrors, "chMod", self.taskChMod, (sRemotePath, fMode)); + + def syncChMod(self, sRemotePath, fMode, cMsTimeout = 30000, fIgnoreErrors = False): + """Synchronous version.""" + return self.asyncToSync(self.asyncChMod, sRemotePath, fMode, cMsTimeout, fIgnoreErrors); + + def asyncChOwn(self, sRemotePath, idUser, idGroup, cMsTimeout = 30000, fIgnoreErrors = False): + """ + Initiates a chown task. + + Returns True on success, False on failure (logged). + + The task returns True on success, False on failure (logged). + """ + return self.startTask(cMsTimeout, fIgnoreErrors, "chOwn", self.taskChOwn, (sRemotePath, idUser, idGroup)); + + def syncChOwn(self, sRemotePath, idUser, idGroup, cMsTimeout = 30000, fIgnoreErrors = False): + """Synchronous version.""" + return self.asyncToSync(self.asyncChMod, sRemotePath, idUser, idGroup, cMsTimeout, fIgnoreErrors); + + def asyncIsDir(self, sRemoteDir, cMsTimeout = 30000, fIgnoreErrors = False): + """ + Initiates a is-dir query task. + + Returns True on success, False on failure (logged). + + The task returns True if it's a directory, False if it isn't, and + None on error (logged). + """ + return self.startTask(cMsTimeout, fIgnoreErrors, "isDir", self.taskIsDir, (sRemoteDir,)); + + def syncIsDir(self, sRemoteDir, cMsTimeout = 30000, fIgnoreErrors = False): + """Synchronous version.""" + return self.asyncToSync(self.asyncIsDir, sRemoteDir, cMsTimeout, fIgnoreErrors); + + def asyncIsFile(self, sRemoteFile, cMsTimeout = 30000, fIgnoreErrors = False): + """ + Initiates a is-file query task. + + Returns True on success, False on failure (logged). + + The task returns True if it's a file, False if it isn't, and None on + error (logged). + """ + return self.startTask(cMsTimeout, fIgnoreErrors, "isFile", self.taskIsFile, (sRemoteFile,)); + + def syncIsFile(self, sRemoteFile, cMsTimeout = 30000, fIgnoreErrors = False): + """Synchronous version.""" + return self.asyncToSync(self.asyncIsFile, sRemoteFile, cMsTimeout, fIgnoreErrors); + + def asyncIsSymlink(self, sRemoteSymlink, cMsTimeout = 30000, fIgnoreErrors = False): + """ + Initiates a is-symbolic-link query task. + + Returns True on success, False on failure (logged). + + The task returns True if it's a symbolic linke, False if it isn't, and + None on error (logged). + """ + return self.startTask(cMsTimeout, fIgnoreErrors, "isSymlink", self.taskIsSymlink, (sRemoteSymlink,)); + + def syncIsSymlink(self, sRemoteSymlink, cMsTimeout = 30000, fIgnoreErrors = False): + """Synchronous version.""" + return self.asyncToSync(self.asyncIsSymlink, sRemoteSymlink, cMsTimeout, fIgnoreErrors); + + #def "STAT " + #def "LSTAT " + #def "LIST " + + @staticmethod + def calcFileXferTimeout(cbFile): + """ + Calculates a reasonable timeout for an upload/download given the file size. + + Returns timeout in milliseconds. + """ + return 30000 + cbFile / 32; # 32 KiB/s (picked out of thin air) + + @staticmethod + def calcUploadTimeout(sLocalFile): + """ + Calculates a reasonable timeout for an upload given the file (will stat it). + + Returns timeout in milliseconds. + """ + try: cbFile = os.path.getsize(sLocalFile); + except: cbFile = 1024*1024; + return Session.calcFileXferTimeout(cbFile); + + def asyncCopyFile(self, sSrcFile, sDstFile, + fMode = 0, fFallbackOkay = True, cMsTimeout = 30000, fIgnoreErrors = False): + """ + Initiates a file copying task on the remote. + + Returns True on success, False on failure (logged). + + The task returns True on success, False on failure (logged). + """ + return self.startTask(cMsTimeout, fIgnoreErrors, "cpfile", + self.taskCopyFile, (sSrcFile, sDstFile, fMode, fFallbackOkay)); + + def syncCopyFile(self, sSrcFile, sDstFile, fMode = 0, cMsTimeout = 30000, fIgnoreErrors = False): + """Synchronous version.""" + return self.asyncToSync(self.asyncCopyFile, sSrcFile, sDstFile, fMode, cMsTimeout, fIgnoreErrors); + + def asyncUploadFile(self, sLocalFile, sRemoteFile, + fMode = 0, fFallbackOkay = True, cMsTimeout = 30000, fIgnoreErrors = False): + """ + Initiates a download query task. + + Returns True on success, False on failure (logged). + + The task returns True on success, False on failure (logged). + """ + return self.startTask(cMsTimeout, fIgnoreErrors, "upload", + self.taskUploadFile, (sLocalFile, sRemoteFile, fMode, fFallbackOkay)); + + def syncUploadFile(self, sLocalFile, sRemoteFile, fMode = 0, fFallbackOkay = True, cMsTimeout = 0, fIgnoreErrors = False): + """Synchronous version.""" + if cMsTimeout <= 0: + cMsTimeout = self.calcUploadTimeout(sLocalFile); + return self.asyncToSync(self.asyncUploadFile, sLocalFile, sRemoteFile, fMode, fFallbackOkay, cMsTimeout, fIgnoreErrors); + + def asyncUploadString(self, sContent, sRemoteFile, + fMode = 0, fFallbackOkay = True, cMsTimeout = 0, fIgnoreErrors = False): + """ + Initiates a upload string task. + + Returns True on success, False on failure (logged). + + The task returns True on success, False on failure (logged). + """ + if cMsTimeout <= 0: + cMsTimeout = self.calcFileXferTimeout(len(sContent)); + return self.startTask(cMsTimeout, fIgnoreErrors, "uploadString", + self.taskUploadString, (sContent, sRemoteFile, fMode, fFallbackOkay)); + + def syncUploadString(self, sContent, sRemoteFile, fMode = 0, fFallbackOkay = True, cMsTimeout = 0, fIgnoreErrors = False): + """Synchronous version.""" + if cMsTimeout <= 0: + cMsTimeout = self.calcFileXferTimeout(len(sContent)); + return self.asyncToSync(self.asyncUploadString, sContent, sRemoteFile, fMode, fFallbackOkay, cMsTimeout, fIgnoreErrors); + + def asyncDownloadFile(self, sRemoteFile, sLocalFile, cMsTimeout = 120000, fIgnoreErrors = False): + """ + Initiates a download file task. + + Returns True on success, False on failure (logged). + + The task returns True on success, False on failure (logged). + """ + return self.startTask(cMsTimeout, fIgnoreErrors, "downloadFile", self.taskDownloadFile, (sRemoteFile, sLocalFile)); + + def syncDownloadFile(self, sRemoteFile, sLocalFile, cMsTimeout = 120000, fIgnoreErrors = False): + """Synchronous version.""" + return self.asyncToSync(self.asyncDownloadFile, sRemoteFile, sLocalFile, cMsTimeout, fIgnoreErrors); + + def asyncDownloadString(self, sRemoteFile, sEncoding = 'utf-8', fIgnoreEncodingErrors = True, + cMsTimeout = 30000, fIgnoreErrors = False): + """ + Initiates a download string task. + + Returns True on success, False on failure (logged). + + The task returns a byte string on success, False on failure (logged). + """ + return self.startTask(cMsTimeout, fIgnoreErrors, "downloadString", + self.taskDownloadString, (sRemoteFile, sEncoding, fIgnoreEncodingErrors)); + + def syncDownloadString(self, sRemoteFile, sEncoding = 'utf-8', fIgnoreEncodingErrors = True, + cMsTimeout = 30000, fIgnoreErrors = False): + """Synchronous version.""" + return self.asyncToSync(self.asyncDownloadString, sRemoteFile, sEncoding, fIgnoreEncodingErrors, + cMsTimeout, fIgnoreErrors); + + def asyncPackFile(self, sRemoteFile, sRemoteSource, cMsTimeout = 120000, fIgnoreErrors = False): + """ + Initiates a packing file/directory task. + + Returns True on success, False on failure (logged). + + The task returns True on success, False on failure (logged). + """ + return self.startTask(cMsTimeout, fIgnoreErrors, "packFile", self.taskPackFile, + (sRemoteFile, sRemoteSource)); + + def syncPackFile(self, sRemoteFile, sRemoteSource, cMsTimeout = 120000, fIgnoreErrors = False): + """Synchronous version.""" + return self.asyncToSync(self.asyncPackFile, sRemoteFile, sRemoteSource, cMsTimeout, fIgnoreErrors); + + def asyncUnpackFile(self, sRemoteFile, sRemoteDir, cMsTimeout = 120000, fIgnoreErrors = False): + """ + Initiates a unpack file task. + + Returns True on success, False on failure (logged). + + The task returns True on success, False on failure (logged). + """ + return self.startTask(cMsTimeout, fIgnoreErrors, "unpackFile", self.taskUnpackFile, + (sRemoteFile, sRemoteDir)); + + def syncUnpackFile(self, sRemoteFile, sRemoteDir, cMsTimeout = 120000, fIgnoreErrors = False): + """Synchronous version.""" + return self.asyncToSync(self.asyncUnpackFile, sRemoteFile, sRemoteDir, cMsTimeout, fIgnoreErrors); + + def asyncExpandString(self, sString, cMsTimeout = 120000, fIgnoreErrors = False): + """ + Initiates an expand string task. + + Returns expanded string on success, False on failure (logged). + + The task returns True on success, False on failure (logged). + """ + return self.startTask(cMsTimeout, fIgnoreErrors, "expandString", + self.taskExpandString, (sString,)); + + def syncExpandString(self, sString, cMsTimeout = 120000, fIgnoreErrors = False): + """Synchronous version.""" + return self.asyncToSync(self.asyncExpandString, sString, cMsTimeout, fIgnoreErrors); + + +class TransportTcp(TransportBase): + """ + TCP transport layer for the TXS client session class. + """ + + def __init__(self, sHostname, uPort, fReversedSetup): + """ + Save the parameters. The session will call us back to make the + connection later on its worker thread. + """ + TransportBase.__init__(self, utils.getCallerName()); + self.sHostname = sHostname; + self.fReversedSetup = fReversedSetup; + self.uPort = uPort if uPort is not None else 5042 if fReversedSetup is False else 5048; + self.oSocket = None; + self.oWakeupW = None; + self.oWakeupR = None; + self.fConnectCanceled = False; + self.fIsConnecting = False; + self.oCv = threading.Condition(); + self.abReadAhead = array.array('B'); + + def toString(self): + return '<%s sHostname=%s, fReversedSetup=%s, uPort=%s, oSocket=%s,'\ + ' fConnectCanceled=%s, fIsConnecting=%s, oCv=%s, abReadAhead=%s>' \ + % (TransportBase.toString(self), self.sHostname, self.fReversedSetup, self.uPort, self.oSocket, + self.fConnectCanceled, self.fIsConnecting, self.oCv, self.abReadAhead); + + def __isInProgressXcpt(self, oXcpt): + """ In progress exception? """ + try: + if isinstance(oXcpt, socket.error): + try: + if oXcpt.errno == errno.EINPROGRESS: + return True; + except: pass; + # Windows? + try: + if oXcpt.errno == errno.EWOULDBLOCK: + return True; + except: pass; + except: + pass; + return False; + + def __isWouldBlockXcpt(self, oXcpt): + """ Would block exception? """ + try: + if isinstance(oXcpt, socket.error): + try: + if oXcpt.errno == errno.EWOULDBLOCK: + return True; + except: pass; + try: + if oXcpt.errno == errno.EAGAIN: + return True; + except: pass; + except: + pass; + return False; + + def __isConnectionReset(self, oXcpt): + """ Connection reset by Peer or others. """ + try: + if isinstance(oXcpt, socket.error): + try: + if oXcpt.errno == errno.ECONNRESET: + return True; + except: pass; + try: + if oXcpt.errno == errno.ENETRESET: + return True; + except: pass; + except: + pass; + return False; + + def _closeWakeupSockets(self): + """ Closes the wakup sockets. Caller should own the CV. """ + oWakeupR = self.oWakeupR; + self.oWakeupR = None; + if oWakeupR is not None: + oWakeupR.close(); + + oWakeupW = self.oWakeupW; + self.oWakeupW = None; + if oWakeupW is not None: + oWakeupW.close(); + + return None; + + def cancelConnect(self): + # This is bad stuff. + self.oCv.acquire(); + reporter.log2('TransportTcp::cancelConnect: fIsConnecting=%s oSocket=%s' % (self.fIsConnecting, self.oSocket)); + self.fConnectCanceled = True; + if self.fIsConnecting: + oSocket = self.oSocket; + self.oSocket = None; + if oSocket is not None: + reporter.log2('TransportTcp::cancelConnect: closing the socket'); + oSocket.close(); + + oWakeupW = self.oWakeupW; + self.oWakeupW = None; + if oWakeupW is not None: + reporter.log2('TransportTcp::cancelConnect: wakeup call'); + try: oWakeupW.send(b'cancelled!\n'); + except: reporter.logXcpt(); + try: oWakeupW.shutdown(socket.SHUT_WR); + except: reporter.logXcpt(); + oWakeupW.close(); + self.oCv.release(); + + def _connectAsServer(self, oSocket, oWakeupR, cMsTimeout): + """ Connects to the TXS server as server, i.e. the reversed setup. """ + assert(self.fReversedSetup); + + reporter.log2('TransportTcp::_connectAsServer: oSocket=%s, cMsTimeout=%u' % (oSocket, cMsTimeout)); + + # Workaround for bind() failure... + try: + oSocket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1); + except: + reporter.errorXcpt('socket.listen(1) failed'); + return None; + + # Bind the socket and make it listen. + try: + oSocket.bind((self.sHostname, self.uPort)); + except: + reporter.errorXcpt('socket.bind((%s,%s)) failed' % (self.sHostname, self.uPort)); + return None; + try: + oSocket.listen(1); + except: + reporter.errorXcpt('socket.listen(1) failed'); + return None; + + # Accept connections. + oClientSocket = None; + tClientAddr = None; + try: + (oClientSocket, tClientAddr) = oSocket.accept(); + except socket.error as e: + if not self.__isInProgressXcpt(e): + raise; + + # Do the actual waiting. + reporter.log2('TransportTcp::accept: operation in progress (%s)...' % (e,)); + try: + select.select([oSocket, oWakeupR], [], [oSocket, oWakeupR], cMsTimeout / 1000.0); + except socket.error as oXctp: + if oXctp.errno != errno.EBADF or not self.fConnectCanceled: + raise; + reporter.log('socket.select() on accept was canceled'); + return None; + except: + reporter.logXcpt('socket.select() on accept'); + + # Try accept again. + try: + (oClientSocket, tClientAddr) = oSocket.accept(); + except socket.error as oXcpt: + if not self.__isInProgressXcpt(e): + if oXcpt.errno != errno.EBADF or not self.fConnectCanceled: + raise; + reporter.log('socket.accept() was canceled'); + return None; + reporter.log('socket.accept() timed out'); + return False; + except: + reporter.errorXcpt('socket.accept() failed'); + return None; + except: + reporter.errorXcpt('socket.accept() failed'); + return None; + + # Store the connected socket and throw away the server socket. + self.oCv.acquire(); + if not self.fConnectCanceled: + self.oSocket.close(); + self.oSocket = oClientSocket; + self.sHostname = "%s:%s" % (tClientAddr[0], tClientAddr[1]); + self.oCv.release(); + return True; + + def _connectAsClient(self, oSocket, oWakeupR, cMsTimeout): + """ Connects to the TXS server as client. """ + assert(not self.fReversedSetup); + + # Connect w/ timeouts. + rc = None; + try: + oSocket.connect((self.sHostname, self.uPort)); + rc = True; + except socket.error as oXcpt: + iRc = oXcpt.errno; + if self.__isInProgressXcpt(oXcpt): + # Do the actual waiting. + reporter.log2('TransportTcp::connect: operation in progress (%s)...' % (oXcpt,)); + try: + ttRc = select.select([oWakeupR], [oSocket], [oSocket, oWakeupR], cMsTimeout / 1000.0); + if len(ttRc[1]) + len(ttRc[2]) == 0: + raise socket.error(errno.ETIMEDOUT, 'select timed out'); + iRc = oSocket.getsockopt(socket.SOL_SOCKET, socket.SO_ERROR); + rc = iRc == 0; + except socket.error as oXcpt2: + iRc = oXcpt2.errno; + except: + iRc = -42; + reporter.fatalXcpt('socket.select() on connect failed'); + + if rc is True: + pass; + elif iRc in (errno.ECONNREFUSED, errno.EHOSTUNREACH, errno.EINTR, errno.ENETDOWN, errno.ENETUNREACH, errno.ETIMEDOUT): + rc = False; # try again. + else: + if iRc != errno.EBADF or not self.fConnectCanceled: + reporter.fatalXcpt('socket.connect((%s,%s)) failed; iRc=%s' % (self.sHostname, self.uPort, iRc)); + reporter.log2('TransportTcp::connect: rc=%s iRc=%s' % (rc, iRc)); + except: + reporter.fatalXcpt('socket.connect((%s,%s)) failed' % (self.sHostname, self.uPort)); + return rc; + + + def connect(self, cMsTimeout): + # Create a non-blocking socket. + reporter.log2('TransportTcp::connect: cMsTimeout=%s sHostname=%s uPort=%s' % (cMsTimeout, self.sHostname, self.uPort)); + try: + oSocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM, 0); + except: + reporter.fatalXcpt('socket.socket() failed'); + return None; + try: + oSocket.setblocking(0); + except: + oSocket.close(); + reporter.fatalXcpt('socket.socket() failed'); + return None; + + # Create wakeup socket pair for unix (select doesn't wake up on socket close on Linux). + oWakeupR = None; + oWakeupW = None; + if hasattr(socket, 'socketpair'): + try: (oWakeupR, oWakeupW) = socket.socketpair(); # pylint: disable=no-member + except: reporter.logXcpt('socket.socketpair() failed'); + + # Update the state. + self.oCv.acquire(); + rc = None; + if not self.fConnectCanceled: + self.oSocket = oSocket; + self.oWakeupW = oWakeupW; + self.oWakeupR = oWakeupR; + self.fIsConnecting = True; + self.oCv.release(); + + # Try connect. + if oWakeupR is None: + oWakeupR = oSocket; # Avoid select failure. + if self.fReversedSetup: + rc = self._connectAsServer(oSocket, oWakeupR, cMsTimeout); + else: + rc = self._connectAsClient(oSocket, oWakeupR, cMsTimeout); + oSocket = None; + + # Update the state and cleanup on failure/cancel. + self.oCv.acquire(); + if rc is True and self.fConnectCanceled: + rc = False; + self.fIsConnecting = False; + + if rc is not True: + if self.oSocket is not None: + self.oSocket.close(); + self.oSocket = None; + self._closeWakeupSockets(); + self.oCv.release(); + + reporter.log2('TransportTcp::connect: returning %s' % (rc,)); + return rc; + + def disconnect(self, fQuiet = False): + if self.oSocket is not None: + self.abReadAhead = array.array('B'); + + # Try a shutting down the socket gracefully (draining it). + try: + self.oSocket.shutdown(socket.SHUT_WR); + except: + if not fQuiet: + reporter.error('shutdown(SHUT_WR)'); + try: + self.oSocket.setblocking(0); # just in case it's not set. + sData = "1"; + while sData: + sData = self.oSocket.recv(16384); + except: + pass; + + # Close it. + self.oCv.acquire(); + try: self.oSocket.setblocking(1); + except: pass; + self.oSocket.close(); + self.oSocket = None; + else: + self.oCv.acquire(); + self._closeWakeupSockets(); + self.oCv.release(); + + def sendBytes(self, abBuf, cMsTimeout): + if self.oSocket is None: + reporter.error('TransportTcp.sendBytes: No connection.'); + return False; + + # Try send it all. + try: + cbSent = self.oSocket.send(abBuf); + if cbSent == len(abBuf): + return True; + except Exception as oXcpt: + if not self.__isWouldBlockXcpt(oXcpt): + reporter.errorXcpt('TranportTcp.sendBytes: %s bytes' % (len(abBuf))); + return False; + cbSent = 0; + + # Do a timed send. + msStart = base.timestampMilli(); + while True: + cMsElapsed = base.timestampMilli() - msStart; + if cMsElapsed > cMsTimeout: + reporter.error('TranportTcp.sendBytes: %s bytes timed out (1)' % (len(abBuf))); + break; + + # wait. + try: + ttRc = select.select([], [self.oSocket], [self.oSocket], (cMsTimeout - cMsElapsed) / 1000.0); + if ttRc[2] and not ttRc[1]: + reporter.error('TranportTcp.sendBytes: select returned with exception'); + break; + if not ttRc[1]: + reporter.error('TranportTcp.sendBytes: %s bytes timed out (2)' % (len(abBuf))); + break; + except: + reporter.errorXcpt('TranportTcp.sendBytes: select failed'); + break; + + # Try send more. + try: + cbSent += self.oSocket.send(abBuf[cbSent:]); + if cbSent == len(abBuf): + return True; + except Exception as oXcpt: + if not self.__isWouldBlockXcpt(oXcpt): + reporter.errorXcpt('TranportTcp.sendBytes: %s bytes' % (len(abBuf))); + break; + + return False; + + def __returnReadAheadBytes(self, cb): + """ Internal worker for recvBytes. """ + assert(len(self.abReadAhead) >= cb); + abRet = self.abReadAhead[:cb]; + self.abReadAhead = self.abReadAhead[cb:]; + return abRet; + + def recvBytes(self, cb, cMsTimeout, fNoDataOk): + if self.oSocket is None: + reporter.error('TransportTcp.recvBytes(%s,%s): No connection.' % (cb, cMsTimeout)); + return None; + + # Try read in some more data without bothering with timeout handling first. + if len(self.abReadAhead) < cb: + try: + abBuf = self.oSocket.recv(cb - len(self.abReadAhead)); + if abBuf: + self.abReadAhead.extend(array.array('B', abBuf)); + except Exception as oXcpt: + if not self.__isWouldBlockXcpt(oXcpt): + reporter.errorXcpt('TranportTcp.recvBytes: 0/%s bytes' % (cb,)); + return None; + + if len(self.abReadAhead) >= cb: + return self.__returnReadAheadBytes(cb); + + # Timeout loop. + msStart = base.timestampMilli(); + while True: + cMsElapsed = base.timestampMilli() - msStart; + if cMsElapsed > cMsTimeout: + if not fNoDataOk or self.abReadAhead: + reporter.error('TranportTcp.recvBytes: %s/%s bytes timed out (1)' % (len(self.abReadAhead), cb)); + break; + + # Wait. + try: + ttRc = select.select([self.oSocket], [], [self.oSocket], (cMsTimeout - cMsElapsed) / 1000.0); + if ttRc[2] and not ttRc[0]: + reporter.error('TranportTcp.recvBytes: select returned with exception'); + break; + if not ttRc[0]: + if not fNoDataOk or self.abReadAhead: + reporter.error('TranportTcp.recvBytes: %s/%s bytes timed out (2) fNoDataOk=%s' + % (len(self.abReadAhead), cb, fNoDataOk)); + break; + except: + reporter.errorXcpt('TranportTcp.recvBytes: select failed'); + break; + + # Try read more. + try: + abBuf = self.oSocket.recv(cb - len(self.abReadAhead)); + if not abBuf: + reporter.error('TranportTcp.recvBytes: %s/%s bytes (%s) - connection has been shut down' + % (len(self.abReadAhead), cb, fNoDataOk)); + self.disconnect(); + return None; + + self.abReadAhead.extend(array.array('B', abBuf)); + + except Exception as oXcpt: + reporter.log('recv => exception %s' % (oXcpt,)); + if not self.__isWouldBlockXcpt(oXcpt): + if not fNoDataOk or not self.__isConnectionReset(oXcpt) or self.abReadAhead: + reporter.errorXcpt('TranportTcp.recvBytes: %s/%s bytes (%s)' % (len(self.abReadAhead), cb, fNoDataOk)); + break; + + # Done? + if len(self.abReadAhead) >= cb: + return self.__returnReadAheadBytes(cb); + + #reporter.log('recv => None len(self.abReadAhead) -> %d' % (len(self.abReadAhead), )); + return None; + + def isConnectionOk(self): + if self.oSocket is None: + return False; + try: + ttRc = select.select([], [], [self.oSocket], 0.0); + if ttRc[2]: + return False; + + self.oSocket.send(array.array('B')); # send zero bytes. + except: + return False; + return True; + + def isRecvPending(self, cMsTimeout = 0): + try: + ttRc = select.select([self.oSocket], [], [], cMsTimeout / 1000.0); + if not ttRc[0]: + return False; + except: + pass; + return True; + + +def openTcpSession(cMsTimeout, sHostname, uPort = None, fReversedSetup = False, cMsIdleFudge = 0, fnProcessEvents = None): + """ + Opens a connection to a Test Execution Service via TCP, given its name. + + The optional fnProcessEvents callback should be set to vbox.processPendingEvents + or similar. + """ + reporter.log2('openTcpSession(%s, %s, %s, %s, %s)' % + (cMsTimeout, sHostname, uPort, fReversedSetup, cMsIdleFudge)); + try: + oTransport = TransportTcp(sHostname, uPort, fReversedSetup); + oSession = Session(oTransport, cMsTimeout, cMsIdleFudge, fnProcessEvents = fnProcessEvents); + except: + reporter.errorXcpt(None, 15); + return None; + return oSession; + + +def tryOpenTcpSession(cMsTimeout, sHostname, uPort = None, fReversedSetup = False, cMsIdleFudge = 0, fnProcessEvents = None): + """ + Tries to open a connection to a Test Execution Service via TCP, given its name. + + This differs from openTcpSession in that it won't log a connection failure + as an error. + """ + try: + oTransport = TransportTcp(sHostname, uPort, fReversedSetup); + oSession = Session(oTransport, cMsTimeout, cMsIdleFudge, fTryConnect = True, fnProcessEvents = fnProcessEvents); + except: + reporter.errorXcpt(None, 15); + return None; + return oSession; |