Creating and Securing GCS Buckets with Pulumi (Python)

Creating a Google Cloud Storage bucket with Pulumi Python is one resource; securing it is the rest of the work — uniform bucket-level access, scoped IAM bindings, object versioning, and customer-managed encryption, part of the broader GCP Provider Configuration workflow. A default bucket inherits legacy ACLs and project-wide IAM; this guide closes those gaps with a typed, repeatable pattern.

This guide provisions a bucket with uniform bucket-level access enabled, versioning on, a CMEK from Cloud KMS, and a single least-privilege IAM binding — then verifies that public access is impossible.

Context

A misconfigured GCS bucket is one of the most common cloud data leaks: legacy ACLs let an object be made public even when project IAM looks locked down. Uniform bucket-level access removes ACLs entirely, so IAM is the only access path — which makes the bucket auditable. Doing this at creation time costs nothing; retrofitting it onto a bucket already holding objects with per-object ACLs is a migration. The encryption and IAM discipline here is the same you apply when deploying a GKE cluster with Pulumi (Python) and pulls its credential routing from the parent provider configuration.

Three GCS controls are frequently confused, and knowing which one does what determines how much of the risk you have actually removed. Uniform bucket-level access turns off the ACL evaluation path; requests are authorised purely by IAM at the project, folder, or bucket level. Public access prevention is a separate switch that rejects any grant naming allUsers or allAuthenticatedUsers, so even a bucket without uniform access cannot be opened to the internet while it is enforced. The organization policy constraint constraints/storage.publicAccessPrevention applies the same rule from above, and when it is set at the org node it overrides whatever the bucket resource says — which is why a pulumi preview showing ~ publicAccessPrevention: "inherited" => "enforced" is a no-op in some projects and a real change in others.

Uniform access also has a switch-back window. GCP allows you to disable it for 90 days after enabling; after that the lockedTime on the bucket passes and the setting is permanent. Pulumi has no special handling for this: setting uniform_bucket_level_access=False on a locked bucket returns googleapi: Error 400: Cannot disable uniform bucket-level access after the locked time, invalid, and the only remedy is to create a new bucket and copy objects across.

Context Context: Context with 4 facets. Context GCS key element IAM key element GKE key element Pulumi key element
Context: how GCS, IAM, GKE relate in this pattern.

Prerequisites

Prerequisites Prerequisites: layered from GCP_PROJECT down to GCP. GCP_PROJECT GCP_REGION mypy Python GCP
Prerequisites: the building blocks this section assembles.
  • Python 3.9+ with pulumi>=3.0 and pulumi-gcp>=7.0.
  • A GCP project with billing enabled and GCP_PROJECT / GCP_REGION set, or the provider configured per the parent guide.
  • IAM permissions for storage.buckets.create, storage.buckets.setIamPolicy, and (for CMEK) cloudkms.cryptoKeys.get plus the ability to grant the storage service agent encrypt/decrypt.
  • An existing Cloud KMS key ring and crypto key, or permission to create them.
  • mypy for static checking of the typed config.

Implementation

1. Define a typed bucket configuration

Implementation Implementation: 1. Define a typed then 2. Create the then 3. Grant a single then 4. Wire the CMEK 1. Define a typed 2. Create the 3. Grant a single 4. Wire the CMEK
Implementation: the stages run left to right — 1. Define a typed, 2. Create the, 3. Grant a single, 4. Wire the CMEK.

The config object is what makes the bucket reviewable in a pull request: every security-relevant knob has a name and a default, and mypy --strict refuses a call that omits one. Note the two fields that carry a replacement penalty — name and location are immutable in the GCS API, so a change to either shows up as ++ gcp:storage/bucket:Bucket acme-data-us replace and destroys the object contents along with it.

# infra/bucket_config.py
# CLI: mypy --strict infra/
from __future__ import annotations
from dataclasses import dataclass
from typing import Optional

