path: root/ansible_collections/azure/azcollection/plugins/modules/azure_rm_keyvaultsecret.py

#!/usr/bin/python
# Copyright: Ansible Project
# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)

from __future__ import absolute_import, division, print_function

__metaclass__ = type


DOCUMENTATION = '''
---
module: azure_rm_keyvaultsecret
version_added: "0.1.2"
short_description: Use Azure KeyVault Secrets
description:
    - Create or delete a secret within a given keyvault.
    - By using Key Vault, you can encrypt keys and secrets.
    - Such as authentication keys, storage account keys, data encryption keys, .PFX files, and passwords.
options:
    keyvault_uri:
        description:
            - URI of the keyvault endpoint.
        required: true
        type: str
    content_type:
        description:
            - Type of the secret value such as a password.
        type: str
    secret_name:
        description:
            - Name of the keyvault secret.
        required: true
        type: str
    secret_value:
        description:
            - Secret to be secured by keyvault.
        type: str
    secret_expiry:
        description:
            - Optional expiry datetime for secret
        type: str
    secret_valid_from:
        description:
            - Optional valid-from datetime for secret
        type: str
    recover_if_need:
        description:
            - Whether to permanently recover delete secrets.
        type: bool
    purge_if_need:
        description:
            - Whether to permanently delete secrets.
        type: bool
    state:
        description:
            - Assert the state of the subnet. Use C(present) to create or update a secret and C(absent) to delete a secret .
        type: str
        default: present
        choices:
            - absent
            - present

extends_documentation_fragment:
    - azure.azcollection.azure
    - azure.azcollection.azure_tags

author:
    - Ian Philpot (@iphilpot)

'''

EXAMPLES = '''
- name: Create a secret
  azure_rm_keyvaultsecret:
    secret_name: MySecret
    secret_value: My_Pass_Sec
    keyvault_uri: https://contoso.vault.azure.net/
    tags:
      testing: testing
      delete: never

- name: Delete a secret
  azure_rm_keyvaultsecret:
    secret_name: MySecret
    keyvault_uri: https://contoso.vault.azure.net/
    state: absent

- name: Recover a delete secret
  azure_rm_keyvaultsecret:
    secret_name: MySecret
    keyvault_uri: https://contoso.vault.azure.net/
    recover_if_need: true

- name: Purge a delete secret
  azure_rm_keyvaultsecret:
    secret_name: MySecret
    keyvault_uri: https://contoso.vault.azure.net/
    purge_if_need: true
'''

RETURN = '''
state:
    description:
        - Current state of the secret.
    returned: success
    type: complex
    contains:
        secret_id:
          description:
              - Secret resource path.
          type: str
          example: https://contoso.vault.azure.net/secrets/hello/e924f053839f4431b35bc54393f98423
'''

from ansible_collections.azure.azcollection.plugins.module_utils.azure_rm_common import AzureRMModuleBase

try:
    from azure.keyvault.secrets import SecretClient
    from azure.core.exceptions import ResourceNotFoundError
    from azure.core.exceptions import HttpResponseError
    import dateutil.parser
except ImportError:
    # This is handled in azure_rm_common
    pass


