root/foolscap/call.py

from twisted.python import failure, reflect
from twisted.internet import defer

from foolscap import copyable, slicer, tokens
from foolscap.copyable import AttributeDictConstraint
from foolscap.constraint import ByteStringConstraint
from foolscap.slicers.list import ListConstraint
from tokens import BananaError, Violation
from foolscap.util import AsyncAND
from foolscap.logging import log


class FailureConstraint(AttributeDictConstraint):
    opentypes = [("copyable", "twisted.python.failure.Failure")]
    name = "FailureConstraint"
    klass = failure.Failure

    def __init__(self):
        attrs = [('type', ByteStringConstraint(200)),
                 ('value', ByteStringConstraint(1000)),
                 ('traceback', ByteStringConstraint(2000)),
                 ('parents', ListConstraint(ByteStringConstraint(200))),
                 ]
        AttributeDictConstraint.__init__(self, *attrs)

    def checkObject(self, obj, inbound):
        if not isinstance(obj, self.klass):
            raise Violation("is not an instance of %s" % self.klass)


class PendingRequest(object):
    # this object is a local representation of a message we have sent to
    # someone else, that will be executed on their end.
    active = True

    def __init__(self, reqID, rref=None):
        self.reqID = reqID
        self.rref = rref # keep it alive
        self.broker = None # if set, the broker knows about us
        self.deferred = defer.Deferred()
        self.constraint = None # this constrains the results
        self.failure = None

    def setConstraint(self, constraint):
        self.constraint = constraint

    def complete(self, res):
        if self.broker:
            self.broker.removeRequest(self)
        if self.active:
            self.active = False
            self.deferred.callback(res)
        else:
            log.msg("PendingRequest.complete called on an inactive request")

    def fail(self, why):
        if self.active:
            if self.broker:
                self.broker.removeRequest(self)
            self.active = False
            self.failure = why
            if (self.broker and
                self.broker.tub and
                self.broker.tub.logRemoteFailures):

                my_short_tubid = "??"
                if self.broker.tub: # for tests
                    my_short_tubid = self.broker.tub.getShortTubID()
                their_short_tubid = self.broker.remote_tubref.getShortTubID()

                lp = log.msg("an outbound callRemote (that we [%s] sent to "
                             "someone else [%s]) failed on the far end"
                             % (my_short_tubid, their_short_tubid),
                             level=log.UNUSUAL)
                methname = ".".join([self.interfaceName or "?",
                                     self.methodName or "?"])
                log.msg(" reqID=%d, rref=%s, methname=%s"
                        % (self.reqID, self.rref, methname),
                        level=log.NOISY, parent=lp)
                #stack = why.getTraceback()
                # TODO: include the first few letters of the remote tubID in
                # this REMOTE tag
                #stack = "REMOTE: " + stack.replace("\n", "\nREMOTE: ")
                log.msg(" the REMOTE failure was:", failure=why,
                        level=log.NOISY, parent=lp)
                #log.msg(stack, level=log.NOISY, parent=lp)
            self.deferred.errback(why)
        else:
            log.msg("WEIRD: fail() on an inactive request", traceback=True)
            if self.failure:
                log.msg("multiple failures")
                log.msg("first one was:", self.failure)
                log.msg("this one was:", why)
                log.err("multiple failures indicate a problem")

class ArgumentSlicer(slicer.ScopedSlicer):
    opentype = ('arguments',)

    def __init__(self, args, kwargs):
        slicer.ScopedSlicer.__init__(self, None)
        self.args = args
        self.kwargs = kwargs
        self.which = ""

    def sliceBody(self, streamable, banana):
        yield len(self.args)
        for i,arg in enumerate(self.args):
            self.which = "arg[%d]" % i
            yield arg
        keys = self.kwargs.keys()
        keys.sort()
        for argname in keys:
            self.which = "arg[%s]" % argname
            yield argname
            yield self.kwargs[argname]

    def describe(self):
        return "<%s>" % self.which


class CallSlicer(slicer.ScopedSlicer):
    opentype = ('call',)

    def __init__(self, reqID, clid, methodname, args, kwargs):
        slicer.ScopedSlicer.__init__(self, None)
        self.reqID = reqID
        self.clid = clid
        self.methodname = methodname
        self.args = args
        self.kwargs = kwargs

    def sliceBody(self, streamable, banana):
        yield self.reqID
        yield self.clid
        yield self.methodname
        yield ArgumentSlicer(self.args, self.kwargs)

    def describe(self):
        return "<call-%s-%s-%s>" % (self.reqID, self.clid, self.methodname)

