azkaban-aplcache

diff --git a/azkaban-common/src/main/c/execute-as-user.c b/azkaban-common/src/main/c/execute-as-user.c
index f27d5a1..daaa66b 100644
--- a/azkaban-common/src/main/c/execute-as-user.c
+++ b/azkaban-common/src/main/c/execute-as-user.c
@@ -39,18 +39,23 @@ int INVALID_INPUT = 30;
 
 /*
  *  Change the real and effective user and group from super user to the specified user
- *  
+ *
  *  Adopted from:
  *  ./hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/native/container-executor/impl/container-executor.c
- *  
+ *
  */
 
-int change_user(uid_t user, gid_t group) {
+int change_user(char *username, uid_t user, gid_t group) {
     if (user == getuid() && user == geteuid() &&
             group == getgid() && group == getegid()) {
         return 0;
     }
 
+    if (initgroups(username, group) != 0) {
+        fprintf(LOGFILE, "Error setting supplementary groups for user %s: %s\n",
+            username, strerror(errno));
+        return SETUID_OPER_FAILED;
+    }
     if (seteuid(0) != 0) {
         fprintf(LOGFILE, "unable to reacquire root - %s\n", strerror(errno));
         fprintf(LOGFILE, "Real: %d:%d; Effective: %d:%d\n",
@@ -85,24 +90,24 @@ int main(int argc, char **argv){
     }
 
     if (argc < 3) {
-        fprintf(ERRORFILE, "Requires at least 3 variables: ./execute-as-user uid command [args]");
+        fprintf(ERRORFILE, "Requires at least 3 variables: ./execute-as-user username command [args]");
         return INVALID_INPUT;
     }
 
-    char *uid = argv[1];
+    char *username = argv[1];
 
     // gather information about user
-    struct passwd *user_info = getpwnam(uid);
+    struct passwd *user_info = getpwnam(username);
     if (user_info == NULL){
-        fprintf(LOGFILE, "user does not exist: %s", uid);
+        fprintf(LOGFILE, "user does not exist: %s", username);
         return USER_NOT_FOUND;
     }
 
     // try to change user
-    fprintf(LOGFILE, "Changing user: user: %s, uid: %d, gid: %d\n", uid, user_info->pw_uid, user_info->pw_gid);
-    int retval = change_user(user_info->pw_uid, user_info->pw_gid);
+    fprintf(LOGFILE, "Changing user: user: %s, uid: %d, gid: %d\n", username, user_info->pw_uid, user_info->pw_gid);
+    int retval = change_user(username, user_info->pw_uid, user_info->pw_gid);
     if (retval != 0){
-        fprintf(LOGFILE, "Error changing user to %s\n", uid);
+        fprintf(LOGFILE, "Error changing user to %s\n", username);
         return SETUID_OPER_FAILED;
     }
 

diff --git a/azkaban-common/src/main/java/azkaban/Constants.java b/azkaban-common/src/main/java/azkaban/Constants.java
index 54234cd..7aba594 100644
--- a/azkaban-common/src/main/java/azkaban/Constants.java
+++ b/azkaban-common/src/main/java/azkaban/Constants.java
@@ -87,6 +87,12 @@ public class Constants {
     // List of users we prevent azkaban from running flows as. (ie: root, azkaban)
     public static final String BLACK_LISTED_USERS = "azkaban.server.blacklist.users";
 
+    // Path name of execute-as-user executable
+    public static final String AZKABAN_SERVER_NATIVE_LIB_FOLDER = "azkaban.native.lib";
+
+    // Name of *nix group associated with the process running Azkaban
+    public static final String AZKABAN_SERVER_GROUP_NAME = "azkaban.group.name";
+
     // 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 feed8c1..369c3c8 100644
--- a/azkaban-common/src/main/java/azkaban/jobExecutor/ProcessJob.java
+++ b/azkaban-common/src/main/java/azkaban/jobExecutor/ProcessJob.java
@@ -16,6 +16,8 @@
 
 package azkaban.jobExecutor;
 
+import static azkaban.Constants.ConfigurationKeys.AZKABAN_SERVER_GROUP_NAME;
+import static azkaban.Constants.ConfigurationKeys.AZKABAN_SERVER_NATIVE_LIB_FOLDER;
 import static azkaban.ServiceProvider.SERVICE_PROVIDER;
 
 import azkaban.Constants;
@@ -23,11 +25,13 @@ import azkaban.flow.CommonJobProperties;
 import azkaban.jobExecutor.utils.process.AzkabanProcess;
 import azkaban.jobExecutor.utils.process.AzkabanProcessBuilder;
 import azkaban.metrics.CommonMetrics;
+import azkaban.utils.ExecuteAsUser;
 import azkaban.utils.Pair;
 import azkaban.utils.Props;
 import azkaban.utils.SystemMemoryInfo;
 import com.google.common.annotations.VisibleForTesting;
 import java.io.File;
+import java.io.IOException;
 import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Arrays;
@@ -46,12 +50,18 @@ public class ProcessJob extends AbstractProcessJob {
 
   public static final String COMMAND = "command";
   public static final String AZKABAN_MEMORY_CHECK = "azkaban.memory.check";
+  // Use azkaban.Constants.ConfigurationKeys.AZKABAN_SERVER_NATIVE_LIB_FOLDER instead
+  @Deprecated
   public static final String NATIVE_LIB_FOLDER = "azkaban.native.lib";
   public static final String EXECUTE_AS_USER = "execute.as.user";
   public static final String USER_TO_PROXY = "user.to.proxy";
   public static final String KRB5CCNAME = "KRB5CCNAME";
   private static final Duration KILL_TIME = Duration.ofSeconds(30);
   private static final String MEMCHECK_ENABLED = "memCheck.enabled";
+  private static final String CHOWN = "chown";
+  private static final String CREATE_FILE = "touch";
+  private static final int SUCCESSFUL_EXECUTION = 0;
+  private static final String TEMP_FILE_NAME = "user_can_write";
   private final CommonMetrics commonMetrics;
   private volatile AzkabanProcess process;
   private volatile boolean killed = false;
@@ -221,7 +231,7 @@ public class ProcessJob extends AbstractProcessJob {
     // 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);
+      final String nativeLibFolder = this.sysProps.getString(AZKABAN_SERVER_NATIVE_LIB_FOLDER);
       executeAsUserBinaryPath = String.format("%s/%s", nativeLibFolder, "execute-as-user");
       effectiveUser = getEffectiveUser(this.jobProps);
       // Throw exception if Azkaban tries to run flow as a prohibited user
@@ -230,6 +240,11 @@ public class ProcessJob extends AbstractProcessJob {
             String.format("Not permitted to proxy as '%s' through Azkaban", effectiveUser)
         );
       }
+      // Set parent directory permissions to <uid>:azkaban so user can write in their execution directory
+      // if the directory is not permissioned correctly already (should happen once per execution)
+      if (!canWriteInCurrentWorkingDirectory(effectiveUser)) {
+        assignUserDirOwnership(effectiveUser);
+      }
     }
 
     for (String command : commands) {
@@ -271,8 +286,8 @@ public class ProcessJob extends AbstractProcessJob {
         this.process = builder.build();
       }
       try {
-          this.process.run();
-          this.success = true;
+        this.process.run();
+        this.success = true;
       } catch (final Throwable e) {
         for (final File file : propFiles) {
           if (file != null && file.exists()) {
@@ -345,6 +360,47 @@ public class ProcessJob extends AbstractProcessJob {
   }
 
   /**
+   * Checks to see if user has write access to current working directory which many users
+   * need for their jobs to store temporary data/jars on the executor.
+   *
+   * Accomplishes this by using execute-as-user to try to create an empty file in the cwd.
+   *
+   * @param effectiveUser user/proxy user running the job
+   * @return true if user has write permissions in current working directory otherwise false
+   */
+  private boolean canWriteInCurrentWorkingDirectory(final String effectiveUser)
+      throws IOException {
+    final ExecuteAsUser executeAsUser = new ExecuteAsUser(
+        this.sysProps.getString(AZKABAN_SERVER_NATIVE_LIB_FOLDER));
+    final List<String> checkIfUserCanWriteCommand = Arrays
+        .asList(CREATE_FILE, getWorkingDirectory() + "/" + TEMP_FILE_NAME);
+    final int result = executeAsUser.execute(effectiveUser, checkIfUserCanWriteCommand);
+    return result == SUCCESSFUL_EXECUTION;
+  }
+
+  /**
+   * Changes permission on current working directory so that the directory is owned by the user
+   * and the group remains azkaban.
+   *
+   * Leverages execute-as-user with "root" as the user to run the command.
+   *
+   * @param effectiveUser user/proxy user running the job
+   */
+  private void assignUserDirOwnership(final String effectiveUser) throws IOException {
+    final ExecuteAsUser executeAsUser = new ExecuteAsUser(
+        this.sysProps.getString(AZKABAN_SERVER_NATIVE_LIB_FOLDER));
+    final String groupName = this.sysProps.getString(AZKABAN_SERVER_GROUP_NAME, "azkaban");
+    final List<String> changeOwnershipCommand = Arrays
+        .asList(CHOWN, effectiveUser + ":" + groupName, getWorkingDirectory());
+    info("Change current working directory ownership to " + effectiveUser + ":" + groupName + ".");
+    final int result = executeAsUser.execute("root", changeOwnershipCommand);
+    if (result != 0) {
+      error("Failed to change current working directory ownership. Error code: " + Integer
+          .toString(result));
+    }
+  }
+
+  /**
    * This is used to get the min/max memory size requirement by processes.
    * SystemMemoryInfo can use the info to determine if the memory request can be
    * fulfilled. For Java process, this should be Xms/Xmx setting.

diff --git a/azkaban-hadoop-security-plugin/src/main/java/azkaban/security/HadoopSecurityManager_H_2_0.java b/azkaban-hadoop-security-plugin/src/main/java/azkaban/security/HadoopSecurityManager_H_2_0.java
index 7bfc055..b5d9cda 100644
--- a/azkaban-hadoop-security-plugin/src/main/java/azkaban/security/HadoopSecurityManager_H_2_0.java
+++ b/azkaban-hadoop-security-plugin/src/main/java/azkaban/security/HadoopSecurityManager_H_2_0.java
@@ -16,8 +16,11 @@
 
 package azkaban.security;
 
+import static azkaban.Constants.ConfigurationKeys.AZKABAN_SERVER_NATIVE_LIB_FOLDER;
+
 import azkaban.security.commons.HadoopSecurityManager;
 import azkaban.security.commons.HadoopSecurityManagerException;
+import azkaban.utils.ExecuteAsUser;
 import azkaban.utils.Props;
 import azkaban.utils.UndefinedPropertyException;
 import java.io.DataOutputStream;
@@ -66,15 +69,9 @@ import org.apache.thrift.TException;
 
 public class HadoopSecurityManager_H_2_0 extends HadoopSecurityManager {
 
-  /**
-   * TODO Remove duplicated constants from plugins.
-   *
-   * Azkaban plugins don't depend on a common submodule from which they both can inherit code. Thus,
-   * constants are copied around and any changes to the constant values will break Azkaban. This
-   * needs to be fixed as part of a plugin infrastructure implementation.
-   */
+  // Use azkaban.Constants.ConfigurationKeys.AZKABAN_SERVER_NATIVE_LIB_FOLDER instead
+  @Deprecated
   public static final String NATIVE_LIB_FOLDER = "azkaban.native.lib";
-
   /**
    * TODO: This should be exposed as a configurable parameter
    *
@@ -137,7 +134,7 @@ public class HadoopSecurityManager_H_2_0 extends HadoopSecurityManager {
 
   private HadoopSecurityManager_H_2_0(final Props props)
       throws HadoopSecurityManagerException, IOException {
-    this.executeAsUser = new ExecuteAsUser(props.getString(NATIVE_LIB_FOLDER));
+    this.executeAsUser = new ExecuteAsUser(props.getString(AZKABAN_SERVER_NATIVE_LIB_FOLDER));
 
     // for now, assume the same/compatible native library, the same/compatible
     // hadoop-core jar