ansible-collections/google.cloud
1
0
Fork 0
mirror of https://github.com/ansible-collections/google.cloud.git synced 2025-04-06 10:50:28 -07:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

Adding support for Google Secret Manager for issue 543

Browse source
This commit is contained in:
Dave Costakos 2023-06-21 16:24:14 -07:00
parent 30a4e66363
commit 76eb024c25
No known key found for this signature in database
GPG key ID: C4DC31A1B32AC45C
2 changed files with 626 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

213
plugins/lookup/gcp_secret_manager.py Normal file
View file

@ -0,0 +1,213 @@
# GNU General Public License v3.0+ (see LICENSES/GPL-3.0-or-later.txt or https://www.gnu.org/licenses/gpl-3.0.txt)
# SPDX-License-Identifier: GPL-3.0-or-later
from __future__ import (absolute_import, division, print_function)
__metaclass__ = type
DOCUMENTATION = '''
author:
- Dave Costakos <dcostako@redhat.com>
name: gcp_secret_manager
short_description: Get Secrets from Google Cloud as a Lookup plugin
description:
- retrieve secret keys in Secret Manager for use in playbooks
- see https://cloud.google.com/iam/docs/service-account-creds for details on creating
credentials for Google Cloud and the format of such credentials
- once a secret value is retreived, it is returned decoded. It is up to the developer
to maintain secrecy of this value once returned.
options:
key:
description:
- the key of the secret to look up in Secret Manager
type: str
required: True
project:
description:
- The name of the google cloud project
- defaults to OS env variable GCP_PROJECT if not present
type: str
auth_kind:
description:
- the type of authentication to use with Google Cloud (i.e. serviceaccount or machineaccount)
- defaults to OS env variable GCP_AUTH_KIND if not present
type: str
version:
description:
- the version name of your secret to retrieve
type: str
default: latest
required: False
service_account_email:
description:
- email associated with the service account
- defaults to OS env variable GCP_SERVICE_ACCOUNT_EMAIL if not present
type: str
required: False
service_account_file:
description:
- JSON Credential file obtained from Google Cloud
- defaults to OS env variable GCP_SERVICE_ACCOUNT_FILE if not present
- see https://cloud.google.com/iam/docs/service-account-creds for details
type: str
required: False
service_account_info:
description:
- JSON Object representing the contents of a service_account_file obtained from Google Cloud
- defaults to OS env variable GCP_SERVICE_ACCOUNT_INFO if not present
type: jsonarg
required: False
errors:
description:
- how to handle errors
choices: ['strict','warn','ignore']
default: strict
'''
EXAMPLES = '''
- name: Test secret using env variables for credentials
ansible.builtin.debug:
msg: "{{ lookup('google.cloud.gcp_secret_manager', key='secret_key') }}"
- name: Test secret using explicit credentials
ansible.builtin.debug:
msg: "{{ lookup('google.cloud.gcp_secret_manager', key='secret_key', project='project', auth_kind='serviceaccount', service_account_file='file.json') }}"
- name: Test getting specific version of a secret (old version)
ansible.builtin.debug:
msg: "{{ lookup('google.cloud.gcp_secret_manager', key='secret_key', version='1') }}"
- name: Test getting specific version of a secret (new version)
ansible.builtin.debug:
msg: "{{ lookup('google.cloud.gcp_secret_manager', key='secret_key', version='2') }}"
'''
RETURN = '''
_raw:
description: the contents of the secret requested (please use "no_log" to not expose this secret)
type: list
elements: str
'''
################################################################################
# Imports
################################################################################
import json
import os
import base64
from ansible.plugins.lookup import LookupBase
try:
import requests
HAS_REQUESTS = True
except ImportError:
HAS_REQUESTS = False
try:
import google.auth
from google.oauth2 import service_account
from google.auth.transport.requests import AuthorizedSession
HAS_GOOGLE_LIBRARIES = True
except ImportError:
HAS_GOOGLE_LIBRARIES = False
from ansible_collections.google.cloud.plugins.module_utils.gcp_utils import GcpSession, GcpRequest
from ansible.errors import AnsibleError
class GcpLookupException(Exception):
pass
class LookupModule(LookupBase):
def run(self, terms, variables, **kwargs):
self.set_options(var_options=variables, direct=kwargs)
self.scopes = ["https://www.googleapis.com/auth/cloud-platform"]
self._validate()
self.service_acct_creds = self._credentials()
session = AuthorizedSession(self.service_acct_creds)
response = session.get("https://secretmanager.googleapis.com/v1/projects/{project}/secrets/{key}/versions/{version}:access".format(**self.get_options()))
if response.status_code == 200:
result_data = response.json()
secret_value = base64.b64decode(result_data['payload']['data'])
return [ secret_value ]
else:
if self.get_option('errors') == 'warn':
self.warn(f"secret request returned bad status: {response.status_code} {response.json()}")
return [ '' ]
elif self.get_option('error') == 'ignore':
return [ '' ]
else:
raise AnsibleError(f"secret request returned bad status: {response.status_code} {response.json()}")
def _validate(self):
if HAS_GOOGLE_LIBRARIES == False:
raise AnsibleError("Please install the google-auth library")
if HAS_REQUESTS == False:
raise AnsibleError("Please install the requests library")
if self.get_option('key') == None:
raise AnsibleError("'key' is a required parameter")
if self.get_option('version') == None:
self.set_option('version', 'latest')
self._set_from_env('project', 'GCP_PROJECT', True)
self._set_from_env('auth_kind', 'GCP_AUTH_KIND', True)
self._set_from_env('service_account_email', 'GCP_SERVICE_ACCOUNT_EMAIL')
self._set_from_env('service_account_file', 'GCP_SERVICE_ACCOUNT_FILE')
self._set_from_env('service_account_info', 'GCP_SERVICE_ACCOUNT_INFO')
def _set_from_env(self, var=None, env_name=None, raise_on_empty=False):
if self.get_option(var) == None:
if env_name is not None and env_name in os.environ:
fallback = os.environ[env_name]
self.set_option(var, fallback)
if self.get_option(var) == None and raise_on_empty:
msg = f"No key '{var}' provided"
if env_name is not None:
msg += f" and no fallback to env['{env_name}'] available"
raise AnsibleError(msg)
def _credentials(self):
cred_type = self.get_option('auth_kind')
if cred_type == 'application':
credentials, project_id = google.auth.default(scopes=self.scopes)
return credentials
if cred_type == 'serviceaccount':
if self.get_option('service_account_file') is not None:
path = os.path.realpath(os.path.expanduser(self.get_option('service_account_file')))
try:
svc_acct_creds = service_account.Credentials.from_service_account_file(path)
except OSError as e:
raise GcpLookupException("Unable to read service_account_file at %s: %s" % (path, e.strerror))
elif self.get_option('service_account_contents') is not None:
try:
info = json.loads(self.get_option('service_account_contents'))
except json.decoder.JSONDecodeError as e:
raise GcpLookupException("Unable to decode service_account_contents as JSON: %s" % e)
svc_acct_creds = service_account.Credentials.from_service_account_info(info)
else:
raise GcpLookupException('Service Account authentication requires setting either service_account_file or service_account_contents')
return svc_acct_creds.with_scopes(self.scopes)
if cred_type == 'machineaccount':
self.svc_acct_creds = google.auth.compute_engine.Credentials(self.service_account_email)
return self.svc_acct_creds
raise GcpLookupException("Credential type '%s' not implemented" % cred_type)

