gcp-dataflow-exfiltrate-via-sink

A Dataflow worker SA with storage.objects.create on an external GCS bucket can stream pipeline output (data records, query results) to that external location.

derived gcp emits CanExfiltrate

match

A conjunctive graph pattern. Variables (?x) bind node ids; every clause must hold.

where

emit

source type	`Compute`
target type	`Storage`
source	`<Dataflow job>`
target	`<external GCS bucket>`
permissions	`storage.objects.create`
conditions	`iam_permission` `network_reachability`
state logic	ACTIVE when the worker SA's effective grant includes storage.objects.create on the external bucket AND no VPC Service Controls perimeter prevents the write. BLOCKED if the Dataflow project is within a VPC SC perimeter that restricts storage.googleapis.com writes to external buckets. CONDITIONAL(iam_permission) if the grant is unresolved.

source type

Compute

target type

Storage

source

<Dataflow job>

target

<external GCS bucket>

permissions

storage.objects.create

conditions

iam_permission network_reachability

state logic

ACTIVE when the worker SA's effective grant includes storage.objects.create on the external bucket AND no VPC Service Controls perimeter prevents the write. BLOCKED if the Dataflow project is within a VPC SC perimeter that restricts storage.googleapis.com writes to external buckets. CONDITIONAL(iam_permission) if the grant is unresolved.

Narrative

Dataflow job {job.name} runs as worker service account {sa.name}, which has storage.objects.create on external GCS bucket {externalBucket.name}; pipeline output steps can stream data records to that external location (UNLOAD/exfiltration).

Raw rule rules/derived/gcp/dataflow.yaml

id: gcp-dataflow-exfiltrate-via-sink
emits: CanExfiltrate
description: A Dataflow worker SA with storage.objects.create on an external GCS bucket can stream pipeline
  output (data records, query results) to that external location.
match:
- - ?job
  - ExecutesAs
  - ?sa
- - ?sa
  - CanWriteData
  - ?externalBucket
where:
- node_type(?job) == BatchJob
- 'node_type(?externalBucket) == ObjectStorage  # GCS bucket'
- ?externalBucket.project != ?job.project OR ?externalBucket is outside the VPC SC perimeter
emit:
  source_type: Compute
  target_type: Storage
  source: <Dataflow job>
  target: <external GCS bucket>
  permissions:
  - storage.objects.create
  conditions:
  - iam_permission
  - network_reachability
  state_logic: ACTIVE when the worker SA's effective grant includes storage.objects.create on the external
    bucket AND no VPC Service Controls perimeter prevents the write. BLOCKED if the Dataflow project is
    within a VPC SC perimeter that restricts storage.googleapis.com writes to external buckets. CONDITIONAL(iam_permission)
    if the grant is unresolved.
  confidence: min(contributing_confidences) * 0.85
  derived_from:
  - <ExecutesAs edge_id for ?job -> ?sa>
  - <CanWriteData edge_id for ?sa -> ?externalBucket>
  false_positive_note: "Exfiltration requires BOTH that the worker SA can write to the external bucket\
    \ AND that no VPC SC perimeter blocks the storage API call. If the Dataflow project is in a VPC SC\
    \ perimeter, cross-perimeter storage writes are BLOCKED unless an access policy explicitly allows\
    \ them. Confirm perimeter configuration before emitting ACTIVE. This edge is only meaningful when\
    \ combined with a CanExecuteAs or CanModifyCode edge \u2014 a job the attacker cannot influence is\
    \ not an exfil path even if the worker SA has broad GCS access."
  narrative: Dataflow job {job.name} runs as worker service account {sa.name}, which has storage.objects.create
    on external GCS bucket {externalBucket.name}; pipeline output steps can stream data records to that
    external location (UNLOAD/exfiltration).