Mukul Sharma
83cbf62ec6
Renames src/com/meesho -> src/com/homelab, resources/com/meesho -> resources/com/homelab, resources/org/meesho -> resources/org/homelab (via git mv, preserving history), and sweeps every remaining occurrence of "meesho" (any casing) out of package declarations, imports, libraryResource() paths, and comments across the whole repo. Also drops the per-user allowlist in vars/eksCICD.groovy, which hardcoded real former-colleagues' emails and doesn't apply to a single-person homelab — that branch is now permanently skipped rather than deleted outright, to avoid hand-editing the escape-sequence-heavy echo blocks it guards (eksCICD.groovy itself is unused legacy code, not called by homelabPipeline.groovy). Does not touch the ~114 files that were already missing from the working tree but still tracked in the prior commit — that's unrelated pre-existing state, left as-is.
1262 lines
44 KiB
Python
1262 lines
44 KiB
Python
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 tabulate import tabulate
|
|
|
|
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.homelab.int', 'prd.homelab.int'],
|
|
'int': ['stg.homelab.int'],
|
|
'prd': ['int.homelab.int', 'stg.homelab.int']
|
|
}
|
|
valid_zk_patterns = {
|
|
'dev': ['dev.homelab.int:2181', 'dev.homelab.int'],
|
|
'stg': ['dev.homelab.int:2181', 'dev.homelab.int'],
|
|
'int': ['int.homelab.int:2181', 'int.homelab.int'],
|
|
'prd': ['config.prd.homelab.int:2181','mlp.prd.homelab.int:2181','comms.prd.homelab.int:2181', 'config.prd.homelab.int','mlp.prd.homelab.int','comms.prd.homelab.int']
|
|
}
|
|
|
|
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 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 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 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 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": "BasicAuthDetector"
|
|
},
|
|
{
|
|
"name": "CloudantDetector"
|
|
},
|
|
{
|
|
"name": "DiscordBotTokenDetector"
|
|
},
|
|
{
|
|
"name": "GitHubTokenDetector"
|
|
},
|
|
{
|
|
"name": "GitLabTokenDetector"
|
|
},
|
|
{
|
|
"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 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))
|
|
|
|
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
|
|
"""
|
|
final_result= True
|
|
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)
|
|
error_rows = []
|
|
for section_name, section_data in data.items():
|
|
if section_name == "default":
|
|
continue
|
|
if section_data is None:
|
|
error_rows.append([
|
|
color_env(file_path.split('-')[-1].split('.')[0]), # Environment
|
|
section_name,
|
|
"",
|
|
f"Section data is empty for {file_path} against schema {schema_file} for section {section_name}"
|
|
])
|
|
continue
|
|
flattened_data = flatten_dict(normalise_keys(section_data))
|
|
result = schema.validate(flattened_data, section_name, strict=True)
|
|
if not result.isValid():
|
|
for err in result.errors:
|
|
# Try to extract the key and error message
|
|
# Example error: "spring.cloud.zookeeper.config.enabled: Unexpected element"
|
|
if ':' in err:
|
|
key, issue = err.split(':', 1)
|
|
else:
|
|
key, issue = "", err
|
|
error_rows.append([
|
|
color_env(file_path.split('-')[-1].split('.')[0]), # Environment
|
|
section_name,
|
|
key.strip(),
|
|
issue.strip()
|
|
])
|
|
final_result = False
|
|
|
|
if error_rows:
|
|
print("\n" + tabulate(
|
|
error_rows,
|
|
headers=["Environment", "Deployable Name", "Key", "Error/Issue"],
|
|
tablefmt="fancy_grid"
|
|
) + "\n", flush=True)
|
|
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
|
|
error_rows = []
|
|
|
|
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):
|
|
error_rows.append([
|
|
color_env(env),
|
|
key,
|
|
value,
|
|
"Invalid cross-environment value"
|
|
])
|
|
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):
|
|
error_rows.append([
|
|
color_env(env),
|
|
key,
|
|
element,
|
|
"Invalid zookeeper connect string"
|
|
])
|
|
status = False
|
|
|
|
if error_rows:
|
|
print("\n" + tabulate(
|
|
error_rows,
|
|
headers=["Environment", "Key", "Value", "Error/Issue"],
|
|
tablefmt="fancy_grid"
|
|
) + "\n", flush=True)
|
|
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
|
|
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.<service-name>.<some_value>}}'.
|
|
|
|
: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
|
|
if not os.path.exists(configs_path):
|
|
return 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
|
|
|
|
|
|
def color_env(env):
|
|
color_map = {
|
|
"dev": "\033[94m", # Blue
|
|
"stg": "\033[93m", # Yellow
|
|
"int": "\033[92m", # Green
|
|
"prd": "\033[91m", # Red
|
|
}
|
|
reset = "\033[0m"
|
|
return f"{color_map.get(env, '')}{env}{reset}"
|
|
|
|
|
|
if __name__ == "__main__":
|
|
if deep_validate():
|
|
print("Validation successful.", flush=True)
|
|
os._exit(0)
|
|
else:
|
|
print("Validation failed.", flush=True)
|
|
os._exit(1)
|