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

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.

Change-Id: I9f87cdf07066d316722e798cd0755f0fff5c2a02
Closes-Bug: #1552457

magnum/common/keystone.py

@@ -187,16 +187,30 @@ class KeystoneClientV3(object):
                 trustee_user_id=trustee_user)
         return trust
 
-    def delete_trust(self, trust_id):
+    def delete_trust(self, context, bay):
-        if trust_id is None:
+        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)