Store hash of access as primary key for ec2 type.

In current v3 credential implementation when
a new ec2 credential is created we store it with
new credential id in the backend. This fix is provided
to store the hash of access key id as credential
id when a new ec2 v3 credential is created.

Fixes Bug #1185582

Change-Id: I0ad9aaa3282b101adb6228582f0d24349232ce18

keystone/credential/controllers.py

@@ -14,13 +14,38 @@
 # License for the specific language governing permissions and limitations
 # under the License.
 
+import hashlib
+import json
+
 from keystone.common import controller
+from keystone import exception
 
 
 class CredentialV3(controller.V3Controller):
     collection_name = 'credentials'
     member_name = 'credential'
 
+    def _assign_unique_id(self, ref):
+        # Generates and assigns a unique identifer to
+        # a credential reference.
+        if ref.get('type', '').lower() == 'ec2':
+            try:
+                blob = json.loads(ref.get('blob'))
+            except (ValueError, TypeError):
+                raise exception.ValidationError(
+                    message=_('Invalid blob in credential'))
+            if not blob or not isinstance(blob, dict):
+                raise exception.ValidationError(attribute='blob',
+                                                target='credential')
+            if blob.get('access') is None:
+                raise exception.ValidationError(attribute='access',
+                                                target='blob')
+            ref = ref.copy()
+            ref['id'] = hashlib.sha256(blob['access']).hexdigest()
+            return ref
+        else:
+            return super(CredentialV3, self)._assign_unique_id(ref)
+
     @controller.protected
     def create_credential(self, context, credential):
         ref = self._assign_unique_id(self._normalize_dict(credential))

keystone/tests/test_v3_credential.py

@@ -14,6 +14,8 @@
 # License for the specific language governing permissions and limitations
 # under the License.
 
+import hashlib
+import json
 import uuid
 
 import test_v3
@@ -76,3 +78,53 @@ def test_delete_credential(self):
         self.delete(
             '/credentials/%(credential_id)s' % {
                 'credential_id': self.credential_id})
+
+    def test_create_ec2_credential(self):
+        """Call ``POST /credentials`` for creating ec2 credential."""
+        ref = self.new_credential_ref(user_id=self.user['id'])
+        blob = {"access": uuid.uuid4().hex,
+                "secret": uuid.uuid4().hex}
+        ref['blob'] = json.dumps(blob)
+        ref['type'] = 'ec2'
+        r = self.post(
+            '/credentials',
+            body={'credential': ref})
+        self.assertValidCredentialResponse(r, ref)
+        # Assert credential id is same as hash of access key id for
+        # ec2 credentials
+        self.assertEqual(r.result['credential']['id'],
+                         hashlib.sha256(blob['access']).hexdigest())
+        # Create second ec2 credential with the same access key id and check
+        # for conflict.
+        self.post(
+            '/credentials',
+            body={'credential': ref}, expected_status=409)
+
+    def test_create_non_ec2_credential(self):
+        """Call ``POST /credentials`` for creating non-ec2 credential."""
+        ref = self.new_credential_ref(user_id=self.user['id'])
+        blob = {"access": uuid.uuid4().hex,
+                "secret": uuid.uuid4().hex}
+        ref['blob'] = json.dumps(blob)
+        r = self.post(
+            '/credentials',
+            body={'credential': ref})
+        self.assertValidCredentialResponse(r, ref)
+        # Assert credential id is not same as hash of access key id for
+        # non-ec2 credentials
+        self.assertNotEqual(r.result['credential']['id'],
+                            hashlib.sha256(blob['access']).hexdigest())
+
+    def test_create_ec2_credential_with_invalid_blob(self):
+        """Call ``POST /credentials`` for creating ec2
+           credential with invalid blob.
+        """
+        ref = self.new_credential_ref(user_id=self.user['id'])
+        ref['blob'] = '{"abc":"def"d}'
+        ref['type'] = 'ec2'
+        # Assert 400 status for bad request containing invalid
+        # blob
+        response = self.post(
+            '/credentials',
+            body={'credential': ref}, expected_status=400)
+        self.assertValidErrorResponse(response)