class InboundDelivery:
    """An inbound message that has not yet been delivered.

    This is created when a 'call' sequence has finished being received. The
    Broker will add it to a queue. The delivery at the head of the queue is
    serviced when all of its arguments have been resolved.

    The only way that the arguments might not all be available is if one of
    the Unslicers which created them has provided a 'ready_deferred' along
    with the prospective object. The only standard Unslicer which does this
    is the TheirReferenceUnslicer, which handles introductions. (custom
    Unslicers might also provide a ready_deferred, for example a URL
    slicer/unslicer pair for which the receiving end fetches the target of
    the URL as its value, or a UnixFD slicer/unslicer that had to wait for a
    side-channel unix-domain socket to finish transferring control over the
    FD to the recipient before being ready).

    Most Unslicers refuse to accept unready objects as their children (most
    implementations of receiveChild() do 'assert ready_deferred is None').
    The CallUnslicer is fairly unique in not rejecting such objects.

    We do require, however, that all of the arguments be at least
    referenceable. This is not generally a problem: the only time an
    unslicer's receiveChild() can get a non-referenceable object (represented
    by a Deferred) is if that unslicer is participating in a reference cycle
    that has not yet completed, and CallUnslicers only live at the top level,
    above any cycles.
    """

    def __init__(self, broker, reqID, obj,
                 interface, methodname, methodSchema,
                 allargs):
        self.broker = broker
        self.reqID = reqID
        self.obj = obj
        self.interface = interface
        self.methodname = methodname
        self.methodSchema = methodSchema
        self.allargs = allargs

    def logFailure(self, f):
        # called if tub.logLocalFailures is True
        my_short_tubid = "??"
        if self.broker.tub: # for tests
            my_short_tubid = self.broker.tub.getShortTubID()
        their_short_tubid = "<unauth>"
        if self.broker.remote_tubref:
            their_short_tubid = self.broker.remote_tubref.getShortTubID()
        lp = log.msg("an inbound callRemote that we [%s] executed (on behalf "
                     "of someone else, TubID %s) failed"
                     % (my_short_tubid, their_short_tubid),
                     level=log.UNUSUAL)
        if self.interface:
            methname = self.interface.getName() + "." + self.methodname
        else:
            methname = self.methodname
        log.msg(" reqID=%d, rref=%s, methname=%s" %
                (self.reqID, self.obj, methname),
                level=log.NOISY, parent=lp)
        log.msg(" args=%s" % (self.allargs.args,), level=log.NOISY, parent=lp)
        log.msg(" kwargs=%s" % (self.allargs.kwargs,),
                level=log.NOISY, parent=lp)
        #if isinstance(f.type, str):
        #    stack = "getTraceback() not available for string exceptions\n"
        #else:
        #    stack = f.getTraceback()
        # TODO: trim stack to everything below Broker._doCall
        #stack = "LOCAL: " + stack.replace("\n", "\nLOCAL: ")
        log.msg(" the LOCAL failure was:", failure=f,
                level=log.NOISY, parent=lp)
        #log.msg(stack, level=log.NOISY, parent=lp)

