Verified Syntax

# ✅ Correct
condition = core::semverconstraint(meta.version, ">= 4.0.0, < 5.0.0")

# ❌ Wrong
condition = meta.version >= 4.0  # Don't use direct comparison

# ✅ Correct
filter = core::try(attrs.encrypted, false) == true
is_allowed = core::length([for v in local.version_checks : v if v]) > 0
error_message = "Allowed: ${core::join(", ", local.allowed_versions)}"

# ❌ Wrong
filter = try(attrs.encrypted, false) == true  # Missing core:: prefix

# Sentinel: strings.has_prefix(p.version_constraint, ">")   →  non-convertible
# TFPolicy nearest equivalent — enforce resolved version range instead:
provider_policy "*" "provider_version_range" {
  enforce {
    condition     = core::semverconstraint(meta.version, ">= 4.0.0, < 5.0.0")
    error_message = "Provider version '${meta.version}' must satisfy '>= 4.0.0, < 5.0.0'. Pin the provider to a tested version range to prevent major version upgrades."
  }
}

# ✅ BLOCK access (default_tags in AWS provider)
# Schema: default_tags = [ { tags = { "Env" = "Dev" } } ]
default_tags_map = core::try(attrs.default_tags[0].tags, {})  # Need [0] index

# ✅ ATTRIBUTE access (tags on resources)
# Schema: tags = { "Env" = "Dev" }
has_tags = attrs.tags != null && core::length(attrs.tags) > 0  # Direct access, no [0]

# Note: core::length() works directly on maps, no need for core::keys()

# ✅ Correct — parameterized, overridable per policy set
input "allowed_providers" {
  type    = list(string)
  default = ["registry.terraform.io/hashicorp/aws"]
}

provider_policy "*" "providers_allowlist" {
  enforce {
    condition     = core::contains(input.allowed_providers, meta.source)
    error_message = "Provider '${meta.source}' not allowed. Permitted: ${core::join(", ", input.allowed_providers)}."
  }
}

# ❌ Wrong — hardcoding what should be configurable
locals {
  allowed_providers = ["registry.terraform.io/hashicorp/aws"]  # Never override without editing policy file
}

# ✅ Fires only on create and update — never on destroy
resource_policy "tfe_workspace" "require_project" {
  operations = ["create", "update"]
  enforce {
    condition     = core::try(attrs.project_id, "") != ""
    error_message = "tfe_workspace must have project_id set."
  }
}

# ✅ Delete-gate: check prior state before workspace is destroyed
resource_policy "tfe_workspace" "deny_delete_without_tag" {
  operations = ["delete"]   # prior_attrs is available when "create" is NOT in operations
  locals {
    prior_tag_names = core::try(prior_attrs.tag_names, [])
    had_delete_tag  = core::contains(local.prior_tag_names, "delete")
  }
  enforce {
    condition     = local.had_delete_tag
    error_message = "Add 'delete' tag before destroying a workspace."
  }
}

# ✅ Day-of-week restriction
input "restricted_weekdays_utc" {
  type    = list(string)
  default = ["Friday", "Saturday", "Sunday"]
}

resource_policy "tfe_workspace" "deny_apply_day_of_week" {
  locals {
    current_weekday = core::formatdate("EEEE", core::timestamp())  # "Monday", "Friday", etc.
    is_restricted   = core::contains(input.restricted_weekdays_utc, local.current_weekday)
  }
  enforce {
    condition     = !local.is_restricted
    error_message = "Apply blocked on ${local.current_weekday} (UTC). Restricted days: ${core::join(", ", input.restricted_weekdays_utc)}."
  }
}

# ✅ Hour-of-day restriction
input "restricted_hours_utc" {
  type    = list(number)
  default = [8, 9, 10, 11, 12]
}

resource_policy "tfe_workspace" "deny_apply_hour_of_day" {
  locals {
    current_hour = core::parseint(core::formatdate("HH", core::timestamp()), 10)
    is_restricted = core::contains(input.restricted_hours_utc, local.current_hour)
  }
  enforce {
    condition     = !local.is_restricted
    error_message = "Apply blocked during hour ${local.current_hour} UTC."
  }
}

# ✅ Correct — descriptive snake_case
resource_policy "aws_s3_bucket" "versioning_required" { }
resource_policy "aws_iam_policy" "no_admin_privileges" { }
resource_policy "aws_ecs_task_definition" "secure_networking_mode_and_user" { }

# ❌ Wrong — generic names that don't describe the check
resource_policy "aws_s3_bucket" "check" { }
resource_policy "aws_s3_bucket" "all_checks" { }
resource_policy "aws_s3_bucket" "policy" { }

# ✅ Correct defaults by type
encrypted        = core::try(attrs.encrypted, false)          # boolean
bucket_name      = core::try(attrs.bucket, "")                 # string
tag_names        = core::try(attrs.tag_names, [])              # list
tags             = core::try(attrs.tags, {})                   # map
optional_config  = core::try(attrs.config, null) != null       # detect presence

# Wrong
condition = meta.version < 5.0

# Wrong - Crashes when attribute doesn't exist
has_region = attrs.region != null

# Wrong - Still crashes even with double check!
condition = attrs.encrypted != null && attrs.encrypted == true

# Correct
region_value = core::try(attrs.region, null)
has_region = local.region_value != null

# Language server shows error, but syntax is valid:
provider_policy "aws" "example" {
    locals {  # ❌ False error: "No declaration found"
        version_check = ...
    }
}

# Suboptimal - doesn't filter empty collections
filter = attrs.ingress != null

