Merge "Configurable token hash algorithm"

etc/keystone.conf.sample

@@ -1302,6 +1302,13 @@
 # value)
 #revoke_by_id=true
 
+# The hash algorithm to use for PKI tokens. This can be set to
+# any algorithm that hashlib supports. WARNING: Before
+# changing this value, the auth_token middleware must be
+# configured with the hash_algorithms, otherwise token
+# revocation will not be processed correctly. (string value)
+#hash_algorithm=md5
+
 
 [trust]
 

keystone/common/config.py

@@ -216,6 +216,13 @@ FILE_OPTIONS = {
                     'list of tokens to revoke. Only disable if you are '
                     'switching to using the Revoke extension with a '
                     'backend other than KVS, which stores events in memory.'),
+        cfg.StrOpt('hash_algorithm', default='md5',
+                   help="The hash algorithm to use for PKI tokens. This can "
+                        "be set to any algorithm that hashlib supports. "
+                        "WARNING: Before changing this value, the auth_token "
+                        "middleware must be configured with the "
+                        "hash_algorithms, otherwise token revocation will "
+                        "not be processed correctly."),
     ],
     'revoke': [
         cfg.StrOpt('driver',

keystone/tests/test_backend.py

@@ -3221,9 +3221,9 @@ class TokenTests(object):
         self.assertIn(token_id, revoked_tokens)
         self.assertIn(token2_id, revoked_tokens)
 
-    def test_predictable_revoked_pki_token_id(self):
+    def _test_predictable_revoked_pki_token_id(self, hash_fn):
         token_id = self._create_token_id()
-        token_id_hash = hashlib.md5(token_id).hexdigest()
+        token_id_hash = hash_fn(token_id).hexdigest()
         token = {'user': {'id': uuid.uuid4().hex}}
 
         self.token_api.create_token(token_id, token)
@@ -3235,6 +3235,13 @@ class TokenTests(object):
         for t in self.token_api.list_revoked_tokens():
             self.assertIn('expires', t)
 
+    def test_predictable_revoked_pki_token_id_default(self):
+        self._test_predictable_revoked_pki_token_id(hashlib.md5)
+
+    def test_predictable_revoked_pki_token_id_sha256(self):
+        self.config_fixture.config(group='token', hash_algorithm='sha256')
+        self._test_predictable_revoked_pki_token_id(hashlib.sha256)
+
     def test_predictable_revoked_uuid_token_id(self):
         token_id = uuid.uuid4().hex
         token = {'user': {'id': uuid.uuid4().hex}}

keystone/tests/test_content_types.py

@@ -12,9 +12,12 @@
 # License for the specific language governing permissions and limitations
 # under the License.
 
+import json
 import uuid
 
+from keystoneclient.common import cms
 import six
+from testtools import matchers
 
 from keystone.common import extension
 from keystone import config
@@ -1209,6 +1212,54 @@ class JsonTestCase(RestfulTestCase, CoreApiTests, LegacyV2UsernameTests):
     def assertValidRevocationListResponse(self, response):
         self.assertIsNotNone(response.result['signed'])
 
+    def _fetch_parse_revocation_list(self):
+
+        token1 = self.get_scoped_token()
+
+        token2 = self.get_scoped_token()
+        self.admin_request(method='DELETE',
+                           path='/v2.0/tokens/%s' % token2,
+                           token=token1)
+
+        r = self.admin_request(
+            method='GET',
+            path='/v2.0/tokens/revoked',
+            token=token1,
+            expected_status=200)
+        signed_text = r.result['signed']
+
+        data_json = cms.cms_verify(signed_text, CONF.signing.certfile,
+                                   CONF.signing.ca_certs)
+
+        data = json.loads(data_json)
+
+        return (data, token2)
+
+    def test_fetch_revocation_list_md5(self):
+        """If the server is configured for md5, then the revocation list has
+           tokens hashed with MD5.
+        """
+
+        # The default hash algorithm is md5.
+        hash_algorithm = 'md5'
+
+        (data, token) = self._fetch_parse_revocation_list()
+        token_hash = cms.cms_hash_token(token, mode=hash_algorithm)
+        self.assertThat(token_hash, matchers.Equals(data['revoked'][0]['id']))
+
+    def test_fetch_revocation_list_sha256(self):
+        """If the server is configured for sha256, then the revocation list has
+           tokens hashed with SHA256
+        """
+
+        hash_algorithm = 'sha256'
+        self.config_fixture.config(group='token',
+                                   hash_algorithm=hash_algorithm)
+
+        (data, token) = self._fetch_parse_revocation_list()
+        token_hash = cms.cms_hash_token(token, mode=hash_algorithm)
+        self.assertThat(token_hash, matchers.Equals(data['revoked'][0]['id']))
+
     def test_create_update_user_json_invalid_enabled_type(self):
         # Enforce usage of boolean for 'enabled' field in JSON
         token = self.get_scoped_token()
@@ -1334,6 +1385,12 @@ class RevokeApiJsonTestCase(JsonTestCase):
     def test_fetch_revocation_list_admin_200(self):
         self.skipTest('Revoke API disables revocation_list.')
 
+    def test_fetch_revocation_list_md5(self):
+        self.skipTest('Revoke API disables revocation_list.')
+
+    def test_fetch_revocation_list_sha256(self):
+        self.skipTest('Revoke API disables revocation_list.')
+
 
 class XmlTestCase(RestfulTestCase, CoreApiTests, LegacyV2UsernameTests):
     xmlns = 'http://docs.openstack.org/identity/api/v2.0'

keystone/token/core.py

@@ -122,7 +122,7 @@ class Manager(manager.Manager):
                   returns the passed-in value (such as a UUID token ID or an
                   existing hash).
         """
-        return cms.cms_hash_token(token_id)
+        return cms.cms_hash_token(token_id, mode=CONF.token.hash_algorithm)
 
     def _assert_valid(self, token_id, token_ref):
         """Raise TokenNotFound if the token is expired."""