adding wrapper script for python

.github/workflows/controls_behave_testing_storage.yml

@@ -0,0 +1,34 @@
+name: "Controls Behave Testing - Storage"
+
+on:
+  workflow_call:
+
+jobs:
+  run_behave_tests_storage:
+    runs-on: ubuntu-latest
+
+    steps:
+    - name: Checkout code
+      uses: actions/checkout@v4
+
+    - name: Configure AWS credentials
+      uses: aws-actions/configure-aws-credentials@v4
+      with:
+        aws-access-key-id: ${{ secrets.AWS_ACCESS_KEY_ID }}
+        aws-secret-access-key: ${{ secrets.AWS_SECRET_ACCESS_KEY }}
+        aws-region: "eu-west-3"
+
+    - uses: 'google-github-actions/auth@v2'
+      with:
+        credentials_json: '${{ secrets.GOOGLE_CREDENTIALS }}'
+
+    - name: Set up Python
+      uses: actions/setup-python@v2  # This action sets up Python
+
+    - name: Install dependencies
+      working-directory: ./src/control-catalog/behave_tests
+      run: pip install -r requirements.txt  # Install your Python dependencies
+
+    - name: Run Behave Tests
+      working-directory: ./src/control-catalog/behave_tests
+      run: python app.py storage

.github/workflows/default_pr_controls_testing_workflow.yml

@@ -13,8 +13,8 @@ jobs: # Can expand on this later if we'd like to
         secrets: inherit
 
     # Once all has been applied, execute the python script for testing/validation
-    run-behave-tests:
-        uses: ./.github/workflows/controls_testing_python_script.yml
+    run-storage-behave-tests:
+        uses: ./.github/workflows/controls_behave_testing_storage.yml
         needs: run-terraform-apply
         permissions:
           pull-requests: write
@@ -24,7 +24,7 @@ jobs: # Can expand on this later if we'd like to
     # Then, tear resources down
     run-terraform-destroy:
       uses: ./.github/workflows/controls_terraform_destroy.yml
-      needs: run-behave-tests
+      needs: run-storage-behave-tests
       permissions:
         pull-requests: write
         contents: read

src/control-catalog/behave_tests/app.py

@@ -0,0 +1,43 @@
+import sys
+import subprocess
+import logging
+
+from os import environ
+from glob import glob
+
+if environ.get("AWS_DEFAULT_REGION") is None:
+    environ["AWS_DEFAULT_REGION"] = "eu-west-3"  # Set to France
+
+
+class BehaveTester:
+
+    def __init__(self) -> None:
+        logging.basicConfig(level=logging.INFO)
+        self.logging = logging.getLogger(__name__)
+
+    def run_behave(self, service_name):
+        abs_behave_file_paths = glob("./**/*.feature", recursive=True)
+
+        for abs_behave_file_path in abs_behave_file_paths:
+            behave_test_dir = "/".join(abs_behave_file_path.split("/")[:-2])
+            if service_name in behave_test_dir:
+                print(f"Running behave for {service_name}")
+                try:
+                    subprocess.run(["behave", behave_test_dir], check=True)
+                except subprocess.CalledProcessError as err:
+                    self.logging.error(f"Behave tested failed: {err}")
+                    sys.exit(1)
+            else:
+                self.logging.info(f"Behave for service {service_name} not found!")
+                sys.exit(1)
+
+        self.logging.info(f"Behave tested passed for the {service_name} service!")
+
+
+if __name__ == "__main__":
+    if len(sys.argv) != 2:
+        print("Usage: python3 app.py <service_name>")
+        sys.exit(1)
+
+    tester = BehaveTester()
+    tester.run_behave(sys.argv[1])

src/control-catalog/behave_tests/ccc.os.c1/steps/ccc_os_c1.py → src/control-catalog/behave_tests/storage/object/ccc.os.c1/steps/ccc_os_c1.py

@@ -6,7 +6,7 @@
 from google.cloud import storage
 from behave import given, then, when
 
-logging.basicConfig(level=logging.DEBUG)
+logging.basicConfig(level=logging.INFO)
 
 STORAGE_BUCKET_NAME = "malicious-sb-ccc-os-c1"
 

src/control-catalog/behave_tests/ccc.os.c2/features/ccc_os_c2.feature → src/control-catalog/behave_tests/storage/object/ccc.os.c2/features/ccc_os_c2.feature

@@ -1,6 +1,8 @@
 Feature: (CCC.OS.C2) - Prevent object storage data encrypted for impact
 
-Scenario: Test Control CCC.OS.C2 AWS
+Scenario: Test Control CCC.OS.C2
     GIVEN you own the object storage bucket in AWS
-    WHEN a data plane request with an untrusted KMS key is made to the object storage bucket
+    AND you own the object storage bucket in GCP
+    WHEN a data plane request with an untrusted KMS key is made to the AWS object storage bucket
+    AND a data plane request with an untrusted KMS key is made to the GCP object storage bucket
     THEN the request should be denied 

src/control-catalog/behave_tests/ccc.os.c2/steps/ccc_os_c2.py → src/control-catalog/behave_tests/storage/object/ccc.os.c2/steps/ccc_os_c2.py

@@ -1,25 +1,34 @@
 import logging
 import boto3
 
+from google.cloud import storage
 from botocore.exceptions import ClientError
 from behave import given, then, when
 
-logging.basicConfig(level=logging.DEBUG)
+logging.basicConfig(level=logging.INFO)
 
 STORAGE_BUCKET_NAME = "malicious-sb-ccc-os-c2"
