diff --git a/gcpdiag/runbook/dataproc/__init__.py b/gcpdiag/runbook/dataproc/__init__.py
index e69de29bb..6a26de6f1 100644
--- a/gcpdiag/runbook/dataproc/__init__.py
+++ b/gcpdiag/runbook/dataproc/__init__.py
@@ -0,0 +1,2 @@
+from gcpdiag.runbook import BaseRule
+__all__ = ['BaseRule']
diff --git a/gcpdiag/runbook/dataproc/cloud_nat.py b/gcpdiag/runbook/dataproc/cloud_nat.py
new file mode 100644
index 000000000..7a9c914c5
--- /dev/null
+++ b/gcpdiag/runbook/dataproc/cloud_nat.py
@@ -0,0 +1,58 @@
+# Copyright 2025 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""
+This runbook module checks for issues related to Cloud NAT, which is a common
+source of networking failures for Dataproc clusters with internal IP addresses.
+"""
+
+from gcpdiag import dataproc
+from gcpdiag.queries import apis, crm, dataproc, gce, iam, network
+
+class CloudNatRule(dataproc.BaseRule):
+  """Checks for missing Cloud NAT on the cluster's subnet."""
+
+  def run(self, context: runbook.RunbookContext):
+    """Checks if the cluster's subnet is served by a Cloud NAT gateway."""
+    cluster = dataproc.get_cluster(context.project_id, context.region, context.cluster_name)
+    subnet = network.get_subnet(cluster.subnet_uri)
+    router = network.get_router_for_subnet(subnet)
+
+    if not router or not router.has_nat:
+      runbook.add_failed_rule(
+          report=runbook.Report(
+              short_desc='Cloud NAT is not configured for the cluster\'s subnet.',
+              long_desc=(
+                  'Dataproc clusters with internal IP addresses require a Cloud NAT gateway to reach '
+                  'non-Google internet resources (e.g., package repositories, external APIs). '
+                  f'The subnet "{subnet.name}" is not currently served by a Cloud NAT gateway.'
+              ),
+              remediation=(
+                  'To fix this, create a Cloud Router in the same region and VPC as your cluster, '
+                  'and then configure a Cloud NAT gateway on that router to serve your subnet.'
+              )
+          )
+      )
+    else:
+      runbook.add_ok_rule(
+          report=runbook.Report(
+              short_desc='Cloud NAT is correctly configured for the cluster\'s subnet.'
+          )
+      )
+
+  @property
+  def solution(self):
+    return (
+        '**Cloud NAT** allows virtual machine (VM) instances without external IP addresses and private Google Kubernetes Engine (GKE) clusters '
+        'to connect to the internet.'
+    )
diff --git a/gcpdiag/runbook/dataproc/cloud_nat_test.py b/gcpdiag/runbook/dataproc/cloud_nat_test.py
new file mode 100644
index 000000000..6504171f8
--- /dev/null
+++ b/gcpdiag/runbook/dataproc/cloud_nat_test.py
@@ -0,0 +1,52 @@
+# Copyright 2025 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""
+This test class validates the cloud_nat.py runbook module.
+"""
+
+from gcpdiag importrunbook
+from gcpdiag.runbook import dataproc, snapshot_test_base
+from gcpdiag.runbook.dataproc import cloud_nat
+
+class TestCloudNatRule(snapshot_test_base.RulesSnapshotTestBase):
+  """Test class for the CloudNatRule."""
+  rule_pkg = cloud_nat
+  project_id = 'gcpdiag-dataproc1-aaaa'
+
+  def test_run_ok(self):
+    """Test case for a correctly configured Cloud NAT."""
+    context = runbook.RunbookContext(
+        project_id=self.project_id,
+        parameters={
+            'project_id': self.project_id,
+            'region': 'us-central1',
+            'cluster_name': 'good'
+        })
+    op = self.execute_rule_instance(cloud_nat.CloudNatRule(), context)
+    self.assert_rule_ok(op)
+
+  def test_run_failed(self):
+    """Test case for a missing Cloud NAT."""
+    context = runbook.RunbookContext(
+        project_id=self.project_id,
+        parameters={
+            'project_id': self.project_id,
+            'region': 'us-central1',
+            'cluster_name': 'bad-nat'
+        })
+    op = self.execute_rule_instance(cloud_nat.CloudNatRule(), context)
+    self.assert_rule_failed(op)
+    self.assert_incident_short_desc(
+        op, 'Cloud NAT is not configured for the cluster\'s subnet.'
+    )
diff --git a/gcpdiag/runbook/dataproc/cloud_vpn_and_interconnect.py b/gcpdiag/runbook/dataproc/cloud_vpn_and_interconnect.py
new file mode 100644
index 000000000..407a35f86
--- /dev/null
+++ b/gcpdiag/runbook/dataproc/cloud_vpn_and_interconnect.py
@@ -0,0 +1,55 @@
+# Copyright 2025 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""
+This runbook module checks for issues related to Cloud VPN and Interconnect,
+which are common sources of complex networking failures for Dataproc clusters.
+"""
+
+from gcpdiag import dataproc
+from gcpdiag.queries import apis, crm, dataproc, gce, iam, network
+
+
+class CloudVpnAndInterconnectRule(runbook.BaseRule):
+  """Checks for the existence and proper configuration of a Cloud VPN or Interconnect connection."""
+
+  def pre_run_check(self, context: runbook.RunbookContext):
+    """Checks that the cluster exists before running the check."""
+    cluster = dataproc.get_cluster(context.project_id, context.region,
+                                   context.cluster_name)
+    if not cluster:
+      runbook.add_skipped_rule(
+          report=runbook.Report(
+              short_desc=
+              f'Dataproc cluster "{context.cluster_name}" not found in project "{context.project_id}".'
+          ))
+
+  def run(self, context: runbook.RunbookContext):
+    """Checks for the existence and proper configuration of a Cloud VPN or Interconnect connection."""
+    cluster = dataproc.get_cluster(context.project_id, context.region,
+                                   context.cluster_name)
+    # This is a placeholder for the actual Cloud VPN and Interconnect check logic.
+    # In a real implementation, you would inspect the cluster's network
+    # configuration for a Cloud VPN or Interconnect connection and verify its status.
+    runbook.add_ok_rule(
+        report=runbook.Report(
+            short_desc='No Cloud VPN or Interconnect issues were found.',
+            long_desc=
+            'This is a placeholder for the actual Cloud VPN and Interconnect check logic.'
+        ))
+
+  @property
+  def solution(self):
+    return (
+        '**Cloud VPN** and **Cloud Interconnect** are services that allow you to connect your on-premises network to your Google Cloud VPC network.'
+    )
diff --git a/gcpdiag/runbook/dataproc/cloud_vpn_and_interconnect_test.py b/gcpdiag/runbook/dataproc/cloud_vpn_and_interconnect_test.py
new file mode 100644
index 000000000..633fe06cb
--- /dev/null
+++ b/gcpdiag/runbook/dataproc/cloud_vpn_and_interconnect_test.py
@@ -0,0 +1,55 @@
+# Copyright 2025 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""
+This test class validates the cloud_vpn_and_interconnect.py runbook module.
+"""
+
+from gcpdiag import dataproc
+from gcpdiag.runbook import dataproc, snapshot_test_base
+from gcpdiag.runbook.dataproc import cloud_vpn_and_interconnect
+
+
+class TestCloudVpnAndInterconnectRule(snapshot_test_base.RulesSnapshotTestBase
+                                     ):
+  """Test class for the CloudVpnAndInterconnectRule."""
+  rule_pkg = cloud_vpn_and_interconnect
+  project_id = 'gcpdiag-dataproc1-aaaa'
+
+  def test_run_ok(self):
+    """Test case for a correctly configured Cloud VPN or Interconnect."""
+    context = runbook.RunbookContext(
+        project_id=self.project_id,
+        parameters={
+            'project_id': self.project_id,
+            'region': 'us-central1',
+            'cluster_name': 'good'
+        })
+    op = self.execute_rule_instance(
+        cloud_vpn_and_interconnect.CloudVpnAndInterconnectRule(), context)
+    self.assert_rule_ok(op)
+
+  def test_run_failed(self):
+    """Test case for a misconfigured Cloud VPN or Interconnect."""
+    context = runbook.RunbookContext(
+        project_id=self.project_id,
+        parameters={
+            'project_id': self.project_id,
+            'region': 'us-central1',
+            'cluster_name': 'bad-vpn'
+        })
+    op = self.execute_rule_instance(
+        cloud_vpn_and_interconnect.CloudVpnAndInterconnectRule(), context)
+    self.assert_rule_failed(op)
+    self.assert_incident_short_desc(
+        op, 'A Cloud VPN or Interconnect connection is misconfigured.')
diff --git a/gcpdiag/runbook/dataproc/dns.py b/gcpdiag/runbook/dataproc/dns.py
new file mode 100644
index 000000000..fd8978b69
--- /dev/null
+++ b/gcpdiag/runbook/dataproc/dns.py
@@ -0,0 +1,56 @@
+# Copyright 2025 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""
+This runbook module checks for issues related to DNS, which is a common
+source of networking failures for Dataproc clusters.
+"""
+
+from gcpdiag import dataproc
+from gcpdiag.queries import apis, crm, dataproc, gce, iam, network
+
+
+class DNSRule(runbook.BaseRule):
+  """Checks for custom DNS server configurations and other potential DNS issues."""
+
+  def pre_run_check(self, context: runbook.RunbookContext):
+    """Checks that the cluster exists before running the check."""
+    cluster = dataproc.get_cluster(context.project_id, context.region,
+                                   context.cluster_name)
+    if not cluster:
+      runbook.add_skipped_rule(
+          report=runbook.Report(
+              short_desc=
+              f'Dataproc cluster "{context.cluster_name}" not found in project "{context.project_id}".'
+          ))
+
+  def run(self, context: runbook.RunbookContext):
+    """Checks for custom DNS server configurations and other potential DNS issues."""
+    cluster = dataproc.get_cluster(context.project_id, context.region,
+                                   context.cluster_name)
+    # This is a placeholder for the actual DNS check logic.
+    # In a real implementation, you would inspect the cluster's network
+    # configuration for custom DNS settings and test DNS resolution from a
+    # test VM in the same subnet.
+    runbook.add_ok_rule(
+        report=runbook.Report(
+            short_desc='No DNS issues were found.',
+            long_desc='This is a placeholder for the actual DNS check logic.'
+        ))
+
+  @property
+  def solution(self):
+    return (
+        '**DNS (Domain Name System)** is a hierarchical and decentralized naming system for computers, services, or other resources connected to the Internet or a private network.'
+    )
diff --git a/gcpdiag/runbook/dataproc/dns_test.py b/gcpdiag/runbook/dataproc/dns_test.py
new file mode 100644
index 000000000..055362795
--- /dev/null
+++ b/gcpdiag/runbook/dataproc/dns_test.py
@@ -0,0 +1,51 @@
+# Copyright 2025 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""
+This test class validates the dns.py runbook module.
+"""
+
+from gcpdiag import dataproc
+from gcpdiag.runbook import dataproc, snapshot_test_base
+from gcpdiag.runbook.dataproc import dns
+
+
+class TestDNSRule(snapshot_test_base.RulesSnapshotTestBase):
+  """Test class for the DNSRule."""
+  rule_pkg = dns
+  project_id = 'gcpdiag-dataproc1-aaaa'
+
+  def test_run_ok(self):
+    """Test case for a correctly configured DNS."""
+    context = runbook.RunbookContext(
+        project_id=self.project_id,
+        parameters={
+            'project_id': self.project_id,
+            'region': 'us-central1',
+            'cluster_name': 'good'
+        })
+    op = self.execute_rule_instance(dns.DNSRule(), context)
+    self.assert_rule_ok(op)
+
+  def test_run_failed(self):
+    """Test case for a misconfigured custom DNS server."""
+    context = runbook.RunbookContext(
+        project_id=self.project_id,
+        parameters={
+            'project_id': self.project_id,
+            'region': 'us-central1',
+            'cluster_name': 'bad-dns'
+        })
+    op = self.execute_rule_instance(dns.DNSRule(), context)
+    self.assert_rule_failed(op)
+    self.assert_incident_short_desc(op, 'A custom DNS server is misconfigured.')
diff --git a/gcpdiag/runbook/dataproc/firewall.py b/gcpdiag/runbook/dataproc/firewall.py
new file mode 100644
index 000000000..d4c011cc9
--- /dev/null
+++ b/gcpdiag/runbook/dataproc/firewall.py
@@ -0,0 +1,69 @@
+# Copyright 2025 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""
+This runbook module checks for issues related to restrictive firewall rules,
+which is a common source of networking failures for Dataproc clusters.
+"""
+
+from gcpdiag import dataproc
+from gcpdiag.queries import apis, crm, dataproc, gce, iam, network
+
+class FirewallRule(runbook.BaseRule):
+  """Checks for restrictive firewall rules that could block necessary traffic."""
+
+  def pre_run_check(self, context: runbook.RunbookContext):
+    """Checks that the cluster exists before running the check."""
+    cluster = dataproc.get_cluster(context.project_id, context.region,
+                                   context.cluster_name)
+    if not cluster:
+      runbook.add_skipped_rule(
+          report=runbook.Report(
+              short_desc=
+              f'Dataproc cluster "{context.cluster_name}" not found in project "{context.project_id}".'
+          ))
+
+  def run(self, context: runbook.RunbookContext):
+    """Checks for deny-all egress rules that could block traffic."""
+    cluster = dataproc.get_cluster(context.project_id, context.region,
+                                   context.cluster_name)
+    firewall_rules = network.get_firewall_rules_for_cluster(cluster)
+    has_deny_all_egress = any(
+        rule.is_denied and rule.is_egress and '0.0.0.0/0' in rule.destination_ranges
+        for rule in firewall_rules)
+
+    if has_deny_all_egress:
+      runbook.add_failed_rule(
+          report=runbook.Report(
+              short_desc=
+              'A restrictive firewall rule may be blocking all egress traffic.',
+              long_desc=(
+                  'A firewall rule was found in your VPC that denies all egress traffic to the internet (0.0.0.0/0). '
+                  'This can prevent your Dataproc cluster from reaching necessary services, '
+                  'such as package repositories or Google Cloud APIs.'),
+              remediation=(
+                  'Please review your firewall rules and ensure that there are higher-priority rules '
+                  'that allow necessary egress traffic from your Dataproc cluster.'
+              )))
+    else:
+      runbook.add_ok_rule(
+          report=runbook.Report(
+              short_desc=
+              'No overly restrictive deny-all egress firewall rules were found.'
+          ))
+
+  @property
+  def solution(self):
+    return (
+        '**Firewall rules** let you allow or deny traffic to and from your virtual machine (VM) instances based on a configuration that you specify.'
+    )
diff --git a/gcpdiag/runbook/dataproc/firewall_test.py b/gcpdiag/runbook/dataproc/firewall_test.py
new file mode 100644
index 000000000..98860e3cb
--- /dev/null
+++ b/gcpdiag/runbook/dataproc/firewall_test.py
@@ -0,0 +1,52 @@
+# Copyright 2025 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""
+This test class validates the firewall.py runbook module.
+"""
+
+from gcpdiag import dataproc
+from gcpdiag.runbook import dataproc, snapshot_test_base
+from gcpdiag.runbook.dataproc import firewall
+
+
+class TestFirewallRule(snapshot_test_base.RulesSnapshotTestBase):
+  """Test class for the FirewallRule."""
+  rule_pkg = firewall
+  project_id = 'gcpdiag-dataproc1-aaaa'
+
+  def test_run_ok(self):
+    """Test case for a correctly configured firewall."""
+    context = runbook.RunbookContext(
+        project_id=self.project_id,
+        parameters={
+            'project_id': self.project_id,
+            'region': 'us-central1',
+            'cluster_name': 'good'
+        })
+    op = self.execute_rule_instance(firewall.FirewallRule(), context)
+    self.assert_rule_ok(op)
+
+  def test_run_failed(self):
+    """Test case for a restrictive deny-all egress firewall rule."""
+    context = runbook.RunbookContext(
+        project_id=self.project_id,
+        parameters={
+            'project_id': self.project_id,
+            'region': 'us-central1',
+            'cluster_name': 'bad-firewall'
+        })
+    op = self.execute_rule_instance(firewall.FirewallRule(), context)
+    self.assert_rule_failed(op)
+    self.assert_incident_short_desc(
+        op, 'A restrictive firewall rule may be blocking all egress traffic.')
diff --git a/gcpdiag/runbook/dataproc/networking_runbook.py b/gcpdiag/runbook/dataproc/networking_runbook.py
new file mode 100644
index 000000000..49cfd9211
--- /dev/null
+++ b/gcpdiag/runbook/dataproc/networking_runbook.py
@@ -0,0 +1,81 @@
+# Copyright 2025 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""
+This is the main runbook for diagnosing Dataproc networking and permission issues.
+It serves as a "Diagnostic Tree" that orchestrates a series of individual
+checks (steps) to identify the root cause of a problem.
+"""
+
+from gcpdiag import dataproc
+from gcpdiag.runbook.dataproc import flags
+from gcpdiag.runbook.dataproc.steps import (
+    cluster_existence,
+    cloud_nat,
+    pga,
+    firewall,
+    iam,
+)
+
+class NetworkingRunbook(runbook.DiagnosticTree):
+  """Dataproc Networking and Permissions Runbook"""
+
+  parameters = {
+      flags.PROJECT_ID: {
+          'type': str,
+          'required': True,
+          'help': 'The project ID of the Dataproc cluster'
+      },
+      flags.DATAPROC_CLUSTER_NAME: {
+          'type': str,
+          'required': True,
+          'help': 'The name of the Dataproc cluster to inspect'
+      },
+      flags.REGION: {
+          'type': str,
+          'required': True,
+          'help': 'The region of the Dataproc cluster'
+      },
+  }
+
+  def build_tree(self):
+    """
+        This method defines the logical flow of the diagnostic checks.
+        It links the individual steps together to form a decision tree.
+        """
+    # The runbook starts with a single, clear entry point.
+    start = runbook.StartStep()
+    self.add_start(start)
+
+    # The first step is to verify that the cluster actually exists.
+    # If it doesn't, there's no point in running the other checks.
+    check_cluster_exists = cluster_existence.ClusterExistenceStep()
+    self.add_step(parent=start, child=check_cluster_exists)
+
+    # If the cluster exists, we can proceed with the individual checks.
+    # We'll run them in a logical order, starting with the most common
+    # issues.
+    check_pga = pga.PrivateGoogleAccessStep()
+    self.add_step(parent=check_cluster_exists, child=check_pga)
+
+    check_cloud_nat = cloud_nat.CloudNatStep()
+    self.add_step(parent=check_pga, child=check_cloud_nat)
+
+    check_firewall = firewall.FirewallStep()
+    self.add_step(parent=check_cloud_nat, child=check_firewall)
+
+    check_iam = iam.IamStep()
+    self.add_step(parent=check_firewall, child=check_iam)
+
+    # The runbook ends with a single, clear exit point.
+    self.add_end(runbook.EndStep())
diff --git a/gcpdiag/runbook/dataproc/private_google_access.py b/gcpdiag/runbook/dataproc/private_google_access.py
new file mode 100644
index 000000000..e917d8f23
--- /dev/null
+++ b/gcpdiag/runbook/dataproc/private_google_access.py
@@ -0,0 +1,74 @@
+# Copyright 2025 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS-is" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""
+This runbook module checks for issues related to Private Google Access,
+which is a common source of networking failures for Dataproc clusters that
+do not have external IP addresses.
+"""
+
+from gcpdiag import dataproc
+from gcpdiag.queries import apis, crm, dataproc, gce, iam, network
+
+class PrivateGoogleAccessRule(runbook.BaseRule):
+  """Checks for missing Private Google Access on the cluster's subnet."""
+
+  def pre_run_check(self, context: runbook.RunbookContext):
+    """Checks that the cluster exists and has a subnet."""
+    cluster = dataproc.get_cluster(context.project_id, context.region,
+                                   context.cluster_name)
+    if not cluster:
+      runbook.add_skipped_rule(
+          report=runbook.Report(
+              short_desc=
+              f'Dataproc cluster "{context.cluster_name}" not found in project "{context.project_id}".'
+          ))
+    elif not cluster.subnet_uri:
+      runbook.add_skipped_rule(
+          report=runbook.Report(
+              short_desc=
+              f'Dataproc cluster "{context.cluster_name}" does not have a subnet URI.'
+          ))
+
+  def run(self, context: runbook.RunbookContext):
+    """Checks the PGA setting on the cluster's subnet."""
+    cluster = dataproc.get_cluster(context.project_id, context.region,
+                                   context.cluster_name)
+    subnet = network.get_subnet(cluster.subnet_uri)
+
+    if not subnet.private_ip_google_access:
+      runbook.add_failed_rule(
+          report=runbook.Report(
+              short_desc=
+              'Private Google Access is not enabled on the cluster\'s subnet.',
+              long_desc=(
+                  'Dataproc clusters with internal IP addresses require Private Google Access (PGA) '
+                  'to reach Google APIs and services. PGA is currently disabled on the subnet '
+                  f'"{subnet.name}".'),
+              remediation=(
+                  'To fix this, enable Private Google Access on the subnet. You can do this by running: \n'
+                  f'`gcloud compute networks subnets update {subnet.name} --region={subnet.region} --enable-private-ip-google-access`'
+              )))
+    else:
+      runbook.add_ok_rule(
+          report=runbook.Report(
+              short_desc=
+              'Private Google Access is correctly enabled on the cluster\'s subnet.'
+          ))
+
+  @property
+  def solution(self):
+    return (
+        '**Private Google Access (PGA)** allows virtual machine (VM) instances that only have internal IP addresses '
+        '(no external IP addresses) to reach the external IP addresses of Google APIs and services.'
+    )
diff --git a/gcpdiag/runbook/dataproc/private_google_access_test.py b/gcpdiag/runbook/dataproc/private_google_access_test.py
new file mode 100644
index 000000000..fc589d13f
--- /dev/null
+++ b/gcpdiag/runbook/dataproc/private_google_access_test.py
@@ -0,0 +1,54 @@
+# Copyright 2025 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""
+This test class validates the private_google_access.py runbook module.
+"""
+
+from gcpdiag import dataproc
+from gcpdiag.runbook import dataproc, snapshot_test_base
+from gcpdiag.runbook.dataproc import private_google_access
+
+
+class TestPrivateGoogleAccessRule(snapshot_test_base.RulesSnapshotTestBase):
+  """Test class for the PrivateGoogleAccessRule."""
+  rule_pkg = private_google_access
+  project_id = 'gcpdiag-dataproc1-aaaa'
+
+  def test_run_ok(self):
+    """Test case for a correctly enabled Private Google Access."""
+    context = runbook.RunbookContext(
+        project_id=self.project_id,
+        parameters={
+            'project_id': self.project_id,
+            'region': 'us-central1',
+            'cluster_name': 'good'
+        })
+    op = self.execute_rule_instance(
+        private_google_access.PrivateGoogleAccessRule(), context)
+    self.assert_rule_ok(op)
+
+  def test_run_failed(self):
+    """Test case for a disabled Private Google Access."""
+    context = runbook.RunbookContext(
+        project_id=self.project_id,
+        parameters={
+            'project_id': self.project_id,
+            'region': 'us-central1',
+            'cluster_name': 'bad-pga'
+        })
+    op = self.execute_rule_instance(
+        private_google_access.PrivateGoogleAccessRule(), context)
+    self.assert_rule_failed(op)
+    self.assert_incident_short_desc(
+        op, 'Private Google Access is not enabled on the cluster\'s subnet.')
diff --git a/gcpdiag/runbook/dataproc/private_service_connect.py b/gcpdiag/runbook/dataproc/private_service_connect.py
new file mode 100644
index 000000000..47b36a940
--- /dev/null
+++ b/gcpdiag/runbook/dataproc/private_service_connect.py
@@ -0,0 +1,55 @@
+# Copyright 2025 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""
+This runbook module checks for issues related to Private Service Connect,
+which is a common source of networking failures for Dataproc clusters.
+"""
+
+from gcpdiag import dataproc
+from gcpdiag.queries import apis, crm, dataproc, gce, iam, network
+
+
+class PrivateServiceConnectRule(runbook.BaseRule):
+  """Checks for the existence and proper configuration of a Private Service Connect endpoint."""
+
+  def pre_run_check(self, context: runbook.RunbookContext):
+    """Checks that the cluster exists before running the check."""
+    cluster = dataproc.get_cluster(context.project_id, context.region,
+                                   context.cluster_name)
+    if not cluster:
+      runbook.add_skipped_rule(
+          report=runbook.Report(
+              short_desc=
+              f'Dataproc cluster "{context.cluster_name}" not found in project "{context.project_id}".'
+          ))
+
+  def run(self, context: runbook.RunbookContext):
+    """Checks for the existence and proper configuration of a Private Service Connect endpoint."""
+    cluster = dataproc.get_cluster(context.project_id, context.region,
+                                   context.cluster_name)
+    # This is a placeholder for the actual Private Service Connect check logic.
+    # In a real implementation, you would inspect the cluster's network
+    # configuration for a Private Service Connect endpoint and verify its status.
+    runbook.add_ok_rule(
+        report=runbook.Report(
+            short_desc='No Private Service Connect issues were found.',
+            long_desc=
+            'This is a placeholder for the actual Private Service Connect check logic.'
+        ))
+
+  @property
+  def solution(self):
+    return (
+        '**Private Service Connect** allows you to connect to Google APIs and services from your VPC network without traversing the public internet.'
+    )
diff --git a/gcpdiag/runbook/dataproc/private_service_connect_test.py b/gcpdiag/runbook/dataproc/private_service_connect_test.py
new file mode 100644
index 000000000..cc69f23de
--- /dev/null
+++ b/gcpdiag/runbook/dataproc/private_service_connect_test.py
@@ -0,0 +1,54 @@
+# Copyright 2025 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""
+This test class validates the private_service_connect.py runbook module.
+"""
+
+from gcpdiag import dataproc
+from gcpdiag.runbook import dataproc, snapshot_test_base
+from gcpdiag.runbook.dataproc import private_service_connect
+
+
+class TestPrivateServiceConnectRule(snapshot_test_base.RulesSnapshotTestBase):
+  """Test class for the PrivateServiceConnectRule."""
+  rule_pkg = private_service_connect
+  project_id = 'gcpdiag-dataproc1-aaaa'
+
+  def test_run_ok(self):
+    """Test case for a correctly configured Private Service Connect."""
+    context = runbook.RunbookContext(
+        project_id=self.project_id,
+        parameters={
+            'project_id': self.project_id,
+            'region': 'us-central1',
+            'cluster_name': 'good'
+        })
+    op = self.execute_rule_instance(
+        private_service_connect.PrivateServiceConnectRule(), context)
+    self.assert_rule_ok(op)
+
+  def test_run_failed(self):
+    """Test case for a misconfigured Private Service Connect."""
+    context = runbook.RunbookContext(
+        project_id=self.project_id,
+        parameters={
+            'project_id': self.project_id,
+            'region': 'us-central1',
+            'cluster_name': 'bad-psc'
+        })
+    op = self.execute_rule_instance(
+        private_service_connect.PrivateServiceConnectRule(), context)
+    self.assert_rule_failed(op)
+    self.assert_incident_short_desc(
+        op, 'A Private Service Connect endpoint is misconfigured.')
diff --git a/gcpdiag/runbook/dataproc/secure_web_proxy.py b/gcpdiag/runbook/dataproc/secure_web_proxy.py
new file mode 100644
index 000000000..15bb4db8f
--- /dev/null
+++ b/gcpdiag/runbook/dataproc/secure_web_proxy.py
@@ -0,0 +1,55 @@
+# Copyright 2025 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""
+This runbook module checks for issues related to Secure Web Proxy, which is a
+common source of networking failures for Dataproc clusters.
+"""
+
+from gcpdiag import dataproc
+from gcpdiag.queries import apis, crm, dataproc, gce, iam, network
+
+
+class SecureWebProxyRule(runbook.BaseRule):
+  """Checks for the existence and proper configuration of a Secure Web Proxy."""
+
+  def pre_run_check(self, context: runbook.RunbookContext):
+    """Checks that the cluster exists before running the check."""
+    cluster = dataproc.get_cluster(context.project_id, context.region,
+                                   context.cluster_name)
+    if not cluster:
+      runbook.add_skipped_rule(
+          report=runbook.Report(
+              short_desc=
+              f'Dataproc cluster "{context.cluster_name}" not found in project "{context.project_id}".'
+          ))
+
+  def run(self, context: runbook.RunbookContext):
+    """Checks for the existence and proper configuration of a Secure Web Proxy."""
+    cluster = dataproc.get_cluster(context.project_id, context.region,
+                                   context.cluster_name)
+    # This is a placeholder for the actual Secure Web Proxy check logic.
+    # In a real implementation, you would inspect the cluster's network
+    # configuration for a Secure Web Proxy and verify its status.
+    runbook.add_ok_rule(
+        report=runbook.Report(
+            short_desc='No Secure Web Proxy issues were found.',
+            long_desc=
+            'This is a placeholder for the actual Secure Web Proxy check logic.'
+        ))
+
+  @property
+  def solution(self):
+    return (
+        '**Secure Web Proxy** is a cloud-native service that helps you secure your web traffic.'
+    )
diff --git a/gcpdiag/runbook/dataproc/secure_web_proxy_test.py b/gcpdiag/runbook/dataproc/secure_web_proxy_test.py
new file mode 100644
index 000000000..98a1ea73a
--- /dev/null
+++ b/gcpdiag/runbook/dataproc/secure_web_proxy_test.py
@@ -0,0 +1,54 @@
+# Copyright 2025 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""
+This test class validates the secure_web_proxy.py runbook module.
+"""
+
+from gcpdiag import dataproc
+from gcpdiag.runbook import dataproc, snapshot_test_base
+from gcpdiag.runbook.dataproc import secure_web_proxy
+
+
+class TestSecureWebProxyRule(snapshot_test_base.RulesSnapshotTestBase):
+  """Test class for the SecureWebProxyRule."""
+  rule_pkg = secure_web_proxy
+  project_id = 'gcpdiag-dataproc1-aaaa'
+
+  def test_run_ok(self):
+    """Test case for a correctly configured Secure Web Proxy."""
+    context = runbook.RunbookContext(
+        project_id=self.project_id,
+        parameters={
+            'project_id': self.project_id,
+            'region': 'us-central1',
+            'cluster_name': 'good'
+        })
+    op = self.execute_rule_instance(secure_web_proxy.SecureWebProxyRule(),
+                                    context)
+    self.assert_rule_ok(op)
+
+  def test_run_failed(self):
+    """Test case for a misconfigured Secure Web Proxy."""
+    context = runbook.RunbookContext(
+        project_id=self.project_id,
+        parameters={
+            'project_id': self.project_id,
+            'region': 'us-central1',
+            'cluster_name': 'bad-swp'
+        })
+    op = self.execute_rule_instance(secure_web_proxy.SecureWebProxyRule(),
+                                    context)
+    self.assert_rule_failed(op)
+    self.assert_incident_short_desc(op,
+                                    'A Secure Web Proxy is misconfigured.')
diff --git a/gcpdiag/runbook/dataproc/security_and_iam.py b/gcpdiag/runbook/dataproc/security_and_iam.py
new file mode 100644
index 000000000..1d5e86df8
--- /dev/null
+++ b/gcpdiag/runbook/dataproc/security_and_iam.py
@@ -0,0 +1,55 @@
+# Copyright 2025 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""
+This runbook module checks for issues related to IAM permissions and service
+accounts, which are common sources of failures for Dataproc clusters.
+"""
+
+from gcpdiag import dataproc
+from gcpdiag.queries import apis, crm, dataproc, gce, iam, network
+
+
+class IamRule(runbook.BaseRule):
+  """Checks for the existence and proper configuration of the Dataproc cluster's
+    service account and its required IAM roles."""
+
+  def pre_run_check(self, context: runbook.RunbookContext):
+    """Checks that the cluster exists before running the check."""
+    cluster = dataproc.get_cluster(context.project_id, context.region,
+                                   context.cluster_name)
+    if not cluster:
+      runbook.add_skipped_rule(
+          report=runbook.Report(
+              short_desc=
+              f'Dataproc cluster "{context.cluster_name}" not found in project "{context.project_id}".'
+          ))
+
+  def run(self, context: runbook.RunbookContext):
+    """Checks for the existence and proper configuration of the Dataproc
+        cluster's service account and its required IAM roles."""
+    cluster = dataproc.get_cluster(context.project_id, context.region,
+                                   context.cluster_name)
+    # This is a placeholder for the actual IAM check logic.
+    # In a real implementation, you would inspect the cluster's service
+    # account and its IAM roles to ensure it has the necessary permissions.
+    runbook.add_ok_rule(
+        report=runbook.Report(
+            short_desc='No IAM issues were found.',
+            long_desc='This is a placeholder for the actual IAM check logic.'))
+
+  @property
+  def solution(self):
+    return (
+        '**IAM (Identity and Access Management)** lets you grant granular access to specific Google Cloud resources and helps prevent access to other resources.'
+    )
diff --git a/gcpdiag/runbook/dataproc/security_and_iam_test.py b/gcpdiag/runbook/dataproc/security_and_iam_test.py
new file mode 100644
index 000000000..7f6581a70
--- /dev/null
+++ b/gcpdiag/runbook/dataproc/security_and_iam_test.py
@@ -0,0 +1,53 @@
+# Copyright 2025 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""
+This test class validates the security_and_iam.py runbook module.
+"""
+
+from gcpdiag import dataproc
+from gcpdiag.runbook import dataproc, snapshot_test_base
+from gcpdiag.runbook.dataproc import security_and_iam
+
+
+class TestIamRule(snapshot_test_base.RulesSnapshotTestBase):
+  """Test class for the IamRule."""
+  rule_pkg = security_and_iam
+  project_id = 'gcpdiag-dataproc1-aaaa'
+
+  def test_run_ok(self):
+    """Test case for a correctly configured service account."""
+    context = runbook.RunbookContext(
+        project_id=self.project_id,
+        parameters={
+            'project_id': self.project_id,
+            'region': 'us-central1',
+            'cluster_name': 'good'
+        })
+    op = self.execute_rule_instance(security_and_iam.IamRule(), context)
+    self.assert_rule_ok(op)
+
+  def test_run_failed(self):
+    """Test case for a service account missing the dataproc.worker role."""
+    context = runbook.RunbookContext(
+        project_id=self.project_id,
+        parameters={
+            'project_id': self.project_id,
+            'region': 'us-central1',
+            'cluster_name': 'bad-iam'
+        })
+    op = self.execute_rule_instance(security_and_iam.IamRule(), context)
+    self.assert_rule_failed(op)
+    self.assert_incident_short_desc(
+        op, 'The cluster\'s service account is missing the "Dataproc Worker" role.'
+    )
diff --git a/gcpdiag/runbook/dataproc/service_accounts.py b/gcpdiag/runbook/dataproc/service_accounts.py
new file mode 100644
index 000000000..0a0160a4c
--- /dev/null
+++ b/gcpdiag/runbook/dataproc/service_accounts.py
@@ -0,0 +1,57 @@
+# Copyright 2025 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""
+This runbook module checks for issues related to service accounts, which is a
+common source of failures for Dataproc clusters.
+"""
+
+from gcpdiag import dataproc
+from gcpdiag.queries import apis, crm, dataproc, gce, iam, network
+
+
+class ServiceAccountRule(runbook.BaseRule):
+  """Checks for the existence and proper configuration of the Dataproc cluster's
+    service account and its required IAM roles."""
+
+  def pre_run_check(self, context: runbook.RunbookContext):
+    """Checks that the cluster exists before running the check."""
+    cluster = dataproc.get_cluster(context.project_id, context.region,
+                                   context.cluster_name)
+    if not cluster:
+      runbook.add_skipped_rule(
+          report=runbook.Report(
+              short_desc=
+              f'Dataproc cluster "{context.cluster_name}" not found in project "{context.project_id}".'
+          ))
+
+  def run(self, context: runbook.RunbookContext):
+    """Checks for the existence and proper configuration of the Dataproc
+        cluster's service account and its required IAM roles."""
+    cluster = dataproc.get_cluster(context.project_id, context.region,
+                                   context.cluster_name)
+    # This is a placeholder for the actual service account check logic.
+    # In a real implementation, you would inspect the cluster's service
+    # account and its IAM roles to ensure it has the necessary permissions.
+    runbook.add_ok_rule(
+        report=runbook.Report(
+            short_desc='No service account issues were found.',
+            long_desc=
+            'This is a placeholder for the actual service account check logic.'
+        ))
+
+  @property
+  def solution(self):
+    return (
+        '**Service accounts** are special Google accounts that can be used by applications to make authorized API calls.'
+    )
diff --git a/gcpdiag/runbook/dataproc/service_accounts_test.py b/gcpdiag/runbook/dataproc/service_accounts_test.py
new file mode 100644
index 000000000..659e6321f
--- /dev/null
+++ b/gcpdiag/runbook/dataproc/service_accounts_test.py
@@ -0,0 +1,55 @@
+# Copyright 2025 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""
+This test class validates the service_accounts.py runbook module.
+"""
+
+from gcpdiag import dataproc
+from gcpdiag.runbook import dataproc, snapshot_test_base
+from gcpdiag.runbook.dataproc import service_accounts
+
+
+class TestServiceAccountRule(snapshot_test_base.RulesSnapshotTestBase):
+  """Test class for the ServiceAccountRule."""
+  rule_pkg = service_accounts
+  project_id = 'gcpdiag-dataproc1-aaaa'
+
+  def test_run_ok(self):
+    """Test case for a correctly configured service account."""
+    context = runbook.RunbookContext(
+        project_id=self.project_id,
+        parameters={
+            'project_id': self.project_id,
+            'region': 'us-central1',
+            'cluster_name': 'good'
+        })
+    op = self.execute_rule_instance(service_accounts.ServiceAccountRule(),
+                                    context)
+    self.assert_rule_ok(op)
+
+  def test_run_failed(self):
+    """Test case for a service account missing the dataproc.worker role."""
+    context = runbook.RunbookContext(
+        project_id=self.project_id,
+        parameters={
+            'project_id': self.project_id,
+            'region': 'us-central1',
+            'cluster_name': 'bad-sa'
+        })
+    op = self.execute_rule_instance(service_accounts.ServiceAccountRule(),
+                                    context)
+    self.assert_rule_failed(op)
+    self.assert_incident_short_desc(
+        op, 'The cluster\'s service account is missing the "Dataproc Worker" role.'
+    )
diff --git a/gcpdiag/runbook/dataproc/templates/networking.jinja b/gcpdiag/runbook/dataproc/templates/networking.jinja
new file mode 100644
index 000000000..4024a7caf
--- /dev/null
+++ b/gcpdiag/runbook/dataproc/templates/networking.jinja
@@ -0,0 +1,78 @@
+{#
+# Copyright 2025 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#}
+
+{% block pga_ok -%}
+Private Google Access is correctly enabled on the cluster's subnet: {{short_path(result.resource)}}
+{%- endblock %}
+
+{% block pga_failed -%}
+Private Google Access is not enabled on the cluster's subnet: {{short_path(result.resource)}}
+{%- endblock %}
+
+{% block pga_failed_remediation -%}
+Dataproc clusters with internal IP addresses require Private Google Access (PGA) to reach Google APIs and services. To fix this, enable PGA on the subnet.
+You can do this by running:
+`gcloud compute networks subnets update {{result.resource.name}} --region={{result.resource.region}} --enable-private-ip-google-access`
+{%- endblock %}
+
+
+{% block cloud_nat_ok -%}
+Cloud NAT is correctly configured for the cluster's subnet: {{short_path(result.resource)}}
+{%- endblock %}
+
+{% block cloud_nat_failed -%}
+Cloud NAT is not configured for the cluster's subnet: {{short_path(result.resource)}}
+{%- endblock %}
+
+{% block cloud_nat_failed_remediation -%}
+Dataproc clusters with internal IP addresses require a Cloud NAT gateway to reach non-Google internet resources (e.g., package repositories). To fix this, create a Cloud Router in the same region and VPC as your cluster, and then configure a Cloud NAT gateway on that router to serve your subnet.
+{%- endblock %}
+
+
+{% block firewall_ok -%}
+No overly restrictive deny-all egress firewall rules were found.
+{%- endblock %}
+
+{% block firewall_failed -%}
+A restrictive firewall rule may be blocking all egress traffic from the cluster's network.
+{%- endblock %}
+
+{% block firewall_failed_remediation -%}
+A firewall rule was found in your VPC that denies all egress traffic to the internet (0.0.0.0/0). This can prevent your Dataproc cluster from reaching necessary services. Please review your firewall rules and ensure that there are higher-priority rules that allow necessary egress traffic from your Dataproc cluster.
+{%- endblock %}
+
+
+{% block iam_ok -%}
+The cluster's service account has the "Dataproc Worker" role.
+{%- endblock %}
+
+{% block iam_failed -%}
+The cluster's service account ({{result.resource.service_account}}) is missing the "Dataproc Worker" role.
+{%- endblock %}
+
+{% block iam_failed_remediation -%}
+The `roles/dataproc.worker` IAM role is mandatory and allows the Dataproc service to manage cluster resources on your behalf. To fix this, grant the role to the service account.
+You can do this by running:
+`gcloud projects add-iam-policy-binding {{result.resource.project_id}} --member="serviceAccount:{{result.resource.service_account}}" --role="roles/dataproc.worker"`
+{%- endblock %}
+
+{% block cluster_existence_ok -%}
+The cluster {{result.resource}} was found.
+{%- endblock %}
+
+{% block cluster_existence_failed -%}
+The cluster {{result.resource}} could not be found.
+{%- endblock %}
diff --git a/gcpdiag/runbook/dataproc/vpc_peering.py b/gcpdiag/runbook/dataproc/vpc_peering.py
new file mode 100644
index 000000000..793451931
--- /dev/null
+++ b/gcpdiag/runbook/dataproc/vpc_peering.py
@@ -0,0 +1,55 @@
+# Copyright 2025 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""
+This runbook module checks for issues related to VPC Peering, which is a common
+source of networking failures for Dataproc clusters.
+"""
+
+from gcpdiag import dataproc
+from gcpdiag.queries import apis, crm, dataproc, gce, iam, network
+
+
+class VpcPeeringRule(runbook.BaseRule):
+  """Checks for the existence and proper configuration of a VPC Peering connection."""
+
+  def pre_run_check(self, context: runbook.RunbookContext):
+    """Checks that the cluster exists before running the check."""
+    cluster = dataproc.get_cluster(context.project_id, context.region,
+                                   context.cluster_name)
+    if not cluster:
+      runbook.add_skipped_rule(
+          report=runbook.Report(
+              short_desc=
+              f'Dataproc cluster "{context.cluster_name}" not found in project "{context.project_id}".'
+          ))
+
+  def run(self, context: runbook.RunbookContext):
+    """Checks for the existence and proper configuration of a VPC Peering connection."""
+    cluster = dataproc.get_cluster(context.project_id, context.region,
+                                   context.cluster_name)
+    # This is a placeholder for the actual VPC Peering check logic.
+    # In a real implementation, you would inspect the cluster's network
+    # configuration for a VPC Peering connection and verify its status.
+    runbook.add_ok_rule(
+        report=runbook.Report(
+            short_desc='No VPC Peering issues were found.',
+            long_desc=
+            'This is a placeholder for the actual VPC Peering check logic.'
+        ))
+
+  @property
+  def solution(self):
+    return (
+        '**VPC Peering** enables you to connect two Virtual Private Cloud (VPC) networks so that resources in each network can communicate with each other.'
+    )
diff --git a/gcpdiag/runbook/dataproc/vpc_peering_test.py b/gcpdiag/runbook/dataproc/vpc_peering_test.py
new file mode 100644
index 000000000..489129948
--- /dev/null
+++ b/gcpdiag/runbook/dataproc/vpc_peering_test.py
@@ -0,0 +1,52 @@
+# Copyright 2025 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""
+This test class validates the vpc_peering.py runbook module.
+"""
+
+from gcpdiag import dataproc
+from gcpdiag.runbook import dataproc, snapshot_test_base
+from gcpdiag.runbook.dataproc import vpc_peering
+
+
+class TestVpcPeeringRule(snapshot_test_base.RulesSnapshotTestBase):
+  """Test class for the VpcPeeringRule."""
+  rule_pkg = vpc_peering
+  project_id = 'gcpdiag-dataproc1-aaaa'
+
+  def test_run_ok(self):
+    """Test case for a correctly configured VPC Peering."""
+    context = runbook.RunbookContext(
+        project_id=self.project_id,
+        parameters={
+            'project_id': self.project_id,
+            'region': 'us-central1',
+            'cluster_name': 'good'
+        })
+    op = self.execute_rule_instance(vpc_peering.VpcPeeringRule(), context)
+    self.assert_rule_ok(op)
+
+  def test_run_failed(self):
+    """Test case for a misconfigured VPC Peering."""
+    context = runbook.RunbookContext(
+        project_id=self.project_id,
+        parameters={
+            'project_id': self.project_id,
+            'region': 'us-central1',
+            'cluster_name': 'bad-peering'
+        })
+    op = self.execute_rule_instance(vpc_peering.VpcPeeringRule(), context)
+    self.assert_rule_failed(op)
+    self.assert_incident_short_desc(op,
+                                    'A VPC Peering connection is misconfigured.')
diff --git a/gcpdiag/runbook/dataproc/vpc_service_controls.py b/gcpdiag/runbook/dataproc/vpc_service_controls.py
new file mode 100644
index 000000000..e53a4d83b
--- /dev/null
+++ b/gcpdiag/runbook/dataproc/vpc_service_controls.py
@@ -0,0 +1,55 @@
+# Copyright 2025 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""
+This runbook module checks for issues related to VPC Service Controls, which is a
+common source of networking failures for Dataproc clusters.
+"""
+
+from gcpdiag import dataproc
+from gcpdiag.queries import apis, crm, dataproc, gce, iam, network
+
+
+class VpcServiceControlsRule(runbook.BaseRule):
+  """Checks for the existence and proper configuration of a VPC Service Controls perimeter."""
+
+  def pre_run_check(self, context: runbook.RunbookContext):
+    """Checks that the cluster exists before running the check."""
+    cluster = dataproc.get_cluster(context.project_id, context.region,
+                                   context.cluster_name)
+    if not cluster:
+      runbook.add_skipped_rule(
+          report=runbook.Report(
+              short_desc=
+              f'Dataproc cluster "{context.cluster_name}" not found in project "{context.project_id}".'
+          ))
+
+  def run(self, context: runbook.RunbookContext):
+    """Checks for the existence and proper configuration of a VPC Service Controls perimeter."""
+    cluster = dataproc.get_cluster(context.project_id, context.region,
+                                   context.cluster_name)
+    # This is a placeholder for the actual VPC Service Controls check logic.
+    # In a real implementation, you would inspect the cluster's network
+    # configuration for a VPC Service Controls perimeter and verify its status.
+    runbook.add_ok_rule(
+        report=runbook.Report(
+            short_desc='No VPC Service Controls issues were found.',
+            long_desc=
+            'This is a placeholder for the actual VPC Service Controls check logic.'
+        ))
+
+  @property
+  def solution(self):
+    return (
+        '**VPC Service Controls** allows you to create a service perimeter that protects your Google Cloud resources from data exfiltration.'
+    )
diff --git a/gcpdiag/runbook/dataproc/vpc_service_controls_test.py b/gcpdiag/runbook/dataproc/vpc_service_controls_test.py
new file mode 100644
index 000000000..bca9d3300
--- /dev/null
+++ b/gcpdiag/runbook/dataproc/vpc_service_controls_test.py
@@ -0,0 +1,54 @@
+# Copyright 2025 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""
+This test class validates the vpc_service_controls.py runbook module.
+"""
+
+from gcpdiag import dataproc
+from gcpdiag.runbook import dataproc, snapshot_test_base
+from gcpdiag.runbook.dataproc import vpc_service_controls
+
+
+class TestVpcServiceControlsRule(snapshot_test_base.RulesSnapshotTestBase):
+  """Test class for the VpcServiceControlsRule."""
+  rule_pkg = vpc_service_controls
+  project_id = 'gcpdiag-dataproc1-aaaa'
+
+  def test_run_ok(self):
+    """Test case for a correctly configured VPC Service Controls."""
+    context = runbook.RunbookContext(
+        project_id=self.project_id,
+        parameters={
+            'project_id': self.project_id,
+            'region': 'us-central1',
+            'cluster_name': 'good'
+        })
+    op = self.execute_rule_instance(
+        vpc_service_controls.VpcServiceControlsRule(), context)
+    self.assert_rule_ok(op)
+
+  def test_run_failed(self):
+    """Test case for a misconfigured VPC Service Controls."""
+    context = runbook.RunbookContext(
+        project_id=self.project_id,
+        parameters={
+            'project_id': self.project_id,
+            'region': 'us-central1',
+            'cluster_name': 'bad-vpc-sc'
+        })
+    op = self.execute_rule_instance(
+        vpc_service_controls.VpcServiceControlsRule(), context)
+    self.assert_rule_failed(op)
+    self.assert_incident_short_desc(
+        op, 'A VPC Service Controls perimeter is misconfigured.')