aws-functions.sentinel

# Common functions for use with the AWS provider

##### Imports #####
import "tfconfig-functions" as config
import "tfplan/v2" as tfplan
import "tfconfig/v2" as tfconfig
import "strings"
import "types"

##### Functions #####

### find_resources_with_standard_tags ###
find_resources_with_standard_tags = func(resource_types) {
  resources = filter tfplan.resource_changes as address, rc {
    rc.provider_name matches "(.*)aws$" and
    rc.type in resource_types and
  	rc.mode is "managed" and
    (rc.change.actions contains "create" or rc.change.actions contains "update" or
     rc.change.actions contains "read" or rc.change.actions contains "no-op")
  }

  return resources
}

### determine_role_arn ###
# This can only determine the role_arn if it is set to either a hard-coded value
# or to a reference to a single Terraform variable.
# It sets the role to "complex" if it finds a single non-variable reference
# or if it finds multiple references.
# It sets the role to "none" if no role arn is found.
determine_role_arn = func(address, data) {

  # Return empty string if provider does not assume a role
  role_arn_value = "none"

  # Check for role_arn
  if (length(data.config) else 0) > 0 and
     (length(data.config.assume_role) else 0) > 0 and
     data.config.assume_role[0].role_arn else null is not null {
    role_arn_data = data.config.assume_role[0].role_arn
    # Check for constant value or references
    if role_arn_data.constant_value else null is not null {
      # role_arn of AWS provider was hard-coded role_arn
      #print("Found a single constant value for role_arn.")
      role_arn_value = role_arn_data.constant_value
    } else if role_arn_data.references else [] is not [] {
      if length(role_arn_data.references) == 1 {
        # Process references
        role_arn = role_arn_data.references[0]
        if role_arn matches "var\\.(.*)" {
          # role_arn of AWS provider was a variable
          #print("Found a single variable reference for role_arn.")
          role_arn_variable = strings.trim_prefix(role_arn, "var.")
          role_arn_value = tfplan.variables[role_arn_variable].value
        } else {
          # reference was not a variable
          print("Found a single reference in the role_arn attribute,",
                "for provider", address, "but it was not a variable.")
          print("This policy only handles a role_arn attribute that is",
                "a constant value or a single reference to a variable.")
          # Set role_arn_value to null to cause failure of policy
          role_arn_value = "complex"
        } // end if role_arn is variable
      } else {
        print("Found more than one reference in the role_arn attribute",
              "for provider", address)
        print("This policy only handles a role_arn attribute that is",
              "a constant value or a single reference to a variable.")
        # Set role_arn_value to null to cause failure of policy
        role_arn_value = "complex"
      } // end if single reference
    } // end if constant_value or references
  } else {
    #print("Did not find role_arn.")
  } // end if assume_role.role_arn in config

  return role_arn_value
}

### get_assumed_roles ###
# Get assumed roles from all AWS providers
# Please note that the assumed roles returned could include "none" or "complex";
# See the comments for the determine_role_arn function above.
get_assumed_roles = func() {

  # Initialize empty map of roles indexed by aliases
  assumed_roles = {}

  # Get all AWS provider aliases
  aws_providers = config.find_providers_by_type("aws")

  # Iterate through all AWS provider aliases
  for aws_providers as address, data {
    assumed_roles[address] = determine_role_arn(address, data)
  } // end aws_providers

  return assumed_roles

}

### validate_assumed_roles_with_list ###
# Validate that all assumed roles are allowed.
# If you want to the policy to pass if an assumed role contains a single
# non-variable reference or if it finds multiple references, then include a role
# called "complex" in the allowed_roles list.
validate_assumed_roles_with_list = func(allowed_roles) {

  validated = true

  assumed_roles = get_assumed_roles()

  for assumed_roles as address, role {
    if role is not "none" and role not in allowed_roles {
      print("AWS provider", address, "has assumed role",
            role, "that is not allowed.")
      validated = false
    }
  }

  return validated
}

### validate_assumed_roles_with_map ###
# Validate that all assumed roles are allowed for the current workspace.
# If you want to a policy to pass if an assumed role contains a single
# non-variable reference or if it finds multiple references, then include a role
# called "complex" in the map passed to this function and associate it
# with workspaces.
validate_assumed_roles_with_map = func(roles_map, workspace_name) {

  validated = true

  assumed_roles = get_assumed_roles()

  for assumed_roles as address, role {
    if role is not "none" {
      if role not in keys(roles_map) {
        print("AWS provider", address, "has assumed role",
              role, "that is not allowed.")
        validated = false
      } else {
        # Validate that role is allowed for current workspace
        matched = false
        for roles_map[role] as workspace_regex {
          if workspace_name matches workspace_regex {
            matched = true
          }
        } // end for workspace_regex
        if not matched {
          print("Workspace", workspace_name, "is not allowed to use role", role)
          print("It used that role in the AWS provider", address)
          validated = false
        } // end matched check
      } // end role in roles_map
    } // end if role is not ""
  } // end for assumed_roles

  return validated
}

