Source code for cowbird.config

import logging
import os
import re
from typing import Any, Dict, List, Literal, Tuple, Union, cast, overload

import yaml
from schema import And, Optional, Regex, Schema

from cowbird.typedefs import (
    ConfigDict,
    ConfigResTokenInfo,
    ConfigSegment,
    HandlerConfig,
    SyncPermissionConfig,
    SyncPointConfig
)
from cowbird.utils import get_logger, print_log, raise_log

[docs]LOGGER = get_logger(__name__)
[docs]MULTI_TOKEN = "**" # nosec: B105
[docs]BIDIRECTIONAL_ARROW = "<->"
[docs]RIGHT_ARROW = "->"
[docs]LEFT_ARROW = "<-"
[docs]PERMISSION_REGEX = r"[\w-]+"
# Either a single word, or a list of words in array
[docs]PERMISSIONS_REGEX = rf"({PERMISSION_REGEX}|\[\s*{PERMISSION_REGEX}(?:\s*,\s*{PERMISSION_REGEX})*\s*\])"
[docs]DIRECTION_REGEX = rf"({BIDIRECTIONAL_ARROW}|{LEFT_ARROW}|{RIGHT_ARROW})"
# Mapping format # <res_key1> : <permission(s)> <direction> <res_key2> : <permission(s)>
[docs]MAPPING_REGEX = r"(\w+)\s*:\s*" + PERMISSIONS_REGEX + r"\s*" + DIRECTION_REGEX + r"\s*(\w+)\s*:\s*" + PERMISSIONS_REGEX
[docs]NAMED_TOKEN_REGEX = r"^\{\s*(\w+)\s*\}$" # nosec: B105
[docs]class ConfigError(RuntimeError): """ Generic error during configuration loading. """
[docs]class ConfigErrorInvalidTokens(ConfigError): """ Config error specific to invalid SINGLE_TOKEN or MULTI_TOKEN tokens. """
[docs]class ConfigErrorInvalidServiceKey(ConfigError): """ Config error for invalid service keys. """
[docs]class ConfigErrorInvalidResourceKey(ConfigError): """ Config error for invalid resource keys. """
[docs]def _load_config(path_or_dict: Union[str, ConfigDict], section: str, allow_missing: bool = False) -> ConfigDict: """ Loads a file path or dictionary as YAML/JSON configuration. """ try: if isinstance(path_or_dict, str): with open(path_or_dict, "r", encoding="utf-8") as f: cfg = yaml.safe_load(f) else: cfg = path_or_dict return _expand_all(cfg[section]) except KeyError: msg = f"Config file section [{section!s}] not found." if allow_missing: print_log(msg, level=logging.WARNING, logger=LOGGER) return {} raise_log(msg, exception=ConfigError, logger=LOGGER) except Exception as exc: raise_log(f"Invalid config file [{exc!r}]", exception=ConfigError, logger=LOGGER)
@overload
[docs]def get_all_configs( # type: ignore[misc,unused-ignore] path_or_dict: Union[str, ConfigDict], section: Literal["handlers"], allow_missing: bool = False, ) -> List[Dict[str, HandlerConfig]]: ...
@overload def get_all_configs( # type: ignore[misc,unused-ignore] path_or_dict: Union[str, ConfigDict], section: Literal["sync_permissions"], allow_missing: bool = False, ) -> List[SyncPointConfig]: ... def get_all_configs( # type: ignore[misc,unused-ignore] path_or_dict: Union[str, ConfigDict], section: str, allow_missing: bool = False, ) -> List[Union[ConfigDict, Dict[str, HandlerConfig], SyncPointConfig]]: """ Loads all configuration files specified by the path (if a directory), a single configuration (if a file) or directly returns the specified dictionary section (if a configuration dictionary). :returns: - list of configurations loaded if input was a directory path - list of single configuration if input was a file path - list of single configuration if input was a JSON dict - empty list if none of the other cases where matched .. note:: Order of file loading will be resolved by alphabetically sorted filename if specifying a directory path. """ if isinstance(path_or_dict, str): if os.path.isdir(path_or_dict): dir_path = os.path.abspath(path_or_dict) known_extensions = [".cfg", ".yml", ".yaml", ".json"] cfg_names = list(sorted({fn for fn in os.listdir(dir_path) if any(fn.endswith(ext) for ext in known_extensions)})) return [_load_config(os.path.join(dir_path, fn), section, allow_missing) for fn in cfg_names] if os.path.isfile(path_or_dict): return [_load_config(path_or_dict, section, allow_missing)] elif isinstance(path_or_dict, dict): return [_load_config(path_or_dict, section, allow_missing)] return []
[docs]def _expand_all(config: ConfigDict) -> ConfigDict: """ Applies environment variable expansion recursively to all applicable fields of a configuration definition. """ if isinstance(config, dict): for cfg in list(config): cfg_key = os.path.expandvars(cfg) if cfg_key != cfg: config[cfg_key] = config.pop(cfg) config[cfg_key] = _expand_all(cast(ConfigDict, config[cfg_key])) elif isinstance(config, (list, set)): for i, cfg in enumerate(config): config[i] = _expand_all(cfg) elif isinstance(config, str): config = os.path.expandvars(str(config)) elif isinstance(config, (int, bool, float, type(None))): pass else: raise NotImplementedError(f"unknown parsing of config of type: {type(config)}") return config
[docs]def validate_handlers_config_schema(handlers_cfg: Dict[str, HandlerConfig]) -> None: """ Validates the schema of the `handlers` section found in the config. """ str_not_empty_validator = And(str, lambda s: len(s) > 0) schema = Schema({ str: { # Handler name # parameters common to all handlers Optional("active"): bool, Optional("priority"): int, Optional("url"): str, Optional("workspace_dir"): str, # parameters for specific handlers Optional("jupyterhub_user_data_dir"): str_not_empty_validator, Optional("wps_outputs_dir"): str_not_empty_validator, Optional("secure_data_proxy_name"): str_not_empty_validator, Optional("wps_outputs_res_name"): str_not_empty_validator, Optional("notebooks_dir_name"): str_not_empty_validator, Optional("public_workspace_wps_outputs_subpath"): str_not_empty_validator, Optional("user_wps_outputs_dir_name"): str_not_empty_validator, } }, ignore_extra_keys=True) schema.validate(handlers_cfg)
[docs]def validate_sync_perm_config_schema(sync_cfg: SyncPointConfig) -> None: """ Validates the schema of the `sync_permissions` section found in the config. """ schema = Schema({ Optional(str): { "services": { str: { # Service type, must correspond to an actual Magpie service type str: [ # Resource key, used to identify the resource here and in the permissions_mapping {"name": str, "type": str} ] } }, "permissions_mapping": [Regex(MAPPING_REGEX)] } }) schema.validate(sync_cfg)
[docs]def validate_and_get_resource_info(res_key: str, segments: List[ConfigSegment]) -> ConfigResTokenInfo: """ Validates a resource_key and its related info from the config and returns some resource info relevant to the config mapping validation. Returned info contains the following: - if the resource uses a MULTI_TOKEN in its resource_path - the list of named tokens found in the resource_path """ named_tokens = set() has_multi_token = False for seg in segments: if seg["name"] == MULTI_TOKEN: if has_multi_token: raise ConfigErrorInvalidTokens(f"Invalid config value for resource key {res_key}. Only one " f"`{MULTI_TOKEN}` token is permitted per resource.") has_multi_token = True else: matched_groups = re.match(NAMED_TOKEN_REGEX, seg["name"]) if matched_groups: # Save the first group as a named token, since there's only 1 matching group in the regex. if matched_groups.groups()[0] in named_tokens: raise ConfigErrorInvalidTokens(f"Invalid config value for resource key {res_key}. Named token " f"{matched_groups.groups()[0]} was found in multiple segments of " "the resource path. Each named token should only be used once in a " "resource path.") named_tokens.add(matched_groups.groups()[0]) return {"has_multi_token": has_multi_token, "named_tokens": named_tokens}
[docs]def validate_bidirectional_mapping(mapping: str, res_info: Dict[str, ConfigResTokenInfo], res_key1: str, res_key2: str) -> None: """ Validates if both resources of a bidirectional mapping respect validation rules. Both should either use MULTI_TOKEN or not use it and both should use exactly the same named tokens. """ if res_info[res_key1]["has_multi_token"] != res_info[res_key2]["has_multi_token"]: raise ConfigErrorInvalidTokens(f"Invalid permission mapping `{mapping}`. For a bidirectional mapping, " f"either all mapped resources should have `{MULTI_TOKEN}` " "or none should use them.") if res_info[res_key1]["named_tokens"] != res_info[res_key2]["named_tokens"]: raise ConfigErrorInvalidTokens(f"Invalid permission mapping `{mapping}`. For a bidirectional mapping, " "both resources should have exactly the same named_tokens. " f"({res_key1}: {res_info[res_key1]['named_tokens']}, " f"{res_key2}: {res_info[res_key2]['named_tokens']})")
[docs]def validate_unidirectional_mapping(mapping: str, src_info: ConfigResTokenInfo, tgt_info: ConfigResTokenInfo) -> None: """ Validates if both source and target resource of a unidirectional mapping respect validation rules. Source resource should use MULTI_TOKEN if target uses it, and source resource should include all named tokens found in the target resource. """ if not src_info["has_multi_token"] and tgt_info["has_multi_token"]: raise ConfigErrorInvalidTokens(f"Invalid permission mapping `{mapping}`. For a unidirectional mapping, " "the source resource should use a MULTI_TOKEN " "if the target is using one.") missing_named_tokens = tgt_info["named_tokens"] - src_info["named_tokens"] if missing_named_tokens: raise ConfigErrorInvalidTokens(f"Invalid permission mapping `{mapping}`. For a unidirectional mapping, " "all named tokens found in the target resource should also be found in " f"the source resource, but the tokens `{missing_named_tokens}` are " f"missing from the source.")
[docs]def get_mapping_info(mapping: str) -> Tuple[Union[str, Any], ...]: """ Obtain the different info found in a mapping string from the config. Returns the following matching groups : (res_key1, permission1, direction, res_key2, permission2) """ matched_groups = re.match(MAPPING_REGEX, mapping) if not matched_groups or len(matched_groups.groups()) != 5: raise ConfigError(f"Error parsing mapping `{mapping}`. " "Couldn't find all mapping info because of invalid format.") return matched_groups.groups()
[docs]def get_permissions_from_str(permissions: str) -> List[str]: """ Returns a tuple of all permissions found in a string. Used for permission strings found in the config, which can either be a single permission or a list of permissions. """ matched_groups = re.findall(PERMISSION_REGEX, permissions) if not matched_groups: raise ConfigError("Couldn't find permission, invalid format.") return matched_groups
[docs]def validate_sync_mapping_config(sync_cfg: SyncPermissionConfig, res_info: Dict[str, ConfigResTokenInfo]) -> None: """ Validates if mappings in the config have valid resource keys and use tokens properly. """ for mapping in sync_cfg["permissions_mapping"]: res_key1, _, direction, res_key2, _ = get_mapping_info(mapping) for res_key in [res_key1, res_key2]: if res_key not in res_info: raise ConfigErrorInvalidResourceKey(f"Invalid config mapping references resource {res_key} which is " "not defined in any service.") if direction == BIDIRECTIONAL_ARROW: validate_bidirectional_mapping(mapping, res_info, res_key1, res_key2) elif direction == RIGHT_ARROW: validate_unidirectional_mapping(mapping, src_info=res_info[res_key1], tgt_info=res_info[res_key2]) elif direction == LEFT_ARROW: validate_unidirectional_mapping(mapping, src_info=res_info[res_key2], tgt_info=res_info[res_key1]) else: raise ConfigError(f"Invalid direction `{direction}` found in the permissions_mapping.")
[docs]def validate_sync_config(sync_cfg: SyncPermissionConfig) -> None: # validate and get all resources info res_info = {} for resources in sync_cfg["services"].values(): for res_key in resources: if res_key in res_info: raise ConfigErrorInvalidResourceKey(f"Found duplicate resource key {res_key} in config. Config resource" " keys should be unique even between different services.") res_info[res_key] = validate_and_get_resource_info(res_key, resources[res_key]) validate_sync_mapping_config(sync_cfg, res_info)
[docs]def validate_sync_config_services(sync_cfg: SyncPermissionConfig, available_services: List[str]) -> None: """ Validates if all services used in the sync config are actual available services. All services should correspond to actual services available in Magpie. """ for svc in sync_cfg["services"]: if svc not in available_services: raise ConfigErrorInvalidServiceKey(f"Service `{svc}` used in sync config is not valid since it was not " f"found in Magpie services ({available_services}).")