Add policy checking for basic image operations

Partially implements bp interim-glance-authz-service.
This checks basic image operations: get_images, get_image,
add_image, modify_image, delete_image. It looks for a
policy json file next to our config files.

Change-Id: I07f29c11934c68d38a6bdadd39f9dc841b61648a

.mailmap

@@ -1,6 +1,7 @@
 # Format is:
 # <preferred e-mail> <other e-mail 1>
 # <preferred e-mail> <other e-mail 2>
+<brian.waldon@rackspace.com> <bcwaldon@gmail.com>
 <corywright@gmail.com> <cory.wright@rackspace.com>
 <jsuh@isi.edu> <jsuh@bespin>
 <josh@jk0.org> <josh.kearney@rackspace.com>

etc/policy.json

@@ -0,0 +1,3 @@
+{
+    "default": []
+}

glance/api/policy.py

@@ -0,0 +1,102 @@
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+
+# Copyright (c) 2011 OpenStack, LLC.
+# All Rights Reserved.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+"""Policy Engine For Glance"""
+
+import json
+import os.path
+
+from glance.common import cfg
+from glance.common import exception
+from glance.common import policy
+
+
+class Enforcer(object):
+    """Responsible for loading and enforcing rules"""
+
+    policy_opts = (
+        cfg.StrOpt('policy_file', default=None),
+        cfg.StrOpt('policy_default_rule', default='default'),
+    )
+
+    def __init__(self, conf):
+        for opt in self.policy_opts:
+            conf.register_opt(opt)
+
+        self.default_rule = conf.policy_default_rule
+        self.policy_path = self._find_policy_file(conf)
+        self.policy_file_mtime = None
+        self.policy_file_contents = None
+
+    def set_rules(self, rules):
+        """Create a new Brain based on the provided dict of rules"""
+        brain = policy.Brain(rules, self.default_rule)
+        policy.set_brain(brain)
+
+    def load_rules(self):
+        """Set the rules found in the json file on disk"""
+        rules = self._read_policy_file()
+        self.set_rules(rules)
+
+    @staticmethod
+    def _find_policy_file(conf):
+        """Locate the policy json data file"""
+        if conf.policy_file:
+            return conf.policy_file
+
+        matches = cfg.find_config_files('glance', 'policy', 'json')
+
+        try:
+            return matches[0]
+        except IndexError:
+            raise cfg.ConfigFilesNotFoundError(('policy.json',))
+
+    def _read_policy_file(self):
+        """Read contents of the policy file
+
+        This re-caches policy data if the file has been changed.
+        """
+        mtime = os.path.getmtime(self.policy_path)
+        if not self.policy_file_contents or mtime != self.policy_file_mtime:
+            with open(self.policy_path) as fap:
+                raw_contents = fap.read()
+                self.policy_file_contents = json.loads(raw_contents)
+            self.policy_file_mtime = mtime
+        return self.policy_file_contents
+
+    def enforce(self, context, action, target):
+        """Verifies that the action is valid on the target in this context.
+
+           :param context: Glance request context
+           :param action: String representing the action to be checked
+           :param object: Dictionary representing the object of the action.
+           :raises: `glance.common.exception.NotAuthorized`
+           :returns: None
+        """
+        self.load_rules()
+
+        match_list = ('rule:%s' % action,)
+        credentials = {
+            'roles': context.roles,
+            'user': context.user,
+            'tenant': context.tenant,
+        }
+
+        try:
+            policy.enforce(match_list, target, credentials)
+        except policy.NotAuthorized:
+            raise exception.NotAuthorized(action=action)

glance/api/v1/images.py

