diff --git a/assets/queries/terraform/gcp_bom/dataflow/metadata.json b/assets/queries/terraform/gcp_bom/dataflow/metadata.json
new file mode 100644
index 00000000000..18d85e00fe7
--- /dev/null
+++ b/assets/queries/terraform/gcp_bom/dataflow/metadata.json
@@ -0,0 +1,12 @@
+{
+    "id": "895ed0d9-6fec-4567-8614-d7a74b599a53",
+    "queryName": "BOM - GCP Dataflow",
+    "severity": "TRACE",
+    "category": "Bill Of Materials",
+    "descriptionText": "A list of Dataflow resources found. Unified stream and batch data processing that's serverless, fast, and cost-effective.",
+    "descriptionUrl": "https://kics.io",
+    "platform": "Terraform",
+    "descriptionID": "5d614ad5",
+    "cloudProvider": "gcp"
+  }
+  
\ No newline at end of file
diff --git a/assets/queries/terraform/gcp_bom/dataflow/query.rego b/assets/queries/terraform/gcp_bom/dataflow/query.rego
new file mode 100644
index 00000000000..8200193ed69
--- /dev/null
+++ b/assets/queries/terraform/gcp_bom/dataflow/query.rego
@@ -0,0 +1,41 @@
+package Cx
+
+import data.generic.common as common_lib
+import data.generic.terraform as tf_lib
+
+CxPolicy[result] {
+	dataflow_job := input.document[i].resource.google_dataflow_job[name]
+
+	bom_output = {
+		"resource_type": "google_dataflow_job",
+		"resource_name": tf_lib.get_resource_name(dataflow_job, name),
+		"resource_accessibility": check_accessability(dataflow_job),
+		"resource_encryption": check_encrytion(dataflow_job),
+		"resource_vendor": "GCP",
+		"resource_category": "Streaming",
+	}
+
+	result := {
+		"documentId": input.document[i].id,
+		"searchKey": sprintf("google_dataflow_job[%s]", [name]),
+		"issueType": "BillOfMaterials",
+		"keyExpectedValue": "",
+		"keyActualValue": "",
+		"searchLine": common_lib.build_search_line(["resource", "google_dataflow_job", name], []),
+		"value": json.marshal(bom_output),
+	}
+}
+
+check_encrytion(resource) = enc_status {
+	common_lib.valid_key(resource, "kms_key_name")
+	enc_status := "encrypted"
+} else = enc_status {
+	enc_status := "unencrypted"
+}
+
+check_accessability(resource) = acc_status {
+	resource.ip_configuration == "WORKER_IP_PUBLIC"
+	acc_status := "public"
+} else = acc_status {
+	acc_status := "unknown"
+}
diff --git a/assets/queries/terraform/gcp_bom/dataflow/test/negative.tf b/assets/queries/terraform/gcp_bom/dataflow/test/negative.tf
new file mode 100644
index 00000000000..eea55ba7b0e
--- /dev/null
+++ b/assets/queries/terraform/gcp_bom/dataflow/test/negative.tf
@@ -0,0 +1,22 @@
+# negative sample
+resource "google_bigquery_dataset" "negative1" {
+  dataset_id                  = "example_dataset"
+  friendly_name               = "test"
+  description                 = "This is a test description"
+  location                    = "EU"
+  default_table_expiration_ms = 3600000
+
+  labels = {
+    env = "default"
+  }
+
+  access {
+    role          = "OWNER"
+    user_by_email = google_service_account.bqowner.email
+  }
+
+  access {
+    role   = "READER"
+    domain = "hashicorp.com"
+  }
+}
diff --git a/assets/queries/terraform/gcp_bom/dataflow/test/positive.tf b/assets/queries/terraform/gcp_bom/dataflow/test/positive.tf
new file mode 100644
index 00000000000..9d0f1fa8bba
--- /dev/null
+++ b/assets/queries/terraform/gcp_bom/dataflow/test/positive.tf
@@ -0,0 +1,33 @@
+resource "google_dataflow_job" "pubsub_stream" {
+    name = "tf-test-dataflow-job1"
+    template_gcs_path = "gs://my-bucket/templates/template_file"
+    temp_gcs_location = "gs://my-bucket/tmp_dir"
+    enable_streaming_engine = true
+    parameters = {
+      inputFilePattern = "${google_storage_bucket.bucket1.url}/*.json"
+      outputTopic    = google_pubsub_topic.topic.id
+    }
+    transform_name_mapping = {
+        name = "test_job"
+        env = "test"
+    }
+    on_delete = "cancel"
+}
+
+resource "google_dataflow_job" "pubsub_stream2" {
+    name = "tf-test-dataflow-job1"
+    template_gcs_path = "gs://my-bucket/templates/template_file"
+    temp_gcs_location = "gs://my-bucket/tmp_dir"
+    enable_streaming_engine = true
+    parameters = {
+      inputFilePattern = "${google_storage_bucket.bucket1.url}/*.json"
+      outputTopic    = google_pubsub_topic.topic.id
+    }
+    transform_name_mapping = {
+        name = "test_job"
+        env = "test"
+    }
+    on_delete = "cancel"
+    kms_key_name = "somekey"
+    ip_configuration = "WORKER_IP_PUBLIC"
+}
diff --git a/assets/queries/terraform/gcp_bom/dataflow/test/positive_expected_result.json b/assets/queries/terraform/gcp_bom/dataflow/test/positive_expected_result.json
new file mode 100644
index 00000000000..2af3f798c94
--- /dev/null
+++ b/assets/queries/terraform/gcp_bom/dataflow/test/positive_expected_result.json
@@ -0,0 +1,14 @@
+[
+	{
+		"queryName": "BOM - GCP Dataflow",
+		"severity": "TRACE",
+		"line": 1,
+		"fileName": "positive.tf"
+	},
+	{
+		"queryName": "BOM - GCP Dataflow",
+		"severity": "TRACE",
+		"line": 17,
+		"fileName": "positive.tf"
+	}
+]