Enable use of service user token with admin context

When the [service_user] section is configured in nova.conf, nova will
have the ability to send a service user token alongside the user's
token. The service user token is sent when nova calls other services'
REST APIs to authenticate as a service, and service calls can sometimes
have elevated privileges.

Currently, nova does not however have the ability to send a service user
token with an admin context. This means that when nova makes REST API
calls to other services with an anonymous admin RequestContext (such as
in nova-manage or periodic tasks), it will not be authenticated as a
service.

This adds a keyword argument to service_auth.get_auth_plugin() to
enable callers to provide a user_auth object instead of attempting to
extract the user_auth from the RequestContext.

The cinder and neutron client modules are also adjusted to make use of
the new user_auth keyword argument so that nova calls made with
anonymous admin request contexts can authenticate as a service when
configured.

Related-Bug: #2004555

Change-Id: I14df2d55f4b2f0be58f1a6ad3f19e48f7a6bfcb4

1 files changed, 16 insertions, 0 deletions

diff --git a/nova/tests/unit/network/test_neutron.py b/nova/tests/unit/network/test_neutron.py
index 9aa970aca1..c551191e4c 100644
--- a/nova/tests/unit/network/test_neutron.py
+++ b/nova/tests/unit/network/test_neutron.py
@@ -142,6 +142,22 @@ class TestNeutronClient(test.NoDBTestCase):
         self.assertIsInstance(cl.httpclient.auth,
                               service_token.ServiceTokenAuthWrapper)
 
+    @mock.patch('nova.service_auth._SERVICE_AUTH')
+    @mock.patch('nova.network.neutron._ADMIN_AUTH')
+    @mock.patch.object(ks_loading, 'load_auth_from_conf_options')
+    def test_admin_with_service_token(
+        self, mock_load, mock_admin_auth, mock_service_auth
+    ):
+        self.flags(send_service_user_token=True, group='service_user')
+
+        admin_context = context.get_admin_context()
+
+        cl = neutronapi.get_client(admin_context)
+        self.assertIsInstance(cl.httpclient.auth,
+                              service_token.ServiceTokenAuthWrapper)
+        self.assertEqual(mock_admin_auth, cl.httpclient.auth.user_auth)
+        self.assertEqual(mock_service_auth, cl.httpclient.auth.service_auth)
+
     @mock.patch.object(client.Client, "list_networks",
                        side_effect=exceptions.Unauthorized())
     def test_Unauthorized_user(self, mock_list_networks):