标签：encode   nis   with open   system权限   des   方式   inf   ups   hook   

这2年一直在做一个运维项目,如果你要在windwos上实现不同进程的权限切换，那么这篇文章会对你有用的!

最重要的函数CreateProcessWithLogonW

在UNIX平台上如果你要对一个进程降权有跟多办法,不管是su命令 还是python的os函数都可以很容易的达到目的。
但是在windwos上要降权就不是很容易了。基本原则就是高级权限切低级权限不需要密码，低级权限使用高级权限需要密码。

依赖模块代码如下 文件名:win_startup.py

# -*- coding: utf-8 -*-
'''
    特定于windwos平台 根据以下内容做适度调整
    https://stackoverflow.com/a/43233332
    http://stackoverflow.com/questions/29566330
'''
from __future__ import absolute_import, unicode_literals
import os
import collections
import psutil

import ctypes
from ctypes import wintypes


import win32con
import win32api
import win32process
import win32security
import win32service
import ntsecuritycon

from logger import aglogger
log = aglogger.logger

ntdll = ctypes.WinDLL('ntdll')
secur32 = ctypes.WinDLL('secur32')
kernel32 = ctypes.WinDLL('kernel32', use_last_error=True)
advapi32 = ctypes.WinDLL('advapi32', use_last_error=True)
userenv = ctypes.WinDLL('userenv', use_last_error=True)

SYSTEM_SID = "S-1-5-18"
LOCAL_SRV_SID = "S-1-5-19"
NETWORK_SRV_SID = "S-1-5-19"

LOGON_WITH_PROFILE = 0x00000001

WINSTA_ALL = (
    win32con.WINSTA_ACCESSCLIPBOARD |
    win32con.WINSTA_ACCESSGLOBALATOMS |
    win32con.WINSTA_CREATEDESKTOP |
    win32con.WINSTA_ENUMDESKTOPS |
    win32con.WINSTA_ENUMERATE |
    win32con.WINSTA_EXITWINDOWS |
    win32con.WINSTA_READATTRIBUTES |
    win32con.WINSTA_READSCREEN |
    win32con.WINSTA_WRITEATTRIBUTES |
    win32con.DELETE |
    win32con.READ_CONTROL |
    win32con.WRITE_DAC |
    win32con.WRITE_OWNER
)

DESKTOP_ALL = (
    win32con.DESKTOP_CREATEMENU |
    win32con.DESKTOP_CREATEWINDOW |
    win32con.DESKTOP_ENUMERATE |
    win32con.DESKTOP_HOOKCONTROL |
    win32con.DESKTOP_JOURNALPLAYBACK |
    win32con.DESKTOP_JOURNALRECORD |
    win32con.DESKTOP_READOBJECTS |
    win32con.DESKTOP_SWITCHDESKTOP |
    win32con.DESKTOP_WRITEOBJECTS |
    win32con.DELETE |
    win32con.READ_CONTROL |
    win32con.WRITE_DAC |
    win32con.WRITE_OWNER
)

MAX_COMPUTER_NAME_LENGTH = 15

SECURITY_LOGON_TYPE = wintypes.ULONG
Interactive = 2
Network = 3
Batch = 4
Service = 5

LOGON_SUBMIT_TYPE = wintypes.ULONG
PROFILE_BUFFER_TYPE = wintypes.ULONG

MsV1_0InteractiveLogon = 2
MsV1_0Lm20Logon = 3
MsV1_0NetworkLogon = 4
MsV1_0WorkstationUnlockLogon = 7
MsV1_0S4ULogon = 12
MsV1_0NoElevationLogon = 82

KerbInteractiveLogon = 2
KerbWorkstationUnlockLogon = 7
KerbS4ULogon = 12

MSV1_0_S4U_LOGON_FLAG_CHECK_LOGONHOURS = 0x2

KERB_S4U_LOGON_FLAG_CHECK_LOGONHOURS = 0x2
KERB_S4U_LOGON_FLAG_IDENTITY = 0x8

TOKEN_SOURCE_LENGTH = 8

NEGOTIATE_PACKAGE_NAME = b'Negotiate'
MICROSOFT_KERBEROS_NAME = b'Kerberos'
MSV1_0_PACKAGE_NAME = b'MICROSOFT_AUTHENTICATION_PACKAGE_V1_0'

DELETE = 0x00010000
READ_CONTROL = 0x00020000
WRITE_DAC = 0x00040000
WRITE_OWNER = 0x00080000

