import os from typing import Union from aws.secret_manager import get_secrets as get_secrets_from_secret_manager from aws.s3_apis.s3 import upload_file_to_s3, copy_s3_file, check_file_exists_in_s3, delete_file_from_s3 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 git.utils import get_dev_version from aws.constants import PROD_ENVS, PRODUCTION, PROD_ACCOUNT_ENVS from env_files.create_dot_env import create_dot_env_from_json from env_files.constants import DEFAULT_ENV_FILES_PATH, ENV_FILES_SUBFOLDERS from master.calculate_version import Version import json def get_backup_file_name_and_bucket(env_name, source): dev_version = get_dev_version(env_name) bucket_name = get_env_bucket(env_name) return bucket_name, f"{source}_bkp_{dev_version}" def backup_dot_env(env_name, source): bucket_name, bkp_file = get_backup_file_name_and_bucket(env_name, source) logger.info(f"backing up current .env file {source} to {bkp_file}") if check_file_exists_in_s3(bucket_name, source): copy_s3_file(bucket_name, source, bucket_name, bkp_file) else: logger.warning(f"source {source} .env file not exists in s3 in {bucket_name} for {env_name}, skipping backup") def restore_dot_env(env_name, source): bucket_name, bkp_file = get_backup_file_name_and_bucket(env_name, source) logger.info(f"restoring .env from backup {bkp_file} to {source}") env_name_to_assume_role = env_name if env_name not in PROD_ACCOUNT_ENVS else '' if check_file_exists_in_s3(bucket_name, bkp_file, env_name_to_assume_role): copy_s3_file(bucket_name, bkp_file, bucket_name, source) else: logger.warning(f"backup {bkp_file} .env file not exists in s3 in {bucket_name} for {env_name}, nothing to restore") def delete_backup_dot_env(env_name, source): bucket_name, bkp_file = get_backup_file_name_and_bucket(env_name, source) logger.info(f"deleting backup .env {bkp_file}") try: env_name_to_assume_role = env_name if env_name not in PROD_ACCOUNT_ENVS else '' delete_file_from_s3(bucket_name, bkp_file, env_name_to_assume_role) logger.info(f"deleted backup .env {bkp_file}") except Exception as e: logger.error(f"Object {bkp_file} couldn't be deleted due to error: {e}") # new bucket tree: # ms.env # ms/ # version_history.txt * ignore for now # v1.0.0/ # ms.env def copy_versioned_dot_env_to_root(env_version: Union[str, Version], bucket_name, service_name): try: logger.debug(f"service_name: {service_name}, env_version: {env_version}, bucket_name: {bucket_name}") source = os.path.join(service_name, str(env_version), f"{service_name}.env") logger.info(f"copying versioned dot env file {bucket_name}/{source} to the root of the bucket") if not check_file_exists_in_s3(bucket_name, source): exit_on_error_and_write_summary(f"versioned dot env file {source} not exists in s3 {bucket_name}") dst = f"{service_name}.env" copy_s3_file(bucket_name, source, bucket_name, dst) logger.info(f"copying versioned dot env file {source} to the root of the {bucket_name} done!") except Exception as e: raise RuntimeError(f"error in copy_versioned_dot_env_to_root: {e}") def get_svc_name_from_file_path(env_file) -> str: """ will get the svc name from the file path if it single service repo will return empty string :param env_file: :return: svc name or empty string """ try: empty_svc_name = "" if not env_file.startswith(DEFAULT_ENV_FILES_PATH): logger.warning( f"env file {env_file} not starts with {DEFAULT_ENV_FILES_PATH} folder, returning empty svc name") return empty_svc_name path_without_base = env_file.replace(f"{DEFAULT_ENV_FILES_PATH}/", "") potential_svc_name = path_without_base.split("/")[0] if potential_svc_name in ENV_FILES_SUBFOLDERS: logger.warning(f"env file {env_file} is not in multi service repo structure, returning empty svc name") return empty_svc_name return potential_svc_name except Exception as e: logger.warning(f"error in get_svc_name_from_file_path: {e}. will return empty string") return "" def load_secrets_to_env(secret_id, env_name): try: secrets_res = get_secrets_from_secret_manager(secret_id, env_name) if not secrets_res: logger.warning(f"received empty response for secret_id: {secret_id}") return secrets = json.loads(secrets_res) secret_prefix = secret_id.replace("/", "_").replace("-", "_").upper() for secret in secrets: key = secret value = secrets[key] upper_key = key.replace('-', '_').upper() key_with_prefix = f"{secret_prefix}_{upper_key}" if isinstance(value, dict): for sub_key in value: sub_value = value[sub_key] upper_sub_key = sub_key.replace('-', '_').upper() os.environ[f"{key_with_prefix}_{upper_sub_key}"] = str(sub_value) logger.debug(f"secret {key_with_prefix}_{upper_sub_key} loaded to env") else: os.environ[key_with_prefix] = value logger.debug(f"secret {key_with_prefix} loaded to env") except Exception as e: logger.warning(f"error in load_secrets_to_env: {e}") def render_secret_id_and_load_secrets_to_env(svc_name, env_name): """ get the secrets from aws secret manager and load the into the environment the load part will be same as the aws-actions/aws-secretsmanager-get-secrets@v2 action. woth pars-jaon set to true :param svc_name: :param env_name: :return: """ try: if not svc_name and not env_name: return if env_name: env_level = env_name if env_name in PROD_ENVS: env_level = PRODUCTION secret_id = f"env/{env_level}" load_secrets_to_env(secret_id, env_name) if svc_name: secret_id = f"global/{svc_name}" load_secrets_to_env(secret_id, env_name) if svc_name and env_name: secret_id = f"env/{env_level}/{svc_name}" load_secrets_to_env(secret_id, env_name) except RuntimeError as e: logger.warning(f"couldn't get and load the secrets: {e}") except Exception as e: logger.error(f"error in render_secret_id_and_load_secrets_to_env: {e}") def build_and_upload_dot_envs(env_versions_data: dict, is_multi_service_repo: bool, env_file_path: str = DEFAULT_ENV_FILES_PATH): """ will create dot env files from json files and upload them to s3 :param env_versions_data: dict with env file path as key and version as value :param is_multi_service_repo: :return: """ try: logger.info("will generate and upload env files to s3:") is_merged = os.getenv('EVENT_TYPE', '') == 'merged' svc_name = None # TODO: here and in produce_dot_env.py we need a more elegant way to handle multi-service and double-service secret paths in SM # e.g. sx-dns: env/$env_name/sx-dns svc_sm_path_suffix = None if not is_multi_service_repo: svc_name = get_required_env_var('SVC_NAME') svc_sm_path_suffix = svc_name logger.debug(f"will parse this data to create dot env files: {env_versions_data}") for env_file in env_versions_data: if is_multi_service_repo: # multi service repo, will get the svc name from file path svc_name = get_svc_name_from_file_path(env_file) svc_sm_path_suffix = get_required_env_var('SVC_NAME') env_name = os.path.basename(env_file).split('.json')[0] env_version = env_versions_data[env_file] logger.info(f"creating dot env file for {env_name} {env_version} {svc_name}") bucket_name = get_env_bucket(env_name) dot_env_file = f"{svc_name}.env" # if not is_dev: # only happen in master build (auto) render_secret_id_and_load_secrets_to_env(svc_name=svc_sm_path_suffix, env_name=env_name) create_dot_env_from_json(env_file, env_name, svc_name, is_multi_service_repo, base_env_files_path=env_file_path) file_path_in_s3 = f"{svc_name}/{env_version}/{dot_env_file}" # if is_dev: # file_path_in_s3 = dot_env_file env_name_to_assume = env_name if is_merged else '' upload_file_to_s3(bucket_name=bucket_name, file_name=dot_env_file, file_path=file_path_in_s3, env_name=env_name_to_assume) logger.info(f"removing file {dot_env_file}") os.remove(dot_env_file) logger.info(f"uploading dot env file for {env_name} {env_version} {svc_name} done!") except Exception as e: raise RuntimeError(f"Failed to build and upload dot envs: {e}")