openstack
/
keystone
Code Issues Proposed changes

Reject create endpoint with invalid urls 

The patch is not for general URL validation. It just implements
a check to make sure any substitutions in the URL will be valid
when building the catalog.

The bug also happen in v3. So we have to fix this bug in both
v2 and v3.

Closes-Bug: #1471034
Change-Id: I6728edaae26b140fc19afec6331674c57856985c
This commit is contained in:
jiaxi 2015-07-23 04:33:13 -04:00
parent a9c794ae5e
commit 312d6c9fd0
5 changed files with 168 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
keystone/catalog/controllers.py
View File

@ -17,6 +17,7 @@ import uuid
import six
from keystone.catalog import core
from keystone.catalog import schema
from keystone.common import controller
from keystone.common import dependency
@ -100,6 +101,14 @@ class Endpoint(controller.V2Controller):
# service_id is necessary
self._require_attribute(endpoint, 'service_id')
# we should check publicurl, adminurl, internalurl
# if invalid, we should raise an exception to reject
# the request
for interface in INTERFACES:
interface_url = endpoint.get(interface + 'url')
if interface_url:
core.check_endpoint_url(interface_url)
initiator = notifications._get_request_audit_info(context)
if endpoint.get('region') is not None:
@ -301,6 +310,7 @@ class EndpointV3(controller.V3Controller):
@controller.protected()
@validation.validated(schema.endpoint_create, 'endpoint')
def create_endpoint(self, context, endpoint):
core.check_endpoint_url(endpoint['url'])
ref = self._assign_unique_id(self._normalize_dict(endpoint))
ref = self._validate_endpoint_region(ref, context)
initiator = notifications._get_request_audit_info(context)

32
keystone/catalog/core.py
View File

@ -16,6 +16,7 @@
"""Main entry point into the Catalog service."""
import abc
import itertools
from oslo_config import cfg
from oslo_log import log
@ -35,6 +36,10 @@ from keystone import notifications
CONF = cfg.CONF
LOG = log.getLogger(__name__)
MEMOIZE = cache.get_memoization_decorator(section='catalog')
WHITELISTED_PROPERTIES = [
'tenant_id', 'user_id', 'public_bind_host', 'admin_bind_host',
'compute_host', 'admin_port', 'public_port',
'public_endpoint', 'admin_endpoint', ]
def format_url(url, substitutions, silent_keyerror_failures=None):
@ -48,11 +53,6 @@ def format_url(url, substitutions, silent_keyerror_failures=None):
"""
WHITELISTED_PROPERTIES = [
'tenant_id', 'user_id', 'public_bind_host', 'admin_bind_host',
'compute_host', 'admin_port', 'public_port',
'public_endpoint', 'admin_endpoint', ]
substitutions = utils.WhiteListedItemFilter(
WHITELISTED_PROPERTIES,
substitutions)
@ -85,6 +85,28 @@ def format_url(url, substitutions, silent_keyerror_failures=None):
return result
def check_endpoint_url(url):
"""Check substitution of url.
The invalid urls are as follows:
urls with substitutions that is not in the whitelist
Check the substitutions in the URL to make sure they are valid
and on the whitelist.
:param str url: the URL to validate
:rtype: None
:raises keystone.exception.URLValidationError: if the URL is invalid
"""
# check whether the property in the path is exactly the same
# with that in the whitelist below
substitutions = dict(zip(WHITELISTED_PROPERTIES, itertools.repeat('')))
try:
url.replace('$(', '%(') % substitutions
except (KeyError, TypeError, ValueError):
raise exception.URLValidationError(url)
@dependency.provider('catalog_api')
class Manager(manager.Manager):
"""Default pivot point for the Catalog backend.

5
keystone/exception.py
View File

@ -84,6 +84,11 @@ class ValidationError(Error):
title = 'Bad Request'
class URLValidationError(ValidationError):
message_format = _("Cannot create an endpoint with an invalid URL:"
" %(url)s")
class SchemaValidationError(ValidationError):
# NOTE(lbragstad): For whole OpenStack message consistency, this error
# message has been written in a format consistent with WSME.

87
keystone/tests/unit/test_catalog.py
View File