class ArgumentUnslicer(slicer.ScopedUnslicer):
    methodSchema = None
    debug = False

    def setConstraint(self, methodSchema):
        self.methodSchema = methodSchema

    def start(self, count):
        if self.debug:
            log.msg("%s.start: %s" % (self, count))
        self.numargs = None
        self.args = []
        self.kwargs = {}
        self.argname = None
        self.argConstraint = None
        self.num_unreferenceable_children = 0
        self._all_children_are_referenceable_d = None
        self._ready_deferreds = []
        self.closed = False

    def checkToken(self, typebyte, size):
        if self.numargs is None:
            # waiting for positional-arg count
            if typebyte != tokens.INT:
                raise BananaError("posarg count must be an INT")
            return
        if len(self.args) < self.numargs:
            # waiting for a positional arg
            if self.argConstraint:
                self.argConstraint.checkToken(typebyte, size)
            return
        if self.argname is None:
            # waiting for the name of a keyword arg
            if typebyte not in (tokens.STRING, tokens.VOCAB):
                raise BananaError("kwarg name must be a STRING")
            # TODO: limit to longest argument name of the method?
            return
        # waiting for the value of a kwarg
        if self.argConstraint:
            self.argConstraint.checkToken(typebyte, size)

    def doOpen(self, opentype):
        if self.argConstraint:
            self.argConstraint.checkOpentype(opentype)
        unslicer = self.open(opentype)
        if unslicer:
            if self.argConstraint:
                unslicer.setConstraint(self.argConstraint)
        return unslicer

    def receiveChild(self, token, ready_deferred=None):
        if self.debug:
            log.msg("%s.receiveChild: %s %s %s %s %s args=%s kwargs=%s" %
                    (self, self.closed, self.num_unreferenceable_children,
                     len(self._ready_deferreds), token, ready_deferred,
                     self.args, self.kwargs))
        if self.numargs is None:
            # this token is the number of positional arguments
            assert isinstance(token, int)
            assert ready_deferred is None
            self.numargs = token
            if self.numargs:
                ms = self.methodSchema
                if ms:
                    accept, self.argConstraint = \
                            ms.getPositionalArgConstraint(0)
                    assert accept
            return

        if len(self.args) < self.numargs:
            # this token is a positional argument
            argvalue = token
            argpos = len(self.args)
            self.args.append(argvalue)
            if isinstance(argvalue, defer.Deferred):
                # this may occur if the child is a gift which has not
                # resolved yet.
                self.num_unreferenceable_children += 1
                argvalue.addCallback(self.updateChild, argpos)
            if ready_deferred:
                if self.debug:
                    log.msg("%s.receiveChild got an unready posarg" % self)
                self._ready_deferreds.append(ready_deferred)
            if len(self.args) < self.numargs:
                # more to come
                ms = self.methodSchema
                if ms:
                    nextargnum = len(self.args)
                    accept, self.argConstraint = \
                            ms.getPositionalArgConstraint(nextargnum)
                    assert accept
            return

        if self.argname is None:
            # this token is the name of a keyword argument
            assert ready_deferred is None
            self.argname = token
            # if the argname is invalid, this may raise Violation
            ms = self.methodSchema
            if ms:
                accept, self.argConstraint = \
                        ms.getKeywordArgConstraint(self.argname,
                                                   self.numargs,
                                                   self.kwargs.keys())
                assert accept
            return

        # this token is the value of a keyword argument
        argvalue = token
        self.kwargs[self.argname] = argvalue
        if isinstance(argvalue, defer.Deferred):
            self.num_unreferenceable_children += 1
            argvalue.addCallback(self.updateChild, self.argname)
        if ready_deferred:
            if self.debug:
                log.msg("%s.receiveChild got an unready kwarg" % self)
            self._ready_deferreds.append(ready_deferred)
        self.argname = None
        return

    def updateChild(self, obj, which):
        # one of our arguments has just now become referenceable. Normal
        # types can't trigger this (since the arguments to a method form a
        # top-level serialization domain), but special Unslicers might. For
        # example, the Gift unslicer will eventually provide us with a
        # RemoteReference, but for now all we get is a Deferred as a
        # placeholder.

        if self.debug:
            log.msg("%s.updateChild, [%s] became referenceable: %s" %
                    (self, which, obj))
        if isinstance(which, int):
            self.args[which] = obj
        else:
            self.kwargs[which] = obj
        self.num_unreferenceable_children -= 1
        if self.num_unreferenceable_children == 0:
            if self._all_children_are_referenceable_d:
                self._all_children_are_referenceable_d.callback(None)
        return obj


    def receiveClose(self):
        if self.debug:
            log.msg("%s.receiveClose: %s %s %s" %
                    (self, self.closed, self.num_unreferenceable_children,
                     len(self._ready_deferreds)))
        if (self.numargs is None or
            len(self.args) < self.numargs or
            self.argname is not None):
            raise BananaError("'arguments' sequence ended too early")
        self.closed = True
        dl = []
        if self.num_unreferenceable_children:
            d = self._all_children_are_referenceable_d = defer.Deferred()
            dl.append(d)
        dl.extend(self._ready_deferreds)
        ready_deferred = None
        if dl:
            ready_deferred = AsyncAND(dl)
        return self, ready_deferred

    def describe(self):
        s = "<arguments"
        if self.numargs is not None:
            if len(self.args) < self.numargs:
                s += " arg[%d]" % len(self.args)
            else:
                if self.argname is not None:
                    s += " arg[%s]" % self.argname
                else:
                    s += " arg[?]"
        if self.closed:
            s += " closed"
            # TODO: it would be nice to indicate if we still have unready
            # children
        s += ">"
        return s


