twisted/web/distrib.py

   1 # -*- test-case-name: twisted.web.test.test_distrib -*-
   2 # Copyright (c) Twisted Matrix Laboratories.
   3 # See LICENSE for details.
   4
   5 """
   6 Distributed web servers.
   7
   8 This is going to have to be refactored so that argument parsing is done
   9 by each subprocess and not by the main web server (i.e. GET, POST etc.).
  10 """
  11
  12 # System Imports
  13 import types, os, copy, cStringIO
  14 try:
  15     import pwd
  16 except ImportError:
  17     pwd = None
  18
  19 from xml.dom.minidom import Element, Text
  20
  21 # Twisted Imports
  22 from twisted.spread import pb
  23 from twisted.spread.banana import SIZE_LIMIT
  24 from twisted.web import http, resource, server, html, static
  25 from twisted.web.http_headers import Headers
  26 from twisted.python import log
  27 from twisted.persisted import styles
  28 from twisted.internet import address, reactor
  29
  30
  31 class _ReferenceableProducerWrapper(pb.Referenceable):
  32     def __init__(self, producer):
  33         self.producer = producer
  34
  35     def remote_resumeProducing(self):
  36         self.producer.resumeProducing()
  37
  38     def remote_pauseProducing(self):
  39         self.producer.pauseProducing()
  40
  41     def remote_stopProducing(self):
  42         self.producer.stopProducing()
  43
  44
  45 class Request(pb.RemoteCopy, server.Request):
  46     """
  47     A request which was received by a L{ResourceSubscription} and sent via
  48     PB to a distributed node.
  49     """
  50     def setCopyableState(self, state):
  51         """
  52         Initialize this L{twisted.web.distrib.Request} based on the copied
  53         state so that it closely resembles a L{twisted.web.server.Request}.
  54         """
  55         for k in 'host', 'client':
  56             tup = state[k]
  57             addrdesc = {'INET': 'TCP', 'UNIX': 'UNIX'}[tup[0]]
  58             addr = {'TCP': lambda: address.IPv4Address(addrdesc,
  59                                                        tup[1], tup[2]),
  60                     'UNIX': lambda: address.UNIXAddress(tup[1])}[addrdesc]()
  61             state[k] = addr
  62         state['requestHeaders'] = Headers(dict(state['requestHeaders']))
  63         pb.RemoteCopy.setCopyableState(self, state)
  64         # Emulate the local request interface --
  65         self.content = cStringIO.StringIO(self.content_data)
  66         self.finish           = self.remote.remoteMethod('finish')
  67         self.setHeader        = self.remote.remoteMethod('setHeader')
  68         self.addCookie        = self.remote.remoteMethod('addCookie')
  69         self.setETag          = self.remote.remoteMethod('setETag')
  70         self.setResponseCode  = self.remote.remoteMethod('setResponseCode')
  71         self.setLastModified  = self.remote.remoteMethod('setLastModified')
  72
  73         # To avoid failing if a resource tries to write a very long string
  74         # all at once, this one will be handled slightly differently.
  75         self._write = self.remote.remoteMethod('write')
  76
  77
  78     def write(self, bytes):
  79         """
  80         Write the given bytes to the response body.
  81
  82         @param bytes: The bytes to write.  If this is longer than 640k, it
  83             will be split up into smaller pieces.
  84         """
  85         start = 0
  86         end = SIZE_LIMIT
  87         while True:
  88             self._write(bytes[start:end])
  89             start += SIZE_LIMIT
  90             end += SIZE_LIMIT
  91             if start >= len(bytes):
  92                 break
  93
  94
  95     def registerProducer(self, producer, streaming):
  96         self.remote.callRemote("registerProducer",
  97                                _ReferenceableProducerWrapper(producer),
  98                                streaming).addErrback(self.fail)
  99
 100     def unregisterProducer(self):
 101         self.remote.callRemote("unregisterProducer").addErrback(self.fail)
 102
 103     def fail(self, failure):
 104         log.err(failure)
 105
 106
 107 pb.setUnjellyableForClass(server.Request, Request)
 108
 109 class Issue:
 110     def __init__(self, request):
 111         self.request = request
 112
 113     def finished(self, result):
 114         if result != server.NOT_DONE_YET:
 115             assert isinstance(result, types.StringType),\
 116                    "return value not a string"
 117             self.request.write(result)
 118             self.request.finish()
 119
 120     def failed(self, failure):
 121         #XXX: Argh. FIXME.
 122         failure = str(failure)
 123         self.request.write(
 124             resource.ErrorPage(http.INTERNAL_SERVER_ERROR,
 125                                "Server Connection Lost",
 126                                "Connection to distributed server lost:" +
 127                                html.PRE(failure)).
 128             render(self.request))
 129         self.request.finish()
 130         log.msg(failure)
 131
 132
 133 class ResourceSubscription(resource.Resource):
 134     isLeaf = 1
 135     waiting = 0
 136     def __init__(self, host, port):
 137         resource.Resource.__init__(self)
 138         self.host = host
 139         self.port = port
 140         self.pending = []
 141         self.publisher = None
 142
 143     def __getstate__(self):
 144         """Get persistent state for this ResourceSubscription.
 145         """
 146         # When I unserialize,
 147         state = copy.copy(self.__dict__)
 148         # Publisher won't be connected...
 149         state['publisher'] = None
 150         # I won't be making a connection
 151         state['waiting'] = 0
 152         # There will be no pending requests.
 153         state['pending'] = []
 154         return state
 155
 156     def connected(self, publisher):
 157         """I've connected to a publisher; I'll now send all my requests.
 158         """
 159         log.msg('connected to publisher')
 160         publisher.broker.notifyOnDisconnect(self.booted)
 161         self.publisher = publisher
 162         self.waiting = 0
 163         for request in self.pending:
 164             self.render(request)
 165         self.pending = []
 166
 167     def notConnected(self, msg):
 168         """I can't connect to a publisher; I'll now reply to all pending
 169         requests.
 170         """
 171         log.msg("could not connect to distributed web service: %s" % msg)
 172         self.waiting = 0
 173         self.publisher = None
 174         for request in self.pending:
 175             request.write("Unable to connect to distributed server.")
 176             request.finish()
 177         self.pending = []
 178
 179     def booted(self):
 180         self.notConnected("connection dropped")
 181
 182     def render(self, request):
 183         """Render this request, from my server.
 184
 185         This will always be asynchronous, and therefore return NOT_DONE_YET.
 186         It spins off a request to the pb client, and either adds it to the list
 187         of pending issues or requests it immediately, depending on if the
 188         client is already connected.
 189         """
 190         if not self.publisher:
 191             self.pending.append(request)
 192             if not self.waiting:
 193                 self.waiting = 1
 194                 bf = pb.PBClientFactory()
 195                 timeout = 10
 196                 if self.host == "unix":
 197                     reactor.connectUNIX(self.port, bf, timeout)
 198                 else:
 199                     reactor.connectTCP(self.host, self.port, bf, timeout)
 200                 d = bf.getRootObject()
 201                 d.addCallbacks(self.connected, self.notConnected)
 202
 203         else:
 204             i = Issue(request)
 205             self.publisher.callRemote('request', request).addCallbacks(i.finished, i.failed)
 206         return server.NOT_DONE_YET
 207
 208
 209
 210 class ResourcePublisher(pb.Root, styles.Versioned):
 211     """
 212     L{ResourcePublisher} exposes a remote API which can be used to respond
 213     to request.
 214
 215     @ivar site: The site which will be used for resource lookup.
 216     @type site: L{twisted.web.server.Site}
 217     """
 218     def __init__(self, site):
 219         self.site = site
 220
 221     persistenceVersion = 2
 222
 223     def upgradeToVersion2(self):
 224         self.application.authorizer.removeIdentity("web")
 225         del self.application.services[self.serviceName]
 226         del self.serviceName
 227         del self.application
 228         del self.perspectiveName
 229
 230     def getPerspectiveNamed(self, name):
 231         return self
 232
 233
 234     def remote_request(self, request):
 235         """
 236         Look up the resource for the given request and render it.
 237         """
 238         res = self.site.getResourceFor(request)
 239         log.msg( request )
 240         result = res.render(request)
 241         if result is not server.NOT_DONE_YET:
 242             request.write(result)
 243             request.finish()
 244         return server.NOT_DONE_YET
 245
 246
 247
 248 class UserDirectory(resource.Resource):
 249     """
 250     A resource which lists available user resources and serves them as
 251     children.
 252
 253     @ivar _pwd: An object like L{pwd} which is used to enumerate users and
 254         their home directories.
 255     """
 256
 257     userDirName = 'public_html'
 258     userSocketName = '.twistd-web-pb'
 259
 260     template = """
 261 <html>
 262     <head>
 263     <title>twisted.web.distrib.UserDirectory</title>
 264     <style>
 265
 266     a
 267     {
 268         font-family: Lucida, Verdana, Helvetica, Arial, sans-serif;
 269         color: #369;
 270         text-decoration: none;
 271     }
 272
 273     th
 274     {
 275         font-family: Lucida, Verdana, Helvetica, Arial, sans-serif;
 276         font-weight: bold;
 277         text-decoration: none;
 278         text-align: left;
 279     }
 280
 281     pre, code
 282     {
 283         font-family: "Courier New", Courier, monospace;
 284     }
 285
 286     p, body, td, ol, ul, menu, blockquote, div
 287     {
 288         font-family: Lucida, Verdana, Helvetica, Arial, sans-serif;
 289         color: #000;
 290     }
 291     </style>
 292     </head>
 293
 294     <body>
 295     <h1>twisted.web.distrib.UserDirectory</h1>
 296
 297     %(users)s
 298 </body>
 299 </html>
 300 """
 301
 302     def __init__(self, userDatabase=None):
 303         resource.Resource.__init__(self)
 304         if userDatabase is None:
 305             userDatabase = pwd
 306         self._pwd = userDatabase
 307
 308
 309     def _users(self):
 310         """
 311         Return a list of two-tuples giving links to user resources and text to
 312         associate with those links.
 313         """
 314         users = []
 315         for user in self._pwd.getpwall():
 316             name, passwd, uid, gid, gecos, dir, shell = user
 317             realname = gecos.split(',')[0]
 318             if not realname:
 319                 realname = name
 320             if os.path.exists(os.path.join(dir, self.userDirName)):
 321                 users.append((name, realname + ' (file)'))
 322             twistdsock = os.path.join(dir, self.userSocketName)
 323             if os.path.exists(twistdsock):
 324                 linkName = name + '.twistd'
 325                 users.append((linkName, realname + ' (twistd)'))
 326         return users
 327
 328
 329     def render_GET(self, request):
 330         """
 331         Render as HTML a listing of all known users with links to their
 332         personal resources.
 333         """
 334         listing = Element('ul')
 335         for link, text in self._users():
 336             linkElement = Element('a')
 337             linkElement.setAttribute('href', link + '/')
 338             textNode = Text()
 339             textNode.data = text
 340             linkElement.appendChild(textNode)
 341             item = Element('li')
 342             item.appendChild(linkElement)
 343             listing.appendChild(item)
 344         return self.template % {'users': listing.toxml()}
 345
 346
 347     def getChild(self, name, request):
 348         if name == '':
 349             return self
 350
 351         td = '.twistd'
 352
 353         if name[-len(td):] == td:
 354             username = name[:-len(td)]
 355             sub = 1
 356         else:
 357             username = name
 358             sub = 0
 359         try:
 360             pw_name, pw_passwd, pw_uid, pw_gid, pw_gecos, pw_dir, pw_shell \
 361                      = self._pwd.getpwnam(username)
 362         except KeyError:
 363             return resource.NoResource()
 364         if sub:
 365             twistdsock = os.path.join(pw_dir, self.userSocketName)
 366             rs = ResourceSubscription('unix',twistdsock)
 367             self.putChild(name, rs)
 368             return rs
 369         else:
 370             path = os.path.join(pw_dir, self.userDirName)
 371             if not os.path.exists(path):
 372                 return resource.NoResource()
 373             return static.File(path)