@dataclass(frozen=True)
class BucketConfig:
    name: str
    location: str = "US"
    storage_class: str = "STANDARD"
    kms_key_id: Optional[str] = None
    reader_member: Optional[str] = None  # e.g. "serviceAccount:[email protected]"
    retention_days: Optional[int] = None
    log_sink_bucket: Optional[str] = None
    # State implication: changing `name` or `location` forces replacement.

    def __post_init__(self) -> None:
        if self.storage_class not in {"STANDARD", "NEARLINE", "COLDLINE", "ARCHIVE"}:
            raise ValueError(f"unsupported storage class: {self.storage_class}")
        if self.kms_key_id and not self.kms_key_id.startswith("projects/"):
            raise ValueError("kms_key_id must be a fully-qualified crypto key resource name")

Validating in __post_init__ means a bad value fails during pulumi preview with a Python traceback rather than 40 seconds later with a googleapi: Error 400: Invalid argument from the GCS API. That difference matters most in CI, where the second form gives you an opaque provider error and no line number.

2. Create the bucket with UBLA, versioning, and CMEK

uniform_bucket_level_access=True is the single most important field — it disables object ACLs. public_access_prevention="enforced" blocks any IAM grant to allUsers/allAuthenticatedUsers.

# infra/bucket.py
# CLI: pulumi preview --diff
from __future__ import annotations
import pulumi_gcp as gcp
from infra.bucket_config import BucketConfig

def build_bucket(cfg: BucketConfig) -> gcp.storage.Bucket:
    encryption = (
        gcp.storage.BucketEncryptionArgs(default_kms_key_name=cfg.kms_key_id)
        if cfg.kms_key_id else None
    )
    return gcp.storage.Bucket(
        cfg.name,
        name=cfg.name,
        location=cfg.location,
        storage_class=cfg.storage_class,
        # Provider note: UBLA removes ACLs so IAM is the only access path.
        uniform_bucket_level_access=True,
        public_access_prevention="enforced",
        versioning=gcp.storage.BucketVersioningArgs(enabled=True),
        encryption=encryption,
        force_destroy=False,  # State implication: blocks deleting a non-empty bucket
    )

3. Grant a single scoped IAM binding

Use BucketIAMMember (one principal, one role) rather than BucketIAMPolicy (authoritative, overwrites everything). Member is additive and safe; policy can lock you out if it omits your own access.

Three GCS IAM resources in pulumi_gcp Three GCS IAM resources in pulumi_gcp: comparison across Scope it owns, On removal from code, Risk. Resource Scope it owns On removal from code Risk BucketIAMMember One member, one role Only that grant drops Low BucketIAMBinding All members of one role Whole role is cleared Medium BucketIAMPolicy The entire bucket policy Every binding is replaced High
Pick the narrowest IAM resource that expresses the intent; the blast radius grows sharply down the table.
# infra/bucket.py (continued)
# CLI: pulumi up
from __future__ import annotations
import pulumi_gcp as gcp

def grant_reader(
    name: str, bucket: gcp.storage.Bucket, member: str
) -> gcp.storage.BucketIAMMember:
    # Provider note: BucketIAMMember is additive; BucketIAMPolicy is
    # authoritative and will drop bindings it does not list.
    return gcp.storage.BucketIAMMember(
        f"{name}-reader",
        bucket=bucket.name,
        role="roles/storage.objectViewer",
        member=member,
    )

Pass the logical name in rather than reading bucket._name: the underscore-prefixed attribute is Pulumi internals and is not part of the public Python API, so a minor SDK bump can remove it. Note also the role choice. roles/storage.objectViewer grants storage.objects.get and storage.objects.list only; the superficially similar roles/storage.legacyObjectReader also works under uniform access but carries ACL-era semantics, and roles/storage.admin — the one people reach for when a permission error appears — includes storage.buckets.setIamPolicy, which lets the holder grant themselves anything else.

3b. Add retention, lifecycle, and access logs

A secured bucket is not finished until deletion and cost behaviour are also declared. Retention policies are the strongest control here: while a retention period is active, GCS refuses to delete or overwrite an object before it has aged past the period, and Pulumi cannot override that.