@ -128,6 +128,93 @@ class V2CatalogTestCase(rest.RestfulTestCase):
def test_endpoint_create_with_empty_service_id(self):
self._endpoint_create(expected_status=400, service_id='')
def test_endpoint_create_with_valid_url(self):
"""Create endpoint with valid url should be tested,too."""
# list one valid url is enough, no need to list too much
valid_url = 'http://127.0.0.1:8774/v1.1/$(tenant_id)s'
# baseline tests that all valid URLs works
self._endpoint_create(expected_status=200,
publicurl=valid_url,
internalurl=valid_url,
adminurl=valid_url)
def test_endpoint_create_with_invalid_url(self):
"""Test the invalid cases: substitutions is not exactly right.
"""
invalid_urls = [
# using a substitution that is not whitelisted - KeyError
'http://127.0.0.1:8774/v1.1/$(nonexistent)s',
# invalid formatting - ValueError
'http://127.0.0.1:8774/v1.1/$(tenant_id)',
'http://127.0.0.1:8774/v1.1/$(tenant_id)t',
'http://127.0.0.1:8774/v1.1/$(tenant_id',
# invalid type specifier - TypeError
# admin_url is a string not an int
'http://127.0.0.1:8774/v1.1/$(admin_url)d',
]
# list one valid url is enough, no need to list too much
valid_url = 'http://127.0.0.1:8774/v1.1/$(tenant_id)s'
# Case one: publicurl, internalurl and adminurl are
# all invalid
for invalid_url in invalid_urls:
self._endpoint_create(expected_status=400,
publicurl=invalid_url,
internalurl=invalid_url,
adminurl=invalid_url)
# Case two: publicurl, internalurl are invalid
# and adminurl is valid
for invalid_url in invalid_urls:
self._endpoint_create(expected_status=400,
publicurl=invalid_url,
internalurl=invalid_url,
adminurl=valid_url)
# Case three: publicurl, adminurl are invalid
# and internalurl is valid
for invalid_url in invalid_urls:
self._endpoint_create(expected_status=400,
publicurl=invalid_url,
internalurl=valid_url,
adminurl=invalid_url)
# Case four: internalurl, adminurl are invalid
# and publicurl is valid
for invalid_url in invalid_urls:
self._endpoint_create(expected_status=400,
publicurl=valid_url,
internalurl=invalid_url,
adminurl=invalid_url)
# Case five: publicurl is invalid, internalurl
# and adminurl are valid
for invalid_url in invalid_urls:
self._endpoint_create(expected_status=400,
publicurl=invalid_url,
internalurl=valid_url,
adminurl=valid_url)
# Case six: internalurl is invalid, publicurl
# and adminurl are valid
for invalid_url in invalid_urls:
self._endpoint_create(expected_status=400,
publicurl=valid_url,
internalurl=invalid_url,
adminurl=valid_url)
# Case seven: adminurl is invalid, publicurl
# and internalurl are valid
for invalid_url in invalid_urls:
self._endpoint_create(expected_status=400,
publicurl=valid_url,
internalurl=valid_url,
adminurl=invalid_url)
class TestV2CatalogAPISQL(tests.TestCase):

43
keystone/tests/unit/test_v3_catalog.py
View File

@ -629,14 +629,49 @@ class CatalogTestCase(test_v3.RestfulTestCase):
ref['url'] = url_with_space
# add the endpoint to the database
r = self.post('/endpoints', body={'endpoint': ref})
endpoint = r.result['endpoint']
self.catalog_api.create_endpoint(ref['id'], ref)
# delete the endpoint
self.delete('/endpoints/%s' % endpoint['id'])
self.delete('/endpoints/%s' % ref['id'])
# make sure it's deleted (GET should return 404)
self.get('/endpoints/%s' % endpoint['id'], expected_status=404)
self.get('/endpoints/%s' % ref['id'], expected_status=404)
def test_endpoint_create_with_valid_url(self):
"""Create endpoint with valid url should be tested,too."""
# list one valid url is enough, no need to list too much
valid_url = 'http://127.0.0.1:8774/v1.1/$(tenant_id)s'
ref = self.new_endpoint_ref(self.service_id)
ref['url'] = valid_url
self.post('/endpoints',
body={'endpoint': ref},
expected_status=201)
def test_endpoint_create_with_invalid_url(self):
"""Test the invalid cases: substitutions is not exactly right.
"""
invalid_urls = [
# using a substitution that is not whitelisted - KeyError
'http://127.0.0.1:8774/v1.1/$(nonexistent)s',
# invalid formatting - ValueError
'http://127.0.0.1:8774/v1.1/$(tenant_id)',
'http://127.0.0.1:8774/v1.1/$(tenant_id)t',
'http://127.0.0.1:8774/v1.1/$(tenant_id',
# invalid type specifier - TypeError
# admin_url is a string not an int
'http://127.0.0.1:8774/v1.1/$(admin_url)d',
]
ref = self.new_endpoint_ref(self.service_id)
for invalid_url in invalid_urls:
ref['url'] = invalid_url
self.post('/endpoints',
body={'endpoint': ref},
expected_status=400)
class TestCatalogAPISQL(tests.TestCase):