api/escalator/common/wsgi.py

   1 # Copyright 2010 United States Government as represented by the
   2 # Administrator of the National Aeronautics and Space Administration.
   3 # Copyright 2010 OpenStack Foundation
   4 # Copyright 2014 IBM Corp.
   5 # All Rights Reserved.
   6 #
   7 #    Licensed under the Apache License, Version 2.0 (the "License"); you may
   8 #    not use this file except in compliance with the License. You may obtain
   9 #    a copy of the License at
  10 #
  11 #         http://www.apache.org/licenses/LICENSE-2.0
  12 #
  13 #    Unless required by applicable law or agreed to in writing, software
  14 #    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
  15 #    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
  16 #    License for the specific language governing permissions and limitations
  17 #    under the License.
  18
  19 """
  20 Utility methods for working with WSGI servers
  21 """
  22 from __future__ import print_function
  23
  24 import errno
  25 import functools
  26 import os
  27 import signal
  28 import sys
  29 import time
  30
  31 import eventlet
  32 from eventlet.green import socket
  33 from eventlet.green import ssl
  34 import eventlet.greenio
  35 import eventlet.wsgi
  36 from oslo_serialization import jsonutils
  37 from oslo_concurrency import processutils
  38 from oslo_config import cfg
  39 from oslo_log import log as logging
  40 from oslo_log import loggers
  41 import routes
  42 import routes.middleware
  43 import six
  44 import webob.dec
  45 import webob.exc
  46 from webob import multidict
  47
  48 from escalator.common import exception
  49 from escalator.common import utils
  50 from escalator import i18n
  51
  52
  53 _ = i18n._
  54 _LE = i18n._LE
  55 _LI = i18n._LI
  56 _LW = i18n._LW
  57
  58 bind_opts = [
  59     cfg.StrOpt('bind_host', default='0.0.0.0',
  60                help=_('Address to bind the server.  Useful when '
  61                       'selecting a particular network interface.')),
  62     cfg.IntOpt('bind_port',
  63                help=_('The port on which the server will listen.')),
  64 ]
  65
  66 socket_opts = [
  67     cfg.IntOpt('backlog', default=4096,
  68                help=_('The backlog value that will be used when creating the '
  69                       'TCP listener socket.')),
  70     cfg.IntOpt('tcp_keepidle', default=600,
  71                help=_('The value for the socket option TCP_KEEPIDLE.  This is '
  72                       'the time in seconds that the connection must be idle '
  73                       'before TCP starts sending keepalive probes.')),
  74     cfg.StrOpt('ca_file', help=_('CA certificate file to use to verify '
  75                                  'connecting clients.')),
  76     cfg.StrOpt('cert_file', help=_('Certificate file to use when starting API '
  77                                    'server securely.')),
  78     cfg.StrOpt('key_file', help=_('Private key file to use when starting API '
  79                                   'server securely.')),
  80 ]
  81
  82 eventlet_opts = [
  83     cfg.IntOpt('workers', default=processutils.get_worker_count(),
  84                help=_('The number of child process workers that will be '
  85                       'created to service requests. The default will be '
  86                       'equal to the number of CPUs available.')),
  87     cfg.IntOpt('max_header_line', default=16384,
  88                help=_('Maximum line size of message headers to be accepted. '
  89                       'max_header_line may need to be increased when using '
  90                       'large tokens (typically those generated by the '
  91                       'Keystone v3 API with big service catalogs')),
  92     cfg.BoolOpt('http_keepalive', default=True,
  93                 help=_('If False, server will return the header '
  94                        '"Connection: close", '
  95                        'If True, server will return "Connection: Keep-Alive" '
  96                        'in its responses. In order to close the client socket '
  97                        'connection explicitly after the response is sent and '
  98                        'read successfully by the client, you simply have to '
  99                        'set this option to False when you create a wsgi '
 100                        'server.')),
 101 ]
 102
 103 profiler_opts = [
 104     cfg.BoolOpt("enabled", default=False,
 105                 help=_('If False fully disable profiling feature.')),
 106     cfg.BoolOpt("trace_sqlalchemy", default=False,
 107                 help=_("If False doesn't trace SQL requests."))
 108 ]
 109
 110
 111 LOG = logging.getLogger(__name__)
 112
 113 CONF = cfg.CONF
 114 CONF.register_opts(bind_opts)
 115 CONF.register_opts(socket_opts)
 116 CONF.register_opts(eventlet_opts)
 117 CONF.register_opts(profiler_opts, group="profiler")
 118
 119 ASYNC_EVENTLET_THREAD_POOL_LIST = []
 120
 121
 122 def get_bind_addr(default_port=None):
 123     """Return the host and port to bind to."""
 124     return (CONF.bind_host, CONF.bind_port or default_port)
 125
 126
 127 def ssl_wrap_socket(sock):
 128     """
 129     Wrap an existing socket in SSL
 130
 131     :param sock: non-SSL socket to wrap
 132
 133     :returns: An SSL wrapped socket
 134     """
 135     utils.validate_key_cert(CONF.key_file, CONF.cert_file)
 136
 137     ssl_kwargs = {
 138         'server_side': True,
 139         'certfile': CONF.cert_file,
 140         'keyfile': CONF.key_file,
 141         'cert_reqs': ssl.CERT_NONE,
 142     }
 143
 144     if CONF.ca_file:
 145         ssl_kwargs['ca_certs'] = CONF.ca_file
 146         ssl_kwargs['cert_reqs'] = ssl.CERT_REQUIRED
 147
 148     return ssl.wrap_socket(sock, **ssl_kwargs)
 149
 150
 151 def get_socket(default_port):
 152     """
 153     Bind socket to bind ip:port in conf
 154
 155     note: Mostly comes from Swift with a few small changes...
 156
 157     :param default_port: port to bind to if none is specified in conf
 158
 159     :returns : a socket object as returned from socket.listen or
 160                ssl.wrap_socket if conf specifies cert_file
 161     """
 162     bind_addr = get_bind_addr(default_port)
 163
 164     # TODO(jaypipes): eventlet's greened socket module does not actually
 165     # support IPv6 in getaddrinfo(). We need to get around this in the
 166     # future or monitor upstream for a fix
 167     address_family = [
 168         addr[0] for addr in socket.getaddrinfo(bind_addr[0],
 169                                                bind_addr[1],
 170                                                socket.AF_UNSPEC,
 171                                                socket.SOCK_STREAM)
 172         if addr[0] in (socket.AF_INET, socket.AF_INET6)
 173     ][0]
 174
 175     use_ssl = CONF.key_file or CONF.cert_file
 176     if use_ssl and (not CONF.key_file or not CONF.cert_file):
 177         raise RuntimeError(_("When running server in SSL mode, you must "
 178                              "specify both a cert_file and key_file "
 179                              "option value in your configuration file"))
 180
 181     sock = utils.get_test_suite_socket()
 182     retry_until = time.time() + 30
 183
 184     while not sock and time.time() < retry_until:
 185         try:
 186             sock = eventlet.listen(bind_addr,
 187                                    backlog=CONF.backlog,
 188                                    family=address_family)
 189         except socket.error as err:
 190             if err.args[0] != errno.EADDRINUSE:
 191                 raise
 192             eventlet.sleep(0.1)
 193     if not sock:
 194         raise RuntimeError(_("Could not bind to %(host)s:%(port)s after"
 195                              " trying for 30 seconds") %
 196                            {'host': bind_addr[0],
 197                             'port': bind_addr[1]})
 198
 199     return sock
 200
 201
 202 def set_eventlet_hub():
 203     try:
 204         eventlet.hubs.use_hub('poll')
 205     except Exception:
 206         try:
 207             eventlet.hubs.use_hub('selects')
 208         except Exception:
 209             msg = _("eventlet 'poll' nor 'selects' hubs are available "
 210                     "on this platform")
 211             raise exception.WorkerCreationFailure(
 212                 reason=msg)
 213
 214
 215 def get_asynchronous_eventlet_pool(size=1000):
 216     """Return eventlet pool to caller.
 217
 218     Also store pools created in global list, to wait on
 219     it after getting signal for graceful shutdown.
 220
 221     :param size: eventlet pool size
 222     :returns: eventlet pool
 223     """
 224     global ASYNC_EVENTLET_THREAD_POOL_LIST
 225
 226     pool = eventlet.GreenPool(size=size)
 227     # Add pool to global ASYNC_EVENTLET_THREAD_POOL_LIST
 228     ASYNC_EVENTLET_THREAD_POOL_LIST.append(pool)
 229
 230     return pool
 231
 232
 233 class Server(object):
 234     """Server class to manage multiple WSGI sockets and applications.
 235
 236     This class requires initialize_escalator_store set to True if
 237     escalator store needs to be initialized.
 238     """
 239
 240     def __init__(self, threads=1000, initialize_escalator_store=False):
 241         os.umask(0o27)  # ensure files are created with the correct privileges
 242         self._logger = logging.getLogger("eventlet.wsgi.server")
 243         self._wsgi_logger = loggers.WritableLogger(self._logger)
 244         self.threads = threads
 245         self.children = set()
 246         self.stale_children = set()
 247         self.running = True
 248         # NOTE(abhishek): Allows us to only re-initialize escalator_store when
 249         # the API's configuration reloads.
 250         self.initialize_escalator_store = initialize_escalator_store
 251         self.pgid = os.getpid()
 252         try:
 253             # NOTE(flaper87): Make sure this process
 254             # runs in its own process group.
 255             os.setpgid(self.pgid, self.pgid)
 256         except OSError:
 257             # NOTE(flaper87): When running escalator-control,
 258             # (escalator's functional tests, for example)
 259             # setpgid fails with EPERM as escalator-control
 260             # creates a fresh session, of which the newly
 261             # launched service becomes the leader (session
 262             # leaders may not change process groups)
 263             #
 264             # Running escalator-(api) is safe and
 265             # shouldn't raise any error here.
 266             self.pgid = 0
 267
 268     def hup(self, *args):
 269         """
 270         Reloads configuration files with zero down time
 271         """
 272         signal.signal(signal.SIGHUP, signal.SIG_IGN)
 273         raise exception.SIGHUPInterrupt
 274
 275     def kill_children(self, *args):
 276         """Kills the entire process group."""
 277         signal.signal(signal.SIGTERM, signal.SIG_IGN)
 278         signal.signal(signal.SIGINT, signal.SIG_IGN)
 279         self.running = False
 280         os.killpg(self.pgid, signal.SIGTERM)
 281
 282     def start(self, application, default_port):
 283         """
 284         Run a WSGI server with the given application.
 285
 286         :param application: The application to be run in the WSGI server
 287         :param default_port: Port to bind to if none is specified in conf
 288         """
 289         self.application = application
 290         self.default_port = default_port
 291         self.configure()
 292         self.start_wsgi()
 293
 294     def start_wsgi(self):
 295
 296         if CONF.workers == 0:
 297             # Useful for profiling, test, debug etc.
 298             self.pool = self.create_pool()
 299             self.pool.spawn_n(self._single_run, self.application, self.sock)
 300             return
 301         else:
 302             LOG.info(_LI("Starting %d workers") % CONF.workers)
 303             signal.signal(signal.SIGTERM, self.kill_children)
 304             signal.signal(signal.SIGINT, self.kill_children)
 305             signal.signal(signal.SIGHUP, self.hup)
 306             while len(self.children) < CONF.workers:
 307                 self.run_child()
 308
 309     def create_pool(self):
 310         return eventlet.GreenPool(size=self.threads)
 311
 312     def _remove_children(self, pid):
 313         if pid in self.children:
 314             self.children.remove(pid)
 315             LOG.info(_LI('Removed dead child %s') % pid)
 316         elif pid in self.stale_children:
 317             self.stale_children.remove(pid)
 318             LOG.info(_LI('Removed stale child %s') % pid)
 319         else:
 320             LOG.warn(_LW('Unrecognised child %s') % pid)
 321
 322     def _verify_and_respawn_children(self, pid, status):
 323         if len(self.stale_children) == 0:
 324             LOG.debug('No stale children')
 325         if os.WIFEXITED(status) and os.WEXITSTATUS(status) != 0:
 326             LOG.error(_LE('Not respawning child %d, cannot '
 327                           'recover from termination') % pid)
 328             if not self.children and not self.stale_children:
 329                 LOG.info(
 330                     _LI('All workers have terminated. Exiting'))
 331                 self.running = False
 332         else:
 333             if len(self.children) < CONF.workers:
 334                 self.run_child()
 335
 336     def wait_on_children(self):
 337         while self.running:
 338             try:
 339                 pid, status = os.wait()
 340                 if os.WIFEXITED(status) or os.WIFSIGNALED(status):
 341                     self._remove_children(pid)
 342                     self._verify_and_respawn_children(pid, status)
 343             except OSError as err:
 344                 if err.errno not in (errno.EINTR, errno.ECHILD):
 345                     raise
 346             except KeyboardInterrupt:
 347                 LOG.info(_LI('Caught keyboard interrupt. Exiting.'))
 348                 break
 349             except exception.SIGHUPInterrupt:
 350                 self.reload()
 351                 continue
 352         eventlet.greenio.shutdown_safe(self.sock)
 353         self.sock.close()
 354         LOG.debug('Exited')
 355
 356     def configure(self, old_conf=None, has_changed=None):
 357         """
 358         Apply configuration settings
 359
 360         :param old_conf: Cached old configuration settings (if any)
 361         :param has changed: callable to determine if a parameter has changed
 362         """
 363         eventlet.wsgi.MAX_HEADER_LINE = CONF.max_header_line
 364         self.configure_socket(old_conf, has_changed)
 365         if self.initialize_escalator_store:
 366             initialize_escalator_store()
 367
 368     def reload(self):
 369         """
 370         Reload and re-apply configuration settings
 371
 372         Existing child processes are sent a SIGHUP signal
 373         and will exit after completing existing requests.
 374         New child processes, which will have the updated
 375         configuration, are spawned. This allows preventing
 376         interruption to the service.
 377         """
 378         def _has_changed(old, new, param):
 379             old = old.get(param)
 380             new = getattr(new, param)
 381             return (new != old)
 382
 383         old_conf = utils.stash_conf_values()
 384         has_changed = functools.partial(_has_changed, old_conf, CONF)
 385         CONF.reload_config_files()
 386         os.killpg(self.pgid, signal.SIGHUP)
 387         self.stale_children = self.children
 388         self.children = set()
 389
 390         # Ensure any logging config changes are picked up
 391         logging.setup(CONF, 'escalator')
 392
 393         self.configure(old_conf, has_changed)
 394         self.start_wsgi()
 395
 396     def wait(self):
 397         """Wait until all servers have completed running."""
 398         try:
 399             if self.children:
 400                 self.wait_on_children()
 401             else:
 402                 self.pool.waitall()
 403         except KeyboardInterrupt:
 404             pass
 405
 406     def run_child(self):
 407         def child_hup(*args):
 408             """Shuts down child processes, existing requests are handled."""
 409             signal.signal(signal.SIGHUP, signal.SIG_IGN)
 410             eventlet.wsgi.is_accepting = False
 411             self.sock.close()
 412
 413         pid = os.fork()
 414         if pid == 0:
 415             signal.signal(signal.SIGHUP, child_hup)
 416             signal.signal(signal.SIGTERM, signal.SIG_DFL)
 417             # ignore the interrupt signal to avoid a race whereby
 418             # a child worker receives the signal before the parent
 419             # and is respawned unnecessarily as a result
 420             signal.signal(signal.SIGINT, signal.SIG_IGN)
 421             # The child has no need to stash the unwrapped
 422             # socket, and the reference prevents a clean
 423             # exit on sighup
 424             self._sock = None
 425             self.run_server()
 426             LOG.info(_LI('Child %d exiting normally') % os.getpid())
 427             # self.pool.waitall() is now called in wsgi's server so
 428             # it's safe to exit here
 429             sys.exit(0)
 430         else:
 431             LOG.info(_LI('Started child %s') % pid)
 432             self.children.add(pid)
 433
 434     def run_server(self):
 435         """Run a WSGI server."""
 436         if cfg.CONF.pydev_worker_debug_host:
 437             utils.setup_remote_pydev_debug(cfg.CONF.pydev_worker_debug_host,
 438                                            cfg.CONF.pydev_worker_debug_port)
 439
 440         eventlet.wsgi.HttpProtocol.default_request_version = "HTTP/1.0"
 441         self.pool = self.create_pool()
 442         try:
 443             eventlet.wsgi.server(self.sock,
 444                                  self.application,
 445                                  log=self._wsgi_logger,
 446                                  custom_pool=self.pool,
 447                                  debug=False,
 448                                  keepalive=CONF.http_keepalive)
 449         except socket.error as err:
 450             if err[0] != errno.EINVAL:
 451                 raise
 452
 453         # waiting on async pools
 454         if ASYNC_EVENTLET_THREAD_POOL_LIST:
 455             for pool in ASYNC_EVENTLET_THREAD_POOL_LIST:
 456                 pool.waitall()
 457
 458     def _single_run(self, application, sock):
 459         """Start a WSGI server in a new green thread."""
 460         LOG.info(_LI("Starting single process server"))
 461         eventlet.wsgi.server(sock, application, custom_pool=self.pool,
 462                              log=self._wsgi_logger,
 463                              debug=False,
 464                              keepalive=CONF.http_keepalive)
 465
 466     def configure_socket(self, old_conf=None, has_changed=None):
 467         """
 468         Ensure a socket exists and is appropriately configured.
 469
 470         This function is called on start up, and can also be
 471         called in the event of a configuration reload.
 472
 473         When called for the first time a new socket is created.
 474         If reloading and either bind_host or bind port have been
 475         changed the existing socket must be closed and a new
 476         socket opened (laws of physics).
 477
 478         In all other cases (bind_host/bind_port have not changed)
 479         the existing socket is reused.
 480
 481         :param old_conf: Cached old configuration settings (if any)
 482         :param has changed: callable to determine if a parameter has changed
 483         """
 484         # Do we need a fresh socket?
 485         new_sock = (old_conf is None or (
 486                     has_changed('bind_host') or
 487                     has_changed('bind_port')))
 488         # Will we be using https?
 489         use_ssl = not (not CONF.cert_file or not CONF.key_file)
 490         # Were we using https before?
 491         old_use_ssl = (old_conf is not None and not (
 492                        not old_conf.get('key_file') or
 493                        not old_conf.get('cert_file')))
 494         # Do we now need to perform an SSL wrap on the socket?
 495         wrap_sock = use_ssl is True and (old_use_ssl is False or new_sock)
 496         # Do we now need to perform an SSL unwrap on the socket?
 497         unwrap_sock = use_ssl is False and old_use_ssl is True
 498
 499         if new_sock:
 500             self._sock = None
 501             if old_conf is not None:
 502                 self.sock.close()
 503             _sock = get_socket(self.default_port)
 504             _sock.setsockopt(socket.SOL_SOCKET,
 505                              socket.SO_REUSEADDR, 1)
 506             # sockets can hang around forever without keepalive
 507             _sock.setsockopt(socket.SOL_SOCKET,
 508                              socket.SO_KEEPALIVE, 1)
 509             self._sock = _sock
 510
 511         if wrap_sock:
 512             self.sock = ssl_wrap_socket(self._sock)
 513
 514         if unwrap_sock:
 515             self.sock = self._sock
 516
 517         if new_sock and not use_ssl:
 518             self.sock = self._sock
 519
 520         # Pick up newly deployed certs
 521         if old_conf is not None and use_ssl is True and old_use_ssl is True:
 522             if has_changed('cert_file') or has_changed('key_file'):
 523                 utils.validate_key_cert(CONF.key_file, CONF.cert_file)
 524             if has_changed('cert_file'):
 525                 self.sock.certfile = CONF.cert_file
 526             if has_changed('key_file'):
 527                 self.sock.keyfile = CONF.key_file
 528
 529         if new_sock or (old_conf is not None and has_changed('tcp_keepidle')):
 530             # This option isn't available in the OS X version of eventlet
 531             if hasattr(socket, 'TCP_KEEPIDLE'):
 532                 self.sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPIDLE,
 533                                      CONF.tcp_keepidle)
 534
 535         if old_conf is not None and has_changed('backlog'):
 536             self.sock.listen(CONF.backlog)
 537
 538
 539 class Middleware(object):
 540     """
 541     Base WSGI middleware wrapper. These classes require an application to be
 542     initialized that will be called next.  By default the middleware will
 543     simply call its wrapped app, or you can override __call__ to customize its
 544     behavior.
 545     """
 546
 547     def __init__(self, application):
 548         self.application = application
 549
 550     @classmethod
 551     def factory(cls, global_conf, **local_conf):
 552         def filter(app):
 553             return cls(app)
 554         return filter
 555
 556     def process_request(self, req):
 557         """
 558         Called on each request.
 559
 560         If this returns None, the next application down the stack will be
 561         executed. If it returns a response then that response will be returned
 562         and execution will stop here.
 563
 564         """
 565         return None
 566
 567     def process_response(self, response):
 568         """Do whatever you'd like to the response."""
 569         return response
 570
 571     @webob.dec.wsgify
 572     def __call__(self, req):
 573         response = self.process_request(req)
 574         if response:
 575             return response
 576         response = req.get_response(self.application)
 577         response.request = req
 578         try:
 579             return self.process_response(response)
 580         except webob.exc.HTTPException as e:
 581             return e
 582
 583
 584 class Debug(Middleware):
 585     """
 586     Helper class that can be inserted into any WSGI application chain
 587     to get information about the request and response.
 588     """
 589
 590     @webob.dec.wsgify
 591     def __call__(self, req):
 592         print(("*" * 40) + " REQUEST ENVIRON")
 593         for key, value in req.environ.items():
 594             print(key, "=", value)
 595         print('')
 596         resp = req.get_response(self.application)
 597
 598         print(("*" * 40) + " RESPONSE HEADERS")
 599         for (key, value) in six.iteritems(resp.headers):
 600             print(key, "=", value)
 601         print('')
 602
 603         resp.app_iter = self.print_generator(resp.app_iter)
 604
 605         return resp
 606
 607     @staticmethod
 608     def print_generator(app_iter):
 609         """
 610         Iterator that prints the contents of a wrapper string iterator
 611         when iterated.
 612         """
 613         print(("*" * 40) + " BODY")
 614         for part in app_iter:
 615             sys.stdout.write(part)
 616             sys.stdout.flush()
 617             yield part
 618         print()
 619
 620
 621 class APIMapper(routes.Mapper):
 622     """
 623     Handle route matching when url is '' because routes.Mapper returns
 624     an error in this case.
 625     """
 626
 627     def routematch(self, url=None, environ=None):
 628         if url is "":
 629             result = self._match("", environ)
 630             return result[0], result[1]
 631         return routes.Mapper.routematch(self, url, environ)
 632
 633
 634 class RejectMethodController(object):
 635
 636     def reject(self, req, allowed_methods, *args, **kwargs):
 637         LOG.debug("The method %s is not allowed for this resource" %
 638                   req.environ['REQUEST_METHOD'])
 639         raise webob.exc.HTTPMethodNotAllowed(
 640             headers=[('Allow', allowed_methods)])
 641
 642
 643 class Router(object):
 644     """
 645     WSGI middleware that maps incoming requests to WSGI apps.
 646     """
 647
 648     def __init__(self, mapper):
 649         """
 650         Create a router for the given routes.Mapper.
 651
 652         Each route in `mapper` must specify a 'controller', which is a
 653         WSGI app to call.  You'll probably want to specify an 'action' as
 654         well and have your controller be a wsgi.Controller, who will route
 655         the request to the action method.
 656
 657         Examples:
 658           mapper = routes.Mapper()
 659           sc = ServerController()
 660
 661           # Explicit mapping of one route to a controller+action
 662           mapper.connect(None, "/svrlist", controller=sc, action="list")
 663
 664           # Actions are all implicitly defined
 665           mapper.resource("server", "servers", controller=sc)
 666
 667           # Pointing to an arbitrary WSGI app.  You can specify the
 668           # {path_info:.*} parameter so the target app can be handed just that
 669           # section of the URL.
 670           mapper.connect(None, "/v1.0/{path_info:.*}", controller=BlogApp())
 671         """
 672         mapper.redirect("", "/")
 673         self.map = mapper
 674         self._router = routes.middleware.RoutesMiddleware(self._dispatch,
 675                                                           self.map)
 676
 677     @classmethod
 678     def factory(cls, global_conf, **local_conf):
 679         return cls(APIMapper())
 680
 681     @webob.dec.wsgify
 682     def __call__(self, req):
 683         """
 684         Route the incoming request to a controller based on self.map.
 685         If no match, return either a 404(Not Found) or 501(Not Implemented).
 686         """
 687         return self._router
 688
 689     @staticmethod
 690     @webob.dec.wsgify
 691     def _dispatch(req):
 692         """
 693         Called by self._router after matching the incoming request to a route
 694         and putting the information into req.environ.  Either returns 404,
 695         501, or the routed WSGI app's response.
 696         """
 697         match = req.environ['wsgiorg.routing_args'][1]
 698         if not match:
 699             implemented_http_methods = ['GET', 'HEAD', 'POST', 'PUT',
 700                                         'DELETE', 'PATCH']
 701             if req.environ['REQUEST_METHOD'] not in implemented_http_methods:
 702                 return webob.exc.HTTPNotImplemented()
 703             else:
 704                 return webob.exc.HTTPNotFound()
 705         app = match['controller']
 706         return app
 707
 708
 709 class Request(webob.Request):
 710     """Add some OpenStack API-specific logic to the base webob.Request."""
 711
 712     def best_match_content_type(self):
 713         """Determine the requested response content-type."""
 714         supported = ('application/json',)
 715         bm = self.accept.best_match(supported)
 716         return bm or 'application/json'
 717
 718     def get_content_type(self, allowed_content_types):
 719         """Determine content type of the request body."""
 720         if "Content-Type" not in self.headers:
 721             raise exception.InvalidContentType(content_type=None)
 722
 723         content_type = self.content_type
 724
 725         if content_type not in allowed_content_types:
 726             raise exception.InvalidContentType(content_type=content_type)
 727         else:
 728             return content_type
 729
 730     def best_match_language(self):
 731         """Determines best available locale from the Accept-Language header.
 732
 733         :returns: the best language match or None if the 'Accept-Language'
 734                   header was not available in the request.
 735         """
 736         if not self.accept_language:
 737             return None
 738         langs = i18n.get_available_languages('escalator')
 739         return self.accept_language.best_match(langs)
 740
 741     def get_content_range(self):
 742         """Return the `Range` in a request."""
 743         range_str = self.headers.get('Content-Range')
 744         if range_str is not None:
 745             range_ = webob.byterange.ContentRange.parse(range_str)
 746             if range_ is None:
 747                 msg = _('Malformed Content-Range header: %s') % range_str
 748                 raise webob.exc.HTTPBadRequest(explanation=msg)
 749             return range_
 750
 751
 752 class JSONRequestDeserializer(object):
 753     valid_transfer_encoding = frozenset(['chunked', 'compress', 'deflate',
 754                                          'gzip', 'identity'])
 755
 756     def has_body(self, request):
 757         """
 758         Returns whether a Webob.Request object will possess an entity body.
 759
 760         :param request:  Webob.Request object
 761         """
 762         request_encoding = request.headers.get('transfer-encoding', '').lower()
 763         is_valid_encoding = request_encoding in self.valid_transfer_encoding
 764         if is_valid_encoding and request.is_body_readable:
 765             return True
 766         elif request.content_length > 0:
 767             return True
 768
 769         return False
 770
 771     @staticmethod
 772     def _sanitizer(obj):
 773         """Sanitizer method that will be passed to jsonutils.loads."""
 774         return obj
 775
 776     def from_json(self, datastring):
 777         try:
 778             return jsonutils.loads(datastring, object_hook=self._sanitizer)
 779         except ValueError:
 780             msg = _('Malformed JSON in request body.')
 781             raise webob.exc.HTTPBadRequest(explanation=msg)
 782
 783     def default(self, request):
 784         if self.has_body(request):
 785             return {'body': self.from_json(request.body)}
 786         else:
 787             return {}
 788
 789
 790 class JSONResponseSerializer(object):
 791
 792     def _sanitizer(self, obj):
 793         """Sanitizer method that will be passed to jsonutils.dumps."""
 794         if hasattr(obj, "to_dict"):
 795             return obj.to_dict()
 796         if isinstance(obj, multidict.MultiDict):
 797             return obj.mixed()
 798         return jsonutils.to_primitive(obj)
 799
 800     def to_json(self, data):
 801         return jsonutils.dumps(data, default=self._sanitizer)
 802
 803     def default(self, response, result):
 804         response.content_type = 'application/json'
 805         response.body = self.to_json(result)
 806
 807
 808 def translate_exception(req, e):
 809     """Translates all translatable elements of the given exception."""
 810
 811     # The RequestClass attribute in the webob.dec.wsgify decorator
 812     # does not guarantee that the request object will be a particular
 813     # type; this check is therefore necessary.
 814     if not hasattr(req, "best_match_language"):
 815         return e
 816
 817     locale = req.best_match_language()
 818
 819     if isinstance(e, webob.exc.HTTPError):
 820         e.explanation = i18n.translate(e.explanation, locale)
 821         e.detail = i18n.translate(e.detail, locale)
 822         if getattr(e, 'body_template', None):
 823             e.body_template = i18n.translate(e.body_template, locale)
 824     return e
 825
 826
 827 class Resource(object):
 828     """
 829     WSGI app that handles (de)serialization and controller dispatch.
 830
 831     Reads routing information supplied by RoutesMiddleware and calls
 832     the requested action method upon its deserializer, controller,
 833     and serializer. Those three objects may implement any of the basic
 834     controller action methods (create, update, show, index, delete)
 835     along with any that may be specified in the api router. A 'default'
 836     method may also be implemented to be used in place of any
 837     non-implemented actions. Deserializer methods must accept a request
 838     argument and return a dictionary. Controller methods must accept a
 839     request argument. Additionally, they must also accept keyword
 840     arguments that represent the keys returned by the Deserializer. They
 841     may raise a webob.exc exception or return a dict, which will be
 842     serialized by requested content type.
 843     """
 844
 845     def __init__(self, controller, deserializer=None, serializer=None):
 846         """
 847         :param controller: object that implement methods created by routes lib
 848         :param deserializer: object that supports webob request deserialization
 849                              through controller-like actions
 850         :param serializer: object that supports webob response serialization
 851                            through controller-like actions
 852         """
 853         self.controller = controller
 854         self.serializer = serializer or JSONResponseSerializer()
 855         self.deserializer = deserializer or JSONRequestDeserializer()
 856
 857     @webob.dec.wsgify(RequestClass=Request)
 858     def __call__(self, request):
 859         """WSGI method that controls (de)serialization and method dispatch."""
 860         action_args = self.get_action_args(request.environ)
 861         action = action_args.pop('action', None)
 862
 863         try:
 864             deserialized_request = self.dispatch(self.deserializer,
 865                                                  action, request)
 866             action_args.update(deserialized_request)
 867             action_result = self.dispatch(self.controller, action,
 868                                           request, **action_args)
 869         except webob.exc.WSGIHTTPException as e:
 870             exc_info = sys.exc_info()
 871             raise translate_exception(request, e), None, exc_info[2]
 872
 873         try:
 874             response = webob.Response(request=request)
 875             self.dispatch(self.serializer, action, response, action_result)
 876             return response
 877         except webob.exc.WSGIHTTPException as e:
 878             return translate_exception(request, e)
 879         except webob.exc.HTTPException as e:
 880             return e
 881         # return unserializable result (typically a webob exc)
 882         except Exception:
 883             return action_result
 884
 885     def dispatch(self, obj, action, *args, **kwargs):
 886         """Find action-specific method on self and call it."""
 887         try:
 888             method = getattr(obj, action)
 889         except AttributeError:
 890             method = getattr(obj, 'default')
 891
 892         return method(*args, **kwargs)
 893
 894     def get_action_args(self, request_environment):
 895         """Parse dictionary created by routes library."""
 896         try:
 897             args = request_environment['wsgiorg.routing_args'][1].copy()
 898         except Exception:
 899             return {}
 900
 901         try:
 902             del args['controller']
 903         except KeyError:
 904             pass
 905
 906         try:
 907             del args['format']
 908         except KeyError:
 909             pass
 910
 911         return args