Package pyamf :: Package remoting :: Package gateway :: Module twisted

[hide private]
[frames] | no frames]

# Source Code for Module pyamf.remoting.gateway.twisted

      1  # Copyright (c) 2007-2009 The PyAMF Project.
      2  # See LICENSE for details.
      3
      4  """
      5  Twisted server implementation.
      6
      7  This gateway allows you to expose functions in Twisted to AMF
      8  clients and servers.
      9
     10  @see: U{Twisted homepage (external)<http://twistedmatrix.com>}
     11
     12  @since: 0.1.0
     13  """
     14
     15  import sys, os.path
     16
     17  try:
     18      sys.path.remove('')
     19  except ValueError:
     20      pass
     21
     22  try:
     23      sys.path.remove(os.path.dirname(os.path.abspath(__file__)))
     24  except ValueError:
     25      pass
     26
     27  twisted = __import__('twisted')
     28  __import__('twisted.internet.defer')
     29  __import__('twisted.internet.threads')
     30  __import__('twisted.web.resource')
     31  __import__('twisted.web.server')
     32
     33  defer = twisted.internet.defer
     34  threads = twisted.internet.threads
     35  resource = twisted.web.resource
     36  server = twisted.web.server
     37
     38  import pyamf
     39  from pyamf import remoting
     40  from pyamf.remoting import gateway, amf0, amf3
     41
     42  __all__ = ['TwistedGateway']
     43
     44 -class AMF0RequestProcessor(amf0.RequestProcessor):
     45      """
     46      A Twisted friendly implementation of
     47      L{amf0.RequestProcessor<pyamf.remoting.amf0.RequestProcessor>}
     48      """
     49
     50 -    def __call__(self, request, *args, **kwargs):
     51          """
     52          Calls the underlying service method.
     53
     54          @return: A C{Deferred} that will contain the AMF L{Response}.
     55          @rtype: C{twisted.internet.defer.Deferred}
     56          """
     57          try:
     58              service_request = self.gateway.getServiceRequest(request,
     59                                                               request.target)
     60          except gateway.UnknownServiceError, e:
     61              return defer.succeed(self.buildErrorResponse(request))
     62
     63          response = remoting.Response(None)
     64          deferred_response = defer.Deferred()
     65
     66          def eb(failure):
     67              errMesg = "%s: %s" % (failure.type, failure.getErrorMessage())
     68              self.gateway.logger.error(errMesg)
     69              self.gateway.logger.info(failure.getTraceback())
     70              deferred_response.callback(self.buildErrorResponse(
     71                  request, (failure.type, failure.value, failure.tb)))
     72
     73          def response_cb(result):
     74              self.gateway.logger.debug("AMF Response: %s" % (result,))
     75              response.body = result
     76
     77              deferred_response.callback(response)
     78
     79          def preprocess_cb(result):
     80              d = defer.maybeDeferred(self._getBody, request, response,
     81                                      service_request, **kwargs)
     82              d.addCallback(response_cb).addErrback(eb)
     83
     84          def auth_cb(result):
     85              if result is not True:
     86                  response.status = remoting.STATUS_ERROR
     87                  response.body = remoting.ErrorFault(code='AuthenticationError',
     88                      description='Authentication failed')
     89
     90                  deferred_response.callback(response)
     91
     92                  return
     93
     94              d = defer.maybeDeferred(self.gateway.preprocessRequest,
     95                                      service_request, *args, **kwargs)
     96              d.addCallback(preprocess_cb).addErrback(eb)
     97
     98          # we have a valid service, now attempt authentication
     99          d = defer.maybeDeferred(self.authenticateRequest, request,
    100                                  service_request, **kwargs)
    101          d.addCallback(auth_cb).addErrback(eb)
    102
    103          return deferred_response
    104
    105 -class AMF3RequestProcessor(amf3.RequestProcessor):
    106      """
    107      A Twisted friendly implementation of
    108      L{amf3.RequestProcessor<pyamf.remoting.amf3.RequestProcessor>}
    109      """
    110
    111 -    def _processRemotingMessage(self, amf_request, ro_request, **kwargs):
    112          ro_response = amf3.generate_acknowledgement(ro_request)
    113          amf_response = remoting.Response(ro_response, status=remoting.STATUS_OK)
    114
    115          try:
    116              service_name = ro_request.operation
    117
    118              if hasattr(ro_request, 'destination') and ro_request.destination:
    119                  service_name = '%s.%s' % (ro_request.destination, service_name)
    120
    121              service_request = self.gateway.getServiceRequest(amf_request,
    122                                                               service_name)
    123          except gateway.UnknownServiceError, e:
    124              return defer.succeed(remoting.Response(
    125                  self.buildErrorResponse(ro_request),
    126                  status=remoting.STATUS_ERROR))
    127
    128          deferred_response = defer.Deferred()
    129
    130          def eb(failure):
    131              errMesg = "%s: %s" % (failure.type, failure.getErrorMessage())
    132              self.gateway.logger.error(errMesg)
    133              self.gateway.logger.info(failure.getTraceback())
    134              ro_response = self.buildErrorResponse(ro_request, (failure.type,
    135                                                    failure.value, failure.tb))
    136              deferred_response.callback(remoting.Response(ro_response,
    137                                          status=remoting.STATUS_ERROR))
    138
    139          def response_cb(result):
    140              ro_response.body = result
    141              res = remoting.Response(ro_response)
    142              self.gateway.logger.debug("AMF Response: %r" % (res,))
    143
    144              deferred_response.callback(res)
    145
    146          def process_cb(result):
    147              d = defer.maybeDeferred(self.gateway.callServiceRequest,
    148                                      service_request, *ro_request.body, **kwargs)
    149              d.addCallback(response_cb).addErrback(eb)
    150
    151          d = defer.maybeDeferred(self.gateway.preprocessRequest, service_request,
    152                                  *ro_request.body, **kwargs)
    153          d.addCallback(process_cb).addErrback(eb)
    154
    155          return deferred_response
    156
    157 -    def __call__(self, amf_request, **kwargs):
    158          """
    159          Calls the underlying service method.
    160
    161          @return: A C{deferred} that will contain the AMF L{Response}.
    162          @rtype: C{Deferred<twisted.internet.defer.Deferred>}
    163          """
    164          deferred_response = defer.Deferred()
    165          ro_request = amf_request.body[0]
    166
    167          def cb(amf_response):
    168              deferred_response.callback(amf_response)
    169
    170          def eb(failure):
    171              errMesg = "%s: %s" % (failure.type, failure.getErrorMessage())
    172              self.gateway.logger.error(errMesg)
    173              self.gateway.logger.info(failure.getTraceback())
    174              deferred_response.callback(self.buildErrorResponse(ro_request,
    175                  (failure.type, failure.value, failure.tb)))
    176
    177          d = defer.maybeDeferred(self._getBody, amf_request, ro_request, **kwargs)
    178          d.addCallback(cb).addErrback(eb)
    179
    180          return deferred_response
    181
    182 -class TwistedGateway(gateway.BaseGateway, resource.Resource):
    183      """
    184      Twisted Remoting gateway for C{twisted.web}.
    185
    186      @ivar expose_request: Forces the underlying HTTP request to be the first
    187          argument to any service call.
    188      @type expose_request: C{bool}
    189      """
    190
    191      allowedMethods = ('POST',)
    192
    193 -    def __init__(self, *args, **kwargs):
    194          if 'expose_request' not in kwargs:
    195              kwargs['expose_request'] = True
    196
    197          gateway.BaseGateway.__init__(self, *args, **kwargs)
    198          resource.Resource.__init__(self)
    199
    200 -    def _finaliseRequest(self, request, status, content, mimetype='text/plain'):
    201          """
    202          Finalises the request.
    203
    204          @param request: The HTTP Request.
    205          @type request: C{http.Request}
    206          @param status: The HTTP status code.
    207          @type status: C{int}
    208          @param content: The content of the response.
    209          @type content: C{str}
    210          @param mimetype: The MIME type of the request.
    211          @type mimetype: C{str}
    212          """
    213          request.setResponseCode(status)
    214
    215          request.setHeader("Content-Type", mimetype)
    216          request.setHeader("Content-Length", str(len(content)))
    217          request.setHeader("Server", gateway.SERVER_NAME)
    218
    219          request.write(content)
    220          request.finish()
    221
    222 -    def render_POST(self, request):
    223          """
    224          Read remoting request from the client.
    225
    226          @type request: The HTTP Request.
    227          @param request: C{twisted.web.http.Request}
    228          """
    229          def handleDecodeError(failure):
    230              """
    231              Return HTTP 400 Bad Request.
    232              """
    233              errMesg = "%s: %s" % (failure.type, failure.getErrorMessage())
    234              self.logger.error(errMesg)
    235              self.logger.info(failure.getTraceback())
    236
    237              body = "400 Bad Request\n\nThe request body was unable to " \
    238                  "be successfully decoded."
    239
    240              if self.debug:
    241                  body += "\n\nTraceback:\n\n%s" % failure.getTraceback()
    242
    243              self._finaliseRequest(request, 400, body)
    244
    245          request.content.seek(0, 0)
    246          context = pyamf.get_context(pyamf.AMF0)
    247
    248          d = threads.deferToThread(remoting.decode, request.content.read(),
    249              context, strict=self.strict)
    250
    251          def cb(amf_request):
    252              self.logger.debug("AMF Request: %r" % amf_request)
    253              x = self.getResponse(request, amf_request)
    254
    255              x.addCallback(self.sendResponse, request, context)
    256
    257          # Process the request
    258          d.addCallback(cb).addErrback(handleDecodeError)
    259
    260          return server.NOT_DONE_YET
    261
    262 -    def sendResponse(self, amf_response, request, context):
    263          def cb(result):
    264              self._finaliseRequest(request, 200, result.getvalue(),
    265                  remoting.CONTENT_TYPE)
    266
    267          def eb(failure):
    268              """
    269              Return 500 Internal Server Error.
    270              """
    271              errMesg = "%s: %s" % (failure.type, failure.getErrorMessage())
    272              self.logger.error(errMesg)
    273              self.logger.info(failure.getTraceback())
    274
    275              body = "500 Internal Server Error\n\nThere was an error encoding" \
    276                  " the response."
    277
    278              if self.debug:
    279                  body += "\n\nTraceback:\n\n%s" % failure.getTraceback()
    280
    281              self._finaliseRequest(request, 500, body)
    282
    283          d = threads.deferToThread(remoting.encode, amf_response, context, strict=self.strict)
    284
    285          d.addCallback(cb).addErrback(eb)
    286
    287 -    def getProcessor(self, request):
    288          """
    289          Determines the request processor, based on the request.
    290
    291          @param request: The AMF message.
    292          @type request: L{Request<pyamf.remoting.Request>}
    293          """
    294          if request.target == 'null':
    295              return AMF3RequestProcessor(self)
    296
    297          return AMF0RequestProcessor(self)
    298
    299 -    def getResponse(self, http_request, amf_request):
    300          """
    301          Processes the AMF request, returning an AMF L{Response}.
    302
    303          @param http_request: The underlying HTTP Request
    304          @type http_request: C{twisted.web.http.Request}
    305          @param amf_request: The AMF Request.
    306          @type amf_request: L{Envelope<pyamf.remoting.Envelope>}
    307          """
    308          response = remoting.Envelope(amf_request.amfVersion,
    309                                       amf_request.clientType)
    310          dl = []
    311
    312          def cb(body, name):
    313              response[name] = body
    314
    315          for name, message in amf_request:
    316              processor = self.getProcessor(message)
    317
    318              d = defer.maybeDeferred(processor, message,
    319                                      http_request=http_request)
    320              d.addCallback(cb, name)
    321
    322              dl.append(d)
    323
    324          def cb2(result):
    325              return response
    326
    327          def eb(failure):
    328              """
    329              Return 500 Internal Server Error.
    330              """
    331              errMesg = "%s: %s" % (failure.type, failure.getErrorMessage())
    332
    333              self.logger.error(errMesg)
    334              self.logger.info(failure.getTraceback())
    335
    336              body = "500 Internal Server Error\n\nThe request was unable to " \
    337                  "be successfully processed."
    338
    339              if self.debug:
    340                  body += "\n\nTraceback:\n\n%s" % failure.getTraceback()
    341
    342              self._finaliseRequest(http_request, 500, body)
    343
    344          d = defer.DeferredList(dl)
    345
    346          return d.addCallback(cb2).addErrback(eb)
    347
    348 -    def authenticateRequest(self, service_request, username, password, **kwargs):
    349          """
    350          Processes an authentication request. If no authenticator is supplied,
    351          then authentication succeeds.
    352
    353          @return: C{Deferred}.
    354          @rtype: C{twisted.internet.defer.Deferred}
    355          """
    356          authenticator = self.getAuthenticator(service_request)
    357          self.logger.debug('Authenticator expands to: %r' % authenticator)
    358
    359          if authenticator is None:
    360              return defer.succeed(True)
    361
    362          args = (username, password)
    363
    364          if hasattr(authenticator, '_pyamf_expose_request'):
    365              http_request = kwargs.get('http_request', None)
    366              args = (http_request,) + args
    367
    368          return defer.maybeDeferred(authenticator, *args)
    369
    370 -    def preprocessRequest(self, service_request, *args, **kwargs):
    371          """
    372          Preprocesses a request.
    373          """
    374          processor = self.getPreprocessor(service_request)
    375          self.logger.debug('Preprocessor expands to: %r' % processor)
    376
    377          if processor is None:
    378              return
    379
    380          args = (service_request,) + args
    381
    382          if hasattr(processor, '_pyamf_expose_request'):
    383              http_request = kwargs.get('http_request', None)
    384              args = (http_request,) + args
    385
    386          return defer.maybeDeferred(processor, *args)
    387