diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/util/CombinedIPList.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/util/CombinedIPList.java
new file mode 100644
index 0000000000..1e985e411d
--- /dev/null
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/util/CombinedIPList.java
@@ -0,0 +1,59 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 org.apache.hadoop.util;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Util class to stores ips/hosts/subnets.
+ */
+public class CombinedIPList implements IPList {
+
+  public static final Logger LOG =
+      LoggerFactory.getLogger(CombinedIPList.class);
+
+  private final IPList[] networkLists;
+
+  public CombinedIPList(String fixedBlackListFile,
+      String variableBlackListFile, long cacheExpiryInSeconds) {
+
+    IPList fixedNetworkList = new FileBasedIPList(fixedBlackListFile);
+    if (variableBlackListFile != null) {
+      IPList variableNetworkList = new CacheableIPList(
+          new FileBasedIPList(variableBlackListFile), cacheExpiryInSeconds);
+      networkLists = new IPList[]{fixedNetworkList, variableNetworkList};
+    } else {
+      networkLists = new IPList[]{fixedNetworkList};
+    }
+  }
+
+  @Override
+  public boolean isIn(String ipAddress) {
+    if (ipAddress == null) {
+      throw new IllegalArgumentException("ipAddress is null");
+    }
+
+    for (IPList networkList : networkLists) {
+      if (networkList.isIn(ipAddress)) {
+        return true;
+      }
+    }
+    return false;
+  }
+}
\ No newline at end of file
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/datatransfer/BlackListBasedTrustedChannelResolver.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/datatransfer/BlackListBasedTrustedChannelResolver.java
new file mode 100644
index 0000000000..4fb24167ca
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/datatransfer/BlackListBasedTrustedChannelResolver.java
@@ -0,0 +1,143 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 org.apache.hadoop.hdfs.protocol.datatransfer;
+
+import java.net.InetAddress;
+import java.net.UnknownHostException;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.util.CombinedIPList;
+
+/**
+ * Implements {@link TrustedChannelResolver}
+ * to trust ips/host/subnets based on a blackList.
+ */
+public class BlackListBasedTrustedChannelResolver extends
+    TrustedChannelResolver {
+
+  private CombinedIPList blackListForServer;
+  private CombinedIPList blackListForClient;
+
+  private static final String FIXED_BLACK_LIST_DEFAULT_LOCATION = "/etc/hadoop"
+      + "/fixedBlackList";
+
+  private static final String VARIABLE_BLACK_LIST_DEFAULT_LOCATION = "/etc/"
+      + "hadoop/blackList";
+
+  /**
+   * Path to the file containing subnets and ip addresses to form
+   * fixed BlackList. Server side config.
+   */
+  public static final String DFS_DATATRANSFER_SERVER_FIXED_BLACK_LIST_FILE =
+      "dfs.datatransfer.server.fixedBlackList.file";
+  /**
+   * Enables/Disables variable BlackList. Server side config.
+   */
+  public static final String DFS_DATATRANSFER_SERVER_VARIABLE_BLACK_LIST_ENABLE
+      = "dfs.datatransfer.server.variableBlackList.enable";
+  /**
+   * Path to the file containing subnets and ip addresses to form
+   * variable BlackList. Server side config.
+   */
+  public static final String DFS_DATATRANSFER_SERVER_VARIABLE_BLACK_LIST_FILE =
+      "dfs.datatransfer.server.variableBlackList.file";
+  /**
+   * Time in seconds after which the variable BlackList file is checked for
+   * updates. Server side config.
+   */
+  public static final String
+      DFS_DATATRANSFER_SERVER_VARIABLE_BLACK_LIST_CACHE_SECS = "dfs."
+      + "datatransfer.server.variableBlackList.cache.secs";
+
+  /**
+   * Path to the file containing subnets and ip addresses to
+   * form fixed BlackList. This key is for client.
+   */
+  public static final String DFS_DATATRANSFER_CLIENT_FIXED_BLACK_LIST_FILE =
+      "dfs.datatransfer.client.fixedBlackList.file";
+  /**
+   * Enables/Disables variable BlackList. This key is for client.
+   */
+  public static final String DFS_DATATRANSFER_CLIENT_VARIABLE_BLACK_LIST_ENABLE
+      = "dfs.datatransfer.client.variableBlackList.enable";
+  /**
+   * Path to the file to containing subnets and ip addresses to form variable
+   * BlackList. This key is for client.
+   */
+  public static final String DFS_DATATRANSFER_CLIENT_VARIABLE_BLACK_LIST_FILE =
+      "dfs.datatransfer.client.variableBlackList.file";
+  /**
+   * Time in seconds after which the variable BlackList file is
+   * checked for updates. This key is for client.
+   */
+  public static final String
+      DFS_DATATRANSFER_CLIENT_VARIABLE_BLACK_LIST_CACHE_SECS =
+      "dfs.datatransfer.client.variableBlackList.cache.secs";
+
+  @Override
+  public void setConf(Configuration conf) {
+    super.setConf(conf);
+    String fixedFile = conf.get(DFS_DATATRANSFER_SERVER_FIXED_BLACK_LIST_FILE,
+        FIXED_BLACK_LIST_DEFAULT_LOCATION);
+    String variableFile = null;
+    long expiryTime = 0;
+
+    if (conf
+        .getBoolean(DFS_DATATRANSFER_SERVER_VARIABLE_BLACK_LIST_ENABLE,
+            false)) {
+      variableFile = conf.get(DFS_DATATRANSFER_SERVER_VARIABLE_BLACK_LIST_FILE,
+          VARIABLE_BLACK_LIST_DEFAULT_LOCATION);
+      expiryTime =
+          conf.getLong(DFS_DATATRANSFER_SERVER_VARIABLE_BLACK_LIST_CACHE_SECS,
+              3600) * 1000;
+    }
+
+    blackListForServer = new CombinedIPList(fixedFile, variableFile,
+        expiryTime);
+
+    fixedFile = conf
+        .get(DFS_DATATRANSFER_CLIENT_FIXED_BLACK_LIST_FILE, fixedFile);
+    expiryTime = 0;
+
+    if (conf
+        .getBoolean(DFS_DATATRANSFER_CLIENT_VARIABLE_BLACK_LIST_ENABLE,
+            false)) {
+      variableFile = conf
+          .get(DFS_DATATRANSFER_CLIENT_VARIABLE_BLACK_LIST_FILE, variableFile);
+      expiryTime =
+          conf.getLong(DFS_DATATRANSFER_CLIENT_VARIABLE_BLACK_LIST_CACHE_SECS,
+              3600) * 1000;
+    }
+
+    blackListForClient = new CombinedIPList(fixedFile, variableFile,
+        expiryTime);
+  }
+
+  public boolean isTrusted() {
+    try {
+      return !blackListForClient
+          .isIn(InetAddress.getLocalHost().getHostAddress());
+    } catch (UnknownHostException e) {
+      return true;
+    }
+  }
+
+  public boolean isTrusted(InetAddress clientAddress) {
+    return !blackListForServer.isIn(clientAddress.getHostAddress());
+  }
+}
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/datatransfer/package-info.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/datatransfer/package-info.java
new file mode 100644
index 0000000000..a13c7d8638
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/datatransfer/package-info.java
@@ -0,0 +1,24 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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.
+ */
+@InterfaceStability.Evolving
+package org.apache.hadoop.hdfs.protocol.datatransfer;
+import org.apache.hadoop.classification.InterfaceStability;
+
+/**
+ * This package contains classes related to hdfs data transfer protocol.
+ */
\ No newline at end of file
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/protocol/datatransfer/sasl/TestBlackListBasedTrustedChannelResolver.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/protocol/datatransfer/sasl/TestBlackListBasedTrustedChannelResolver.java
new file mode 100644
index 0000000000..75e1a4d023
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/protocol/datatransfer/sasl/TestBlackListBasedTrustedChannelResolver.java
@@ -0,0 +1,89 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 org.apache.hadoop.hdfs.protocol.datatransfer.sasl;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+import java.io.File;
+import java.io.IOException;
+import java.net.InetAddress;
+import java.net.UnknownHostException;
+import org.apache.commons.io.FileUtils;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hdfs.protocol.datatransfer.BlackListBasedTrustedChannelResolver;
+import org.apache.hadoop.test.GenericTestUtils;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+/**
+ * Test class for  {@link BlackListBasedTrustedChannelResolver}.
+ */
+public class TestBlackListBasedTrustedChannelResolver {
+
+  private final static String FILE_NAME = "blacklistfile.txt";
+  private File blacklistFile;
+  private final static String BLACK_LISTED = "127.0.0.1\n216.58.216.174\n";
+  private BlackListBasedTrustedChannelResolver resolver;
+
+  @Before
+  public void setup() {
+    blacklistFile = new File(GenericTestUtils.getTestDir(), FILE_NAME);
+    resolver
+        = new BlackListBasedTrustedChannelResolver();
+    try {
+      FileUtils.write(blacklistFile, BLACK_LISTED);
+    } catch (IOException e) {
+      fail("Setup for TestBlackListBasedTrustedChannelResolver failed.");
+    }
+  }
+
+  @After
+  public void cleanUp() {
+    FileUtils.deleteQuietly(blacklistFile);
+  }
+
+  @Test
+  public void testBlackListIpClient() throws IOException {
+    Configuration conf = new Configuration();
+    FileUtils.write(blacklistFile,
+        InetAddress.getLocalHost().getHostAddress(), true);
+    conf.set(BlackListBasedTrustedChannelResolver
+            .DFS_DATATRANSFER_CLIENT_FIXED_BLACK_LIST_FILE,
+        blacklistFile.getAbsolutePath());
+
+    resolver.setConf(conf);
+    assertFalse(resolver.isTrusted());
+
+  }
+
+  @Test
+  public void testBlackListIpServer() throws UnknownHostException {
+    Configuration conf = new Configuration();
+    conf.set(BlackListBasedTrustedChannelResolver
+            .DFS_DATATRANSFER_SERVER_FIXED_BLACK_LIST_FILE,
+        blacklistFile.getAbsolutePath());
+
+    resolver.setConf(conf);
+    assertTrue(resolver.isTrusted());
+    assertFalse(resolver.isTrusted(InetAddress
+        .getByName("216.58.216.174")));
+  }
+}