terraform_plan: Add references_to & referenced_by to direct_refer…

…ences

.github/workflows/build_test.yml

@@ -7,15 +7,24 @@ jobs:
     runs-on: ubuntu-latest
     strategy:
       matrix:
-        python-version: ["3.8"]
+        python-version: ["3.8, 3.9, 3.10, 3.11"]
     name: Run unittests
     steps:
     - uses: actions/checkout@v3
     - uses: actions/setup-python@v4
+    - uses: hashicorp/setup-terraform@v2  # For testing the terraform_plan provider
       with:
         python-version: ${{ matrix.python-version }}
         cache: 'pip'
         cache-dependency-path: Pipfile.lock
     - run: pip install . && pip install pipenv && pipenv install --system -d
+    - run: |
+        mkdir ~/.aws
+        cat > ~/.aws/credentials <<EOL
+        [default]
+        aws_access_key_id = verysecret
+        aws_secret_access_key = verysecretttt
+        region=ap-southeast-1
+        EOL
     - run: pytest
 

.gitignore

@@ -131,4 +131,5 @@ dmypy.json
 .pyre/
 
 .DS_Store
-.local
+.local
+.test_tmp

src/tirith/providers/terraform_plan/handler.py

@@ -205,7 +205,118 @@ def direct_dependencies_operator(input_data: dict, provider_inputs: dict, output
         )
 
 
+def direct_references_operator_referenced_by(input_data: dict, provider_inputs: dict, outputs: list):
+    # Verify that all of the terraform_resource_type instances are
+    # referenced by `referenced_by`
+    # Idea:
+    # - Get all of the resource_type instances id save it in a list
+    # - Iterate through all of the referenced_by instances, read what it references to,
+    #   check if it is in the list of resource_type instances, if it is, pop the element
+    # - If the list is empty, then all of the resource_type instances are
+    #   referenced by `referenced_by` return true, otherwise false
+    config_resources = input_data.get("configuration", {}).get("root_module", {}).get("resources", [])
+    resource_type = provider_inputs.get("terraform_resource_type")
+    referenced_by = provider_inputs.get("referenced_by")
+
+    reference_target = set()
+    is_resource_found = False
+
+    # Loop for adding reference_target
+    for config_resource in config_resources:
+        if config_resource.get("type") != resource_type:
+            continue
+        reference_target.add(config_resource.get("address"))
+        is_resource_found = True
+
+    if not is_resource_found:
+        outputs.append(
+            {
+                "value": ProviderError(severity_value=1),
+                "err": f"resource_type: '{resource_type}' is not found (severity_value: 1)",
+                "meta": config_resources,
+            }
+        )
+        return
+
+    # Loop for removing reference_target
+    for config_resource in config_resources:
+        if config_resource.get("type") != referenced_by:
+            continue
+
+        for expression_val_dict in config_resource.get("expressions", {}).values():
+            if not isinstance(expression_val_dict, dict):
+                continue
+
+            for reference in expression_val_dict.get("references", []):
+                if reference in reference_target:
+                    reference_target.remove(reference)
+
+    is_all_referenced = len(reference_target) == 0
+    outputs.append({"value": is_all_referenced, "meta": config_resources})
+
+
+def direct_references_operator_references_to(input_data: dict, provider_inputs: dict, outputs: list):
+    # The exact opposite of `direct_references_operator_referenced_by`
+    config_resources = input_data.get("configuration", {}).get("root_module", {}).get("resources", [])
+    resource_type = provider_inputs.get("terraform_resource_type")
+    references_to = provider_inputs.get("references_to")
+
+    resource_type_count = 0
+    reference_count = 0
+    is_resource_found = False
+
+    for config_resource in config_resources:
+        if config_resource.get("type") != resource_type:
+            continue
+        is_resource_found = True
+        resource_type_count += 1
+
+        for expression_val_dict in config_resource.get("expressions", {}).values():
+            if not isinstance(expression_val_dict, dict):
+                continue
+
+            for reference in expression_val_dict.get("references", []):
+                reference_res_type = reference.split(".")[0]
+                if reference_res_type == references_to:
+                    reference_count += 1
+                    # We break early because most of the times the references
+                    # list contains something like this:
+                    # ["aws_s3_bucket.a.id", "aws_s3_bucket.a"]
+                    break
+
+    if not is_resource_found:
+        outputs.append(
+            {
+                "value": ProviderError(severity_value=1),
+                "err": f"resource_type: '{resource_type}' is not found (severity_value: 1)",
+                "meta": config_resources,
+            }
+        )
+        return
+
+    is_all_resource_type_references_to = resource_type_count == reference_count
+    outputs.append({"value": is_all_resource_type_references_to, "meta": config_resources})
+
+
 def direct_references_operator(input_data: dict, provider_inputs: dict, outputs: list):