# Better - filters both null and empty
filter = attrs.ingress != null && core::length(attrs.ingress) > 0

# Wrong - causes syntax errors
is_exact = local.has_version &&
           !core::contains(meta.version, "~>") &&
           !core::contains(meta.version, ">")

# Correct - all on one line
is_exact = local.has_version && !core::contains(meta.version, "~>") && !core::contains(meta.version, ">")

# Wrong - module inputs not accessible yet
module_policy "*" "check" {
    locals {
        dns_enabled = attrs.enable_dns_hostnames  # ❌ Not supported
    }
}

# Wrong - substring matching doesn't work
module_policy "vpc" "check" { }  # Won't match modules with "vpc" in source

# Correct - use full source path or wildcard
module_policy "app.terraform.io/myorg/vpc/aws" "check" { }
module_policy "*" "check" { }  # For all modules

# Wrong - default_tags is a Block, needs [0]
provider_policy "aws" "check" {
    locals {
        tags = attrs.default_tags.tags  # ❌ Error: no indices
    }
}

# Correct - use [0] for Blocks
provider_policy "aws" "check" {
    locals {
        tags = attrs.default_tags[0].tags  # ✅ Works!
    }
}

# Wrong - can't use source in first label
provider_policy "hashicorp/aws" "check" { }  # Doesn't work

# Correct - use provider TYPE
provider_policy "aws" "check" {
    # meta.source available inside policy
    # Example: meta.source = "registry.terraform.io/hashicorp/aws"
}

# Unnecessarily complex
has_tags = core::try(core::length(core::keys(attrs.tags)), 0) > 0

# Simpler - core::length() works directly on maps
has_tags = core::try(core::length(attrs.tags), 0) > 0

# Most readable
has_tags = attrs.tags != null && core::length(attrs.tags) > 0

# ❌ WRONG — null or empty filter inside resource_policy runs for EVERY resource (O(N) cost)
resource_policy "aws_s3_bucket" "check" {
    locals {
        all_policies = core::getresources("aws_s3_bucket_policy", null)  # unscoped
    }
}
# ❌ ALSO WRONG — empty filter {} is equally unscoped inside resource_policy
resource_policy "aws_s3_bucket" "check" {
    locals {
        all_policies = core::getresources("aws_s3_bucket_policy", {})
    }
}

# ✅ CORRECT — cache at top-level locals for plan-time join (runs once total)
locals {
    all_policies = core::getresources("aws_s3_bucket_policy", {})
}

# Example of a LINKING attribute (makes inline core::getresources() correct):
# aws_emr_cluster has: security_configuration = aws_emr_security_configuration.x.name
# → The cluster references the security config by name → inline lookup is valid.

# Example of NO linking attribute (makes inline core::getresources() WRONG):
# aws_instance has NO attribute that references aws_ec2_instance_metadata_defaults.
# aws_ec2_instance_metadata_defaults has NO attribute that references aws_instance.
# → They are independent → each gets its own separate resource_policy block.

# ❌ WRONG — top-level {} cache, then HCL-filtered by attrs.* value inside resource_policy
locals {
  all_sec_configs = core::getresources("aws_emr_security_configuration", {})
}
resource_policy "aws_emr_cluster" "check" {
  locals {
    # ❌ Wrong pattern: fetches all globally and re-filters per cluster
    matching = [for sc in local.all_sec_configs : sc if sc.name == attrs.security_configuration]
  }
}

# ❌ ALSO WRONG — same anti-pattern restructured as a top-level lookup map.
# Building { name => value } at the top level and indexing by an attrs.*-derived key
# inside resource_policy is semantically equivalent to the {} cache + HCL for-loop above.
# The lookup key (local.security_config_name = core::try(attrs.security_configuration, ""))
# is still derived from attrs.*, so this violates the same rule.
locals {
  all_security_configs = core::getresources("aws_emr_security_configuration", {})
  security_config_map  = {
    for sc in local.all_security_configs :
    core::try(sc.name, "") => core::try(sc.configuration, "")
  }
}
resource_policy "aws_emr_cluster" "check" {
  locals {
    security_config_name = core::try(attrs.security_configuration, "")
    # ❌ Wrong: map lookup by attrs.*-derived key is an attrs.* filter in disguise
    security_config_json = core::try(local.security_config_map[local.security_config_name], "")
  }
}

# ✅ CORRECT — inline call with filter derived from attrs.*
resource_policy "aws_emr_cluster" "check" {
  locals {
    matching = core::getresources("aws_emr_security_configuration", {
      name = attrs.security_configuration
    })
  }
}

# ✅ CORRECT — inline call with per-resource filter using attrs.id
# NOTE: This policy contains a cross-resource reference that will not resolve during plan time,
# but the policy will run successfully during apply time.
resource_policy "aws_s3_bucket" "public_access_required" {
  locals {
    public_access_block     = core::getresources("aws_s3_bucket_public_access_block", {
      bucket = attrs.id  # Use attrs.id: Terraform sets child.bucket = parent.id
    })
    block_public_acls       = core::try(local.public_access_block[0].block_public_acls, false)
    block_public_policy     = core::try(local.public_access_block[0].block_public_policy, false)
    ignore_public_acls      = core::try(local.public_access_block[0].ignore_public_acls, false)
    restrict_public_buckets = core::try(local.public_access_block[0].restrict_public_buckets, false)
  }
  enforce {
    condition     = local.block_public_acls && local.block_public_policy && local.ignore_public_acls && local.restrict_public_buckets
    error_message = "S3 bucket must have all public access block settings enabled."
  }
}

