""" Class to manage all the prompts during a guided sam deploy """ import logging from typing import Any, Dict, List, Optional import click from click import confirm, prompt from click.types import FuncParamType from samcli.commands._utils.options import _space_separated_list_func_type from samcli.commands._utils.template import ( get_template_parameters, ) from samcli.commands.deploy.auth_utils import auth_per_resource from samcli.commands.deploy.code_signer_utils import ( extract_profile_name_and_owner_from_existing, prompt_profile_name, prompt_profile_owner, signer_config_per_function, ) from samcli.commands.deploy.exceptions import GuidedDeployFailedError from samcli.commands.deploy.guided_config import GuidedConfig from samcli.commands.deploy.utils import sanitize_parameter_overrides from samcli.lib.bootstrap.bootstrap import manage_stack from samcli.lib.bootstrap.companion_stack.companion_stack_manager import CompanionStackManager, sync_ecr_stack from samcli.lib.config.samconfig import DEFAULT_CONFIG_FILE_NAME, DEFAULT_ENV from samcli.lib.intrinsic_resolver.intrinsics_symbol_table import IntrinsicsSymbolTable from samcli.lib.package.ecr_utils import is_ecr_url from samcli.lib.package.image_utils import NoImageFoundException, NonLocalImageException, tag_translation from samcli.lib.providers.provider import Function, ResourceIdentifier, Stack, get_resource_full_path_by_id from samcli.lib.providers.sam_function_provider import SamFunctionProvider from samcli.lib.providers.sam_stack_provider import SamLocalStackProvider from samcli.lib.utils.colors import Colored from samcli.lib.utils.defaults import get_default_aws_region from samcli.lib.utils.packagetype import IMAGE LOG = logging.getLogger(__name__) class GuidedContext: # pylint: disable=too-many-statements def __init__( self, template_file, stack_name, s3_bucket, image_repository, image_repositories, s3_prefix, resolve_s3=False, resolve_image_repos=False, region=None, profile=None, confirm_changeset=None, capabilities=None, signing_profiles=None, parameter_overrides=None, save_to_config=True, config_section=None, config_env=None, config_file=None, disable_rollback=None, ): self.template_file = template_file self.stack_name = stack_name self.s3_bucket = s3_bucket self.image_repository = image_repository self.image_repositories = image_repositories self.s3_prefix = s3_prefix self.region = region self.profile = profile self.confirm_changeset = confirm_changeset self.capabilities = (capabilities,) self.parameter_overrides_from_cmdline = parameter_overrides self.save_to_config = save_to_config self.config_section = config_section self.config_env = config_env self.config_file = config_file self.guided_stack_name = None self.guided_s3_bucket = None self.guided_image_repository = None self.guided_image_repositories = None self.guided_s3_prefix = None self.guided_region = None self.guided_profile = None self.resolve_s3 = resolve_s3 self.resolve_image_repositories = resolve_image_repos self.signing_profiles = signing_profiles self._capabilities = None self._parameter_overrides = None self.start_bold = "\033[1m" self.end_bold = "\033[0m" self.color = Colored() self.function_provider = None self.disable_rollback = disable_rollback @property def guided_capabilities(self): return self._capabilities @property def guided_parameter_overrides(self): return self._parameter_overrides # pylint: disable=too-many-statements def guided_prompts(self, parameter_override_keys): """ Start an interactive cli prompt to collection information for deployment Parameters ---------- parameter_override_keys The keys of parameters to override, for each key, customers will be asked to provide a value """ default_stack_name = self.stack_name or "sam-app" default_region = self.region or get_default_aws_region() default_capabilities = self.capabilities[0] or ("CAPABILITY_IAM",) default_config_env = self.config_env or DEFAULT_ENV default_config_file = self.config_file or DEFAULT_CONFIG_FILE_NAME input_capabilities = None config_env = None config_file = None click.echo( self.color.yellow( "\n\tSetting default arguments for 'sam deploy'\n\t=========================================" ) ) stack_name = prompt( f"\t{self.start_bold}Stack Name{self.end_bold}", default=default_stack_name, type=click.STRING ) region = prompt(f"\t{self.start_bold}AWS Region{self.end_bold}", default=default_region, type=click.STRING) global_parameter_overrides = {IntrinsicsSymbolTable.AWS_REGION: region} input_parameter_overrides = self.prompt_parameters( parameter_override_keys, self.parameter_overrides_from_cmdline, self.start_bold, self.end_bold ) stacks, _ = SamLocalStackProvider.get_stacks( self.template_file, parameter_overrides=sanitize_parameter_overrides(input_parameter_overrides), global_parameter_overrides=global_parameter_overrides, ) click.secho("\t#Shows you resources changes to be deployed and require a 'Y' to initiate deploy") confirm_changeset = confirm( f"\t{self.start_bold}Confirm changes before deploy{self.end_bold}", default=self.confirm_changeset ) click.secho("\t#SAM needs permission to be able to create roles to connect to the resources in your template") capabilities_confirm = confirm( f"\t{self.start_bold}Allow SAM CLI IAM role creation{self.end_bold}", default=True ) if not capabilities_confirm: input_capabilities = prompt( f"\t{self.start_bold}Capabilities{self.end_bold}", default=list(default_capabilities), type=FuncParamType(func=_space_separated_list_func_type), ) click.secho("\t#Preserves the state of previously provisioned resources when an operation fails") disable_rollback = confirm(f"\t{self.start_bold}Disable rollback{self.end_bold}", default=self.disable_rollback) self.prompt_authorization(stacks) self.prompt_code_signing_settings(stacks) save_to_config = confirm( f"\t{self.start_bold}Save arguments to configuration file{self.end_bold}", default=True ) if save_to_config: config_file = prompt( f"\t{self.start_bold}SAM configuration file{self.end_bold}", default=default_config_file, type=click.STRING, ) config_env = prompt( f"\t{self.start_bold}SAM configuration environment{self.end_bold}", default=default_config_env, type=click.STRING, ) click.echo("\n\tLooking for resources needed for deployment:") managed_s3_bucket = manage_stack(profile=self.profile, region=region) click.secho(f"\n\tManaged S3 bucket: {managed_s3_bucket}", bold=True) click.echo( "\tA different default S3 bucket can be set in samconfig.toml" " and auto resolution of buckets turned off by setting resolve_s3=False" ) image_repositories = ( sync_ecr_stack( self.template_file, stack_name, region, managed_s3_bucket, self.s3_prefix, self.image_repositories ) if self.resolve_image_repositories else self.prompt_image_repository( stack_name, stacks, self.image_repositories, region, managed_s3_bucket, self.s3_prefix ) ) self.guided_stack_name = stack_name self.guided_s3_bucket = managed_s3_bucket self.guided_image_repositories = image_repositories # NOTE(sriram-mv): The resultant s3 bucket is ALWAYS the managed_s3_bucket. There is no user flow to set it # within guided. self.resolve_s3 = True if self.guided_s3_bucket else False self.guided_s3_prefix = stack_name self.guided_region = region self.guided_profile = self.profile self._capabilities = input_capabilities if input_capabilities else default_capabilities self._parameter_overrides = ( input_parameter_overrides if input_parameter_overrides else self.parameter_overrides_from_cmdline ) self.save_to_config = save_to_config self.config_env = config_env if config_env else default_config_env self.config_file = config_file if config_file else default_config_file self.confirm_changeset = confirm_changeset self.disable_rollback = disable_rollback def prompt_authorization(self, stacks: List[Stack]): auth_required_per_resource = auth_per_resource(stacks) for resource, authorization_required in auth_required_per_resource: if not authorization_required: auth_confirm = confirm( f"\t{self.start_bold}{resource} has no authentication. Is this okay?{self.end_bold}", default=False, ) if not auth_confirm: raise GuidedDeployFailedError(msg="Security Constraints Not Satisfied!") def prompt_code_signing_settings(self, stacks: List[Stack]): """ Prompt code signing settings to ask whether customers want to code sign their code and display signing details. Parameters ---------- stacks : List[Stack] List of stacks to search functions and layers """ (functions_with_code_sign, layers_with_code_sign) = signer_config_per_function(stacks) # if no function or layer definition found with code signing, skip it if not functions_with_code_sign and not layers_with_code_sign: LOG.debug("No function or layer definition found with code sign config, skipping") return click.echo("\n\t#Found code signing configurations in your function definitions") sign_functions = confirm( f"\t{self.start_bold}Do you want to sign your code?{self.end_bold}", default=True, ) if not sign_functions: LOG.debug("User skipped code signing, continuing rest of the process") self.signing_profiles = None return if not self.signing_profiles: self.signing_profiles = {} click.echo("\t#Please provide signing profile details for the following functions & layers") for function_name in functions_with_code_sign: (profile_name, profile_owner) = extract_profile_name_and_owner_from_existing( function_name, self.signing_profiles ) click.echo(f"\t#Signing profile details for function '{function_name}'") profile_name = prompt_profile_name(profile_name, self.start_bold, self.end_bold) profile_owner = prompt_profile_owner(profile_owner, self.start_bold, self.end_bold) self.signing_profiles[function_name] = {"profile_name": profile_name, "profile_owner": profile_owner} self.signing_profiles[function_name]["profile_owner"] = "" if not profile_owner else profile_owner for layer_name, functions_use_this_layer in layers_with_code_sign.items(): (profile_name, profile_owner) = extract_profile_name_and_owner_from_existing( layer_name, self.signing_profiles ) click.echo( f"\t#Signing profile details for layer '{layer_name}', " f"which is used by functions {functions_use_this_layer}" ) profile_name = prompt_profile_name(profile_name, self.start_bold, self.end_bold) profile_owner = prompt_profile_owner(profile_owner, self.start_bold, self.end_bold) self.signing_profiles[layer_name] = {"profile_name": profile_name, "profile_owner": profile_owner} self.signing_profiles[layer_name]["profile_owner"] = "" if not profile_owner else profile_owner LOG.debug("Signing profile names and owners %s", self.signing_profiles) def prompt_parameters( self, parameter_override_from_template, parameter_override_from_cmdline, start_bold, end_bold ): _prompted_param_overrides = {} if parameter_override_from_template: for parameter_key, parameter_properties in parameter_override_from_template.items(): no_echo = parameter_properties.get("NoEcho", False) if no_echo: parameter = prompt( f"\t{start_bold}Parameter {parameter_key}{end_bold}", type=click.STRING, hide_input=True ) _prompted_param_overrides[parameter_key] = {"Value": parameter, "Hidden": True} else: parameter = prompt( f"\t{start_bold}Parameter {parameter_key}{end_bold}", default=_prompted_param_overrides.get( parameter_key, self._get_parameter_value( parameter_key, parameter_properties, parameter_override_from_cmdline ), ), type=click.STRING, ) _prompted_param_overrides[parameter_key] = {"Value": parameter, "Hidden": False} return _prompted_param_overrides def prompt_image_repository( self, stack_name, stacks: List[Stack], image_repositories: Optional[Dict[str, str]], region: str, s3_bucket: str, s3_prefix: str, ) -> Dict[str, str]: """ Prompt for the image repository to push the images. For each image function found in build artifacts, it will prompt for an image repository. Parameters ---------- stack_name : List[Stack] Name of the stack to be deployed. stacks : List[Stack] List of stacks to look for image functions. image_repositories: Dict[str, str] Dictionary with function logical ID as key and image repo URI as value. region: str Region for the image repos. s3_bucket: str s3 bucket URI to be used for uploading companion stack template s3_prefix: str s3 prefix to be used for uploading companion stack template Returns ------- Dict[str, str] A dictionary contains image function logical ID as key, image repository as value. """ image_repositories = image_repositories if image_repositories is not None else {} updated_repositories = {} for image_repo_func_id, image_repo_uri in image_repositories.items(): repo_full_path = get_resource_full_path_by_id(stacks, ResourceIdentifier(image_repo_func_id)) if repo_full_path: updated_repositories[repo_full_path] = image_repo_uri self.function_provider = SamFunctionProvider(stacks, ignore_code_extraction_warnings=True) manager = CompanionStackManager(stack_name, region, s3_bucket, s3_prefix) function_logical_ids = [ function.full_path for function in self.function_provider.get_all() if function.packagetype == IMAGE ] functions_without_repo = [ function_logical_id for function_logical_id in function_logical_ids if function_logical_id not in updated_repositories ] manager.set_functions(function_logical_ids, updated_repositories) create_all_repos = self.prompt_create_all_repos( function_logical_ids, functions_without_repo, updated_repositories ) if create_all_repos: updated_repositories.update(manager.get_repository_mapping()) else: updated_repositories = self.prompt_specify_repos(functions_without_repo, updated_repositories) manager.set_functions(function_logical_ids, updated_repositories) updated_repositories = self.prompt_delete_unreferenced_repos( [manager.get_repo_uri(repo) for repo in manager.get_unreferenced_repos()], updated_repositories ) GuidedContext.verify_images_exist_locally(self.function_provider.functions) manager.sync_repos() return updated_repositories def prompt_specify_repos( self, functions_without_repos: List[str], image_repositories: Dict[str, str], ) -> Dict[str, str]: """ Show prompts for each function that isn't associated with a image repo Parameters ---------- functions_without_repos: List[str] List of functions without associating repos image_repositories: Dict[str, str] Current image repo dictionary with function logical ID as key and image repo URI as value. Returns ------- Dict[str, str] Updated image repo dictionary with values(image repo URIs) filled by user input """ updated_repositories = image_repositories.copy() for function_logical_id in functions_without_repos: image_uri = prompt( f"\t {self.start_bold}ECR repository for {function_logical_id}{self.end_bold}", type=click.STRING, ) if not is_ecr_url(image_uri): raise GuidedDeployFailedError(f"Invalid Image Repository ECR URI: {image_uri}") # NOTE(sriram-mv): If a prompt to accept an ECR URI succeeded, then one does not any longer # resolve image repositories automatically. self.resolve_image_repositories = False updated_repositories[function_logical_id] = image_uri return updated_repositories def prompt_create_all_repos( self, functions: List[str], functions_without_repo: List[str], existing_mapping: Dict[str, str] ) -> bool: """ Prompt whether to create all repos Parameters ---------- functions: List[str] List of function logical IDs that are image based functions_without_repo: List[str] List of function logical IDs that do not have an ECR image repo specified existing_mapping: Dict[str, str] Current image repo dictionary with function logical ID as key and image repo URI as value. This dict will be shown in the terminal. Returns ------- Boolean Returns False if there is no missing function or denied by prompt """ if not functions: return False # Case for when all functions do not have mapped repo if functions == functions_without_repo: click.echo("\t Image repositories: Not found.") click.echo( "\t #Managed repositories will be deleted when " "their functions are removed from the template and deployed" ) return confirm( f"\t {self.start_bold}Create managed ECR repositories for all functions?{self.end_bold}", default=True ) functions_with_repo_count = len(functions) - len(functions_without_repo) click.echo( "\t Image repositories: " f"Found ({functions_with_repo_count} of {len(functions)})" " #Different image repositories can be set in samconfig.toml" ) for function_logical_id, repo_uri in existing_mapping.items(): click.echo(f"\t {function_logical_id}: {repo_uri}") # Case for all functions do have mapped repo if not functions_without_repo: return False click.echo( "\t #Managed repositories will be deleted when their functions are " "removed from the template and deployed" ) return ( confirm( f"\t {self.start_bold}Create managed ECR repositories for the " f"{len(functions_without_repo)} functions without?{self.end_bold}", default=True, ) if functions_without_repo else True ) def prompt_delete_unreferenced_repos( self, unreferenced_repo_uris: List[str], image_repositories: Dict[str, str] ) -> Dict[str, str]: """ Prompt user for deleting unreferenced companion stack image repos. Throws GuidedDeployFailedError if delete repos has been denied by the user. This function does not actually remove the functions from the stack. Parameters ---------- unreferenced_repo_uris: List[str] List of unreferenced image repos that need to be deleted. image_repositories: Dict[str, str] Dictionary of image repo URIs with key as function logical ID and value as image repo URI Returns ------- Dict[str, str] Copy of image_repositories that have unreferenced image repos removed """ output_image_repositories = image_repositories.copy() if not unreferenced_repo_uris: return output_image_repositories click.echo("\t Checking for unreferenced ECR repositories to clean-up: " f"{len(unreferenced_repo_uris)} found") for repo_uri in unreferenced_repo_uris: click.echo(f"\t {repo_uri}") delete_repos = confirm( f"\t {self.start_bold}Delete the unreferenced repositories listed above when deploying?{self.end_bold}", default=False, ) if not delete_repos: click.echo("\t Deployment aborted!") click.echo( "\t #The deployment was aborted to prevent " "unreferenced managed ECR repositories from being deleted.\n" "\t #You may remove repositories from the AWS SAM CLI " "managed stack to retain them and resolve this unreferenced check." ) raise GuidedDeployFailedError("Unreferenced Auto Created ECR Repos Must Be Deleted.") for function_logical_id, repo_uri in image_repositories.items(): if repo_uri in unreferenced_repo_uris: del output_image_repositories[function_logical_id] break return output_image_repositories @staticmethod def verify_images_exist_locally(functions: Dict[str, Function]) -> None: """ Verify all images associated with deploying functions exist locally. Parameters ---------- functions: Dict[str, Function] Dictionary of functions in the stack to be deployed with key as their logical ID. """ for _, function_prop in functions.items(): if function_prop.packagetype != IMAGE: continue image = function_prop.imageuri try: tag_translation(image) except NonLocalImageException: LOG.debug("Image URI is not pointing to local. Skipping verification.") except NoImageFoundException as ex: raise GuidedDeployFailedError("No images found to deploy, try running sam build") from ex def run(self): try: _parameter_override_keys = get_template_parameters(template_file=self.template_file) except ValueError as ex: LOG.debug("Failed to parse SAM template", exc_info=ex) raise GuidedDeployFailedError(str(ex)) from ex guided_config = GuidedConfig(template_file=self.template_file, section=self.config_section) guided_config.read_config_showcase( self.config_file or DEFAULT_CONFIG_FILE_NAME, ) self.guided_prompts(_parameter_override_keys) if self.save_to_config: guided_config.save_config( self._parameter_overrides, self.config_env or DEFAULT_ENV, self.config_file or DEFAULT_CONFIG_FILE_NAME, stack_name=self.guided_stack_name, resolve_s3=self.resolve_s3, s3_prefix=self.guided_s3_prefix, image_repositories=self.guided_image_repositories if not self.resolve_image_repositories else None, resolve_image_repos=self.resolve_image_repositories, region=self.guided_region, profile=self.guided_profile, confirm_changeset=self.confirm_changeset, capabilities=self._capabilities, signing_profiles=self.signing_profiles, disable_rollback=self.disable_rollback, ) @staticmethod def _get_parameter_value( parameter_key: str, parameter_properties: Dict, parameter_override_from_cmdline: Dict ) -> Any: """ This function provide the value of a parameter. If the command line/config file have "override_parameter" whose key exist in the template file parameters, it will use the corresponding value. Otherwise, it will use its default value in template file. :param parameter_key: key of parameter :param parameter_properties: properties of that parameters from template file :param parameter_override_from_cmdline: parameter_override from command line/config file """ if parameter_override_from_cmdline and parameter_override_from_cmdline.get(parameter_key, None): return parameter_override_from_cmdline[parameter_key] # Make sure the default is casted to a string. return str(parameter_properties.get("Default", ""))