YARN-1063. Augmented Hadoop common winutils to have the ability to create containers as domain users. Contributed by Remus Rusanu.

Committed as a YARN patch even though all the code changes are in common.

hadoop-common-project/hadoop-common/src/main/winutils/chown.c

@@ -63,11 +63,11 @@ static DWORD ChangeFileOwnerBySid(__in LPCWSTR path,
   // SID is not contained in the caller's token, and have the SE_GROUP_OWNER
   // permission enabled.
   //
-  if (!EnablePrivilege(L"SeTakeOwnershipPrivilege"))
+  if (EnablePrivilege(L"SeTakeOwnershipPrivilege") != ERROR_SUCCESS)
   {
     fwprintf(stdout, L"INFO: The user does not have SeTakeOwnershipPrivilege.\n");
   }
-  if (!EnablePrivilege(L"SeRestorePrivilege"))
+  if (EnablePrivilege(L"SeRestorePrivilege") != ERROR_SUCCESS)
   {
     fwprintf(stdout, L"INFO: The user does not have SeRestorePrivilege.\n");
   }

hadoop-common-project/hadoop-common/src/main/winutils/include/winutils.h

@@ -27,6 +27,8 @@
 #include <accctrl.h>
 #include <strsafe.h>
 #include <lm.h>
+#include <ntsecapi.h>
+#include <userenv.h>
 
 enum EXIT_CODE
 {
@@ -153,6 +155,26 @@ DWORD ChangeFileModeByMask(__in LPCWSTR path, INT mode);
 DWORD GetLocalGroupsForUser(__in LPCWSTR user,
   __out LPLOCALGROUP_USERS_INFO_0 *groups, __out LPDWORD entries);
 
-BOOL EnablePrivilege(__in LPCWSTR privilegeName);
-
 void GetLibraryName(__in LPCVOID lpAddress, __out LPWSTR *filename);
+
+DWORD EnablePrivilege(__in LPCWSTR privilegeName);
+
+void AssignLsaString(__inout LSA_STRING * target, __in const char *strBuf);
+
+DWORD RegisterWithLsa(__in const char *logonProcessName, __out HANDLE * lsaHandle);
+
+void UnregisterWithLsa(__in HANDLE lsaHandle);
+
+DWORD LookupKerberosAuthenticationPackageId(__in HANDLE lsaHandle, __out ULONG * packageId);
+
+DWORD CreateLogonForUser(__in HANDLE lsaHandle,
+                         __in const char * tokenSourceName, 
+                         __in const char * tokenOriginName, 
+                         __in ULONG authnPkgId, 
+                         __in const wchar_t* principalName, 
+                         __out HANDLE *tokenHandle);
+
+DWORD LoadUserProfileForLogon(__in HANDLE logonHandle, __out PROFILEINFO * pi);
+
+DWORD UnloadProfileForLogon(__in HANDLE logonHandle, __in PROFILEINFO * pi);
+

hadoop-common-project/hadoop-common/src/main/winutils/libwinutils.c

@@ -17,6 +17,8 @@
 
 #pragma comment(lib, "authz.lib")
 #pragma comment(lib, "netapi32.lib")
+#pragma comment(lib, "Secur32.lib")
+#pragma comment(lib, "Userenv.lib")
 #include "winutils.h"
 #include <authz.h>
 #include <sddl.h>
@@ -797,7 +799,6 @@ DWORD FindFileOwnerAndPermission(
   __out_opt PINT pMask)
 {
   DWORD dwRtnCode = 0;
-
   PSECURITY_DESCRIPTOR pSd = NULL;
 
   PSID psidOwner = NULL;
@@ -1638,11 +1639,12 @@ GetLocalGroupsForUserEnd:
 //  to the process's access token.
 //
 // Returns:
-//	TRUE: on success
+//  ERROR_SUCCESS on success
+//  GetLastError() on error
 //
 // Notes:
 //
-BOOL EnablePrivilege(__in LPCWSTR privilegeName)
+DWORD EnablePrivilege(__in LPCWSTR privilegeName)
 {
   HANDLE hToken = INVALID_HANDLE_VALUE;
   TOKEN_PRIVILEGES tp = { 0 };
@@ -1651,28 +1653,31 @@ BOOL EnablePrivilege(__in LPCWSTR privilegeName)
   if (!OpenProcessToken(GetCurrentProcess(),
     TOKEN_ADJUST_PRIVILEGES | TOKEN_QUERY, &hToken))
   {
-    ReportErrorCode(L"OpenProcessToken", GetLastError());
+    dwErrCode = GetLastError();
-    return FALSE;
+    ReportErrorCode(L"OpenProcessToken", dwErrCode);
+    return dwErrCode;
   }
 
   tp.PrivilegeCount = 1;
   if (!LookupPrivilegeValueW(NULL,
     privilegeName, &(tp.Privileges[0].Luid)))
   {
-    ReportErrorCode(L"LookupPrivilegeValue", GetLastError());
+    dwErrCode = GetLastError();
+    ReportErrorCode(L"LookupPrivilegeValue", dwErrCode);
     CloseHandle(hToken);
-    return FALSE;
+    return dwErrCode;
   }
   tp.Privileges[0].Attributes = SE_PRIVILEGE_ENABLED;
 
   // As stated on MSDN, we need to use GetLastError() to check if
   // AdjustTokenPrivileges() adjusted all of the specified privileges.
   //
-  AdjustTokenPrivileges(hToken, FALSE, &tp, 0, NULL, NULL);
+  if( !AdjustTokenPrivileges(hToken, FALSE, &tp, 0, NULL, NULL) ) {
   dwErrCode = GetLastError();
+  }
   CloseHandle(hToken);
 
-  return dwErrCode == ERROR_SUCCESS;
+  return dwErrCode;
 }
 
 //----------------------------------------------------------------------------
@@ -1716,9 +1721,6 @@ void ReportErrorCode(LPCWSTR func, DWORD err)
 // Description:
 //  Given an address, get the file name of the library from which it was loaded.
 //
-// Returns:
-//  None
-//
 // Notes:
 // - The function allocates heap memory and points the filename out parameter to
 //   the newly allocated memory, which will contain the name of the file.
@@ -1757,3 +1759,290 @@ cleanup:
     *filename = NULL;
   }
 }
+
+// Function: AssignLsaString
+//
+// Description:
+//  fills in values of LSA_STRING struct to point to a string buffer
+//
+// Returns:
+//  None
+//
+//  IMPORTANT*** strBuf is not copied. It must be globally immutable
+//
+void AssignLsaString(__inout LSA_STRING * target, __in const char *strBuf)
+{
+  target->Length = (USHORT)(sizeof(char)*strlen(strBuf));
+  target->MaximumLength = target->Length;
+  target->Buffer = (char *)(strBuf);
+}
+
+//----------------------------------------------------------------------------
+// Function: RegisterWithLsa
+//
+// Description:
+//  Registers with local security authority and sets handle for use in later LSA
+//  operations
+//
+// Returns:
+//  ERROR_SUCCESS on success
+//  Other error code on failure
+//
+// Notes:
+//
+DWORD RegisterWithLsa(__in const char *logonProcessName, __out HANDLE * lsaHandle) 
+{
+  LSA_STRING processName; 
+  LSA_OPERATIONAL_MODE o_mode; // never useful as per msdn docs
+  NTSTATUS registerStatus;
+  *lsaHandle = 0;
+  
+  AssignLsaString(&processName, logonProcessName);
+  registerStatus = LsaRegisterLogonProcess(&processName, lsaHandle, &o_mode); 
+  
+  return LsaNtStatusToWinError( registerStatus );
+}
+
+//----------------------------------------------------------------------------
+// Function: UnregisterWithLsa
+//
+// Description:
+//  Closes LSA handle allocated by RegisterWithLsa()
+//
+// Returns:
+//  None
+//
+// Notes:
+//
+void UnregisterWithLsa(__in HANDLE lsaHandle)
+{
+  LsaClose(lsaHandle);
+}
+
+//----------------------------------------------------------------------------
+// Function: LookupKerberosAuthenticationPackageId
+//
+// Description:
+//  Looks of the current id (integer index) of the Kerberos authentication package on the local
+//  machine.
+//
+// Returns:
+//  ERROR_SUCCESS on success
+//  Other error code on failure
+//
+// Notes:
+//
+DWORD LookupKerberosAuthenticationPackageId(__in HANDLE lsaHandle, __out ULONG * packageId)
+{
+  NTSTATUS lookupStatus; 
+  LSA_STRING pkgName;
+
+  AssignLsaString(&pkgName, MICROSOFT_KERBEROS_NAME_A);
+  lookupStatus = LsaLookupAuthenticationPackage(lsaHandle, &pkgName, packageId);
+  return LsaNtStatusToWinError( lookupStatus );
+}
+  
+//----------------------------------------------------------------------------
+// Function: CreateLogonForUser
+//
+// Description:
+//  Contacts the local LSA and performs a logon without credential for the 
+//  given principal. This logon token will be local machine only and have no 
+//  network credentials attached.
+//
+// Returns:
+//  ERROR_SUCCESS on success
+//  Other error code on failure
+//
+// Notes:
+//  This call assumes that all required privileges have already been enabled (TCB etc).
+//  IMPORTANT ****  tokenOriginName must be immutable!
+//
+DWORD CreateLogonForUser(__in HANDLE lsaHandle,
+                         __in const char * tokenSourceName, 
+                         __in const char * tokenOriginName, // must be immutable, will not be copied!
+                         __in ULONG authnPkgId, 
+                         __in const wchar_t* principalName, 
+                         __out HANDLE *tokenHandle) 
+{ 
+  DWORD logonStatus = ERROR_ASSERTION_FAILURE; // Failure to set status should trigger error
+  TOKEN_SOURCE tokenSource;
+  LSA_STRING originName;
+  void * profile = NULL;
+
+  // from MSDN:
+  // The ClientUpn and ClientRealm members of the KERB_S4U_LOGON 
+  // structure must point to buffers in memory that are contiguous 
+  // to the structure itself. The value of the 
+  // AuthenticationInformationLength parameter must take into 
+  // account the length of these buffers.
+  const int principalNameBufLen = lstrlen(principalName)*sizeof(*principalName);
+  const int totalAuthInfoLen = sizeof(KERB_S4U_LOGON) + principalNameBufLen;
+  KERB_S4U_LOGON* s4uLogonAuthInfo = (KERB_S4U_LOGON*)calloc(totalAuthInfoLen, 1);
+  if (s4uLogonAuthInfo == NULL ) {
+    logonStatus = ERROR_NOT_ENOUGH_MEMORY;
+    goto done;
+  }
+  s4uLogonAuthInfo->MessageType = KerbS4ULogon;
+  s4uLogonAuthInfo->ClientUpn.Buffer = (wchar_t*)((char*)s4uLogonAuthInfo + sizeof *s4uLogonAuthInfo);
+  CopyMemory(s4uLogonAuthInfo->ClientUpn.Buffer, principalName, principalNameBufLen);
+  s4uLogonAuthInfo->ClientUpn.Length        = (USHORT)principalNameBufLen;
+  s4uLogonAuthInfo->ClientUpn.MaximumLength = (USHORT)principalNameBufLen;
+
+  AllocateLocallyUniqueId(&tokenSource.SourceIdentifier);
+  StringCchCopyA(tokenSource.SourceName, TOKEN_SOURCE_LENGTH, tokenSourceName );
+  AssignLsaString(&originName, tokenOriginName);
+
+  {
+    DWORD cbProfile = 0;
+    LUID logonId;
+    QUOTA_LIMITS quotaLimits;
+    NTSTATUS subStatus;
+
+    NTSTATUS logonNtStatus = LsaLogonUser(lsaHandle,
+      &originName,
+      Batch, // SECURITY_LOGON_TYPE
+      authnPkgId,
+      s4uLogonAuthInfo, 
+      totalAuthInfoLen,
+      0,
+      &tokenSource,
+      &profile, 
+      &cbProfile,
+      &logonId, 
+      tokenHandle,
+      &quotaLimits,
+      &subStatus);
+    logonStatus = LsaNtStatusToWinError( logonNtStatus );
+  }
+done:
+  // clean up
+  if (s4uLogonAuthInfo != NULL) {
+    free(s4uLogonAuthInfo);
+  }
+  if (profile != NULL) {
+    LsaFreeReturnBuffer(profile);
+  }
+  return logonStatus;
+}
+
+// NOTE: must free allocatedName
+DWORD GetNameFromLogonToken(__in HANDLE logonToken, __out wchar_t **allocatedName)
+{
+  DWORD userInfoSize = 0;
+  PTOKEN_USER  user = NULL;
+  DWORD userNameSize = 0;
+  wchar_t * userName = NULL;
+  DWORD domainNameSize = 0; 
+  wchar_t * domainName = NULL;
+  SID_NAME_USE sidUse = SidTypeUnknown;
+  DWORD getNameStatus = ERROR_ASSERTION_FAILURE; // Failure to set status should trigger error
+  BOOL tokenInformation = FALSE;
+
+  // call for sid size then alloc and call for sid
+  tokenInformation = GetTokenInformation(logonToken, TokenUser, NULL, 0, &userInfoSize);
+  assert (FALSE == tokenInformation);
+  
+  // last call should have failed and filled in allocation size
+  if ((getNameStatus = GetLastError()) != ERROR_INSUFFICIENT_BUFFER)
+  {
+    goto done; 
+  }
+  user = (PTOKEN_USER)calloc(userInfoSize,1);
+  if (user == NULL)
+  {
+    getNameStatus = ERROR_NOT_ENOUGH_MEMORY;
+    goto done;
+  }
+  if (!GetTokenInformation(logonToken, TokenUser, user, userInfoSize, &userInfoSize)) {
+      getNameStatus = GetLastError();
+      goto done;
+  }
+  LookupAccountSid( NULL, user->User.Sid, NULL, &userNameSize, NULL, &domainNameSize, &sidUse );
+  // last call should have failed and filled in allocation size
+  if ((getNameStatus = GetLastError()) != ERROR_INSUFFICIENT_BUFFER)
+  {
+    goto done;
+  }
+  userName = (wchar_t *)calloc(userNameSize, sizeof(wchar_t));
+  if (userName == NULL) {
+    getNameStatus = ERROR_NOT_ENOUGH_MEMORY;
+    goto done;
+  }
+  domainName = (wchar_t *)calloc(domainNameSize, sizeof(wchar_t));
+  if (domainName == NULL) {
+    getNameStatus = ERROR_NOT_ENOUGH_MEMORY;
+    goto done;
+  }
+  if (!LookupAccountSid( NULL, user->User.Sid, userName, &userNameSize, domainName, &domainNameSize, &sidUse )) {
+      getNameStatus = GetLastError();
+      goto done;
+  }
+
+  getNameStatus = ERROR_SUCCESS;
+  *allocatedName = userName;
+  userName = NULL;
+done:
+  if (user != NULL) {
+    free( user );
+    user = NULL;
+  }
+  if (userName != NULL) {
+    free( userName );
+    userName = NULL;
+  }
+  if (domainName != NULL) {
+    free( domainName );
+    domainName = NULL;
+  }
+  return getNameStatus;
+}
+
+DWORD LoadUserProfileForLogon(__in HANDLE logonHandle, __out PROFILEINFO * pi)
+{
+  wchar_t *userName = NULL;
+  DWORD loadProfileStatus = ERROR_ASSERTION_FAILURE; // Failure to set status should trigger error
+
+  loadProfileStatus = GetNameFromLogonToken( logonHandle, &userName );
+  if (loadProfileStatus != ERROR_SUCCESS) {
+    goto done;
+  }
+
+  assert(pi);
+
+  ZeroMemory( pi, sizeof(*pi) );
+  pi->dwSize = sizeof(*pi); 
+  pi->lpUserName = userName;
+  pi->dwFlags = PI_NOUI;
+
+  // if the profile does not exist it will be created
+  if ( !LoadUserProfile( logonHandle, pi ) ) {      
+    loadProfileStatus = GetLastError();
+    goto done;
+  }
+
+  loadProfileStatus = ERROR_SUCCESS;
+done:
+  return loadProfileStatus;
+}
+
+DWORD UnloadProfileForLogon(__in HANDLE logonHandle, __in PROFILEINFO * pi)
+{
+  DWORD touchProfileStatus = ERROR_ASSERTION_FAILURE; // Failure to set status should trigger error
+
+  assert(pi);
+
+  if ( !UnloadUserProfile(logonHandle, pi->hProfile ) ) {
+    touchProfileStatus = GetLastError();
+    goto done;
+  }
+  if (pi->lpUserName != NULL) {
+    free(pi->lpUserName);
+    pi->lpUserName = NULL;
+  }
+  ZeroMemory( pi, sizeof(*pi) );
+
+  touchProfileStatus = ERROR_SUCCESS;
+done:
+  return touchProfileStatus;
+}

hadoop-common-project/hadoop-common/src/main/winutils/symlink.c

@@ -77,7 +77,7 @@ int Symlink(__in int argc, __in_ecount(argc) wchar_t *argv[])
   // This is just an additional step to do the privilege check by not using
   // error code from CreateSymbolicLink() method.
   //
-  if (!EnablePrivilege(L"SeCreateSymbolicLinkPrivilege"))
+  if (EnablePrivilege(L"SeCreateSymbolicLinkPrivilege") != ERROR_SUCCESS)
   {
     fwprintf(stderr,
       L"No privilege to create symbolic links.\n");

hadoop-common-project/hadoop-common/src/main/winutils/task.c

@@ -18,6 +18,7 @@
 #include "winutils.h"
 #include <errno.h>
 #include <psapi.h>
+#include <malloc.h>
 
 #define PSAPI_VERSION 1
 #pragma comment(lib, "psapi.lib")
@@ -28,12 +29,18 @@
 // process exits with 128 + signal.  For SIGKILL, this would be 128 + 9 = 137.
 #define KILLED_PROCESS_EXIT_CODE 137
 
+// Name for tracking this logon process when registering with LSA
+static const char *LOGON_PROCESS_NAME="Hadoop Container Executor";
+// Name for token source, must be less or eq to TOKEN_SOURCE_LENGTH (currently 8) chars
+static const char *TOKEN_SOURCE_NAME = "HadoopEx";
+
 // List of different task related command line options supported by
 // winutils.
 typedef enum TaskCommandOptionType
 {
   TaskInvalid,
   TaskCreate,
+  TaskCreateAsUser,
   TaskIsAlive,
   TaskKill,
   TaskProcessList
@@ -86,37 +93,53 @@ static BOOL ParseCommandLine(__in int argc,
     }
   }
 
+  if (argc >= 6) {
+    if (wcscmp(argv[1], L"createAsUser") == 0)
+    {
+      *command = TaskCreateAsUser;
+      return TRUE;
+    }
+  }
+
   return FALSE;
 }
 
 //----------------------------------------------------------------------------
-// Function: createTask
+// Function: CreateTaskImpl
 //
 // Description:
 //  Creates a task via a jobobject. Outputs the
 //  appropriate information to stdout on success, or stderr on failure.
+//  logonHandle may be NULL, in this case the current logon will be utilized for the 
+//  created process
 //
 // Returns:
 // ERROR_SUCCESS: On success
 // GetLastError: otherwise
-DWORD createTask(__in PCWSTR jobObjName,__in PWSTR cmdLine) 
+DWORD CreateTaskImpl(__in_opt HANDLE logonHandle, __in PCWSTR jobObjName,__in PWSTR cmdLine) 
 {
-  DWORD err = ERROR_SUCCESS;
+  DWORD dwErrorCode = ERROR_SUCCESS;
   DWORD exitCode = EXIT_FAILURE;
+  DWORD currDirCnt = 0;
   STARTUPINFO si;
   PROCESS_INFORMATION pi;
   HANDLE jobObject = NULL;
   JOBOBJECT_EXTENDED_LIMIT_INFORMATION jeli = { 0 };
+  void * envBlock = NULL;
+  BOOL createProcessResult = FALSE;
+
+  wchar_t* curr_dir = NULL;
+  FILE *stream = NULL;
 
   // Create un-inheritable job object handle and set job object to terminate 
   // when last handle is closed. So winutils.exe invocation has the only open 
   // job object handle. Exit of winutils.exe ensures termination of job object.
   // Either a clean exit of winutils or crash or external termination.
   jobObject = CreateJobObject(NULL, jobObjName);
-  err = GetLastError();
+  dwErrorCode = GetLastError();
-  if(jobObject == NULL || err ==  ERROR_ALREADY_EXISTS)
+  if(jobObject == NULL || dwErrorCode ==  ERROR_ALREADY_EXISTS)
   {
-    return err;
+    return dwErrorCode;
   }
   jeli.BasicLimitInformation.LimitFlags = JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE;
   if(SetInformationJobObject(jobObject, 
@@ -124,36 +147,102 @@ DWORD createTask(__in PCWSTR jobObjName,__in PWSTR cmdLine)
                              &jeli, 
                              sizeof(jeli)) == 0)
   {
-    err = GetLastError();
+    dwErrorCode = GetLastError();
     CloseHandle(jobObject);
-    return err;
+    return dwErrorCode;
   }      
 
   if(AssignProcessToJobObject(jobObject, GetCurrentProcess()) == 0)
   {
-    err = GetLastError();
+    dwErrorCode = GetLastError();
     CloseHandle(jobObject);
-    return err;
+    return dwErrorCode;
   }
 
   // the child JVM uses this env var to send the task OS process identifier 
   // to the TaskTracker. We pass the job object name.
   if(SetEnvironmentVariable(L"JVM_PID", jobObjName) == 0)
   {
-    err = GetLastError();
+    dwErrorCode = GetLastError();
-    CloseHandle(jobObject);
+    // We have to explictly Terminate, passing in the error code
-    return err;
+    // simply closing the job would kill our own process with success exit status
+    TerminateJobObject(jobObject, dwErrorCode);
+    return dwErrorCode;
   }
 
   ZeroMemory( &si, sizeof(si) );
   si.cb = sizeof(si);
   ZeroMemory( &pi, sizeof(pi) );
 
-  if (CreateProcess(NULL, cmdLine, NULL, NULL, TRUE, 0, NULL, NULL, &si, &pi) == 0)
+  if( logonHandle != NULL ) {
-  {
+    // create user environment for this logon
-    err = GetLastError();
+    if(!CreateEnvironmentBlock(&envBlock,
-    CloseHandle(jobObject);
+                              logonHandle,
-    return err;
+                              TRUE )) {
+        dwErrorCode = GetLastError();
+        // We have to explictly Terminate, passing in the error code
+        // simply closing the job would kill our own process with success exit status
+        TerminateJobObject(jobObject, dwErrorCode);
+        return dwErrorCode; 
+    }
+  }
+
+  // Get the required buffer size first
+  currDirCnt = GetCurrentDirectory(0, NULL);
+  if (0 < currDirCnt) {
+    curr_dir = (wchar_t*) alloca(currDirCnt * sizeof(wchar_t));
+    assert(curr_dir);
+    currDirCnt = GetCurrentDirectory(currDirCnt, curr_dir);
+  }
+  
+  if (0 == currDirCnt) {
+     dwErrorCode = GetLastError();
+     // We have to explictly Terminate, passing in the error code
+     // simply closing the job would kill our own process with success exit status
+     TerminateJobObject(jobObject, dwErrorCode);
+     return dwErrorCode;     
+  }
+
+  if (logonHandle == NULL) {
+    createProcessResult = CreateProcess(
+                  NULL,         // ApplicationName
+                  cmdLine,      // command line
+                  NULL,         // process security attributes
+                  NULL,         // thread security attributes
+                  TRUE,         // inherit handles
+                  0,            // creation flags
+                  NULL,         // environment
+                  curr_dir,     // current directory
+                  &si,          // startup info
+                  &pi);         // process info
+  }
+  else {
+    createProcessResult = CreateProcessAsUser(
+                  logonHandle,  // logon token handle
+                  NULL,         // Application handle
+                  cmdLine,      // command line
+                  NULL,         // process security attributes
+                  NULL,         // thread security attributes
+                  FALSE,        // inherit handles
+                  CREATE_UNICODE_ENVIRONMENT, // creation flags
+                  envBlock,     // environment
+                  curr_dir,     // current directory
+                  &si,          // startup info
+                  &pi);         // process info
+  }
+  
+  if (FALSE == createProcessResult) {
+    dwErrorCode = GetLastError();
+    if( envBlock != NULL ) {
+      DestroyEnvironmentBlock( envBlock );
+      envBlock = NULL;
+    }
+    // We have to explictly Terminate, passing in the error code
+    // simply closing the job would kill our own process with success exit status
+    TerminateJobObject(jobObject, dwErrorCode);
+
+    // This is tehnically dead code, we cannot reach this condition
+    return dwErrorCode;
   }
 
   CloseHandle(pi.hThread);
@@ -162,10 +251,15 @@ DWORD createTask(__in PCWSTR jobObjName,__in PWSTR cmdLine)
   WaitForSingleObject( pi.hProcess, INFINITE );
   if(GetExitCodeProcess(pi.hProcess, &exitCode) == 0)
   {
-    err = GetLastError();
+    dwErrorCode = GetLastError();
   }
   CloseHandle( pi.hProcess );
 
+  if( envBlock != NULL ) {
+    DestroyEnvironmentBlock( envBlock );
+    envBlock = NULL;
+  }
+
   // Terminate job object so that all spawned processes are also killed.
   // This is needed because once this process closes the handle to the job 
   // object and none of the spawned objects have the handle open (via 
@@ -173,21 +267,134 @@ DWORD createTask(__in PCWSTR jobObjName,__in PWSTR cmdLine)
   // program (say winutils task kill) to terminate this job object via its name.
   if(TerminateJobObject(jobObject, exitCode) == 0)
   {
-    err = GetLastError();
+    dwErrorCode = GetLastError();
   }
 
-  // comes here only on failure or TerminateJobObject
+  // comes here only on failure of TerminateJobObject
   CloseHandle(jobObject);
 
-  if(err != ERROR_SUCCESS)
+  if(dwErrorCode != ERROR_SUCCESS)
   {
-    return err;
+    return dwErrorCode;
   }
   return exitCode;
 }
 
 //----------------------------------------------------------------------------
-// Function: isTaskAlive
+// Function: CreateTask
+//
+// Description:
+//  Creates a task via a jobobject. Outputs the
+//  appropriate information to stdout on success, or stderr on failure.
+//
+// Returns:
+// ERROR_SUCCESS: On success
+// GetLastError: otherwise
+DWORD CreateTask(__in PCWSTR jobObjName,__in PWSTR cmdLine) 
+{
+  // call with null logon in order to create tasks utilizing the current logon
+  return CreateTaskImpl( NULL, jobObjName, cmdLine );
+}
+//----------------------------------------------------------------------------
+// Function: CreateTask
+//
+// Description:
+//  Creates a task via a jobobject. Outputs the
+//  appropriate information to stdout on success, or stderr on failure.
+//
+// Returns:
+// ERROR_SUCCESS: On success
+// GetLastError: otherwise
+DWORD CreateTaskAsUser(__in PCWSTR jobObjName,__in PWSTR user, __in PWSTR pidFilePath, __in PWSTR cmdLine)
+{
+  DWORD err = ERROR_SUCCESS;
+  DWORD exitCode = EXIT_FAILURE;
+  ULONG authnPkgId;
+  HANDLE lsaHandle = INVALID_HANDLE_VALUE;
+  PROFILEINFO  pi;
+  BOOL profileIsLoaded = FALSE;
+  FILE* pidFile = NULL;
+
+  DWORD retLen = 0;
+  HANDLE logonHandle = NULL;
+
+  err = EnablePrivilege(SE_TCB_NAME);
+  if( err != ERROR_SUCCESS ) {
+    fwprintf(stdout, L"INFO: The user does not have SE_TCB_NAME.\n");
+    goto done;
+  }
+  err = EnablePrivilege(SE_ASSIGNPRIMARYTOKEN_NAME);
+  if( err != ERROR_SUCCESS ) {
+    fwprintf(stdout, L"INFO: The user does not have SE_ASSIGNPRIMARYTOKEN_NAME.\n");
+    goto done;
+  }
+  err = EnablePrivilege(SE_INCREASE_QUOTA_NAME);
+  if( err != ERROR_SUCCESS ) {
+    fwprintf(stdout, L"INFO: The user does not have SE_INCREASE_QUOTA_NAME.\n");
+    goto done;
+  }
+  err = EnablePrivilege(SE_RESTORE_NAME);
+  if( err != ERROR_SUCCESS ) {
+    fwprintf(stdout, L"INFO: The user does not have SE_RESTORE_NAME.\n");
+    goto done;
+  }
+
+  err = RegisterWithLsa(LOGON_PROCESS_NAME ,&lsaHandle);
+  if( err != ERROR_SUCCESS ) goto done;
+
+  err = LookupKerberosAuthenticationPackageId( lsaHandle, &authnPkgId );
+  if( err != ERROR_SUCCESS ) goto done;
+
+  err =  CreateLogonForUser(lsaHandle,
+    LOGON_PROCESS_NAME, 
+    TOKEN_SOURCE_NAME,
+    authnPkgId, 
+    user, 
+    &logonHandle); 
+  if( err != ERROR_SUCCESS ) goto done;
+
+  err = LoadUserProfileForLogon(logonHandle, &pi);
+  if( err != ERROR_SUCCESS ) goto done;
+  profileIsLoaded = TRUE; 
+
+  // Create the PID file
+
+  if (!(pidFile = _wfopen(pidFilePath, "w"))) {
+      err = GetLastError();
+      goto done;
+  }
+
+  if (0 > fprintf_s(pidFile, "%ls", jobObjName)) {
+    err = GetLastError();
+  }
+  
+  fclose(pidFile);
+    
+  if (err != ERROR_SUCCESS) {
+    goto done;
+  }
+  
+  err = CreateTaskImpl(logonHandle, jobObjName, cmdLine);
+
+done:	
+  if( profileIsLoaded ) {
+    UnloadProfileForLogon( logonHandle, &pi );
+    profileIsLoaded = FALSE;
+  }
+  if( logonHandle != NULL ) {
+    CloseHandle(logonHandle);
+  }
+
+  if (INVALID_HANDLE_VALUE != lsaHandle) {
+    UnregisterWithLsa(lsaHandle);
+  }
+
+  return err;
+}
+
+
+//----------------------------------------------------------------------------
+// Function: IsTaskAlive
 //
 // Description:
 //  Checks if a task is alive via a jobobject. Outputs the
@@ -196,7 +403,7 @@ DWORD createTask(__in PCWSTR jobObjName,__in PWSTR cmdLine)
 // Returns:
 // ERROR_SUCCESS: On success
 // GetLastError: otherwise
-DWORD isTaskAlive(const WCHAR* jobObjName, int* isAlive, int* procsInJob)
+DWORD IsTaskAlive(const WCHAR* jobObjName, int* isAlive, int* procsInJob)
 {
   PJOBOBJECT_BASIC_PROCESS_ID_LIST procList;
   HANDLE jobObject = NULL;
@@ -247,7 +454,7 @@ DWORD isTaskAlive(const WCHAR* jobObjName, int* isAlive, int* procsInJob)
 }
 
 //----------------------------------------------------------------------------
-// Function: killTask
+// Function: KillTask
 //
 // Description:
 //  Kills a task via a jobobject. Outputs the
@@ -256,7 +463,7 @@ DWORD isTaskAlive(const WCHAR* jobObjName, int* isAlive, int* procsInJob)
 // Returns:
 // ERROR_SUCCESS: On success
 // GetLastError: otherwise
-DWORD killTask(PCWSTR jobObjName)
+DWORD KillTask(PCWSTR jobObjName)
 {
   HANDLE jobObject = OpenJobObject(JOB_OBJECT_TERMINATE, FALSE, jobObjName);
   if(jobObject == NULL)
@@ -280,7 +487,7 @@ DWORD killTask(PCWSTR jobObjName)
 }
 
 //----------------------------------------------------------------------------
-// Function: printTaskProcessList
+// Function: PrintTaskProcessList
 //
 // Description:
 // Prints resource usage of all processes in the task jobobject
@@ -288,7 +495,7 @@ DWORD killTask(PCWSTR jobObjName)
 // Returns:
 // ERROR_SUCCESS: On success
 // GetLastError: otherwise
-DWORD printTaskProcessList(const WCHAR* jobObjName)
+DWORD PrintTaskProcessList(const WCHAR* jobObjName)
 {
   DWORD i;
   PJOBOBJECT_BASIC_PROCESS_ID_LIST procList;
@@ -372,6 +579,21 @@ int Task(__in int argc, __in_ecount(argc) wchar_t *argv[])
 {
   DWORD dwErrorCode = ERROR_SUCCESS;
   TaskCommandOption command = TaskInvalid;
+  wchar_t* cmdLine = NULL;
+  wchar_t buffer[16*1024] = L""; // 32K max command line
+  size_t charCountBufferLeft = sizeof
(buffer)/sizeof(wchar_t);
+  int crtArgIndex = 0;
+  size_t argLen = 0;
+  size_t wscatErr = 0;
+  wchar_t* insertHere = NULL;
+
+  enum {
+               ARGC_JOBOBJECTNAME = 2,
+               ARGC_USERNAME,
+               ARGC_PIDFILE,
+               ARGC_COMMAND,
+               ARGC_COMMAND_ARGS
+       };
 
   if (!ParseCommandLine(argc, argv, &command)) {
     dwErrorCode = ERROR_INVALID_COMMAND_LINE;
@@ -385,10 +607,57 @@ int Task(__in int argc, __in_ecount(argc) wchar_t *argv[])
   {
     // Create the task jobobject
     //
-    dwErrorCode = createTask(argv[2], argv[3]);
+    dwErrorCode = CreateTask(argv[2], argv[3]);
+    if (dwErrorCode != ERROR_SUCCESS)
+    {
+      ReportErrorCode(L"CreateTask", dwErrorCode);
+      goto TaskExit;
+    }
+  } else if (command == TaskCreateAsUser)
+  {
+    // Create the task jobobject as a domain user
+    // createAsUser accepts an open list of arguments. All arguments after the command are
+    // to be passed as argumrnts to the command itself.Here we're concatenating all 
+    // arguments after the command into a single arg entry.
+    //
+    cmdLine = argv[ARGC_COMMAND];
+    if (argc > ARGC_COMMAND_ARGS) {
+        crtArgIndex = ARGC_COMMAND;
+        insertHere = buffer;
+        while (crtArgIndex < argc) {
+            argLen = wcslen(argv[crtArgIndex]);
+            wscatErr = wcscat_s(insertHere, charCountBufferLeft, argv[crtArgIndex]);
+            switch (wscatErr) {
+            case 0:
+              // 0 means success;
+              break;
+            case EINVAL:
+              dwErrorCode = ERROR_INVALID_PARAMETER;
+              goto TaskExit;
+            case ERANGE:
+              dwErrorCode = ERROR_INSUFFICIENT_BUFFER;
+              goto TaskExit;
+            default:
+              // This case is not MSDN documented.
+              dwErrorCode = ERROR_GEN_FAILURE;
+              goto TaskExit;
+            }
+            insertHere += argLen;
+            charCountBufferLeft -= argLen;
+            insertHere[0] = L' ';
+            insertHere += 1;
+            charCountBufferLeft -= 1;
+            insertHere[0] = 0;
+            ++crtArgIndex;
+        }
+        cmdLine = buffer;
+    }
+
+    dwErrorCode = CreateTaskAsUser(
+      argv[ARGC_JOBOBJECTNAME], argv[ARGC_USERNAME], argv[ARGC_PIDFILE], cmdLine);
     if (dwErrorCode != ERROR_SUCCESS)
     {
-      ReportErrorCode(L"createTask", dwErrorCode);
+      ReportErrorCode(L"CreateTaskAsUser", dwErrorCode);
       goto TaskExit;
     }
   } else if (command == TaskIsAlive)
@@ -397,10 +666,10 @@ int Task(__in int argc, __in_ecount(argc) wchar_t *argv[])
     //
     int isAlive;
     int numProcs;
-    dwErrorCode = isTaskAlive(argv[2], &isAlive, &numProcs);
+    dwErrorCode = IsTaskAlive(argv[2], &isAlive, &numProcs);
     if (dwErrorCode != ERROR_SUCCESS)
     {
-      ReportErrorCode(L"isTaskAlive", dwErrorCode);
+      ReportErrorCode(L"IsTaskAlive", dwErrorCode);
       goto TaskExit;
     }
 
@@ -412,27 +681,27 @@ int Task(__in int argc, __in_ecount(argc) wchar_t *argv[])
     else
     {
       dwErrorCode = ERROR_TASK_NOT_ALIVE;
-      ReportErrorCode(L"isTaskAlive returned false", dwErrorCode);
+      ReportErrorCode(L"IsTaskAlive returned false", dwErrorCode);
       goto TaskExit;
     }
   } else if (command == TaskKill)
   {
     // Check if task jobobject
     //
-    dwErrorCode = killTask(argv[2]);
+    dwErrorCode = KillTask(argv[2]);
     if (dwErrorCode != ERROR_SUCCESS)
     {
-      ReportErrorCode(L"killTask", dwErrorCode);
+      ReportErrorCode(L"KillTask", dwErrorCode);
       goto TaskExit;
     }
   } else if (command == TaskProcessList)
   {
     // Check if task jobobject
     //
-    dwErrorCode = printTaskProcessList(argv[2]);
+    dwErrorCode = PrintTaskProcessList(argv[2]);
     if (dwErrorCode != ERROR_SUCCESS)
     {
-      ReportErrorCode(L"printTaskProcessList", dwErrorCode);
+      ReportErrorCode(L"PrintTaskProcessList", dwErrorCode);
       goto TaskExit;
     }
   } else
@@ -453,10 +722,12 @@ void TaskUsage()
   // ProcessTree.isSetsidSupported()
   fwprintf(stdout, L"\
     Usage: task create [TASKNAME] [COMMAND_LINE] |\n\
+          task createAsUser [TASKNAME] [USERNAME] [PIDFILE] [COMMAND_LINE] |\n\
           task isAlive [TASKNAME] |\n\
           task kill [TASKNAME]\n\
           task processList [TASKNAME]\n\
     Creates a new task jobobject with taskname\n\
+    Creates a new task jobobject with taskname as the user provided\n\
     Checks if task jobobject is alive\n\
     Kills task jobobject\n\
     Prints to stdout a list of processes in the task\n\

hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/util/TestWinUtils.java

@@ -20,10 +20,12 @@ package org.apache.hadoop.util;
 
 import static org.junit.Assert.*;
 import static org.junit.Assume.assumeTrue;
+import static org.junit.matchers.JUnitMatchers.containsString;
 
 import java.io.File;
 import java.io.FileInputStream;
 import java.io.FileOutputStream;
+import java.io.FileWriter;
 import java.io.IOException;
 
 import org.apache.commons.io.FileUtils;
@@ -33,7 +35,7 @@ import org.apache.hadoop.fs.FileUtil;
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
-import static org.junit.Assume.*;
+
 import static org.hamcrest.CoreMatchers.*;
 
 /**
@@ -521,4 +523,26 @@ public class TestWinUtils {
       assertThat(ece.getExitCode(), is(1));
     }
   }
+  
+  @SuppressWarnings("deprecation")
+  @Test(timeout=10000)
+  public void testTaskCreate() throws IOException {
+    File batch = new File(TEST_DIR, "testTaskCreate.cmd");
+    File proof = new File(TEST_DIR, "testTaskCreate.out");
+    FileWriter fw = new FileWriter(batch);
+    String testNumber = String.format("%f", Math.random());
+    fw.write(String.format("echo %s > \"%s\"", testNumber, proof.getAbsolutePath()));
+    fw.close();
+    
+    assertFalse(proof.exists());
+    
+    Shell.execCommand(Shell.WINUTILS, "task", "create", "testTaskCreate" + testNumber, 
+        batch.getAbsolutePath());
+    
+    assertTrue(proof.exists());
+    
+    String outNumber = FileUtils.readFileToString(proof);
+    
+    assertThat(outNumber, containsString(testNumber));
+  }
 }

hadoop-yarn-project/CHANGES.txt

@@ -115,6 +115,9 @@ Release 2.6.0 - UNRELEASED
     YARN-2581. Passed LogAggregationContext to NM via ContainerTokenIdentifier.
     (Xuan Gong via zjshen)
 
+    YARN-1063. Augmented Hadoop common winutils to have the ability to create
+    containers as domain users. (Remus Rusanu via vinodkv)
+
   IMPROVEMENTS
 
     YARN-2197. Add a link to YARN CHANGES.txt in the left side of doc