Prevent a UnicodeDecodeError in the s3token middleware

Mixing "str" and "unicode" can lead to a UnicodeDecodeError. We encode unicode
values before using them with text strings.

The bug occurs if the URL contains a non-ASCII character in the path:
"hého" in "/v1/AUTH_cfa/c/hého" ("/v1/AUTH_cfa/c/h\xc3\xa9ho" in UTF-8) for
example.

The bug occurs on Python 2 in s3_token.py because the tenant id is retrieved
from identity_info['access']['token']['tenant'] and identity_info comes from
resp.json().

The problem is that in Python, the JSON decoder always create Unicode strings.
Example in Python 2:

>>> json.loads('{"key": "value"}') {u'key': u'value'}

There is no issue in Python 3, since all text strings are Unicode.

Change-Id: Ib7fdf60f8369ea9546fcd92f1ac385c777478d10
Closes-Bug: #1428706
Co-Authored-By: Victor Stinner <vstinner@redhat.com>

keystonemiddleware/s3_token.py

@@ -251,6 +251,8 @@ class S3Token(object):
 
         req.headers['X-Auth-Token'] = token_id
         tenant_to_connect = force_tenant or tenant['id']
+        if six.PY2 and isinstance(tenant_to_connect, six.text_type):
+            tenant_to_connect = tenant_to_connect.encode('utf-8')
         self._logger.debug('Connecting with tenant: %s', tenant_to_connect)
         new_tenant_name = '%s%s' % (self._reseller_prefix, tenant_to_connect)
         environ['PATH_INFO'] = environ['PATH_INFO'].replace(account,

keystonemiddleware/tests/unit/test_s3_token_middleware.py

@@ -17,6 +17,7 @@ from oslo_serialization import jsonutils
 import requests
 from requests_mock.contrib import fixture as rm_fixture
 import six
+from six.moves import urllib
 import webob
 
 from keystonemiddleware import s3_token
@@ -165,6 +166,13 @@ class S3TokenMiddlewareTestGood(S3TokenMiddlewareTestBase):
         middleware = s3_token.filter_factory(config)(FakeApp())
         self.assertIs('false_ind', middleware._verify)
 
+    def test_unicode_path(self):
+        url = u'/v1/AUTH_cfa/c/euro\u20ac'.encode('utf8')
+        req = webob.Request.blank(urllib.parse.quote(url))
+        req.headers['Authorization'] = 'access:signature'
+        req.headers['X-Storage-Token'] = 'token'
+        req.get_response(self.middleware)
+
 
 class S3TokenMiddlewareTestBad(S3TokenMiddlewareTestBase):
     def setUp(self):