openstack
/
deb-python-os-cloud-config
Code Issues Proposed changes

Add []'s around ipv6 addresses when creating url's 

Anywhere a url is formatted in order to create a Keystone endpoint, []'s
must surround the host portion of the url if it's an ipv6 address.

Change-Id: If4cf3892761ac17c08779d87c790f74f3a1ce88a
This commit is contained in:
James Slagle 2016-01-18 14:59:56 -05:00
parent 643c50f103
commit ebc1352b11
2 changed files with 89 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

27
os_cloud_config/keystone.py
View File

@ -15,6 +15,7 @@
from __future__ import print_function
import logging
import socket
import subprocess
import time
@ -343,6 +344,16 @@ def setup_endpoints(endpoints, public_host=None, region=None, client=None,
_register_endpoint(client, service, conf, region)
def is_valid_ipv6_address(address):
try:
socket.inet_pton(socket.AF_INET6, address)
except socket.error: # not a valid address
return False
except TypeError: # Not a string, e.g. None
return False
return True
def _register_endpoint(keystone, service, data, region=None):
"""Create single service endpoint in Keystone.
@ -353,11 +364,15 @@ def _register_endpoint(keystone, service, data, region=None):
"""
path = data.get('path', '/')
internal_host = data.get('internal_host')
if is_valid_ipv6_address(internal_host):
internal_host = '[{host}]'.format(host=internal_host)
port = data.get('port')
internal_uri = 'http://{host}:{port}{path}'.format(
host=internal_host, port=port, path=path)
public_host = data.get('public_host')
if is_valid_ipv6_address(public_host):
public_host = '[{host}]'.format(host=public_host)
public_protocol = 'http'
public_port = port
if public_host and 'ssl_port' in data:
@ -495,6 +510,18 @@ def _create_keystone_endpoint(keystone, host, region, ssl, public, admin,
LOG.debug('Create keystone public endpoint')
service = _create_service(keystone, 'keystone', 'identity',
description='Keystone Identity Service')
if is_valid_ipv6_address(host):
host = '[{host}]'.format(host=host)
if is_valid_ipv6_address(ssl):
ssl = '[{host}]'.format(host=ssl)
if is_valid_ipv6_address(public):
public = '[{host}]'.format(host=public)
if is_valid_ipv6_address(admin):
admin = '[{host}]'.format(host=admin)
if is_valid_ipv6_address(internal):
internal = '[{host}]'.format(host=internal)
public_url = 'http://%s:5000/v2.0' % host
if ssl:
public_url = 'https://%s:13000/v2.0' % ssl

62
os_cloud_config/tests/test_keystone.py
View File

@ -297,6 +297,27 @@ class KeystoneTest(base.TestCase):
self.client, '192.0.0.3', 'regionTwo', None, None, None, None)
self.assert_endpoint('192.0.0.3', region='regionTwo')
def test_create_keystone_endpoint_ipv6(self):
self._patch_client()
self.client.services.findall.return_value = []
self.client.endpoints.findall.return_value = []
keystone._create_keystone_endpoint(
self.client, '2001:db8:fd00:1000:f816:3eff:fec2:8e7c',
'regionOne',
None,
'2001:db8:fd00:1000:f816:3eff:fec2:8e7d',
'2001:db8:fd00:1000:f816:3eff:fec2:8e7e',
'2001:db8:fd00:1000:f816:3eff:fec2:8e7f')
pe = 'http://[2001:db8:fd00:1000:f816:3eff:fec2:8e7d]:5000/v2.0'
ae = 'http://[2001:db8:fd00:1000:f816:3eff:fec2:8e7e]:35357/v2.0'
ie = 'http://[2001:db8:fd00:1000:f816:3eff:fec2:8e7f]:5000/v2.0'
self.assert_endpoint(
'[2001:db8:fd00:1000:f816:3eff:fec2:8e7c]',
region='regionOne', public_endpoint=pe, admin_endpoint=ae,
internal_endpoint=ie)
@mock.patch('time.sleep')
def test_create_roles_retry(self, sleep):
self._patch_client()
@ -347,6 +368,47 @@ class KeystoneTest(base.TestCase):
'http://192.0.0.3:8774/v2/$(tenant_id)s',
'http://192.0.0.3:8774/v2/$(tenant_id)s')
def test_setup_endpoints_ipv6(self):
self.client = mock.MagicMock()
self.client.users.find.side_effect = ksclient_v2.exceptions.NotFound()
self.client.services.findall.return_value = []
self.client.endpoints.findall.return_value = []
keystone.setup_endpoints(
{'nova': {'password': 'pass', 'type': 'compute',
'ssl_port': 1234}},
public_host='2001:db8:fd00:1000:f816:3eff:fec2:8e7c',
region='region', client=self.client,
os_auth_url='https://[2001:db8:fd00:1000:f816:3eff:fec2:8e7c]')
self.client.users.find.assert_called_once_with(name='nova')
self.client.tenants.find.assert_called_once_with(name='service')
self.client.roles.find.assert_called_once_with(name='admin')
self.client.services.findall.assert_called_once_with(type='compute')
self.client.endpoints.findall.assert_called_once_with(
publicurl='https://[2001:db8:fd00:1000:f816:3eff:fec2:8e7c]'
':1234/v2/$(tenant_id)s')
self.client.users.create.assert_called_once_with(
'nova', 'pass',
tenant_id=self.client.tenants.find.return_value.id,
email='email=nobody@example.com')
self.client.roles.add_user_role.assert_called_once_with(
self.client.users.create.return_value,
self.client.roles.find.return_value,
self.client.tenants.find.return_value)
self.client.services.create.assert_called_once_with(
'nova', 'compute', description='Nova Compute Service')
ipv6_addr = '2001:db8:fd00:1000:f816:3eff:fec2:8e7c'
self.client.endpoints.create.assert_called_once_with(
'region',
self.client.services.create.return_value.id,
'https://[%s]:1234/v2/$(tenant_id)s' % ipv6_addr,
'http://[%s]:8774/v2/$(tenant_id)s' % ipv6_addr,
'http://[%s]:8774/v2/$(tenant_id)s' % ipv6_addr)
@mock.patch('os_cloud_config.keystone._create_service')
def test_create_ssl_endpoint_no_ssl_port(self, mock_create_service):
client = mock.Mock()