tfconfig-functions.sentinel

# Common functions that use the tfconfig/v2 import

##### Imports #####
import "tfconfig/v2" as tfconfig
import "strings"
import "types"

##### Functions #####

### find_all_resources ###
# Find all resources of all types using the tfconfig/v2 import.
find_all_resources = func() {
  resources = filter tfconfig.resources as address, r {
  	r.mode is "managed"
  }

  return resources
}

### find_resources_by_type ###
# Find all resources of a specific type using the tfconfig/v2 import.
# The parameter, type, should be a string like "aws_instance".
find_resources_by_type = func(type) {
  resources = filter tfconfig.resources as address, r {
  	r.type is type and
  	r.mode is "managed"
  }

  return resources
}

### find_resources_in_module ###
# Find all resources from a specific module using the tfconfig/v2 import.
find_resources_in_module = func(module_address) {
  resources = filter tfconfig.resources as address, r {
  	r.module_address is module_address and
  	r.mode is "managed"
  }

  return resources
}

### find_resources_by_provider ###
# Find all resources from a specific provider using the tfconfig/v2 import.
# The parameter, provider, should be given as a string such as "aws".
find_resources_by_provider = func(provider) {
  resources = filter tfconfig.resources as address, r {
  	r.provider_config_key matches "(.*:)?" + provider + "(\\..*)?" and
  	r.mode is "managed"
  }

  return resources
}

### find_all_datasources ###
# Find all data sources of all types using the tfconfig/v2 import.
find_all_datasources = func() {
  datasources = filter tfconfig.resources as address, d {
  	d.mode is "data"
  }

  return datasources
}

### find_datasources_by_type ###
# Find all data sources of a specific type using the tfconfig/v2 import.
# The parameter, type, should be a string like "aws_ami".
find_datasources_by_type = func(type) {
  datasources = filter tfconfig.resources as address, d {
  	d.type is type and
  	d.mode is "data"
  }

  return datasources
}

### find_datasources_in_module ###
# Find all data sources from a specific module using the tfconfig/v2 import.
find_datasources_in_module = func(module_address) {
  datasources = filter tfconfig.resources as address, d {
  	d.module_address is module_address and
  	d.mode is "data"
  }

  return datasources
}

### find_datasources_by_provider ###
# Find all data sources from a specific provider using the tfconfig/v2 import.
# The parameter, provider, should be given as a string such as "aws".
find_datasources_by_provider = func(provider) {
  datasources = filter tfconfig.resources as address, d {
  	d.provider_config_key matches "(.*:)?" + provider + "(\\..*)?" and
  	d.mode is "data"
  }

  return datasources
}

### find_all_provisioners ###
# Find all provisioners using the tfconfig/v2 import.
find_all_provisioners = func() {
  return tfconfig.provisioners
}

### find_provisioners_by_type ###
# Find all provisioners of a specific type using the tfconfig/v2 import.
# The parameter, type, should be a string like "local_exec".
find_provisioners_by_type = func(type) {
  provisioners = filter tfconfig.provisioners as address, p {
  	p.type is type
  }

  return provisioners
}

### find_all_providers ###
# Find all providers using the tfconfig/v2 import.
find_all_providers = func() {
  return tfconfig.providers
}

### find_providers_by_type ###
# Find all providers of a specific type using the tfconfig/v2 import.
# The parameter, provider, should be given as a string such as "aws".
find_providers_by_type = func(type) {
  providers = filter tfconfig.providers as address, p {
  	p.provider_config_key matches "(.*:)?" + type + "(\\..*)?"
  }

  return providers
}

### find_providers_in_module ###
# Find all providers from a specific module using the tfconfig/v2 import.
find_providers_in_module = func(module_address) {
  providers = filter tfconfig.providers as address, p {
  	p.module_address is module_address
  }

  return providers
}

### find_all_variables ###
# Find all variables using the tfconfig/v2 import.
find_all_variables = func() {
  return tfconfig.variables
}

### find_variables_in_module ###
# Find all variables from a specific module using the tfconfig/v2 import.
find_variables_in_module = func(module_address) {
  variables = filter tfconfig.variables as address, v {
  	v.module_address is module_address
  }

  return variables
}

### find_all_outputs ###
# Find all outputs using the tfconfig/v2 import.
find_all_outputs = func() {
  return tfconfig.outputs
}

