feat: Lambda Code Signer Support

requirements/base.txt

@@ -6,7 +6,7 @@ boto3~=1.14.23
 jmespath~=0.10.0
 PyYAML~=5.3
 cookiecutter~=1.7.2
-aws-sam-translator==1.30.1
+aws-sam-translator==1.31.0
 #docker minor version updates can include breaking changes. Auto update micro version only.
 docker~=4.2.0
 dateparser~=0.7

requirements/reproducible-linux.txt

@@ -12,10 +12,10 @@ attrs==19.3.0 \
     --hash=sha256:08a96c641c3a74e44eb59afb61a24f2cb9f4d7188748e76ba4bb5edfa3cb7d1c \
     --hash=sha256:f7b7ce16570fe9965acd6d30101a28f62fb4a7f9e926b3bbc9b61f8b04247e72 \
     # via jsonschema
-aws-sam-translator==1.30.1 \
-    --hash=sha256:1a1903fd1fda55bac5ba66b97b24c4b6b599973844e4ea45d0d1bcf381c22825 \
-    --hash=sha256:22b89488e449a3f368ae8ab1eaacc1ebb190970de31d6492904252ab21117419 \
-    --hash=sha256:bd9dbdf0773b52bb5bf652954bf47ab6bd5e2bd752255209207c8a4e52bb3735 \
+aws-sam-translator==1.31.0 \
+    --hash=sha256:3a1d73d098161e60966b0d53bb310c98e4f66101688cce3d1697903643782d79 \
+    --hash=sha256:65749571042e704027bbcaabe6dddd5d13ab54959c46e60c347918d6b85551fd \
+    --hash=sha256:b87a61cffba8f29e4f1b5eb43a4abafdfb14b76aedc716517d48dbdf0b1f989a \
     # via aws-sam-cli (setup.py)
 aws_lambda_builders==1.1.0 \
     --hash=sha256:2b40a0003c2c05143e1aa816fed758c7d78f3e5c8e115be681aa2478f2655056 \

samcli/__init__.py

@@ -2,4 +2,4 @@
 SAM CLI version
 """
 
-__version__ = "1.11.0"
+__version__ = "1.12.0"

samcli/cli/types.py

@@ -12,7 +12,6 @@
 
 
 def _generate_match_regex(match_pattern, delim):
-
     """
     Creates a regex string based on a match pattern (also a regex) that is to be
     run on a string (which may contain escaped quotes) that is separated by delimiters.
@@ -272,3 +271,82 @@ def _space_separated_key_value_parser(tag_value):
                 return False, None
             tags_dict = {**tags_dict, **parsed_tag}
         return True, tags_dict
+
+
+class SigningProfilesOptionType(click.ParamType):
+    """
+    Custom parameter type to parse Signing Profile options
+    Example options could be;
+    - MyFunctionOrLayerToBeSigned=MySigningProfile
+    - MyFunctionOrLayerToBeSigned=MySigningProfile:MyProfileOwner
+
+    See convert function docs for details
+    """
+
+    pattern = r"(?:(?: )([A-Za-z0-9\"]+)=(\"(?:\\.|[^\"\\]+)*\"|(?:\\.|[^ \"\\]+)+))"
+
+    # Note: this is required, otherwise it is failing when running help
+    name = ""
+
+    def convert(self, value, param, ctx):
+        """
+        Converts given Signing Profile options to a dictionary where Function or Layer name would be key,
+        and signing profile details would be the value.
+
+        Since this method is also been used when we are reading the config details from samconfig.toml,
+        If value is already a dictionary, don't need to do anything, just return it.
+        If value is an array, then each value will correspond to a function or layer config, and here it is parsed
+        and converted into a dictionary
+        """
+        result = {}
+
+        # Empty tuple
+        if value == ("",):
+            return result
+
+        value = (value,) if isinstance(value, str) else value
+        for val in value:
+            val.strip()
+            # Add empty string to start of the string to help match `_pattern2`
+            val = " " + val
+
+            signing_profiles = re.findall(self.pattern, val)
+
+            # if no signing profiles found by regex, then fail
+            if not signing_profiles:
+                return self.fail(
+                    f"{value} is not a valid code sign config, it should look like this 'MyFunction=SigningProfile'",
+                    param,
+                    ctx,
+                )
+
+            for function_name, param_value in signing_profiles:
+                (signer_profile_name, signer_profile_owner) = self._split_signer_profile_name_owner(
+                    _unquote_wrapped_quotes(param_value)
+                )
+
+                # code signing requires profile name, if it is not present then fail
+                if not signer_profile_name:
+                    return self.fail(
+                        "Signer profile option has invalid format, it should look like this "
+                        "MyFunction=MySigningProfile or MyFunction=MySigningProfile:MySigningProfileOwner",
+                        param,
+                        ctx,
+                    )
+
+                result[_unquote_wrapped_quotes(function_name)] = {
+                    "profile_name": signer_profile_name,
+                    "profile_owner": signer_profile_owner,
+                }
+
+        return result
+
+    @staticmethod
+    def _split_signer_profile_name_owner(signing_profile):
+        equals_count = signing_profile.count(":")
+
+        if equals_count > 1:
+            return None, None
+        if equals_count == 1:
+            return signing_profile.split(":")
+        return signing_profile, ""