# infra/bucket_hardening.py
# CLI: pulumi up --diff
from __future__ import annotations
import pulumi_gcp as gcp
from infra.bucket_config import BucketConfig


def harden(name: str, cfg: BucketConfig) -> gcp.storage.Bucket:
    """Bucket with lifecycle transitions, retention, and access logging."""
    return gcp.storage.Bucket(
        name,
        name=cfg.name,
        location=cfg.location,
        uniform_bucket_level_access=True,
        public_access_prevention="enforced",
        versioning=gcp.storage.BucketVersioningArgs(enabled=True),
        lifecycle_rules=[
            gcp.storage.BucketLifecycleRuleArgs(
                action=gcp.storage.BucketLifecycleRuleActionArgs(
                    type="SetStorageClass", storage_class="NEARLINE"
                ),
                condition=gcp.storage.BucketLifecycleRuleConditionArgs(age=30),
            ),
            gcp.storage.BucketLifecycleRuleArgs(
                action=gcp.storage.BucketLifecycleRuleActionArgs(type="Delete"),
                # Prune superseded versions so versioning does not grow without bound.
                condition=gcp.storage.BucketLifecycleRuleConditionArgs(
                    num_newer_versions=3, with_state="ARCHIVED"
                ),
            ),
        ],
        # State implication: once set, retention blocks pulumi destroy until it
        # expires — GCS returns 403 "Object is subject to bucket's retention policy".
        retention_policy=(
            gcp.storage.BucketRetentionPolicyArgs(
                retention_period=cfg.retention_days * 86400, is_locked=False
            )
            if cfg.retention_days
            else None
        ),
        logging=(
            gcp.storage.BucketLoggingArgs(log_bucket=cfg.log_sink_bucket)
            if cfg.log_sink_bucket
            else None
        ),
    )

is_locked=False is deliberate. Locking a retention policy is irreversible for the lifetime of the bucket — not even a project owner can shorten it — so it belongs behind an explicit compliance decision rather than a default.

4. Wire the CMEK grant and export

Before the bucket can use a CMEK, the GCS service agent needs roles/cloudkms.cryptoKeyEncrypterDecrypter on the key. Grant it in the same program so the dependency is explicit.

# __main__.py
# CLI: pulumi up && pulumi stack output bucketUrl
import pulumi
import pulumi_gcp as gcp
from infra.bucket_config import BucketConfig
from infra.bucket import build_bucket, grant_reader

project = gcp.organizations.get_project()
sa = gcp.storage.get_project_service_account()

key_id = "projects/p/locations/us/keyRings/r/cryptoKeys/data"
gcp.kms.CryptoKeyIAMMember(
    "gcs-cmek-grant",
    crypto_key_id=key_id,
    role="roles/cloudkms.cryptoKeyEncrypterDecrypter",
    member=sa.email_address.apply(lambda e: f"serviceAccount:{e}"),
)

cfg = BucketConfig(name="acme-data-us", kms_key_id=key_id,
                   reader_member="serviceAccount:[email protected]")
bucket = build_bucket(cfg)
if cfg.reader_member:
    grant_reader("acme-data", bucket, cfg.reader_member)

pulumi.export("bucketUrl", bucket.url)

gcp.storage.get_project_service_account() is an invoke, not a resource: it runs during the preview and returns the service-<project-number>@gs-project-accounts.iam.gserviceaccount.com identity that GCS uses on your behalf. Because the KMS grant and the bucket both reference values derived from it, Pulumi orders the CryptoKeyIAMMember before the Bucket automatically — but only if the grant is in the same stack. Split them across stacks and the first deploy of the bucket races IAM propagation, which is the most common cause of the CMEK permission error described below.

Verification

Assert UBLA and public-access prevention are on, then confirm out of band that the bucket rejects a public grant.

