import re import yamale, validators from collections.abc import MutableMapping from deepmerge import merger from yamale.validators import DefaultValidators, Validator import yaml import copy import os from collections import Counter import tempfile import json from urllib.parse import urlparse from detect_secrets import SecretsCollection from detect_secrets.settings import transient_settings from deepmerge import Merger merger = Merger( [(dict, ["merge"])], ["override"], ["override"] ) deployable_section_format = "" invalid_patterns = { 'stg': ['int.meesho.int', 'prd.meesho.int'], 'int': ['stg.meesho.int'], 'prd': ['int.meesho.int', 'stg.meesho.int'] } valid_zk_patterns = { 'dev': ['dev.meesho.int:2181'], 'stg': ['dev.meesho.int:2181'], 'int': ['int.meesho.int:2181'], 'prd': ['config.prd.meesho.int:2181','mlp.prd.meesho.int:2181','comms.prd.meesho.int:2181'] } whitelists_for_preprod = ['consumer', 'scheduler', 'cron'] db_patterns = [ # PostgreSQL r'^postgresql://(?:\S+):(?:\S+)@(?:\S+):\d+/\S+$', # PostgreSQL (standard) r'^jdbc:postgresql://(?:\S+):(?:\S+)@(?:\S+):\d+/\S+$', # PostgreSQL (JDBC) # MySQL r'^mysql://(?:\S+):(?:\S+)@(?:\S+):\d+/\S+$', # MySQL (standard) r'^jdbc:mysql://*', # MySQL (JDBC) with optional query parameters # MongoDB r'^mongodb://(?:\S+):(?:\S+)@(?:\S+):\d+/\S+$', # MongoDB (standard) r'^mongodb\+srv://(?:\S+):(?:\S+)@(?:\S+)(?:/\S+)?$', # MongoDB (SRV) # SQLite r'^sqlite://(?:\S+)$', # SQLite # Redis r'^redis://(?:\S+):\d+$' # Redis (host:port) ] config_files_props = [ { 'name': 'application-prd.yml', 'type': 'static', 'schema-file-name': 'application-schema.yml', 'env': 'prd' }, { 'name': 'application-int.yml', 'type': 'static', 'schema-file-name': 'application-schema.yml', 'env': 'int' }, { 'name': 'application-stg.yml', 'type': 'static', 'schema-file-name': 'application-schema.yml', 'env': 'stg' }, { 'name': 'application-dev.yml', 'type': 'static', 'schema-file-name': 'application-schema.yml', 'env': 'dev' }, { 'name': 'application-dyn-prd.yml', 'type': 'dynamic', 'schema-file-name': 'application-schema.yml', 'env': 'prd' }, { 'name': 'application-dyn-int.yml', 'type': 'dynamic', 'schema-file-name': 'application-schema.yml', 'env': 'int' }, { 'name': 'application-dyn-stg.yml', 'type': 'dynamic', 'schema-file-name': 'application-schema.yml', 'env': 'stg' }, { 'name': 'application-dyn-dev.yml', 'type': 'dynamic', 'schema-file-name': 'application-schema.yml', 'env': 'dev' } ] class NonRequiredValidator(Validator): """ Wrapper around existing validators to make fields optional. """ def __init__(self, base_validator, *args, **kwargs): super().__init__(*args, **kwargs) self.base_validator = base_validator def is_valid(self, value): """ Allow missing values by returning True when value is None """ if value is None: return True return self.base_validator.is_valid(value) def to_python(self, value): return self.base_validator.to_python(value) class NumOrStringNum(Validator): tag = 'num' def _is_valid(self, value): if isinstance(value, (int, float)): return True if isinstance(value, str): try: float(value) # check if string can be parsed as number return True except ValueError: return False return False def _format(self, value): """Converts a valid string representation of a number into int or float.""" if isinstance(value, str): try: # convert to int if it's whole number, else float f = float(value) return int(f) if f.is_integer() else f except ValueError: return value return value class JsonStr(Validator): tag = 'json_str' # Custom tag to use in the schema def _is_valid(self, value): # Check if value is already a dict or list (already parsed JSON) if isinstance(value, (dict, list)): return True # Check if value is a string that can be parsed as JSON if isinstance(value, str): try: json.loads(value) return True except (json.JSONDecodeError, TypeError): return False return False def _format(self, value): """Converts a valid JSON string into a parsed JSON object.""" if isinstance(value, str): try: return json.loads(value) except (json.JSONDecodeError, TypeError): pass return value class IntOrStringInt(Validator): tag = 'int' # Custom tag to use in the schema def _is_valid(self, value): # Check if value is an integer if isinstance(value, int): return True # Check if value is a string that can be converted to an integer if isinstance(value, str): try: int(value) return True except ValueError: return False return False def _format(self, value): """Converts a valid string representation of an integer into an integer.""" if isinstance(value, str): try: return int(value) except ValueError: pass return value class BoolOrStringBool(Validator): tag = 'bool' def _is_valid(self, value): if isinstance(value, bool): return True if isinstance(value, str): value = value.lower() if value in ['true', 'false']: return True return False def _format(self, value): """Converts a valid string representation of a boolean into a boolean.""" if isinstance(value, str): value = value.lower() if value == 'true': return True elif value == 'false': return False return value class StringCustom(Validator): tag = 'str' def __init__(self, max=None, required=False): super().__init__(required=required) self.max_length = max self.required = required def _is_valid(self, value): errors = [] if self.required and value is None: errors.append(self.fail("Value is required but missing.")) if not isinstance(value, (str, int, bool)): errors.append(self.fail(f"Invalid type '{type(value).__name__}'. Expected str, int, or bool.")) if isinstance(value, str) and self.max_length is not None and len(value) > self.max_length: errors.append(self.fail(f"Value exceeds maximum length of {self.max_length} characters.")) return errors # Return list of errors instead of True/False def fail(self, message): """Override to define a custom fail message""" return f"Validation Error: {message}" def validate(self, value): """ Override validate to return a list of validation errors. """ errors = self._is_valid(value) # Get errors from _is_valid # Validate constraints (if any exist) for constraint in self._constraints_inst: error = constraint.is_valid(value) if error: if isinstance(error, list): errors.extend(error) else: errors.append(error) return errors # Return full list of validation errors def is_valid(self, value): return not self.validate(value) # Returns True if no errors def _format(self, value): if isinstance(value, int): value = str(value) if isinstance(value, str) and self.max_length is not None and len(value) > self.max_length: return value[:self.max_length] return value def is_db_url(key, value): """ Check if the given key-value pair is a valid database URL or name. Parameters: key_value_pair (tuple): A tuple containing the key and value to check. Returns: bool: True if the value is a valid database URL or name, False otherwise. """ # Regex patterns for different database URL formats db_patterns = [ r'^(postgresql://|mysql://|sqlite://|mongodb://|oracle://|mssql://|cockroachdb://|redis://|cassandra://|neo4j://|jdbc:mysql://).+', r'^[a-zA-Z0-9_]+$', # Just a database name (alphanumeric + underscores) r'^[a-zA-Z0-9_]+\.([a-zA-Z0-9_]+)$' # Database name with an optional prefix (e.g., schema) ] # Check if the value matches any of the patterns for pattern in db_patterns: if re.match(pattern, value): return True return False class Reader: """ This class provides methods to read the yml configuration file which has sections. File structure should be as follows: default_Data: --- spring: profiles: sections_1 sections_1_data --- spring: profiles: sections_2 sections_2_data --- spring: profiles: sections_3 sections_3_data """ def __init__(self, config_file_path: str): self.config_file_path = config_file_path def read_config_file(self): """ Read the config file and return the data in the form of dictionary. """ data = yamale.make_data(self.config_file_path) return data # Function to process each YAML file def read_without_sections(self): with open(self.config_file_path, 'r') as file: try: data = yaml.safe_load(file) return data except yaml.YAMLError as exc: print(f"Error reading {self.config_file_path}: {exc}", flush=True) return None class DictUtils: @staticmethod def sub_dict(keys: list[str], data: dict) -> dict: """ Function to extract the dictionary from the keys :param keys: keys to extract :param data: dictionary of data :return: dictionary with the keys """ return {key: data[key] for key in keys} @staticmethod def clean_up(data, keys): """ Delete the specified nested key and clean up empty parent keys. :param data: The dictionary to clean up. :param keys: A list of keys specifying the path to the nested key. """ if not keys: return current_key = keys[0] if len(keys) == 1: data.pop(current_key, None) else: next_level = data.get(current_key, None) if next_level is not None and isinstance(next_level, dict): DictUtils.clean_up(next_level, keys[1:]) if not next_level: data.pop(current_key, None) @staticmethod def remove_common_keys(dict_of_dicts, common_data): for key in dict_of_dicts: for common_key in common_data.keys(): if common_key in dict_of_dicts[key]: del dict_of_dicts[key][common_key] return dict_of_dicts @staticmethod def find_common_values(dict_of_dicts): dict_list = list(dict_of_dicts.values()) common_values = dict_list[0].copy() for d in dict_list[1:]: common_values = {k: v for k, v in common_values.items() if k in d and d[k] == v} return common_values @staticmethod def remove_keys(dictionary: dict, keys: list): for key in keys: if key in dictionary: del dictionary[key] @staticmethod def extract_dict_with_chosen_keys(dictionary: dict, keys: list): new_dict = dict() for key in keys: if key in dictionary: new_dict[key] = dictionary[key] return new_dict @staticmethod def extract_common_data_outside(nested_dict: dict): common_values = DictUtils.find_common_values(nested_dict) dict_list = DictUtils.remove_common_keys(nested_dict, common_values) # Assuming common_values and dict_list are defined merged_dict = common_values.copy() # Start with keys and values from common_values merged_dict.update(dict_list) # Adds keys and values from each dictionary in dict_list to merged_dict return merged_dict @staticmethod def extract_common_majority(dict_of_dicts): dict_keys = list(dict_of_dicts.keys()) if not dict_keys: return dict_of_dicts # Step 1: Identify common keys across all inner dictionaries common_keys = set(dict_of_dicts[dict_keys[0]].keys()) for key in dict_keys[1:]: common_keys.intersection_update(dict_of_dicts[key].keys()) common_dict = {} for key in common_keys: if key in ["spring~application~name", "spring~profiles", "spring~config~activate~on-profile","deployable-name"]: continue serialized_values = [json.dumps(dict_of_dicts[d][key], sort_keys=True) for d in dict_keys] most_common_serialized_value, count = Counter(serialized_values).most_common(1)[0] most_common_value = json.loads(most_common_serialized_value) common_dict[key] = most_common_value for d in set(dict_keys): if json.dumps(dict_of_dicts[d][key], sort_keys=True) == most_common_serialized_value: del dict_of_dicts[d][key] dict_of_dicts["common"] = common_dict return dict_of_dicts @staticmethod def extract_common(dict_of_dicts, common_data: dict): """ Extract common data from the dictionary of dictionaries and if some keys are there which are common in dict_of_dicts but not present in common_data then for them we will use the majority value :param dict_of_dicts: dict: deployable -> config_key -> config_value :param common_data: dict: config_key -> config_value :return: """ for key, value in common_data.items(): for deployable in dict_of_dicts.keys(): if key in dict_of_dicts[deployable] and dict_of_dicts[deployable][key] == value: dict_of_dicts[deployable].pop(key) dict_of_dicts = DictUtils.extract_common_majority(dict_of_dicts) dict_of_dicts["common"] = merger.merge(common_data, dict_of_dicts.get("common", {})) return dict_of_dicts class JsonFileUtils: @staticmethod def get_keys_by_line_numbers(file_path: str, line_numbers: list[int]) -> list[str]: keys = [] with open(file_path, 'r') as file: for i, line in enumerate(file, start=1): if i in line_numbers: # Extract the key from the line key = line.strip().split(':')[0].replace('"', '').strip() keys.append(key) return keys @staticmethod def write_dict_to_json_file(data: dict, file_path: str): with open(file_path, "w") as json_file: json.dump(data, json_file, indent=4) class SecretFinder: def __init__(self, file_path: str, is_exclude_keyword_detector: bool = False): self.file_path = file_path self.secrets = SecretsCollection() self.settings = { "plugins_used": [ { "name": "ArtifactoryDetector" }, { "name": "AWSKeyDetector" }, { "name": "AzureStorageKeyDetector" }, # { # "name": "Base64HighEntropyString", # "limit": 4.5 # }, { "name": "BasicAuthDetector" }, { "name": "CloudantDetector" }, { "name": "DiscordBotTokenDetector" }, { "name": "GitHubTokenDetector" }, { "name": "GitLabTokenDetector" }, # { # "name": "HexHighEntropyString", # "limit": 3.0 # }, { "name": "IbmCloudIamDetector" }, { "name": "IbmCosHmacDetector" }, { "name": "IPPublicDetector" }, { "name": "JwtTokenDetector" }, { "name": "MailchimpDetector" }, { "name": "NpmDetector" }, { "name": "OpenAIDetector" }, { "name": "PrivateKeyDetector" }, { "name": "PypiTokenDetector" }, { "name": "SendGridDetector" }, { "name": "SoftlayerDetector" }, { "name": "SquareOAuthDetector" }, { "name": "StripeDetector" }, { "name": "TelegramBotTokenDetector" }, { "name": "TwilioKeyDetector" } ], "filters_used": [ { "path": "detect_secrets.filters.allowlist.is_line_allowlisted" }, { "path": "detect_secrets.filters.common.is_ignored_due_to_verification_policies", "min_level": 2 }, { "path": "detect_secrets.filters.heuristic.is_indirect_reference" }, { "path": "detect_secrets.filters.heuristic.is_likely_id_string" }, { "path": "detect_secrets.filters.heuristic.is_lock_file" }, { "path": "detect_secrets.filters.heuristic.is_not_alphanumeric_string" }, { "path": "detect_secrets.filters.heuristic.is_potential_uuid" }, { "path": "detect_secrets.filters.heuristic.is_prefixed_with_dollar_sign" }, { "path": "detect_secrets.filters.heuristic.is_sequential_string" }, { "path": "detect_secrets.filters.heuristic.is_swagger_file" }, { "path": "detect_secrets.filters.heuristic.is_templated_secret" } ] } if not is_exclude_keyword_detector: self.settings["plugins_used"].append({ "name": "KeywordDetector", "keyword_exclude": "" } ) def fetch_secrets(self) -> dict: """ Function to check for secretes in the filex :return: line numbers of the secretes in the file """ with transient_settings(self.settings): self.secrets.scan_file(self.file_path) return self.secrets.json().get(self.file_path, {}) class SecretHelper: def __init__(self): pass @staticmethod def extract_secrets_positions(secrets: dict) -> list[int]: """ Function to extract the line numbers of the secrets :param secrets: secret find result from the secret finder :return: line numbers of the secretes in the file """ line_numbers = set() if len(secrets) > 0: for secret in secrets: line_numbers.add(secret['line_number']) return sorted(list(line_numbers)) @staticmethod def is_secret_present(file_path: str, is_exclude_keyword_detector: bool = False) -> bool: """ Function to check for secretes in the file :param is_exclude_keyword_detector: if we want to exclude keyword detector :param file_path: config file_path :return: error_message and status of the check """ secrets = SecretFinder(file_path, is_exclude_keyword_detector).fetch_secrets() if len(secrets) != 0: print(json.dumps(secrets, indent=2), flush=True) secrets_line_numbers = SecretHelper.extract_secrets_positions(secrets) return len(secrets_line_numbers) > 0 @staticmethod def fetch_secrets_keys_from_json(file_path: str) -> list[str]: """ Function to fetch the secret keys from the json file :return: list of secret keys """ secrets = SecretFinder(file_path).fetch_secrets() secrets_line_numbers = SecretHelper.extract_secrets_positions(secrets) secret_keys = JsonFileUtils.get_keys_by_line_numbers(file_path, secrets_line_numbers) return secret_keys @staticmethod def filter_secrets_from_dict(data: dict, secret_keys: list[str]) -> dict: """ Function to filter the secrets from the dictionary :param secret_keys: secret keys :param data: dictionary of data :return: dictionary without the secrets """ return DictUtils.sub_dict(secret_keys, data) @staticmethod def find_secrets_in_json_files(directory_path: str) -> dict: file_to_secrets_dict = dict() for filename in os.listdir(directory_path): if filename.endswith(".json"): file_path = os.path.join(directory_path, filename) secrets = SecretHelper.fetch_secrets_keys_from_json(file_path) file_to_secrets_dict[filename] = secrets return file_to_secrets_dict class ParsingHelper: @staticmethod def get_default_section(reader: Reader): return reader.read_config_file()[0][0] @staticmethod def get_section_data(reader: Reader, section_name: str) -> dict: data = reader.read_config_file() default_section = data[0][0] is_first = True for section in data: if is_first: is_first = False continue if (section[0].get('spring', {}).get('profiles', None) == section_name or section[0].get('spring', {}).get('config', {}).get("activate", {}).get("on-profile", None) == section_name): requested_section = default_section requested_section['spring']['profiles'] = section[0].get('spring', {}).get('profiles', None) requested_section['spring']['config']['activate']['on-profile'] = section[0].get('spring', {}).get( 'config', {}).get("activate", {}).get("on-profile", None) requested_section.update(section[0]) return requested_section if (section[0].get('deployable-name', None) == section_name): requested_section = default_section requested_section['deployable-name'] = section[0].get('deployable-name', None) return None @staticmethod def get_all_sections(reader: Reader, merge_on_default_section: bool = True, update_section_data: bool = True) -> dict: data = reader.read_config_file() default_section = data[0][0] sections = {"default": ParsingHelper.flatten_dict(default_section)} is_first = True for section in data: if is_first: is_first = False continue current_section_name = ParsingHelper.get_current_section_name(section) if update_section_data: ParsingHelper.remove_spring_section_header(section[0]) if current_section_name is not None: current_section_data = copy.deepcopy(default_section) if merge_on_default_section else {} current_section_data = merger.merge(ParsingHelper.flatten_dict(current_section_data), copy.deepcopy(section[0])) sections[current_section_name] = ParsingHelper.flatten_dict(current_section_data) return sections @staticmethod def remove_spring_section_header(section: dict, hack: bool = True): if section and section.get('spring', {}).get('profiles', None) is not None and type( section.get('spring', {}).get('profiles', None)) == str: del section['spring']['profiles'] if hack: # IMPORTANT: This is a hack to make the code work with the current schema global deployable_section_format deployable_section_format = "spring.profiles" if section and section.get('spring', {}).get('config', {}).get("activate", {}).get("on-profile", None) is not None: del section['spring']['config']['activate']['on-profile'] if section and "spring~profiles" in section: del section["spring~profiles"] if section and "spring~config~activate~on-profile" in section: del section["spring~config~activate~on-profile"] if section and section.get('deployable-name', None) is not None and type( section.get('deployable-name', None)) == str: del section['deployable-name'] if hack: # IMPORTANT: This is a hack to make the code work with the current schema deployable_section_format = "deployable-name" if section and section.get('deployable-name', None) is not None: del section['deployable-name'] if section and "deployable-name" in section: del section["deployable-name"] @staticmethod def get_current_section_name(section): if section and isinstance(section, tuple) and len(section) > 0: profiles = section[0].get('spring', {}).get('profiles', None) if isinstance(profiles, str): return profiles on_profile = section[0].get('spring', {}).get('config', {}).get("activate", {}).get("on-profile", None) if isinstance(on_profile, str): return on_profile if section and isinstance(section, tuple) and len(section) > 0: return section[0].get('deployable-name', None) return None @staticmethod def flatten_dict(d, parent_key='', sep='~'): items = [] for k, v in d.items(): new_key = f"{parent_key}{sep}{k}" if parent_key else k if isinstance(v, dict): items.extend(ParsingHelper.flatten_dict(v, new_key, sep=sep).items()) else: items.append((new_key, v)) return dict(items) @staticmethod def unflatten_dict(d, separator='~'): result_dict = {} for key, value in d.items(): parts = key.split(separator) last_d_ref = None d_ref = result_dict for part in parts[:-1]: if not isinstance(d_ref, dict): last_d_ref[".".join(parts[list(parts).index(part):])] = value break else: if part not in d_ref: d_ref[part] = {} last_d_ref = d_ref d_ref = d_ref[part] if not isinstance(d_ref, dict): last_d_ref[".".join(parts[-2:])] = value else: d_ref[parts[-1]] = value return result_dict @staticmethod def dump_properties_into_string(data: dict, priority_keys_order: list = ["spring"]): unflatten_data = ParsingHelper.unflatten_dict(data) dump_str = "" for config_key in priority_keys_order: spring_data = unflatten_data.get(config_key) if spring_data: dump_str = dump_str + yaml.dump({config_key: spring_data}, default_flow_style=False) del unflatten_data[config_key] if not unflatten_data: return dump_str return dump_str + yaml.dump(unflatten_data, default_flow_style=False) @staticmethod def write_to_application_file(env_deployable_dictionary: dict, config_data: dict, target_dir: str, prefix: str): # if prefix is there then adding a seperator as - if len(prefix) > 0: prefix = f"-{prefix}" for env_name in env_deployable_dictionary.keys(): if env_name not in config_data: continue env_specific_data = config_data[env_name].copy() os.makedirs(target_dir, exist_ok=True) with open(f'{target_dir}/application{prefix}-{env_name}.yml', 'w') as application_file: application_file.write(ParsingHelper.dump_properties_into_string(env_specific_data["common"])) for deployable_name in env_deployable_dictionary.get(env_name): if not config_data[env_name].get(deployable_name, {}): continue application_file.write(f"\n\n---\n") env_deployable_data = dict(config_data[env_name][deployable_name]) ParsingHelper.remove_spring_section_header(env_deployable_data, False) env_deployable_data[deployable_section_format] = deployable_name application_file.write( ParsingHelper.dump_properties_into_string(env_deployable_data)) @staticmethod def write_to_application_file_with_default_section(env_deployable_dictionary: dict, env_data: dict, prefix: str): for env_name in env_deployable_dictionary.keys(): with open(f'application{prefix}-{env_name}.yml', 'w') as application_file: default_section_data = env_data[env_name]["default"] application_file.write(ParsingHelper.dump_properties_into_string(default_section_data)) for deployable_name in env_deployable_dictionary.get(env_name): if not env_data[env_name].get(deployable_name, {}) or deployable_name == "default": continue application_file.write(f"\n\n---\n") env_deployable_data = dict(env_data[env_name][deployable_name]) env_deployable_data[deployable_section_format] = deployable_name application_file.write(ParsingHelper.dump_properties_into_string(env_deployable_data)) def is_url(value): """ Check if a value is a URL using validators library. :param value: Value to check. :return: True if the value is a URL, False otherwise. """ return validators.url(value) def read_yaml_data_with_sections(file_path: str, base_data: dict = {}): Filereader = Reader(file_path) file_data = ParsingHelper.get_all_sections(Filereader) # base data mai ek section hai file data mai many # base data dyn data # file data static data working fine in this # base data static data # file data dyn data copy_data = copy.deepcopy(base_data) for key in file_data.keys(): if key in base_data: base_data[key] = merger.merge(file_data[key], copy_data[key]) else: base_data[key] = merger.merge(file_data[key], copy_data["default"]) # if "default" in file_data: # del file_data["default"] for key in base_data.keys(): if key != "default": base_data[key] = flatten_dict(merger.merge(base_data[key], base_data["default"])) return base_data def write_schema_to_temp_file(schema_dict): flattened_schema = flatten_dict(normalise_keys(schema_dict)) if 'spring.profiles' in flattened_schema: del flattened_schema['spring.profiles'] if 'spring.config.activate.on-profile' in flattened_schema: del flattened_schema['spring.config.activate.on-profile'] if 'spring.application.name' in flattened_schema: del flattened_schema['spring.application.name'] if 'deployable-name' in flattened_schema: del flattened_schema['deployable-name'] with tempfile.NamedTemporaryFile(delete=False, mode='w') as temp_file: yaml.dump(flattened_schema, temp_file) return temp_file.name def normalise_keys(d, old_char='~', new_char='.'): """ Recursively replace old_char with new_char in the keys of the dictionary. :param d: The dictionary to process. :param old_char: The character to replace. :param new_char: The character to replace with. :return: A new dictionary with corrected keys. """ if not isinstance(d, dict): return d corrected_dict = {} for k, v in d.items(): new_key = str(k).replace(old_char, new_char) if isinstance(v, dict): corrected_dict[new_key] = normalise_keys(v, old_char, new_char) else: corrected_dict[new_key] = v return corrected_dict def validate_config_file_against_schema(file_path: str, schema_file: str, configs_directory_path: str, all_key_required: bool = True): """ Function to validate the config file against the schema It checks following things 1. check consistency of the config across all the files as schema will have the structure of the config_files 2. type error if any 3. rule against the value :param file_path: file path of the config file :param schema_file: file path of the schema file :param all_key_required: config file should have all the keys present in the schema file :return: True if validation succeeds, False otherwise """ temp_schema_path = os.path.join(configs_directory_path, schema_file) with open(temp_schema_path, 'r') as temp_file: schema_dict = yaml.safe_load(temp_file) temp_schema_path_to_use = write_schema_to_temp_file(schema_dict) try: _validators = DefaultValidators.copy() _validators['int'] = IntOrStringInt _validators['str'] = StringCustom _validators['bool'] = BoolOrStringBool _validators['json_str'] = JsonStr _validators['num'] = NumOrStringNum base_data = {} if not all_key_required: static_file_reader = Reader(file_path.replace("-dyn", "")) base_data = ParsingHelper.get_all_sections(static_file_reader) else: dyn_file_path = re.sub(r'(-[^-]+$)', r'-dyn\1', file_path) dyn_file_reader = Reader(dyn_file_path) base_data = ParsingHelper.get_all_sections(dyn_file_reader) schema = yamale.make_schema(temp_schema_path_to_use, validators=_validators) data = read_yaml_data_with_sections(file_path, base_data) temp_dict = copy.deepcopy(schema.dict) schema.dict = flatten_dict(temp_dict) final_result = True for section_name, section_data in data.items(): if section_name == "default": continue if section_data is None: print(f"Section data is empty for {file_path} against schema {schema_file} for section {section_name}", flush=True) continue flattened_data = flatten_dict(normalise_keys(section_data)) result = schema.validate(flattened_data, section_name, strict=False) if not result.isValid(): print( f"\033[91mValidation failed for environment {file_path.split('-')[-1].split('.')[0]} against schema {schema_file} for section {section_name}\033[0m", flush=True) print(result, flush=True) final_result = False return final_result except yamale.YamaleError as e: print("YAML validation failed:", flush=True) print(e.message, flush=True) return False def validate_no_cross_environment_endpoint(file_path: str, env: str) -> bool: """ Function to validate that the config file does not contain cross-environment references. :param file_path: Path to the config file. :param env: Environment name (e.g., 'prd', 'int', 'stg'). :return: True if no cross-environment references are found, False otherwise. """ status = True with open(file_path, 'r') as file: try: data = list(yaml.safe_load_all(file)) # Use safe_load_all for multiple documents for section_content in data: flattened_data = flatten_dict(section_content) for key, value in flattened_data.items(): # Check if any invalid patterns are found in the value for the environment invalid_patterns_for_env = invalid_patterns.get(env, []) if isinstance(value, str) and any( pattern in value for pattern in invalid_patterns_for_env) and not is_db_url(key, value): print(f"\033[91mInvalid value '{value}' found in {file_path} for environment {env}\033[0m", flush=True) status = False # Check for specific zookeeper connect string patterns if 'zookeeper.connect-string' in key or 'zookeeper~connect-string' in key or 'zookeeper.server' in key or 'zookeeper~server' in key: elements = value.split(',') valid_patterns = valid_zk_patterns.get(env, []) for element in elements: if not any(pattern in element for pattern in valid_patterns): print( f"\033[91mInvalid value '{value}' found in {file_path} for environment {env}\033[0m", flush=True) status = False return status except yaml.YAMLError as exc: print(f"Error reading {file_path}: {exc}", flush=True) return False def process_config_files_of_module(configs_directory_path: str): """ Function to process the config files of a module :param configs_directory_path: path where for the given modules files are residing :return: True if all check passed else false """ is_invalid = False validation_status = True module_name = os.path.basename(configs_directory_path) for config_file in config_files_props: config_file_path = os.path.join(configs_directory_path, config_file['name']) if not os.path.exists(config_file_path) and '-dev' in config_file['name']: continue if not os.path.exists(config_file_path): canSkip = False if '-int' in config_file['name']: for whitelists in whitelists_for_preprod: if whitelists in module_name: canSkip = True break if canSkip: continue print(f"Validation failed for {config_file_path}", flush=True) print(f"File {config_file_path} does not exist.", flush=True) validation_status = False continue if check_for_secretes(config_file_path): validation_status = False if config_file['type'] == 'static': if not validate_config_file_against_schema(config_file_path, config_file['schema-file-name'], configs_directory_path): is_invalid = True print(f"Validation failed for {config_file_path}", flush=True) if not validate_no_cross_environment_endpoint(config_file_path, config_file['env']): is_invalid = True else: if not validate_config_file_against_schema(config_file_path, config_file['schema-file-name'], configs_directory_path, False): is_invalid = True if is_invalid: validation_status = False return validation_status def check_for_secretes(file_path: str): """ Function to check for secretes in the file :param file_path: config file_path :return: error_message and status of the check """ if SecretHelper.is_secret_present(file_path, True): print("\033[91mFound secrets in the config file.\033[0m", flush=True) return True return False def find_non_matching_dicts(dict_list, dict_a): """ Find dictionaries in a list where specific keys do not match the given pattern. :param dict_list: List of tuples (dict, str). :param dict_a: Dictionary to flatten and search for URL values. :return: True if all keys match the pattern in the given dictionaries, False otherwise. """ # Step 1: Flatten dict A flattened_dict_a = flatten_dict(dict_a) # Step 2: Identify keys with URL values url_keys = [key for key, value in flattened_dict_a.items() if is_url(value)] # Step 3: Filter dicts in the list where keys do not match the given pattern all_keys_match = True for key in url_keys: key_match_found = False for d, s in dict_list: if key in d and matches_pattern(d[key]): key_match_found = True if not key_match_found: print(f"Key is not in proper format, name is : {key}", flush=True) all_keys_match = False return all_keys_match def flatten_dict(d, parent_key='', sep='.'): """ Flatten a nested dictionary. :param d: Dictionary to flatten. :param parent_key: String to use as a prefix for keys. :param sep: Separator for nested keys. :return: Flattened dictionary. """ items = [] for k, v in d.items(): new_key = f'{parent_key}{sep}{k}' if parent_key else k if isinstance(v, MutableMapping): items.extend(flatten_dict(v, new_key, sep=sep).items()) else: items.append((new_key, v)) return dict(items) def matches_pattern(value): """ Check if a value matches the pattern '{{stateless_service..}}'. :param value: Value to check. :return: True if the value matches the pattern, False otherwise. """ pattern = re.compile(r'\{\{(?:stateless_service|stateful_service)\.[^.]+\.[^}]+\}\}') return bool(pattern.match(value)) def clean_up(data, keys): """ Delete the specified nested key and clean up empty parent keys. :param data: The dictionary to clean up. :param keys: A list of keys specifying the path to the nested key. """ if not keys: return current_key = keys[0] if len(keys) == 1: data.pop(current_key, None) else: next_level = data.get(current_key, None) if next_level is not None and isinstance(next_level, dict): clean_up(next_level, keys[1:]) if not next_level: data.pop(current_key, None) def deep_validate(): configs_path = os.path.join(os.getenv("REPO_ROOT", ""), "configs") status = True for root, dirs, files in os.walk(configs_path): for dir_name in dirs: if not process_config_files_of_module(os.path.join(root, dir_name)): status = False return status if __name__ == "__main__": if deep_validate(): print("Validation successful.", flush=True) os._exit(0) else: print("Validation failed.", flush=True) os._exit(1)