samcli/commands/_utils/options.py

@@ -10,7 +10,7 @@
 from click.types import FuncParamType
 
 from samcli.commands._utils.template import get_template_data, TemplateNotFoundException
-from samcli.cli.types import CfnParameterOverridesType, CfnMetadataType, CfnTags
+from samcli.cli.types import CfnParameterOverridesType, CfnMetadataType, CfnTags, SigningProfilesOptionType
 from samcli.commands._utils.custom_options.option_nargs import OptionNargs
 
 _TEMPLATE_OPTION_DEFAULT_VALUE = "template.[yaml|yml]"
@@ -189,6 +189,23 @@ def no_progressbar_option(f):
     return no_progressbar_click_option()(f)
 
 
+def signing_profiles_click_option():
+    return click.option(
+        "--signing-profiles",
+        cls=OptionNargs,
+        type=SigningProfilesOptionType(),
+        default={},
+        help="Optional. A string that contains Code Sign configuration parameters as "
+        "FunctionOrLayerNameToSign=SigningProfileName:SigningProfileOwner "
+        "Since signing profile owner is optional, it could also be written as "
+        "FunctionOrLayerNameToSign=SigningProfileName",
+    )
+
+
+def signing_profiles_option(f):
+    return signing_profiles_click_option()(f)
+
+
 def metadata_click_option():
     return click.option(
         "--metadata",

samcli/commands/deploy/code_signer_utils.py

@@ -0,0 +1,66 @@
+"""
+Utilities for code signing process
+"""
+
+import logging
+from click import prompt, STRING
+
+from samcli.lib.providers.sam_function_provider import SamFunctionProvider
+
+LOG = logging.getLogger(__name__)
+
+
+def prompt_profile_name(profile_name, start_bold, end_bold):
+    return prompt(f"\t{start_bold}Signing Profile Name{end_bold}", type=STRING, default=profile_name)
+
+
+def prompt_profile_owner(profile_owner, start_bold, end_bold):
+    # click requires to have non None value for passing
+    if not profile_owner:
+        profile_owner = ""
+
+    profile_owner = prompt(
+        f"\t{start_bold}Signing Profile Owner Account ID (optional){end_bold}",
+        type=STRING,
+        default=profile_owner,
+        show_default=len(profile_owner) > 0,
+    )
+
+    return profile_owner
+
+
+def extract_profile_name_and_owner_from_existing(function_or_layer_name, signing_profiles):
+    profile_name = None
+    profile_owner = None
+    # extract any code sign config that is passed via command line
+    if function_or_layer_name in signing_profiles:
+        profile_name = signing_profiles[function_or_layer_name]["profile_name"]
+        profile_owner = signing_profiles[function_or_layer_name]["profile_owner"]
+
+    return profile_name, profile_owner
+
+
+def signer_config_per_function(parameter_overrides, template_dict):
+    functions_with_code_sign = set()
+    layers_with_code_sign = {}
+
+    sam_functions = SamFunctionProvider(template_dict=template_dict, parameter_overrides=parameter_overrides)
+
+    for sam_function in sam_functions.get_all():
+        if sam_function.codesign_config_arn:
+            function_name = sam_function.name
+            LOG.debug("Found the following function with a code signing config %s", function_name)
+            functions_with_code_sign.add(function_name)
+
+            if sam_function.layers:
+                for layer in sam_function.layers:
+                    layer_name = layer.name
+                    LOG.debug("Found following layers inside the function %s", layer_name)
+                    if layer_name in layers_with_code_sign:
+                        layers_with_code_sign[layer_name].add(function_name)
+                    else:
+                        functions_that_is_referring_to_function = set()
+                        functions_that_is_referring_to_function.add(function_name)
+                        layers_with_code_sign[layer_name] = functions_that_is_referring_to_function
+
+    return functions_with_code_sign, layers_with_code_sign

samcli/commands/deploy/command.py

@@ -16,6 +16,7 @@
     no_progressbar_option,
     tags_override_option,
     template_click_option,
+    signing_profiles_option,
 )
 from samcli.commands.deploy.utils import sanitize_parameter_overrides
 from samcli.lib.telemetry.metrics import track_command
@@ -140,6 +141,7 @@
 @notification_arns_override_option
 @tags_override_option
 @parameter_override_option
+@signing_profiles_option
 @no_progressbar_option
 @capabilities_override_option
 @aws_creds_options
