Enable support for application credentials (#39)

* Avoid using AsyncMagicMock in python >=3.8

* Add support for application credentials

Author: giffels

README.rst

@@ -35,7 +35,7 @@ Usage
     from asyncopenstackclient import NovaClient, GlanceClient, CinderClient, AuthPassword
 
     # you can either pass credentials explicitly (as shown below)
-    # or use enviormental variables from OpenStack RC file
+    # or use environmental variables from OpenStack RC file
     # https://docs.openstack.org/mitaka/cli-reference/common/cli_set_environment_variables_using_openstack_rc.html
     auth = AuthPassword(
         auth_url='https://keystone:5999/v3'
@@ -44,6 +44,15 @@ Usage
         user_domain_name='default',
         project_domain_name='foo.bar'
     )
+
+    # alternatively you can also use application_credentials to authenticate with the OpenStack Keystone API
+    # https://docs.openstack.org/keystone/queens/user/application_credentials.html
+    alternative_auth = AuthPassword(
+        auth_url='https://keystone:5999/v3'
+        application_credential_id="ID",
+        application_credential_secret="SECRET"
+    )
+
     nova = NovaClient(session=auth)
     glance = GlanceClient(session=auth)
     cinder = CinderClient(session=auth)

src/asyncopenstackclient/auth.py

@@ -15,6 +15,8 @@ def __init__(self):
         self._project_id = None
         self._project_name = None
         self._region_name = None
+        self._application_credential_id = None
+        self._application_credential_secret = None
 
     async def authenticate(self):
         raise NotImplementedError
@@ -51,51 +53,83 @@ def os_project_name(self):
     def os_region_name(self):
         return self._region_name or os.environ.get('OS_REGION_NAME')
 
+    @property
+    def os_application_credential_id(self):
+        return self._application_credential_id or os.environ.get('OS_APPLICATION_CREDENTIAL_ID')
+
+    @property
+    def os_application_credential_secret(self):
+        return self._application_credential_secret or os.environ.get('OS_APPLICATION_CREDENTIAL_SECRET')
+
 
 class AuthPassword(AuthModel):
 
-    def __init__(self, auth_url=None, username=None, password=None, project_name=None, user_domain_name=None, project_domain_name=None):
+    def __init__(self, auth_url=None, username=None, password=None, project_name=None,
+                 user_domain_name=None, project_domain_name=None,
+                 application_credential_id=None, application_credential_secret=None):
         super().__init__()
         self._auth_url = auth_url
         self._username = username
         self._password = password
         self._project_name = project_name
         self._user_domain_name = user_domain_name
         self._project_domain_name = project_domain_name
+        self._application_credential_id = application_credential_id
+        self._application_credential_secret = application_credential_secret
 
         self._auth_endpoint = self.os_auth_url + '/auth/tokens'
         self.token = None
         self.token_expires_at = 0
         self.headers = {
             'Content-Type': 'application/json'
         }
-        self._auth_payload = {
-            'auth': {
-                'identity': {
-                    'methods': ['password'],
-                    'password': {
-                        'user': {
-                            'domain': {
-                                'name': self.os_user_domain_name
-                            },
-                            'name': self.os_username,
-                            'password': self.os_password
-                        }
-                    }
-                },
-                'scope': {
-                    "project": {
-                        "domain": {
-                            "name": self.os_project_domain_name
+
+        self._auth_payload = {'auth': {}}
+        for key_from_property in (self._identity, self._scope):
+            self._auth_payload['auth'].update(key_from_property)
+
+    def is_token_valid(self):
+        return self.token_expires_at - time() > 0
+
+    @property
+    def _identity(self):
+        if self.os_application_credential_id and self.os_application_credential_secret:
+            return {"identity": {
+                "methods": ["application_credential"],
+                "application_credential": {
+                    "id": self.os_application_credential_id,
+                    "secret": self.os_application_credential_secret
+                }
+            }
+            }
+        else:
+            return {"identity": {
+                'methods': ['password'],
+                'password': {
+                    'user': {
+                        'domain': {
+                            'name': self.os_user_domain_name
                         },
-                        "name": self.os_project_name
+                        'name': self.os_username,
+                        'password': self.os_password
                     }
                 }
-            }
-        }
+            }}
 
-    def is_token_valid(self):
-        return self.token_expires_at - time() > 0
+    @property
+    def _scope(self):
+        if self.os_application_credential_id and self.os_application_credential_secret:
+            # The scope is automatically determined from the application_credential
+            return {}
+        else:
+            return {"scope": {
+                "project": {
+                    "domain": {
+                        "name": self.os_project_domain_name
+                    },
+                    "name": self.os_project_name
+                }
+            }}
 
     async def get_token(self):
         async with aiohttp.ClientSession() as session:

tests/test_auth.py

@@ -2,7 +2,7 @@
 from aioresponses import aioresponses
 from aiounittest import AsyncTestCase, futurized
 from asyncopenstackclient import AuthPassword
-from unittest.mock import patch
+from unittest.mock import MagicMock, patch
 
 
 class TestAuth(AsyncTestCase):
@@ -19,33 +19,63 @@ def tearDown(self):
                 del os.environ[name]
 
     async def test_create_object(self):
-        expected_payload = {'auth': {
+        expected_payload_password = {'auth': {
             'identity': {'methods': ['password'], 'password': {'user': {
                 'domain': {'name': 'm_user_domain'},
                 'name': 'm_user', 'password': 'm_pass'
             }}},
             'scope': {'project': {'domain': {'name': 'm_project_domain'}, 'name': 'm_project'}}
         }}
-        self.assertEqual(self.auth._auth_payload, expected_payload)
-        self.assertEqual(self.auth._auth_endpoint, 'http://url/auth/tokens')
-        self.assertTrue('Content-Type' in self.auth.headers)
+
+        expected_payload_application_credential = {'auth': {
+            'identity': {'methods': ['application_credential'],
+                         'application_credential': {'id': 'm_app_id',
+                                                    'secret': 'm_app_secret'}}
+        }}
+
+        auth_args_application_credentials = ('http://url', None, None, None, None,
+                                             None, 'm_app_id', 'm_app_secret')
+
+        auth_application_credentials = AuthPassword(*auth_args_application_credentials)
+
+        for auth, expected_payload in ((self.auth, expected_payload_password),
+                                       (auth_application_credentials, expected_payload_application_credential)):
+            self.assertEqual(auth._auth_payload, expected_payload)
+            self.assertEqual(auth._auth_endpoint, 'http://url/auth/tokens')
+            self.assertTrue('Content-Type' in auth.headers)
 
     async def test_create_object_use_environ(self):
-        expected_payload = {'auth': {
+        expected_payload_password = {'auth': {
             'identity': {'methods': ['password'], 'password': {'user': {'domain': {'name': 'udm'}, 'name': 'uuu', 'password': 'ppp'}}},
             'scope': {'project': {'domain': {'name': 'udm'}, 'name': 'prj'}}
         }}
-        env = {
+        env_password = {
             'OS_AUTH_URL': 'https://keystone/v3',
             'OS_PASSWORD': 'ppp', 'OS_USERNAME': 'uuu',
             'OS_USER_DOMAIN_NAME': 'udm', 'OS_PROJECT_NAME': 'prj'
         }
 
-        with patch.dict('os.environ', env, clear=True):
-            auth = AuthPassword()
-        self.assertEqual(auth._auth_payload, expected_payload)
-        self.assertEqual(auth._auth_endpoint, 'https://keystone/v3/auth/tokens')
-        self.assertTrue('Content-Type' in auth.headers)
+        expected_payload_application_credentials = {'auth': {
+            'identity': {'methods': ['application_credential'],
+                         'application_credential': {'id': 'iid',
+                                                    'secret': 'ssecret'
+                                                    }
+                         }
+        }}
+        env_application_credentials = {
+            'OS_AUTH_URL': 'https://keystone/v3',
+            'OS_APPLICATION_CREDENTIAL_ID': 'iid',
+            'OS_APPLICATION_CREDENTIAL_SECRET': 'ssecret',
+            'OS_USER_DOMAIN_NAME': 'udm', 'OS_PROJECT_NAME': 'prj'
+        }
+
+        for env, expected_payload in ((env_password, expected_payload_password),
+                                      (env_application_credentials, expected_payload_application_credentials)):
+            with patch.dict('os.environ', env, clear=True):
+                auth = AuthPassword()
+            self.assertEqual(auth._auth_payload, expected_payload)
+            self.assertEqual(auth._auth_endpoint, 'https://keystone/v3/auth/tokens')
+            self.assertTrue('Content-Type' in auth.headers)
 
     async def test_get_token(self):
         body = {
@@ -107,7 +137,8 @@ async def test_authenticate_first_time(self):
             1100
         ]
 
-        patch('asyncopenstackclient.auth.AuthPassword.get_token', side_effect=mock_get_token_results).start()
+        get_token_mock = patch('asyncopenstackclient.auth.AuthPassword.get_token', new=MagicMock()).start()
+        get_token_mock.side_effect = mock_get_token_results
         patch('asyncopenstackclient.auth.time', side_effect=mock_time_results).start()
 
         # first time token should be None and get_token shall be called