class AzureRMKeyVaultSecret(AzureRMModuleBase):
    ''' Module that creates or deletes secrets in Azure KeyVault '''

    def __init__(self):

        self.module_arg_spec = dict(
            secret_name=dict(type='str', required=True),
            secret_value=dict(type='str', no_log=True),
            secret_valid_from=dict(type='str', no_log=True),
            secret_expiry=dict(type='str', no_log=True),
            keyvault_uri=dict(type='str', no_log=True, required=True),
            state=dict(type='str', default='present', choices=['present', 'absent']),
            recover_if_need=dict(type='bool'),
            purge_if_need=dict(type='bool'),
            content_type=dict(type='str')
        )

        required_if = [
            ('state', 'present', ['secret_value'])
        ]

        self.results = dict(
            changed=False,
            state=dict()
        )

        self.secret_name = None
        self.secret_value = None
        self.secret_valid_from = None
        self.secret_expiry = None
        self.keyvault_uri = None
        self.state = None
        self.data_creds = None
        self.client = None
        self.tags = None
        self.recover_if_need = None
        self.purge_if_need = None
        self.content_type = None

        super(AzureRMKeyVaultSecret, self).__init__(self.module_arg_spec,
                                                    supports_check_mode=True,
                                                    required_if=required_if,
                                                    supports_tags=True)

    def exec_module(self, **kwargs):

        for key in list(self.module_arg_spec.keys()) + ['tags']:
            setattr(self, key, kwargs[key])

        # Create KeyVault Client
        self.client = self.get_keyvault_client()

        results = dict()
        changed = False

        try:
            results = self.get_secret(self.secret_name)

            # Secret exists and will be deleted
            if self.state == 'absent':
                changed = True
            elif self.secret_value and results['secret_value'] != self.secret_value:
                changed = True

        except ResourceNotFoundError as ec:
            # Secret doesn't exist
            if self.state == 'present':
                changed = True
        except Exception as ec2:
            self.fail("Find the key vault secret got exception, exception as {0}".format(str(ec2)))

        self.results['changed'] = changed
        self.results['state'] = results

        valid_from = self.secret_valid_from
        if isinstance(valid_from, str) and len(valid_from) > 0:
            valid_from = dateutil.parser.parse(valid_from)

        expiry = self.secret_expiry
        if isinstance(expiry, str) and len(expiry) > 0:
            expiry = dateutil.parser.parse(expiry)

        if not self.check_mode:
            # Create secret
            if self.state == 'present' and changed:
                if self.get_delete_secret(self.secret_name):
                    if self.recover_if_need:
                        results['secret_id'] = self.recover_delete_secret(self.secret_name)
                        status = 'Recover'
                    elif self.purge_if_need:
                        self.purge_deleted_secret(self.secret_name)
                        status = 'Purged'
                    else:
                        self.fail("Secret {0} is currently in a deleted but recoverable state, and its name cannot be reused; in this state,\
                                  the secret can only be recovered or purged.".format(self.secret_name))
                else:
                    results['secret_id'] = self.create_update_secret(self.secret_name, self.secret_value, self.tags, self.content_type, valid_from, expiry)
                    status = 'Created'
                self.results['state'] = results
                self.results['state']['status'] = status
            # Delete secret
            elif self.state == 'absent' and changed:
                results['secret_id'] = self.delete_secret(self.secret_name)
                self.results['state'] = results
                self.results['state']['status'] = 'Deleted'
        else:
            if self.state == 'present' and changed:
                self.results['state']['status'] = 'Created'
            elif self.state == 'absent' and changed:
                self.results['state']['status'] = 'Deleted'

        return self.results

    def get_keyvault_client(self):

        return SecretClient(vault_url=self.keyvault_uri, credential=self.azure_auth.azure_credential_track2)

    def get_secret(self, name, version=''):
        ''' Gets an existing secret '''
        secret_bundle = self.client.get_secret(name=name, version=version)

        if secret_bundle:
            return dict(secret_id=secret_bundle.id, secret_value=secret_bundle.value)
        return None

    def create_update_secret(self, name, secret, tags, content_type, valid_from, expiry):
        ''' Creates/Updates a secret '''
        secret_bundle = self.client.set_secret(name=name,
                                               value=secret,
                                               tags=tags,
                                               content_type=content_type,
                                               expires_on=expiry,
                                               not_before=valid_from)
        return secret_bundle._properties._id

    def delete_secret(self, name):
        ''' Deletes a secret '''
        deleted_secret = self.client.begin_delete_secret(name)
        result = self.get_poller_result(deleted_secret)
        return result.properties._id

    def recover_delete_secret(self, name):
        ''' Recover a delete secret '''
        try:
            recover_delete_secret = self.client.begin_recover_deleted_secret(name)
            result = self.get_poller_result(recover_delete_secret)
            return result._id
        except HttpResponseError as ec:
            self.fail("Recover the delete secret fail, detail info {0}".format(ec))

    def purge_deleted_secret(self, name):
        ''' Purge delete secret '''
        try:
            purge_deleted_secret = self.client.purge_deleted_secret(name)
            return purge_deleted_secret
        except HttpResponseError as ec:
            self.fail("Purge delete secret fail, detail info {0}".format(ec))

    def get_delete_secret(self, name):
        ''' Get delete secret '''
        try:
            self.client.get_deleted_secret(name=name)
        except ResourceNotFoundError:
            return False
        return True


def main():
    AzureRMKeyVaultSecret()


if __name__ == '__main__':
    main()