# ❌ WRONG - Makes API calls for EVERY resource!
resource_policy "aws_s3_bucket" "check" {
    locals {
        account_id = core::getdatasource("aws_caller_identity", {})
    }
}

# ✅ CORRECT - Cache in top-level locals
locals {
    account_id = core::getdatasource("aws_caller_identity", {})
}

# ❌ WRONG
locals {
    all_roles = core::getresources("aws_iam_role", {})
}
resource_policy "aws_iam_role" "check" {
    locals {
        name = local.all_roles[0].attrs.name  # ERROR!
    }
}

# ✅ CORRECT
resource_policy "aws_iam_role" "check" {
    locals {
        name = local.all_roles[0].name  # Works!
    }
}

# ❌ WRONG - rule is a SET, cannot index
resource_policy "aws_s3_bucket_server_side_encryption_configuration" "check" {
    locals {
        sse_algo = attrs.rule[0].sse_algorithm  # ERROR!
    }
}

# ✅ CORRECT - Convert set to list using for loop
resource_policy "aws_s3_bucket_server_side_encryption_configuration" "check" {
    locals {
        sse_algo = [for rule in attrs.rule : rule][0].sse_algorithm
    }
}

# ❌ Redundant in condition = expressions (core::try catches the out-of-range error)
condition = core::length(local.list) > 0 && core::try(local.list[0].value, "") == "expected"

# ✅ Simpler — safe in condition = expressions
condition = core::try(local.list[0].value, "") == "expected"

# ✅ PREFERRED for core::getresources() results — explicit guard before [0]
locals {
  bucket_acl_resources = core::getresources("aws_s3_bucket_acl", { bucket = attrs.id })
  has_acl              = core::length(local.bucket_acl_resources) > 0
  acl_value            = local.has_acl ? core::try(local.bucket_acl_resources[0].acl, "") : ""
}

# ✅ PREFERRED for nested block lists from getresources() — guard each level
locals {
  sse_configs     = core::getresources("aws_s3_bucket_server_side_encryption_configuration", { bucket = attrs.id })
  has_sse_config  = core::length(local.sse_configs) > 0
  sse_rules       = local.has_sse_config ? core::try([for r in local.sse_configs[0].rule : r], []) : []
  has_sse_rule    = core::length(local.sse_rules) > 0
  sse_apply_block = local.has_sse_rule ? core::try([for a in local.sse_rules[0].apply_server_side_encryption_by_default : a], []) : []
  has_apply_block = core::length(local.sse_apply_block) > 0
  sse_algorithm   = local.has_apply_block ? core::try(local.sse_apply_block[0].sse_algorithm, "") : ""
}

# ❌ AVOID for getresources() results — relies on core::try to mask absent child resource
locals {
  sse_configs   = core::getresources("aws_s3_bucket_server_side_encryption_configuration", { bucket = attrs.id })
  sse_rules     = core::try([for r in local.sse_configs[0].rule : r], [])       # no guard: absence silently swallowed
  sse_algorithm = core::try(local.sse_rules[0].apply_server_side_encryption_by_default[0].sse_algorithm, "")
}

# ❌ Won't match during initial creation
resource "aws_s3_bucket_server_side_encryption_configuration" "example" {
  bucket = aws_s3_bucket.example.id  # Reference not resolved at policy time
}

# ✅ For testing, use literals
resource "aws_s3_bucket_server_side_encryption_configuration" "example" {
  bucket = "my-bucket"  # Literal value
}

# ❌ WRONG — meta.address is UNDEFINED in resource_policy real-plan evaluation
resource_policy "aws_s3_bucket" "check" {
    enforce {
        condition = !local.has_public_acl
        error_message = "S3 bucket '${meta.address}' has a public ACL."  # ERROR!
    }
}

# ✅ CORRECT — use static strings or safe attrs interpolation
resource_policy "aws_s3_bucket" "check" {
    enforce {
        condition = !local.has_public_acl
        error_message = "S3 bucket has a prohibited public ACL. Set acl to 'private' or remove it."
        # Or with dynamic context using a known attribute:
        # error_message = "S3 bucket '${attrs.bucket}' has a prohibited public ACL."
    }
}

# ❌ WRONG — Only covers 2 of 4 SG resource types; misses modern VPC API
resource_policy "aws_security_group" "check" { ... }
resource_policy "aws_security_group_rule" "check" { ... }

# ✅ CORRECT — Cover all 4 AWS security group resource types
resource_policy "aws_security_group" "check" {
    # inline ingress/egress blocks; cidr_blocks in rule.cidr_blocks
}
resource_policy "aws_security_group_rule" "check" {
    # standalone rules; cidr_blocks in attrs.cidr_blocks, type in attrs.type
}
resource_policy "aws_vpc_security_group_ingress_rule" "check" {
    # modern VPC API (recommended); uses attrs.cidr_ipv4 NOT cidr_blocks
    locals {
        has_public_cidr = core::try(attrs.cidr_ipv4, "") == "0.0.0.0/0"
    }
}
resource_policy "aws_default_security_group" "check" {
    # default SG; same inline structure as aws_security_group
}

# ❌ WRONG — Only covers Classic ELB; ALB/NLB (the modern standard) are silently skipped
resource_policy "aws_elb" "ssl_policy_check" { ... }

