Fix GCS path parsing

server/src/main/scala/io/delta/sharing/server/CloudFileSigner.scala

@@ -22,6 +22,7 @@ import java.util.concurrent.TimeUnit.SECONDS
 
 import com.amazonaws.HttpMethod
 import com.amazonaws.services.s3.model.GeneratePresignedUrlRequest
+import com.google.cloud.hadoop.gcsio.StorageResourceId
 import com.google.cloud.storage.BlobId
 import com.google.cloud.storage.BlobInfo
 import com.google.cloud.storage.Storage
@@ -202,13 +203,18 @@ class GCSFileSigner(
   private val storage = StorageOptions.newBuilder.build.getService
 
   override def sign(path: Path): String = {
-    val absPath = path.toUri
-    val bucketName = absPath.getHost
-    val objectName = absPath.getPath.stripPrefix("/")
+    val (bucketName, objectName) = GCSFileSigner.getBucketAndObjectNames(path)
     assert(objectName.nonEmpty, s"cannot get object key from $path")
     val blobInfo = BlobInfo.newBuilder(BlobId.of(bucketName, objectName)).build
     storage.signUrl(
       blobInfo, preSignedUrlTimeoutSeconds, SECONDS, Storage.SignUrlOption.withV4Signature())
       .toString
   }
 }
+
+object GCSFileSigner {
+  def getBucketAndObjectNames(path: Path): (String, String) = {
+    val resourceId = StorageResourceId.fromUriPath(path.toUri, false /* = allowEmptyObjectName */)
+    (resourceId.getBucketName, resourceId.getObjectName)
+  }
+}

server/src/test/scala/io/delta/sharing/server/CloudFileSignerSuite.scala

@@ -0,0 +1,30 @@
+/*
+ * Copyright (2021) The Delta Lake Project Authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.delta.sharing.server
+
+import org.apache.hadoop.fs.Path
+import org.scalatest.FunSuite
+
+class CloudFileSignerSuite extends FunSuite {
+
+  test("GCSFileSigner.getBucketAndObjectNames") {
+    assert(GCSFileSigner.getBucketAndObjectNames(new Path("gs://delta-sharing-test/foo"))
+      == ("delta-sharing-test", "foo"))
+    assert(GCSFileSigner.getBucketAndObjectNames(new Path("gs://delta_sharing_test/foo"))
+      == ("delta_sharing_test", "foo"))
+  }
+}