### find_outputs_in_module ###
# Find all providers from a specific module using the tfconfig/v2 import.
find_outputs_in_module = func(module_address) {
  outputs = filter tfconfig.outputs as address, o {
  	o.module_address is module_address
  }

  return outputs
}

### find_outputs_by_sensitivity ###
# Find all providers of specific sensitivity using the tfconfig/v2 import.
# The parameter, sensitive, should be true or false (without quotes)
find_outputs_by_sensitivity = func(sensitive) {
  outputs = filter tfconfig.outputs as address, o {
  	o.sensitive is sensitive
  }

  return outputs
}

### find_all_module_calls ###
# Find all module calls using the tfconfig/v2 import.
find_all_module_calls = func() {
  return tfconfig.module_calls
}

### find_module_calls_in_module ###
# Find all direct module calls made from a specific module
# using the tfconfig/v2 import.
# The parameter, module_address, should be "" for the root module,
# "module.A" for a module A called by the root module,
# "module.A.module.B" for module B called by module A called by the root module,
# and so on.
find_module_calls_in_module = func(module_address) {
  module_calls = filter tfconfig.module_calls as address, mc {
  	mc.module_address is module_address
  }

  return module_calls
}

### find_descendant_modules ###
# Find addresses of all modules called directly or indirectly by a module.
# The provided module address is included.
# To find all module addresses, call find_descendant_modules("")
# After calling this function against "", you can call find_module_calls_in_module
# against any item in the list that is returned.
find_descendant_modules = func(module_address) {
  module_addresses = [module_address]
  mcs = find_module_calls_in_module(module_address)
  if length(mcs) > 0 {
    for mcs as ma, mc {
      if mc.module_address is "" {
        new_module_address = "module." + mc.name
      } else {
        new_module_address = mc.module_address + ".module." + mc.name
      }
      module_addresses += find_descendant_modules(new_module_address)
    }
  }
  return module_addresses
}

### print_violations ###
# Prints violations returned by any of the filter functions defined below.
# This would normally only be called if the filter function had been called
# with prtmsg set to false, which is sometimes done when processing resources
# and their blocks.
# If the result of a filter function is assigned to a map like violatingIRs,
# then you should pass violatingIRs["message"] as the first argument.
# The prefix argument is printed before the message of each resource.
print_violations = func(messages, prefix) {
  for messages as address, message {
    print(prefix, message)
  }
  return true
}

### to_string ###
# Convert objects of unknown type to string
# It is used to build messages added to the messages map returned by the
# filter functions
to_string = func(obj) {
  case types.type_of(obj) {
    when "string":
      return obj
    when "int", "float", "bool":
      return string(obj)
    when "null":
      return "null"
    when "undefined":
      return "undefined"
    when "list":
      output = "["
      lastIndex = length(obj) - 1
      for obj as index, value {
        if index < lastIndex {
          output += to_string(value) + ", "
        } else {
          output += to_string(value)
        }
      }
      output += "]"
      return output
    when "map":
      output = "{"
      theKeys = keys(obj)
      lastIndex = length(theKeys) - 1
      for theKeys as index, key {
        if index < lastIndex {
          output += to_string(key) + ": " + to_string(obj[key]) + ", "
        } else {
          output += to_string(key) + ": " + to_string(obj[key])
        }
      }
      output += "}"
      return output
    else:
      return ""
  }
}

### evaluate_attribute ###
# Evaluates an attribute
# In general, the attribute should be a top-level attribute of item, but
# we do special processing for attributes with form "config.x"
# `item` is the item with the attribute
# `attribute` is the attribute
evaluate_attribute = func(item, attribute) {
  # Split the attribute into a list, using "." as the separator
  attributes = strings.split(attribute, ".")
  if length(attributes) > 2 {
    print("An attribute passed to evaluate_attribute can only have 1 or 2 fields")
    return null
  }
  if attributes[0] is "config" {
    config = item.config[attributes[1]] else {}
    if "constant_value" in config {
      # Found constant_value in config
      return config.constant_value
    } else if "references" in config {
      # Found references in config
      return config.references
    } else {
      # Did not find constant_value or references in config
      return null
    }
  } else {
    # Return the original attribute or the item
    return item[attribute]
  }
}