# ✅ CORRECT — Cover all 3 ELB resource types
resource_policy "aws_elb" "ssl_policy_check" {
    # Classic ELB: check aws_load_balancer_policy + aws_load_balancer_listener_policy
    # listeners use attrs.listener[*].lb_protocol
}
resource_policy "aws_lb_listener" "ssl_policy_check" {
    # ALB/NLB: ssl_policy is a direct attribute on the listener
    filter = core::contains(["HTTPS", "TLS"], core::try(attrs.protocol, ""))
    locals {
        ssl_policy = core::try(attrs.ssl_policy, "")
    }
    enforce {
        condition     = core::contains(local.allowed_policies, local.ssl_policy)
        error_message = "ALB/NLB listener must use an approved SSL/TLS security policy."
    }
}
resource_policy "aws_alb_listener" "ssl_policy_check" {
    # aws_alb_listener is an alias for aws_lb_listener — same attributes, same checks
    filter = core::contains(["HTTPS", "TLS"], core::try(attrs.protocol, ""))
    locals {
        ssl_policy = core::try(attrs.ssl_policy, "")
    }
    enforce {
        condition     = core::contains(local.allowed_policies, local.ssl_policy)
        error_message = "ALB/NLB listener must use an approved SSL/TLS security policy."
    }
}

# ❌ WRONG — resources where acl is NOT SET get defaulted to "private" and silently pass.
# This does NOT affect resources where acl IS set to "public-read" — those still fail correctly.
# The problem is resources with no acl configured are treated as compliant when they may not be.
resource_policy "aws_s3_bucket" "check" {
    locals {
        acl_value = core::try(attrs.acl, "private")  # missing acl → "private" → passes silently
        is_violation = core::contains(["public-read", "public-read-write"], local.acl_value)
    }
    enforce {
        condition = !local.is_violation
    }
}

# ✅ CORRECT — use filter to only evaluate resources that have the attribute set
resource_policy "aws_s3_bucket" "check" {
    filter = core::try(attrs.acl, null) != null  # skip resources with no acl configured

    locals {
        acl_value = attrs.acl  # safe after filter
        is_violation = core::contains(["public-read", "public-read-write"], local.acl_value)
    }
    enforce {
        condition = !local.is_violation
    }
}

# ❌ WRONG — tfpolicy test does not support expect_failure on data blocks
data "aws_iam_policy_document" "test" {
    expect_failure = true  # ERROR: unsupported argument
    attrs = { ... }
}

# ✅ CORRECT — expect_failure is only valid on resource blocks
resource "aws_s3_bucket" "test_violation" {
    expect_failure = true
    attrs = { acl = "public-read" }
}

# ❌ WRONG — Adds a false limitation and requires from_port == to_port,
# which incorrectly rejects valid port-range rules.
locals {
  # LIMITATION: TF Policy cannot dynamically iterate integer ranges.
  # This implementation requires from_port == to_port (single-port rule only).
  is_single_port  = local.from_port == local.to_port
  port_authorized = core::contains(local.authorized_ports, local.from_port)
  is_compliant    = local.is_single_port && local.port_authorized
}

# ❌ ALSO WRONG — core::range() with dynamic attrs.* values silently returns
# an empty list in the policytest framework (policytest limitation only).
# Do not use core::range() with dynamic port attributes.
locals {
  ports_in_range   = core::range(local.from_port, local.to_port + 1)  # empty in policytest!
  all_authorized   = core::alltrue([for p in local.ports_in_range : core::contains(local.authorized_ports, p)])
}

# ✅ CORRECT — count how many authorized ports fall inside [from_port, to_port].
# If the count equals the total number of ports in the range, all are authorized.
# Works with dynamic attrs.* values at both plan time and in policytest.
locals {
  authorized_ports     = [80, 443]   # or from input block
  from_port            = core::try(attrs.from_port, 0)
  to_port              = core::try(attrs.to_port, 0)
  authorized_in_range  = [for p in local.authorized_ports : p if p >= local.from_port && p <= local.to_port]
  all_ports_authorized = core::length(local.authorized_in_range) == (local.to_port - local.from_port + 1)
}

# ❌ WRONG — ebs_block_device is optional; absent on instances without EBS.
# If the block is missing, attrs.ebs_block_device[0] is null,
# local.ebs.encrypted is unknown, and condition = unknown throws:
#   Error: unknown condition
resource_policy "aws_instance" "ebs_encrypted" {
  locals {
    ebs          = attrs.ebs_block_device[0]     # ❌ null if block absent
    is_encrypted = local.ebs.encrypted           # ❌ unknown propagation
  }
  enforce {
    condition     = local.is_encrypted == true   # ❌ "unknown condition" ERROR
    error_message = "EBS block devices must be encrypted."
  }
}

# ❌ ALSO WRONG — Multiple optional nested blocks combined without null guards.
# If either block is absent, the condition evaluates to unknown.
resource_policy "aws_instance" "check" {
  locals {
    ebs_encrypted = attrs.ebs_block_device[0].encrypted        # ❌ crashes if absent
    nic_sg        = attrs.network_interface[0].security_groups  # ❌ crashes if absent
  }
  enforce {
    condition = local.ebs_encrypted == true && core::length(local.nic_sg) > 0
  }
}

# ✅ CORRECT — Use filter to scope to resources with the block,
# convert set to list, then use core::try for safe attribute access.
resource_policy "aws_instance" "ebs_encrypted" {
  # Only evaluate instances that have at least one EBS block device configured
  filter = core::try(attrs.ebs_block_device, null) != null && core::try(core::length(attrs.ebs_block_device), 0) > 0

  locals {
    ebs_devices   = [for d in attrs.ebs_block_device : d]  # set → list
    # Count devices that are NOT encrypted; if zero, all are encrypted
    unencrypted   = [for d in local.ebs_devices : d if !core::try(d.encrypted, false)]
    all_encrypted = core::length(local.unencrypted) == 0
  }

  enforce {
    condition     = local.all_encrypted
    error_message = "All EBS block devices on the instance must have encryption enabled."
  }
}

