Merge pull request #11787 from liortamari/hashicorp-vault-kubernetes-auth

Add Hashicorp Vault Kubernetes authentication method

awx/main/credential_plugins/hashivault.py

@@ -47,15 +47,24 @@ base_inputs = {
             'multiline': False,
             'help_text': _('Name of the namespace to use when authenticate and retrieve secrets'),
         },
+        {
+            'id': 'kubernetes_role',
+            'label': _('Kubernetes role'),
+            'type': 'string',
+            'multiline': False,
+            'help_text': _(
+                'The Role for Kubernetes Authentication.'
+                ' This is the named role, configured in Vault server, for AWX pod auth policies.'
+                ' see https://www.vaultproject.io/docs/auth/kubernetes#configuration'
+            ),
+        },
         {
             'id': 'default_auth_path',
-            'label': _('Path to Approle Auth'),
+            'label': _('Path to Auth'),
             'type': 'string',
             'multiline': False,
             'default': 'approle',
-            'help_text': _(
-                'The AppRole Authentication path to use if one isn\'t provided in the metadata when linking to an input field. Defaults to \'approle\''
-            ),
+            'help_text': _('The Authentication path to use if one isn\'t provided in the metadata when linking to an input field. Defaults to \'approle\''),
         },
     ],
     'metadata': [
@@ -151,16 +160,30 @@ def handle_auth(**kwargs):
     if kwargs.get('token'):
         token = kwargs['token']
     elif kwargs.get('role_id') and kwargs.get('secret_id'):
-        token = approle_auth(**kwargs)
+        token = method_auth(**kwargs, auth_param=approle_auth(**kwargs))
+    elif kwargs.get('kubernetes_role'):
+        token = method_auth(**kwargs, auth_param=kubernetes_auth(**kwargs))
     else:
-        raise Exception('Either token or AppRole parameters must be set')
+        raise Exception('Either token or AppRole/Kubernetes authentication parameters must be set')
 
     return token
 
 
 def approle_auth(**kwargs):
-    role_id = kwargs['role_id']
-    secret_id = kwargs['secret_id']
+    return {'role_id': kwargs['role_id'], 'secret_id': kwargs['secret_id']}
+
+
+def kubernetes_auth(**kwargs):
+    jwt_file = pathlib.Path('/var/run/secrets/kubernetes.io/serviceaccount/token')
+    with jwt_file.open('r') as jwt_fo:
+        jwt = jwt_fo.read().rstrip()
+    return {'role': kwargs['kubernetes_role'], 'jwt': jwt}
+
+
+def method_auth(**kwargs):
+    # get auth method specific params
+    request_kwargs = {'json': kwargs['auth_param'], 'timeout': 30}
+
     # we first try to use the 'auth_path' from the metadata
     # if not found we try to fetch the 'default_auth_path' from inputs
     auth_path = kwargs.get('auth_path') or kwargs['default_auth_path']
@@ -168,9 +191,6 @@ def approle_auth(**kwargs):
     url = urljoin(kwargs['url'], 'v1')
     cacert = kwargs.get('cacert', None)
 
-    request_kwargs = {'timeout': 30}
-    # AppRole Login
-    request_kwargs['json'] = {'role_id': role_id, 'secret_id': secret_id}
     sess = requests.Session()
     # Namespace support
     if kwargs.get('namespace'):

awx/main/tests/functional/test_credential_plugins.py

@@ -1,3 +1,8 @@
+import pytest
+from unittest import mock
+from awx.main.credential_plugins import hashivault
+
+
 def test_imported_azure_cloud_sdk_vars():
     from awx.main.credential_plugins import azure_kv
 
@@ -5,3 +10,69 @@ def test_imported_azure_cloud_sdk_vars():
     assert all([hasattr(c, 'name') for c in azure_kv.clouds])
     assert all([hasattr(c, 'suffixes') for c in azure_kv.clouds])
     assert all([hasattr(c.suffixes, 'keyvault_dns') for c in azure_kv.clouds])
+
+
+def test_hashivault_approle_auth():
+    kwargs = {
+        'role_id': 'the_role_id',
+        'secret_id': 'the_secret_id',
+    }
+    expected_res = {
+        'role_id': 'the_role_id',
+        'secret_id': 'the_secret_id',
+    }
+    res = hashivault.approle_auth(**kwargs)
+    assert res == expected_res
+
+
+def test_hashivault_kubernetes_auth():
+    kwargs = {
+        'kubernetes_role': 'the_kubernetes_role',
+    }
+    expected_res = {
+        'role': 'the_kubernetes_role',
+        'jwt': 'the_jwt',
+    }
+    with mock.patch('pathlib.Path') as path_mock:
+        mock.mock_open(path_mock.return_value.open, read_data='the_jwt')
+        res = hashivault.kubernetes_auth(**kwargs)
+        path_mock.assert_called_with('/var/run/secrets/kubernetes.io/serviceaccount/token')
+        assert res == expected_res
+
+
+def test_hashivault_handle_auth_token():
+    kwargs = {
+        'token': 'the_token',
+    }
+    token = hashivault.handle_auth(**kwargs)
+    assert token == kwargs['token']
+
+
+def test_hashivault_handle_auth_approle():
+    kwargs = {
+        'role_id': 'the_role_id',
+        'secret_id': 'the_secret_id',
+    }
+    with mock.patch.object(hashivault, 'method_auth') as method_mock:
+        method_mock.return_value = 'the_token'
+        token = hashivault.handle_auth(**kwargs)
+        method_mock.assert_called_with(**kwargs, auth_param=kwargs)
+        assert token == 'the_token'
+
+
+def test_hashivault_handle_auth_kubernetes():
+    kwargs = {
+        'kubernetes_role': 'the_kubernetes_role',
+    }
+    with mock.patch.object(hashivault, 'method_auth') as method_mock:
+        with mock.patch('pathlib.Path') as path_mock:
+            mock.mock_open(path_mock.return_value.open, read_data='the_jwt')
+            method_mock.return_value = 'the_token'
+            token = hashivault.handle_auth(**kwargs)
+            method_mock.assert_called_with(**kwargs, auth_param={'role': 'the_kubernetes_role', 'jwt': 'the_jwt'})
+            assert token == 'the_token'
+
+
+def test_hashivault_handle_auth_not_enough_args():
+    with pytest.raises(Exception):
+        hashivault.handle_auth()