STANDARD_RIGHTS_REQUIRED = (
    DELETE |
    READ_CONTROL |
    WRITE_DAC |
    WRITE_OWNER
)

TOKEN_ASSIGN_PRIMARY = 0x0001
TOKEN_DUPLICATE = 0x0002
TOKEN_IMPERSONATE = 0x0004
TOKEN_QUERY = 0x0008
TOKEN_QUERY_SOURCE = 0x0010
TOKEN_ADJUST_PRIVILEGES = 0x0020
TOKEN_ADJUST_GROUPS = 0x0040
TOKEN_ADJUST_DEFAULT = 0x0080
TOKEN_ADJUST_SESSIONID = 0x0100

TOKEN_ALL_ACCESS = (
    STANDARD_RIGHTS_REQUIRED |
    TOKEN_ASSIGN_PRIMARY |
    TOKEN_DUPLICATE |
    TOKEN_IMPERSONATE |
    TOKEN_QUERY |
    TOKEN_QUERY_SOURCE |
    TOKEN_ADJUST_PRIVILEGES |
    TOKEN_ADJUST_GROUPS |
    TOKEN_ADJUST_DEFAULT |
    TOKEN_ADJUST_SESSIONID
)

DUPLICATE_CLOSE_SOURCE = 0x00000001
DUPLICATE_SAME_ACCESS = 0x00000002

TOKEN_TYPE = wintypes.ULONG
TokenPrimary = 1
TokenImpersonation = 2

SECURITY_IMPERSONATION_LEVEL = wintypes.ULONG
SecurityAnonymous = 0
SecurityIdentification = 1
SecurityImpersonation = 2
SecurityDelegation = 3


class NTSTATUS(wintypes.LONG):

    def to_error(self):
        return ntdll.RtlNtStatusToDosError(self)

    def __repr__(self):
        name = self.__class__.__name__
        status = wintypes.ULONG.from_buffer(self)
        return '{}({})'.format(name, status.value)


PNTSTATUS = ctypes.POINTER(NTSTATUS)


class BOOL(wintypes.BOOL):
    def __repr__(self):
        name = self.__class__.__name__
        return '{}({})'.format(name, bool(self))


class HANDLE(wintypes.HANDLE):
    __slots__ = 'closed',

    def __int__(self):
        return self.value or 0

    def Detach(self):
        if not getattr(self, 'closed', False):
            self.closed = True
            value = int(self)
            self.value = None
            return value
        raise ValueError("already closed")

    def Close(self, CloseHandle=kernel32.CloseHandle):
        if self and not getattr(self, 'closed', False):
            CloseHandle(self.Detach())

    __del__ = Close

    def __repr__(self):
        return "{}({})".format(self.__class__.__name__, int(self))


class LARGE_INTEGER(wintypes.LARGE_INTEGER):
    # https://msdn.microsoft.com/en-us/library/ff553204
    ntdll.RtlSecondsSince1970ToTime.restype = None
    _unix_epoch = wintypes.LARGE_INTEGER()
    ntdll.RtlSecondsSince1970ToTime(0, ctypes.byref(_unix_epoch))
    _unix_epoch = _unix_epoch.value

    def __int__(self):
        return self.value

    def __repr__(self):
        name = self.__class__.__name__
        return '{}({})'.format(name, self.value)

    def as_time(self):
        time100ns = self.value - self._unix_epoch
        if time100ns >= 0:
            return time100ns / 1e7
        raise ValueError('value predates the Unix epoch')

    @classmethod
    def from_time(cls, t):
        time100ns = int(t * 10**7)
        return cls(time100ns + cls._unix_epoch)


CHAR = ctypes.c_char
WCHAR = ctypes.c_wchar
PCHAR = ctypes.POINTER(CHAR)
PWCHAR = ctypes.POINTER(WCHAR)


class STRING(ctypes.Structure):
    _fields_ = (
        ('Length', wintypes.USHORT),
        ('MaximumLength', wintypes.USHORT),
        ('Buffer', PCHAR),
    )


LPSTRING = ctypes.POINTER(STRING)


class UNICODE_STRING(ctypes.Structure):
    _fields_ = (
        ('Length', wintypes.USHORT),
        ('MaximumLength', wintypes.USHORT),
        ('Buffer', PWCHAR),
    )


