Fix duplicate handling for user-specified IDs

For resources such as federation protocols and federation mappings, the
database primary keys are ID strings specified by the user creating
them. If the user created such a resource that happened to have the
substrings 'id' or 'name' in the identifier, and then by accident tried
to create it again, it would fail with a message that did not appear to
relate to the entry being a duplicate:

 string indices must be integers (HTTP 400)

This was because the method that is supposed to form a user-friendly
message receives all the arguments as a tuple and iterates over it,
looking for a dictionary with the keys 'id' or 'name' to figure out what
was trying to be duplicated. However, it can't distinguish between a
dictionary with 'id' or 'name' as a key and a string with 'id' or 'name'
as a substring, and trips if it finds such a string. This logic for
looking for 'id', 'name', or 'domain_id' in an object really only makes
sense if the object is a dict, so this patch adds a check to ensure it
is a dict before looking for keys in it.

Change-Id: If3c23a28eb5594efaa49c6a15d8db11cfc8d9057
Closes-bug: #1668563

keystone/common/sql/core.py

@@ -449,14 +449,15 @@ def handle_conflicts(conflict_type='object'):
                 # We want to store the duplicate objects name in the error
                 # message for the user. If name is not available we use the id.
                 for arg in params:
-                    if 'name' in arg:
-                        field = 'name'
-                        name = arg['name']
-                    elif 'id' in arg:
-                        field = 'ID'
-                        name = arg['id']
-                    if 'domain_id' in arg:
-                        domain_id = arg['domain_id']
+                    if isinstance(arg, dict):
+                        if 'name' in arg:
+                            field = 'name'
+                            name = arg['name']
+                        elif 'id' in arg:
+                            field = 'ID'
+                            name = arg['id']
+                        if 'domain_id' in arg:
+                            domain_id = arg['domain_id']
                 msg = _('Duplicate entry')
                 if name and domain_id:
                     msg = _('Duplicate entry found with %(field)s %(name)s '

keystone/tests/unit/common/test_database_conflicts.py

@@ -134,6 +134,19 @@ class DuplicateTestCase(test_v3.RestfulTestCase):
         else:
             self.fail("Create duplicate mapping did not raise a conflict")
 
+    def test_mapping_duplicate_conflict_with_id_in_id(self):
+        self.mapping = mapping_fixtures.MAPPING_EPHEMERAL_USER
+        self.mapping['id'] = 'mapping_with_id_in_the_id'
+        self.federation_api.create_mapping(self.mapping['id'],
+                                           self.mapping)
+        try:
+            self.federation_api.create_mapping(self.mapping['id'],
+                                               self.mapping)
+        except exception.Conflict as e:
+            self.assertIn("Duplicate entry found with ID %s"
+                          % self.mapping['id'], repr(e))
+        # Any other exception will cause the test to fail
+
     def test_region_duplicate_conflict_gives_name(self):
         region_ref = unit.new_region_ref()
         self.catalog_api.create_region(region_ref)
@@ -172,6 +185,60 @@ class DuplicateTestCase(test_v3.RestfulTestCase):
         else:
             self.fail("Create duplicate region did not raise a conflict")
 
+    def test_federation_protocol_duplicate_conflict_with_id_in_id(self):
+        self.idp = {
+            'id': uuid.uuid4().hex,
+            'enabled': True,
+            'description': uuid.uuid4().hex
+        }
+        self.federation_api.create_idp(self.idp['id'], self.idp)
+        self.mapping = mapping_fixtures.MAPPING_EPHEMERAL_USER
+        self.mapping['id'] = uuid.uuid4().hex
+        self.federation_api.create_mapping(self.mapping['id'],
+                                           self.mapping)
+        protocol = {
+            'id': 'federation_protocol_with_id_in_the_id',
+            'mapping_id': self.mapping['id']
+        }
+        protocol_ret = self.federation_api.create_protocol(self.idp['id'],
+                                                           protocol['id'],
+                                                           protocol)
+        try:
+            self.federation_api.create_protocol(self.idp['id'],
+                                                protocol['id'],
+                                                protocol)
+        except exception.Conflict as e:
+            self.assertIn("Duplicate entry found with ID %s"
+                          % protocol_ret['id'], repr(e))
+        # Any other exception will fail the test
+
+    def test_federation_protocol_duplicate_conflict_with_id_in_idp_id(self):
+        self.idp = {
+            'id': 'myidp',
+            'enabled': True,
+            'description': uuid.uuid4().hex
+        }
+        self.federation_api.create_idp(self.idp['id'], self.idp)
+        self.mapping = mapping_fixtures.MAPPING_EPHEMERAL_USER
+        self.mapping['id'] = uuid.uuid4().hex
+        self.federation_api.create_mapping(self.mapping['id'],
+                                           self.mapping)
+        protocol = {
+            'id': uuid.uuid4().hex,
+            'mapping_id': self.mapping['id']
+        }
+        protocol_ret = self.federation_api.create_protocol(self.idp['id'],
+                                                           protocol['id'],
+                                                           protocol)
+        try:
+            self.federation_api.create_protocol(self.idp['id'],
+                                                protocol['id'],
+                                                protocol)
+        except exception.Conflict as e:
+            self.assertIn("Duplicate entry found with ID %s"
+                          % protocol_ret['id'], repr(e))
+        # Any other exception will fail the test
+
     def test_sp_duplicate_conflict_gives_name(self):
         sp = {
             'auth_url': uuid.uuid4().hex,