handle unicode objects and utf-8 strings in url and params and encode them to utf-8 when serializing

Author: zookos

oauth2/__init__.py

@@ -90,16 +90,47 @@ def build_xoauth_string(url, consumer, token=None):
 
 def to_unicode(s):
     """ Convert to unicode, raise exception with instructive error
-    message if s is not unicode or ascii. """
+    message if s is not unicode, ascii, or utf-8. """
     if not isinstance(s, unicode):
         if not isinstance(s, str):
             raise TypeError('You are required to pass either unicode or string here, not: %r (%s)' % (type(s), s))
         try:
-            s = s.decode('ascii')
+            s = s.decode('utf-8')
         except UnicodeDecodeError, le:
-            raise TypeError('You are required to pass either a unicode object or an ascii string here. You passed a Python string object which contained non-ascii: %r. The UnicodeDecodeError that resulted from attempting to interpret it as ascii was: %s' % (s, le,))
+            raise TypeError('You are required to pass either a unicode object or a utf-8 string here. You passed a Python string object which contained non-utf-8: %r. The UnicodeDecodeError that resulted from attempting to interpret it as utf-8 was: %s' % (s, le,))
     return s
 
+def to_utf8(s):
+    return to_unicode(s).encode('utf-8')
+
+def to_unicode_if_string(s):
+    if isinstance(s, basestring):
+        return to_unicode(s)
+    else:
+        return s
+
+def to_utf8_if_string(s):
+    if isinstance(s, basestring):
+        return to_utf8(s)
+    else:
+        return s
+
+def to_unicode_optional_iterator(x):
+    """
+    Raise TypeError if x is a str containing non-utf8 bytes or if x is
+    an iterable which contains such a str.
+    """
+    if isinstance(x, basestring):
+        return to_unicode(x)
+
+    try:
+        l = list(x)
+    except TypeError, e:
+        assert 'is not iterable' in str(e)
+        return x
+    else:
+            return [ to_unicode(e) for e in l ]
+
 def escape(s):
     """Escape a URL including any /."""
     s = to_unicode(s)
@@ -292,8 +323,12 @@ def __init__(self, method=HTTP_METHOD, url=None, parameters=None):
             self.url = to_unicode(url)
         self.method = method
         if parameters is not None:
-            self.update(parameters)
- 
+            for k, v in parameters.iteritems():
+                k = to_unicode(k)
+                v = to_unicode_optional_iterator(v)
+                self[k] = v
+
+
     @setter
     def url(self, value):
         self.__dict__['url'] = value
@@ -383,7 +418,7 @@ def get_parameter(self, parameter):
             raise Error('Parameter not found: %s' % parameter)
 
         return ret
- 
+
     def get_normalized_parameters(self):
         """Return a string that contains the parameters that must be signed."""
         items = []
@@ -392,16 +427,22 @@ def get_normalized_parameters(self):
                 continue
             # 1.0a/9.1.1 states that kvp must be sorted by key, then by value,
             # so we unpack sequence values into multiple items for sorting.
-            if hasattr(value, '__iter__'):
-                items.extend((key, item) for item in value)
+            if isinstance(value, basestring):
+                items.append((to_utf8_if_string(key), to_utf8(value)))
             else:
-                items.append((key, value))
+                try:
+                    value = list(value)
+                except TypeError, e:
+                    assert 'is not iterable' in str(e)
+                    items.append((to_utf8_if_string(key), to_utf8_if_string(value)))
+                else:
+                    items.extend((to_utf8_if_string(key), to_utf8_if_string(item)) for item in value)
 
         # Include any query string parameters from the provided URL
         query = urlparse.urlparse(self.url)[4]
-        
+
         url_items = self._split_url_string(query).items()
-        non_oauth_url_items = list([(k, v) for k, v in url_items  if not k.startswith('oauth_')])
+        non_oauth_url_items = list([(to_utf8(k), to_utf8(v)) for k, v in url_items  if not k.startswith('oauth_')])
         items.extend(non_oauth_url_items)
 
         encoded_str = urllib.urlencode(sorted(items))
@@ -410,7 +451,7 @@ def get_normalized_parameters(self):
         # (http://tools.ietf.org/html/draft-hammer-oauth-07#section-3.6)
         # Spaces must be encoded with "%20" instead of "+"
         return encoded_str.replace('+', '%20').replace('%7E', '~')
- 
+
     def sign_request(self, signature_method, consumer, token):
         """Set the signature parameter to the result of sign."""
 

tests/test_oauth.py

@@ -248,7 +248,24 @@ def test_from_string(self):
         new = oauth.Token.from_string(string)
         self._compare_tokens(new)
 
-class TestRequest(unittest.TestCase):
+class ReallyEqualMixin:
+    def failUnlessReallyEqual(self, a, b, msg=None):
+        self.failUnlessEqual(a, b, msg=msg)
+        self.failUnlessEqual(type(a), type(b), msg="a :: %r, b :: %r, %r" % (a, b, msg))
+
+class TestFuncs(unittest.TestCase):
+    def test_to_unicode(self):
+        self.failUnlessRaises(TypeError, oauth.to_unicode, '\xae')
+        self.failUnlessRaises(TypeError, oauth.to_unicode_optional_iterator, '\xae')
+        self.failUnlessRaises(TypeError, oauth.to_unicode_optional_iterator, ['\xae'])
+
+        self.failUnlessEqual(oauth.to_unicode(':-)'), u':-)')
+        self.failUnlessEqual(oauth.to_unicode(u'\u00ae'), u'\u00ae')
+        self.failUnlessEqual(oauth.to_unicode('\xc2\xae'), u'\u00ae')
+        self.failUnlessEqual(oauth.to_unicode_optional_iterator([':-)']), [u':-)'])
+        self.failUnlessEqual(oauth.to_unicode_optional_iterator([u'\u00ae']), [u'\u00ae'])
+
+class TestRequest(unittest.TestCase, ReallyEqualMixin):
     def test_setter(self):
         url = "http://example.com"
         method = "GET"
@@ -342,9 +359,11 @@ def test_get_nonoauth_parameters(self):
         }
 
         other_params = {
-            'foo': 'baz',
-            'bar': 'foo',
-            'multi': ['FOO','BAR']
+            u'foo': u'baz',
+            u'bar': u'foo',
+            u'multi': [u'FOO',u'BAR'],
+            u'uni_utf8': u'\xae',
+            u'uni_unicode': u'\u00ae'
         }
 
         params = oauth_params
@@ -461,6 +480,24 @@ def test_to_url_with_query(self):
         self.assertEquals(b['max-contacts'], ['10'])
         self.assertEquals(a, b)
 
+    def test_signature_base_string_nonascii(self):
+        consumer = oauth.Consumer('consumer_token', 'consumer_secret')
+
+        url = "http://api.simplegeo.com:80/1.0/places/address.json?q=monkeys&category=animal&address=41+Decatur+St%2C+San+Francisc%E2%9D%A6%2C+CA"
+        req = oauth.Request("GET", url)
+        self.failUnlessReallyEqual(req.normalized_url, u'http://api.simplegeo.com/1.0/places/address.json')
+        self.assertEquals(req.url, u'http://api.simplegeo.com:80/1.0/places/address.json?q=monkeys&category=animal&address=41+Decatur+St%2C+San+Francisc%E2%9D%A6%2C+CA')
+        req.sign_request(oauth.SignatureMethod_HMAC_SHA1(), consumer, None)
+
+    def test_signature_base_string_nonascii_nonutf8(self):
+        consumer = oauth.Consumer('consumer_token', 'consumer_secret')
+
+        url = "http://api.simplegeo.com:80/1.0/places/address.json?q=monkeys&category=animal&address=41+Decatur+St%2C+San+Francisc%E2%9D%A6%2C+CA"
+        req = oauth.Request("GET", url)
+        self.failUnlessReallyEqual(req.normalized_url, u'http://api.simplegeo.com/1.0/places/address.json')
+        self.assertEquals(req.url, u'http://api.simplegeo.com:80/1.0/places/address.json?q=monkeys&category=animal&address=41+Decatur+St%2C+San+Francisc%E2%9D%A6%2C+CA')
+        req.sign_request(oauth.SignatureMethod_HMAC_SHA1(), consumer, None)
+
     def test_signature_base_string_with_query(self):
         url = "https://www.google.com/m8/feeds/contacts/default/full/?alt=json&max-contacts=10"
         params = {
@@ -495,16 +532,18 @@ def test_get_normalized_parameters(self):
             'oauth_consumer_key': "0685bd9184jfhq22",
             'oauth_signature_method': "HMAC-SHA1",
             'oauth_token': "ad180jjd733klru7",
-            'multi': ['FOO','BAR'],
+            'multi': ['FOO','BAR', u'\u00ae', '\xc2\xae'],
+            'uni_utf8': '\xc2\xae',
+            'uni_unicode': u'\u00ae'
         }
 
         req = oauth.Request("GET", url, params)
 
         res = req.get_normalized_parameters()
-        
-        srtd = [(k, v if type(v) != ListType else sorted(v)) for k,v in sorted(params.items())]
 
-        self.assertEquals(urllib.urlencode(srtd, True), res)
+        expected='multi=BAR&multi=FOO&multi=%C2%AE&multi=%C2%AE&oauth_consumer_key=0685bd9184jfhq22&oauth_nonce=4572616e48616d6d65724c61686176&oauth_signature_method=HMAC-SHA1&oauth_timestamp=137131200&oauth_token=ad180jjd733klru7&oauth_version=1.0&uni_unicode=%C2%AE&uni_utf8=%C2%AE'
+
+        self.assertEquals(expected, res)
 
     def test_get_normalized_parameters_ignores_auth_signature(self):
         url = "http://sp.example.com/"
@@ -558,24 +597,63 @@ def test_get_normalized_string_escapes_spaces_properly(self):
         expected = urllib.urlencode(sorted(params.items())).replace('+', '%20')
         self.assertEqual(expected, res)
 
-    def test_request_nonascii_bytes(self):
-        # If someone has a sequence of bytes which is not ascii, we'll
-        # raise an exception as early as possible.
-        url = "http://sp.example.com/\x92"
+    @mock.patch('oauth2.Request.make_timestamp')
+    @mock.patch('oauth2.Request.make_nonce')
+    def test_request_nonascii_bytes(self, mock_make_nonce, mock_make_timestamp):
+        mock_make_nonce.return_value = 5
+        mock_make_timestamp.return_value = 6
 
+        tok = oauth.Token(key="tok-test-key", secret="tok-test-secret")
+        con = oauth.Consumer(key="con-test-key", secret="con-test-secret")
         params = {
             'oauth_version': "1.0",
             'oauth_nonce': "4572616e48616d6d65724c61686176",
-            'oauth_timestamp': "137131200"
+            'oauth_timestamp': "137131200",
+            'oauth_token': tok.key,
+            'oauth_consumer_key': con.key
         }
 
-        tok = oauth.Token(key="tok-test-key", secret="tok-test-secret")
-        con = oauth.Consumer(key="con-test-key", secret="con-test-secret")
+        # If someone passes a sequence of bytes which is not ascii for
+        # url, we'll raise an exception as early as possible.
+        url = "http://sp.example.com/\x92" # It's actually cp1252-encoding...
+        self.assertRaises(TypeError, oauth.Request, method="GET", url=url, parameters=params)
 
-        params['oauth_token'] = tok.key
-        params['oauth_consumer_key'] = con.key
+        # And if they pass an unicode, then we'll use it.
+        url = u'http://sp.example.com/\u2019'
+        req = oauth.Request(method="GET", url=url, parameters=params)
+        req.sign_request(oauth.SignatureMethod_HMAC_SHA1(), con, None)
+        self.failUnlessReallyEqual(req['oauth_signature'], '/DgF7cY2friC01cmOAFdu8S0z+A=')
+
+        # And if it is a utf-8-encoded-then-percent-encoded non-ascii
+        # thing, we'll decode it and use it.
+        url = "http://sp.example.com/%E2%80%99"
+        req = oauth.Request(method="GET", url=url, parameters=params)
+        req.sign_request(oauth.SignatureMethod_HMAC_SHA1(), con, None)
+        self.failUnlessReallyEqual(req['oauth_signature'], 'anzjnpdqCUJWvePgDiwMb7Q8g28=')
+
+        # Same thing with the params.
+        url = "http://sp.example.com/"
+
+        # If someone passes a sequence of bytes which is not ascii in
+        # params, we'll raise an exception as early as possible.
+        params['non_oauth_thing'] = '\xae', # It's actually cp1252-encoding...
         self.assertRaises(TypeError, oauth.Request, method="GET", url=url, parameters=params)
 
+        # And if they pass a unicode, then we'll use it.
+        params['non_oauth_thing'] = u'\u2019'
+        req = oauth.Request(method="GET", url=url, parameters=params)
+        req.sign_request(oauth.SignatureMethod_HMAC_SHA1(), con, None)
+        self.failUnlessReallyEqual(req['oauth_signature'], 'QcgQMe9XzNxDWpechlQKFCd2orw=')
+
+        # And if it is a utf-8-encoded non-ascii thing, we'll decode
+        # it and use it.
+        params['non_oauth_thing'] = '\xc2\xae'
+        req = oauth.Request(method="GET", url=url, parameters=params)
+        req.sign_request(oauth.SignatureMethod_HMAC_SHA1(), con, None)
+        self.failUnlessReallyEqual(req['oauth_signature'], 'OuMkgNFhlgcmEA1gIMII7aWLDgE=')
+
+
+
     def test_sign_request(self):
         url = "http://sp.example.com/"