Draft: PoC for testing and option

plugins/doc_fragments/auth.py

@@ -104,6 +104,9 @@ class ModuleDocFragment(object):
         description: For C(cert) auth, path to the private key file to authenticate with, in PEM format.
         type: path
         version_added: 1.4.0
+      revoke_ephemeral_token:
+        description: Foobar
+        type: bool
     '''
 
     PLUGINS = r'''

plugins/module_utils/_auth_method_approle.py

@@ -40,4 +40,4 @@ def authenticate(self, client, use_token=True):
         return response
 
     def should_revoke_token(self):
-        return True
+        return self._options.get_option("revoke_ephemeral_token")

plugins/module_utils/_authenticator.py

@@ -56,6 +56,7 @@ class HashiVaultAuthenticator():
         aws_iam_server_id=dict(type='str'),
         cert_auth_private_key=dict(type='path', no_log=False),
         cert_auth_public_key=dict(type='path'),
+        revoke_ephemeral_token=dict(type='bool', default=False),
     )
 
     def __init__(self, option_adapter, warning_callback, deprecate_callback):

tests/integration/targets/auth_approle/tasks/approle_test_revocation.yml

@@ -0,0 +1,57 @@
+- name: "Test block"
+  module_defaults:
+    community.hashi_vault.vault_read:
+      url: '{{ ansible_hashi_vault_url }}'
+      auth_method: 'approle'
+      mount_point: '{{ this_path }}'
+      role_id: '{{ role_id_cmd.result.data.role_id | default(omit) }}'
+      secret_id: "{{ secret_id_cmd.result.data.secret_id | default(omit) }}"
+  block:
+    - name: 'Fetch the RoleID of the AppRole'
+      vault_ci_read:
+        path: 'auth/{{ this_path }}/role/{{ approle_name }}/role-id'
+      register: role_id_cmd
+
+    - name: 'Get a SecretID issued against the AppRole'
+      vault_ci_write:
+        path: 'auth/{{ this_path }}/role/{{ approle_name }}/secret-id'
+        data: {}
+      register: secret_id_cmd
+
+    - name: Read token information using plugin
+      community.hashi_vault.vault_read:
+        path: "auth/token/lookup-self"
+        revoke_ephemeral_token: "{{ revoke_token }}"
+      register: token_info
+
+    - name: Check if token can still be used
+      community.hashi_vault.vault_read:
+        path: "auth/token/lookup-self"
+        token: '{{ token_info.data.data.id }}'
+        auth_method: token
+      ignore_errors: true
+      register: token_check
+
+    - assert:
+        fail_msg: "A token from vault_read was unexpectedly (un-)usable"
+        that:
+        - token_check is failed or not revoke_token
+        - token_check is not failed or revoke_token
+
+    - name: Read token information using lookup
+      set_fact:
+        token: "{{ lookup('community.hashi_vault.vault_read', 'auth/token/lookup-self', auth_method='approle', mount_point=this_path, url=ansible_hashi_vault_url, role_id=role_id_cmd.result.data.role_id, secret_id=secret_id_cmd.result.data.secret_id, revoke_ephemeral_token=revoke_token) }}"
+
+    - name: Check if token can still be used
+      community.hashi_vault.vault_read:
+        path: "auth/token/lookup-self"
+        token: '{{ token.data.id }}'
+        auth_method: token
+      ignore_errors: true
+      register: token_check
+
+    - assert:
+        fail_msg: "A token from vault_read was unexpectedly (un-)usable"
+        that:
+        - token_check is failed or not revoke_token
+        - token_check is not failed or revoke_token

tests/integration/targets/auth_approle/tasks/main.yml

@@ -50,3 +50,16 @@
           module_defaults:
             assert:
               quiet: yes
+
+    - name: Run approle ephemeral revocation tests
+      loop: '{{ [True, False] }}'
+      include_tasks:
+        file: approle_test_revocation.yml
+        apply:
+          vars:
+            this_path: '{{ ansible_hashi_vault_auth_method }}'
+            approle_name: '{{ secret_id_role }}'
+            revoke_token: '{{ item }}'
+          module_defaults:
+            assert:
+              quiet: yes

tests/integration/targets/auth_token/tasks/main.yml

@@ -32,3 +32,13 @@
   module_defaults:
     assert:
       quiet: yes
+
+- loop: '{{ [True, False] }}'
+  include_tasks:
+    file: token_test_revocation.yml
+    apply:
+      vars:
+        revoke_token: '{{ item }}'
+      module_defaults:
+        assert:
+          quiet: yes

tests/integration/targets/auth_token/tasks/token_test_revocation.yml