LPUNICODE_STRING = ctypes.POINTER(UNICODE_STRING)


class LUID(ctypes.Structure):
    _fields_ = (
        ('LowPart', wintypes.DWORD),
        ('HighPart', wintypes.LONG),
    )

    def __new__(cls, value=0):
        return cls.from_buffer_copy(ctypes.c_ulonglong(value))

    def __int__(self):
        return ctypes.c_ulonglong.from_buffer(self).value

    def __repr__(self):
        name = self.__class__.__name__
        return '{}({})'.format(name, int(self))


LPLUID = ctypes.POINTER(LUID)
PSID = wintypes.LPVOID


class SID_AND_ATTRIBUTES(ctypes.Structure):
    _fields_ = (
        ('Sid', PSID),
        ('Attributes', wintypes.DWORD),
    )


LPSID_AND_ATTRIBUTES = ctypes.POINTER(SID_AND_ATTRIBUTES)


class TOKEN_GROUPS(ctypes.Structure):
    _fields_ = (
        ('GroupCount', wintypes.DWORD),
        ('Groups', SID_AND_ATTRIBUTES * 1),
    )


LPTOKEN_GROUPS = ctypes.POINTER(TOKEN_GROUPS)


class TOKEN_SOURCE(ctypes.Structure):
    _fields_ = (
        ('SourceName', CHAR * TOKEN_SOURCE_LENGTH),
        ('SourceIdentifier', LUID),
    )

    def __init__(self, SourceName=None, SourceIdentifier=None):
        super(TOKEN_SOURCE, self).__init__()
        if SourceName is not None:
            if not isinstance(SourceName, bytes):
                SourceName = SourceName.encode('mbcs')
            self.SourceName = SourceName
        if SourceIdentifier is None:
            luid = self.SourceIdentifier  # pylint: disable=access-member-before-definition
            ntdll.NtAllocateLocallyUniqueId(ctypes.byref(luid))
        else:
            self.SourceIdentifier = SourceIdentifier


LPTOKEN_SOURCE = ctypes.POINTER(TOKEN_SOURCE)
py_source_context = TOKEN_SOURCE(b"PYTHON  ")
py_origin_name = __name__.encode()
py_logon_process_name = "{}-{}".format(py_origin_name, os.getpid())
SIZE_T = ctypes.c_size_t


class QUOTA_LIMITS(ctypes.Structure):
    _fields_ = (('PagedPoolLimit', SIZE_T),
                ('NonPagedPoolLimit', SIZE_T),
                ('MinimumWorkingSetSize', SIZE_T),
                ('MaximumWorkingSetSize', SIZE_T),
                ('PagefileLimit', SIZE_T),
                ('TimeLimit', wintypes.LARGE_INTEGER))


LPQUOTA_LIMITS = ctypes.POINTER(QUOTA_LIMITS)
LPULONG = ctypes.POINTER(wintypes.ULONG)
LSA_OPERATIONAL_MODE = wintypes.ULONG
LPLSA_OPERATIONAL_MODE = LPULONG
LPHANDLE = ctypes.POINTER(wintypes.HANDLE)
LPLPVOID = ctypes.POINTER(wintypes.LPVOID)
LPDWORD = ctypes.POINTER(wintypes.DWORD)


