1
0
Fork 0
mirror of https://github.com/ansible-collections/community.general.git synced 2024-09-14 20:13:21 +02:00
community.general/plugins/module_utils/consul.py

322 lines
10 KiB
Python
Raw Normal View History

# -*- coding: utf-8 -*-
# Copyright (c) 2022, Håkon Lerring
# GNU General Public License v3.0+ (see COPYING 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
import copy
import json
from ansible.module_utils.six.moves.urllib import error as urllib_error
from ansible.module_utils.six.moves.urllib.parse import urlencode
from ansible.module_utils.urls import open_url
def get_consul_url(configuration):
return "%s://%s:%s/v1" % (
configuration.scheme,
configuration.host,
configuration.port,
)
def get_auth_headers(configuration):
if configuration.token is None:
return {}
else:
return {"X-Consul-Token": configuration.token}
class RequestError(Exception):
def __init__(self, status, response_data=None):
self.status = status
self.response_data = response_data
def __str__(self):
if self.response_data is None:
# self.status is already the message (backwards compat)
return self.status
return "HTTP %d: %s" % (self.status, self.response_data)
def handle_consul_response_error(response):
if 400 <= response.status_code < 600:
raise RequestError("%d %s" % (response.status_code, response.content))
AUTH_ARGUMENTS_SPEC = dict(
host=dict(default="localhost"),
port=dict(type="int", default=8500),
scheme=dict(default="http"),
validate_certs=dict(type="bool", default=True),
token=dict(no_log=True),
ca_path=dict(),
)
def camel_case_key(key):
parts = []
for part in key.split("_"):
if part in {"id", "ttl", "jwks", "jwt", "oidc", "iam", "sts"}:
parts.append(part.upper())
else:
parts.append(part.capitalize())
return "".join(parts)
STATE_PARAMETER = "state"
STATE_PRESENT = "present"
STATE_ABSENT = "absent"
OPERATION_READ = "read"
OPERATION_CREATE = "create"
OPERATION_UPDATE = "update"
OPERATION_DELETE = "remove"
def _normalize_params(params, arg_spec):
final_params = {}
for k, v in params.items():
if k not in arg_spec: # Alias
continue
spec = arg_spec[k]
if (
spec.get("type") == "list"
and spec.get("elements") == "dict"
and spec.get("options")
and v
):
v = [_normalize_params(d, spec["options"]) for d in v]
elif spec.get("type") == "dict" and spec.get("options") and v:
v = _normalize_params(v, spec["options"])
final_params[k] = v
return final_params
class _ConsulModule:
"""Base class for Consul modules.
This class is considered private, till the API is fully fleshed out.
As such backwards incompatible changes can occur even in bugfix releases.
"""
api_endpoint = None # type: str
unique_identifier = None # type: str
result_key = None # type: str
create_only_fields = set()
params = {}
def __init__(self, module):
self._module = module
self.params = _normalize_params(module.params, module.argument_spec)
self.api_params = {
k: camel_case_key(k)
for k in self.params
if k not in STATE_PARAMETER and k not in AUTH_ARGUMENTS_SPEC
}
def execute(self):
obj = self.read_object()
changed = False
diff = {}
if self.params[STATE_PARAMETER] == STATE_PRESENT:
obj_from_module = self.module_to_obj(obj is not None)
if obj is None:
operation = OPERATION_CREATE
new_obj = self.create_object(obj_from_module)
diff = {"before": {}, "after": new_obj}
changed = True
else:
operation = OPERATION_UPDATE
if self._needs_update(obj, obj_from_module):
new_obj = self.update_object(obj, obj_from_module)
diff = {"before": obj, "after": new_obj}
changed = True
else:
new_obj = obj
elif self.params[STATE_PARAMETER] == STATE_ABSENT:
operation = OPERATION_DELETE
if obj is not None:
self.delete_object(obj)
changed = True
diff = {"before": obj, "after": {}}
else:
diff = {"before": {}, "after": {}}
new_obj = None
else:
raise RuntimeError("Unknown state supplied.")
result = {"changed": changed}
if changed:
result["operation"] = operation
if self._module._diff:
result["diff"] = diff
if self.result_key:
result[self.result_key] = new_obj
self._module.exit_json(**result)
def module_to_obj(self, is_update):
obj = {}
for k, v in self.params.items():
result = self.map_param(k, v, is_update)
if result:
obj[result[0]] = result[1]
return obj
def map_param(self, k, v, is_update):
def helper(item):
return {camel_case_key(k): v for k, v in item.items()}
def needs_camel_case(k):
spec = self._module.argument_spec[k]
return (
spec.get("type") == "list"
and spec.get("elements") == "dict"
and spec.get("options")
) or (spec.get("type") == "dict" and spec.get("options"))
if k in self.api_params and v is not None:
if isinstance(v, dict) and needs_camel_case(k):
v = helper(v)
elif isinstance(v, (list, tuple)) and needs_camel_case(k):
v = [helper(i) for i in v]
if is_update and k in self.create_only_fields:
return
return camel_case_key(k), v
def _needs_update(self, api_obj, module_obj):
api_obj = copy.deepcopy(api_obj)
module_obj = copy.deepcopy(module_obj)
return self.needs_update(api_obj, module_obj)
def needs_update(self, api_obj, module_obj):
for k, v in module_obj.items():
if k not in api_obj:
return True
if api_obj[k] != v:
return True
return False
def prepare_object(self, existing, obj):
operational_attributes = {"CreateIndex", "CreateTime", "Hash", "ModifyIndex"}
existing = {
k: v for k, v in existing.items() if k not in operational_attributes
}
for k, v in obj.items():
existing[k] = v
return existing
def endpoint_url(self, operation, identifier=None):
if operation == OPERATION_CREATE:
return self.api_endpoint
elif identifier:
return "/".join([self.api_endpoint, identifier])
raise RuntimeError("invalid arguments passed")
def read_object(self):
url = self.endpoint_url(OPERATION_READ, self.params.get(self.unique_identifier))
try:
return self.get(url)
except RequestError as e:
if e.status == 404:
return
elif e.status == 403 and b"ACL not found" in e.response_data:
return
raise
def create_object(self, obj):
if self._module.check_mode:
return obj
else:
return self.put(self.api_endpoint, data=self.prepare_object({}, obj))
def update_object(self, existing, obj):
url = self.endpoint_url(
OPERATION_UPDATE, existing.get(camel_case_key(self.unique_identifier))
)
merged_object = self.prepare_object(existing, obj)
if self._module.check_mode:
return merged_object
else:
return self.put(url, data=merged_object)
def delete_object(self, obj):
if self._module.check_mode:
return {}
else:
url = self.endpoint_url(
OPERATION_DELETE, obj.get(camel_case_key(self.unique_identifier))
)
return self.delete(url)
def _request(self, method, url_parts, data=None, params=None):
module_params = self.params
if not isinstance(url_parts, (tuple, list)):
url_parts = [url_parts]
if params:
# Remove values that are None
params = {k: v for k, v in params.items() if v is not None}
ca_path = module_params.get("ca_path")
base_url = "%s://%s:%s/v1" % (
module_params["scheme"],
module_params["host"],
module_params["port"],
)
url = "/".join([base_url] + list(url_parts))
headers = {}
token = self.params.get("token")
if token:
headers["X-Consul-Token"] = token
try:
if data:
data = json.dumps(data)
headers["Content-Type"] = "application/json"
if params:
url = "%s?%s" % (url, urlencode(params))
response = open_url(
url,
method=method,
data=data,
headers=headers,
validate_certs=module_params["validate_certs"],
ca_path=ca_path,
)
response_data = response.read()
status = (
response.status if hasattr(response, "status") else response.getcode()
)
except urllib_error.URLError as e:
if isinstance(e, urllib_error.HTTPError):
status = e.code
response_data = e.fp.read()
else:
self._module.fail_json(
msg="Could not connect to consul agent at %s:%s, error was %s"
% (module_params["host"], module_params["port"], str(e))
)
raise
if 400 <= status < 600:
raise RequestError(status, response_data)
return json.loads(response_data)
def get(self, url_parts, **kwargs):
return self._request("GET", url_parts, **kwargs)
def put(self, url_parts, **kwargs):
return self._request("PUT", url_parts, **kwargs)
def delete(self, url_parts, **kwargs):
return self._request("DELETE", url_parts, **kwargs)