import os from pathlib import Path from botocore.exceptions import ClientError from packaging.version import Version from actions_logging.app_logging import logger from github.env import (exit_on_error_and_write_summary, get_required_env_var, write_github_env) from terragrunt.tg_common import (get_bucket_name, get_from_env_config, s3_client, sanitize_work_dir, ConfigError) from terragrunt.constants import (INITIAL_VERSION, TOOLS_REQUIREMENTS, APPLY, DESTROY) def check_bucket_exists(env_name): bucket_name = get_bucket_name(env_name) try: logger.info(f"Checking if {bucket_name} exists") s3_client.list_objects(Bucket=bucket_name, MaxKeys=1) logger.info(f"bucket exists") except ClientError as e: # https://docs.aws.amazon.com/AmazonS3/latest/API/API_HeadBucket.html if e.response['Error']['Code'] in ['404', '403', '400']: exit_on_error_and_write_summary(f"bucket {bucket_name} doesn't exist") else: exit_on_error_and_write_summary(f"Unexpected error: {e.response['Error']['Message']} while trying to check bucket {bucket_name} existance") def get_semver_obj(semver): try: return Version(semver) except Exception as e: exit_on_error_and_write_summary(f"failed to cast to Version {semver} due to error: {e}") def validate_plan_version(plan_version, exec_mode): """ Validate the plan version based on the execution mode and save plan flag. This function checks the plan version according to the provided execution mode. It ensures that a valid semantic version string is provided when running in "apply" mode and that the version is not less than the initial version. The validation is bypassed if save_plan is provided. Args: plan_version (str): The semantic version string of the plan. exec_mode (str): The execution mode ('apply' or 'destroy'). save_plan (bool): Flag to indicate if the plan should be saved. Raises: SystemExit: If validation fails due to missing plan version or invalid version. """ if exec_mode == DESTROY: logger.info('running in destroy mode - ignoring PLAN_VERSION input') return if not plan_version and exec_mode == APPLY: exit_on_error_and_write_summary('semantic PLAN_VERSION string is required parameter when running in "apply" mode') semver = get_semver_obj(plan_version) if semver < Version(INITIAL_VERSION): exit_on_error_and_write_summary(f"Plan version cannot be less than {INITIAL_VERSION}, {plan_version} provided.") def validate_env_work_dir(env_name, work_dir): """ Validate the environment name and working directory combination. This function checks if the provided environment name exists in the environment data and if the top-level directory path of the working directory matches the expected top level directory for the environment set in .github/iac_deploy_config.json. Args: env_name (str): The name of the environment. work_dir (str): The working directory path. Raises: SystemExit: If the environment name is unknown or if the working directory does not match the expected top-level directory for the environment. """ try: top_level_path = get_from_env_config(env_name, 'top_level_path') if not work_dir.startswith(top_level_path): exit_on_error_and_write_summary(f"Bad work_dir {work_dir} and environment '{env_name}' inputs combination provided, work_dir should start from {top_level_path}") except ConfigError as ce: exit_on_error_and_write_summary(f"couldn't validate work_dir for {env_name}, error occured: {ce}") def get_required_tool_version(env_name, tool, file_name): """ This function checks the provided version against the required version for the tool found in the $top_level_path/infra/.$toolname-version file. if file is found the version is taken from it where version was supplied or not. If no file was found but version was supplied on inputs - it is valided and used. If neither exist the process fails. """ try: top_level_path = get_from_env_config(env_name, 'top_level_path') requirements_file_path = os.path.join(top_level_path, file_name) logger.info(f"looking for {requirements_file_path} for {env_name}") with open(requirements_file_path, 'r') as f: required_version = str(Version(f.readline())) logger.warning(f"Required {tool} version is: {required_version}") return required_version except ConfigError as ce: exit_on_error_and_write_summary(f"Don't know where to look for the {file_name} due to deploy misconfiguration for {env_name}: {ce}") except FileNotFoundError: logger.warning(f"Requirement file {file_name} for {tool} doesn't exist for {env_name}") return None except Exception as e: exit_on_error_and_write_summary(f"error while reading version from {file_name} for {tool} for {env_name} due to error: {e}") def uses_hcl_json(search_path, env_name): """ Check if the provided path or any of its parent directories contain .hcl.json files up to top_level_path. """ try: search_path = Path(search_path) logger.debug(f"Searching for hcl.json files in {search_path}") top_level_path = get_from_env_config(env_name, 'top_level_path') files = list(search_path.glob('*')) hcl_json_files = [file for file in files if file.name.endswith('.hcl.json')] if not hcl_json_files: if search_path == Path(top_level_path): logger.info(f"No hcl.json files found in {search_path} or it's parent") return False return uses_hcl_json(search_path.parent, env_name) logger.warning(f"hcl.json files found in the manifest directory or one of it's parents: {search_path}") return True except Exception as e: exit_on_error_and_write_summary(f"error while searching for hcl.json files in {search_path} and it's parent folders: {e}") def validate_and_update_tool_version(version, work_dir, env_name, tool): """ Validate and enforce the version of the specified tool based on the requirements file or file extensions. Args: version (str): The version of the tool to validate and update. work_dir (str): The working directory containing the IaC files. env_name (str): The environment name we running the tool in. tool (str): The tool to validate and update ('terraform' or 'terragrunt'). Returns: str: The validated and updated version of the tool. Raises: SystemExit: If the tool requirements are undefined or if any error occurs during the process. Example: >>> validate_and_update_tool_version('0.12.0', '/path/to/work_dir', 'terraform') '0.12.0' """ logger.info(f"Validating and probably enforcing {tool} version to use") try: tool_reqs = TOOLS_REQUIREMENTS[tool] except KeyError: raise ConfigError(f"requirements for {tool} are undefined") if tool == 'terragrunt' and uses_hcl_json(Path(work_dir), env_name): tg_old_version = tool_reqs.get("old_version") logger.warning(f"terragrunt {tg_old_version} will be used") return tg_old_version try: requirements_file = tool_reqs["requirements_file"] except KeyError: raise ConfigError(f"{tool}'s {requirements_file} property is undefined") required_version = get_required_tool_version(env_name, tool, requirements_file) if required_version: return required_version if not version: exit_on_error_and_write_summary(f"{tool} version was not provided on input and {requirements_file} not found in {work_dir}") logger.warning(f"{requirements_file} not found for the {tool}, will try using version from input: {version}") if version == "latest": logger.warning(f"Using {tool}'s latest version") return version input_version = str(get_semver_obj(version)) logger.info(f"using input version {input_version} for {tool}") return input_version def select_iac_tool(work_dir): logger.info("Selecting tool to run the IaC commands with") tool = 'terragrunt' all_files = list(Path(work_dir).glob('*')) terragrunt_files = [file.as_posix() for file in all_files if file.suffix == '.hcl' or file.name.endswith('.hcl.json')] logger.info(f"terragrunt files: {terragrunt_files}") if not terragrunt_files: tool = 'terraform' logger.warning(f"tool selected is {tool}") return tool def main(): work_dir = sanitize_work_dir(get_required_env_var('WORK_DIR')) env_name = get_required_env_var('ENV_NAME') exec_mode = get_required_env_var('EXECUTION_MODE') validate_env_work_dir(env_name, work_dir) plan_version = os.getenv('PLAN_VERSION', None) tg_version = os.getenv('TG_VERSION', None) tf_version = os.getenv('TF_VERSION', None) save_plan = os.getenv('SAVE_PLAN') logger.info(f"save_plan: {save_plan}") # TODO: what can we validate/configure here? command_args = os.getenv('COMMAND_ARGS') check_bucket_exists(env_name) if save_plan is None: validate_plan_version(plan_version, exec_mode) tool = select_iac_tool(work_dir) try: tf_version = validate_and_update_tool_version(tf_version, work_dir, env_name, 'terraform') tg_version = validate_and_update_tool_version(tg_version, work_dir, env_name, 'terragrunt') except Exception as e: exit_on_error_and_write_summary(f"couldn't update tool version due to an error: {e}") write_github_env(tf_version, 'TF_VERSION') write_github_env(tg_version, 'TG_VERSION') write_github_env(tool, 'TOOL') if __name__ == "__main__": main()