# ✅ CORRECT — Multiple optional nested blocks: guard each independently.
resource_policy "aws_instance" "check" {
  locals {
    ebs_raw       = core::try(attrs.ebs_block_device, null)
    has_ebs       = local.ebs_raw != null ? core::length(local.ebs_raw) > 0 : false
    ebs_devices   = local.has_ebs ? [for d in local.ebs_raw : d] : []
    # ✅ Use "no unencrypted devices" pattern — core::alltrue() does NOT exist
    unencrypted   = [for d in local.ebs_devices : d if core::try(d.encrypted, false) != true]
    all_encrypted = !local.has_ebs || core::length(local.unencrypted) == 0
  }

  enforce {
    condition     = local.all_encrypted
    error_message = "All EBS block devices must be encrypted."
  }
}

Error: Call to unknown function
There is no function named "anytrue" in namespace core::.

Error: Call to unknown function
There is no function named "alltrue" in namespace core::.

# ❌ WRONG — core::alltrue() does NOT exist in tfpolicy runtime
locals {
  all_encrypted = core::alltrue([for d in local.devices : core::try(d.encrypted, false)])
}

# ❌ WRONG — core::anytrue() does NOT exist in tfpolicy runtime
locals {
  any_public = core::anytrue([for r in local.rules : r.cidr == "0.0.0.0/0"])
}

# ✅ CORRECT — use core::length() with list comprehension instead of core::alltrue()
locals {
  # "all encrypted" = no unencrypted devices exist
  unencrypted   = [for d in local.devices : d if !core::try(d.encrypted, false)]
  all_encrypted = core::length(local.unencrypted) == 0
}

# ✅ CORRECT — use core::length() instead of core::anytrue()
locals {
  # "any public" = at least one public rule exists
  public_rules = [for r in local.rules : r if r.cidr == "0.0.0.0/0"]
  any_public   = core::length(local.public_rules) > 0
}

# ✅ CORRECT — boolean conditions in for...if: use plain && / || operators
locals {
  bad_rules = [
    for rule in attrs.ingress : rule
    if (rule.protocol == "tcp" && rule.from_port == 22)  # ✅ plain boolean — safe
  ]

  wide_open_rules = [
    for rule in attrs.ingress : rule
    if (rule.cidr_blocks == ["0.0.0.0/0"] || rule.ipv6_cidr_blocks == ["::/0"])  # ✅ safe
  ]
}

# ✅ ALSO CORRECT — two-pass pattern for complex conditions
locals {
  ingress_with_flags = [
    for rule in attrs.ingress : {
      rule        = rule
      is_ssh_tcp  = rule.protocol == "tcp" && rule.from_port == 22
    }
  ]
  ssh_tcp_rules = [for r in local.ingress_with_flags : r.rule if r.is_ssh_tcp]  # ✅ safe
}

# ❌ WRONG — destination_ranges exists but is null; core::try() does NOT catch null values.
# core::try() only catches attribute-access errors (missing attributes / index-out-of-bounds).
# When the attribute is present but set to null, core::try() returns null — NOT the default [].
# Passing null to core::length(), core::contains(), or a for-loop then causes:
#   "Invalid value for "list" parameter: argument must not be null"
resource_policy "google_compute_firewall" "check" {
  locals {
    dest_ranges = core::try(attrs.destination_ranges, [])            # ❌ returns null, not []
    bad_ranges  = [for r in local.dest_ranges : r if r == "0.0.0.0/0"]  # ❌ crashes
  }
}

# ❌ ALSO WRONG — looks safe but crashes in `locals`! TFPolicy does NOT short-circuit `&&` in locals.
# Both sides of `&&` are always evaluated, so core::length(null) is called even when
# logging_raw is null, causing: "Invalid value for 'collection': argument must not be null"
resource_policy "google_storage_bucket" "check" {
  locals {
    logging_raw     = core::try(attrs.logging, null)
    logging_present = local.logging_raw != null && core::length(local.logging_raw) > 0  # ❌ crashes when null!
  }
}

# ✅ CORRECT — explicitly check for null after core::try(), then fall back to [].
resource_policy "google_compute_firewall" "check" {
  locals {
    dest_ranges_raw = core::try(attrs.destination_ranges, null)
    dest_ranges     = local.dest_ranges_raw != null ? local.dest_ranges_raw : []
    bad_ranges      = [for r in local.dest_ranges : r if r == "0.0.0.0/0"]
  }
}

# ✅ CORRECT — use ternary to guard core::length() call; ternary DOES short-circuit.
resource_policy "google_storage_bucket" "check" {
  locals {
    logging_raw      = core::try(attrs.logging, null)
    logging_not_null = local.logging_raw != null
    logging_length   = local.logging_not_null ? core::length(local.logging_raw) : 0  # ✅ ternary safe
    logging_present  = local.logging_not_null && local.logging_length > 0
  }
}

# ✅ ALSO CORRECT — inline ternary in one line (equivalent to above).
locals {
  dest_ranges = core::try(attrs.destination_ranges, null) != null ? attrs.destination_ranges : []
}

# ❌ WRONG — condition = also does NOT short-circuit; crashes when local.X is null
enforce {
  condition = local.X != null && core::contains(["a", "b"], local.X)  # ❌ crashes!
}

# ✅ CORRECT — use ternary inside the locals block, then reference in condition
locals {
  is_allowed = local.X != null ? core::contains(["a", "b"], local.X) : false
}
enforce {
  condition = local.is_allowed
}

