Add CSEK decryption for storage.get(BlobId,..)

google-cloud-clients/google-cloud-storage/src/main/java/com/google/cloud/storage/Blob.java

@@ -119,6 +119,8 @@ private Storage.BlobGetOption toGetOption(BlobInfo blobInfo) {
           return Storage.BlobGetOption.metagenerationNotMatch(blobInfo.getMetageneration());
         case USER_PROJECT:
           return Storage.BlobGetOption.userProject((String) getValue());
+        case CUSTOMER_SUPPLIED_KEY:
+          return Storage.BlobGetOption.decryptionKey((String) getValue());
         default:
           throw new AssertionError("Unexpected enum value");
       }

google-cloud-clients/google-cloud-storage/src/main/java/com/google/cloud/storage/Storage.java

@@ -748,6 +748,25 @@ public static BlobGetOption fields(BlobField... fields) {
     public static BlobGetOption userProject(String userProject) {
       return new BlobGetOption(StorageRpc.Option.USER_PROJECT, userProject);
     }
+
+    /**
+     * Returns an option to set a customer-supplied AES256 key for server-side decryption of the
+     * blob.
+     */
+    public static BlobGetOption decryptionKey(Key key) {
+      String base64Key = BaseEncoding.base64().encode(key.getEncoded());
+      return new BlobGetOption(StorageRpc.Option.CUSTOMER_SUPPLIED_KEY, base64Key);
+    }
+
+    /**
+     * Returns an option to set a customer-supplied AES256 key for server-side decryption of the
+     * blob.
+     *
+     * @param key the AES256 encoded in base64
+     */
+    public static BlobGetOption decryptionKey(String key) {
+      return new BlobGetOption(StorageRpc.Option.CUSTOMER_SUPPLIED_KEY, key);
+    }
   }
 
   /** Class for specifying bucket list options. */
@@ -1612,6 +1631,21 @@ Blob create(
    * Blob blob = storage.get(blobId, BlobGetOption.metagenerationMatch(blobMetageneration));
    * }</pre>
    *
+   * <p>Example of getting information on a blob encrypted using Customer Supplied Encryption Keys,
+   * only if supplied Decrpytion Key decrypts the blob successfully, otherwise a {@link
+   * StorageException} is thrown. For more information review
+   *
+   * @see <a
+   *     href="https://cloud.google.com/storage/docs/encryption/customer-supplied-keys#encrypted-elements">Encrypted
+   *     Elements</a>
+   *     <pre>{@code
+   * String bucketName = "my_unique_bucket";
+   * String blobName = "my_blob_name";
+   * String blobEncryptionKey = "";
+   * BlobId blobId = BlobId.of(bucketName, blobName);
+   * Blob blob = storage.get(blobId, BlobGetOption.decryptionKey(blobEncryptionKey));
+   * }</pre>
+   *
    * @throws StorageException upon failure
    */
   Blob get(BlobId blob, BlobGetOption... options);

google-cloud-clients/google-cloud-storage/src/main/java/com/google/cloud/storage/spi/v1/HttpStorageRpc.java

@@ -414,10 +414,9 @@ public Bucket get(Bucket bucket, Map<Option, ?> options) {
 
   private Storage.Objects.Get getCall(StorageObject object, Map<Option, ?> options)
       throws IOException {
-    return storage
-        .objects()
-        .get(object.getBucket(), object.getName())
-        .setGeneration(object.getGeneration())
+    Storage.Objects.Get get = storage.objects().get(object.getBucket(), object.getName());
+    setEncryptionHeaders(get.getRequestHeaders(), ENCRYPTION_KEY_PREFIX, options);
+    return get.setGeneration(object.getGeneration())
         .setProjection(DEFAULT_PROJECTION)
         .setIfMetagenerationMatch(Option.IF_METAGENERATION_MATCH.getLong(options))
         .setIfMetagenerationNotMatch(Option.IF_METAGENERATION_NOT_MATCH.getLong(options))

google-cloud-clients/google-cloud-storage/src/test/java/com/google/cloud/storage/it/ITStorageTest.java

@@ -481,7 +481,7 @@ public void testCreateBlobMd5Crc32cFromHexString() {
   }
 
   @Test
-  public void testCreateBlobWithEncryptionKey() {
+  public void testCreateGetBlobWithEncryptionKey() {
     String blobName = "test-create-with-customer-key-blob";
     BlobInfo blob = BlobInfo.newBuilder(BUCKET, blobName).build();
     Blob remoteBlob =
@@ -492,6 +492,13 @@ public void testCreateBlobWithEncryptionKey() {
     byte[] readBytes =
         storage.readAllBytes(BUCKET, blobName, Storage.BlobSourceOption.decryptionKey(BASE64_KEY));
     assertArrayEquals(BLOB_BYTE_CONTENT, readBytes);
+    remoteBlob =
+        storage.get(
+            blob.getBlobId(),
+            Storage.BlobGetOption.decryptionKey(BASE64_KEY),
+            Storage.BlobGetOption.fields(BlobField.CRC32C, BlobField.MD5HASH));
+    assertNotNull(remoteBlob.getCrc32c());
+    assertNotNull(remoteBlob.getMd5());
   }
 
   @Test