rwed

"""O_o Copyright (c) 2006, Milos Negovanovic o_OAll rights reserved.Redistribution and use in source and binary forms, with or withoutmodification, are permitted provided that the following conditionsare met:    1. Redistributions of source code must retain the above copyright notice,    this list of conditions and the following disclaimer.    2. Redistributions in binary form must reproduce the above copyright    notice, this list of conditions and the following disclaimer in the    documentation and/or other materials provided with the distribution.    3. All advertising materials mentioning features or use of this software    must display the following acknowledgement:        This product includes software developed by Milos Negovanovic.    4. Neither the name of the Milos Negovanovic nor the names of its    contributors may be used to endorse or promote products derived from this    software without specific prior written permission.THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THEIMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE AREDISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLEFOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIALDAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS ORSERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVERCAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USEOF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE."""import osimport signalimport sysfrom twisted.internet import reactor, protocol, defer, task, threads, sslfrom twisted.protocols.basic import LineReceiverfrom twisted.python import failureimport timeimport StringIOdeferred = threads.deferToThread.__get__DELIMITER =             "<-RWE->"EXEC =                  "EXEC"EXEC_MULTI =            "EXEC_MULTI"EXEC_STATUS =           "EXEC_STATUS"READ =                  "READ"WRITE =                 "WRITE"WATCH =                 "WATCH"MAXPOUT = 65536#@deferreddef readFile(tag, RS, filename):    record = RS.records[tag]    try:        f = open(filename)    except IOError, (errno, strerror):        #self.d.errback(failure.Failure(self.tag        #    + "OFD_delimiterI/O error(%s): %s" % (errno, strerror)))        record["all"] = "I/O error(%s): %s" % (errno, strerror)        record["err"] = "I/O error(%s): %s" % (errno, strerror)        record["out"] = ""        record["exitcode"] = 1        record["done"] = True        return            content = f.read()    f.close()    record["all"] = content    record["err"] = ""    record["out"] = content    record["exitcode"] = 0    record["done"] = True#@deferreddef writeFile(tag, RS, filename, content):    record = RS.records[tag]    try:        f = open(filename, "w")    except IOError, (errno, strerror):        #self.d.errback(failure.Failure(self.tag        #    + "OFD_delimiterI/O error(%s): %s" % (errno, strerror)))        record["all"] = "I/O error(%s): %s" % (errno, strerror)        record["err"] = "I/O error(%s): %s" % (errno, strerror)        record["out"] = ""        record["exitcode"] = 1        record["done"] = True        return        f.write(content)    f.close()    record["all"] = ""    record["err"] = ""    record["out"] = ""    record["exitcode"] = 0    record["done"] = Truedef getProcessOutput(record, argid = None, env={}, path='.', reactor=reactor):    if record["type"] == EXEC or record["type"] == EXEC_STATUS:        executable = record["fileORcom"]        args = record["args"][0]        return __spawnProcess( __RWEDPP(record, argid), executable, args, env,                path, reactor)        elif record["type"] == EXEC_MULTI:        executable = record["fileORcom"]        args = record["args"][argid]        return __spawnProcess( __RWEDPP(record, argid), executable, args, env,                path, reactor)def __spawnProcess(protocol, executable, args, env, path, reactor):    p = protocol    reactor.spawnProcess(p, executable, (executable,)+tuple(args), env, path)    return p    class __RWEDPP(protocol.ProcessProtocol):    def __init__(self, record, argid):        self.all = StringIO.StringIO()        self.err = StringIO.StringIO()        self.out = StringIO.StringIO()        self.argid = argid        self.record = record        def connectionMade(self):        self.record["processprotocol"] = self        self.transport.closeStdin() # tell them we're done        def outReceived(self, text):        self.all.write(text)        self.out.write(text)        self.maxpoutcheck()            def errReceived(self, text):        self.all.write(text)        self.err.write(text)        self.maxpoutcheck()        def inConnectionLost(self):        pass        def outConnectionLost(self):        pass        def errConnectionLost(self):        pass        def processEnded(self, status_object):        #print "processEnded, status %d" % status_object.value.exitCode        if self.record["type"] == EXEC or self.record["type"] == EXEC_STATUS:            self.record["all"] = self.all.getvalue()            self.record["out"] = self.out.getvalue()            self.record["err"] = self.err.getvalue()            if self.record["exitcode"] is None:                self.record["exitcode"] = status_object.value.exitCode            self.record["done"] = True                elif self.record["type"] == EXEC_MULTI:            self.record["all"][self.argid] = self.all.getvalue()            self.record["err"][self.argid] = self.err.getvalue()            self.record["out"][self.argid] = self.out.getvalue()            if self.record["exitcode"][self.argid] is None:                self.record["exitcode"][self.argid] = status_object.value.exitCode            self.record["mrresultcount"] += 1        def maxpoutcheck(self):        if len(self.all.getvalue()) > MAXPOUT:            code = "KILL9 max output reached"            if self.record["type"] == EXEC:                self.record["exitcode"] = code            elif self.record["type"] == EXEC_MULTI:                self.record["exitcode"][self.argid] = code            self.kill()        def statusupdate(self):        self.record["all"] = self.all.getvalue()        self.all = StringIO.StringIO()                self.record["out"] = self.out.getvalue()        self.out = StringIO.StringIO()                self.record["err"] = self.err.getvalue()        self.err = StringIO.StringIO()        def kill(self):        os.kill(self.transport.pid, signal.SIGKILL)class RWEDSSRecordStore:    def __init__(self):        self.records = {}        watcher = task.LoopingCall(self.watcher)        statuswatcher = task.LoopingCall(self.statuswatcher)        watcher.start(0.1) # call every 0.1 seconds        statuswatcher.start(1) # call every 1 seconds        def registerRecord(self, protocol, s):        words = s.split(DELIMITER)        wordcount = len(words)        if wordcount < 2:            return                argstart = 3        tag = words[0]        record = {}        record["protocol"] = protocol        record["done"] = False        record["exitcode"] = None        record["all"] = None        record["err"] = None        record["out"] = None        record["type"] = words[1]        record["fileORcom"] = words[2]                if record["type"] == EXEC_MULTI:            record["delay"] = float(words[3])            record["mrresultcount"] = 0            argstart = 4        elif record["type"] == EXEC_STATUS:            record["processprotocol"] = None            record["delay"] = float(words[3])            record["statuscounter"] = record["delay"]            argstart = 4                record["args"] = []        for i in range(argstart, wordcount):            if record["type"] == WRITE:                record["args"].append(words[i])            else:                record["args"].append(words[i].split())        self.records[tag] = record                        if record["type"] == EXEC or record["type"] == EXEC_STATUS:            p = getProcessOutput(record)                elif record["type"] == EXEC_MULTI:            args = record["args"]            record["exitcode"] = []            record["all"] = []            record["err"] = []            record["out"] = []            for arg in args:                record["exitcode"].append(None)                record["all"].append(None)                record["err"].append(None)                record["out"].append(None)            argid = 0            delay = record["delay"]            for arg in args:                reactor.callLater(argid*delay, getProcessOutput, record, argid)                argid += 1                elif record["type"] == READ:            file = record["fileORcom"]            d = readFile(tag, self, file)                elif record["type"] == WRITE:            file = record["fileORcom"]            content = record["args"][0]            d = writeFile(tag, self, file, content)        def watcher(self):        for key in self.records.keys():            record = self.records[key]            if record["type"] == EXEC:                if record["done"]:                    exitcode = record["exitcode"]                    all = record["all"]                    err = record["err"]                    out = record["out"]                    d = DELIMITER                    resp = "%s%s%s%s%s%s%s%s%s%s%s" % \                            (key, d, EXEC, d, str(exitcode), d, all, d, err, d, out)                                                record["protocol"].sendLine(resp)                    del self.records[key]                        elif record["type"] == EXEC_MULTI:                exitcodes = record["exitcode"]                count = record["mrresultcount"]                if len(exitcodes) == count:                    alls = record["all"]                    errs = record["err"]                    outs = record["out"]                    d = DELIMITER                    resp = "%s%s%s" % (key , d , EXEC_MULTI)                    for i in range(0, count):                        resp = "%s%s%s%s%s%s%s%s%s" % \                                (resp,                                d, str(exitcodes[i]),                                d, alls[i],                                d, errs[i],                                d, outs[i])                                        record["protocol"].sendLine(resp)                    del self.records[key]                        elif record["type"] == READ:                if record["done"]:                    exitcode = record["exitcode"]                    all = record["all"]                    err = record["err"]                    out = record["out"]                    d = DELIMITER                    resp = "%s%s%s%s%s%s%s%s%s%s%s" % \                            (key, d, READ, d, str(exitcode), d, all, d, err, d, out)                                                record["protocol"].sendLine(resp)                    del self.records[key]                        elif record["type"] == WRITE:                if record["done"]:                    exitcode = record["exitcode"]                    all = record["all"]                    err = record["err"]                    out = record["out"]                    d = DELIMITER                    resp = "%s%s%s%s%s%s%s%s%s%s%s" % \                            (key, d, WRITE, d, str(exitcode), d, all, d, err, d, out)                                                record["protocol"].sendLine(resp)                    del self.records[key]        def statuswatcher(self):        for key in self.records.keys():            d = DELIMITER            none = "None"            record = self.records[key]            if record["type"] == EXEC_STATUS:                if record["done"]:                    exitcode = record["exitcode"]                    all = record["all"]                    err = record["err"]                    out = record["out"]                    resp = "%s%s%s%s%s%s%s%s%s%s%s" % \                            (key, d, EXEC_STATUS, d, str(exitcode), d, all, d, err, d, out)                                                record["protocol"].sendLine(resp)                    del self.records[key]                else:                    record["statuscounter"] -= 1                    if record["statuscounter"] != 0:                        continue                    if record["protocol"] is None:                        continue                    record["processprotocol"].statusupdate()                    all = record["all"]                    record["all"] = ""                    err = record["err"]                    record["err"] = ""                    out = record["out"]                    record["out"] = ""                    resp = "%s%s%s%s%s%s%s%s%s%s%s" % \                            (key, d, EXEC_STATUS, d, none, d, all, d, err, d, out)                    record["statuscounter"] = record["delay"]                    record["protocol"].sendLine(resp)        def recordCleanUp(self, protocol):        for key in self.records.keys():            record = self.records[key]            if record["protocol"] is protocol:                record["protocol"] = None                try:                    record["processprotocol"].kill()                except KeyError:                    pass                del self.records[key]class RWEDProtocol(LineReceiver):    def connectionMade(self):        interpreter = sys.executable        self.sendLine("RWED v0.1 " + interpreter)        def lineReceived(self, line):        self.factory.RS.registerRecord(self, line)    def connectionLost(self, reason):        self.factory.RS.recordCleanUp(self)class RWEDFactory(protocol.Factory):    def startFactory(self):        self.RS = RWEDSSRecordStore()if __name__ == '__main__':    factory = RWEDFactory()    factory.protocol = RWEDProtocol        sslContext = ssl.DefaultOpenSSLContextFactory(            '/tmp/privkey.pem',            '/tmp/cacert.pem')    reactor.listenSSL(10999, factory, sslContext)    reactor.run()