Module: check_mk Branch: master Commit: c444ddc3c03372c12a91addaf0da1d7bcfbb303a URL: http://git.mathias-kettner.de/git/?p=check_mk.git;a=commit;h=c444ddc3c03372… Author: Andreas <ab(a)mathias-kettner.de> Date: Fri Jun 29 11:03:41 2018 +0200 webapi.py: moved some helper functions from plugins to htdocs/webapi.py Change-Id: I0f101ee98e1d48bf2bbf57839c58809cda254b74 --- web/htdocs/webapi.py | 106 +++++++++++++++++++++++++++++++++++++++++++ web/plugins/webapi/webapi.py | 104 ------------------------------------------ 2 files changed, 106 insertions(+), 104 deletions(-) diff --git a/web/htdocs/webapi.py b/web/htdocs/webapi.py index a920fc4..a31e6d9 100644 --- a/web/htdocs/webapi.py +++ b/web/htdocs/webapi.py @@ -47,6 +47,111 @@ api_actions = {} loaded_with_language = False +#. +# .--API Helpers-------------------------------------------------------------. +# | _ _ _ | +# | | | | | ___| |_ __ ___ _ __ ___ | +# | | |_| |/ _ \ | '_ \ / _ \ '__/ __| | +# | | _ | __/ | |_) | __/ | \__ \ | +# | |_| |_|\___|_| .__/ \___|_| |___/ | +# | |_| | +# +----------------------------------------------------------------------+ +# TODO: encapsulate in module/class + +def validate_request_keys(request, required_keys = None, optional_keys = None): + if required_keys: + missing = set(required_keys) - set(request.keys()) + if missing: + raise MKUserError(None, _("Missing required key(s): %s") % ", ".join(missing)) + + + all_keys = (required_keys or []) + (optional_keys or []) + for key in request.keys(): + if key not in all_keys: + raise MKUserError(None, _("Invalid key: %s") % key) + + +def check_hostname(hostname, should_exist = True): + # Validate hostname with valuespec + Hostname().validate_value(hostname, "hostname") + + if should_exist: + host = watolib.Host.host(hostname) + if not host: + raise MKUserError(None, _("No such host")) + else: + if watolib.Host.host_exists(hostname): + raise MKUserError(None, _("Host %s already exists in the folder %s") % (hostname, watolib.Host.host(hostname).folder().path())) + + +# Check if the given attribute name exists, no type check +def validate_general_host_attributes(host_attributes): + # inventory_failed and site are no "real" host_attributes (TODO: Clean this up!) + all_host_attribute_names = map(lambda (x, y): x.name(), watolib.all_host_attributes()) + ["inventory_failed", "site"] + for name, value in host_attributes.items(): + if name not in all_host_attribute_names: + raise MKUserError(None, _("Unknown attribute: %s") % html.attrencode(name)) + + # For real host attributes validate the values + try: + attr = watolib.host_attribute(name) + except KeyError: + attr = None + + if attr != None: + if attr.needs_validation("host"): + attr.validate_input(value, "") + + # The site attribute gets an extra check + if name == "site" and value not in config.allsites().keys(): + raise MKUserError(None, _("Unknown site %s") % html.attrencode(value)) + + +# Check if the tag group exists and the tag value is valid +def validate_host_tags(host_tags): + for key, value in host_tags.items(): + for group_entry in config.host_tag_groups(): + if group_entry[0] == key: + for value_entry in group_entry[2]: + if value_entry[0] == value: + break + else: + raise MKUserError(None, _("Unknown host tag %s") % html.attrencode(value)) + break + else: + raise MKUserError(None, _("Unknown host tag group %s") % html.attrencode(key)) + + +def validate_host_attributes(attributes): + validate_general_host_attributes(dict((key, value) for key, value in attributes.items() if not key.startswith("tag_"))) + validate_host_tags(dict((key[4:], value) for key, value in attributes.items() if key.startswith("tag_"))) + + +def compute_config_hash(entity): + import json, md5 + try: + entity_encoded = json.dumps(entity, sort_keys=True) + entity_hash = md5.md5(entity_encoded).hexdigest() + except Exception, e: + logger.error("Error %s" % e) + entity_hash = "0" + + return entity_hash + + +def validate_config_hash(hash_value, entity): + entity_hash = compute_config_hash(entity) + if hash_value != entity_hash: + raise MKUserError(None, _("The configuration has changed in the meantime. "\ + "You need to load the configuration and start another update. " + "If the existing configuration should not be checked, you can " + "remove the configuration_hash value from the request object.")) + + +def add_configuration_hash(response, configuration_object): + response["configuration_hash"] = compute_config_hash(configuration_object) + + class APICallCollection(object): @classmethod def all_classes(cls): @@ -61,6 +166,7 @@ class APICallCollection(object): raise NotImplementedError("This API collection does not register any API call") + def load_plugins(force): global loaded_with_language if loaded_with_language == current_language and not force: diff --git a/web/plugins/webapi/webapi.py b/web/plugins/webapi/webapi.py index 6df2128..7990fb1 100644 --- a/web/plugins/webapi/webapi.py +++ b/web/plugins/webapi/webapi.py @@ -26,110 +26,6 @@ # TODO CLEANUP: Replace MKUserError by MKAPIError or something like that -#. -# .--Helpers-------------------------------------------------------------. -# | _ _ _ | -# | | | | | ___| |_ __ ___ _ __ ___ | -# | | |_| |/ _ \ | '_ \ / _ \ '__/ __| | -# | | _ | __/ | |_) | __/ | \__ \ | -# | |_| |_|\___|_| .__/ \___|_| |___/ | -# | |_| | -# +----------------------------------------------------------------------+ - -def validate_request_keys(request, required_keys = None, optional_keys = None): - if required_keys: - missing = set(required_keys) - set(request.keys()) - if missing: - raise MKUserError(None, _("Missing required key(s): %s") % ", ".join(missing)) - - - all_keys = (required_keys or []) + (optional_keys or []) - for key in request.keys(): - if key not in all_keys: - raise MKUserError(None, _("Invalid key: %s") % key) - - -def check_hostname(hostname, should_exist = True): - # Validate hostname with valuespec - Hostname().validate_value(hostname, "hostname") - - if should_exist: - host = watolib.Host.host(hostname) - if not host: - raise MKUserError(None, _("No such host")) - else: - if watolib.Host.host_exists(hostname): - raise MKUserError(None, _("Host %s already exists in the folder %s") % (hostname, watolib.Host.host(hostname).folder().path())) - - -# Check if the given attribute name exists, no type check -def validate_general_host_attributes(host_attributes): - # inventory_failed and site are no "real" host_attributes (TODO: Clean this up!) - all_host_attribute_names = map(lambda (x, y): x.name(), watolib.all_host_attributes()) + ["inventory_failed", "site"] - for name, value in host_attributes.items(): - if name not in all_host_attribute_names: - raise MKUserError(None, _("Unknown attribute: %s") % html.attrencode(name)) - - # For real host attributes validate the values - try: - attr = watolib.host_attribute(name) - except KeyError: - attr = None - - if attr != None: - if attr.needs_validation("host"): - attr.validate_input(value, "") - - # The site attribute gets an extra check - if name == "site" and value not in config.allsites().keys(): - raise MKUserError(None, _("Unknown site %s") % html.attrencode(value)) - - -# Check if the tag group exists and the tag value is valid -def validate_host_tags(host_tags): - for key, value in host_tags.items(): - for group_entry in config.host_tag_groups(): - if group_entry[0] == key: - for value_entry in group_entry[2]: - if value_entry[0] == value: - break - else: - raise MKUserError(None, _("Unknown host tag %s") % html.attrencode(value)) - break - else: - raise MKUserError(None, _("Unknown host tag group %s") % html.attrencode(key)) - - -def validate_host_attributes(attributes): - validate_general_host_attributes(dict((key, value) for key, value in attributes.items() if not key.startswith("tag_"))) - validate_host_tags(dict((key[4:], value) for key, value in attributes.items() if key.startswith("tag_"))) - - -def compute_config_hash(entity): - import json, md5 - try: - entity_encoded = json.dumps(entity, sort_keys=True) - entity_hash = md5.md5(entity_encoded).hexdigest() - except Exception, e: - logger.error("Error %s" % e) - entity_hash = "0" - - return entity_hash - - -def validate_config_hash(hash_value, entity): - entity_hash = compute_config_hash(entity) - if hash_value != entity_hash: - raise MKUserError(None, _("The configuration has changed in the meantime. "\ - "You need to load the configuration and start another update. " - "If the existing configuration should not be checked, you can " - "remove the configuration_hash value from the request object.")) - - -def add_configuration_hash(response, configuration_object): - response["configuration_hash"] = compute_config_hash(configuration_object) - - #.