feat(storage): add ManagedFolder and Folder, and fix three defects in existing Storage code

CHANGELOG.md

@@ -12,6 +12,33 @@ formae agent.
 
 ### Added
 
+- `GCP::Storage::ManagedFolder` — an IAM boundary inside a bucket, letting a
+  policy be attached to a prefix without granting it over the whole bucket.
+  Requires uniform bucket-level access.
+- `GCP::Storage::Folder` — a real directory node, available only in a bucket
+  created with a hierarchical namespace. Renaming one moves everything beneath
+  it, where a managed folder only governs who may read a prefix.
+- `GCP::Storage::Bucket` models `iamConfiguration.uniformBucketLevelAccess` and
+  `hierarchicalNamespace`. Neither folder type can exist without them, and
+  hierarchical namespace is fixed at creation — a bucket is created flat or
+  hierarchical and cannot convert.
+
+### Fixed
+
+- `GCP::Storage::Bucket`'s resolvable names properties the resource actually
+  has. It pointed at `"Id"`, `"SelfLink"` and `"Name"` — capitalised, matching
+  nothing — so `bkt.res.name` resolved to no value and **any resource
+  referencing a bucket reached the plugin with the reference unresolved**. It
+  went unnoticed because nothing in the repository referenced a bucket until the
+  folder types did. `selfLink` is removed: the bucket has no such property.
+- Storage names containing a slash survive a native-ID round trip. Both folder
+  types are named with a **trailing slash that is part of the identity**
+  ("reports/" is not "reports"), and the parser took a single path segment, so
+  the slash was dropped and the rebuilt URL addressed a folder that does not
+  exist. The name is now taken whole and escaped when addressed — a no-op for
+  every pre-existing storage name, none of which contains a slash.
+
+### Fixed
 - `GCP::DNS::ResourceRecordSet` — what a managed zone actually serves: one name,
   one record type, and the data behind it. This completes Cloud DNS.
 

pkg/resources/storage/api.go