### filter_attribute_not_in_list ###
# Filter a list of items such as providers to those with a specified
# attribute (attr) that is not in a given list of allowed values (allowed).
# The parameter, attr, can only be a top-level attribute of the collection, items.
# Set prtmsg to `true` (without quotes) if you want to print violation messages.
# If you want to disallow null, include "null" in the list (forbidden).
filter_attribute_not_in_list = func(items, attr, allowed, prtmsg) {
  violators = {}
  messages = {}

  # Iterate over items
  for items as index, item {
    val = evaluate_attribute(item, attr) else null
    # Check if the value is null
    if val is null {
      val = "null"
    }
    # Process lists and maps
    if types.type_of(val) in ["list", "map"] {
      message = ""
      # Check each item of list or map
      for val as i, v {
        if v not in allowed {
          # Add the item and a warning message to the violators list
          message = to_string(index) + " has " + to_string(attr) + " with value " +
                    to_string(v) + " that is not in the allowed list: " +
                    to_string(allowed)
        }
        if message is not "" {
          # Add the item and warning message to the violators list
          violators[index] = item
      		messages[index] = message
          if prtmsg {
            print(message)
          }
        } // end message not ""
      } // end for
    } else {
      # Process single item
      if val not in allowed {
        # Add the item and a warning message to the violators list
        message = to_string(index) + " has " + to_string(attr) +
                  " with value " + to_string(val) +
                  " that is not in the allowed list: " +
                  to_string(allowed)
        violators[index] = item
        messages[index] = message
        if prtmsg {
          print(message)
        }
      } // end if single item not matches
    } // end single item
  } // end for items
  return {"items":violators,"messages":messages}
}

### filter_attribute_in_list ###
# Filter a list of items such as providers to those with a specified
# attribute (attr) that is in a given list of forbidden values (forbidden).
# The parameter, attr, can only be a top-level attribute of the collection, items.
# Set prtmsg to `true` (without quotes) if you want to print violation messages.
# If you want to disallow null, include "null" in the list (forbidden).
filter_attribute_in_list = func(items, attr, forbidden, prtmsg) {
  violators = {}
  messages = {}

  # Iterate over items
  for items as index, item {
    val = evaluate_attribute(item, attr) else null
    # Check if the value is null
    if val is null {
      val = "null"
    }
    # Process lists and maps
    if types.type_of(val) in ["list", "map"] {
      message = ""
      # Check each item of list or map
      for val as i, v {
        if v in forbidden {
          # Add the item and a warning message to the violators list
          message = to_string(index) + " has " + to_string(attr) + " with value " +
                    to_string(v) + " that is in the forbidden list: " +
                    to_string(forbidden)
        }
        if message is not "" {
          # Add the item and warning message to the violators list
          violators[index] = item
      		messages[index] = message
          if prtmsg {
            print(message)
          }
        } // end message not ""
      } // end for
    } else {
      # Process single item
      if val in forbidden {
        # Add the item and a warning message to the violators list
        message = to_string(index) + " has " + to_string(attr) +
                  " with value " + to_string(val) +
                  " that is in the forbidden list: " +
                  to_string(forbidden)
        violators[index] = item
        messages[index] = message
        if prtmsg {
          print(message)
        }
      } // end if single item not matches
    } // end single item
  } // end for items

  return {"items":violators,"messages":messages}
}

### filter_attribute_does_not_match_regex ###
# Filter a list of items such as resources to those with a specified
# attribute (attr) that does not match a regular expression (expr).
# The parameter, attr, can only be a top-level attribute of items or
# an attribute in the form "config.x".
# Set prtmsg to `true` (without quotes) if you want to print violation messages.
filter_attribute_does_not_match_regex = func(items, attr, expr, prtmsg) {
  violators = {}
	messages = {}
  for items as index, item {
    val = evaluate_attribute(item, attr) else null
    if val is null {
      # Add the item and a warning message to the violators list
      message = to_string(index) + " has " + to_string(attr) +
                " that is null or undefined. " + "It is supposed to " +
                "match the regex " + to_string(expr)
      violators[index] = item
			messages[index] = message
      if prtmsg {
        print(message)
      }
    } else {
      # Process lists and maps
      if types.type_of(val) in ["list", "map"] {
        message = ""
        # Check each item of list or map
        for val as i, v {
          if v not matches expr {
            # Add to the warning message
            message += to_string(index) + " has " + to_string(attr) +
                      " with value " + to_string(v) +
                      " that does not match the regex " + to_string(expr) + "\n"
          }
          if message is not "" {
            # Add the item and warning message to the violators list
            violators[index] = item
      			messages[index] = message
            if prtmsg {
              print(message)
            }
          } // end message not ""
        } // end for
      } else {
        # Process single item
        if val not matches expr {
          # Add the item and a warning message to the violators list
          message = to_string(index) + " has " + to_string(attr) +
                    " with value " + to_string(val) +
                    " that does not match the regex " + to_string(expr)
          violators[index] = item
          messages[index] = message
          if prtmsg {
            print(message)
          }
        } // end if single item not matches
      } // end single item
    } // end not null
  } // end for items
  return {"items":violators,"messages":messages}
}

