Merge pull request #417 from bjmc/sanitize-logging

Improves sanitizing sensitive data from Request.__repr__

2 files changed, 29 insertions, 4 deletions

diff --git a/oauthlib/common.py b/oauthlib/common.py
index e0e80c8..6b96de8 100644
--- a/oauthlib/common.py
+++ b/oauthlib/common.py
@@ -36,7 +36,7 @@ UNICODE_ASCII_CHARACTER_SET = ('abcdefghijklmnopqrstuvwxyz'
 CLIENT_ID_CHARACTER_SET = (r' !"#$%&\'()*+,-./0123456789:;<=>?@ABCDEFGHIJKLMN'
                            'OPQRSTUVWXYZ[\\]^_`abcdefghijklmnopqrstuvwxyz{|}')
 
-PASSWORD_PATTERN = re.compile(r'password=[^&]+')
+SANITIZE_PATTERN = re.compile(r'([^&;]*(?:password|token)[^=]*=)[^&;]+', re.IGNORECASE)
 INVALID_HEX_PATTERN = re.compile(r'%[^0-9A-Fa-f]|%[0-9A-Fa-f][^0-9A-Fa-f]')
 
 always_safe = ('ABCDEFGHIJKLMNOPQRSTUVWXYZ'
@@ -414,10 +414,13 @@ class Request(object):
 
     def __repr__(self):
         body = self.body
-        if body and 'password=' in body:
-            body = PASSWORD_PATTERN.sub('password=***', body)
+        headers = self.headers.copy()
+        if body:
+            body = SANITIZE_PATTERN.sub('\1<SANITIZED>', body)
+        if 'Authorization' in headers:
+            headers['Authorization'] = '<SANITIZED>'
         return '<oauthlib.Request url="%s", http_method="%s", headers="%s", body="%s">' % (
-            self.uri, self.http_method, self.headers, body)
+            self.uri, self.http_method, headers, body)
 
     @property
     def uri_query(self):
diff --git a/tests/test_common.py b/tests/test_common.py
index fb28fa4..078b67b 100644
--- a/tests/test_common.py
+++ b/tests/test_common.py
@@ -187,14 +187,36 @@ class RequestTest(TestCase):
         with self.assertRaises(AttributeError):
             getattr(r, 'does_not_exist')
 
+    def test_sanitizing_authorization_header(self):
+        r = Request(URI, headers={'Accept': 'application/json',
+                                  'Authorization': 'Basic Zm9vOmJhcg=='}
+                    )
+        self.assertNotIn('Zm9vOmJhcg==', repr(r))
+        self.assertIn('<SANITIZED>', repr(r))
+        # Double-check we didn't modify the underlying object:
+        self.assertEqual(r.headers['Authorization'], 'Basic Zm9vOmJhcg==')
+
+    def test_token_body(self):
+        payload = 'client_id=foo&refresh_token=bar'
+        r = Request(URI, body=payload)
+        self.assertNotIn('bar', repr(r))
+        self.assertIn('<SANITIZED>', repr(r))
+
+        payload = 'refresh_token=bar&client_id=foo'
+        r = Request(URI, body=payload)
+        self.assertNotIn('bar', repr(r))
+        self.assertIn('<SANITIZED>', repr(r))
+
     def test_password_body(self):
         payload = 'username=foo&password=bar'
         r = Request(URI, body=payload)
         self.assertNotIn('bar', repr(r))
+        self.assertIn('<SANITIZED>', repr(r))
 
         payload = 'password=bar&username=foo'
         r = Request(URI, body=payload)
         self.assertNotIn('bar', repr(r))
+        self.assertIn('<SANITIZED>', repr(r))
 
 
 class CaseInsensitiveDictTest(TestCase):