Upstream fix for GCS object owner is optional (#5618) (#11006)

* GCS object owner is optional

- There are some cases (one case is described in #10342 (comment)),
  where `Object.owner` is missing, which leads to nil pointer dereference.

Fixes #10342

* remove comment

* skip vcr test

Co-authored-by: Aniruddha Maru <[email protected]>
Signed-off-by: Modular Magician <[email protected]>

Co-authored-by: Aniruddha Maru <[email protected]>

Author: modular-magician

.changelog/5618.txt

@@ -0,0 +1,3 @@
+```release-note:bug
+storage: Fixed bug where the provider crashes when `Object.owner` is missing when using `google_storage_object_acl`
+```

google/resource_storage_object_acl.go

@@ -80,7 +80,10 @@ func resourceStorageObjectAclDiff(_ context.Context, diff *schema.ResourceDiff,
 		return nil
 	}
 
-	objectOwner := sObject.Owner.Entity
+	var objectOwner string
+	if sObject.Owner != nil {
+		objectOwner = sObject.Owner.Entity
+	}
 	ownerRole := fmt.Sprintf("%s:%s", "OWNER", objectOwner)
 	oldRoleEntity, newRoleEntity := diff.GetChange("role_entity")
 
@@ -146,7 +149,10 @@ func resourceStorageObjectAclCreate(d *schema.ResourceData, meta interface{}) er
 			return fmt.Errorf("Error reading object %s in %s: %v", object, bucket, err)
 		}
 
-		objectOwner := sObject.Owner.Entity
+		var objectOwner string
+		if sObject.Owner != nil {
+			objectOwner = sObject.Owner.Entity
+		}
 		roleEntitiesUpstream, err := getRoleEntitiesAsStringsFromApi(config, bucket, object, userAgent)
 		if err != nil {
 			return fmt.Errorf("Error reading object %s in %s: %v", object, bucket, err)
@@ -227,7 +233,10 @@ func resourceStorageObjectAclUpdate(d *schema.ResourceData, meta interface{}) er
 			return fmt.Errorf("Error reading object %s in %s: %v", object, bucket, err)
 		}
 
-		objectOwner := sObject.Owner.Entity
+		var objectOwner string
+		if sObject.Owner != nil {
+			objectOwner = sObject.Owner.Entity
+		}
 
 		o, n := d.GetChange("role_entity")
 		create, update, remove, err := getRoleEntityChange(

google/resource_storage_object_acl_test.go

@@ -1,11 +1,18 @@
 package google
 
 import (
+	"bytes"
+	"context"
+	"encoding/json"
 	"fmt"
+	"io"
 	"io/ioutil"
+	"net/http"
 	"testing"
 
+	"github.com/hashicorp/terraform-plugin-sdk/v2/diag"
 	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/resource"
+	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/schema"
 	"github.com/hashicorp/terraform-plugin-sdk/v2/terraform"
 )
 
@@ -31,8 +38,7 @@ func TestAccStorageObjectAcl_basic(t *testing.T) {
 			}
 			testAccPreCheck(t)
 		},
-		Providers:    testAccProviders,
-		CheckDestroy: testAccStorageObjectAclDestroyProducer(t),
+		Providers: testAccProviders,
 		Steps: []resource.TestStep{
 			{
 				Config: testGoogleStorageObjectsAclBasic1(bucketName, objectName),
@@ -278,6 +284,82 @@ func TestAccStorageObjectAcl_unordered(t *testing.T) {
 	})
 }
 
+// a round tripper that returns fake response for get object API removing `owner` attribute
+// it only modifies the response once, since otherwise resource will fail to delete.
+type testRoundTripper struct {
+	http.RoundTripper
+	bucketName, objectName string
+	done                   bool
+}
+
+func (t *testRoundTripper) RoundTrip(r *http.Request) (*http.Response, error) {
+	response, err := t.RoundTripper.RoundTrip(r)
+	if err != nil {
+		return response, err
+	}
+	expectedPath := fmt.Sprintf("/storage/v1/b/%s/o/%s", t.bucketName, t.objectName)
+	if t.done || r.URL.Path != expectedPath || r.Host != "storage.googleapis.com" {
+		return response, err
+	}
+	t.done = true
+	responseBytes, err := ioutil.ReadAll(response.Body)
+	if err != nil {
+		return response, err
+	}
+	var responseMap map[string]json.RawMessage
+	err = json.Unmarshal(responseBytes, &responseMap)
+	if err != nil {
+		return response, err
+	}
+	delete(responseMap, "owner")
+	responseBytes, err = json.Marshal(responseMap)
+	if err != nil {
+		return response, err
+	}
+	response.Body = io.NopCloser(bytes.NewBuffer(responseBytes))
+	return response, nil
+}
+
+// Test that we don't fail if there's no owner for object
+func TestAccStorageObjectAcl_noOwner(t *testing.T) {
+	skipIfVcr(t)
+	t.Parallel()
+
+	bucketName := testBucketName(t)
+	objectName := testAclObjectName(t)
+	objectData := []byte("data data data")
+	if err := ioutil.WriteFile(tfObjectAcl.Name(), objectData, 0644); err != nil {
+		t.Errorf("error writing file: %v", err)
+	}
+	provider := Provider()
+	oldConfigureFunc := provider.ConfigureContextFunc
+	provider.ConfigureContextFunc = func(ctx context.Context, d *schema.ResourceData) (interface{}, diag.Diagnostics) {
+		c, diagnostics := oldConfigureFunc(ctx, d)
+		config := c.(*Config)
+		roundTripper := &testRoundTripper{RoundTripper: config.client.Transport, bucketName: bucketName, objectName: objectName}
+		config.client.Transport = roundTripper
+		return c, diagnostics
+	}
+	providers := map[string]*schema.Provider{
+		"google": provider,
+	}
+	vcrTest(t, resource.TestCase{
+		PreCheck: func() {
+			if errObjectAcl != nil {
+				panic(errObjectAcl)
+			}
+			testAccPreCheck(t)
+		},
+		Providers: providers,
+		Steps: []resource.TestStep{
+			{
+				Config:             testGoogleStorageObjectsAclBasic1(bucketName, objectName),
+				ExpectNonEmptyPlan: true,
+			},
+		},
+	})
+}
+
 func testAccCheckGoogleStorageObjectAcl(t *testing.T, bucket, object, roleEntityS string) resource.TestCheckFunc {
 	return func(s *terraform.State) error {
 		roleEntity, _ := getRoleEntityPair(roleEntityS)