# ❌ WRONG — for...if predicate also does NOT short-circuit; crashes when r.field is absent
violating = [
  for r in local.rules : r
  if core::try(r.field, null) != null && core::contains(["a", "b"], r.field)  # ❌ crashes!
]

# ✅ CORRECT — use ternary in for...if predicate; also wrap the second access with core::try
violating = [
  for r in local.rules : r
  if (core::try(r.field, null) != null ? core::contains(["a", "b"], core::try(r.field, "")) : false)
]

# ❌ WRONG — The same allowlist is copied into every resource_policy block.
# Changing the list requires editing multiple blocks, which is error-prone.
resource_policy "azurerm_linux_virtual_machine" "allowed_sizes" {
  locals {
    allowed_sizes = ["Standard_D2s_v3", "Standard_D4s_v3"]
  }
  enforce {
    condition     = core::contains(local.allowed_sizes, core::try(attrs.size, ""))
    error_message = "VM size is not in the allowed list."
  }
}

resource_policy "azurerm_windows_virtual_machine" "allowed_sizes" {
  locals {
    allowed_sizes = ["Standard_D2s_v3", "Standard_D4s_v3"]  # ❌ duplicated value
  }
  enforce {
    condition     = core::contains(local.allowed_sizes, core::try(attrs.size, ""))
    error_message = "VM size is not in the allowed list."
  }
}

# ✅ CORRECT — Extract the shared value to a top-level locals block.
# All policy blocks reference it via local.<name>. One change updates every policy.
locals {
  allowed_sizes = ["Standard_D2s_v3", "Standard_D4s_v3"]
}

resource_policy "azurerm_linux_virtual_machine" "allowed_sizes" {
  enforce {
    condition     = core::contains(local.allowed_sizes, core::try(attrs.size, ""))
    error_message = "VM size is not in the allowed list."
  }
}

resource_policy "azurerm_windows_virtual_machine" "allowed_sizes" {
  enforce {
    condition     = core::contains(local.allowed_sizes, core::try(attrs.size, ""))
    error_message = "VM size is not in the allowed list."
  }
}

# ❌ WRONG — The + operator does not concatenate strings in tfpolicy
locals {
  pattern = "^com\\.amazonaws\\..+\\." + input.service_name  # ❌ runtime error
}

# ✅ CORRECT — Use ${ } interpolation for dynamic string building
locals {
  pattern        = "^com\\.amazonaws\\..+\\.${input.service_name}"
  error_msg      = "Service name must match pattern com.amazonaws.<region>.${input.service_name}."
}

# ✅ ALSO CORRECT — Avoid dynamic regex patterns entirely; use core::contains_substring
locals {
  service_ok = core::startswith(local.svc, "com.amazonaws.") &&
               core::contains_substring(local.svc, input.service_name)
}

# ❌ WRONG — enforcement_level appears twice, once before each enforce block
resource_policy "aws_vpc_endpoint" "check" {
  locals { ... }

  enforcement_level = "advisory"
  enforce {
    condition     = local.is_interface
    error_message = "..."
  }

  enforcement_level = "advisory"   # ❌ ERROR: already defined above
  enforce {
    condition     = local.service_matches
    error_message = "..."
  }
}

# ✅ CORRECT — enforcement_level appears exactly once for the whole block
resource_policy "aws_vpc_endpoint" "check" {
  locals { ... }

  enforcement_level = "advisory"   # ✅ declared once

  enforce {
    condition     = local.is_interface
    error_message = "..."
  }

  enforce {
    condition     = local.service_matches
    error_message = "..."
  }
}

# ❌ WRONG — keys defined inside the for-object literal are NOT in local scope
locals {
  origin_checks = [
    for origin in local.origins : {
      origin_domain  = core::try(origin.domain_name, "")
      is_s3_origin   = core::contains_substring(local.origin_domain, ".s3.")  # ❌ local.origin_domain is unknown here
      is_compliant   = !local.is_s3_origin || local.has_oac_id                # ❌ local.is_s3_origin is unknown here
    }
  ]
}

# ✅ CORRECT — repeat the expression inline; each key is independent
locals {
  origin_checks = [
    for origin in local.origins : {
      origin_domain  = core::try(origin.domain_name, "")
      is_s3_origin   = core::contains_substring(core::try(origin.domain_name, ""), ".s3.")
      is_compliant   = !core::contains_substring(core::try(origin.domain_name, ""), ".s3.") ||
                       (core::try(origin.oac_id, null) != null && core::try(origin.oac_id, "") != "")
    }
  ]
}

# ✅ ALSO CORRECT — two-pass pattern: compute properties first, then combine
locals {
  origins_with_props = [
    for origin in local.origins : {
      domain       = core::try(origin.domain_name, "")
      has_oac      = core::try(origin.origin_access_control_id, null) != null
    }
  ]
  # Now the second pass can reference the object's own keys by iterating:
  origin_checks = [
    for o in local.origins_with_props : {
      is_s3        = core::contains_substring(o.domain, ".s3.")
      is_compliant = !core::contains_substring(o.domain, ".s3.") || o.has_oac
    }
  ]
}

# ❌ WRONG — the true branch produces an object with "Statement" key,
# but the false branch is an empty object {} without that key
locals {
  policy_doc = local.appears_inline ? core::try(core::jsondecode(local.policy_value), {}) : {}
  statements = core::try(local.policy_doc.Statement, [])
}

# ✅ CORRECT — both branches must produce objects with the SAME set of keys
locals {
  policy_doc = local.appears_inline
    ? core::try(core::jsondecode(local.policy_value), { Statement = [] })
    : { Statement = [] }
  statements = core::try(local.policy_doc.Statement, [])
}