class ContiguousUnicode(ctypes.Structure):
    # _string_names_: sequence matched to underscore-prefixed fields
    def __init__(self, *args, **kwargs):
        super(ContiguousUnicode, self).__init__(*args, **kwargs)

    def _get_unicode_string(self, name):
        wchar_size = ctypes.sizeof(WCHAR)
        s = getattr(self, '_{}'.format(name))
        length = s.Length // wchar_size
        buf = s.Buffer
        if buf:
            return buf[:length]
        return None

    def _set_unicode_buffer(self, value):
        cls = type(self)
        wchar_size = ctypes.sizeof(WCHAR)
        bufsize = (len(value) + 1) * wchar_size
        ctypes.resize(self, ctypes.sizeof(cls) + bufsize)
        addr = ctypes.addressof(self) + ctypes.sizeof(cls)
        ctypes.memmove(addr, value, bufsize)

    def _set_unicode_string(self, name, value):
        values = []
        for n in self._string_names_:
            if n == name:
                values.append(value or '')
            else:
                values.append(getattr(self, n) or '')
        self._set_unicode_buffer('\x00'.join(values))

        cls = type(self)
        wchar_size = ctypes.sizeof(WCHAR)
        addr = ctypes.addressof(self) + ctypes.sizeof(cls)
        for n, v in zip(self._string_names_, values):
            ptr = ctypes.cast(addr, PWCHAR)
            ustr = getattr(self, '_{}'.format(n))
            length = ustr.Length = len(v) * wchar_size
            full_length = length + wchar_size
            if ((n == name and value is None) or
                (n != name and not (length or ustr.Buffer))):
                ustr.Buffer = None
                ustr.MaximumLength = 0
            else:
                ustr.Buffer = ptr
                ustr.MaximumLength = full_length
            addr += full_length

    def __getattr__(self, name):
        if name not in self._string_names_:
            raise AttributeError
        return self._get_unicode_string(name)

    def __setattr__(self, name, value):
        if name in self._string_names_:
            self._set_unicode_string(name, value)
        else:
            super(ContiguousUnicode, self).__setattr__(name, value)

    @classmethod
    def from_address_copy(cls, address, size=None):
        x = ctypes.Structure.__new__(cls)
        if size is not None:
            ctypes.resize(x, size)
        ctypes.memmove(ctypes.byref(x), address, ctypes.sizeof(x))
        delta = ctypes.addressof(x) - address
        for n in cls._string_names_:
            ustr = getattr(x, '_{}'.format(n))
            addr = ctypes.c_void_p.from_buffer(ustr.Buffer)
            if addr:
                addr.value += delta
        return x


class AuthInfo(ContiguousUnicode):
    # _message_type_: from a logon-submit-type enumeration
    def __init__(self):
        super(AuthInfo, self).__init__()
        self.MessageType = self._message_type_


class MSV1_0_INTERACTIVE_LOGON(AuthInfo):
    _message_type_ = MsV1_0InteractiveLogon
    _string_names_ = 'LogonDomainName', 'UserName', 'Password'

    _fields_ = (('MessageType', LOGON_SUBMIT_TYPE),
                ('_LogonDomainName', UNICODE_STRING),
                ('_UserName', UNICODE_STRING),
                ('_Password', UNICODE_STRING))

    def __init__(self, UserName=None, Password=None, LogonDomainName=None):
        super(MSV1_0_INTERACTIVE_LOGON, self).__init__()
        if LogonDomainName is not None:
            self.LogonDomainName = LogonDomainName
        if UserName is not None:
            self.UserName = UserName
        if Password is not None:
            self.Password = Password


class S4ULogon(AuthInfo):
    _string_names_ = 'UserPrincipalName', 'DomainName'

    _fields_ = (('MessageType', LOGON_SUBMIT_TYPE),
                ('Flags', wintypes.ULONG),
                ('_UserPrincipalName', UNICODE_STRING),
                ('_DomainName', UNICODE_STRING))

    def __init__(self, UserPrincipalName=None, DomainName=None, Flags=0):
        super(S4ULogon, self).__init__()
        self.Flags = Flags
        if UserPrincipalName is not None:
            self.UserPrincipalName = UserPrincipalName
        if DomainName is not None:
            self.DomainName = DomainName


class MSV1_0_S4U_LOGON(S4ULogon):
    _message_type_ = MsV1_0S4ULogon


class KERB_S4U_LOGON(S4ULogon):
    _message_type_ = KerbS4ULogon


PMSV1_0_S4U_LOGON = ctypes.POINTER(MSV1_0_S4U_LOGON)
PKERB_S4U_LOGON = ctypes.POINTER(KERB_S4U_LOGON)


class ProfileBuffer(ContiguousUnicode):
    # _message_type_
    def __init__(self):
        super(ProfileBuffer, self).__init__()
        self.MessageType = self._message_type_


class MSV1_0_INTERACTIVE_PROFILE(ProfileBuffer):
    _message_type_ = MsV1_0InteractiveLogon
    _string_names_ = ('LogonScript', 'HomeDirectory', 'FullName',
                      'ProfilePath', 'HomeDirectoryDrive', 'LogonServer')
    _fields_ = (('MessageType', PROFILE_BUFFER_TYPE),
                ('LogonCount', wintypes.USHORT),
                ('BadPasswordCount', wintypes.USHORT),
                ('LogonTime', LARGE_INTEGER),
                ('LogoffTime', LARGE_INTEGER),
                ('KickOffTime', LARGE_INTEGER),
                ('PasswordLastSet', LARGE_INTEGER),
                ('PasswordCanChange', LARGE_INTEGER),
                ('PasswordMustChange', LARGE_INTEGER),
                ('_LogonScript', UNICODE_STRING),
                ('_HomeDirectory', UNICODE_STRING),
                ('_FullName', UNICODE_STRING),
                ('_ProfilePath', UNICODE_STRING),
                ('_HomeDirectoryDrive', UNICODE_STRING),
                ('_LogonServer', UNICODE_STRING),
                ('UserFlags', wintypes.ULONG))


