gcp-composer-dag-write-modify-code
storage.objects.create on the Composer environment's GCS dags/ prefix injects a Python DAG file that Airflow executes as the environment SA; cite can-execute-as for the CanExecuteAs roll-up.
match (effective permission)
{
"action": "storage.objects.create",
"resource_type": "google.storage.Bucket"
}
where
bucket is the environment's DAG bucket (dagGcsPrefix prefix — gs://<env-bucket>/dags/)
object prefix matches dags/ OR plugins/ (plugins/ is equivalent: Airflow loads them at scheduler/worker startup)
emit
| source type | Identity |
|---|---|
| target type | AnalyticsService |
| source | <principal> |
| target | <Composer Environment node (AnalyticsService)> |
| permissions | storage.objects.create |
| state logic | ACTIVE: storage.objects.create on the dags/ prefix is a direct data-plane write that the Airflow scheduler processes automatically (no manual trigger required). Airflow picks up the new DAG file within the DAG scan interval (~30s Composer 2, ~5s Composer 1); once parsed the tasks are scheduled and executed as the environment SA. This is ACTIVE (not CONDITIONAL) because the scan is automatic and requires no further action by the attacker. CONDITIONAL only if a VPC Service Controls perimeter blocks the storage.googleapis.com API for the principal's origin network. BLOCKED if an IAM deny policy or a VPC-SC perimeter enforces a deny on storage.objects.create at this scope. |
Narrative
{principal.name} has storage.objects.create on the Cloud Composer environment {env.name}'s GCS dags/ bucket prefix; uploading a Python DAG file causes the Airflow scheduler to parse and execute it within ~30 seconds as the environment service account {sa.name}. No Composer-level permission is required - a single GCS write yields remote code execution as {sa.name}. Cite can-execute-as execute-as-via-code-modify for the CanExecuteAs collapse (CanModifyCode + ExecutesAs → CanExecuteAs).
Raw rule rules/derived/gcp/composer.yaml
id: gcp-composer-dag-write-modify-code
emits: CanModifyCode
description: storage.objects.create on the Composer environment's GCS dags/ prefix injects a Python DAG
file that Airflow executes as the environment SA; cite can-execute-as for the CanExecuteAs roll-up.
match_effective_permission:
action: storage.objects.create
resource_type: google.storage.Bucket
where:
- "bucket is the environment's DAG bucket (dagGcsPrefix prefix \u2014 gs://<env-bucket>/dags/)"
- 'object prefix matches dags/ OR plugins/ (plugins/ is equivalent: Airflow loads them at scheduler/worker
startup)'
emit:
source_type: Identity
target_type: AnalyticsService
source: <principal>
target: <Composer Environment node (AnalyticsService)>
permissions:
- storage.objects.create
conditions: []
state_logic: 'ACTIVE: storage.objects.create on the dags/ prefix is a direct data-plane write that the
Airflow scheduler processes automatically (no manual trigger required). Airflow picks up the new DAG
file within the DAG scan interval (~30s Composer 2, ~5s Composer 1); once parsed the tasks are scheduled
and executed as the environment SA. This is ACTIVE (not CONDITIONAL) because the scan is automatic
and requires no further action by the attacker. CONDITIONAL only if a VPC Service Controls perimeter
blocks the storage.googleapis.com API for the principal''s origin network. BLOCKED if an IAM deny
policy or a VPC-SC perimeter enforces a deny on storage.objects.create at this scope.'
confidence: 0.97
derived_from:
- <storage.objects.create effective permission on dags/ bucket prefix>
false_positive_note: "Confirm the write is scoped to the dags/ or plugins/ prefix, not merely data/\
\ or logs/ (data/ write is CanWriteData / ContainsSecret, not code execution). Check for IAM Conditions\
\ on the storage.objects.create binding that restrict by prefix (storage.googleapis.com/objectNamePrefix\
\ condition). A VPC Service Controls perimeter covering storage.googleapis.com for the principal's\
\ network makes this CONDITIONAL(scp_or_org_policy) for out-of-perimeter principals \u2014 not ACTIVE.\
\ The DAG scan interval is a minor delay (~5-30s), not a blocking control; emit ACTIVE. Do NOT emit\
\ if the bucket is a different GCS bucket (not the env's dagGcsPrefix). Cite can-execute-as execute-as-via-code-modify\
\ for the CanExecuteAs collapse \u2014 do not re-derive it."
narrative: "{principal.name} has storage.objects.create on the Cloud Composer environment {env.name}'s\
\ GCS dags/ bucket prefix; uploading a Python DAG file causes the Airflow scheduler to parse and execute\
\ it within ~30 seconds as the environment service account {sa.name}. No Composer-level permission\
\ is required \u2014 a single GCS write yields remote code execution as {sa.name}. Cite can-execute-as\
\ execute-as-via-code-modify for the CanExecuteAs collapse (CanModifyCode + ExecutesAs \u2192 CanExecuteAs)."