# ✅ ALSO CORRECT — avoid the ternary entirely; use core::try for the safe path
locals {
  # Always parse; core::try returns empty-statement fallback if decoding fails or is not applicable
  policy_doc = core::try(core::jsondecode(local.policy_value), { Statement = [] })
  statements = core::try(local.policy_doc.Statement, [])
}

# ❌ WRONG — inputs block placed INSIDE a resource test block
resource "aws_db_instance" "pass_instance" {
  inputs = {            # ❌ inputs is not valid inside resource {}
    resource_type = "aws_db_instance"
    source_type   = "db-instance"
  }
  attrs = { ... }
}

# ✅ CORRECT — inputs is a TOP-LEVEL block in the policytest file,
# and it applies to ALL test cases in that file
policytest {
  targets = ["my-policy.policy.hcl"]
}

inputs {                # ✅ top-level block, applies to all resources below
  resource_type = "aws_db_instance"
  source_type   = "db-instance"
}

resource "aws_db_instance" "pass_instance" {
  attrs = { ... }
}

# ❌ WRONG — data_policy is NOT a valid policy block type
data_policy "aws_iam_policy_document" "permissive_actions_denied" {
  enforce {
    condition     = local.no_star_actions
    error_message = "IAM policy documents must not use wildcard actions."
  }
}

# ✅ CORRECT — use resource_policy for actual Terraform resources
# To check IAM policy content, parse the inline_policy or the policy document
# that is attached to an actual resource (aws_iam_policy, aws_iam_role, etc.)
resource_policy "aws_iam_policy" "permissive_actions_denied" {
  locals {
    policy_doc = core::try(core::jsondecode(core::try(attrs.policy, "{}")), { Statement = [] })
    statements = core::try(local.policy_doc.Statement, [])
    star_stmts = [for s in local.statements : s if core::contains(core::try(s.Action, []), "*")]
    no_star_actions = core::length(local.star_stmts) == 0
  }
  enforce {
    condition     = local.no_star_actions
    error_message = "IAM managed policies must not use wildcard (*) actions."
  }
}

# ❌ WRONG — same local variable name defined twice in the same locals {} block
resource_policy "aws_vpc" "flow_logging_enabled" {
  locals {
    flow_logs      = core::getresources("aws_flow_log", {})
    matching_logs  = [for fl in local.flow_logs : fl if fl.vpc_id == attrs.id]
    flow_logs      = core::try(attrs.enable_dns_support, false)  # ❌ duplicate name!
  }
  enforce { ... }
}

# ❌ WRONG — same resource_policy block defined twice (same type + name)
resource_policy "aws_vpc" "flow_logging_enabled" {
  locals { ... }
  enforce { condition = ... }
}

resource_policy "aws_vpc" "flow_logging_enabled" {  # ❌ duplicate!
  locals { ... }
  enforce { condition = ... }
}

# ✅ CORRECT — unique local names + all checks in a single resource_policy block
# Fixes both mistakes above:
#   1. Each local has a distinct name (flow_logs vs dns_support) — no duplicate-name error.
#   2. Both checks live in one "aws_vpc" / "flow_logging_enabled" block with two enforce
#      blocks — no duplicate-policy-block error.
# NOTE: aws_flow_log.vpc_id links to the parent VPC — filter inline by attrs.id; never use
# a top-level core::getresources("aws_flow_log", {}) + HCL for-loop filter (Mistake 13).
resource_policy "aws_vpc" "flow_logging_enabled" {
  locals {
    flow_logs    = core::getresources("aws_flow_log", { vpc_id = attrs.id })
    dns_support  = core::try(attrs.enable_dns_support, false)  # unique name — no conflict
    has_flow_log = core::length(local.flow_logs) > 0
  }
  enforce {
    condition     = local.has_flow_log
    error_message = "VPC must have at least one flow log configured."
  }
  enforce {
    condition     = local.dns_support
    error_message = "VPC must have DNS support enabled."
  }
}

# ❌ WRONG — verbose chain, hard to maintain
locals {
  ssl_mode   = core::try(attrs.ssl_mode, null)
  is_valid   = local.ssl_mode == "require" || local.ssl_mode == "verify-ca" || local.ssl_mode == "verify-full"
}

# ✅ CORRECT — core::contains() with a named list local
locals {
  ssl_mode     = core::try(attrs.ssl_mode, "none")
  valid_modes  = ["require", "verify-ca", "verify-full"]
  is_valid     = core::contains(local.valid_modes, local.ssl_mode)
}
enforce {
  condition     = local.is_valid
  error_message = "Attribute 'ssl_mode' must be one of: require, verify-ca, verify-full."
}

# ❌ WRONG — 5 separate startswith calls for versions 1.x through 5.x
locals {
  is_version_lt_6 = (
    core::startswith(local.engine_version, "1.") ||
    core::startswith(local.engine_version, "2.") ||
    core::startswith(local.engine_version, "3.") ||
    core::startswith(local.engine_version, "4.") ||
    core::startswith(local.engine_version, "5.")
  )
}

# ✅ CORRECT — core::regex() matches all versions with major version 1–5 in one expression
# Use in filter to skip resources where engine_version is >= 6.0 or unset
filter = core::try(attrs.engine_version, "") != "" &&
         core::try(core::regex("^[1-5]\\.", core::try(attrs.engine_version, "")), null) != null