@@ -20,7 +20,6 @@
 """
 
 import errno
-import json
 import logging
 import traceback
 
@@ -28,12 +27,11 @@ from webob.exc import (HTTPNotFound,
                        HTTPConflict,
                        HTTPBadRequest,
                        HTTPForbidden,
-                       HTTPNoContent,
                        HTTPUnauthorized)
 
+from glance.api import policy
 import glance.api.v1
 from glance.api.v1 import controller
-from glance import image_cache
 from glance.common import cfg
 from glance.common import exception
 from glance.common import wsgi
@@ -48,8 +46,7 @@ from glance.store import (get_from_backend,
                           get_size_from_backend,
                           schedule_delete_from_backend,
                           get_store_from_location,
-                          get_store_from_scheme,
-                          UnsupportedBackend)
+                          get_store_from_scheme)
 from glance import registry
 from glance import notifier
 
@@ -86,6 +83,14 @@ class Controller(controller.BaseController):
         glance.store.create_stores(conf)
         self.notifier = notifier.Notifier(conf)
         registry.configure_registry_client(conf)
+        self.policy = policy.Enforcer(conf)
+
+    def _enforce(self, req, action):
+        """Authorize an action against our policies"""
+        try:
+            self.policy.enforce(req.context, action, {})
+        except exception.NotAuthorized:
+            raise HTTPUnauthorized()
 
     def index(self, req):
         """
@@ -110,6 +115,7 @@ class Controller(controller.BaseController):
                  'size': <SIZE>}, ...
             ]}
         """
+        self._enforce(req, 'get_images')
         params = self._get_query_params(req)
         try:
             images = registry.get_images_list(req.context, **params)
@@ -142,6 +148,7 @@ class Controller(controller.BaseController):
                  'properties': {'distro': 'Ubuntu 10.04 LTS', ...}}, ...
             ]}
         """
+        self._enforce(req, 'get_images')
         params = self._get_query_params(req)
         try:
             images = registry.get_images_detail(req.context, **params)
@@ -194,6 +201,7 @@ class Controller(controller.BaseController):
 
         :raises HTTPNotFound if image metadata is not available to user
         """
+        self._enforce(req, 'get_image')
         image_meta = self.get_image_meta_or_404(req, id)
         del image_meta['location']
         return {
@@ -210,6 +218,7 @@ class Controller(controller.BaseController):
 
         :raises HTTPNotFound if image is not available to user
         """
+        self._enforce(req, 'get_image')
         image_meta = self.get_active_image_meta_or_404(req, id)
 
         def get_from_store(image_meta):
@@ -470,6 +479,7 @@ class Controller(controller.BaseController):
                 and the request body is not application/octet-stream
                 image data.
         """
+        self._enforce(req, 'add_image')
         if req.context.read_only:
             msg = _("Read-only access")
             logger.debug(msg)
@@ -501,6 +511,7 @@ class Controller(controller.BaseController):
 
         :retval Returns the updated image information as a mapping
         """
+        self._enforce(req, 'modify_image')
         if req.context.read_only:
             msg = _("Read-only access")
             logger.debug(msg)
@@ -565,6 +576,7 @@ class Controller(controller.BaseController):
         :raises HttpNotAuthorized if image or any chunk is not
                 deleteable by the requesting user
         """
+        self._enforce(req, 'delete_image')
         if req.context.read_only:
             msg = _("Read-only access")
             logger.debug(msg)

glance/common/cfg.py

@@ -298,7 +298,7 @@ class ConfigFileValueError(Error):
     pass
 
 
-def find_config_files(project=None, prog=None):
+def find_config_files(project=None, prog=None, filetype="conf"):
     """Return a list of default configuration files.
 
     We default to two config files: [${project}.conf, ${prog}.conf]
@@ -331,7 +331,8 @@ def find_config_files(project=None, prog=None):
         fix_path(os.path.join('~', '.' + project)) if project else None,
         fix_path('~'),
         os.path.join('/etc', project) if project else None,
-        '/etc'
+        '/etc',
+        'etc',
         ]
     cfg_dirs = filter(bool, cfg_dirs)
 
@@ -342,9 +343,12 @@ def find_config_files(project=None, prog=None):
                 return path
 
     config_files = []
+
     if project:
-        config_files.append(search_dirs(cfg_dirs, '%s.conf' % project))
-    config_files.append(search_dirs(cfg_dirs, '%s.conf' % prog))
+        project_config = search_dirs(cfg_dirs, '%s.%s' % (project, filetype))
+        config_files.append(project_config)
+
+    config_files.append(search_dirs(cfg_dirs, '%s.%s' % (prog, filetype)))
 
     return filter(bool, config_files)
 