def _check_status(result, func, args):
    if result.value 

执行bat脚本主函数如下

# -*- coding: utf-8 -*-
'''
此module适用于在system或者win上以普通用户的权限执行某个命令  或者以普通用户的权限切换到其他用户执行命令(需要密码)
'''
import sys
reload(sys)
sys.setdefaultencoding('utf8')

import ctypes
import os
import logging
import psutil
try:
    import psutil
    HAS_PSUTIL = True
except ImportError:
    HAS_PSUTIL = False

try:
    import win32api
    import win32con
    import win32process
    import win32security
    import win32pipe
    import win32event
    import win32profile
    import msvcrt
    HAS_WIN32 = True
except ImportError:
    HAS_WIN32 = False

from logger import aglogger
import win_startup
log = aglogger.logger

class CommandExecutionError(Exception):
    pass

def __virtual__():

    if not HAS_WIN32 or not HAS_PSUTIL:
        return False, 'This utility requires pywin32 and psutil'
    return 'win_runas'

def proc_pid(file_name,pid = 0):
    try:
        tmp_read = dict()
        if os.path.isfile(file_name):
            with open(file_name, "r+") as fd:
                tmp_read = fd.read()
            tmp_writd = eval(tmp_read)
            if isinstance(tmp_writd, dict) and pid:
                if isinstance(tmp_writd.get("pid"), list):
                    tmp_writd["pid"].append(str(pid))
                else:
                    tmp_writd["pid"] = [str(pid)]
                with open(file_name, "w") as fd:
                    fd.write(str(tmp_writd))
        else:
            log.error("file not find {}".format(file_name))
    except Exception as e:
        log.error(e)


def split_username(username):
    # TODO: Is there a windows api for this?
    domain = '.'
    if '@' in username:
        username, domain = username.split('@')
    if '\\' in username:
        domain, username = username.split('\\')
    return username, domain