locals {
  auth_token = core::try(attrs.auth_token, "")
}
enforce {
  condition     = local.auth_token != null && local.auth_token != ""
  error_message = "Attribute 'auth_token' must be set when 'engine_version' < 6.0."
}

# ❌ WRONG — anchored on the companion type; buckets with no public_access_block resource silently pass
resource_policy "aws_s3_bucket_public_access_block" "block_public_access" {
  locals {
    block_public_acls = core::try(attrs.block_public_acls, false)
  }
  enforce {
    condition     = local.block_public_acls == true
    error_message = "S3 bucket must block public ACLs."
  }
}

# ✅ CORRECT — anchored on the parent; a missing companion resource = false → violation is caught
# NOTE: apply-time cross-resource reference; resolves correctly at apply time.
resource_policy "aws_s3_bucket" "block_public_access" {
  locals {
    public_access_block     = core::getresources("aws_s3_bucket_public_access_block", {
      bucket = attrs.id
    })
    block_public_acls       = core::try(local.public_access_block[0].block_public_acls, false)
    block_public_policy     = core::try(local.public_access_block[0].block_public_policy, false)
    ignore_public_acls      = core::try(local.public_access_block[0].ignore_public_acls, false)
    restrict_public_buckets = core::try(local.public_access_block[0].restrict_public_buckets, false)
  }

  enforce {
    condition     = local.block_public_acls && local.block_public_policy && local.ignore_public_acls && local.restrict_public_buckets
    error_message = "S3 bucket '${attrs.bucket}' must have all four public access block settings enabled."
  }
}

# ❌ INCOMPLETE — only catches standalone managed policies
resource_policy "aws_iam_policy" "no_admin_privileges" {
  locals {
    statements    = core::try(core::jsondecode(core::try(attrs.policy, "{}")).Statement, [])
    admin_stmts   = [for s in local.statements : s if ...]
  }
  enforce { condition = core::length(local.admin_stmts) == 0 ... }
}
# Missing: aws_iam_role_policy, aws_iam_user_policy, aws_iam_group_policy

# ✅ CORRECT — cover all 4 inline policy resource types; each has attrs.policy (JSON string)
# Each resource_policy block is fully self-contained — attrs is only available inside a
# resource_policy, module_policy, or provider_policy block, not in top-level locals.

resource_policy "aws_iam_policy" "no_admin_privileges" {
  locals {
    statements  = core::try(core::jsondecode(core::try(attrs.policy, "{}")).Statement, [])
    admin_stmts = [for s in local.statements : s if core::lower(core::try(s.Effect, "Allow")) == "allow" && core::contains(core::try(s.Action, []), "*") && core::contains(core::try(s.Resource, []), "*")]
  }
  enforce {
    condition     = core::length(local.admin_stmts) == 0
    error_message = "IAM policies must not grant full admin privileges (*:* on *)."
  }
}

resource_policy "aws_iam_role_policy" "no_admin_privileges" {
  locals {
    statements  = core::try(core::jsondecode(core::try(attrs.policy, "{}")).Statement, [])
    admin_stmts = [for s in local.statements : s if core::lower(core::try(s.Effect, "Allow")) == "allow" && core::contains(core::try(s.Action, []), "*") && core::contains(core::try(s.Resource, []), "*")]
  }
  enforce {
    condition     = core::length(local.admin_stmts) == 0
    error_message = "IAM role inline policies must not grant full admin privileges (*:* on *)."
  }
}

resource_policy "aws_iam_user_policy" "no_admin_privileges" {
  locals {
    statements  = core::try(core::jsondecode(core::try(attrs.policy, "{}")).Statement, [])
    admin_stmts = [for s in local.statements : s if core::lower(core::try(s.Effect, "Allow")) == "allow" && core::contains(core::try(s.Action, []), "*") && core::contains(core::try(s.Resource, []), "*")]
  }
  enforce {
    condition     = core::length(local.admin_stmts) == 0
    error_message = "IAM user inline policies must not grant full admin privileges (*:* on *)."
  }
}

resource_policy "aws_iam_group_policy" "no_admin_privileges" {
  locals {
    statements  = core::try(core::jsondecode(core::try(attrs.policy, "{}")).Statement, [])
    admin_stmts = [for s in local.statements : s if core::lower(core::try(s.Effect, "Allow")) == "allow" && core::contains(core::try(s.Action, []), "*") && core::contains(core::try(s.Resource, []), "*")]
  }
  enforce {
    condition     = core::length(local.admin_stmts) == 0
    error_message = "IAM group inline policies must not grant full admin privileges (*:* on *)."
  }
}

# ❌ WRONG — core::try(rule.from_port, 0) appears 4 times; core::try(rule.to_port, 0) appears
# 4 times; core::try(rule.protocol, "") appears 5 times in one predicate.
violating_rules = [for rule in local.ingress_rules : rule if (core::contains(core::try(rule.cidr_blocks, []), "0.0.0.0/0") || core::contains(core::try(rule.ipv6_cidr_blocks, []), "::/0")) && (core::try(rule.protocol, "") == "all" || core::try(rule.protocol, "") == "-1" || (core::try(rule.protocol, "") == "tcp" && core::length([for p in input.authorized_tcp_ports : p if p >= core::try(rule.from_port, 0) && p <= core::try(rule.to_port, 0)]) != (core::try(rule.to_port, 0) - core::try(rule.from_port, 0) + 1)) || (core::try(rule.protocol, "") == "udp" && core::length([for p in input.authorized_udp_ports : p if p >= core::try(rule.from_port, 0) && p <= core::try(rule.to_port, 0)]) != (core::try(rule.to_port, 0) - core::try(rule.from_port, 0) + 1)))]