-UNTRUSTED_KEY_ALIAS = "alias/malicious-sb-untrusted-ccc-os-c2"
+UNTRUSTED_KEY_NAME = "malicious-sb-untrusted-ccc-os-c2"
+UNTRUSTED_KEY_ALIAS = f"alias/{UNTRUSTED_KEY_NAME}"
 
 
 @given("you own the object storage bucket in AWS")
-def verify_aws_bucket_exists(context):
+def aws_verify_bucket_exists(context):
     context.s3_client = boto3.client("s3")
     context.s3_client.get_bucket_policy(Bucket=STORAGE_BUCKET_NAME)
 
 
+@given("you own the object storage bucket in GCP")
+def gcp_verify_bucket_exists(context):
+    context.storage_client = storage.Client()
+    context.bucket = context.storage_client.bucket(STORAGE_BUCKET_NAME)
+    context.bucket.get_iam_policy()
+
+
 @when(
-    "a data plane request with an untrusted KMS key is made to the object storage bucket"
+    "a data plane request with an untrusted KMS key is made to the AWS object storage bucket"
 )
-def upload_obj_with_untrusted_key(context):
+def aws_upload_obj_with_untrusted_key(context):
     context.kms_client = boto3.client("kms")
     untrusted_key_arn = context.kms_client.describe_key(KeyId=UNTRUSTED_KEY_ALIAS)[
         "KeyMetadata"
@@ -38,6 +47,19 @@ def upload_obj_with_untrusted_key(context):
         context.s3_publish_error = str(err)
 
 
+@when(
+    "a data plane request with an untrusted KMS key is made to the GCP object storage bucket"
+)
+def gcp_upload_obj_with_untrusted_key(context):
+    # This control needs to be reviewed in more detail - we 
+    # can upload to the bucket with an untrusted key.
+    key_name = f"projects/common-cloud-controls-testing/locations/us-central1/keyRings/{STORAGE_BUCKET_NAME}-keyring/cryptoKeys/{UNTRUSTED_KEY_NAME}"
+    bucket = storage.Bucket(context.storage_client, STORAGE_BUCKET_NAME)
+    bucket.blob("test.txt", kms_key_name=key_name).upload_from_string(
+        "Hello, World"
+    )
+
+
 @then("the request should be denied")
 def validate_request_denied(context):
     print(context.s3_publish_error)

src/control-catalog/behave_tests/ccc.os.c3/steps/ccc_os_c3.py → src/control-catalog/behave_tests/storage/object/ccc.os.c3/steps/ccc_os_c3.py

@@ -6,7 +6,7 @@
 from google.cloud import storage
 from behave import given, then, when
 
-logging.basicConfig(level=logging.DEBUG)
+logging.basicConfig(level=logging.INFO)
 
 STORAGE_BUCKET_NAME = "malicious-sb-ccc-os-c3"
 

src/control-catalog/terraform_modules/storage/main.tf

@@ -13,6 +13,11 @@ module "aws_storage_object_ccc_os_c2" {
   bucket_name = var.bucket_name
 }
 
+module "gcp_storage_object_ccc_os_c2" {
+  source      = "./object/ccc.os.c2/gcp"
+  bucket_name = var.bucket_name
+}
+
 module "aws_storage_object_ccc_os_c3" {
   source      = "./object/ccc.os.c3/aws"
   bucket_name = var.bucket_name

src/control-catalog/terraform_modules/storage/object/ccc.os.c2/gcp/main.tf

@@ -1,9 +1,56 @@
 resource "google_storage_bucket" "malicious_storage_bucket" {
   name          = "${var.bucket_name}-ccc-os-c2"
-  location      = "US"
+  location      = "us-central1"
   force_destroy = true
 
   versioning {
     enabled = true
   }
-}
+  encryption {
+    default_kms_key_name = google_kms_crypto_key.trusted_cmek.id
+  }
+
+  uniform_bucket_level_access = true
+}
+
+data "google_iam_policy" "policy" {
+  binding {
+    role    = "roles/storage.objectCreator"
+    members = ["user:*"]
+    condition {
+        title      = "Deny unencrypted uploads"
+        description = "Only allow objects to be uploaded with a specific KMS key"
+        expression = "resource.name.startsWith(\"projects/common-cloud-controls-testing/buckets/${google_storage_bucket.malicious_storage_bucket.name}/objects\") && !request.resource.labels.kms_key_name.startsWith(\"projects/common-cloud-controls-testing/locations/us-central1/keyRings/${google_kms_key_ring.keyring.id}/cryptoKeys/${google_kms_crypto_key.trusted_cmek.name}\")"
+      }
+  }
+}
+
+resource "google_storage_bucket_iam_policy" "name" {
+  bucket = google_storage_bucket.malicious_storage_bucket.name
+  policy_data = data.google_iam_policy.policy.policy_data
+}
+
+resource "google_kms_key_ring" "keyring" {
+  name     = "${var.bucket_name}-ccc-os-c2-keyring"
+  location = "us-central1"
+}
+
+resource "google_kms_crypto_key" "trusted_cmek" {
+  name            = "${var.bucket_name}-trusted-ccc-os-c2"
+  key_ring        = google_kms_key_ring.keyring.id
+  rotation_period = "7776000s"
+
+  lifecycle {
+    prevent_destroy = false
+  }
+}
+
+resource "google_kms_crypto_key" "untrusted_cmek" {
+  name            = "${var.bucket_name}-untrusted-ccc-os-c2"
+  key_ring        = google_kms_key_ring.keyring.id
+  rotation_period = "7776000s"
+
+  lifecycle {
+    prevent_destroy = false
+  }
+}