def runas(cmdLine, username, password=None, cwd = 'C:/xxx' ):
    '''
    用其他用户的权限执行一个命令。如果这个进程 运行的权限是admin或者system权限不需要密码。
    其他非特权用户需要密码。命令以提供账户的级别权限运行。
    '''

    # 从当前进程获取令牌
    access = (
        win32security.TOKEN_QUERY |
        win32security.TOKEN_ADJUST_PRIVILEGES
    )
    th = win32security.OpenProcessToken(win32api.GetCurrentProcess(), access)
    win_startup.elevate_token(th)

    #尝试模拟system用户  此进程用户需要被授予SeImpersonatePrivilege权限 Administrator用户默认有这个权限
    try:
        impersonation_token = win_startup.impersonate_sid(
            win_startup.SYSTEM_SID,
            session_id=0,
            privs=['SeTcbPrivilege'],
        )
    except WindowsError:  # pylint: disable=undefined-variable
        log.debug("Unable to impersonate SYSTEM user")
        impersonation_token = None

    # 模拟system用户失败 回退到非特权用户
    if not impersonation_token:
        log.debug("No impersonation token, using unprivileged runas")
        return runas_unpriv(cmdLine, username, password, cwd)

    username, domain = split_username(username)
    #验证这个用户的域和sid是否账户(UAC验证)
    try:
        _, domain, _ = win32security.LookupAccountName(domain, username)
    except Exception as exc:
        log.error("exc {}".format(exc))
        raise CommandExecutionError(exc)

    if domain == 'NT AUTHORITY':
        #作为系统级账户登录  SYSTEM, LOCAL SERVICE, or NETWORK SERVICE.
        logonType = win32con.LOGON32_LOGON_SERVICE
        user_token = win32security.LogonUser(
            username,
            domain,
            '',
            win32con.LOGON32_LOGON_SERVICE,
            win32con.LOGON32_PROVIDER_DEFAULT,
        )
    elif password:
        # 使用密码登录
        user_token = win32security.LogonUser(
            username,
            domain,
            password,
            win32con.LOGON32_LOGON_INTERACTIVE,
            win32con.LOGON32_PROVIDER_DEFAULT,
        )
    else:
        #不使用密码登录  总是返回最高级别的token
        user_token = win_startup.logon_msv1_s4u(username).Token

    # 如果需要获取要提升的用户令牌
    elevation_type = win32security.GetTokenInformation(
        user_token, win32security.TokenElevationType
    )
    if elevation_type > 1:
        user_token = win32security.GetTokenInformation(
            user_token,
            win32security.TokenLinkedToken
        )

    # 提升用户令牌
    win_startup.elevate_token(user_token)

    #确保用户令牌有访问桌面权限
    win_startup.grant_winsta_and_desktop(user_token)

    # 创建输入输出管道 包括正确输出和错误输出
    security_attributes = win32security.SECURITY_ATTRIBUTES()
    security_attributes.bInheritHandle = 1

    stdin_read, stdin_write = win32pipe.CreatePipe(security_attributes, 0)
    stdin_read = win_startup.make_inheritable(stdin_read)

    stdout_read, stdout_write = win32pipe.CreatePipe(security_attributes, 0)
    stdout_write = win_startup.make_inheritable(stdout_write)

    stderr_read, stderr_write = win32pipe.CreatePipe(security_attributes, 0)
    stderr_write = win_startup.make_inheritable(stderr_write)

    #运行进程但是不显示gui界面
    creationflags = (
        win32process.CREATE_NO_WINDOW |
        win32process.CREATE_NEW_CONSOLE |
        win32process.CREATE_SUSPENDED
    )

    # startup_info = win_startup.STARTUPINFO(
    #     dwFlags=win32con.STARTF_USESTDHANDLES,
    #     hStdInput=stdin_read.handle,
    #     hStdOutput=stdout_write.handle,
    #     hStdError=stderr_write.handle,
    # )
    startup_info = win_startup.STARTUPINFO(
        dwFlags=win32con.STARTF_USESTDHANDLES,
        hStdInput=stdin_read.handle,
        hStdOutput=stdout_write.handle,
        hStdError=stdout_write.handle,
    )

    # 得到对应用户的环境变量信息
    env = win32profile.CreateEnvironmentBlock(user_token, False)

    #创建一个挂起状态的进程
    process_info = win_startup.CreateProcessWithTokenW(
        int(user_token),
        logonflags=1,
        applicationname=None,
        commandline=cmdLine,
        currentdirectory=cwd,
        creationflags=creationflags,
        startupinfo=startup_info,
        environment=env,
    )

    hProcess = process_info.hProcess
    hThread = process_info.hThread
    dwProcessId = process_info.dwProcessId
    dwThreadId = process_info.dwThreadId

    win_startup.kernel32.CloseHandle(stdin_write.handle)
    win_startup.kernel32.CloseHandle(stdout_write.handle)
    win_startup.kernel32.CloseHandle(stderr_write.handle)

    ret = {'pid': dwProcessId}
    # 让这个进程运行
    log.info('winsubproc_account pid {}'.format(dwProcessId))
    psutil.Process(dwProcessId).resume()

    # 等待进程退出 并且返回退出码
    if win32event.WaitForSingleObject(hProcess, win32event.INFINITE) == win32con.WAIT_OBJECT_0:
        exitcode = win32process.GetExitCodeProcess(hProcess)
        ret['retcode'] = exitcode

    # 读取标准输出
    fd_out = msvcrt.open_osfhandle(stdout_read.handle, os.O_RDONLY | os.O_TEXT)
    # with os.fdopen(fd_out, 'r') as f_out:
        # stdout = f_out.read()
    # fo = os.fdopen(fd_out, "r")
    # os.lseek(fo, 0, 0)
    #将标准输出和错误重定向到同一个管道
    ret['stdout'] = os.read(fd_out, 1000 * 1024 * 1024)

    # 读取错误输出
    # fd_err = msvcrt.open_osfhandle(stderr_read.handle, os.O_RDONLY | os.O_TEXT)

    # fo = os.fdopen(fd_err, "w+")
    # fo.write("Python is a great language.")
    # ret['stderr'] = os.read(fd_err, 10 * 1024 * 1024)
    ret['stderr'] = ''

    win_startup.kernel32.CloseHandle(hProcess)
    win32api.CloseHandle(user_token)
    if impersonation_token:
        win32security.RevertToSelf()
    win32api.CloseHandle(impersonation_token)
    return ret


