distros: replace invalid characters in mirror URLs with hyphens (#291)

This modifies _get_package_mirror_info to convert the hostnames of generated mirror URLs to their IDNA form, and then iterate through them replacing any invalid characters (i.e. anything other than letters, digits or a hyphen) with a hyphen.

This commit introduces the following changes in behaviour:

* generated mirror URLs with Unicode characters in their hostnames will have their hostnames converted to their all-ASCII IDNA form
* generated mirror URLs with invalid-for-hostname characters in their hostname will have those characters converted to hyphens
* generated mirror URLs which cannot be parsed by `urllib.parse.urlsplit` will not be considered for use
  * other configured patterns will still be considered
  * if all configured patterns fail to produce a URL that parses then the fallback mirror URL will be used

LP: #1868232

1 files changed, 67 insertions, 17 deletions

diff --git a/cloudinit/distros/tests/test_init.py b/cloudinit/distros/tests/test_init.py
index 707a0a49..daa81ab8 100644
--- a/cloudinit/distros/tests/test_init.py
+++ b/cloudinit/distros/tests/test_init.py
@@ -9,7 +9,18 @@ from unittest import mock
 
 import pytest
 
-from cloudinit.distros import _get_package_mirror_info
+from cloudinit.distros import _get_package_mirror_info, LDH_ASCII_CHARS
+
+
+# Define a set of characters we would expect to be replaced
+INVALID_URL_CHARS = [
+    chr(x) for x in range(127) if chr(x) not in LDH_ASCII_CHARS
+]
+for separator in [":", ".", "/", "#", "?", "@", "[", "]"]:
+    # Remove from the set characters that either separate hostname parts (":",
+    # "."), terminate hostnames ("/", "#", "?", "@"), or cause Python to be
+    # unable to parse URLs ("[", "]").
+    INVALID_URL_CHARS.remove(separator)
 
 
 class TestGetPackageMirrorInfo:
@@ -25,14 +36,16 @@ class TestGetPackageMirrorInfo:
         # Empty info gives empty return
         ({}, {}),
         # failsafe values used if present
-        ({'failsafe': {'primary': 'value', 'security': 'other'}},
-         {'primary': 'value', 'security': 'other'}),
+        ({'failsafe': {'primary': 'http://value', 'security': 'http://other'}},
+         {'primary': 'http://value', 'security': 'http://other'}),
         # search values used if present
-        ({'search': {'primary': ['value'], 'security': ['other']}},
-         {'primary': ['value'], 'security': ['other']}),
+        ({'search': {'primary': ['http://value'],
+                     'security': ['http://other']}},
+         {'primary': ['http://value'], 'security': ['http://other']}),
         # failsafe values used if search value not present
-        ({'search': {'primary': ['value']}, 'failsafe': {'security': 'other'}},
-         {'primary': ['value'], 'security': 'other'})
+        ({'search': {'primary': ['http://value']},
+          'failsafe': {'security': 'http://other'}},
+         {'primary': ['http://value'], 'security': 'http://other'})
     ])
     def test_get_package_mirror_info_failsafe(self, mirror_info, expected):
         """
@@ -48,26 +61,63 @@ class TestGetPackageMirrorInfo:
     def test_failsafe_used_if_all_search_results_filtered_out(self):
         """Test the failsafe option used if all search options eliminated."""
         mirror_info = {
-            'search': {'primary': ['value']}, 'failsafe': {'primary': 'other'}
+            'search': {'primary': ['http://value']},
+            'failsafe': {'primary': 'http://other'}
         }
-        assert {'primary': 'other'} == _get_package_mirror_info(
+        assert {'primary': 'http://other'} == _get_package_mirror_info(
             mirror_info, mirror_filter=lambda x: False)
 
     @pytest.mark.parametrize('availability_zone,region,patterns,expected', (
         # Test ec2_region alone
-        ('fk-fake-1f', None, ['EC2-%(ec2_region)s'], ['EC2-fk-fake-1']),
+        ('fk-fake-1f', None, ['http://EC2-%(ec2_region)s/ubuntu'],
+         ['http://ec2-fk-fake-1/ubuntu']),
         # Test availability_zone alone
-        ('fk-fake-1f', None, ['AZ-%(availability_zone)s'], ['AZ-fk-fake-1f']),
+        ('fk-fake-1f', None, ['http://AZ-%(availability_zone)s/ubuntu'],
+         ['http://az-fk-fake-1f/ubuntu']),
         # Test region alone
-        (None, 'fk-fake-1', ['RG-%(region)s'], ['RG-fk-fake-1']),
+        (None, 'fk-fake-1', ['http://RG-%(region)s/ubuntu'],
+         ['http://rg-fk-fake-1/ubuntu']),
         # Test that ec2_region is not available for non-matching AZs
         ('fake-fake-1f', None,
-         ['EC2-%(ec2_region)s', 'AZ-%(availability_zone)s'],
-         ['AZ-fake-fake-1f']),
+         ['http://EC2-%(ec2_region)s/ubuntu',
+          'http://AZ-%(availability_zone)s/ubuntu'],
+         ['http://az-fake-fake-1f/ubuntu']),
         # Test that template order maintained
-        (None, 'fake-region', ['RG-%(region)s-2', 'RG-%(region)s-1'],
-         ['RG-fake-region-2', 'RG-fake-region-1']),
-    ))
+        (None, 'fake-region',
+         ['http://RG-%(region)s-2/ubuntu', 'http://RG-%(region)s-1/ubuntu'],
+         ['http://rg-fake-region-2/ubuntu', 'http://rg-fake-region-1/ubuntu']),
+        # Test that non-ASCII hostnames are IDNA encoded;
+        # "IDNA-ТεЅТ̣".encode('idna') == b"xn--idna--4kd53hh6aba3q"
+        (None, 'ТεЅТ̣', ['http://www.IDNA-%(region)s.com/ubuntu'],
+         ['http://www.xn--idna--4kd53hh6aba3q.com/ubuntu']),
+        # Test that non-ASCII hostnames with a port are IDNA encoded;
+        # "IDNA-ТεЅТ̣".encode('idna') == b"xn--idna--4kd53hh6aba3q"
+        (None, 'ТεЅТ̣', ['http://www.IDNA-%(region)s.com:8080/ubuntu'],
+         ['http://www.xn--idna--4kd53hh6aba3q.com:8080/ubuntu']),
+        # Test that non-ASCII non-hostname parts of URLs are unchanged
+        (None, 'ТεЅТ̣', ['http://www.example.com/%(region)s/ubuntu'],
+         ['http://www.example.com/ТεЅТ̣/ubuntu']),
+        # Test that IPv4 addresses are unchanged
+        (None, 'fk-fake-1', ['http://192.168.1.1:8080/%(region)s/ubuntu'],
+         ['http://192.168.1.1:8080/fk-fake-1/ubuntu']),
+        # Test that IPv6 addresses are unchanged
+        (None, 'fk-fake-1',
+         ['http://[2001:67c:1360:8001::23]/%(region)s/ubuntu'],
+         ['http://[2001:67c:1360:8001::23]/fk-fake-1/ubuntu']),
+        # Test that unparseable URLs are filtered out of the mirror list
+        (None, 'inv[lid',
+         ['http://%(region)s.in.hostname/should/be/filtered',
+          'http://but.not.in.the.path/%(region)s'],
+         ['http://but.not.in.the.path/inv[lid']),
+    ) + (
+        # Dynamically generate a test case for each non-LDH
+        # (Letters/Digits/Hyphen) ASCII character, testing that it is
+        # substituted with a hyphen
+        tuple(
+            (None, 'fk{0}fake{0}1'.format(invalid_char),
+             ['http://%(region)s/ubuntu'], ['http://fk-fake-1/ubuntu'])
+            for invalid_char in INVALID_URL_CHARS))
+    )
     def test_substitution(self, availability_zone, region, patterns, expected):
         """Test substitution works as expected."""
         m_data_source = mock.Mock(