@@ -6,6 +6,7 @@ package storage
 
 import (
 	"fmt"
+	"net/url"
 	"strings"
 
 	"github.com/platform-engineering-labs/formae-plugin-gcp/pkg/resources/base"
@@ -77,7 +78,10 @@ func storagePathBuilder(ctx base.PathContext) string {
 	// Bucket-scoped
 	bucketName := bucketInfo
 	if ctx.ResourceName != "" {
-		return fmt.Sprintf("/b/%s/%s/%s", bucketName, ctx.ResourceType, ctx.ResourceName)
+		// A name containing a slash - a managed folder or a folder - has to be
+		// escaped, or the slash reads as another path segment. PathEscape is a
+		// no-op for every other storage name, which contains no slash.
+		return fmt.Sprintf("/b/%s/%s/%s", bucketName, ctx.ResourceType, url.PathEscape(ctx.ResourceName))
 	}
 	return fmt.Sprintf("/b/%s/%s", bucketName, ctx.ResourceType)
 }
@@ -153,7 +157,12 @@ func parseStorageNativeID(nativeID string) (base.PathContext, error) {
 	if len(parts) >= 4 && parts[0] == "b" && parts[2] != "o" {
 		ctx.ParentResource = parts[1] // bucket name
 		ctx.ResourceType = parts[2]
-		ctx.ResourceName = parts[3]
+		// Everything after the collection is the name. For an ACL entity or an
+		// anywhere cache that is a single segment and this is exactly parts[3];
+		// a managed folder or a folder is named with a trailing slash that is
+		// part of its identity ("reports/" is not "reports"), and taking one
+		// segment would silently truncate it.
+		ctx.ResourceName = strings.Join(parts[3:], "/")
 		return ctx, nil
 	}
 

pkg/resources/storage/bucket_walking_list.go

@@ -0,0 +1,115 @@
+// © 2025 Platform Engineering Labs Inc.
+//
+// SPDX-License-Identifier: FSL-1.1-ALv2
+
+package storage
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/platform-engineering-labs/formae-plugin-gcp/pkg/config"
+	"github.com/platform-engineering-labs/formae-plugin-gcp/pkg/resources/prov"
+	"github.com/platform-engineering-labs/formae-plugin-gcp/pkg/resources/registry"
+	"github.com/platform-engineering-labs/formae-plugin-gcp/pkg/transport"
+	"github.com/platform-engineering-labs/formae-plugin-gcp/pkg/utils"
+	"github.com/platform-engineering-labs/formae/pkg/plugin/resource"
+)
+
+// Both folder types live in a bucket, and discovery lists with no properties -
+// so it can name no bucket, the path builder falls through to the
+// project-scoped branch and asks /projects/{p}/managedFolders, which addresses
+// nothing. GCS has no wildcard for the bucket segment, so the only way to
+// discover either is to walk the buckets.
+//
+// This is the sixth copy of this shape in the plugin (Service Directory,
+// Spanner, Cloud SQL, DNS, Bigtable and now Storage). It belongs in base.
+type bucketWalkingListProvisioner struct {
+	prov.Provisioner
+	cfg        *config.Config
+	collection string
+}
+
+// registerBucketWalkingLists is called from the package init in resources.go so
+// the generic registration is guaranteed to have landed first.
+func registerBucketWalkingLists() {
+	for _, spec := range []struct {
+		resourceType string
+		collection   string
+	}{
+		{ManagedFolderResourceType, "managedFolders"},
+		{FolderResourceType, "folders"},
+	} {
+		spec := spec
+		registry.Register(spec.resourceType,
+			[]resource.Operation{resource.OperationList},
+			func(cfg *config.Config) prov.Provisioner {
+				return &bucketWalkingListProvisioner{
+					Provisioner: storageRegistry.CreateProvisioner(cfg, spec.resourceType),
+					cfg:         cfg,
+					collection:  spec.collection,
+				}
+			})
+	}
+}
+
+func (p *bucketWalkingListProvisioner) List(
+	ctx context.Context, request *resource.ListRequest,
+) (*resource.ListResult, error) {
+	// A named bucket is the caller telling us where to look.
+	if request.AdditionalProperties != nil && request.AdditionalProperties["bucket"] != "" {
+		return p.Provisioner.List(ctx, request)
+	}
+
+	cfg := config.PathFromTargetConfig(request.TargetConfig)
+	if cfg.Project == "" {
+		return &resource.ListResult{NativeIDs: []string{}}, nil
+	}
+
+	client, err := transport.NewClient(ctx, p.cfg)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create transport client: %w", err)
+	}
+
+	bucketsURL := fmt.Sprintf("%s/b?project=%s", StorageAPI.BaseURL, cfg.Project)
+	resp, err := client.SendRequest(ctx, transport.RequestOptions{Method: "GET", URL: bucketsURL})
+	if err != nil {
+		wrapped := transport.WrapError(err, "failed to list storage buckets")
+		return nil, fmt.Errorf("%s", wrapped.Message)
+	}
+
+	nativeIDs := []string{}
+	buckets, _ := resp.Body["items"].([]interface{})
+	for _, raw := range buckets {
+		bucket, ok := raw.(map[string]interface{})
+		if !ok {
+			continue
+		}
+		bucketName := utils.GetString(bucket, "name")
+		if bucketName == "" {
+			continue
+		}
+		itemsResp, listErr := client.SendRequest(ctx, transport.RequestOptions{
+			Method: "GET",
+			URL:    fmt.Sprintf("%s/b/%s/%s", StorageAPI.BaseURL, bucketName, p.collection),
+		})
+		if listErr != nil {
+			// A flat bucket has no folders and a bucket with ACLs has no managed
+			// folders; neither is a reason to hide the rest.
+			continue
+		}
+		items, _ := itemsResp.Body["items"].([]interface{})
+		for _, rawItem := range items {
+			item, ok := rawItem.(map[string]interface{})
+			if !ok {
+				continue
+			}
+			// The name carries its trailing slash, which the native ID keeps.
+			if name := utils.GetString(item, "name"); name != "" {
+				nativeIDs = append(nativeIDs,
+					fmt.Sprintf("b/%s/%s/%s", bucketName, p.collection, name))
+			}
+		}
+	}
+	return &resource.ListResult{NativeIDs: nativeIDs}, nil
+}

