From 2c367b464c86a7d67a2b8dd82ae804d169957573 Mon Sep 17 00:00:00 2001
From: Andrew Wang <wang@apache.org>
Date: Mon, 26 Jun 2017 11:20:07 -0700
Subject: [PATCH] HDFS-11956. Do not require a storage ID or target storage IDs
 when writing a block. Contributed by Ewan Higgs.

---
 .../token/block/BlockTokenSecretManager.java  |  1 +
 .../hdfs/server/datanode/DataXceiver.java     | 17 ++++++---
 .../security/token/block/TestBlockToken.java  | 35 +++++++++++++++++++
 3 files changed, 48 insertions(+), 5 deletions(-)

diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/security/token/block/BlockTokenSecretManager.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/security/token/block/BlockTokenSecretManager.java
index 8be22d966ec..da830a6d419 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/security/token/block/BlockTokenSecretManager.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/security/token/block/BlockTokenSecretManager.java
@@ -112,6 +112,7 @@ public BlockTokenSecretManager(long keyUpdateInterval,
    * @param blockPoolId block pool ID
    * @param encryptionAlgorithm encryption algorithm to use
    * @param numNNs number of namenodes possible
+   * @param useProto should we use new protobuf style tokens
    */
   public BlockTokenSecretManager(long keyUpdateInterval,
       long tokenLifetime, int nnIndex, int numNNs,  String blockPoolId,
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DataXceiver.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DataXceiver.java
index d42e3307e5a..8ffd3a4f3fc 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DataXceiver.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DataXceiver.java
@@ -695,11 +695,18 @@ public void writeBlock(final ExtendedBlock block,
     if (targetStorageTypes.length > 0) {
       System.arraycopy(targetStorageTypes, 0, storageTypes, 1, nst);
     }
-    int nsi = targetStorageIds.length;
-    String[] storageIds = new String[nsi + 1];
-    storageIds[0] = storageId;
-    if (targetStorageTypes.length > 0) {
-      System.arraycopy(targetStorageIds, 0, storageIds, 1, nsi);
+
+    // To support older clients, we don't pass in empty storageIds
+    final int nsi = targetStorageIds.length;
+    final String[] storageIds;
+    if (nsi > 0) {
+      storageIds = new String[nsi + 1];
+      storageIds[0] = storageId;
+      if (targetStorageTypes.length > 0) {
+        System.arraycopy(targetStorageIds, 0, storageIds, 1, nsi);
+      }
+    } else {
+      storageIds = new String[0];
     }
     checkAccess(replyOut, isClient, block, blockToken, Op.WRITE_BLOCK,
         BlockTokenIdentifier.AccessMode.WRITE,
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/security/token/block/TestBlockToken.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/security/token/block/TestBlockToken.java
index 747f2952d2c..aaddb3654aa 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/security/token/block/TestBlockToken.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/security/token/block/TestBlockToken.java
@@ -774,4 +774,39 @@ public void testBlockTokenSerialization() throws IOException {
     testBlockTokenSerialization(false);
     testBlockTokenSerialization(true);
   }
+
+  private void testBadStorageIDCheckAccess(boolean enableProtobuf)
+      throws IOException {
+    BlockTokenSecretManager sm = new BlockTokenSecretManager(
+        blockKeyUpdateInterval, blockTokenLifetime, 0, 1, "fake-pool", null,
+        enableProtobuf);
+    StorageType[] storageTypes = new StorageType[] {StorageType.DISK};
+    String[] storageIds = new String[] {"fake-storage-id"};
+    String[] badStorageIds = new String[] {"BAD-STORAGE-ID"};
+    String[] emptyStorageIds = new String[] {};
+    BlockTokenIdentifier.AccessMode mode = BlockTokenIdentifier.AccessMode.READ;
+    BlockTokenIdentifier id = generateTokenId(sm, block3,
+        EnumSet.of(mode), storageTypes, storageIds);
+    sm.checkAccess(id, null, block3, mode, storageTypes, storageIds);
+
+    try {
+      sm.checkAccess(id, null, block3, mode, storageTypes, badStorageIds);
+      fail("Expected strict BlockTokenSecretManager to fail");
+    } catch(SecretManager.InvalidToken e) {
+    }
+    // We allow empty storageId tokens for backwards compatibility. i.e. old
+    // clients may not have known to pass the storageId parameter to the
+    // writeBlock api.
+    sm.checkAccess(id, null, block3, mode, storageTypes,
+        emptyStorageIds);
+    sm.checkAccess(id, null, block3, mode, storageTypes,
+        null);
+  }
+
+  @Test
+  public void testBadStorageIDCheckAccess() throws IOException {
+    testBadStorageIDCheckAccess(false);
+    testBadStorageIDCheckAccess(true);
+  }
+
 }