diff options
| -rw-r--r-- | ChangeLog | 1 | ||||
| -rw-r--r-- | cloudinit/user_data.py | 28 | ||||
| -rw-r--r-- | tests/unittests/test_data.py | 12 |
3 files changed, 29 insertions, 12 deletions
@@ -125,6 +125,7 @@ - Remove trailing dot from GCE metadata URL (LP: #1581200) [Phil Roche] - support network rendering to sysconfig (for centos and RHEL) - write_files: if no permissions are given, just use default without warn. + - user_data: fix error when user-data is not utf-8 decodable (LP: #1532072) 0.7.6: - open 0.7.6 diff --git a/cloudinit/user_data.py b/cloudinit/user_data.py index f0631906..393bf0bb 100644 --- a/cloudinit/user_data.py +++ b/cloudinit/user_data.py @@ -334,19 +334,23 @@ def is_skippable(part): # Coverts a raw string into a mime message -def convert_string(raw_data, headers=None): +def convert_string(raw_data, content_type=NOT_MULTIPART_TYPE): if not raw_data: raw_data = '' - if not headers: - headers = {} - data = util.decode_binary(util.decomp_gzip(raw_data)) - if "mime-version:" in data[0:4096].lower(): - msg = util.message_from_string(data) - for (key, val) in headers.items(): - _replace_header(msg, key, val) - else: - mtype = headers.get(CONTENT_TYPE, NOT_MULTIPART_TYPE) - maintype, subtype = mtype.split("/", 1) - msg = MIMEBase(maintype, subtype, *headers) + + def create_binmsg(data, content_type): + maintype, subtype = content_type.split("/", 1) + msg = MIMEBase(maintype, subtype) msg.set_payload(data) + return msg + + try: + data = util.decode_binary(util.decomp_gzip(raw_data)) + if "mime-version:" in data[0:4096].lower(): + msg = util.message_from_string(data) + else: + msg = create_binmsg(data, content_type) + except UnicodeDecodeError: + msg = create_binmsg(raw_data, content_type) + return msg diff --git a/tests/unittests/test_data.py b/tests/unittests/test_data.py index 1923e2af..17951117 100644 --- a/tests/unittests/test_data.py +++ b/tests/unittests/test_data.py @@ -557,3 +557,15 @@ class TestUDProcess(helpers.ResourceUsingTestCase): ud_proc = ud.UserDataProcessor(self.getCloudPaths()) message = ud_proc.process(msg) self.assertTrue(count_messages(message) == 1) + + +class TestConvert(helpers.TestCase): + def test_handles_binary(self): + blob = b'\x32\x99' + msg = ud.convert_string(blob) + self.assertEqual(blob, msg.get_payload(decode=True)) + + def test_handle_headers(self): + text = "hi mom" + msg = ud.convert_string(text) + self.assertEqual(text, msg.get_payload(decode=False)) |