@@ -166,6 +168,7 @@ def cli(
     metadata,
     guided,
     confirm_changeset,
+    signing_profiles,
     resolve_s3,
     config_file,
     config_env,
@@ -193,6 +196,7 @@ def cli(
         confirm_changeset,
         ctx.region,
         ctx.profile,
+        signing_profiles,
         resolve_s3,
         config_file,
         config_env,
@@ -220,6 +224,7 @@ def do_cli(
     confirm_changeset,
     region,
     profile,
+    signing_profiles,
     resolve_s3,
     config_file,
     config_env,
@@ -240,6 +245,7 @@ def do_cli(
             profile=profile,
             confirm_changeset=confirm_changeset,
             capabilities=capabilities,
+            signing_profiles=signing_profiles,
             parameter_overrides=parameter_overrides,
             config_section=CONFIG_SECTION,
             config_env=config_env,
@@ -269,6 +275,7 @@ def do_cli(
             on_deploy=True,
             region=guided_context.guided_region if guided else region,
             profile=profile,
+            signing_profiles=guided_context.signing_profiles if guided else signing_profiles,
         ) as package_context:
             package_context.run()
 
@@ -292,5 +299,6 @@ def do_cli(
             region=guided_context.guided_region if guided else region,
             profile=profile,
             confirm_changeset=guided_context.confirm_changeset if guided else confirm_changeset,
+            signing_profiles=guided_context.signing_profiles if guided else signing_profiles,
         ) as deploy_context:
             deploy_context.run()

samcli/commands/deploy/deploy_context.py

@@ -61,6 +61,7 @@ def __init__(
         region,
         profile,
         confirm_changeset,
+        signing_profiles,
     ):
         self.template_file = template_file
         self.stack_name = stack_name
@@ -81,6 +82,7 @@ def __init__(
         self.s3_uploader = None
         self.deployer = None
         self.confirm_changeset = confirm_changeset
+        self.signing_profiles = signing_profiles
 
     def __enter__(self):
         return self
@@ -129,6 +131,7 @@ def run(self):
             self.capabilities,
             self.parameter_overrides,
             self.confirm_changeset,
+            self.signing_profiles,
         )
         return self.deploy(
             self.stack_name,

samcli/commands/deploy/guided_config.py

@@ -41,7 +41,9 @@ def read_config_showcase(self, config_file=None):
         if not config_sanity and samconfig.exists():
             raise GuidedDeployFailedError(msg)
 
-    def save_config(self, parameter_overrides, config_env=DEFAULT_ENV, config_file=None, **kwargs):
+    def save_config(
+        self, parameter_overrides, config_env=DEFAULT_ENV, config_file=None, signing_profiles=None, **kwargs
+    ):
 
         ctx, samconfig = self.get_config_ctx(config_file)
 
@@ -53,16 +55,8 @@ def save_config(self, parameter_overrides, config_env=DEFAULT_ENV, config_file=N
             if value:
                 samconfig.put(cmd_names, self.section, key, value, env=config_env)
 
-        if parameter_overrides:
-            _params = []
-            for key, value in parameter_overrides.items():
-                if isinstance(value, dict):
-                    if not value.get("Hidden"):
-                        _params.append(f"{key}={self.quote_parameter_values(value.get('Value'))}")
-                else:
-                    _params.append(f"{key}={self.quote_parameter_values(value)}")
-            if _params:
-                samconfig.put(cmd_names, self.section, "parameter_overrides", " ".join(_params), env=config_env)
+        self._save_parameter_overrides(cmd_names, config_env, parameter_overrides, samconfig)
+        self._save_signing_profiles(cmd_names, config_env, samconfig, signing_profiles)
 
         samconfig.flush()
 
@@ -75,5 +69,29 @@ def save_config(self, parameter_overrides, config_env=DEFAULT_ENV, config_file=N
             "developerguide/serverless-sam-cli-config.html"
         )
 
+    def _save_signing_profiles(self, cmd_names, config_env, samconfig, signing_profiles):
+        if signing_profiles:
+            _params = []
+            for key, value in signing_profiles.items():
+                if value.get("profile_owner", None):
+                    signing_profile_with_owner = f"{value['profile_name']}:{value['profile_owner']}"
+                    _params.append(f"{key}={self.quote_parameter_values(signing_profile_with_owner)}")
+                else:
+                    _params.append(f"{key}={self.quote_parameter_values(value['profile_name'])}")
+            if _params:
+                samconfig.put(cmd_names, self.section, "signing_profiles", " ".join(_params), env=config_env)
+
+    def _save_parameter_overrides(self, cmd_names, config_env, parameter_overrides, samconfig):
+        if parameter_overrides:
+            _params = []
+            for key, value in parameter_overrides.items():
+                if isinstance(value, dict):
+                    if not value.get("Hidden"):
+                        _params.append(f"{key}={self.quote_parameter_values(value.get('Value'))}")
+                else:
+                    _params.append(f"{key}={self.quote_parameter_values(value)}")
+            if _params:
+                samconfig.put(cmd_names, self.section, "parameter_overrides", " ".join(_params), env=config_env)
+
     def quote_parameter_values(self, parameter_value):
         return '"{}"'.format(parameter_value)