glance/common/policy.py

@@ -0,0 +1,184 @@
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+
+# Copyright (c) 2011 OpenStack, LLC.
+# All Rights Reserved.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+"""Common Policy Engine Implementation"""
+
+import json
+import urllib
+import urllib2
+
+
+class NotAuthorized(Exception):
+    pass
+
+
+_BRAIN = None
+
+
+def set_brain(brain):
+    """Set the brain used by enforce().
+
+    Defaults use Brain() if not set.
+
+    """
+    global _BRAIN
+    _BRAIN = brain
+
+
+def reset():
+    """Clear the brain used by enforce()."""
+    global _BRAIN
+    _BRAIN = None
+
+
+def enforce(match_list, target_dict, credentials_dict):
+    """Enforces authorization of some rules against credentials.
+
+    :param match_list: nested tuples of data to match against
+    The basic brain supports three types of match lists:
+        1) rules
+            looks like: ('rule:compute:get_instance',)
+            Retrieves the named rule from the rules dict and recursively
+            checks against the contents of the rule.
+        2) roles
+            looks like: ('role:compute:admin',)
+            Matches if the specified role is in credentials_dict['roles'].
+        3) generic
+            ('tenant_id:%(tenant_id)s',)
+            Substitutes values from the target dict into the match using
+            the % operator and matches them against the creds dict.
+
+    Combining rules:
+        The brain returns True if any of the outer tuple of rules match
+        and also True if all of the inner tuples match. You can use this to
+        perform simple boolean logic.  For example, the following rule would
+        return True if the creds contain the role 'admin' OR the if the
+        tenant_id matches the target dict AND the the creds contains the
+        role 'compute_sysadmin':
+
+        {
+            "rule:combined": (
+                'role:admin',
+                ('tenant_id:%(tenant_id)s', 'role:compute_sysadmin')
+            )
+        }
+
+
+    Note that rule and role are reserved words in the credentials match, so
+    you can't match against properties with those names. Custom brains may
+    also add new reserved words. For example, the HttpBrain adds http as a
+    reserved word.
+
+    :param target_dict: dict of object properties
+    Target dicts contain as much information as we can about the object being
+    operated on.
+
+    :param credentials_dict: dict of actor properties
+    Credentials dicts contain as much information as we can about the user
+    performing the action.
+
+    :raises NotAuthorized if the check fails
+
+    """
+    global _BRAIN
+    if not _BRAIN:
+        _BRAIN = Brain()
+    if not _BRAIN.check(match_list, target_dict, credentials_dict):
+        raise NotAuthorized()
+
+
+class Brain(object):
+    """Implements policy checking."""
+    @classmethod
+    def load_json(cls, data, default_rule=None):
+        """Init a brain using json instead of a rules dictionary."""
+        rules_dict = json.loads(data)
+        return cls(rules=rules_dict, default_rule=default_rule)
+
+    def __init__(self, rules=None, default_rule=None):
+        self.rules = rules or {}
+        self.default_rule = default_rule
+
+    def add_rule(self, key, match):
+        self.rules[key] = match
+
+    def _check(self, match, target_dict, cred_dict):
+        match_kind, match_value = match.split(':', 1)
+        try:
+            f = getattr(self, '_check_%s' % match_kind)
+        except AttributeError:
+            if not self._check_generic(match, target_dict, cred_dict):
+                return False
+        else:
+            if not f(match_value, target_dict, cred_dict):
+                return False
+        return True
+
+    def check(self, match_list, target_dict, cred_dict):
+        """Checks authorization of some rules against credentials.
+
+        Detailed description of the check with examples in policy.enforce().
+
+        :param match_list: nested tuples of data to match against
+        :param target_dict: dict of object properties
+        :param credentials_dict: dict of actor properties
+
+        :returns: True if the check passes
+
+        """
+        if not match_list:
+            return True
+        for and_list in match_list:
+            if isinstance(and_list, basestring):
+                and_list = (and_list,)
+            if all([self._check(item, target_dict, cred_dict)
+                    for item in and_list]):
+                return True
+        return False
+
+    def _check_rule(self, match, target_dict, cred_dict):
+        """Recursively checks credentials based on the brains rules."""
+        try:
+            new_match_list = self.rules[match]
+        except KeyError:
+            if self.default_rule and match != self.default_rule:
+                new_match_list = ('rule:%s' % self.default_rule,)
+            else:
+                return False
+
+        return self.check(new_match_list, target_dict, cred_dict)
+
+    def _check_role(self, match, target_dict, cred_dict):
+        """Check that there is a matching role in the cred dict."""
+        return match in cred_dict['roles']
+
+    def _check_generic(self, match, target_dict, cred_dict):
+        """Check an individual match.
+
+        Matches look like:
+
+            tenant:%(tenant_id)s
+            role:compute:admin
+
+        """
+
+        # TODO(termie): do dict inspection via dot syntax
+        match = match % target_dict
+        key, value = match.split(':', 1)
+        if key in cred_dict:
+            return value == cred_dict[key]
+        return False