@@ -0,0 +1,36 @@
+- name: "Test block"
+  module_defaults:
+    community.hashi_vault.vault_read:
+      url: '{{ ansible_hashi_vault_url }}'
+      auth_method: token
+  vars:
+    user_token: '{{ user_token_cmd.result.auth.client_token }}'
+  block:
+    # the token auth method never has ephemeral tokens, so we expect all tokens
+    # to continue to be usable even if `revoke_ephemeral_token` is set to true.
+    - name: Read token information using plugin
+      community.hashi_vault.vault_read:
+        path: "auth/token/lookup-self"
+        token: "{{ user_token }}"
+        revoke_ephemeral_token: "{{ revoke_token }}"
+      register: token_info
+
+    - name: Check if token can still be used
+      community.hashi_vault.vault_read:
+        path: "auth/token/lookup-self"
+        # note: we cannot use token_info.data.data.id here, because that is
+        # identical to the `token` parameter, which is no log, so it is
+        # replaced with the verbatim string
+        # `VALUE_SPECIFIED_IN_NO_LOG_PARAMETER` (for better or worse).
+        token: '{{ user_token }}'
+        auth_method: token
+
+    - name: Read token information using lookup
+      set_fact:
+        _: "{{ lookup('community.hashi_vault.vault_read', 'auth/token/lookup-self', auth_method='token', token=user_token, url=ansible_hashi_vault_url, revoke_ephemeral_token=revoke_token) }}"
+
+    - name: Check if token can still be used
+      community.hashi_vault.vault_read:
+        path: "auth/token/lookup-self"
+        token: '{{ user_token }}'
+        auth_method: token

tests/unit/conftest.py

@@ -56,6 +56,7 @@ def authenticator():
     authenticator = HashiVaultAuthenticator
     authenticator.validate = mock.Mock(wraps=lambda: True)
     authenticator.authenticate = mock.Mock(wraps=lambda client: 'throwaway')
+    authenticator.logout = mock.Mock(warps=lambda: None)
 
     return authenticator
 

tests/unit/plugins/module_utils/authentication/conftest.py

@@ -32,6 +32,7 @@ def __init__(self, option_adapter, warning_callback, deprecate_callback):
 
     validate = mock.MagicMock()
     authenticate = mock.MagicMock()
+    should_revoke_token = mock.MagicMock()
 
 
 @pytest.fixture

tests/unit/plugins/module_utils/authentication/test_hashi_vault_authenticator.py

@@ -68,3 +68,26 @@ def test_get_method_object_implicit(self, authenticator, adapter, fake_auth_clas
         obj = authenticator._get_method_object()
 
         assert isinstance(obj, type(fake_auth_class))
+
+    @pytest.mark.parametrize('kwargs', [
+        {},
+        {'one': 1},
+        {'one': '1', 'two': 2},
+    ])
+    @pytest.mark.parametrize('revoke', [True, False])
+    def test_method_logout_logs_out_with_token_if_revocation_requested(self, authenticator, fake_auth_class, revoke, kwargs):
+        client = mock.MagicMock()
+        fake_auth_class.should_revoke_token.return_value = revoke
+
+        authenticator.logout(client, **kwargs)
+
+        fake_auth_class.should_revoke_token.assert_called_once_with(**kwargs)
+        client.logout.assert_called_once_with(revoke_token=revoke)
+
+    def test_logout_not_implemented(self, authenticator, fake_auth_class):
+        client = mock.MagicMock()
+
+        with pytest.raises(NotImplementedError):
+            authenticator.logout(client, method='missing')
+
+        fake_auth_class.should_revoke_token.assert_not_called()

tests/unit/plugins/modules/test_vault_write.py

@@ -176,3 +176,15 @@ def test_vault_write_vault_exception(self, vault_client, exc, capfd):
 
         assert e.value.code != 0, "result: %r" % (result,)
         assert re.search(exc[1], result['msg']) is not None
+
+    @pytest.mark.parametrize('opt_data', [{}, {'thing': 'one', 'thang': 'two'}])
+    @pytest.mark.parametrize('opt_wrap_ttl', [None, '5m'])
+    @pytest.mark.parametrize('patch_ansible_module', [[_combined_options(), 'data', 'wrap_ttl']], indirect=True)
+    def test_vault_write_logout(self, patch_ansible_module, approle_secret_id_write_response, vault_client, authenticator, opt_data, opt_wrap_ttl):
+        client = vault_client
+        client.write.return_value = approle_secret_id_write_response
+
+        with pytest.raises(SystemExit) as e:
+            vault_write.main()
+
+        authenticator.logout.assert_called_once_with(client)