8744072155

Committed 18 Apr 2024 08:11PM UTC coverage: 59.148% (+0.3%) from 58.859%

Build # 8744072155

Build Type

push

github

Committed by

web-flow

Commit Message

[159] Improve apply arguments parsing logic (#260)

Run Details

226 of 564 branches covered (40.07%)

Branch coverage included in aggregate %.

21 of 26 new or added lines in 1 file covered. (80.77%)

55 existing lines in 1 file now uncovered.

2522 of 4082 relevant lines covered (61.78%)

0.62 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

38.28

/leverage/modules/terraform.py

import os
import re

import click
import dockerpty
import hcl2
from click.exceptions import Exit

from leverage import logger
from leverage._internals import pass_container
from leverage._internals import pass_state
from leverage._utils import tar_directory, AwsCredsContainer, LiveContainer, ExitError
from leverage.container import TerraformContainer
from leverage.container import get_docker_client
from leverage.modules.utils import env_var_option, mount_option, auth_mfa, auth_sso

REGION = (
    r"global|(?:[a-z]{2}-(?:gov-)?"
    r"(?:central|north|south|east|west|northeast|northwest|southeast|southwest|secret|topsecret)-[1-4])"
)


# ###########################################################################
# CREATE THE TERRAFORM GROUP
# ###########################################################################
@click.group()
@mount_option
@env_var_option
@pass_state
def terraform(state, env_var, mount):
    """Run Terraform commands in a custom containerized environment that provides extra functionality when interacting
    with your cloud provider such as handling multi factor authentication for you.
    All terraform subcommands that receive extra args will pass the given strings as is to their corresponding Terraform
    counterparts in the container. For example as in `leverage terraform apply -auto-approve` or
    `leverage terraform init -reconfigure`
    """
    if env_var:
        env_var = dict(env_var)

    state.container = TerraformContainer(get_docker_client(), mounts=mount, env_vars=env_var)
    state.container.ensure_image()


CONTEXT_SETTINGS = {"ignore_unknown_options": True}

# ###########################################################################
# CREATE THE TERRAFORM GROUP'S COMMANDS
# ###########################################################################
#
# --layers is a ordered comma separated list of layer names
# The layer names are the relative paths of those layers relative to the current directory
# e.g. if CLI is called from /home/user/project/management and this is the tree:
# home
# ├── user
# │   └── project
# │       └── management
# │           ├── global
# │           |   └── security-base
# │           |   └── sso
# │           └── us-east-1
# │               └── terraform-backend
#
# Then all three layers can be initialized as follows:
# leverage tf init --layers us-east-1/terraform-backend,global/security-base,global/sso
#
# It is an ordered list because the layers will be visited in the same order they were
# supplied.
#
layers_option = click.option(
    "--layers",
    type=str,
    default="",
    help="Layers to apply the action to. (an ordered, comma-separated list of layer names)",
)


@terraform.command(context_settings=CONTEXT_SETTINGS)
@click.option("--skip-validation", is_flag=True, help="Skip layout validation.")
@layers_option
@click.argument("args", nargs=-1)
@pass_container
@click.pass_context
def init(context, tf: TerraformContainer, skip_validation, layers, args):
    """
    Initialize this layer.
    """
    layers = invoke_for_all_commands(layers, _init, args, skip_validation)

    # now change ownership on all the downloaded modules and providers
    for layer in layers:
        tf.change_file_ownership(tf.paths.guest_base_path / layer.relative_to(tf.paths.root_dir) / ".terraform")
    # and then providers in the cache folder
    if tf.paths.tf_cache_dir:
        tf.change_file_ownership(tf.paths.tf_cache_dir)


@terraform.command(context_settings=CONTEXT_SETTINGS)
@layers_option
@click.argument("args", nargs=-1)
@pass_container
@click.pass_context
def plan(context, tf, layers, args):
    """Generate an execution plan for this layer."""
    invoke_for_all_commands(layers, _plan, args)


@terraform.command(context_settings=CONTEXT_SETTINGS)
@layers_option
@click.argument("args", nargs=-1)
@pass_container
@click.pass_context
def apply(context, tf, layers, args):
    """Build or change the infrastructure in this layer."""
    invoke_for_all_commands(layers, _apply, args)


@terraform.command(context_settings=CONTEXT_SETTINGS)
@layers_option
@click.argument("args", nargs=-1)
@pass_container
@click.pass_context
def output(context, tf, layers, args):
    """Show all output variables of this layer."""
    invoke_for_all_commands(layers, _output, args)


@terraform.command(context_settings=CONTEXT_SETTINGS)
@layers_option
@click.argument("args", nargs=-1)
@pass_container
@click.pass_context
def destroy(context, tf, layers, args):
    """Destroy infrastructure in this layer."""
    invoke_for_all_commands(layers, _destroy, args)


@terraform.command()
@pass_container
def version(tf):
    """Print version."""
    tf.disable_authentication()
    tf.start("version")


@terraform.command()
@auth_mfa
@auth_sso
@pass_container
def shell(tf, mfa, sso):
    """Open a shell into the Terraform container in this layer."""
    tf.disable_authentication()
    if sso:
        tf.enable_sso()

    if mfa:
        tf.enable_mfa()

    tf.start_shell()


@terraform.command("format", context_settings=CONTEXT_SETTINGS)
@click.argument("args", nargs=-1)
@pass_container
def _format(tf, args):
    """Check if all files meet the canonical format and rewrite them accordingly."""
    args = args if "-recursive" in args else (*args, "-recursive")
    tf.disable_authentication()
    tf.start("fmt", *args)


@terraform.command()
@pass_container
def validate(tf):
    """Validate code of the current directory. Previous initialization might be needed."""
    tf.disable_authentication()
    tf.start("validate")


@terraform.command("validate-layout")
@pass_container
def validate_layout(tf):
    """Validate layer conforms to Leverage convention."""
    tf.set_backend_key()
    return _validate_layout()


@terraform.command("import")
@click.argument("address")
@click.argument("_id", metavar="ID")
@pass_container
def _import(tf, address, _id):
    """Import a resource."""
    exit_code = tf.start_in_layer("import", *tf.tf_default_args, address, _id)

    if exit_code:
        raise Exit(exit_code)


@terraform.command("refresh-credentials")
@pass_container
def refresh_credentials(tf):
    """Refresh the AWS credentials used on the current layer."""
    tf.paths.check_for_layer_location()
    if exit_code := tf.refresh_credentials():
        raise Exit(exit_code)


# ###########################################################################
# HANDLER FOR MANAGING THE BASE COMMANDS (init, plan, apply, destroy, output)
# ###########################################################################
@pass_container
def invoke_for_all_commands(tf, layers, command, args, skip_validation=True):
    """
    Invoke helper for "all" commands.

    Args:
        layers: comma separated value of relative layer path
            e.g.: global/security_audit,us-east-1/tf-backend
        command: init, plan, apply
    """

    # convert layers from string to list
    layers = layers.split(",") if len(layers) > 0 else []

    # based on the location type manage the layers parameter
    location_type = tf.paths.get_location_type()
    if location_type == "layer" and len(layers) == 0:
        # running on a layer
        layers = [tf.paths.cwd]
    elif location_type == "layer":
        # running on a layer but --layers was set
        raise ExitError(1, "Can not set [bold]--layers[/bold] inside a layer.")
    elif location_type in ["account", "layers-group"] and len(layers) == 0:
        # running on an account but --layers was not set
        raise ExitError(1, "[bold]--layers[/bold] has to be set.")
    elif location_type not in ["account", "layer", "layers-group"]:
        # running outside a layer and account
        raise ExitError(1, "This command has to be run inside a layer or account directory.")
    else:
        # running on an account with --layers set
        layers = [tf.paths.cwd / x for x in layers]

    # get current location
    original_location = tf.paths.cwd
    original_working_dir = tf.container_config["working_dir"]

    # validate each layer before calling the execute command
    for layer in layers:
        logger.debug(f"Checking for layer {layer}...")
        # change to current dir and set it in the container
        tf.paths.cwd = layer

        # check layers existence
        if not layer.is_dir():
            logger.error(f"Directory [red]{layer}[/red] does not exist or is not a directory\n")
            raise Exit(1)

        # set the s3 key
        tf.set_backend_key(skip_validation)

        # validate layer
        validate_for_all_commands(layer, skip_validation=skip_validation)

        # change to original dir and set it in the container
        tf.paths.cwd = original_location

    # check layers existence
    for layer in layers:
        if len(layers) > 1:
            logger.info(f"Invoking command for layer {layer}...")

        # change to current dir and set it in the container
        tf.paths.cwd = layer

        # set the working dir
        working_dir = f"{tf.paths.guest_base_path}/{tf.paths.cwd.relative_to(tf.paths.root_dir).as_posix()}"
        tf.container_config["working_dir"] = working_dir

        # execute the actual command
        command(args=args)

        # change to original dir and set it in the container
        tf.paths.cwd = original_location

        # change to original workgindir
        tf.container_config["working_dir"] = original_working_dir

    return layers


def validate_for_all_commands(layer, skip_validation=False):
    """
    Validate existence of layer and, if set, all the Leverage related stuff
    of each of them

    Args:
        layer: a full layer directory
    """

    logger.debug(f"Checking layer {layer}...")
    if not skip_validation and not _validate_layout():
        logger.error(
            "Layer configuration doesn't seem to be valid. Exiting.\n"
            "If you are sure your configuration is actually correct "
            "you may skip this validation using the --skip-validation flag."
        )
        raise Exit(1)


# ###########################################################################
# BASE COMMAND EXECUTORS
# ###########################################################################
@pass_container
def _init(tf, args):
    """Initialize this layer."""

    args = [
        arg
        for index, arg in enumerate(args)
        if not arg.startswith("-backend-config") or not arg[index - 1] == "-backend-config"
    ]
    args.append(f"-backend-config={tf.paths.backend_tfvars}")

    tf.paths.check_for_layer_location()

    with LiveContainer(tf) as container:
        # create the .ssh directory
        container.exec_run("mkdir -p /root/.ssh")
        # copy the entire ~/.ssh/ folder
        tar_bytes = tar_directory(tf.paths.home / ".ssh")
        # into /root/.ssh
        container.put_archive("/root/.ssh/", tar_bytes)
        # correct the owner of the files to match with the docker internal user
        container.exec_run("chown root:root -R /root/.ssh/")

        with AwsCredsContainer(container, tf):
            dockerpty.exec_command(
                client=tf.client.api,
                container=container.id,
                command="terraform init " + " ".join(args),
                interactive=bool(int(os.environ.get("LEVERAGE_INTERACTIVE", 1))),
            )


@pass_container
def _plan(tf, args):
    """Generate an execution plan for this layer."""
    exit_code = tf.start_in_layer("plan", *tf.tf_default_args, *args)

    if exit_code:
        raise Exit(exit_code)


def handle_apply_arguments_parsing(args):
    """Parse and process the arguments for the 'apply' command."""
    # Initialize new_args to handle both '-key=value' and '-key value'
    new_args = []
    skip_next = False  # Flag to skip the next argument if it's part of '-key value'

    for i, arg in enumerate(args):
        if skip_next:
            skip_next = False  # Reset flag and skip this iteration
            continue

        if arg.startswith("-") and not arg.startswith("-var"):
            if i + 1 < len(args) and not args[i + 1].startswith("-"):
                # Detected '-key value' pair; append them without merging
                new_args.append(arg)
                new_args.append(args[i + 1])
                skip_next = True  # Mark to skip the next item as it's already processed
                logger.debug(f"Detected '-key value' pair: {arg}, {args[i + 1]}")
            else:
                # Either '-key=value' or a standalone '-key'; just append
                new_args.append(arg)
                logger.debug(f"Appending standard -key=value or standalone argument: {arg}")
        else:
            # Handles '-var' and non '-' starting arguments
            new_args.append(arg)
            logger.debug(f"Appending argument (non '-' or '-var'): {arg}")

    return new_args


@pass_container
def _apply(tf, args):
    """Build or change the infrastructure in this layer."""
    # if there is a plan, remove all "-var" from the default args
    # Preserve the original `-var` removal logic and modify tf_default_args if necessary
    tf_default_args = tf.tf_default_args
    for arg in args:
        if not arg.startswith("-"):
            tf_default_args = [arg for index, arg in enumerate(tf_default_args) if not arg.startswith("-var")]
            break

    # Process arguments using the new parsing logic
    processed_args = handle_apply_arguments_parsing(args)

    logger.debug(f"Original tf_default_args: {tf_default_args}")
    logger.debug(f"Processed argument list for execution: {processed_args}")

    # Execute the command with the modified arguments list
    exit_code = tf.start_in_layer("apply", *tf_default_args, *processed_args)

    if exit_code:
        logger.error(f"Command execution failed with exit code: {exit_code}")
        raise Exit(exit_code)


@pass_container
def _output(tf, args):
    """Show all output variables of this layer."""
    tf.start_in_layer("output", *args)


@pass_container
def _destroy(tf, args):
    """Destroy infrastructure in this layer."""
    exit_code = tf.start_in_layer("destroy", *tf.tf_default_args, *args)

    if exit_code:
        raise Exit(exit_code)


# ###########################################################################
# MISC FUNCTIONS
# ###########################################################################
def _make_layer_backend_key(cwd, account_dir, account_name):
    """Create expected backend key.

    Args:
        cwd (pathlib.Path): Current Working Directory (Layer Directory)
        account_dir (pathlib.Path): Account Directory
        account_name (str): Account Name

    Returns:
        list of lists: Backend bucket key parts
    """
    resp = []

    layer_path = cwd.relative_to(account_dir)
    layer_path = layer_path.as_posix().split("/")
    # Check region directory to keep retro compat
    if re.match(REGION, layer_path[0]):
        layer_paths = [layer_path[1:], layer_path]
    else:
        layer_paths = [layer_path]

    curated_layer_paths = []
    # Remove layer name prefix
    for layer_path in layer_paths:
        curated_layer_path = []
        for lp in layer_path:
            if lp.startswith("base-"):
                lp = lp.replace("base-", "")
            elif lp.startswith("tools-"):
                lp = lp.replace("tools-", "")
            curated_layer_path.append(lp)
        curated_layer_paths.append(curated_layer_path)

    curated_layer_paths_retrocomp = []
    for layer_path in curated_layer_paths:
        curated_layer_paths_retrocomp.append(layer_path)
        # check for tf/terraform variants
        for idx, lp in enumerate(layer_path):
            if lp.startswith("tf-"):
                layer_path_tmp = layer_path.copy()
                layer_path_tmp[idx] = layer_path_tmp[idx].replace("tf-", "terraform-")
                curated_layer_paths_retrocomp.append(layer_path_tmp)
                break
            elif lp.startswith("terraform-"):
                layer_path_tmp = layer_path.copy()
                layer_path_tmp[idx] = layer_path_tmp[idx].replace("terraform-", "tf-")
                curated_layer_paths_retrocomp.append(layer_path_tmp)
                break

    curated_layer_paths_withDR = []
    for layer_path in curated_layer_paths_retrocomp:
        curated_layer_paths_withDR.append(layer_path)
        curated_layer_path = []
        append_str = "-dr"
        for lp in layer_path:
            if re.match(REGION, lp):
                curated_layer_path.append(lp)
            else:
                curated_layer_path.append(lp + append_str)
                append_str = ""
        curated_layer_paths_withDR.append(curated_layer_path)

    for layer_path in curated_layer_paths_withDR:
        resp.append([account_name, *layer_path])

    return resp


@pass_container
def _validate_layout(tf: TerraformContainer):
    tf.paths.check_for_layer_location()

    # Check for `environment = <account name>` in account.tfvars
    account_name = tf.paths.account_conf.get("environment")
    logger.info("Checking environment name definition in [bold]account.tfvars[/bold]...")
    if account_name is None:
        logger.error("[red]✘ FAILED[/red]\n")
        raise Exit(1)
    logger.info("[green]✔ OK[/green]\n")

    # Check if account directory name matches with environment name
    if tf.paths.account_dir.stem != account_name:
        logger.warning(
            "[yellow]‼[/yellow] Account directory name does not match environment name.\n"
            f"  Expected [bold]{account_name}[/bold], found [bold]{tf.paths.account_dir.stem}[/bold]\n"
        )

    backend_key = tf.backend_key.split("/")

    # Flag to report layout validity
    valid_layout = True

    # Check backend bucket key
    expected_backend_keys = _make_layer_backend_key(tf.paths.cwd, tf.paths.account_dir, account_name)
    logger.info("Checking backend key...")
    logger.info(f"Found: '{'/'.join(backend_key)}'")
    backend_key = backend_key[:-1]

    if backend_key in expected_backend_keys:
        logger.info("[green]✔ OK[/green]\n")
    else:
        exp_message = [f"{'/'.join(x)}/terraform.tfstate" for x in expected_backend_keys]
        logger.info(f"Expected one of: {';'.join(exp_message)}")
        logger.error("[red]✘ FAILED[/red]\n")
        valid_layout = False

    backend_tfvars = tf.paths.account_config_dir / tf.paths.BACKEND_TF_VARS  # TODO use paths.backend_tfvars instead?
    backend_tfvars = hcl2.loads(backend_tfvars.read_text()) if backend_tfvars.exists() else {}

    logger.info("Checking [bold]backend.tfvars[/bold]:\n")
    names_prefix = f"{tf.project}-{account_name}"
    names_prefix_bootstrap = f"{tf.project}-bootstrap"

    # Check profile, bucket and dynamo table names:
    for field in ("profile", "bucket", "dynamodb_table"):
        logger.info(f"Checking if {field.replace('_', ' ')} starts with {names_prefix}...")
        if backend_tfvars.get(field, "").startswith(names_prefix) or (
            field == "profile" and backend_tfvars.get(field, "").startswith(names_prefix_bootstrap)
        ):
            logger.info("[green]✔ OK[/green]\n")
        else:
            logger.error("[red]✘ FAILED[/red]\n")
            valid_layout = False

    return valid_layout

1	import os	1✔
2	import re	1✔
3
4	import click	1✔
5	import dockerpty	1✔
6	import hcl2	1✔
7	from click.exceptions import Exit	1✔
8
9	from leverage import logger	1✔
10	from leverage._internals import pass_container	1✔
11	from leverage._internals import pass_state	1✔
12	from leverage._utils import tar_directory, AwsCredsContainer, LiveContainer, ExitError	1✔
13	from leverage.container import TerraformContainer	1✔
14	from leverage.container import get_docker_client	1✔
15	from leverage.modules.utils import env_var_option, mount_option, auth_mfa, auth_sso	1✔
16
17	REGION = (	1✔
18	r"global\|(?:[a-z]{2}-(?:gov-)?"
19	r"(?:central\|north\|south\|east\|west\|northeast\|northwest\|southeast\|southwest\|secret\|topsecret)-[1-4])"
20	)
21
22
23	# ###########################################################################
24	# CREATE THE TERRAFORM GROUP
25	# ###########################################################################
26	@click.group()	1✔
27	@mount_option	1✔
28	@env_var_option	1✔
29	@pass_state	1✔
30	def terraform(state, env_var, mount):	1✔
31	"""Run Terraform commands in a custom containerized environment that provides extra functionality when interacting
32	with your cloud provider such as handling multi factor authentication for you.
33	All terraform subcommands that receive extra args will pass the given strings as is to their corresponding Terraform
34	counterparts in the container. For example as in `leverage terraform apply -auto-approve` or
35	`leverage terraform init -reconfigure`
36	"""
37	if env_var:	×
38	env_var = dict(env_var)	×
39
40	state.container = TerraformContainer(get_docker_client(), mounts=mount, env_vars=env_var)	×
41	state.container.ensure_image()	×
42
43
44	CONTEXT_SETTINGS = {"ignore_unknown_options": True}	1✔
45
46	# ###########################################################################
47	# CREATE THE TERRAFORM GROUP'S COMMANDS
48	# ###########################################################################
49	#
50	# --layers is a ordered comma separated list of layer names
51	# The layer names are the relative paths of those layers relative to the current directory
52	# e.g. if CLI is called from /home/user/project/management and this is the tree:
53	# home
54	# ├── user
55	# │ └── project
56	# │ └── management
57	# │ ├── global
58	# │ \| └── security-base
59	# │ \| └── sso
60	# │ └── us-east-1
61	# │ └── terraform-backend
62	#
63	# Then all three layers can be initialized as follows:
64	# leverage tf init --layers us-east-1/terraform-backend,global/security-base,global/sso
65	#
66	# It is an ordered list because the layers will be visited in the same order they were
67	# supplied.
68	#
69	layers_option = click.option(	1✔
70	"--layers",
71	type=str,
72	default="",
73	help="Layers to apply the action to. (an ordered, comma-separated list of layer names)",
74	)
75
76
77	@terraform.command(context_settings=CONTEXT_SETTINGS)	1✔
78	@click.option("--skip-validation", is_flag=True, help="Skip layout validation.")	1✔
79	@layers_option	1✔
80	@click.argument("args", nargs=-1)	1✔
81	@pass_container	1✔
82	@click.pass_context	1✔
83	def init(context, tf: TerraformContainer, skip_validation, layers, args):	1✔
84	"""
85	Initialize this layer.
86	"""
87	layers = invoke_for_all_commands(layers, _init, args, skip_validation)	×
88
89	# now change ownership on all the downloaded modules and providers
90	for layer in layers:	×
91	tf.change_file_ownership(tf.paths.guest_base_path / layer.relative_to(tf.paths.root_dir) / ".terraform")	×
92	# and then providers in the cache folder
93	if tf.paths.tf_cache_dir:	×
94	tf.change_file_ownership(tf.paths.tf_cache_dir)	×
95
96
97	@terraform.command(context_settings=CONTEXT_SETTINGS)	1✔
98	@layers_option	1✔
99	@click.argument("args", nargs=-1)	1✔
100	@pass_container	1✔
101	@click.pass_context	1✔
102	def plan(context, tf, layers, args):	1✔
103	"""Generate an execution plan for this layer."""
104	invoke_for_all_commands(layers, _plan, args)	×
105
106
107	@terraform.command(context_settings=CONTEXT_SETTINGS)	1✔
108	@layers_option	1✔
109	@click.argument("args", nargs=-1)	1✔
110	@pass_container	1✔
111	@click.pass_context	1✔
112	def apply(context, tf, layers, args):	1✔
113	"""Build or change the infrastructure in this layer."""
114	invoke_for_all_commands(layers, _apply, args)	×
115
116
117	@terraform.command(context_settings=CONTEXT_SETTINGS)	1✔
118	@layers_option	1✔
119	@click.argument("args", nargs=-1)	1✔
120	@pass_container	1✔
121	@click.pass_context	1✔
122	def output(context, tf, layers, args):	1✔
123	"""Show all output variables of this layer."""
124	invoke_for_all_commands(layers, _output, args)	×
125
126
127	@terraform.command(context_settings=CONTEXT_SETTINGS)	1✔
128	@layers_option	1✔
129	@click.argument("args", nargs=-1)	1✔
130	@pass_container	1✔
131	@click.pass_context	1✔
132	def destroy(context, tf, layers, args):	1✔
133	"""Destroy infrastructure in this layer."""
134	invoke_for_all_commands(layers, _destroy, args)	×
135
136
137	@terraform.command()	1✔
138	@pass_container	1✔
139	def version(tf):	1✔
140	"""Print version."""
141	tf.disable_authentication()	×
142	tf.start("version")	×
143
144
145	@terraform.command()	1✔
146	@auth_mfa	1✔
147	@auth_sso	1✔
148	@pass_container	1✔
149	def shell(tf, mfa, sso):	1✔
150	"""Open a shell into the Terraform container in this layer."""
151	tf.disable_authentication()	×
152	if sso:	×
153	tf.enable_sso()	×
154
155	if mfa:	×
156	tf.enable_mfa()	×
157
158	tf.start_shell()	×
159
160
161	@terraform.command("format", context_settings=CONTEXT_SETTINGS)	1✔
162	@click.argument("args", nargs=-1)	1✔
163	@pass_container	1✔
164	def _format(tf, args):	1✔
165	"""Check if all files meet the canonical format and rewrite them accordingly."""
166	args = args if "-recursive" in args else (*args, "-recursive")	×
167	tf.disable_authentication()	×
168	tf.start("fmt", *args)	×
169
170
171	@terraform.command()	1✔
172	@pass_container	1✔
173	def validate(tf):	1✔
174	"""Validate code of the current directory. Previous initialization might be needed."""
175	tf.disable_authentication()	×
176	tf.start("validate")	×
177
178
179	@terraform.command("validate-layout")	1✔
180	@pass_container	1✔
181	def validate_layout(tf):	1✔
182	"""Validate layer conforms to Leverage convention."""
183	tf.set_backend_key()	×
184	return _validate_layout()	×
185
186
187	@terraform.command("import")	1✔
188	@click.argument("address")	1✔
189	@click.argument("_id", metavar="ID")	1✔
190	@pass_container	1✔
191	def _import(tf, address, _id):	1✔
192	"""Import a resource."""
193	exit_code = tf.start_in_layer("import", *tf.tf_default_args, address, _id)	×
194
195	if exit_code:	×
196	raise Exit(exit_code)	×
197
198
199	@terraform.command("refresh-credentials")	1✔
200	@pass_container	1✔
201	def refresh_credentials(tf):	1✔
202	"""Refresh the AWS credentials used on the current layer."""
203	tf.paths.check_for_layer_location()	×
204	if exit_code := tf.refresh_credentials():	×
205	raise Exit(exit_code)	×
206
207
208	# ###########################################################################
209	# HANDLER FOR MANAGING THE BASE COMMANDS (init, plan, apply, destroy, output)
210	# ###########################################################################
211	@pass_container	1✔
212	def invoke_for_all_commands(tf, layers, command, args, skip_validation=True):	1✔
213	"""
214	Invoke helper for "all" commands.
215
216	Args:
217	layers: comma separated value of relative layer path
218	e.g.: global/security_audit,us-east-1/tf-backend
219	command: init, plan, apply
220	"""
221
222	# convert layers from string to list
223	layers = layers.split(",") if len(layers) > 0 else []	×
224
225	# based on the location type manage the layers parameter
226	location_type = tf.paths.get_location_type()	×
227	if location_type == "layer" and len(layers) == 0:	×
228	# running on a layer
229	layers = [tf.paths.cwd]	×
230	elif location_type == "layer":	×
231	# running on a layer but --layers was set
232	raise ExitError(1, "Can not set [bold]--layers[/bold] inside a layer.")	×
233	elif location_type in ["account", "layers-group"] and len(layers) == 0:	×
234	# running on an account but --layers was not set
235	raise ExitError(1, "[bold]--layers[/bold] has to be set.")	×
236	elif location_type not in ["account", "layer", "layers-group"]:	×
237	# running outside a layer and account
238	raise ExitError(1, "This command has to be run inside a layer or account directory.")	×
239	else:
240	# running on an account with --layers set
241	layers = [tf.paths.cwd / x for x in layers]	×
242
243	# get current location
244	original_location = tf.paths.cwd	×
245	original_working_dir = tf.container_config["working_dir"]	×
246
247	# validate each layer before calling the execute command
248	for layer in layers:	×
249	logger.debug(f"Checking for layer {layer}...")	×
250	# change to current dir and set it in the container
251	tf.paths.cwd = layer	×
252
253	# check layers existence
254	if not layer.is_dir():	×
255	logger.error(f"Directory [red]{layer}[/red] does not exist or is not a directory\n")	×
256	raise Exit(1)	×
257
258	# set the s3 key
259	tf.set_backend_key(skip_validation)	×
260
261	# validate layer
262	validate_for_all_commands(layer, skip_validation=skip_validation)	×
263
264	# change to original dir and set it in the container
265	tf.paths.cwd = original_location	×
266
267	# check layers existence
268	for layer in layers:	×
269	if len(layers) > 1:	×
270	logger.info(f"Invoking command for layer {layer}...")	×
271
272	# change to current dir and set it in the container
273	tf.paths.cwd = layer	×
274
275	# set the working dir
276	working_dir = f"{tf.paths.guest_base_path}/{tf.paths.cwd.relative_to(tf.paths.root_dir).as_posix()}"	×
277	tf.container_config["working_dir"] = working_dir	×
278
279	# execute the actual command
280	command(args=args)	×
281
282	# change to original dir and set it in the container
283	tf.paths.cwd = original_location	×
284
285	# change to original workgindir
286	tf.container_config["working_dir"] = original_working_dir	×
287
288	return layers	×
289
290
291	def validate_for_all_commands(layer, skip_validation=False):	1✔
292	"""
293	Validate existence of layer and, if set, all the Leverage related stuff
294	of each of them
295
296	Args:
297	layer: a full layer directory
298	"""
299
300	logger.debug(f"Checking layer {layer}...")	×
301	if not skip_validation and not _validate_layout():	×
302	logger.error(	×
303	"Layer configuration doesn't seem to be valid. Exiting.\n"
304	"If you are sure your configuration is actually correct "
305	"you may skip this validation using the --skip-validation flag."
306	)
307	raise Exit(1)	×
308
309
310	# ###########################################################################
311	# BASE COMMAND EXECUTORS
312	# ###########################################################################
313	@pass_container	1✔
314	def _init(tf, args):	1✔
315	"""Initialize this layer."""
316
317	args = [	1✔
318	arg
319	for index, arg in enumerate(args)
320	if not arg.startswith("-backend-config") or not arg[index - 1] == "-backend-config"
321	]
322	args.append(f"-backend-config={tf.paths.backend_tfvars}")	1✔
323
324	tf.paths.check_for_layer_location()	1✔
325
326	with LiveContainer(tf) as container:	1✔
327	# create the .ssh directory
328	container.exec_run("mkdir -p /root/.ssh")	1✔
329	# copy the entire ~/.ssh/ folder
330	tar_bytes = tar_directory(tf.paths.home / ".ssh")	1✔
331	# into /root/.ssh
332	container.put_archive("/root/.ssh/", tar_bytes)	1✔
333	# correct the owner of the files to match with the docker internal user
334	container.exec_run("chown root:root -R /root/.ssh/")	1✔
335
336	with AwsCredsContainer(container, tf):	1✔
337	dockerpty.exec_command(	1✔
338	client=tf.client.api,
339	container=container.id,
340	command="terraform init " + " ".join(args),
341	interactive=bool(int(os.environ.get("LEVERAGE_INTERACTIVE", 1))),
342	)
343
344
345	@pass_container	1✔
346	def _plan(tf, args):	1✔
347	"""Generate an execution plan for this layer."""
348	exit_code = tf.start_in_layer("plan", tf.tf_default_args, args)	×
349
350	if exit_code:	×
351	raise Exit(exit_code)	×
352
353
354	def handle_apply_arguments_parsing(args):	1✔
355	"""Parse and process the arguments for the 'apply' command."""
356	# Initialize new_args to handle both '-key=value' and '-key value'
357	new_args = []	1✔
358	skip_next = False # Flag to skip the next argument if it's part of '-key value'	1✔
359
360	for i, arg in enumerate(args):	1✔
361	if skip_next:	1✔
362	skip_next = False # Reset flag and skip this iteration	1✔
363	continue	1✔
364
365	if arg.startswith("-") and not arg.startswith("-var"):	1✔
366	if i + 1 < len(args) and not args[i + 1].startswith("-"):	1✔
367	# Detected '-key value' pair; append them without merging
368	new_args.append(arg)	1✔
369	new_args.append(args[i + 1])	1✔
370	skip_next = True # Mark to skip the next item as it's already processed	1✔
371	logger.debug(f"Detected '-key value' pair: {arg}, {args[i + 1]}")	1✔
372	else:
373	# Either '-key=value' or a standalone '-key'; just append
374	new_args.append(arg)	1✔
375	logger.debug(f"Appending standard -key=value or standalone argument: {arg}")	1✔
376	else:
377	# Handles '-var' and non '-' starting arguments
378	new_args.append(arg)	1✔
379	logger.debug(f"Appending argument (non '-' or '-var'): {arg}")	1✔
380
381	return new_args	1✔
382
383
384	@pass_container	1✔
385	def _apply(tf, args):	1✔
386	"""Build or change the infrastructure in this layer."""
387	# if there is a plan, remove all "-var" from the default args
388	# Preserve the original `-var` removal logic and modify tf_default_args if necessary
389	tf_default_args = tf.tf_default_args	×
390	for arg in args:	×
391	if not arg.startswith("-"):	×
392	tf_default_args = [arg for index, arg in enumerate(tf_default_args) if not arg.startswith("-var")]	×
393	break	×
394
395	# Process arguments using the new parsing logic
NEW 396	processed_args = handle_apply_arguments_parsing(args)	×
397
NEW 398	logger.debug(f"Original tf_default_args: {tf_default_args}")	×
NEW 399	logger.debug(f"Processed argument list for execution: {processed_args}")	×
400
401	# Execute the command with the modified arguments list
NEW 402	exit_code = tf.start_in_layer("apply", tf_default_args, processed_args)	×
403
404	if exit_code:	×
NEW 405	logger.error(f"Command execution failed with exit code: {exit_code}")	×
406	raise Exit(exit_code)	×
407
408
409	@pass_container	1✔
410	def _output(tf, args):	1✔
411	"""Show all output variables of this layer."""
412	tf.start_in_layer("output", *args)	×
413
414
415	@pass_container	1✔
416	def _destroy(tf, args):	1✔
417	"""Destroy infrastructure in this layer."""
418	exit_code = tf.start_in_layer("destroy", tf.tf_default_args, args)	×
419
420	if exit_code:	×
421	raise Exit(exit_code)	×
422
423
424	# ###########################################################################
425	# MISC FUNCTIONS
426	# ###########################################################################
427	def _make_layer_backend_key(cwd, account_dir, account_name):	1✔
428	"""Create expected backend key.
429
430	Args:
431	cwd (pathlib.Path): Current Working Directory (Layer Directory)
432	account_dir (pathlib.Path): Account Directory
433	account_name (str): Account Name
434
435	Returns:
436	list of lists: Backend bucket key parts
437	"""
438	resp = []	×
439
440	layer_path = cwd.relative_to(account_dir)	×
441	layer_path = layer_path.as_posix().split("/")	×
442	# Check region directory to keep retro compat
443	if re.match(REGION, layer_path[0]):	×
444	layer_paths = [layer_path[1:], layer_path]	×
445	else:
446	layer_paths = [layer_path]	×
447
448	curated_layer_paths = []	×
449	# Remove layer name prefix
450	for layer_path in layer_paths:	×
451	curated_layer_path = []	×
452	for lp in layer_path:	×
453	if lp.startswith("base-"):	×
454	lp = lp.replace("base-", "")	×
455	elif lp.startswith("tools-"):	×
456	lp = lp.replace("tools-", "")	×
457	curated_layer_path.append(lp)	×
458	curated_layer_paths.append(curated_layer_path)	×
459
460	curated_layer_paths_retrocomp = []	×
461	for layer_path in curated_layer_paths:	×
462	curated_layer_paths_retrocomp.append(layer_path)	×
463	# check for tf/terraform variants
464	for idx, lp in enumerate(layer_path):	×
465	if lp.startswith("tf-"):	×
466	layer_path_tmp = layer_path.copy()	×
467	layer_path_tmp[idx] = layer_path_tmp[idx].replace("tf-", "terraform-")	×
468	curated_layer_paths_retrocomp.append(layer_path_tmp)	×
469	break	×
470	elif lp.startswith("terraform-"):	×
471	layer_path_tmp = layer_path.copy()	×
472	layer_path_tmp[idx] = layer_path_tmp[idx].replace("terraform-", "tf-")	×
473	curated_layer_paths_retrocomp.append(layer_path_tmp)	×
474	break	×
475
476	curated_layer_paths_withDR = []	×
477	for layer_path in curated_layer_paths_retrocomp:	×
478	curated_layer_paths_withDR.append(layer_path)	×
479	curated_layer_path = []	×
480	append_str = "-dr"	×
481	for lp in layer_path:	×
482	if re.match(REGION, lp):	×
483	curated_layer_path.append(lp)	×
484	else:
485	curated_layer_path.append(lp + append_str)	×
486	append_str = ""	×
487	curated_layer_paths_withDR.append(curated_layer_path)	×
488
489	for layer_path in curated_layer_paths_withDR:	×
490	resp.append([account_name, *layer_path])	×
491
492	return resp	×
493
494
495	@pass_container	1✔
496	def _validate_layout(tf: TerraformContainer):	1✔
497	tf.paths.check_for_layer_location()	×
498
499	# Check for `environment = <account name>` in account.tfvars
500	account_name = tf.paths.account_conf.get("environment")	×
501	logger.info("Checking environment name definition in [bold]account.tfvars[/bold]...")	×
502	if account_name is None:	×
503	logger.error("[red]✘ FAILED[/red]\n")	×
504	raise Exit(1)	×
505	logger.info("[green]✔ OK[/green]\n")	×
506
507	# Check if account directory name matches with environment name
508	if tf.paths.account_dir.stem != account_name:	×
509	logger.warning(	×
510	"[yellow]‼[/yellow] Account directory name does not match environment name.\n"
511	f" Expected [bold]{account_name}[/bold], found [bold]{tf.paths.account_dir.stem}[/bold]\n"
512	)
513
514	backend_key = tf.backend_key.split("/")	×
515
516	# Flag to report layout validity
517	valid_layout = True	×
518
519	# Check backend bucket key
520	expected_backend_keys = _make_layer_backend_key(tf.paths.cwd, tf.paths.account_dir, account_name)	×
521	logger.info("Checking backend key...")	×
522	logger.info(f"Found: '{'/'.join(backend_key)}'")	×
523	backend_key = backend_key[:-1]	×
524
525	if backend_key in expected_backend_keys:	×
526	logger.info("[green]✔ OK[/green]\n")	×
527	else:
528	exp_message = [f"{'/'.join(x)}/terraform.tfstate" for x in expected_backend_keys]	×
529	logger.info(f"Expected one of: {';'.join(exp_message)}")	×
530	logger.error("[red]✘ FAILED[/red]\n")	×
531	valid_layout = False	×
532
533	backend_tfvars = tf.paths.account_config_dir / tf.paths.BACKEND_TF_VARS # TODO use paths.backend_tfvars instead?	×
534	backend_tfvars = hcl2.loads(backend_tfvars.read_text()) if backend_tfvars.exists() else {}	×
535
536	logger.info("Checking [bold]backend.tfvars[/bold]:\n")	×
537	names_prefix = f"{tf.project}-{account_name}"	×
538	names_prefix_bootstrap = f"{tf.project}-bootstrap"	×
539
540	# Check profile, bucket and dynamo table names:
541	for field in ("profile", "bucket", "dynamodb_table"):	×
542	logger.info(f"Checking if {field.replace('_', ' ')} starts with {names_prefix}...")	×
543	if backend_tfvars.get(field, "").startswith(names_prefix) or (	×
544	field == "profile" and backend_tfvars.get(field, "").startswith(names_prefix_bootstrap)
545	):
546	logger.info("[green]✔ OK[/green]\n")	×
547	else:
548	logger.error("[red]✘ FAILED[/red]\n")	×
549	valid_layout = False	×
550
551	return valid_layout	×

binbashar / leverage / 8744072155

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous