feat: Added environment variable validation for Terraform arguments (#…

…5809)

* Added environment variable validation

* Added integration tests

* Added typing and more detailed doc string

samcli/hook_packages/terraform/copy_terraform_built_artifacts.py

@@ -56,6 +56,16 @@
 LOG = logging.getLogger(__name__)
 
 TF_BACKEND_OVERRIDE_FILENAME = "z_samcli_backend_override"
+TF_BLOCKED_ARGUMENTS = [
+    "-target",
+    "-destroy",
+]
+TF_ENVIRONMENT_VARIABLE_DELIM = "="
+TF_ENVIRONMENT_VARIABLES = [
+    "TF_CLI_ARGS",
+    "TF_CLI_ARGS_plan",
+    "TF_CLI_ARGS_apply",
+]
 
 
 class ResolverException(Exception):
@@ -276,6 +286,31 @@ def create_backend_override():
         cli_exit()
 
 
+def validate_environment_variables():
+    """
+    Validate that the Terraform environment variables do not contain blocked arguments.
+    """
+    for env_var in TF_ENVIRONMENT_VARIABLES:
+        env_value = os.environ.get(env_var, "")
+
+        trimmed_arguments = []
+        # get all trimmed arguments in a list and split on delim
+        # eg.
+        # "-foo=bar -hello" => ["-foo", "-hello"]
+        for argument in env_value.split(" "):
+            cleaned_argument = argument.strip()
+            cleaned_argument = cleaned_argument.split(TF_ENVIRONMENT_VARIABLE_DELIM)[0]
+
+            trimmed_arguments.append(cleaned_argument)
+
+        if any([argument in TF_BLOCKED_ARGUMENTS for argument in trimmed_arguments]):
+            LOG.error(
+                "Environment variable '%s' contains "
+                "a blocked argument, please validate it does not contain: %s" % (env_var, TF_BLOCKED_ARGUMENTS)
+            )
+            cli_exit()
+
+
 if __name__ == "__main__":
     # Gather inputs and clean them
     argparser = argparse.ArgumentParser(
@@ -314,6 +349,9 @@ def create_backend_override():
     target = arguments.target
     json_str = arguments.json
 
+    # validate environment variables do not contain blocked arguments
+    validate_environment_variables()
+
     if target and json_str:
         LOG.error("Provide either --target or --json. Do not provide both.")
         cli_exit()

samcli/hook_packages/terraform/hooks/prepare/hook.py

@@ -12,7 +12,11 @@
 
 from samcli.hook_packages.terraform.hooks.prepare.constants import CFN_CODE_PROPERTIES
 from samcli.hook_packages.terraform.hooks.prepare.translate import translate_to_cfn
-from samcli.lib.hook.exceptions import PrepareHookException, TerraformCloudException
+from samcli.lib.hook.exceptions import (
+    PrepareHookException,
+    TerraformCloudException,
+    UnallowedEnvironmentVariableArgumentException,
+)
 from samcli.lib.utils import osutils
 from samcli.lib.utils.subprocess_utils import LoadingPatternError, invoke_subprocess_with_loading_pattern
 
@@ -32,6 +36,17 @@
     "a plan file using the --terraform-plan-file flag."
 )
 
