I wanted a more enhanced version so I ended up with a module which allows:
UAC request if needed, printing and logging from nonprivileged instance (uses ipc and a network port) and some other candies. usage is just insert elevateme() in your script: in nonprivileged it listen for privileged print/logs and then exits returning false, in privileged instance it returns true immediately.
Supports pyinstaller.
prototype:
# xlogger : a logger in the server/nonprivileged script
# tport : open port of communication, 0 for no comm [printf in nonprivileged window or silent]
# redir : redirect stdout and stderr from privileged instance
#errFile : redirect stderr to file from privileged instance
def elevateme(xlogger=None, tport=6000, redir=True, errFile=False):
winadmin.py
#!/usr/bin/env python
# -*- coding: utf-8; mode: python; py-indent-offset: 4; indent-tabs-mode: nil -*-
# vim: fileencoding=utf-8 tabstop=4 expandtab shiftwidth=4
# (C) COPYRIGHT © Preston Landers 2010
# (C) COPYRIGHT © Matteo Azzali 2020
# Released under the same license as Python 2.6.5/3.7
import sys, os
from traceback import print_exc
from multiprocessing.connection import Listener, Client
import win32event #win32com.shell.shell, win32process
import builtins as __builtin__ # python3
# debug suffixes for remote printing
dbz=["","","",""] #["J:","K:", "G:", "D:"]
LOGTAG="LOGME:"
wrconn = None
#fake logger for message sending
class fakelogger:
def __init__(self, xlogger=None):
self.lg = xlogger
def write(self, a):
global wrconn
if wrconn is not None:
wrconn.send(LOGTAG+a)
elif self.lg is not None:
self.lg.write(a)
else:
print(LOGTAG+a)
class Writer():
wzconn=None
counter = 0
def __init__(self, tport=6000,authkey=b'secret password'):
global wrconn
if wrconn is None:
address = ('localhost', tport)
try:
wrconn = Client(address, authkey=authkey)
except:
wrconn = None
wzconn = wrconn
self.wrconn = wrconn
self.__class__.counter+=1
def __del__(self):
self.__class__.counter-=1
if self.__class__.counter == 0 and wrconn is not None:
import time
time.sleep(0.1) # slows deletion but is enough to print stderr
wrconn.send('close')
wrconn.close()
def sendx(cls, mesg):
cls.wzconn.send(msg)
def sendw(self, mesg):
self.wrconn.send(msg)
#fake file to be passed as stdout and stderr
class connFile():
def __init__(self, thekind="out", tport=6000):
self.cnt = 0
self.old=""
self.vg=Writer(tport)
if thekind == "out":
self.kind=sys.__stdout__
else:
self.kind=sys.__stderr__
def write(self, *args, **kwargs):
global wrconn
global dbz
from io import StringIO # # Python2 use: from cStringIO import StringIO
mystdout = StringIO()
self.cnt+=1
__builtin__.print(*args, **kwargs, file=mystdout, end = '')
#handles "\n" wherever it is, however usually is or string or \n
if "\n" not in mystdout.getvalue():
if mystdout.getvalue() != "\n":
#__builtin__.print("A:",mystdout.getvalue(), file=self.kind, end='')
self.old += mystdout.getvalue()
else:
#__builtin__.print("B:",mystdout.getvalue(), file=self.kind, end='')
if wrconn is not None:
wrconn.send(dbz[1]+self.old)
else:
__builtin__.print(dbz[2]+self.old+ mystdout.getvalue(), file=self.kind, end='')
self.kind.flush()
self.old=""
else:
vv = mystdout.getvalue().split("\n")
#__builtin__.print("V:",vv, file=self.kind, end='')
for el in vv[:-1]:
if wrconn is not None:
wrconn.send(dbz[0]+self.old+el)
self.old = ""
else:
__builtin__.print(dbz[3]+self.old+ el+"\n", file=self.kind, end='')
self.kind.flush()
self.old=""
self.old=vv[-1]
def open(self):
pass
def close(self):
pass
def flush(self):
pass
def isUserAdmin():
if os.name == 'nt':
import ctypes
# WARNING: requires Windows XP SP2 or higher!
try:
return ctypes.windll.shell32.IsUserAnAdmin()
except:
traceback.print_exc()
print ("Admin check failed, assuming not an admin.")
return False
elif os.name == 'posix':
# Check for root on Posix
return os.getuid() == 0
else:
print("Unsupported operating system for this module: %s" % (os.name,))
exit()
#raise (RuntimeError, "Unsupported operating system for this module: %s" % (os.name,))
def runAsAdmin(cmdLine=None, wait=True, hidden=False):
if os.name != 'nt':
raise (RuntimeError, "This function is only implemented on Windows.")
import win32api, win32con, win32process
from win32com.shell.shell import ShellExecuteEx
python_exe = sys.executable
arb=""
if cmdLine is None:
cmdLine = [python_exe] + sys.argv
elif not isinstance(cmdLine, (tuple, list)):
if isinstance(cmdLine, (str)):
arb=cmdLine
cmdLine = [python_exe] + sys.argv
print("original user", arb)
else:
raise( ValueError, "cmdLine is not a sequence.")
cmd = '"%s"' % (cmdLine[0],)
params = " ".join(['"%s"' % (x,) for x in cmdLine[1:]])
if len(arb) > 0:
params += " "+arb
cmdDir = ''
if hidden:
showCmd = win32con.SW_HIDE
else:
showCmd = win32con.SW_SHOWNORMAL
lpVerb = 'runas' # causes UAC elevation prompt.
# print "Running", cmd, params
# ShellExecute() doesn't seem to allow us to fetch the PID or handle
# of the process, so we can't get anything useful from it. Therefore
# the more complex ShellExecuteEx() must be used.
# procHandle = win32api.ShellExecute(0, lpVerb, cmd, params, cmdDir, showCmd)
procInfo = ShellExecuteEx(nShow=showCmd,
fMask=64,
lpVerb=lpVerb,
lpFile=cmd,
lpParameters=params)
if wait:
procHandle = procInfo['hProcess']
obj = win32event.WaitForSingleObject(procHandle, win32event.INFINITE)
rc = win32process.GetExitCodeProcess(procHandle)
#print "Process handle %s returned code %s" % (procHandle, rc)
else:
rc = procInfo['hProcess']
return rc
# xlogger : a logger in the server/nonprivileged script
# tport : open port of communication, 0 for no comm [printf in nonprivileged window or silent]
# redir : redirect stdout and stderr from privileged instance
#errFile : redirect stderr to file from privileged instance
def elevateme(xlogger=None, tport=6000, redir=True, errFile=False):
global dbz
if not isUserAdmin():
print ("You're not an admin.", os.getpid(), "params: ", sys.argv)
import getpass
uname = getpass.getuser()
if (tport> 0):
address = ('localhost', tport) # family is deduced to be 'AF_INET'
listener = Listener(address, authkey=b'secret password')
rc = runAsAdmin(uname, wait=False, hidden=True)
if (tport> 0):
hr = win32event.WaitForSingleObject(rc, 40)
conn = listener.accept()
print ('connection accepted from', listener.last_accepted)
sys.stdout.flush()
while True:
msg = conn.recv()
# do something with msg
if msg == 'close':
conn.close()
break
else:
if msg.startswith(dbz[0]+LOGTAG):
if xlogger != None:
xlogger.write(msg[len(LOGTAG):])
else:
print("Missing a logger")
else:
print(msg)
sys.stdout.flush()
listener.close()
else: #no port connection, its silent
WaitForSingleObject(rc, INFINITE);
return False
else:
#redirect prints stdout on master, errors in error.txt
print("HIADM")
sys.stdout.flush()
if (tport > 0) and (redir):
vox= connFile(tport=tport)
sys.stdout=vox
if not errFile:
sys.stderr=vox
else:
vfrs=open("errFile.txt","w")
sys.stderr=vfrs
#print("HI ADMIN")
return True
def test():
rc = 0
if not isUserAdmin():
print ("You're not an admin.", os.getpid(), "params: ", sys.argv)
sys.stdout.flush()
#rc = runAsAdmin(["c:\\Windows\\notepad.exe"])
rc = runAsAdmin()
else:
print ("You are an admin!", os.getpid(), "params: ", sys.argv)
rc = 0
x = raw_input('Press Enter to exit.')
return rc
if __name__ == "__main__":
sys.exit(test())