Decode urlencoded properly

2 files changed, 49 insertions, 0 deletions

diff --git a/oauthlib/oauth1/rfc5849/utils.py b/oauthlib/oauth1/rfc5849/utils.py
index 14cdde5..032b364 100644
--- a/oauthlib/oauth1/rfc5849/utils.py
+++ b/oauthlib/oauth1/rfc5849/utils.py
@@ -11,6 +11,7 @@ spec.
 import string
 import time
 from random import getrandbits, choice
+import urlparse
 
 UNICODE_ASCII_CHARACTER_SET = (string.ascii_letters.decode('ascii') +
     string.digits.decode('ascii'))
@@ -161,7 +162,38 @@ def urlencode(query):
         query = query.items()
     return u"&".join([u'='.join([escape(k), escape(v)]) for k, v in query])
 
+urlencoded = set(always_safe) | set(u'=&;%+~')
 
+def urldecode(query):
+    """Decode a query string in x-www-form-urlencoded format into a sequence
+    of two-element tuples. 
+
+    Unlike urlparse.parse_qsl(..., strict_parsing=True) urldecode will enforce
+    correct formatting of the query string by validation. If validation fails
+    a ValueError will be raised. urllib.parse_qsl will only raise errors if
+    name-value pairs include contains a name and omits the equals sign. 
+    """
+    # Check if query contains invalid characters
+    if query and not set(query) <= urlencoded:
+        raise ValueError('Invalid characters in query string.')
+
+    # Check for correctly hex encoded values using a regular expression
+    # All encoded values begin with % followed by two hex characters
+    # correct = %00, %A0, %0A, %FF
+    # invalid = %G0, %5H, %PO
+    import re
+    first_invalid = u'%[^0-9A-Fa-f][0-9A-Fa-f]'
+    last_invalid = '%[0-9A-Fa-f][^0-9A-Fa-f]'
+    both_invalid = u'%[^0-9A-Fa-f][^0-9A-Fa-f]'
+    invalid_hex = u'|'.join((first_invalid, last_invalid, both_invalid))
+    if len(re.findall(invalid_hex, query)):
+        raise ValueError('Invalid hex encoding in query string.')
+
+    # We want to allow queries such as "c2" whereas urlparse.parse_qsl
+    # with the strict_parsing flag will not.  
+    return urlparse.parse_qsl(query, keep_blank_values=True) 
+
+    
 def parse_keqv_list(l):
     """A unicode-safe version of urllib2.parse_keqv_list"""
     parsed = {}
diff --git a/tests/oauth1/rfc5849/test_utils.py b/tests/oauth1/rfc5849/test_utils.py
index 77b913b..901bcb9 100644
--- a/tests/oauth1/rfc5849/test_utils.py
+++ b/tests/oauth1/rfc5849/test_utils.py
@@ -130,6 +130,23 @@ class UtilsTests(TestCase):
         self.assertEqual(urlencode(self.sample_params_unicode_list), "notoauth=shouldnotbehere&oauth_consumer_key=9djdj82h48djs9d2&oauth_token=kkk9d7dh3k39sjv7&notoautheither=shouldnotbehere")
         self.assertEqual(urlencode(self.sample_params_unicode_dict), "notoauth=shouldnotbehere&oauth_consumer_key=9djdj82h48djs9d2&oauth_token=kkk9d7dh3k39sjv7&notoautheither=shouldnotbehere")
 
+    def test_urldecode(self):
+        
+        self.assertEqual(urldecode(u''), [])
+        self.assertEqual(urldecode(u'='), [(u'', u'')])
+        self.assertEqual(urldecode(u'%20'), [(u' ', u'')])
+        self.assertEqual(urldecode(u'+'), [(u' ', u'')])
+        self.assertEqual(urldecode(u'c2'), [(u'c2', u'')])
+        self.assertEqual(urldecode(u'c2='), [(u'c2', u'')])
+        self.assertEqual(urldecode(u'foo=bar'), [(u'foo', u'bar')])
+        self.assertEqual(urldecode(u'foo_%20~=.bar-'), [(u'foo_ ~', u'.bar-')])
+        
+        self.assertRaises(ValueError, urldecode, u'foo bar')
+        self.assertRaises(ValueError, urldecode, u'?')
+        self.assertRaises(ValueError, urldecode, u'%?A')
+        self.assertRaises(ValueError, urldecode, u'%A?')
+        self.assertRaises(ValueError, urldecode, u'%??')
+
     def test_parse_authorization_header(self):
 
         # make us some headers