glance/tests/etc/policy.json

@@ -0,0 +1,3 @@
+{
+    "default": []
+}

glance/tests/functional/__init__.py

@@ -26,7 +26,6 @@ and spinning down the servers.
 import datetime
 import functools
 import os
-import random
 import shutil
 import signal
 import socket
@@ -160,7 +159,8 @@ class ApiServer(Server):
     Server object that starts/stops/manages the API server
     """
 
-    def __init__(self, test_dir, port, registry_port, delayed_delete=False):
+    def __init__(self, test_dir, port, registry_port, policy_file,
+            delayed_delete=False):
         super(ApiServer, self).__init__(test_dir, port)
         self.server_name = 'api'
         self.default_store = 'file'
@@ -195,6 +195,8 @@ class ApiServer(Server):
         self.image_cache_dir = os.path.join(self.test_dir,
                                             'cache')
         self.image_cache_driver = 'sqlite'
+        self.policy_file = policy_file
+        self.policy_default_rule = 'default'
         self.conf_base = """[DEFAULT]
 verbose = %(verbose)s
 debug = %(debug)s
@@ -229,6 +231,8 @@ scrub_time = 5
 scrubber_datadir = %(scrubber_datadir)s
 image_cache_dir = %(image_cache_dir)s
 image_cache_driver = %(image_cache_driver)s
+policy_file = %(policy_file)s
+policy_default_rule = %(policy_default_rule)s
 [paste_deploy]
 flavor = %(deployment_flavor)s
 """
@@ -359,9 +363,13 @@ class FunctionalTest(unittest.TestCase):
         self.api_port = get_unused_port()
         self.registry_port = get_unused_port()
 
+        self.copy_data_file('policy.json', self.test_dir)
+        self.policy_file = os.path.join(self.test_dir, 'policy.json')
+
         self.api_server = ApiServer(self.test_dir,
                                     self.api_port,
-                                    self.registry_port)
+                                    self.registry_port,
+                                    self.policy_file)
         self.registry_server = RegistryServer(self.test_dir,
                                               self.registry_port)
 
@@ -546,3 +554,9 @@ class FunctionalTest(unittest.TestCase):
         engine = create_engine(self.registry_server.sql_connection,
                                pool_recycle=30)
         return engine.execute(sql)
+
+    def copy_data_file(self, file_name, dst_dir):
+        src_file_name = os.path.join('glance/tests/etc', file_name)
+        shutil.copy(src_file_name, dst_dir)
+        dst_file_name = os.path.join(dst_dir, file_name)
+        return dst_file_name

glance/tests/stubs.py

@@ -35,7 +35,7 @@ VERBOSE = False
 DEBUG = False
 
 
-def stub_out_registry_and_store_server(stubs, images_dir):
+def stub_out_registry_and_store_server(stubs, base_dir):
     """
     Mocks calls to 127.0.0.1 on 9191 and 9292 for testing so
     that a real Glance server does not need to be up and
@@ -125,7 +125,8 @@ def stub_out_registry_and_store_server(stubs, images_dir):
                     'registry_host': '0.0.0.0',
                     'registry_port': '9191',
                     'default_store': 'file',
