Merge "Bay can not be deleted by other users in the same project"

magnum/common/keystone.py

@@ -187,16 +187,30 @@ class KeystoneClientV3(object):
                 trustee_user_id=trustee_user)
         return trust
 
-    def delete_trust(self, trust_id):
-        if trust_id is None:
+    def delete_trust(self, context, bay):
+        if bay.trust_id is None:
             return
+
+        # Trust can only be deleted by the user who creates it. So when
+        # other users in the same project want to delete the bay, we need
+        # use the trustee which can impersonate the trustor to delete the
+        # trust.
+        if context.user_id == bay.user_id:
+            client = self.client
+        else:
+            auth = ka_v3.Password(auth_url=self.auth_url,
+                                  user_id=bay.trustee_user_id,
+                                  password=bay.trustee_password,
+                                  trust_id=bay.trust_id)
+            sess = ka_session.Session(auth=auth)
+            client = kc_v3.Client(session=sess)
         try:
-            self.client.trusts.delete(trust_id)
+            client.trusts.delete(bay.trust_id)
         except kc_exception.NotFound:
             pass
         except Exception:
             LOG.exception(_LE('Failed to delete trust'))
-            raise exception.TrustDeleteFailed(trust_id=trust_id)
+            raise exception.TrustDeleteFailed(trust_id=bay.trust_id)
 
     def create_trustee(self, username, password, domain_id):
         try:

magnum/conductor/handlers/bay_conductor.py

@@ -199,7 +199,7 @@ class Handler(object):
             LOG.info(_LI('The stack %s was not be found during bay'
                          ' deletion.'), stack_id)
             try:
-                trust_manager.delete_trustee_and_trust(osc, bay)
+                trust_manager.delete_trustee_and_trust(osc, context, bay)
                 cert_manager.delete_certificates_from_bay(bay)
                 bay.destroy()
             except exception.BayNotFound:
@@ -282,6 +282,7 @@ class HeatPoller(object):
                  % self.bay.stack_id)
         try:
             trust_manager.delete_trustee_and_trust(self.openstack_client,
+                                                   self.context,
                                                    self.bay)
             cert_manager.delete_certificates_from_bay(self.bay)
             self.bay.destroy()

magnum/conductor/handlers/common/trust_manager.py

@@ -42,11 +42,11 @@ def create_trustee_and_trust(osc, bay):
         raise exception.TrusteeOrTrustToBayFailed(bay_uuid=bay.uuid)
 
 
-def delete_trustee_and_trust(osc, bay):
+def delete_trustee_and_trust(osc, context, bay):
     try:
         # The bay which is upgraded from Liberty doesn't have trust_id
         if bay.trust_id:
-            osc.keystone().delete_trust(bay.trust_id)
+            osc.keystone().delete_trust(context, bay)
     except Exception:
         # Exceptions are already logged by keystone().delete_trust
         pass

magnum/tests/unit/common/test_keystone.py

@@ -112,14 +112,18 @@ class KeystoneClientTest(base.TestCase):
     def test_delete_trust(self, mock_ks):
         mock_ks.return_value.trusts.delete.return_value = None
         ks_client = keystone.KeystoneClientV3(self.ctx)
-        self.assertIsNone(ks_client.delete_trust(trust_id='atrust123'))
+        bay = mock.MagicMock()
+        bay.trust_id = 'atrust123'
+        self.assertIsNone(ks_client.delete_trust(self.ctx, bay))
         mock_ks.return_value.trusts.delete.assert_called_once_with('atrust123')
 
     def test_delete_trust_not_found(self, mock_ks):
         mock_delete = mock_ks.return_value.trusts.delete
         mock_delete.side_effect = kc_exception.NotFound()
         ks_client = keystone.KeystoneClientV3(self.ctx)
-        self.assertIsNone(ks_client.delete_trust(trust_id='atrust123'))
+        bay = mock.MagicMock()
+        bay.trust_id = 'atrust123'
+        self.assertIsNone(ks_client.delete_trust(self.ctx, bay))
 
     @mock.patch('magnum.common.keystone.ka_session.Session')
     def test_create_trust_with_all_roles(self, mock_session, mock_ks):

magnum/tests/unit/conductor/handlers/common/test_trust_manager.py

@@ -84,11 +84,13 @@ class TrustManagerTestCase(base.BaseTestCase):
         mock_bay.trustee_user_id = 'trustee_user_id'
         mock_keystone = mock.MagicMock()
         self.osc.keystone.return_value = mock_keystone
+        context = mock.MagicMock()
 
-        trust_manager.delete_trustee_and_trust(self.osc, mock_bay)
+        trust_manager.delete_trustee_and_trust(self.osc, context,
+                                               mock_bay)
 
         mock_keystone.delete_trust.assert_called_once_with(
-            mock_bay.trust_id,
+            context, mock_bay
         )
         mock_keystone.delete_trustee.assert_called_once_with(
             mock_bay.trustee_user_id,
@@ -100,8 +102,10 @@ class TrustManagerTestCase(base.BaseTestCase):
         mock_bay.trustee_user_id = 'trustee_user_id'
         mock_keystone = mock.MagicMock()
         self.osc.keystone.return_value = mock_keystone
+        context = mock.MagicMock()
 
-        trust_manager.delete_trustee_and_trust(self.osc, mock_bay)
+        trust_manager.delete_trustee_and_trust(self.osc, context,
+                                               mock_bay)
 
         self.assertEqual(0, mock_keystone.delete_trust.call_count)
         mock_keystone.delete_trustee.assert_called_once_with(
@@ -114,10 +118,11 @@ class TrustManagerTestCase(base.BaseTestCase):
         mock_bay.trustee_user_id = None
         mock_keystone = mock.MagicMock()
         self.osc.keystone.return_value = mock_keystone
+        context = mock.MagicMock()
 
-        trust_manager.delete_trustee_and_trust(self.osc, mock_bay)
+        trust_manager.delete_trustee_and_trust(self.osc, context, mock_bay)
 
         mock_keystone.delete_trust.assert_called_once_with(
-            mock_bay.trust_id,
+            context, mock_bay
         )
         self.assertEqual(0, mock_keystone.delete_trustee.call_count)