+TF_BLOCKED_ARGUMENTS = [
+    "-target",
+    "-destroy",
+]
+TF_ENVIRONMENT_VARIABLE_DELIM = "="
+TF_ENVIRONMENT_VARIABLES = [
+    "TF_CLI_ARGS",
+    "TF_CLI_ARGS_plan",
+    "TF_CLI_ARGS_apply",
+]
+
 
 def prepare(params: dict) -> dict:
     """
@@ -55,6 +70,8 @@ def prepare(params: dict) -> dict:
     if not output_dir_path:
         raise PrepareHookException("OutputDirPath was not supplied")
 
+    _validate_environment_variables()
+
     LOG.debug("Normalize the terraform application root module directory path %s", terraform_application_dir)
     if not os.path.isabs(terraform_application_dir):
         terraform_application_dir = os.path.normpath(os.path.join(os.getcwd(), terraform_application_dir))
@@ -215,3 +232,36 @@ def _generate_plan_file(skip_prepare_infra: bool, terraform_application_dir: str
         raise PrepareHookException(f"Error occurred when invoking a process:\n{e}") from e
 
     return dict(json.loads(result.stdout))
+
+
+def _validate_environment_variables() -> None:
+    """
+    Validate that the Terraform environment variables do not contain blocked arguments.
+
+    Raises
+    ------
+    UnallowedEnvironmentVariableArgumentException
+        Raised when a Terraform related environment variable contains a blocked value
+    """
+    for env_var in TF_ENVIRONMENT_VARIABLES:
+        env_value = os.environ.get(env_var, "")
+
+        trimmed_arguments = []
+        # get all trimmed arguments in a list and split on delim
+        # eg.
+        # "-foo=bar -hello" => ["-foo", "-hello"]
+        for argument in env_value.split(" "):
+            cleaned_argument = argument.strip()
+            cleaned_argument = cleaned_argument.split(TF_ENVIRONMENT_VARIABLE_DELIM)[0]
+
+            trimmed_arguments.append(cleaned_argument)
+
+        if any([argument in TF_BLOCKED_ARGUMENTS for argument in trimmed_arguments]):
+            message = (
+                "Environment variable '%s' contains a blocked argument, please validate it does not contain: %s"
+                % (
+                    env_var,
+                    TF_BLOCKED_ARGUMENTS,
+                )
+            )
+            raise UnallowedEnvironmentVariableArgumentException(message)

samcli/lib/hook/exceptions.py

@@ -24,3 +24,7 @@ class PrepareHookException(UserException):
 
 class TerraformCloudException(UserException):
     pass
+
+
+class UnallowedEnvironmentVariableArgumentException(UserException):
+    pass

tests/integration/buildcmd/test_build_terraform_applications_other_cases.py

@@ -548,3 +548,37 @@ def test_build_and_invoke_lambda_functions(self, function_identifier, expected_o
             overrides=None,
             expected_result={"statusCode": 200, "body": expected_output},
         )
+
+
+@skipIf(
+    (not RUN_BY_CANARY and not CI_OVERRIDE),
+    "Skip Terraform test cases unless running in CI",
+)
+class TestBuildTerraformApplicationsWithBlockedEnvironVariables(BuildTerraformApplicationIntegBase):
+    terraform_application = Path("terraform/simple_application")
+
+    @parameterized.expand(
+        [
+            ("TF_CLI_ARGS", "-destroy"),
+            ("TF_CLI_ARGS", "-target=some.module"),
+            ("TF_CLI_ARGS_plan", "-destroy"),
+            ("TF_CLI_ARGS_plan", "-target=some.module"),
+            ("TF_CLI_ARGS_apply", "-destroy"),
+            ("TF_CLI_ARGS_apply", "-target=some.module"),
+        ]
+    )
+    def test_blocked_env_variables(self, env_name, env_value):
+        cmdlist = self.get_command_list(hook_name="terraform", beta_features=True)
+
+        env_variables = os.environ.copy()
+        env_variables[env_name] = env_value
+
+        _, stderr, return_code = self.run_command(cmdlist, env=env_variables)
+
+        process_stderr = stderr.strip()
+        self.assertRegex(
+            process_stderr.decode("utf-8"),
+            "Error: Environment variable '%s' contains a blocked argument, please validate it does not contain: ['-destroy', '-target']"
+            % env_name,
+        )
+        self.assertNotEqual(return_code, 0)

tests/unit/hook_packages/terraform/hooks/prepare/test_hook.py

@@ -6,12 +6,18 @@
 from tests.unit.hook_packages.terraform.hooks.prepare.prepare_base import PrepareHookUnitBase
 
 from samcli.hook_packages.terraform.hooks.prepare.hook import (
+    TF_ENVIRONMENT_VARIABLES,
+    _validate_environment_variables,
     prepare,
     _update_resources_paths,
     TF_CLOUD_EXCEPTION_MESSAGE,
     TF_CLOUD_HELP_MESSAGE,
 )
-from samcli.lib.hook.exceptions import PrepareHookException, TerraformCloudException
+from samcli.lib.hook.exceptions import (
+    PrepareHookException,
+    TerraformCloudException,
+    UnallowedEnvironmentVariableArgumentException,
+)
 from samcli.lib.utils.subprocess_utils import LoadingPatternError
 
 
@@ -492,3 +498,38 @@ def test_prints_tf_cloud_help_message(self, mock_subprocess_loader):
             prepare(self.prepare_params)
 
         self.assertEqual(ex.exception.message, TF_CLOUD_HELP_MESSAGE)
+
+    @parameterized.expand(
+        [
+            ("-destroy",),
+            ("-target=my.module.resource",),
+            ("-destroy -target=my.module.resource",),
+            ("-target=my.module.resource -destroy",),
+        ]
+    )
+    @patch("samcli.hook_packages.terraform.hooks.prepare.hook.os")
+    def test_environment_variable_check_fails(self, argument, get_mock):
+        get_mock.environ.get.return_value = argument
+
+        with self.assertRaises(UnallowedEnvironmentVariableArgumentException):
+            _validate_environment_variables()
+
+    @parameterized.expand(
+        [
+            ("-not-actually-argument",),
+            ("something",),
+            ("",),
+        ]
+    )
+    @patch("samcli.hook_packages.terraform.hooks.prepare.hook.os")
+    def test_environment_variable_check_passes(self, argument, get_mock):
+        get_mock.environ.get.return_value = argument
+
+        _validate_environment_variables()
+
+    @patch("samcli.hook_packages.terraform.hooks.prepare.hook._validate_environment_variables")
+    def test_prepare_method_fails_environment_variables(self, validate_mock):
+        validate_mock.side_effect = [UnallowedEnvironmentVariableArgumentException("message")]
+
+        with self.assertRaises(UnallowedEnvironmentVariableArgumentException):
+            prepare(self.prepare_params)