CVE-2012-4456: Improper Authentication (CWE-287)
value: The (1) OS-KSADM/services and (2) tenant APIs in OpenStack Keystone Essex before 2012.1.2 and Folsom before folsom-2 do not properly validate X-Auth-Token, which allow remote attackers to read the roles for an arbitrary user or get, create, or delete arbitrary services.
CWE description: When an actor claims to have a given identity, the product does not prove or insufficiently proves that the claim is correct.
Published: 2012-10-09T15:55Z
Repository: https://github.com/openstack/keystone
Language: Python
Files changed: keystone/catalog/core.py, tests/test_content_types.py
Commit date: 2012-06-03 12:08:44 -0500
Commit message: Require authz for service CRUD (bug 1006822)

Change-Id: Ia90f0aa2b856b9a9874d4865fb92ee913e8125c5

Diff: diff --git a/keystone/catalog/core.py b/keystone/catalog/core.py
index 68bcae24..afff7cf8 100644
--- a/keystone/catalog/core.py
+++ b/keystone/catalog/core.py
@@ -114,33 +114,40 @@ class Driver(object):
 
 
 class ServiceController(wsgi.Application):
     def __init__(self):
         self.catalog_api = Manager()
+        self.identity_api = identity.Manager()
+        self.policy_api = policy.Manager()
+        self.token_api = token.Manager()
         super(ServiceController, self).__init__()
 
     # CRUD extensions
     # NOTE(termie): this OS-KSADM stuff is not very consistent
     def get_services(self, context):
+        self.assert_admin(context)
         service_list = self.catalog_api.list_services(context)
         service_refs = [self.catalog_api.get_service(context, x)
                         for x in service_list]
         return {'OS-KSADM:services': service_refs}
 
     def get_service(self, context, service_id):
+        self.assert_admin(context)
         service_ref = self.catalog_api.get_service(context, service_id)
         if not service_ref:
             raise exception.ServiceNotFound(service_id=service_id)
         return {'OS-KSADM:service': service_ref}
 
     def delete_service(self, context, service_id):
+        self.assert_admin(context)
         service_ref = self.catalog_api.get_service(context, service_id)
         if not service_ref:
             raise exception.ServiceNotFound(service_id=service_id)
         self.catalog_api.delete_service(context, service_id)
 
     def create_service(self, context, OS_KSADM_service):
+        self.assert_admin(context)
         service_id = uuid.uuid4().hex
         service_ref = OS_KSADM_service.copy()
         service_ref['id'] = service_id
         new_service_ref = self.catalog_api.create_service(
                 context, service_id, service_ref)
diff --git a/tests/test_content_types.py b/tests/test_content_types.py
index 639a03dc..df73dff6 100644
--- a/tests/test_content_types.py
+++ b/tests/test_content_types.py
@@ -14,10 +14,11 @@
 # License for the specific language governing permissions and limitations
 # under the License.
 
 import httplib
 import json
+import uuid
 
 from lxml import etree
 import nose.exc
 
 from keystone import test
@@ -552,10 +553,42 @@ class JsonTestCase(RestfulTestCase, CoreApiTests):
             self.assertValidVersion(version)
 
     def assertValidVersionResponse(self, r):
         self.assertValidVersion(r.body.get('version'))
 
+    def test_service_crud_requires_auth(self):
+        """Service CRUD should 401 without an X-Auth-Token (bug 1006822)."""
+        # values here don't matter because we should 401 before they're checked
+        service_path = '/v2.0/OS-KSADM/services/%s' % uuid.uuid4().hex
+        service_body = {
+                'OS-KSADM:service': {
+                    'name': uuid.uuid4().hex,
+                    'type': uuid.uuid4().hex,
+                    },
+                }
+
+        r = self.admin_request(method='GET',
+                               path='/v2.0/OS-KSADM/services',
+                               expected_status=401)
+        self.assertValidErrorResponse(r)
+
+        r = self.admin_request(method='POST',
+                               path='/v2.0/OS-KSADM/services',
+                               body=service_body,
+                               expected_status=401)
+        self.assertValidErrorResponse(r)
+
+        r = self.admin_request(method='GET',
+                               path=service_path,
+                               expected_status=401)
+        self.assertValidErrorResponse(r)
+
+        r = self.admin_request(method='DELETE',
+                               path=service_path,
+                               expected_status=401)
+        self.assertValidErrorResponse(r)
+
 
 class XmlTestCase(RestfulTestCase, CoreApiTests):
     xmlns = 'http://docs.openstack.org/identity/api/v2.0'
     content_type = 'xml'
 

Fixed code:         self.identity_api = identity.Manager()
        self.policy_api = policy.Manager()
        self.token_api = token.Manager()
        self.assert_admin(context)
        self.assert_admin(context)
        self.assert_admin(context)
        self.assert_admin(context)
import uuid
    def test_service_crud_requires_auth(self):
        """Service CRUD should 401 without an X-Auth-Token (bug 1006822)."""
        # values here don't matter because we should 401 before they're checked
        service_path = '/v2.0/OS-KSADM/services/%s' % uuid.uuid4().hex
        service_body = {
                'OS-KSADM:service': {
                    'name': uuid.uuid4().hex,
                    'type': uuid.uuid4().hex,
                    },
                }

        r = self.admin_request(method='GET',
                               path='/v2.0/OS-KSADM/services',
                               expected_status=401)
        self.assertValidErrorResponse(r)

        r = self.admin_request(method='POST',
                               path='/v2.0/OS-KSADM/services',
                               body=service_body,
                               expected_status=401)
        self.assertValidErrorResponse(r)

        r = self.admin_request(method='GET',
                               path=service_path,
                               expected_status=401)
        self.assertValidErrorResponse(r)

        r = self.admin_request(method='DELETE',
                               path=service_path,
                               expected_status=401)
        self.assertValidErrorResponse(r)