Redact x-subject-token from response headers

When you invoke any OpenStack API of any of the OpenStack services
e.g. glance, neutron, cinder, heat, ceilometer, nova, keystone
then it logs readable x-subject-token at the debug log level in the
respective log files.

Simply redacting the x-subject-token in keystone client response header
before logging it.

SecurityImpact
Closes-Bug: #1371355
Change-Id: Iac16c6358250677544761beea9f5c5d8ba29afac

keystoneclient/session.py

@@ -116,6 +116,15 @@ class Session(object):
         if user_agent is not None:
             self.user_agent = user_agent
 
+    @classmethod
+    def process_header(cls, header):
+        """Redacts the secure headers to be logged."""
+        secure_headers = ('authorization', 'x-auth-token',
+                          'x-subject-token',)
+        if header[0].lower() in secure_headers:
+            return (header[0], 'TOKEN_REDACTED')
+        return header
+
     @utils.positional()
     def _http_log_request(self, url, method=None, data=None,
                           json=None, headers=None):
@@ -125,13 +134,6 @@ class Session(object):
             # debug log.
             return
 
-        def process_header(header):
-            secure_headers = ('authorization', 'x-auth-token',
-                              'x-subject-token',)
-            if header[0].lower() in secure_headers:
-                return (header[0], 'TOKEN_REDACTED')
-            return header
-
         string_parts = ['REQ: curl -i']
 
         # NOTE(jamielennox): None means let requests do its default validation
@@ -146,7 +148,8 @@ class Session(object):
 
         if headers:
             for header in six.iteritems(headers):
-                string_parts.append('-H "%s: %s"' % process_header(header))
+                string_parts.append('-H "%s: %s"'
+                                    % Session.process_header(header))
         if json:
             data = jsonutils.dumps(json)
         if data:
@@ -175,7 +178,8 @@ class Session(object):
         if status_code:
             string_parts.append('[%s]' % status_code)
         if headers:
-            string_parts.append('%s' % headers)
+            for header in six.iteritems(headers):
+                string_parts.append('%s: %s' % Session.process_header(header))
         if text:
             string_parts.append('\nRESP BODY: %s\n' % text)
 

keystoneclient/tests/test_session.py

@@ -138,6 +138,10 @@ class SessionTests(utils.TestCase):
                           session.get, self.TEST_URL)
 
     def test_session_debug_output(self):
+        """Test request and response headers in debug logs
+
+        in order to redact secure headers while debug is true.
+        """
         session = client_session.Session(verify=False)
         headers = {'HEADERA': 'HEADERVALB'}
         security_headers = {'Authorization': uuid.uuid4().hex,
@@ -145,10 +149,11 @@ class SessionTests(utils.TestCase):
                             'X-Subject-Token': uuid.uuid4().hex, }
         body = 'BODYRESPONSE'
         data = 'BODYDATA'
-        self.stub_url('POST', text=body)
         all_headers = dict(
             itertools.chain(headers.items(), security_headers.items()))
-        session.post(self.TEST_URL, headers=all_headers, data=data)
+        self.stub_url('POST', text=body, headers=all_headers)
+        resp = session.post(self.TEST_URL, headers=all_headers, data=data)
+        self.assertEqual(resp.status_code, 200)
 
         self.assertIn('curl', self.logger.output)
         self.assertIn('POST', self.logger.output)
@@ -159,8 +164,12 @@ class SessionTests(utils.TestCase):
         for k, v in six.iteritems(headers):
             self.assertIn(k, self.logger.output)
             self.assertIn(v, self.logger.output)
+
+        # Assert that response headers contains actual values and
+        # only debug logs has been masked
         for k, v in six.iteritems(security_headers):
-            self.assertIn(k, self.logger.output)
+            self.assertIn('%s: TOKEN_REDACTED' % k, self.logger.output)
+            self.assertEqual(v, resp.headers[k])
             self.assertNotIn(v, self.logger.output)
 
     def test_connect_retries(self):