+    referenced_by = provider_inputs.get("referenced_by")
+    references_to = provider_inputs.get("references_to")
+
+    if referenced_by is not None and references_to is not None:
+        outputs.append(
+            {
+                "value": ProviderError(severity_value=99),
+                "err": "Only one of `referenced_by` or `references_to` must be provided in the provider input (severity_value: 99))",
+            }
+        )
+        return
+
+    if referenced_by is not None:
+        return direct_references_operator_referenced_by(input_data, provider_inputs, outputs)
+
+    if references_to is not None:
+        return direct_references_operator_references_to(input_data, provider_inputs, outputs)
+
     config_resources = input_data.get("configuration", {}).get("root_module", {}).get("resources", [])
     resource_type = provider_inputs.get("terraform_resource_type")
 
@@ -241,7 +352,7 @@ def direct_references_operator(input_data: dict, provider_inputs: dict, outputs:
         outputs.append(
             {
                 "value": ProviderError(severity_value=1),
-                "err": f"resource_type: '{resource_type}' is not found",
+                "err": f"resource_type: '{resource_type}' is not found (severity_value: 1)",
                 "meta": config_resources,
             }
         )

tests/providers/terraform_plan/direct_references/fixtures/at_least_one_aws_s3_bucket_int_config_refers_to_s3_bucket.tirith.json

@@ -0,0 +1,22 @@
+{
+  "meta": {
+    "required_provider": "stackguardian/terraform_plan",
+    "version": "v1"
+  },
+  "evaluators": [
+    {
+      "id": "s3HasLifeCycleIntelligentTiering",
+      "description": "Make sure IntelligentTieringConfig references to S3 bucket",
+      "provider_args": {
+        "operation_type": "direct_references",
+        "terraform_resource_type": "aws_s3_bucket_intelligent_tiering_configuration"
+      },
+      "condition": {
+        "type": "Contains",
+        "value": "aws_s3_bucket",
+        "error_tolerance": 0
+      }
+    }
+  ],
+  "eval_expression": "s3HasLifeCycleIntelligentTiering"
+}

tests/providers/terraform_plan/direct_references/fixtures/elb_references_to_secroup.tirith.json

@@ -0,0 +1,23 @@
+{
+  "meta": {
+    "required_provider": "stackguardian/terraform_plan",
+    "version": "v1"
+  },
+  "evaluators": [
+    {
+      "id": "elbRefsToSecGroup",
+      "description": "Make sure ELBs references to security groups",
+      "provider_args": {
+        "operation_type": "direct_references",
+        "terraform_resource_type": "aws_elb",
+        "references_to": "aws_security_group"
+      },
+      "condition": {
+        "type": "Equals",
+        "value": true,
+        "error_tolerance": 0
+      }
+    }
+  ],
+  "eval_expression": "elbRefsToSecGroup"
+}

tests/providers/terraform_plan/direct_references/fixtures/elb_with_secgroup.tf

@@ -0,0 +1,38 @@
+provider "aws" {
+  region = "us-west-2"  # Change this to your desired AWS region
+}
+
+resource "aws_security_group" "elb_sg" {
+  name        = "elb-sg"
+  description = "Security Group for ELB"
+
+  ingress {
+    from_port   = 80
+    to_port     = 80
+    protocol    = "tcp"
+    cidr_blocks = ["0.0.0.0/0"]  # Allow all incoming HTTP traffic. Modify as needed.
+  }
+
+  egress {
+    from_port   = 0
+    to_port     = 0
+    protocol    = "-1"
+    cidr_blocks = ["0.0.0.0/0"]  # Allow all outgoing traffic. Modify as needed.
+  }
+}
+
+resource "aws_elb" "example" {
+  name               = "example-elb"
+  availability_zones = ["us-west-2a", "us-west-2b"]  # Change these based on your region and requirements
+
+  listener {
+    instance_port     = 80
+    instance_protocol = "http"
+    lb_port           = 80
+    lb_protocol       = "http"
+  }
+
+  security_groups = [aws_security_group.elb_sg.id]
+
+  # Optionally add health checks, instances, etc.
+}

tests/providers/terraform_plan/direct_references/fixtures/extra_elb_no_secgroup.tf

@@ -0,0 +1,54 @@
+provider "aws" {
+  region = "us-west-2"  # Change this to your desired AWS region
+}
+
+resource "aws_security_group" "elb_sg" {
+  name        = "elb-sg"
+  description = "Security Group for ELB"
+
+  ingress {
+    from_port   = 80
+    to_port     = 80
+    protocol    = "tcp"
+    cidr_blocks = ["0.0.0.0/0"]  # Allow all incoming HTTP traffic. Modify as needed.
+  }
+
+  egress {
+    from_port   = 0
+    to_port     = 0
+    protocol    = "-1"
+    cidr_blocks = ["0.0.0.0/0"]  # Allow all outgoing traffic. Modify as needed.
+  }
+}
+
+resource "aws_elb" "example" {
+  name               = "example-elb"
+  availability_zones = ["us-west-2a", "us-west-2b"]  # Change these based on your region and requirements
+
+  listener {
+    instance_port     = 80
+    instance_protocol = "http"
+    lb_port           = 80
+    lb_protocol       = "http"
+  }
+
+  security_groups = [aws_security_group.elb_sg.id]
+
+  # Optionally add health checks, instances, etc.
+}
+
+resource "aws_elb" "something" {
+  name               = "something-elb"
+  availability_zones = ["us-west-2a", "us-west-2b"]  # Change these based on your region and requirements
+
+  listener {
+    instance_port     = 80
+    instance_protocol = "http"
+    lb_port           = 80
+    lb_protocol       = "http"
+  }
+
+  security_groups = []
+
+  # Optionally add health checks, instances, etc.
+}

tests/providers/terraform_plan/direct_references/fixtures/fail_s3_referenced_by_to.tirith.json

@@ -0,0 +1,24 @@
+{
+  "meta": {
+    "required_provider": "stackguardian/terraform_plan",
+    "version": "v1"
+  },
+  "evaluators": [
+    {
+      "id": "s3HasLifeCycleIntelligentTiering",
+      "description": "Make sure all aws_s3_bucket are referenced by aws_s3_bucket_intelligent_tiering_configuration",
+      "provider_args": {
+        "operation_type": "direct_references",
+        "terraform_resource_type": "aws_s3_bucket",
+        "referenced_by": "aws_s3_bucket_intelligent_tiering_configuration",
+        "references_to": "aws_s3_bucket"
+      },
+      "condition": {
+        "type": "Equals",
+        "value": true,
+        "error_tolerance": 0
+      }
+    }
+  ],
+  "eval_expression": "s3HasLifeCycleIntelligentTiering"
+}

tests/providers/terraform_plan/direct_references/fixtures/s3_bucket_not_all_has_intelligent_tiering.tf

@@ -0,0 +1,25 @@
+provider "aws" {
+  region = "us-west-1" # Adjust the region as needed.
+}
+
+resource "aws_s3_bucket_intelligent_tiering_configuration" "example-entire-bucket" {
+  bucket = aws_s3_bucket.example.id
+  name   = "EntireBucket"
+
+  tiering {
+    access_tier = "DEEP_ARCHIVE_ACCESS"
+    days        = 180
+  }
+  tiering {
+    access_tier = "ARCHIVE_ACCESS"
+    days        = 125
+  }
+}
+
+resource "aws_s3_bucket" "example" {
+  bucket = "example"
+}
+
+resource "aws_s3_bucket" "bb" {
+  bucket = "bb"
+}

tests/providers/terraform_plan/direct_references/fixtures/s3_bucket_with_intelligent_tiering_config.tf

@@ -0,0 +1,21 @@
+provider "aws" {
+  region = "us-west-1" # Adjust the region as needed.
+}
+
+resource "aws_s3_bucket_intelligent_tiering_configuration" "example-entire-bucket" {
+  bucket = aws_s3_bucket.example.id
+  name   = "EntireBucket"
+
+  tiering {
+    access_tier = "DEEP_ARCHIVE_ACCESS"
+    days        = 180
+  }
+  tiering {
+    access_tier = "ARCHIVE_ACCESS"
+    days        = 125
+  }
+}
+
+resource "aws_s3_bucket" "example" {
+  bucket = "example"
+}

tests/providers/terraform_plan/direct_references/fixtures/s3_referenced_by_intelligent_tiering_config.tirith.json

@@ -0,0 +1,23 @@
+{
+  "meta": {
+    "required_provider": "stackguardian/terraform_plan",
+    "version": "v1"
+  },
+  "evaluators": [
+    {
+      "id": "s3HasLifeCycleIntelligentTiering",
+      "description": "Make sure all aws_s3_bucket are referenced by aws_s3_bucket_intelligent_tiering_configuration",
+      "provider_args": {
+        "operation_type": "direct_references",
+        "terraform_resource_type": "aws_s3_bucket",
+        "referenced_by": "aws_s3_bucket_intelligent_tiering_configuration"
+      },
+      "condition": {
+        "type": "Equals",
+        "value": true,
+        "error_tolerance": 0
+      }
+    }
+  ],
+  "eval_expression": "s3HasLifeCycleIntelligentTiering"
+}