### filter_providers_by_regions ###
# Filter instances of the AWS provider to those in a specific region using the
# tfconfig/v2 and tfplan/v2 imports.
# See the comments on the validate_provider_in_allowed_regions() function below
# for details on how this is done.
# The parameter, aws_providers, should be a list of AWS provider aliases
# derived from tfconfig.providers.
# The parameter, allowed_regions, should be given as a list of AWS regions
# such as `["us-east-1" and "eu-west-2"]`.
filter_providers_by_regions = func(aws_providers, allowed_regions) {

  # Initialize empty list of validated AWS provider instances
  validated_providers = {}

  # Process  AWS providers
  for aws_providers as index, p {
    if "config" in keys(p) and "region" in keys(p.config) {
      # provider alias has a region
      validated = validate_provider_in_allowed_regions(p, allowed_regions)
      if validated {
        print("validated provider:", index)
        validated_providers[index] = p
      }
    } else {
      # provider alias does not have region
      # So, we process it as an alias to a root module provider
      # A provider that does not have its own region should look like
      # <module_address>:<provider>.<alias>
      # we want to look at <provider>.<alias>
      p_segments = strings.split(p.provider_config_key, ":")
      if length(p_segments) == 1 {
        # Current provider is already in root module
        # So, it is not an alias to another provider
        # Continue on to next resource in for loop
        continue
      }
      # Get the provider in root module that current provider aliases
      p_alias_name = p_segments[1]
      p_alias = tfconfig.providers[p_alias_name] else null

      # Process p_alias
      if p_alias is not null {
        validated =
              validate_provider_in_allowed_regions(p_alias, allowed_regions)
        if validated {
          print("validated provider:", index)
          validated_providers[index] = p
        }
      } // end p_alias not null
    } // end else no region
  } // end for

  # return validated providers
  return validated_providers

}

### validate_provider_in_allowed_regions ###
# Validate if a specific provider instance is in a list of regions using the # tfconfig/v2 and tfplan/v2 iimports.
# The parameter, p, should be a provider derived from tfconfig.providers
# or from provider_config_key of a resource from tfconfig.resources.
# The parameter, regions, should be given as a list of allowed regions

# It attempts to identify the region of the provider aliases in several ways
# including constant values assigned to their `region` argument and resolution
# of references to variables. It first tries to process references to variables
# as strings, then as maps with a key called "region". It handles references
# to variables in the root module by using tfplan.variables. It handles references
# to variables in non-root modules by examining the module call from the current
# module's parent.

# It even tries to match provider aliases in proxy configuration blocks
# (which do not specify regions) of child modules to similarly-named provider
# aliases in the root module.

# If the alias passed in the module call does not match the alias in the root
# module, Sentinel has no way of linking the two provider aliases. However,
# since all providers that do specify regions will be restricted and since
# provider alias proxies must point to other provider aliases in ancestor modules,
# all provider aliases should be restricted by this policy.
validate_provider_in_allowed_regions = func(p, regions) {
  if "config" in keys(p) and "region" in keys(p.config) {
    # provider has its own region
    if "constant_value" in keys(p.config.region) {
      if p.config.region.constant_value in regions {
        # Found resource from region
        return true
      } else {
        return false
      }
    } else if "references" in keys(p.config.region) {
      # Iterate over references in region
      # Usually, there should only be one reference
      for p.config.region.references as reference {
        # Cross reference variables
        # Note that this function does not handle other types of references
        if strings.has_prefix(reference, "var.") {
          variable_name = strings.split(reference, ".")[1]
          if p.module_address is "" {
            # Get root module variable from tfplan.variables
            variable_value = tfplan.variables[variable_name].value
            # Check type of variable
            variable_type = types.type_of(variable_value)
            if variable_type is "string" and variable_value in regions {
              # Found resource from region
              return true
            } else if variable_type is "map" and
                      "region" in keys(variable_value) and
                      variable_value["region"] in regions {
              # Found resource from region
              return true
            } // end variable_type check
          } else {
            # Get non-root module variable
            # Could be value passed into variable by module call
            # or be the default value of a variable in the module

            # First find parent module
            module_segments = strings.split(p.module_address, ".")
            num_segments = length(module_segments)
            parent_module = strings.join(module_segments[0:num_segments-2], ".")
            current_module_name = module_segments[num_segments-1]

            # Find module call that called current module
            if parent_module is "" {
              # parent module is root module
              mc = tfconfig.module_calls[current_module_name]
            } else {
              # parent module is not root module
              mc = tfconfig.module_calls[parent_module + ":" + current_module_name]
            }

            # Check if referenced variable was passed in the module call
            if variable_name in keys(mc.config) {
              mc_var = mc.config[variable_name]
              if "constant_value" in keys(mc_var) {
                if mc_var.constant_value in regions {
                  # Found resource from region
                  return true
                } else {
                  return false
                }
              } else if "references" in keys(mc_var) and parent_module is "" {
                // Check root module variable references
                for mc_var.references as mc_reference {
                  if strings.has_prefix(mc_reference, "var.") {
                    mc_variable_name = strings.split(mc_reference, ".")[1]
                    mc_variable_value = tfplan.variables[mc_variable_name].value
                    mc_variable_type = types.type_of(mc_variable_value)
                    if mc_variable_type is "string" and mc_variable_value in regions {
                      # Found resource from region
                      return true
                    } else if mc_variable_type is "map" and
                              "region" in keys(mc_variable_value) and
                              mc_variable_value["region"] in regions {
                      # Found resource from region
                      return true
                    } // end mc_variable_type check
                  } // end if mc_reference is a root module variable
                } // end for mc_var.references
              } // end constant_value or references in module call
            } else {
              # check default value of variable in current module
              default_value = tfconfig.variables[variable_name].default
              default_value_type = types.type_of(default_value)
              if default_value_type is "string" and default_value in regions {
                # Found resource from region
                return true
              } else if default_value_type is "map" and
                      "region" in keys(default_value) and
                      default_value["region"] in regions {
                # Found resource from region
                return true
              }
            } // end if matched variable in module config or default value
          } // end else non-root module
        } // end reference is a variable
      } // end for references
    } // end references
  } // end if p has config.region

  # If we did not already return true, then return false
  return false

}