-                    'filesystem_store_datadir': images_dir
+                    'filesystem_store_datadir': base_dir,
+                    'policy_file': os.path.join(base_dir, 'policy.json'),
                     })
             api = version_negotiation.VersionNegotiationFilter(
                 context.ContextMiddleware(router.API(conf), conf),

glance/tests/unit/base.py

@@ -15,13 +15,13 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
+import json
 import os
 import shutil
 import unittest
 
 import stubout
 
-from glance.common import utils
 from glance.tests import stubs
 from glance.tests import utils as test_utils
 
@@ -37,14 +37,27 @@ class IsolatedUnitTest(unittest.TestCase):
         self.test_id, self.test_dir = test_utils.get_isolated_test_env()
         self.stubs = stubout.StubOutForTesting()
         stubs.stub_out_registry_and_store_server(self.stubs, self.test_dir)
+        policy_file = self._copy_data_file('policy.json', self.test_dir)
         options = {'sql_connection': 'sqlite://',
                    'verbose': False,
                    'debug': False,
                    'default_store': 'filesystem',
-                   'filesystem_store_datadir': os.path.join(self.test_dir)}
+                   'filesystem_store_datadir': os.path.join(self.test_dir),
+                   'policy_file': policy_file}
 
         self.conf = test_utils.TestConfigOpts(options)
 
+    def _copy_data_file(self, file_name, dst_dir):
+        src_file_name = os.path.join('glance/tests/etc', file_name)
+        shutil.copy(src_file_name, dst_dir)
+        dst_file_name = os.path.join(dst_dir, file_name)
+        return dst_file_name
+
+    def set_policy_rules(self, rules):
+        fap = open(self.conf.policy_file, 'w')
+        fap.write(json.dumps(rules))
+        fap.close()
+
     def tearDown(self):
         self.stubs.UnsetAll()
         if os.path.exists(self.test_dir):

glance/tests/unit/test_api.py

@@ -2086,6 +2086,24 @@ class TestGlanceAPI(base.IsolatedUnitTest):
         self.assertTrue('/images/%s' % res_body['id']
                         in res.headers['location'])
 
+    def test_add_image_unauthorized(self):
+        rules = {"add_image": [["false:false"]]}
+        self.set_policy_rules(rules)
+        fixture_headers = {'x-image-meta-store': 'file',
+                           'x-image-meta-disk-format': 'vhd',
+                           'x-image-meta-container-format': 'ovf',
+                           'x-image-meta-name': 'fake image #3'}
+
+        req = webob.Request.blank("/images")
+        req.method = 'POST'
+        for k, v in fixture_headers.iteritems():
+            req.headers[k] = v
+
+        req.headers['Content-Type'] = 'application/octet-stream'
+        req.body = "chunk00000remainder"
+        res = req.get_response(self.api)
+        self.assertEquals(res.status_int, 401)
+
     def test_register_and_upload(self):
         """
         Test that the process of registering an image with
@@ -2360,6 +2378,20 @@ class TestGlanceAPI(base.IsolatedUnitTest):
         res = req.get_response(self.api)
         self.assertEquals(res.status_int, 400)
 
+    def test_get_images_detailed_unauthorized(self):
+        rules = {"get_images": [["false:false"]]}
+        self.set_policy_rules(rules)
+        req = webob.Request.blank('/images/detail')
+        res = req.get_response(self.api)
+        self.assertEquals(res.status_int, 401)
+
+    def test_get_images_unauthorized(self):
+        rules = {"get_images": [["false:false"]]}
+        self.set_policy_rules(rules)
+        req = webob.Request.blank('/images/detail')
+        res = req.get_response(self.api)
+        self.assertEquals(res.status_int, 401)
+
     def test_store_location_not_revealed(self):
         """
         Test that the internal store location is NOT revealed
@@ -2514,6 +2546,14 @@ class TestGlanceAPI(base.IsolatedUnitTest):
         for key, value in expected_headers.iteritems():
             self.assertEquals(value, res.headers[key])
 