pkg/resources/storage/resources.go

@@ -18,6 +18,8 @@ const (
 	AnywhereCacheResourceType              = "GCP::Storage::AnywhereCache"
 	BucketAccessControlResourceType        = "GCP::Storage::BucketAccessControl"
 	DefaultObjectAccessControlResourceType = "GCP::Storage::DefaultObjectAccessControl"
+	ManagedFolderResourceType              = "GCP::Storage::ManagedFolder"
+	FolderResourceType                     = "GCP::Storage::Folder"
 	ObjectAccessControlResourceType        = "GCP::Storage::ObjectAccessControl"
 )
 
@@ -144,6 +146,47 @@ func init() {
 			RequestTransformer:  wrapBodyBuilder(aclBodyBuilder),
 			ResponseTransformer: nil,
 		},
+		{
+			// A managed folder is an IAM boundary inside a bucket: it lets a
+			// policy be attached to a prefix without giving it to the whole
+			// bucket. It requires uniform bucket-level access - GCS refuses to
+			// create one where per-object ACLs still apply.
+			//
+			// Its name ends with a slash, which is part of its identity and is
+			// escaped in the URL.
+			ResourceType: ManagedFolderResourceType,
+			ResourceConfig: base.ResourceConfig{
+				ResourceType: "managedFolders",
+				ParentResource: &base.ParentResourceConfig{
+					ParentType:         "bucket",
+					RequiresParent:     true,
+					ParentPathSegments: []string{"b"},
+				},
+				// A managed folder carries nothing but its name; there is
+				// nothing an update could change.
+				SupportsUpdate: false,
+			},
+			RequestTransformer:  base.DropFields("bucket"),
+			ResponseTransformer: base.ResponseTransformerFunc(bucketScopedResponseTransformer),
+		},
+		{
+			// A folder is a real directory, available only in a bucket created
+			// with a hierarchical namespace. Where a managed folder is an IAM
+			// boundary over a prefix, a folder is an actual node - renaming one
+			// moves everything beneath it.
+			ResourceType: FolderResourceType,
+			ResourceConfig: base.ResourceConfig{
+				ResourceType: "folders",
+				ParentResource: &base.ParentResourceConfig{
+					ParentType:         "bucket",
+					RequiresParent:     true,
+					ParentPathSegments: []string{"b"},
+				},
+				SupportsUpdate: false,
+			},
+			RequestTransformer:  base.DropFields("bucket"),
+			ResponseTransformer: base.ResponseTransformerFunc(bucketScopedResponseTransformer),
+		},
 		// NOTE: ObjectAccessControlResourceType requires special handling for object-scoped resources
 		// The base package currently doesn't support resources that need TWO parent properties (bucket + object).
 		// This resource type is commented out pending enhancement to base package's parent extraction mechanism.
@@ -172,4 +215,24 @@ func init() {
 	if err != nil {
 		panic(err)
 	}
+
+	registerBucketWalkingLists()
+}
+
+// bucketScopedResponseTransformer puts back the bucket a folder belongs to and
+// drops what GCS echoes that describes the request rather than the resource.
+// Both folder types report "bucket" themselves, so unlike most nested resources
+// here nothing has to be recovered from the URL.
+func bucketScopedResponseTransformer(
+	props map[string]interface{}, _ base.TransformContext,
+) map[string]interface{} {
+	out := make(map[string]interface{}, len(props))
+	for k, v := range props {
+		switch k {
+		case "kind", "selfLink", "metageneration", "id":
+			continue
+		}
+		out[k] = v
+	}
+	return out
 }

pkg/resources/storage/storage_native_id_unit_test.go

