Utility method reworked, etc.

nova/auth/authutils.py

@@ -0,0 +1,48 @@
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+
+# Copyright 2011 NTT DATA CORPORATION.
+# Copyright 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.
+
+"""
+Auth module specific utilities and helper functions.
+"""
+
+import netaddr
+import string
+
+
+def get_host_only_server_string(server_str):
+    """
+    Returns host part only of the given server_string if it's a combination
+    of host part and port. Otherwise, return null string.
+    """
+
+    # First of all, exclude pure IPv6 address (w/o port).
+    if netaddr.valid_ipv6(server_str):
+        return ''
+
+    # Next, check if this is IPv6 address with port number combination.
+    if server_str.find("]:") != -1:
+        [address, sep, port] = server_str.replace('[', '', 1).partition(']:')
+        return address
+
+    # Third, check if this is a combination of general address and port
+    if server_str.find(':') == -1:
+        return ''
+
+    # This must be a combination of host part and port
+    (address, port) = server_str.split(':')
+    return address

nova/auth/manager.py

@@ -35,6 +35,7 @@ from nova import flags
 from nova import log as logging
 from nova import utils
 from nova.auth import signer
+from nova.auth import authutils
 
 
 FLAGS = flags.FLAGS
@@ -315,7 +316,8 @@ class AuthManager(object):
             LOG.debug(_('expected_signature: %s'), expected_signature)
             LOG.debug(_('signature: %s'), signature)
             if signature != expected_signature:
-                host_only = utils.get_host_only_server_string(server_string)
+                host_only = authutils.get_host_only_server_string(
+                    server_string)
                 # If the given server_string contains port num, try without it.
                 if host_only != '':
                     host_only_signature = signer.Signer(

nova/tests/test_auth.py

@@ -25,6 +25,7 @@ from nova import log as logging
 from nova import test
 from nova.auth import manager
 from nova.api.ec2 import cloud
+from nova.auth import authutils
 
 FLAGS = flags.FLAGS
 LOG = logging.getLogger('nova.tests.auth_unittest')
@@ -339,6 +340,29 @@ class AuthManagerDbTestCase(_AuthManagerBaseTestCase):
     auth_driver = 'nova.auth.dbdriver.DbDriver'
 
 
+class AuthManagerUtilTestCase(test.TestCase):
+    def test_get_host_only_server_string(self):
+        result = authutils.get_host_only_server_string('::1')
+        self.assertEqual('', result)
+        result = authutils.get_host_only_server_string('[::1]:8773')
+        self.assertEqual('::1', result)
+        result = authutils.get_host_only_server_string('2001:db8::192.168.1.1')
+        self.assertEqual('', result)
+        result = authutils.get_host_only_server_string(
+            '[2001:db8::192.168.1.1]:8773')
+        self.assertEqual('2001:db8::192.168.1.1', result)
+        result = authutils.get_host_only_server_string('192.168.1.1')
+        self.assertEqual('', result)
+        result = authutils.get_host_only_server_string('192.168.1.2:8773')
+        self.assertEqual('192.168.1.2', result)
+        result = authutils.get_host_only_server_string('192.168.1.3')
+        self.assertEqual('', result)
+        result = authutils.get_host_only_server_string('www.example.com:8443')
+        self.assertEqual('www.example.com', result)
+        result = authutils.get_host_only_server_string('www.example.com')
+        self.assertEqual('', result)
+
+
 if __name__ == "__main__":
     # TODO: Implement use_fake as an option
     unittest.main()