Verification Verification: Test → Program → Mock/Cloud. Test Program Mock/Cloud invoke declare resolve assert
Verification: the test drives the program and asserts on resolved values.
# tests/test_bucket.py
# CLI: pytest tests/test_bucket.py
from __future__ import annotations
import pulumi
from typing import Any, Dict, Tuple

class Mocks(pulumi.runtime.Mocks):
    def new_resource(self, args: pulumi.runtime.MockResourceArgs) -> Tuple[str, Dict[str, Any]]:
        return (f"{args.name}-id", {**args.inputs, "url": f"gs://{args.inputs.get('name','')}"})
    def call(self, args: pulumi.runtime.MockCallArgs) -> Dict[str, Any]:
        return {"projectId": "p", "emailAddress": "[email protected]"}

pulumi.runtime.set_mocks(Mocks(), preview=False)

import importlib
main = importlib.import_module("__main__")

@pulumi.runtime.test
def test_ubla_enabled() -> pulumi.Output:
    return main.bucket.uniform_bucket_level_access.apply(
        lambda v: None if v else (_ for _ in ()).throw(AssertionError("UBLA not enabled"))
    )
# CLI: confirm enforcement on the live bucket; the public grant must fail
gcloud storage buckets describe gs://acme-data-us \
  --format='value(iamConfiguration.uniformBucketLevelAccess.enabled,iamConfiguration.publicAccessPrevention)'
gcloud storage buckets add-iam-policy-binding gs://acme-data-us \
  --member=allUsers --role=roles/storage.objectViewer   # expected: denied

Gotchas & Edge Cases

Gotchas & Edge Cases Gotchas & Edge Cases: Where it breaks with 4 facets. Where it breaks BucketIAMPolic watch this boundary BucketIAMMembe watch this boundary BucketIAMBindi watch this boundary default_kms_ke watch this boundary
Gotchas & Edge Cases: the boundaries where things break and what to check.

BucketIAMPolicy will silently remove your own access. The authoritative BucketIAMPolicy replaces the entire IAM policy with exactly what you declare. If your declaration omits the bindings GCP or you rely on, those are dropped on the next pulumi up. Prefer BucketIAMMember (or BucketIAMBinding per role) unless you genuinely intend to own the full policy.

CMEK fails until the service agent is granted the key. Bucket creation with default_kms_key_name returns Permission denied on Cloud KMS key if the GCS service agent lacks cryptoKeyEncrypterDecrypter. The service agent is created lazily; if get_project_service_account returns an agent that does not yet exist, the first deploy may need a retry after the agent is provisioned.

force_destroy=False blocks pulumi destroy on a non-empty bucket. This is intentional protection, but it means tearing down a populated bucket fails with bucket is not empty. Empty it first, or set force_destroy=True deliberately for ephemeral test buckets only.

Bucket names are globally unique across all of Google Cloud. acme-data almost certainly belongs to someone else, and the create fails with googleapi: Error 409: Sorry, that name is not available. Please try a different one., conflict. Prefix with the project ID or an org slug, and never build the name from a random suffix — a name that changes between deploys replaces the bucket every time. If you need uniqueness without volatility, use pulumi.get_stack() in the name and let the stack identity supply it.

Rotating the CMEK does not re-encrypt existing objects. Updating default_kms_key_name to a new key version only affects objects written after the change; everything already in the bucket stays encrypted under the old version, and deleting that version makes those objects permanently unreadable. Keep old key versions enabled, or rewrite the objects with gcloud storage objects update --encryption-key before disabling anything.

Versioning plus an unbounded lifecycle rule is a silent cost leak. A bucket with versioning on and no num_newer_versions or archived-age condition retains every overwrite forever. A workload that rewrites a 2 GB file hourly accumulates roughly 1.4 TB a month in noncurrent versions that nothing lists by default. The Delete rule in step 3b is not an optimisation — it is the thing that makes versioning affordable.

Operational Notes

Storage class and lifecycle are the two settings that decide the monthly bill, and unlike the security settings they are cheap to change later. The decision reduces to one question about read frequency.