@@ -0,0 +1,92 @@
+// © 2025 Platform Engineering Labs Inc.
+//
+// SPDX-License-Identifier: FSL-1.1-ALv2
+
+//go:build unit
+
+package storage
+
+import (
+	"testing"
+
+	"github.com/platform-engineering-labs/formae-plugin-gcp/pkg/resources/base"
+)
+
+// The names of every storage type that existed before managed folders are a
+// single segment. These pin that the parser and the path builder still treat
+// them exactly as they did - this is the half of the change that must not move.
+func TestExistingStorageNamesAreUnchanged(t *testing.T) {
+	cases := map[string]struct {
+		nativeID string
+		wantPath string
+	}{
+		"bucket acl entity": {
+			"b/my-bucket/acl/user-someone@example.com",
+			"/b/my-bucket/acl/user-someone@example.com",
+		},
+		"default object acl": {
+			"b/my-bucket/defaultObjectAcl/allUsers",
+			"/b/my-bucket/defaultObjectAcl/allUsers",
+		},
+		"anywhere cache": {
+			"b/my-bucket/anywhereCaches/cache-1",
+			"/b/my-bucket/anywhereCaches/cache-1",
+		},
+	}
+	for name, tc := range cases {
+		ctx, err := parseStorageNativeID(tc.nativeID)
+		if err != nil {
+			t.Errorf("%s: parse: %v", name, err)
+			continue
+		}
+		if got := storagePathBuilder(ctx); got != tc.wantPath {
+			t.Errorf("%s: path = %q, want %q", name, got, tc.wantPath)
+		}
+	}
+}
+
+// A managed folder is named with a trailing slash that is part of its identity:
+// "reports/" is not "reports". Splitting on "/" and taking one segment dropped
+// it, and the rebuilt URL then addressed a folder that does not exist.
+func TestSlashTerminatedNamesSurviveAndAreEscaped(t *testing.T) {
+	ctx, err := parseStorageNativeID("b/my-bucket/managedFolders/reports/")
+	if err != nil {
+		t.Fatalf("parse: %v", err)
+	}
+	if ctx.ParentResource != "my-bucket" || ctx.ResourceType != "managedFolders" {
+		t.Errorf("ctx = %+v", ctx)
+	}
+	if ctx.ResourceName != "reports/" {
+		t.Errorf("name = %q, want %q", ctx.ResourceName, "reports/")
+	}
+	// The slash has to be escaped or it reads as another path segment.
+	want := "/b/my-bucket/managedFolders/reports%2F"
+	if got := storagePathBuilder(ctx); got != want {
+		t.Errorf("path = %q, want %q", got, want)
+	}
+
+	// Nested folders keep every segment.
+	nested, err := parseStorageNativeID("b/my-bucket/folders/a/b/")
+	if err != nil {
+		t.Fatalf("parse nested: %v", err)
+	}
+	if nested.ResourceName != "a/b/" {
+		t.Errorf("nested name = %q", nested.ResourceName)
+	}
+}
+
+// A bucket is still addressed by a bare name, and a project-scoped resource
+// still by its project path.
+func TestBucketAndProjectScopedShapesStillParse(t *testing.T) {
+	b, err := parseStorageNativeID("my-bucket")
+	if err != nil || b.ResourceType != "b" || b.ResourceName != "my-bucket" {
+		t.Errorf("bucket ctx = %+v err=%v", b, err)
+	}
+	p, err := parseStorageNativeID("projects/proj/hmacKeys/GOOG1EXAMPLE")
+	if err != nil || p.ResourceType != "hmacKeys" || p.ResourceName != "GOOG1EXAMPLE" || p.ParentResource != "" {
+		t.Errorf("project ctx = %+v err=%v", p, err)
+	}
+	if got := storagePathBuilder(base.PathContext{Project: "proj", ResourceType: "hmacKeys", ResourceName: "GOOG1EXAMPLE"}); got != "/projects/proj/hmacKeys/GOOG1EXAMPLE" {
+		t.Errorf("project path = %q", got)
+	}
+}