### filter_attribute_matches_regex ###
# Filter a list of items such as resources to those with a specified
# attribute (attr) that matches a regular expression (expr).
# The parameter, attr, can only be a top-level attribute of items or
# an attribute in the form "config.x".
# Set prtmsg to `true` (without quotes) if you want to print violation messages.
# If you want to match null, set expr to "null".
filter_attribute_matches_regex = func(items, attr, expr, prtmsg) {
  violators = {}
	messages = {}
  for items as index, item {
    val = evaluate_attribute(item, attr) else null
    if val is null {
      val = "null"
    }
    # Process lists and maps
    if types.type_of(val) in ["list", "map"] {
      message = ""
      # Check each item of list or map
      for val as i, v {
        if v matches expr {
          # Add to the warning message
          message += to_string(index) + " has " + to_string(attr) +
                    " with value " + to_string(v) +
                    " that matches the regex " + to_string(expr) + "\n"
        }
        if message is not "" {
          # Add the item and warning message to the violators list
          violators[index] = item
          messages[index] = message
          if prtmsg {
            print(message)
          }
        } // end message not ""
      } // end for
    } else {
      # Process single item
      if val matches expr {
        # Add the item and a warning message to the violators list
        message = to_string(index) + " has " + to_string(attr) +
                  " with value " + to_string(val) +
                  " that matches the regex " + to_string(expr)
        violators[index] = item
        messages[index] = message
        if prtmsg {
          print(message)
        }
      } // end if single item not matches
    } // end single item
  } // end for items
  return {"items":violators,"messages":messages}
}

### get_module_source ###
# Get the module source from a module address
# Note that the module_address in many collections in the tfplan/v2, tfconfig/v2,
# and tfstate/v2 imports gives the labels used in the module blocks.
# For instance, a module_address like "module.A.module.B" means that the current
# item is in a module with label "B" that is a module with label "A". But that
# does not give you the source the module labeled "B".
# But if you want to limit creation of resources to specific modules based on
# their source, you need the module source.  This function computes it.
get_module_source = func(module_address) {
  # Check for root module
  if module_address is "" {
    return "root"
  } else {
    # Find parent module
    module_segments = strings.split(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]
    }

    # Set source from the module call
    module_source = mc.source

    return module_source
  }
}

### get_ancestor_module_source ###
# Get the module source of the first ancestor module from a module address that
# is not a local module (one having source starting with "./" or "../").
# Note that the module_address in many collections in the tfplan/v2, tfconfig/v2,
# and tfstate/v2 imports gives the labels used in the module blocks.
# For instance, a module_address like "module.A.module.B" means that the current
# item is in a module with label "B" that is a module with label "A". But that
# does not give you the source the module labeled "B".
# But if you want to limit creation of resources to specific modules based on
# their source, you need the module source.  This function computes it for
# the first ancestor module that is not a local module.
get_ancestor_module_source = func(module_address) {
  # Check for root module
  if module_address is "" {
    return "root"
  } else {
    # Find parent module
    module_segments = strings.split(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]
    }

    # Set source from the module call
    module_source = mc.source

    # Check to see if current module source is a nested module
    if strings.has_prefix(module_source, "./") or strings.has_prefix(module_source, "../") {
      return get_ancestor_module_source(parent_module)
    } else {
      return module_source
    }

  }
}

### get_parent_module_address ###
# Get the address of the parent module of a module from the module address
# return "root" if the given module_address is "" (the root module).
get_parent_module_address = func(module_address) {
  # Check for root module
  if module_address is "" {
    return "root"
  } else {
    # Find parent module
    module_segments = strings.split(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]

    return parent_module
  }
}