def runas_unpriv(cmd, username, password, cwd=None):
    '''
    在非特权的方式工作
    '''
    # 创建子进程和父进程之间的可被继承的输入输出管道
    c2pread, c2pwrite = win_startup.CreatePipe(
        inherit_read=False, inherit_write=True,
    )
    errread, errwrite = win_startup.CreatePipe(
        inherit_read=False, inherit_write=True,
    )

    # 创建输入通道
    stdin = win_startup.kernel32.GetStdHandle(
            win_startup.STD_INPUT_HANDLE,
    )
    dupin = win_startup.DuplicateHandle(srchandle=stdin, inherit=True)

    # 获取要运行cmd的结构信息
    # startup_info = win_startup.STARTUPINFO(
    #     dwFlags=win32con.STARTF_USESTDHANDLES,
    #     hStdInput=dupin,
    #     hStdOutput=c2pwrite,
    #     hStdError=errwrite,
    # )
    startup_info = win_startup.STARTUPINFO(
        dwFlags=win32con.STARTF_USESTDHANDLES,
        hStdInput=dupin,
        hStdOutput=c2pwrite,
        hStdError=c2pwrite,
    )

    username, domain = split_username(username)
    # 返回运行进程的信息 后面用到
    process_info = win_startup.CreateProcessWithLogonW(
            username=username,
            domain=domain,
            password=password,
            logonflags=win_startup.LOGON_WITH_PROFILE,
            commandline=cmd,
            startupinfo=startup_info,
            currentdirectory=cwd)

    win_startup.kernel32.CloseHandle(dupin)
    win_startup.kernel32.CloseHandle(c2pwrite)
    win_startup.kernel32.CloseHandle(errwrite)
    win_startup.kernel32.CloseHandle(process_info.hThread)
    #保存输出信息结构体
    ret = {'pid': process_info.dwProcessId}
    # 正常输出
    fd_out = msvcrt.open_osfhandle(c2pread, os.O_RDONLY | os.O_TEXT)
    # with os.fdopen(fd_out, 'r') as f_out:
    #     ret['stdout'] = f_out.read()
    ret['stdout'] = os.read(fd_out, 1000 * 1024 * 1024)
    # 错误输出
    # fd_err = msvcrt.open_osfhandle(errread, os.O_RDONLY | os.O_TEXT)
    # with os.fdopen(fd_err, 'r') as f_err:
    #     ret['stderr'] = f_err.read()
    ret['stderr'] = ""
    # 返回退出码
    if win_startup.kernel32.WaitForSingleObject(process_info.hProcess, win32event.INFINITE) ==             win32con.WAIT_OBJECT_0:
        exitcode = win_startup.wintypes.DWORD()
        win_startup.kernel32.GetExitCodeProcess(process_info.hProcess,ctypes.byref(exitcode))
        ret['retcode'] = exitcode.value
    # 关闭进程句柄对象
    win_startup.kernel32.CloseHandle(process_info.hProcess)
    return ret

if __name__ == "__main__":
    import tempfile,time
    outfile = './a.txt'

    cmd = 'C:/bb.bat 1'
    comspec = os.environ.get("COMSPEC", "cmd.exe")
    cmd = '{} /c "{}"'.format(comspec, cmd)
    f = open(outfile,'wb+')
    out_fileno = f.fileno()
    try:
        # print cmd
        res = runas(cmd,username,passwd)
        print res
        print res['stdout'].decode('gbk')
        # print type(res['stdout'])
        # print res['pid']
        # print psutil.Process(26672)
    except Exception as e:
        print e
#     stdout = f_out.read()
    # while True:
    #     x = fd.read()
    #     if x:
    #         print x
    #     else:
    #         break

说明

返回值是一个字典，从脚本进程拿输出我把正确输出和错误输出重定向到了正确输出的管道。为了避免bat脚本中的start语句造成读输出操作阻塞，我直接操作了文件描述符，没有os.fdopen转文件对象之后再操作。如果你是administrator或者system之类的特权账号，密码不是必须的。如果是一个普通用户要切换到另外一个账户进行操作则需要密码。

关于windows上的账号(权限)切换

标签：encode   nis   with open   system权限   des   方式   inf   ups   hook   

原文地址：https://www.cnblogs.com/sening/p/10581370.html