class CallUnslicer(slicer.ScopedUnslicer):

    debug = False

    def start(self, count):
        # start=0:reqID, 1:objID, 2:methodname, 3: arguments
        self.stage = 0
        self.reqID = None
        self.obj = None
        self.interface = None
        self.methodname = None
        self.methodSchema = None # will be a MethodArgumentsConstraint
        self._ready_deferreds = []

    def checkToken(self, typebyte, size):
        # TODO: limit strings by returning a number instead of None
        if self.stage == 0:
            if typebyte != tokens.INT:
                raise BananaError("request ID must be an INT")
        elif self.stage == 1:
            if typebyte not in (tokens.INT, tokens.NEG):
                raise BananaError("object ID must be an INT/NEG")
        elif self.stage == 2:
            if typebyte not in (tokens.STRING, tokens.VOCAB):
                raise BananaError("method name must be a STRING")
            # TODO: limit to longest method name of self.obj in the interface
        elif self.stage == 3:
            if typebyte != tokens.OPEN:
                raise BananaError("arguments must be an 'arguments' sequence")
        else:
            raise BananaError("too many objects given to CallUnslicer")

    def doOpen(self, opentype):
        # checkToken insures that this can only happen when we're receiving
        # an arguments object, so we don't have to bother checking self.stage
        assert self.stage == 3
        unslicer = self.open(opentype)
        if self.methodSchema:
            unslicer.setConstraint(self.methodSchema)
        return unslicer

    def reportViolation(self, f):
        # if the Violation is because we received an ABORT, then we know
        # that the sender knows there was a problem, so don't respond.
        if f.value.args[0] == "ABORT received":
            return f

        # if the Violation was raised after we know the reqID, we can send
        # back an Error.
        if self.stage > 0:
            self.broker.callFailed(f, self.reqID)
        return f # give up our sequence

    def receiveChild(self, token, ready_deferred=None):
        assert not isinstance(token, defer.Deferred)
        if self.debug:
            log.msg("%s.receiveChild [s%d]: %s" %
                    (self, self.stage, repr(token)))

        if self.stage == 0: # reqID
            # we don't yet know which reqID to send any failure to
            assert ready_deferred is None
            self.reqID = token
            self.stage = 1
            if self.reqID != 0:
                assert self.reqID not in self.broker.activeLocalCalls
                self.broker.activeLocalCalls[self.reqID] = self
            return

        if self.stage == 1: # objID
            # this might raise an exception if objID is invalid
            assert ready_deferred is None
            self.objID = token
            self.obj = self.broker.getMyReferenceByCLID(token)
            #iface = self.broker.getRemoteInterfaceByName(token)
            if self.objID < 0:
                self.interface = None
            else:
                self.interface = self.obj.getInterface()
            self.stage = 2
            return

        if self.stage == 2: # methodname
            # validate the methodname, get the schema. This may raise an
            # exception for unknown methods

            # must find the schema, using the interfaces

            # TODO: getSchema should probably be in an adapter instead of in
            # a pb.Referenceable base class. Old-style (unconstrained)
            # flavors.Referenceable should be adapted to something which
            # always returns None

            # TODO: make this faster. A likely optimization is to take a
            # tuple of components.getInterfaces(obj) and use it as a cache
            # key. It would be even faster to use obj.__class__, but that
            # would probably violate the expectation that instances can
            # define their own __implements__ (independently from their
            # class). If this expectation were to go away, a quick
            # obj.__class__ -> RemoteReferenceSchema cache could be built.

            assert ready_deferred is None
            self.stage = 3

            if self.objID < 0:
                # the target is a bound method, ignore the methodname
                self.methodSchema = getattr(self.obj, "methodSchema", None)
                self.methodname = None # TODO: give it something useful
                if self.broker.requireSchema and not self.methodSchema:
                    why = "This broker does not accept unconstrained " + \
                          "method calls"
                    raise Violation(why)
                return

            self.methodname = token

            if self.interface:
                # they are calling an interface+method pair
                ms = self.interface.get(self.methodname)
                if not ms:
                    why = "method '%s' not defined in %s" % \
                          (self.methodname, self.interface.__remote_name__)
                    raise Violation(why)
                self.methodSchema = ms

            return

        if self.stage == 3: # arguments
            assert isinstance(token, ArgumentUnslicer)
            self.allargs = token
            # queue the message. It will not be executed until all the
            # arguments are ready. The .args list and .kwargs dict may change
            # before then.
            if ready_deferred:
                self._ready_deferreds.append(ready_deferred)
            self.stage = 4
            return

    def receiveClose(self):
        if self.stage != 4:
            raise BananaError("'call' sequence ended too early")
        # time to create the InboundDelivery object so we can queue it