Choosing a storage class and lifecycle rule Choosing a storage class and lifecycle rule: choose among 3 options. How often is this objectread after 30 days? daily Keep STANDARD,expire noncurrent at30 days monthly SetStorageClass toNEARLINE at 30 days yearly COLDLINE at 90 days,ARCHIVE at 365
Lifecycle rules are evaluated once per day per object, so early-deletion charges apply before the class minimum is met.

Each colder class carries a minimum storage duration — 30 days for NEARLINE, 90 for COLDLINE, 365 for ARCHIVE — and deleting or transitioning an object before that elapses bills the remainder anyway. A rule that moves objects to COLDLINE at 30 days and deletes them at 60 therefore costs more than leaving them in STANDARD. Model the intended access pattern before writing the conditions.

Auditing the bucket from CI

Static policy scanning catches the settings that this guide sets deliberately, which is exactly what you want a second pair of eyes for. Run the plan through a policy engine on every pull request rather than relying on review:

# tests/test_bucket_policy.py
# CLI: pytest tests/test_bucket_policy.py -q
from __future__ import annotations
from typing import Any
import pulumi


def _assert_secure(args: pulumi.runtime.MockResourceArgs) -> None:
    """Fail the test run if a Bucket is declared without the mandatory controls."""
    if args.typ != "gcp:storage/bucket:Bucket":
        return
    inputs: dict[str, Any] = args.inputs
    assert inputs.get("uniformBucketLevelAccess") is True, (
        f"{args.name}: uniform bucket-level access must be enabled"
    )
    assert inputs.get("publicAccessPrevention") == "enforced", (
        f"{args.name}: public access prevention must be enforced"
    )
    assert inputs.get("versioning", {}).get("enabled") is True, (
        f"{args.name}: versioning must be enabled"
    )

Wire that helper into the new_resource mock from the Verification section and every bucket the program declares is checked, including ones added later by a component. The same idea applied to a whole plan is covered in scanning Python IaC with Checkov.

Access logs and their own bucket

logging.log_bucket writes hourly usage and storage reports as objects into a second bucket, and GCS needs roles/storage.objectCreator on that destination for the [email protected] group. Point log buckets at a dedicated project with a short lifecycle rule; sending logs back into the bucket being logged creates a slow feedback loop where each log object generates another log line.

FAQ

What is the difference between uniform bucket-level access and ACLs?

ACLs grant access per object or per bucket outside of IAM, which is how buckets accidentally go public. Uniform bucket-level access disables ACLs so IAM is the only access mechanism, making permissions consistent and auditable. Enable it at creation to avoid a later migration.

How do I make sure the bucket can never be made public?

Set public_access_prevention="enforced". GCP then rejects any IAM binding to allUsers or allAuthenticatedUsers at the bucket and the organization level can enforce it project-wide as well.

Should I use BucketIAMMember, BucketIAMBinding, or BucketIAMPolicy?

Use BucketIAMMember for a single principal/role grant (additive, safest). Use BucketIAMBinding to own all members of one role. Use BucketIAMPolicy only when you intend to manage the entire bucket policy authoritatively, since it overwrites unlisted bindings.

Do I need to grant anything for customer-managed encryption keys?

Yes. The GCS service agent for the project needs roles/cloudkms.cryptoKeyEncrypterDecrypter on the crypto key. Grant it with a CryptoKeyIAMMember in the same program so the bucket's CMEK dependency is explicit.

How does versioning interact with deletion?

With versioning enabled, deleting an object creates a noncurrent version rather than removing the data. Add a lifecycle rule to expire noncurrent versions after N days if you need cost control, and remember force_destroy=False blocks destroying a bucket that still holds any versions.

How do I let a GKE workload write to this bucket without a service-account key?

Use Workload Identity: bind the Kubernetes service account to a Google service account with roles/iam.workloadIdentityUser, then grant that Google service account roles/storage.objectAdmin on the bucket with a BucketIAMMember. The pod's client libraries pick up the federated credential from the metadata server, so no JSON key ever exists to leak or rotate.