413
plugins/modules/gcp_secret_manager.py Normal file
View file

@ -0,0 +1,413 @@
#!/usr/bin/env python
# GNU General Public License v3.0+ (see LICENSES/GPL-3.0-or-later.txt or https://www.gnu.org/licenses/gpl-3.0.txt)
# SPDX-License-Identifier: GPL-3.0-or-later
################################################################################
# Documentation
################################################################################
ANSIBLE_METADATA = {'metadata_version': '1.1', 'status': ["preview"], 'supported_by': 'community'}
DOCUMENTATION='''
---
module: gcp_secret_manager
description:
- Simple create/delete of secrets
- add or delete secret versions
- other features like etags, replication, annontation expected to be managed outside of Ansible
requirements:
- python >= 2.6
- requests >= 2.18.4
- google-auth >= 1.3.0
options:
options:
project:
description:
- The Google Cloud Platform project to use.
type: str
auth_kind:
description:
- The type of credential used.
type: str
required: true
choices:
- application
- machineaccount
- serviceaccount
service_account_contents:
description:
- The contents of a Service Account JSON file, either in a dictionary or as a
JSON string that represents it.
type: jsonarg
service_account_file:
description:
- The path of a Service Account JSON file if serviceaccount is selected as type.
type: path
service_account_email:
description:
- An optional service account email address if machineaccount is selected and
the user does not wish to use the default email.
type: str
scopes:
description:
- Array of scopes to be used
type: list
elements: str
name:
description:
- Name of the secret to be used
type: str
value:
description:
- The secret value that the secret should have
- this will be set upon create
- If the secret value is not this, a new version will be added with this value
type: str
state:
description:
- 'absent' or 'present': whether the secret should exist
type: str
return_value:
description:
- if true, the value of the secret will be returned unencrypted to Ansible
- if false, no value will be returned or decrypted
type: bool
default: true
version:
description:
- A version label to apply to the secret
- Default is "latest" which is the newest version of the secret
- "all" is also acceptable on delete (which will delete all versions of a secret)
type: str
default: 'latest'
'''
EXAMPLES='''
- name: Create a new secret
google.cloud.gcp_secret_manager:
name: secret_key
value: super_secret
state: present
auth_kind: serviceaccount
service_account_file: service_account_creds.json
- name: Ensure the secretexists, fail otherwise and return the value
google.cloud.gcp_secret_manager:
name: secret_key
state: present
- name: Ensure secret exists but don't return the value
google.cloud.gcp_secret_manager:
name: secret_key
state: present
return_value: false
- name: Add a new version of a secret
google.cloud.gcp_secret_manager:
name: secret_key
value: updated super secret
state: present
- name: Delete version 1 of a secret (but not the secret itself)
google.cloud.gcp_secret_manager:
name: secret_key
version: 1
state: absent
- name: Delete all versions of a secret
google.cloud.gcp_secret_manager:
name: secret_key
version: all
state: absent
- name: Get
'''
RETURN = '''
resources:
description: List of resources
returned: always
type: complex
name:
description:
- The name of the secret
returned: success
type: str
version:
description:
- the version number of the secret returned
returned: success
type: str
url:
description:
- the Google Cloud URL used to make the request
returned: success
type: str
status_code:
description:
- the HTTP status code of the response to Google Cloud
returned: success
type: str
msg:
description:
- A message indicating what was done (or not done)
returned: success, failure
type: str
value:
description:
- The decrypted secret value, please use care with this
returned: success
type: str
payload:
description:
- The base 64 secret payload including CRC for validation
retunred: success
type: dict
'''
################################################################################
# Imports
################################################################################
from ansible_collections.google.cloud.plugins.module_utils.gcp_utils import (
navigate_hash,
GcpSession,
GcpModule,
GcpRequest,
remove_nones_from_dict,
replace_resource_dict,
)
import json
# for decoding and validating secrets
import base64
import binascii
def get_auth(module):
return GcpSession(module, 'secret-manager')
def self_access_link(module):
return "https://secretmanager.googleapis.com/v1/projects/{project}/secrets/{name}/versions/{calc_version}:access".format(**module.params)
def self_get_link(module):
return "https://secretmanager.googleapis.com/v1/projects/{project}/secrets/{name}/versions/{calc_version}".format(**module.params)
def self_update_link(module):
return "https://secretmanager.googleapis.com/v1/projects/{project}/secrets/{name}/versions/{calc_version:version}".format(**module.params)
def self_list_link(module):
return "https://secretmanager.googleapis.com/v1/projects/{project}/secrets/{name}/versions?filter=state:ENABLED".format(**module.params)
def self_delete_link(module):
return "https://secretmanager.googleapis.com/v1/projects/{project}/secrets/{name}".format(**module.params)
def fetch_resource(module, allow_not_found=True):
auth = get_auth(module)
# set version to the latest version because
# we can't be sure that "latest" is always going
# to be set if secret versions get disabled
# see https://issuetracker.google.com/issues/286489671
if module.params['version'] == "latest" or module.params['version'] == 'all':
version_list = list_secret_versions(module)
latest_version = None
if version_list is None:
return None
if "versions" in version_list:
latest_version = sorted(version_list['versions'], key=lambda d: d['name'])[-1]['name'].split('/')[-1]
module.params['calc_version'] = latest_version
else:
# if this occurs, there are no available secret versions
# handle the corner case that we tried to delete
# a secret version that doesn't exist
if module.params['state'] == "absent":
return { "action": "delete_secret" }
link = self_access_link(module)
access_obj = return_if_object(module, auth.get(link), allow_not_found)
if access_obj is None:
return None
link = self_get_link(module)
get_obj = return_if_object(module, auth.get(link), allow_not_found)
if get_obj is None:
return None
return merge_dicts(get_obj, access_obj)
def merge_dicts(x, y):
z = x.copy()
z.update(y)
return z
def snake_to_camel(snake):
result = ''
capitalize_next = False
for char in snake:
if char == '_':
capitalize_next = True
else:
if capitalize_next:
result += char.upper()
capitalize_next = False
else:
result += char
return str(result)
# create secret is a create call + an add version call
def create_secret(module):
# build the payload
payload = { "replication": { "automatic": {} } }
url = "https://secretmanager.googleapis.com/v1/projects/{project}/secrets".format(**module.params)
auth = get_auth(module)
post_response = auth.post(url, body=payload, params={'secretId': module.params['name']})
return update_secret(module)
def update_secret(module):
# build the payload
b64_value = base64.b64encode(module.params['value'].encode("utf-8")).decode("utf-8")
payload = {
u'payload': {
u'data': b64_value
}
}
auth = get_auth(module)
url = "https://secretmanager.googleapis.com/v1/projects/{project}/secrets/{name}:addVersion".format(**module.params)
return return_if_object(module, auth.post(url, payload), False)
def list_secret_versions(module):
# filter by only enabled secrets
url = self_list_link(module)
auth = get_auth(module)
return return_if_object(module, auth.get(url), True)
# technically we're destroying the version
def delete_secret(module, destroy_all=False):
# delete secret does not take "latest" as a default version
# get the latest version if it doesn't exist in the request
version = module.params['version']
auth = get_auth(module)
if version.lower() == "all" or destroy_all:
url = self_delete_link(module)
return return_if_object(module, auth.delete(url))
else:
url = self_get_link(module) + ":destroy"
return return_if_object(module, auth.post(url, {}), False)
def return_if_object(module, response, allow_not_found=False):
# If not found, return nothing.
if allow_not_found and response.status_code == 404:
return None
if response.status_code == 409:
module.params['info'] = "exists already"
return None;
# probably a code error
if response.status_code == 400:
module.fail_json(msg=f"unexpected REST failure: {response.json()['error']}")
# If no content, return nothing.
if response.status_code == 204:
return None
try:
module.raise_for_status(response)
result = response.json()
result['url'] = response.request.url
result['status_code'] = response.status_code
if "name" in result:
result['version'] = result['name'].split("/")[-1]
result['name'] = result['name'].split("/")[3]
# base64 decode the value
if "payload" in result and "data" in result['payload']:
result['value'] = base64.b64decode(result['payload']['data']).decode("utf-8")
except getattr(json.decoder, 'JSONDecodeError', ValueError):
module.fail_json(msg="Invalid JSON response with error: %s" % response.text)
if navigate_hash(result, ['error', 'errors']):
module.fail_json(msg=navigate_hash(result, ['error', 'errors']))
return result
def main():
# limited support for parameters described in the "Secret" resource
# in order to simplify and deploy primary use cases
# expectation is customers needing to support additional capabilities
# in the SecretPayload will do so outside of Ansible.
# ref: https://cloud.google.com/secret-manager/docs/reference/rest/v1/projects.secrets#Secret
module = GcpModule(
argument_spec=dict(
state=dict(default='present', choices=['present', 'absent'], type='str'),
name=dict(required=True, type='str', aliases=['key', 'secret']),
value=dict(required=False, type='str'),
version=dict(required=False, type='str', default='latest'),
return_value=dict(required=False, type='bool', default=True)
)
)
if not module.params['scopes']:
module.params['scopes'] = ["https://www.googleapis.com/auth/cloud-platform"]
module.params['calc_version'] = module.params['version']
state = module.params['state']
fetch = fetch_resource(module, allow_not_found=True)
changed = False
# nothing came back, so the secret doesn't exist
if not fetch:
# doesn't exist, must create
if module.params.get('value') and state == 'present':
# create a new secret
fetch = create_secret(module)
changed = True
# specified present but no value
# fail, let the user know
# that no secret could be created without a value to encrypt
elif state == 'present':
module.fail_json(msg="secret '{name}' not present in '{project}' and no value for the secret is provided".format(**module.params)),
# secret is absent, success
else:
fetch = { "msg": "secret '{name}' in project '{project}' not present".format(**module.params)}
else:
# delete the secret version (latest if no version is specified)
if state == "absent":
# delete the secret
fetch = delete_secret(module, ("action" in fetch))
fetch['msg'] = "Secret Destroyed, it may take time to propagate"
changed = True
# check to see if the values are the same, and update if neede
if "value" in fetch and module.params.get('value') is not None:
# Update secret
if fetch['value'] != module.params['value']:
update = update_secret(module)
changed = True
else:
fetch['msg'] = "values identical, no need to update secret"
# pop value data if return_value == false
if module.params['return_value'] == False:
fetch.pop('value')
fetch.pop('payload')
if "msg" in fetch:
fetch['msg'] = f"{fetch['msg']} | not returning secret value since 'return_value is set to false"
else:
fetch['msg'] = "not returning secret value since 'return_value is set to false"
fetch['changed'] = changed
fetch['name'] = module.params['name']
module.exit_json(**fetch)
if __name__ == "__main__":
main()