HADOOP-11337. KeyAuthorizationKeyProvider access checks need to be done atomically. Contributed by Dian Fu.

(cherry picked from commit 9fa2990257)

hadoop-common-project/hadoop-common/CHANGES.txt

@@ -116,6 +116,9 @@ Release 2.7.0 - UNRELEASED
     HADOOP-11333. Fix deadlock in DomainSocketWatcher when the notification
     pipe is full (zhaoyunjiong via cmccabe)
 
+    HADOOP-11337. KeyAuthorizationKeyProvider access checks need to be done
+    atomically. (Dian Fu via wang)
+
 Release 2.6.0 - 2014-11-18
 
   INCOMPATIBLE CHANGES

hadoop-common-project/hadoop-kms/src/main/java/org/apache/hadoop/crypto/key/kms/server/KeyAuthorizationKeyProvider.java

@@ -23,6 +23,9 @@ import java.security.GeneralSecurityException;
 import java.security.NoSuchAlgorithmException;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
 
 import org.apache.hadoop.crypto.key.KeyProvider;
 import org.apache.hadoop.crypto.key.KeyProviderCryptoExtension;
@@ -83,6 +86,8 @@ public class KeyAuthorizationKeyProvider extends KeyProviderCryptoExtension {
 
   private final KeyProviderCryptoExtension provider;
   private final KeyACLs acls;
+  private Lock readLock;
+  private Lock writeLock;
 
   /**
    * The constructor takes a {@link KeyProviderCryptoExtension} and an
@@ -96,6 +101,9 @@ public class KeyAuthorizationKeyProvider extends KeyProviderCryptoExtension {
     super(keyProvider, null);
     this.provider = keyProvider;
     this.acls = acls;
+    ReadWriteLock lock = new ReentrantReadWriteLock(true);
+    readLock = lock.readLock();
+    writeLock = lock.writeLock();
   }
 
   // This method first checks if "key.acl.name" attribute is present as an
@@ -146,50 +154,85 @@ public class KeyAuthorizationKeyProvider extends KeyProviderCryptoExtension {
   @Override
   public KeyVersion createKey(String name, Options options)
       throws NoSuchAlgorithmException, IOException {
+    writeLock.lock();
+    try {
       authorizeCreateKey(name, options, getUser());
       return provider.createKey(name, options);
+    } finally {
+      writeLock.unlock();
+    }
   }
 
   @Override
   public KeyVersion createKey(String name, byte[] material, Options options)
       throws IOException {
+    writeLock.lock();
+    try {
       authorizeCreateKey(name, options, getUser());
       return provider.createKey(name, material, options);
+    } finally {
+      writeLock.unlock();
+    }
   }
 
   @Override
   public KeyVersion rollNewVersion(String name)
       throws NoSuchAlgorithmException, IOException {
+    writeLock.lock();
+    try {
       doAccessCheck(name, KeyOpType.MANAGEMENT);
       return provider.rollNewVersion(name);
+    } finally {
+      writeLock.unlock();
+    }
   }
 
   @Override
   public void deleteKey(String name) throws IOException {
+    writeLock.lock();
+    try {
       doAccessCheck(name, KeyOpType.MANAGEMENT);
       provider.deleteKey(name);
+    } finally {
+      writeLock.unlock();
+    }
   }
 
   @Override
   public KeyVersion rollNewVersion(String name, byte[] material)
       throws IOException {
+    writeLock.lock();
+    try {
       doAccessCheck(name, KeyOpType.MANAGEMENT);
       return provider.rollNewVersion(name, material);
+    } finally {
+      writeLock.unlock();
+    }
   }
 
   @Override
   public void warmUpEncryptedKeys(String... names) throws IOException {
+    readLock.lock();
+    try {
       for (String name : names) {
         doAccessCheck(name, KeyOpType.GENERATE_EEK);
       }
       provider.warmUpEncryptedKeys(names);
+    } finally {
+      readLock.unlock();
+    }
   }
 
   @Override
   public EncryptedKeyVersion generateEncryptedKey(String encryptionKeyName)
       throws IOException, GeneralSecurityException {
+    readLock.lock();
+    try {
       doAccessCheck(encryptionKeyName, KeyOpType.GENERATE_EEK);
       return provider.generateEncryptedKey(encryptionKeyName);
+    } finally {
+      readLock.unlock();
+    }
   }
 
   private void verifyKeyVersionBelongsToKey(EncryptedKeyVersion ekv)
@@ -206,19 +249,29 @@ public class KeyAuthorizationKeyProvider extends KeyProviderCryptoExtension {
   @Override
   public KeyVersion decryptEncryptedKey(EncryptedKeyVersion encryptedKeyVersion)
           throws IOException, GeneralSecurityException {
+    readLock.lock();
+    try {
       verifyKeyVersionBelongsToKey(encryptedKeyVersion);
       doAccessCheck(
           encryptedKeyVersion.getEncryptionKeyName(), KeyOpType.DECRYPT_EEK);
       return provider.decryptEncryptedKey(encryptedKeyVersion);
+    } finally {
+      readLock.unlock();
+    }
   }
 
   @Override
   public KeyVersion getKeyVersion(String versionName) throws IOException {
+    readLock.lock();
+    try {
       KeyVersion keyVersion = provider.getKeyVersion(versionName);
       if (keyVersion != null) {
         doAccessCheck(keyVersion.getName(), KeyOpType.READ);
       }
       return keyVersion;
+    } finally {
+      readLock.unlock();
+    }
   }
 
   @Override
@@ -228,28 +281,48 @@ public class KeyAuthorizationKeyProvider extends KeyProviderCryptoExtension {
 
   @Override
   public List<KeyVersion> getKeyVersions(String name) throws IOException {
+    readLock.lock();
+    try {
       doAccessCheck(name, KeyOpType.READ);
       return provider.getKeyVersions(name);
+    } finally {
+      readLock.unlock();
+    }
   }
 
   @Override
   public Metadata getMetadata(String name) throws IOException {
+    readLock.lock();
+    try {
       doAccessCheck(name, KeyOpType.READ);
       return provider.getMetadata(name);
+    } finally {
+      readLock.unlock();
+    }
   }
 
   @Override
   public Metadata[] getKeysMetadata(String... names) throws IOException {
+    readLock.lock();
+    try {
       for (String name : names) {
         doAccessCheck(name, KeyOpType.READ);
       }
       return provider.getKeysMetadata(names);
+    } finally {
+      readLock.unlock();
+    }
   }
 
   @Override
   public KeyVersion getCurrentKey(String name) throws IOException {
+    readLock.lock();
+    try {
       doAccessCheck(name, KeyOpType.READ);
       return provider.getCurrentKey(name);
+    } finally {
+      readLock.unlock();
+    }
   }
 
   @Override