+    def test_image_meta_unauthorized(self):
+        rules = {"get_image": [["false:false"]]}
+        self.set_policy_rules(rules)
+        req = webob.Request.blank("/images/%s" % UUID2)
+        req.method = 'HEAD'
+        res = req.get_response(self.api)
+        self.assertEquals(res.status_int, 401)
+
     def test_show_image_basic(self):
         req = webob.Request.blank("/images/%s" % UUID2)
         res = req.get_response(self.api)
@@ -2526,6 +2566,13 @@ class TestGlanceAPI(base.IsolatedUnitTest):
         res = req.get_response(self.api)
         self.assertEquals(res.status_int, webob.exc.HTTPNotFound.code)
 
+    def test_show_image_unauthorized(self):
+        rules = {"get_image": [["false:false"]]}
+        self.set_policy_rules(rules)
+        req = webob.Request.blank("/images/%s" % UUID2)
+        res = req.get_response(self.api)
+        self.assertEqual(res.status_int, 401)
+
     def test_delete_image(self):
         req = webob.Request.blank("/images/%s" % UUID2)
         req.method = 'DELETE'
@@ -2596,6 +2643,14 @@ class TestGlanceAPI(base.IsolatedUnitTest):
         res = req.get_response(self.api)
         self.assertEquals(res.status_int, httplib.FORBIDDEN)
 
+    def test_delete_image_unauthorized(self):
+        rules = {"delete_image": [["false:false"]]}
+        self.set_policy_rules(rules)
+        req = webob.Request.blank("/images/%s" % UUID2)
+        req.method = 'DELETE'
+        res = req.get_response(self.api)
+        self.assertEquals(res.status_int, 401)
+
     def test_get_details_invalid_marker(self):
         """
         Tests that the /images/detail registry API returns a 400

glance/tests/unit/test_config.py

@@ -21,10 +21,8 @@ import unittest
 
 import stubout
 
-from glance.api.middleware import version_negotiation
-from glance.api.v1 import images
-from glance.api.v1 import members
-from glance.common import config, context, utils
+from glance.common import config
+from glance.common import context
 from glance.image_cache import pruner
 from glance.tests import utils as test_utils
 
@@ -52,34 +50,36 @@ class TestPasteApp(unittest.TestCase):
                 f.flush()
 
         if paste_copy:
-            paste_from = os.path.join(os.getcwd(), 'etc/glance-api-paste.ini')
+            paste_from = os.path.join(os.getcwd(),
+                                      'etc/glance-registry-paste.ini')
             paste_to = os.path.join(conf.temp_file.replace('.conf',
                                                        '-paste.ini'))
             _appendto(paste_from, paste_to, paste_append)
 
-        app = config.load_paste_app(conf, 'glance-api')
+        app = config.load_paste_app(conf, 'glance-registry')
 
         self.assertEquals(expected_app_type, type(app))
 
     def test_load_paste_app(self):
-        type = version_negotiation.VersionNegotiationFilter
-        self._do_test_load_paste_app(type)
+        expected_middleware = context.ContextMiddleware
+        self._do_test_load_paste_app(expected_middleware)
 
     def test_load_paste_app_with_paste_flavor(self):
         paste_group = {'paste_deploy': {'flavor': 'incomplete'}}
-        pipeline = '[pipeline:glance-api-incomplete]\n' + \
-                   'pipeline = context apiv1app'
+        pipeline = '[pipeline:glance-registry-incomplete]\n' + \
+                   'pipeline = context registryapp'
 
         type = context.ContextMiddleware
         self._do_test_load_paste_app(type, paste_group, paste_append=pipeline)
 
     def test_load_paste_app_with_paste_config_file(self):
         paste_config_file = os.path.join(os.getcwd(),
-                                         'etc/glance-api-paste.ini')
+                                         'etc/glance-registry-paste.ini')
         paste_group = {'paste_deploy': {'config_file': paste_config_file}}
 
-        type = version_negotiation.VersionNegotiationFilter
-        self._do_test_load_paste_app(type, paste_group, paste_copy=False)
+        expected_middleware = context.ContextMiddleware
+        self._do_test_load_paste_app(expected_middleware,
+                                     paste_group, paste_copy=False)
 
     def test_load_paste_app_with_conf_name(self):
         def fake_join(*args):