Skip to content

terraform

Module that contains terraform related functions.

create_tf_files(config, target_path)

Create Terraform files (iam-bindings) based on user-specified resource configuration.

Parameters:

Name Type Description Default
config IAMBindingConfig

IAMBindingConfig collected from user that specifies which resources to generate Terraform IAM bindings for

 required
target_path str

Path to the (temporary) git repo where the tf files should be written

 required

Returns:

Type Description
Dict[str, str]

An IAMBindingConfig (filename -> tf file content)
Source code in dapla_team_cli/tf/iam_bindings/terraform.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
def create_tf_files(config: IAMBindingConfig, target_path: str) -> Dict[str, str]:
    """Create Terraform files (iam-bindings) based on user-specified resource configuration.

    Args:
        config: IAMBindingConfig collected from user that specifies which resources to
            generate Terraform IAM bindings for
        target_path: Path to the (temporary) git repo where the tf files should be written

    Returns:
        An IAMBindingConfig (filename -> tf file content)

    """
    # TODO: Refactor this function..
    tf_files = {}

    bucket_template = jinja_env.get_template("buckets-iam.jinja")
    project_iam_template = jinja_env.get_template("projects-iam.jinja")

    for auth_group in config.auth_groups:
        for env in auth_group.envs:
            filename = f"iam-{auth_group.shortname}-{env.name}.tf"
            filepath = os.path.join(target_path, filename)

            # Projects IAM binding module name
            iam_module = get_iam_module_name(auth_group.shortname, env.name)

            # Find old bindings in file if it exists
            old_bindings = []
            old_buckets = {}
            old_content = ""
            if os.path.exists(filepath):
                with open(filepath, encoding="utf-8") as f:
                    old_content = f.read()
                    hcl_object = hcl2.loads(old_content)

                    # Old projects IAM bindings
                    old_bindings = get_old_bindings(hcl_object, iam_module)
                    # Overwrite old bindings if role name matches
                    new_roles = [r.role.name for r in env.roles]
                    old_bindings = [r for r in old_bindings if r["role"] not in new_roles]

                    # Old bucket bindings
                    for bucket in {b.name for b in env.buckets}:
                        module_name = get_bucket_module_name(auth_group.name, env.name, bucket)
                        old_buckets[bucket] = get_old_bindings(hcl_object, module_name)
                        new_binding_roles = [b.access for b in env.buckets if b.name == bucket]
                        old_buckets[bucket] = [b for b in old_buckets[bucket] if b["role"].split(".")[-1] not in new_binding_roles]

            if env.roles:
                projects_iam = project_iam_template.render(auth_group=auth_group, env=env, old_bindings=old_bindings)
                pattern = re.compile(module_regex(iam_module), flags=re.MULTILINE)
                if pattern.search(old_content):
                    old_content = pattern.sub(projects_iam, old_content)
                else:
                    old_content = "\n".join((old_content, projects_iam))

            for bucket, bindings in igroupby(env.buckets, lambda b: b.name).items():
                module_name = get_bucket_module_name(auth_group.name, env.name, bucket)

                bucket_iam = bucket_template.render(
                    bucket=bucket,
                    auth_group=auth_group,
                    env=env.name,
                    bucket_bindings=bindings,
                    old_bindings=old_buckets.get(bucket, []),
                    team_name=config.team_name,
                )
                pattern = re.compile(module_regex(module_name), flags=re.MULTILINE)
                if pattern.search(old_content):
                    old_content = pattern.sub(bucket_iam, old_content)
                else:
                    old_content = "\n".join((old_content, bucket_iam))

            tf_files[filename] = old_content
    return tf_files

get_bucket_module_name(shortname, env, name)

Get name of the iam module for the given group, environment and bucket.

Source code in dapla_team_cli/tf/iam_bindings/terraform.py 
55
56
57
def get_bucket_module_name(shortname: str, env: str, name: str) -> str:
    """Get name of the iam module for the given group, environment and bucket."""
    return f"buckets-{shortname}-iam-{env}-{name}"

get_iam_module_name(shortname, environment)

Get name of iam-bindings module for the given auth group and environment.

Source code in dapla_team_cli/tf/iam_bindings/terraform.py 
50
51
52
def get_iam_module_name(shortname: str, environment: str) -> str:
    """Get name of iam-bindings module for the given auth group and environment."""
    return f"projects-iam-{shortname}-{environment}"

get_module(modules, module_name)

Return the module asked for, or return None.

Source code in dapla_team_cli/tf/iam_bindings/terraform.py 
60
61
62
63
64
65
66
def get_module(modules: Any, module_name: str) -> Optional[Any]:
    """Return the module asked for, or return None."""
    for module in modules:
        if module_name in module:
            return module

    return None

get_old_bindings(hcl_object, module_name)

Get already-existing bindings, so that we can avoid overwriting them.

Source code in dapla_team_cli/tf/iam_bindings/terraform.py 
69
70
71
72
73
74
75
76
77
78
79
80
81
82
def get_old_bindings(hcl_object: Dict[Any, Any], module_name: str) -> List[Any]:
    """Get already-existing bindings, so that we can avoid overwriting them."""
    # if not os.path.exists(bindings_file):
    #     return []

    # with open(bindings_file, encoding="utf-8") as f:
    if "module" not in hcl_object:
        return []

    module = get_module(hcl_object["module"], module_name)
    if module is None:
        return []

    return list(module[module_name]["conditional_bindings"])

module_regex(module_name)

Returns regex to match a Terraform module block.

Source code in dapla_team_cli/tf/iam_bindings/terraform.py 
85
86
87
def module_regex(module_name: str) -> str:
    """Returns regex to match a Terraform module block."""
    return rf'module "{module_name}" {{(?:\n(?: [ ]+.*$)*)+\n}}'

write_tf_files(config, target_path)

Produce and write terraform files to target_path.

Parameters:

Name Type Description Default
config IAMBindingConfig

user supplied configuration

 required
target_path str

The path to write files to

 required

Returns:

Type Description
List[Any]

a list of terraform files that was written
Source code in dapla_team_cli/tf/iam_bindings/terraform.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
def write_tf_files(config: IAMBindingConfig, target_path: str) -> List[Any]:
    """Produce and write terraform files to `target_path`.

    Args:
        config: user supplied configuration
        target_path: The path to write files to

    Returns:
        a list of terraform files that was written
    """
    # Create terraform files - one file per auth group and environment
    tf_files = create_tf_files(config, target_path)
    target_tf_files = []

    # Write the files to the team's IaC repo
    for tf_file_name, content in tf_files.items():
        file_path = os.path.join(target_path, tf_file_name)
        with open(file_path, mode="w", encoding="utf-8") as tf_file:
            tf_file.write(content)
            target_tf_files.append(tf_file)

    return target_tf_files