azkaban-aplcache

diff --git a/azkaban-common/src/main/java/azkaban/Constants.java b/azkaban-common/src/main/java/azkaban/Constants.java
index 2b228ac..1780e1b 100644
--- a/azkaban-common/src/main/java/azkaban/Constants.java
+++ b/azkaban-common/src/main/java/azkaban/Constants.java
@@ -84,6 +84,9 @@ public class Constants {
     // Hostname for the host, if not specified, canonical hostname will be used
     public static final String AZKABAN_SERVER_HOST_NAME = "azkaban.server.hostname";
 
+    // List of users we prevent azkaban from running flows as. (ie: root, azkaban)
+    public static final String BLACK_LISTED_USERS = "azkaban.server.blacklist.users";
+
     // Legacy configs section, new configs should follow the naming convention of azkaban.server.<rest of the name> for server configs.
 
     // The property is used for the web server to get the host name of the executor when running in SOLO mode.

diff --git a/azkaban-common/src/main/java/azkaban/jobExecutor/ProcessJob.java b/azkaban-common/src/main/java/azkaban/jobExecutor/ProcessJob.java
index c45d25d..ead5147 100644
--- a/azkaban-common/src/main/java/azkaban/jobExecutor/ProcessJob.java
+++ b/azkaban-common/src/main/java/azkaban/jobExecutor/ProcessJob.java
@@ -31,6 +31,9 @@ import java.time.Duration;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
+import java.util.HashSet;
+import java.util.Set;
+import java.util.Arrays;
 import java.util.concurrent.TimeUnit;
 import org.apache.log4j.Logger;
 
@@ -203,18 +206,26 @@ public class ProcessJob extends AbstractProcessJob {
     String effectiveUser = null;
     final boolean isExecuteAsUser = this.sysProps.getBoolean(EXECUTE_AS_USER, true);
 
+    //Get list of users we never execute flows as. (ie: root, azkaban)
+    final Set<String> blackListedUsers = new HashSet<String>(
+        Arrays.asList(
+            this.sysProps.getString(Constants.ConfigurationKeys.BLACK_LISTED_USERS,"root,azkaban").split(",")
+        )
+    );
+
     // nativeLibFolder specifies the path for execute-as-user file,
     // which will change user from Azkaban to effectiveUser
     if (isExecuteAsUser) {
       final String nativeLibFolder = this.sysProps.getString(NATIVE_LIB_FOLDER);
-      executeAsUserBinaryPath =
-          String.format("%s/%s", nativeLibFolder, "execute-as-user");
+      executeAsUserBinaryPath = String.format("%s/%s", nativeLibFolder, "execute-as-user");
       effectiveUser = getEffectiveUser(this.jobProps);
-      if ("root".equals(effectiveUser)) {
-        throw new RuntimeException(
-            "Not permitted to proxy as root through Azkaban");
+        // Throw exception if Azkaban tries to run flow as a prohibited user
+        if (blackListedUsers.contains(effectiveUser)) {
+          throw new RuntimeException(
+              String.format("Not permitted to proxy as '%s' through Azkaban", effectiveUser)
+          );
+        }
       }
-    }
 
     for (String command : commands) {
       AzkabanProcessBuilder builder = null;

diff --git a/azkaban-common/src/test/java/azkaban/jobExecutor/ProcessJobTest.java b/azkaban-common/src/test/java/azkaban/jobExecutor/ProcessJobTest.java
index 07f13c0..e05ca88 100644
--- a/azkaban-common/src/test/java/azkaban/jobExecutor/ProcessJobTest.java
+++ b/azkaban-common/src/test/java/azkaban/jobExecutor/ProcessJobTest.java
@@ -105,6 +105,38 @@ public class ProcessJobTest {
 
   }
 
+  /**
+   * this job should fail because it sets user.to.proxy = root which is black listed
+   */
+  @Test(expected = RuntimeException.class)
+  public void testOneUnixCommandWithRootUser() throws Exception {
+
+    // Initialize the Props
+    this.props.removeLocal(CommonJobProperties.SUBMIT_USER);
+    this.props.put("user.to.proxy", "root");
+    this.props.put("execute.as.user", "true");
+    this.props.put(ProcessJob.COMMAND, "ls -al");
+
+    this.job.run();
+
+  }
+
+  /**
+   * this job should fail because it sets user.to.proxy = azkaban which is black listed
+   */
+  @Test(expected = RuntimeException.class)
+  public void testOneUnixCommandWithAzkabanUser() throws Exception {
+
+    // Initialize the Props
+    this.props.removeLocal(CommonJobProperties.SUBMIT_USER);
+    this.props.put("user.to.proxy", "azkaban");
+    this.props.put("execute.as.user", "true");
+    this.props.put(ProcessJob.COMMAND, "ls -al");
+
+    this.job.run();
+
+  }
+
   @Test
   public void testFailedUnixCommand() throws Exception {
     // Initialize the Props