import os import json from collections import OrderedDict from aws.s3_apis.s3 import upload_file_to_s3, check_and_create_folder from aws.env_info import get_env_bucket from actions_logging.app_logging import logger from github.env import get_required_env_var, exit_on_error_and_write_summary from aws.constants import PROD_ENVS, STG_ENVS, PRODUCTION, STAGING , ENVS from env_files.constants import VERSION_HISTORY_FILE def parse_env_var_file(env_name, global_file=False, sx_extend="", service_name=""): if env_name in PROD_ENVS: dir_name = "production" elif env_name in STG_ENVS: dir_name = "staging" else: dir_name = "dev" secrets = {} vars = {} file_path_prefix = ".github/env_files" if service_name == "common": file_path_prefix = "common_env_files" file_path = f"{file_path_prefix}/{sx_extend}{dir_name}/{env_name}.json" if os.path.exists(file_path): logger.info(f"file {file_path} exists!") else: logger.info(f"file {file_path} does not exist!, taking template.json") env_name = "template" file_path = f"{file_path_prefix}/{sx_extend}{dir_name}/{env_name}.json" if not os.path.exists(file_path): logger.info(f"No file path {file_path} was found! continue only with global.json") return secrets, vars if global_file: file_path = f"{file_path_prefix}/{sx_extend}global.json" with open(file_path, 'r') as env_file: json_data = json.load(env_file) for key in json_data: if json_data[key].startswith('SECRET_'): secret_name = json_data[key][7:] secrets[key] = secret_name else: vars[key] = json_data[key] return secrets, vars def merge_dicts(dict1, dict2): merged_dict = dict1.copy() merged_dict.update(dict2) return merged_dict def handle_secrets(env_name, env, service_name): # save all the pulled secrets to a dict secrets_dict = {} upper_env_name = env_name.upper().replace("-", "_") upper_service_name = service_name.upper().replace("-", "_") global_cicd_prefix = f"GLOBAL_CICD_" global_keys_prefix = f"GLOBAL_KEYS_" global_certificates_prefix = f"GLOBAL_CERTIFICATES_" global_svc_prefix = f"GLOBAL_{upper_service_name}_" env_svc_prefix = f"ENV_{upper_env_name}_{upper_service_name}_" env_prefix = f"ENV_{upper_env_name}_" sorted_env = dict(sorted(env.items(), key=lambda x: x[0], reverse=True)) for key in sorted_env: if key.startswith(global_cicd_prefix): real_key = key.replace(global_cicd_prefix, "") secrets_dict[real_key] = env[key] elif key.startswith(global_keys_prefix): real_key = key.replace(global_keys_prefix, "") secrets_dict[real_key] = env[key] elif key.startswith(global_certificates_prefix): real_key = key.replace(global_certificates_prefix, "") secrets_dict[real_key] = env[key] elif key.startswith(global_svc_prefix): real_key = key.replace(global_svc_prefix, "") secrets_dict[real_key] = env[key] elif key.startswith(env_svc_prefix): real_key = key.replace(env_svc_prefix, "") secrets_dict[real_key] = env[key] elif key.startswith(env_prefix): if key.startswith(env_svc_prefix): real_key = key.replace(env_svc_prefix, "") secrets_dict[real_key] = env[key] else: real_key = key.replace(env_prefix, "") secrets_dict[real_key] = env[key] return secrets_dict def iterate_secrets_and_env_vars(final_vars, secrets_dict, secret_names): for secret_name in secret_names: secret_name_value = secret_names[secret_name] if secrets_dict.get(secret_name_value): final_vars[secret_name] = secrets_dict[secret_name_value] return final_vars def get_app_vars(env): prefix = "APP_" app_vars = {} for key in env: if key.startswith(prefix): origin_key = key new_key = key.replace(prefix, "") app_vars[new_key] = env[origin_key] return app_vars def remove_app_prefix(vars_dict): new_vars_dict = {} for key in vars_dict: if key.startswith("APP_"): new_var = key.replace("APP_", "", 1) else: new_var = key new_vars_dict[new_var] = vars_dict[key] return new_vars_dict def create_env_file(final_vars, file_name): vars = OrderedDict(sorted(final_vars.items())) with open(file_name, "w") as base_file: for var in vars: base_file.write(f"{var}={vars[var]}\n") def remove_env_file(file_name): os.remove(file_name) def main(version_to_use=None): try: env = os.environ env_name = get_required_env_var("ENV_NAME") env_files_path = os.getenv("ENV_FILES_PATH", ".github/env_files") service_name = get_required_env_var("SVC_NAME") version_history_path = f"{env_files_path}/version_history" version_history_file = VERSION_HISTORY_FILE if env_name in PROD_ENVS: sub_folder = PRODUCTION elif env_name in STG_ENVS: sub_folder = STAGING else: sub_folder = "dev" sub_folder += f"/{env_name}" version_history_file_path = os.path.join(version_history_path, sub_folder, VERSION_HISTORY_FILE) secrets_dict = handle_secrets(env_name, env, service_name) double_service = False svc_name = os.getenv('SVC_NAME') paths_extends = [""] if svc_name == "saferx-backend": paths_extends = ["saferx-backend/", "saferx-backend-cli/"] double_service = True elif svc_name == "sx-core-dns": paths_extends = ["sx-core-dns-private/", "sx-core-dns-public/"] double_service = True for path in paths_extends: if double_service: service_name = path[:-1] file_name = f"{service_name}.env" secret_names_from_var_file, vars_dict_from_var_file = parse_env_var_file(env_name, sx_extend=path, service_name=service_name) secret_names_from_global_file, vars_from_all_envs_files = parse_env_var_file(env_name, global_file=True, sx_extend=path, service_name=service_name) secret_names = merge_dicts(secret_names_from_global_file, secret_names_from_var_file) file_name = f"{service_name}.env" vars_from_app_env = get_app_vars(env) # logger.info(f"merged dicts global file and secret names from env: {secret_names}") vars_dict = merge_dicts(vars_from_all_envs_files, vars_dict_from_var_file) # logger.info(f"merged dicts all env vars and env vars: {vars_dict}") vars_dict_with_app = merge_dicts(vars_dict, vars_from_app_env) vars_dict_with_app = remove_app_prefix(vars_dict_with_app) # logger.info(f"merged dicts all env vars and env vars: {vars_dict_with_app}") final_vars = {} final_vars = iterate_secrets_and_env_vars(final_vars, secrets_dict, secret_names) final_vars = remove_app_prefix(final_vars) # logger.info(f"final var1: {final_vars}") final_vars = merge_dicts(final_vars, vars_dict_with_app) # logger.info(f"final var: {final_vars}") create_env_file(final_vars, file_name) bucket_name = get_env_bucket(env_name) upload_file_to_s3(bucket_name, file_name) version = os.getenv("ENV_FILE_VERSION", version_to_use) if version: version = version.split("-")[0] check_and_create_folder(bucket_name, f"{service_name}/", create_if_not_exists=True) check_and_create_folder(bucket_name, f"{service_name}/{version}/", create_if_not_exists=True) upload_file_to_s3(bucket_name, file_name, f"{service_name}/{version}/{file_name}") upload_file_to_s3(bucket_name, version_history_file_path, f"{service_name}/{version_history_file}") else: logger.info("no version provided, not uploading to s3 with new convention") logger.info(f"removing file {file_name}") remove_env_file(file_name) logger.info_green_bg(f"upload env vars and secrets to s3 for {service_name} done!") except Exception as e: exit_on_error_and_write_summary(f"error in upload_env_vars_and_secrets_to_s3: {e}") if __name__ == "__main__": main()