#Restructure_Part3

dev-linux
Ivan Maslov 5 years ago
parent 8b267f0a79
commit b458aae7fc

@ -0,0 +1,23 @@
# A sample distutils script to show to build your own
# extension module which extends pywintypes or pythoncom.
#
# Use 'python setup.py build' to build this extension.
import os
from distutils.core import setup, Extension
from distutils.sysconfig import get_python_lib
sources = ["win32_extension.cpp"]
# Specify the directory where the PyWin32 .h and .lib files are installed.
# If you are doing a win32com extension, you will also need to add
# win32com\Include and win32com\Libs.
ext = Extension("win32_extension", sources,
include_dirs = [os.path.join(get_python_lib(), "win32", "Include")],
library_dirs = [os.path.join(get_python_lib(), "win32", "Libs")],
)
setup(
name="win32 extension sample",
version="0.1",
ext_modules=[ext],
)

@ -0,0 +1,213 @@
# A demo of the Windows CE Remote API
#
# This connects to a CE device, and interacts with it.
import wincerapi
import win32event
import win32api
import win32con
import os
import sys
import getopt
def DumpPythonRegistry():
try:
h = wincerapi.CeRegOpenKeyEx(win32con.HKEY_LOCAL_MACHINE, "Software\\Python\\PythonCore\\%s\\PythonPath" % sys.winver)
except win32api.error:
print("The remote device does not appear to have Python installed")
return 0
path, typ = wincerapi.CeRegQueryValueEx(h, None)
print("The remote PythonPath is '%s'" % (str(path), ))
h.Close()
return 1
def DumpRegistry(root, level=0):
# A recursive dump of the remote registry to test most functions.
h = wincerapi.CeRegOpenKeyEx(win32con.HKEY_LOCAL_MACHINE, None)
level_prefix = " " * level
index = 0
# Enumerate values.
while 1:
try:
name, data, typ = wincerapi.CeRegEnumValue(root, index)
except win32api.error:
break
print("%s%s=%s" % (level_prefix, name, repr(str(data))))
index = index+1
# Now enumerate all keys.
index=0
while 1:
try:
name, klass = wincerapi.CeRegEnumKeyEx(root, index)
except win32api.error:
break
print("%s%s\\" % (level_prefix, name))
subkey = wincerapi.CeRegOpenKeyEx(root, name)
DumpRegistry(subkey, level+1)
index = index+1
def DemoCopyFile():
# Create a file on the device, and write a string.
cefile = wincerapi.CeCreateFile("TestPython", win32con.GENERIC_WRITE, 0, None, win32con.OPEN_ALWAYS, 0, None)
wincerapi.CeWriteFile(cefile, "Hello from Python")
cefile.Close()
# reopen the file and check the data.
cefile = wincerapi.CeCreateFile("TestPython", win32con.GENERIC_READ, 0, None, win32con.OPEN_EXISTING, 0, None)
if wincerapi.CeReadFile(cefile, 100) != "Hello from Python":
print("Couldnt read the data from the device!")
cefile.Close()
# Delete the test file
wincerapi.CeDeleteFile("TestPython")
print("Created, wrote to, read from and deleted a test file!")
def DemoCreateProcess():
try:
hp, ht, pid, tid = wincerapi.CeCreateProcess("Windows\\Python.exe", "", None, None, 0, 0, None, "", None)
# Not necessary, except to see if handle closing raises an exception
# (if auto-closed, the error is suppressed)
hp.Close()
ht.Close()
print("Python is running on the remote device!")
except win32api.error as xxx_todo_changeme1:
(hr, fn, msg) = xxx_todo_changeme1.args
print("Couldnt execute remote process -", msg)
def DumpRemoteMachineStatus():
ACLineStatus, BatteryFlag, BatteryLifePercent, BatteryLifeTime, BatteryFullLifeTime, BackupBatteryFlag, BackupBatteryLifePercent, BackupBatteryLifeTime, BackupBatteryLifeTime = \
wincerapi.CeGetSystemPowerStatusEx()
if ACLineStatus:
power = "AC"
else:
power = "battery"
if BatteryLifePercent==255:
batPerc = "unknown"
else:
batPerc = BatteryLifePercent
print("The batteries are at %s%%, and is currently being powered by %s" % (batPerc, power))
memLoad, totalPhys, availPhys, totalPage, availPage, totalVirt, availVirt = \
wincerapi.CeGlobalMemoryStatus()
print("The memory is %d%% utilized." % (memLoad))
print("%-20s%-10s%-10s" % ("", "Total", "Avail"))
print("%-20s%-10s%-10s" % ("Physical Memory", totalPhys, availPhys))
print("%-20s%-10s%-10s" % ("Virtual Memory", totalVirt, availVirt))
print("%-20s%-10s%-10s" % ("Paging file", totalPage, availPage))
storeSize, freeSize = wincerapi.CeGetStoreInformation()
print("%-20s%-10s%-10s" % ("File store", storeSize, freeSize))
print("The CE temp path is", wincerapi.CeGetTempPath())
print("The system info for the device is", wincerapi.CeGetSystemInfo())
def DumpRemoteFolders():
# Dump all special folders possible.
for name, val in list(wincerapi.__dict__.items()):
if name[:6]=="CSIDL_":
try:
loc = str(wincerapi.CeGetSpecialFolderPath(val))
print("Folder %s is at %s" % (name, loc))
except win32api.error as details:
pass
# Get the shortcut targets for the "Start Menu"
print("Dumping start menu shortcuts...")
try:
startMenu = str(wincerapi.CeGetSpecialFolderPath(wincerapi.CSIDL_STARTMENU))
except win32api.error as details:
print("This device has no start menu!", details)
startMenu = None
if startMenu:
for fileAttr in wincerapi.CeFindFiles(os.path.join(startMenu, "*")):
fileName = fileAttr[8]
fullPath = os.path.join(startMenu, str(fileName))
try:
resolved = wincerapi.CeSHGetShortcutTarget(fullPath)
except win32api.error as xxx_todo_changeme:
(rc, fn, msg) = xxx_todo_changeme.args
resolved = "#Error - %s" % msg
print("%s->%s" % (fileName, resolved))
# print "The start menu is at",
# print wincerapi.CeSHGetShortcutTarget("\\Windows\\Start Menu\\Shortcut to Python.exe.lnk")
def usage():
print("Options:")
print("-a - Execute all demos")
print("-p - Execute Python process on remote device")
print("-r - Dump the remote registry")
print("-f - Dump all remote special folder locations")
print("-s - Dont dump machine status")
print("-y - Perform asynch init of CE connection")
def main():
async_init = bStartPython = bDumpRegistry = bDumpFolders = 0
bDumpStatus = 1
try:
opts, args = getopt.getopt(sys.argv[1:], "apr")
except getopt.error as why:
print("Invalid usage:", why)
usage()
return
for o, v in opts:
if o=="-a":
bStartPython = bDumpRegistry = bDumpStatus = bDumpFolders = asynch_init = 1
if o=="-p":
bStartPython=1
if o=="-r":
bDumpRegistry=1
if o=="-s":
bDumpStatus=0
if o=="-f":
bDumpFolders = 1
if o=="-y":
print("Doing asynch init of CE connection")
async_init = 1
if async_init:
event, rc = wincerapi.CeRapiInitEx()
while 1:
rc = win32event.WaitForSingleObject(event, 500)
if rc==win32event.WAIT_OBJECT_0:
# We connected.
break
else:
print("Waiting for Initialize to complete (picture a Cancel button here :)")
else:
wincerapi.CeRapiInit()
print("Connected to remote CE device.")
try:
verinfo = wincerapi.CeGetVersionEx()
print("The device is running windows CE version %d.%d - %s" % (verinfo[0], verinfo[1], verinfo[4]))
if bDumpStatus:
print("Dumping remote machine status")
DumpRemoteMachineStatus()
if bDumpRegistry:
print("Dumping remote registry...")
DumpRegistry(win32con.HKEY_LOCAL_MACHINE)
if bDumpFolders:
print("Dumping remote folder information")
DumpRemoteFolders()
DemoCopyFile()
if bStartPython:
print("Starting remote Python process")
if DumpPythonRegistry():
DemoCreateProcess()
else:
print("Not trying to start Python, as it's not installed")
finally:
wincerapi.CeRapiUninit()
print("Disconnected")
if __name__=='__main__':
main()

@ -0,0 +1,19 @@
# 'Request' example added jjk 11/20/98
import win32ui
import dde
server = dde.CreateServer()
server.Create("TestClient")
conversation = dde.CreateConversation(server)
conversation.ConnectTo("RunAny", "RunAnyCommand")
conversation.Exec("DoSomething")
conversation.Exec("DoSomethingElse")
conversation.ConnectTo("RunAny", "ComputeStringLength")
s = 'abcdefghi'
sl = conversation.Request(s)
print('length of "%s" is %s'%(s,sl))

@ -0,0 +1,38 @@
# 'Request' example added jjk 11/20/98
import win32ui
from pywin.mfc import object
import dde
class MySystemTopic(object.Object):
def __init__(self):
object.Object.__init__(self, dde.CreateServerSystemTopic())
def Exec(self, cmd):
print("System Topic asked to exec", cmd)
class MyOtherTopic(object.Object):
def __init__(self, topicName):
object.Object.__init__(self, dde.CreateTopic(topicName))
def Exec(self, cmd):
print("Other Topic asked to exec", cmd)
class MyRequestTopic(object.Object):
def __init__(self, topicName):
topic = dde.CreateTopic(topicName)
topic.AddItem(dde.CreateStringItem(""))
object.Object.__init__(self, topic)
def Request(self, aString):
print("Request Topic asked to compute length of:", aString)
return(str(len(aString)))
server = dde.CreateServer()
server.AddTopic(MySystemTopic())
server.AddTopic(MyOtherTopic("RunAnyCommand"))
server.AddTopic(MyRequestTopic("ComputeStringLength"))
server.Create('RunAny')
while 1:
win32ui.PumpWaitingMessages(0, -1)

@ -0,0 +1,166 @@
# Demonstrates using a taskbar icon to create and navigate between desktops
import win32api, win32con, win32gui, win32service, win32process
import pywintypes
import traceback, _thread, time
import io
## "Shell_TrayWnd" is class of system tray window, broadcasts "TaskbarCreated" when initialized
def desktop_name_dlgproc(hwnd,msg,wparam,lparam):
""" Handles messages from the desktop name dialog box """
if msg in (win32con.WM_CLOSE,win32con.WM_DESTROY):
win32gui.DestroyWindow(hwnd)
elif msg == win32con.WM_COMMAND:
if wparam == win32con.IDOK:
desktop_name=win32gui.GetDlgItemText(hwnd, 72)
print('new desktop name: ',desktop_name)
win32gui.DestroyWindow(hwnd)
create_desktop(desktop_name)
elif wparam == win32con.IDCANCEL:
win32gui.DestroyWindow(hwnd)
def get_new_desktop_name(parent_hwnd):
""" Create a dialog box to ask the user for name of desktop to be created """
msgs={win32con.WM_COMMAND:desktop_name_dlgproc,
win32con.WM_CLOSE:desktop_name_dlgproc,
win32con.WM_DESTROY:desktop_name_dlgproc}
# dlg item [type, caption, id, (x,y,cx,cy), style, ex style
style=win32con.WS_BORDER|win32con.WS_VISIBLE|win32con.WS_CAPTION|win32con.WS_SYSMENU ## |win32con.DS_SYSMODAL
h=win32gui.CreateDialogIndirect(
win32api.GetModuleHandle(None),
[['One ugly dialog box !',(100,100,200,100),style,0],
['Button','Create', win32con.IDOK, (10,10,30,20),win32con.WS_VISIBLE|win32con.WS_TABSTOP|win32con.BS_HOLLOW|win32con.BS_DEFPUSHBUTTON],
['Button','Never mind', win32con.IDCANCEL, (45,10,50,20),win32con.WS_VISIBLE|win32con.WS_TABSTOP|win32con.BS_HOLLOW],
['Static','Desktop name:',71,(10,40,70,10),win32con.WS_VISIBLE],
['Edit','',72,(75,40,90,10),win32con.WS_VISIBLE]],
parent_hwnd, msgs) ## parent_hwnd, msgs)
win32gui.EnableWindow(h,True)
hcontrol=win32gui.GetDlgItem(h,72)
win32gui.EnableWindow(hcontrol,True)
win32gui.SetFocus(hcontrol)
def new_icon(hdesk,desktop_name):
""" Runs as a thread on each desktop to create a new tray icon and handle its messages """
global id
id=id+1
hdesk.SetThreadDesktop()
## apparently the threads can't use same hinst, so each needs its own window class
windowclassname='PythonDesktopManager'+desktop_name
wc = win32gui.WNDCLASS()
wc.hInstance = win32api.GetModuleHandle(None)
wc.lpszClassName = windowclassname
wc.style = win32con.CS_VREDRAW | win32con.CS_HREDRAW | win32con.CS_GLOBALCLASS
wc.hCursor = win32gui.LoadCursor( 0, win32con.IDC_ARROW )
wc.hbrBackground = win32con.COLOR_WINDOW
wc.lpfnWndProc = icon_wndproc
windowclass = win32gui.RegisterClass(wc)
style = win32con.WS_OVERLAPPED | win32con.WS_SYSMENU
hwnd = win32gui.CreateWindow(windowclass, 'dm_'+desktop_name, win32con.WS_SYSMENU,
0, 0, win32con.CW_USEDEFAULT, win32con.CW_USEDEFAULT,
0, 0, wc.hInstance, None)
win32gui.UpdateWindow(hwnd)
flags = win32gui.NIF_ICON | win32gui.NIF_MESSAGE | win32gui.NIF_TIP
notify_info = (hwnd, id, flags, win32con.WM_USER+20, hicon, 'Desktop Manager (%s)' %desktop_name)
window_info[hwnd]=notify_info
## wait for explorer to initialize system tray for new desktop
tray_found=0
while not tray_found:
try:
tray_found=win32gui.FindWindow("Shell_TrayWnd",None)
except win32gui.error:
traceback.print_exc
time.sleep(.5)
win32gui.Shell_NotifyIcon(win32gui.NIM_ADD, notify_info)
win32gui.PumpMessages()
def create_desktop(desktop_name, start_explorer=1):
""" Creates a new desktop and spawns a thread running on it
Will also start a new icon thread on an existing desktop
"""
sa=pywintypes.SECURITY_ATTRIBUTES()
sa.bInheritHandle=1
try:
hdesk=win32service.CreateDesktop(desktop_name, 0, win32con.MAXIMUM_ALLOWED, sa)
except win32service.error:
traceback.print_exc()
errbuf=io.StringIO()
traceback.print_exc(None,errbuf)
win32api.MessageBox(0, errbuf.getvalue(), 'Desktop creation failed')
return
if start_explorer:
s=win32process.STARTUPINFO()
s.lpDesktop=desktop_name
prc_info=win32process.CreateProcess(None, "Explorer.exe",None,None,True,win32con.CREATE_NEW_CONSOLE,None,'c:\\',s)
th=_thread.start_new_thread(new_icon,(hdesk,desktop_name))
hdesk.SwitchDesktop()
def icon_wndproc(hwnd, msg, wp, lp):
""" Window proc for the tray icons """
if lp==win32con.WM_LBUTTONDOWN:
## popup menu won't disappear if you don't do this
win32gui.SetForegroundWindow(hwnd)
curr_desktop=win32service.OpenInputDesktop(0,True,win32con.MAXIMUM_ALLOWED)
curr_desktop_name=win32service.GetUserObjectInformation(curr_desktop,win32con.UOI_NAME)
winsta=win32service.GetProcessWindowStation()
desktops=winsta.EnumDesktops()
m=win32gui.CreatePopupMenu()
desktop_cnt=len(desktops)
## *don't* create an item 0
for d in range(1, desktop_cnt+1):
mf_flags=win32con.MF_STRING
## if you switch to winlogon yourself, there's nothing there and you're stuck
if desktops[d-1].lower() in ('winlogon','disconnect'):
mf_flags=mf_flags|win32con.MF_GRAYED|win32con.MF_DISABLED
if desktops[d-1]==curr_desktop_name:
mf_flags=mf_flags|win32con.MF_CHECKED
win32gui.AppendMenu(m, mf_flags, d, desktops[d-1])
win32gui.AppendMenu(m, win32con.MF_STRING, desktop_cnt+1, 'Create new ...')
win32gui.AppendMenu(m, win32con.MF_STRING, desktop_cnt+2, 'Exit')
x,y=win32gui.GetCursorPos()
d=win32gui.TrackPopupMenu(m,win32con.TPM_LEFTBUTTON|win32con.TPM_RETURNCMD|win32con.TPM_NONOTIFY,
x,y, 0, hwnd, None)
win32gui.PumpWaitingMessages()
win32gui.DestroyMenu(m)
if d==desktop_cnt+1: ## Create new
get_new_desktop_name(hwnd)
elif d==desktop_cnt+2: ## Exit
win32gui.PostQuitMessage(0)
win32gui.Shell_NotifyIcon(win32gui.NIM_DELETE, window_info[hwnd])
del window_info[hwnd]
origin_desktop.SwitchDesktop()
elif d>0:
hdesk=win32service.OpenDesktop(desktops[d-1],0,0,win32con.MAXIMUM_ALLOWED)
hdesk.SwitchDesktop()
return 0
else:
return win32gui.DefWindowProc(hwnd, msg, wp, lp)
window_info={}
origin_desktop=win32service.OpenInputDesktop(0, True, win32con.MAXIMUM_ALLOWED)
origin_desktop_name=win32service.GetUserObjectInformation(origin_desktop, win32service.UOI_NAME)
hinst=win32api.GetModuleHandle(None)
try:
hicon=win32gui.LoadIcon(hinst, 1) ## python.exe and pythonw.exe
except win32gui.error:
hicon=win32gui.LoadIcon(hinst, 135) ## pythonwin's icon
id=0
create_desktop(str(origin_desktop_name),0)
## wait for first thread to initialize its icon
while not window_info:
time.sleep(1)
## exit when last tray icon goes away
while window_info:
win32gui.PumpWaitingMessages()
time.sleep(3)

@ -0,0 +1,111 @@
import win32evtlog
import win32api
import win32con
import win32security # To translate NT Sids to account names.
import win32evtlogutil
def ReadLog(computer, logType="Application", dumpEachRecord = 0):
# read the entire log back.
h=win32evtlog.OpenEventLog(computer, logType)
numRecords = win32evtlog.GetNumberOfEventLogRecords(h)
# print "There are %d records" % numRecords
num=0
while 1:
objects = win32evtlog.ReadEventLog(h, win32evtlog.EVENTLOG_BACKWARDS_READ|win32evtlog.EVENTLOG_SEQUENTIAL_READ, 0)
if not objects:
break
for object in objects:
# get it for testing purposes, but dont print it.
msg = win32evtlogutil.SafeFormatMessage(object, logType)
if object.Sid is not None:
try:
domain, user, typ = win32security.LookupAccountSid(computer, object.Sid)
sidDesc = "%s/%s" % (domain, user)
except win32security.error:
sidDesc = str(object.Sid)
user_desc = "Event associated with user %s" % (sidDesc,)
else:
user_desc = None
if dumpEachRecord:
print("Event record from %r generated at %s" % (object.SourceName, object.TimeGenerated.Format()))
if user_desc:
print(user_desc)
try:
print(msg)
except UnicodeError:
print("(unicode error printing message: repr() follows...)")
print(repr(msg))
num = num + len(objects)
if numRecords == num:
print("Successfully read all", numRecords, "records")
else:
print("Couldn't get all records - reported %d, but found %d" % (numRecords, num))
print("(Note that some other app may have written records while we were running!)")
win32evtlog.CloseEventLog(h)
def usage():
print("Writes an event to the event log.")
print("-w : Dont write any test records.")
print("-r : Dont read the event log")
print("-c : computerName : Process the log on the specified computer")
print("-v : Verbose")
print("-t : LogType - Use the specified log - default = 'Application'")
def test():
# check if running on Windows NT, if not, display notice and terminate
if win32api.GetVersion() & 0x80000000:
print("This sample only runs on NT")
return
import sys, getopt
opts, args = getopt.getopt(sys.argv[1:], "rwh?c:t:v")
computer = None
do_read = do_write = 1
logType = "Application"
verbose = 0
if len(args)>0:
print("Invalid args")
usage()
return 1
for opt, val in opts:
if opt == '-t':
logType = val
if opt == '-c':
computer = val
if opt in ['-h', '-?']:
usage()
return
if opt=='-r':
do_read = 0
if opt=='-w':
do_write = 0
if opt=='-v':
verbose = verbose + 1
if do_write:
ph=win32api.GetCurrentProcess()
th = win32security.OpenProcessToken(ph,win32con.TOKEN_READ)
my_sid = win32security.GetTokenInformation(th,win32security.TokenUser)[0]
win32evtlogutil.ReportEvent(logType, 2,
strings=["The message text for event 2","Another insert"],
data = "Raw\0Data".encode("ascii"), sid = my_sid)
win32evtlogutil.ReportEvent(logType, 1, eventType=win32evtlog.EVENTLOG_WARNING_TYPE,
strings=["A warning","An even more dire warning"],
data = "Raw\0Data".encode("ascii"), sid = my_sid)
win32evtlogutil.ReportEvent(logType, 1, eventType=win32evtlog.EVENTLOG_INFORMATION_TYPE,
strings=["An info","Too much info"],
data = "Raw\0Data".encode("ascii"), sid = my_sid)
print("Successfully wrote 3 records to the log")
if do_read:
ReadLog(computer, logType, verbose > 0)
if __name__=='__main__':
test()

@ -0,0 +1,21 @@
import os, win32api
ver_strings=('Comments','InternalName','ProductName',
'CompanyName','LegalCopyright','ProductVersion',
'FileDescription','LegalTrademarks','PrivateBuild',
'FileVersion','OriginalFilename','SpecialBuild')
fname = os.environ["comspec"]
d=win32api.GetFileVersionInfo(fname, '\\')
## backslash as parm returns dictionary of numeric info corresponding to VS_FIXEDFILEINFO struc
for n, v in d.items():
print(n, v)
pairs=win32api.GetFileVersionInfo(fname, '\\VarFileInfo\\Translation')
## \VarFileInfo\Translation returns list of available (language, codepage) pairs that can be used to retreive string info
## any other must be of the form \StringfileInfo\%04X%04X\parm_name, middle two are language/codepage pair returned from above
for lang, codepage in pairs:
print('lang: ', lang, 'codepage:', codepage)
for ver_string in ver_strings:
str_info='\\StringFileInfo\\%04X%04X\\%s' %(lang,codepage,ver_string)
## print str_info
print(ver_string, repr(win32api.GetFileVersionInfo(fname, str_info)))

Binary file not shown.

After

Width:  |  Height:  |  Size: 3.1 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 3.1 KiB

@ -0,0 +1,95 @@
import win32api, mmapfile
import winerror
import tempfile, os
from pywin32_testutil import str2bytes
system_info=win32api.GetSystemInfo()
page_size=system_info[1]
alloc_size=system_info[7]
fname=tempfile.mktemp()
mapping_name=os.path.split(fname)[1]
fsize=8*page_size
print(fname, fsize, mapping_name)
m1=mmapfile.mmapfile(File=fname, Name=mapping_name, MaximumSize=fsize)
m1.seek(100)
m1.write_byte(str2bytes('?'))
m1.seek(-1,1)
assert m1.read_byte()==str2bytes('?')
## A reopened named mapping should have exact same size as original mapping
m2=mmapfile.mmapfile(Name=mapping_name, File=None, MaximumSize=fsize*2)
assert m2.size()==m1.size()
m1.seek(0,0)
m1.write(fsize*str2bytes('s'))
assert m2.read(fsize)==fsize*str2bytes('s')
move_src=100
move_dest=500
move_size=150
m2.seek(move_src,0)
assert m2.tell()==move_src
m2.write(str2bytes('m')*move_size)
m2.move(move_dest, move_src, move_size)
m2.seek(move_dest, 0)
assert m2.read(move_size) == str2bytes('m') * move_size
## m2.write('x'* (fsize+1))
m2.close()
m1.resize(fsize*2)
assert m1.size()==fsize * 2
m1.seek(fsize)
m1.write(str2bytes('w') * fsize)
m1.flush()
m1.close()
os.remove(fname)
## Test a file with size larger than 32 bits
## need 10 GB free on drive where your temp folder lives
fname_large=tempfile.mktemp()
mapping_name='Pywin32_large_mmap'
offsetdata=str2bytes('This is start of offset')
## Deliberately use odd numbers to test rounding logic
fsize = (1024*1024*1024*10) + 333
offset = (1024*1024*32) + 42
view_size = (1024*1024*16) + 111
## round mapping size and view size up to multiple of system page size
if fsize%page_size:
fsize += page_size - (fsize%page_size)
if view_size%page_size:
view_size += page_size - (view_size%page_size)
## round offset down to multiple of allocation granularity
offset -= offset%alloc_size
m1=None
m2=None
try:
try:
m1=mmapfile.mmapfile(fname_large, mapping_name, fsize, 0, offset*2)
except mmapfile.error as exc:
# if we don't have enough disk-space, that's OK.
if exc.winerror!=winerror.ERROR_DISK_FULL:
raise
print("skipping large file test - need", fsize, "available bytes.")
else:
m1.seek(offset)
m1.write(offsetdata)
## When reopening an existing mapping without passing a file handle, you have
## to specify a positive size even though it's ignored
m2=mmapfile.mmapfile(File=None, Name=mapping_name, MaximumSize=1,
FileOffset=offset, NumberOfBytesToMap=view_size)
assert m2.read(len(offsetdata))==offsetdata
finally:
if m1 is not None:
m1.close()
if m2 is not None:
m2.close()
if os.path.exists(fname_large):
os.remove(fname_large)

@ -0,0 +1,16 @@
'''cat.py
a version of unix cat, tweaked to show off runproc.py
'''
import sys
data = sys.stdin.read(1)
sys.stdout.write(data)
sys.stdout.flush()
while data:
data = sys.stdin.read(1)
sys.stdout.write(data)
sys.stdout.flush()
# Just here to have something to read from stderr.
sys.stderr.write("Blah...")
# end of cat.py

@ -0,0 +1,110 @@
'''runproc.py
start a process with three inherited pipes.
Try to write to and read from those.
'''
import win32api
import win32pipe
import win32file
import win32process
import win32security
import win32con
import msvcrt
import os
class Process:
def run(self, cmdline):
# security attributes for pipes
sAttrs = win32security.SECURITY_ATTRIBUTES()
sAttrs.bInheritHandle = 1
# create pipes
hStdin_r, self.hStdin_w = win32pipe.CreatePipe(sAttrs, 0)
self.hStdout_r, hStdout_w = win32pipe.CreatePipe(sAttrs, 0)
self.hStderr_r, hStderr_w = win32pipe.CreatePipe(sAttrs, 0)
# set the info structure for the new process.
StartupInfo = win32process.STARTUPINFO()
StartupInfo.hStdInput = hStdin_r
StartupInfo.hStdOutput = hStdout_w
StartupInfo.hStdError = hStderr_w
StartupInfo.dwFlags = win32process.STARTF_USESTDHANDLES
# Mark doesn't support wShowWindow yet.
# StartupInfo.dwFlags = StartupInfo.dwFlags | win32process.STARTF_USESHOWWINDOW
# StartupInfo.wShowWindow = win32con.SW_HIDE
# Create new output read handles and the input write handle. Set
# the inheritance properties to FALSE. Otherwise, the child inherits
# the these handles; resulting in non-closeable handles to the pipes
# being created.
pid = win32api.GetCurrentProcess()
tmp = win32api.DuplicateHandle(
pid,
self.hStdin_w,
pid,
0,
0, # non-inheritable!!
win32con.DUPLICATE_SAME_ACCESS)
# Close the inhertible version of the handle
win32file.CloseHandle(self.hStdin_w)
self.hStdin_w = tmp
tmp = win32api.DuplicateHandle(
pid,
self.hStdout_r,
pid,
0,
0, # non-inheritable!
win32con.DUPLICATE_SAME_ACCESS)
# Close the inhertible version of the handle
win32file.CloseHandle(self.hStdout_r)
self.hStdout_r = tmp
# start the process.
hProcess, hThread, dwPid, dwTid = win32process.CreateProcess(
None, # program
cmdline,# command line
None, # process security attributes
None, # thread attributes
1, # inherit handles, or USESTDHANDLES won't work.
# creation flags. Don't access the console.
0, # Don't need anything here.
# If you're in a GUI app, you should use
# CREATE_NEW_CONSOLE here, or any subprocesses
# might fall victim to the problem described in:
# KB article: Q156755, cmd.exe requires
# an NT console in order to perform redirection..
None, # no new environment
None, # current directory (stay where we are)
StartupInfo)
# normally, we would save the pid etc. here...
# Child is launched. Close the parents copy of those pipe handles
# that only the child should have open.
# You need to make sure that no handles to the write end of the
# output pipe are maintained in this process or else the pipe will
# not close when the child process exits and the ReadFile will hang.
win32file.CloseHandle(hStderr_w)
win32file.CloseHandle(hStdout_w)
win32file.CloseHandle(hStdin_r)
self.stdin = os.fdopen(msvcrt.open_osfhandle(self.hStdin_w, 0), "wb")
self.stdin.write('hmmmmm\r\n')
self.stdin.flush()
self.stdin.close()
self.stdout = os.fdopen(msvcrt.open_osfhandle(self.hStdout_r, 0), "rb")
print("Read on stdout: ", repr(self.stdout.read()))
self.stderr = os.fdopen(msvcrt.open_osfhandle(self.hStderr_r, 0), "rb")
print("Read on stderr: ", repr(self.stderr.read()))
if __name__ == '__main__':
p = Process()
exe = win32api.GetModuleFileName(0)
p.run(exe + ' cat.py')
# end of runproc.py

@ -0,0 +1,67 @@
import win32print, pywintypes, win32con, win32gui, win32api
pname=win32print.GetDefaultPrinter()
print(pname)
p=win32print.OpenPrinter(pname)
print('Printer handle: ',p)
print_processor=win32print.GetPrinter(p,2)['pPrintProcessor']
## call with last parm set to 0 to get total size needed for printer's DEVMODE
dmsize=win32print.DocumentProperties(0, p, pname, None, None, 0)
## dmDriverExtra should be total size - fixed size
driverextra=dmsize - pywintypes.DEVMODEType().Size ## need a better way to get DEVMODE.dmSize
dm=pywintypes.DEVMODEType(driverextra)
dm.Fields=dm.Fields|win32con.DM_ORIENTATION|win32con.DM_COPIES
dm.Orientation=win32con.DMORIENT_LANDSCAPE
dm.Copies=2
win32print.DocumentProperties(0, p, pname, dm, dm, win32con.DM_IN_BUFFER|win32con.DM_OUT_BUFFER)
pDC=win32gui.CreateDC(print_processor,pname,dm)
printerwidth=win32print.GetDeviceCaps(pDC, win32con.PHYSICALWIDTH)
printerheight=win32print.GetDeviceCaps(pDC, win32con.PHYSICALHEIGHT)
hwnd=win32gui.GetDesktopWindow()
l,t,r,b=win32gui.GetWindowRect(hwnd)
desktopheight=b-t
desktopwidth=r-l
dDC = win32gui.GetWindowDC(hwnd)
dcDC=win32gui.CreateCompatibleDC(dDC)
dcBM = win32gui.CreateCompatibleBitmap(dDC, desktopwidth, desktopheight);
win32gui.SelectObject(dcDC, dcBM)
win32gui.StretchBlt(dcDC, 0, 0, desktopwidth, desktopheight, dDC, 0, 0, desktopwidth, desktopheight, win32con.SRCCOPY)
pcDC=win32gui.CreateCompatibleDC(pDC)
pcBM=win32gui.CreateCompatibleBitmap(pDC, printerwidth, printerheight)
win32gui.SelectObject(pcDC, pcBM)
win32gui.StretchBlt(pcDC, 0, 0, printerwidth, printerheight, dcDC, 0, 0, desktopwidth, desktopheight, win32con.SRCCOPY)
win32print.StartDoc(pDC,('desktop.bmp',None,None,0))
win32print.StartPage(pDC)
win32gui.StretchBlt(pDC, 0, 0, int(printerwidth*.9), int(printerheight*.9), pcDC, 0, 0, printerwidth, printerheight, win32con.SRCCOPY)
font=win32gui.LOGFONT()
font.lfHeight=int(printerheight/20)
font.lfWidth=font.lfHeight
font.lfWeight=150
font.lfItalic=1
font.lfUnderline=1
hf=win32gui.CreateFontIndirect(font)
win32gui.SelectObject(pDC,hf)
win32gui.SetBkMode(pDC, win32con.TRANSPARENT)
win32gui.SetTextColor(pDC,win32api.RGB(0,255,0))
win32gui.DrawText(pDC,'Printed by Python!', -1,
(0,0, int(printerwidth*.9), int(printerheight*.9)),
win32con.DT_RIGHT|win32con.DT_BOTTOM|win32con.DT_SINGLELINE)
win32print.EndPage(pDC)
win32print.EndDoc(pDC)
win32print.ClosePrinter(p)
win32gui.DeleteObject(dcBM)
win32gui.DeleteObject(pcBM)
win32gui.DeleteObject(hf)
win32gui.DeleteDC(dDC)
win32gui.DeleteDC(dcDC)
win32gui.DeleteDC(pDC)
win32gui.DeleteDC(pcDC)

@ -0,0 +1,136 @@
# rastest.py - test/demonstrate the win32ras module.
# Much of the code here contributed by Jethro Wright.
import sys
import os
import win32ras
# Build a little dictionary of RAS states to decent strings.
# eg win32ras.RASCS_OpenPort -> "OpenPort"
stateMap = {}
for name, val in list(win32ras.__dict__.items()):
if name[:6]=="RASCS_":
stateMap[val] = name[6:]
# Use a lock so the callback can tell the main thread when it is finished.
import win32event
callbackEvent = win32event.CreateEvent(None, 0, 0, None)
def Callback( hras, msg, state, error, exterror):
# print "Callback called with ", hras, msg, state, error, exterror
stateName = stateMap.get(state, "Unknown state?")
print("Status is %s (%04lx), error code is %d" % (stateName, state, error))
finished = state in [win32ras.RASCS_Connected]
if finished:
win32event.SetEvent(callbackEvent)
if error != 0 or int( state ) == win32ras.RASCS_Disconnected:
# we know for sure this is a good place to hangup....
print("Detected call failure: %s" % win32ras.GetErrorString( error ))
HangUp( hras )
win32event.SetEvent(callbackEvent)
def ShowConnections():
print("All phone-book entries:")
for (name,) in win32ras.EnumEntries():
print(" ", name)
print("Current Connections:")
for con in win32ras.EnumConnections():
print(" ", con)
def EditEntry(entryName):
try:
win32ras.EditPhonebookEntry(0,None,entryName)
except win32ras.error as xxx_todo_changeme:
(rc, function, msg) = xxx_todo_changeme.args
print("Can not edit/find the RAS entry -", msg)
def HangUp( hras ):
# trap potential, irrelevant errors from win32ras....
try:
win32ras.HangUp( hras )
except:
print("Tried to hang up gracefully on error, but didn't work....")
return None
def Connect(entryName, bUseCallback):
if bUseCallback:
theCallback = Callback
win32event.ResetEvent(callbackEvent)
else:
theCallback = None
# in order to *use* the username/password of a particular dun entry, one must
# explicitly get those params under win95....
try:
dp, b = win32ras.GetEntryDialParams( None, entryName )
except:
print("Couldn't find DUN entry: %s" % entryName)
else:
hras, rc = win32ras.Dial(None, None, (entryName, "", "", dp[ 3 ], dp[ 4 ], ""),theCallback)
# hras, rc = win32ras.Dial(None, None, (entryName, ),theCallback)
# print hras, rc
if not bUseCallback and rc != 0:
print("Could not dial the RAS connection:", win32ras.GetErrorString(rc))
hras = HangUp( hras )
# don't wait here if there's no need to....
elif bUseCallback and win32event.WaitForSingleObject(callbackEvent, 60000)!=win32event.WAIT_OBJECT_0:
print("Gave up waiting for the process to complete!")
# sdk docs state one must explcitly hangup, even if there's an error....
try:
cs = win32ras.GetConnectStatus( hras )
except:
# on error, attempt a hang up anyway....
hras = HangUp( hras )
else:
if int( cs[ 0 ] ) == win32ras.RASCS_Disconnected:
hras = HangUp( hras )
return hras, rc
def Disconnect( rasEntry ):
# Need to find the entry
name = rasEntry.lower()
for hcon, entryName, devName, devType in win32ras.EnumConnections():
if entryName.lower() == name:
win32ras.HangUp( hcon )
print("Disconnected from", rasEntry)
break
else:
print("Could not find an open connection to", entryName)
usage = """
Usage: %s [-s] [-l] [-c connection] [-d connection]
-l : List phone-book entries and current connections.
-s : Show status while connecting/disconnecting (uses callbacks)
-c : Connect to the specified phonebook name.
-d : Disconnect from the specified phonebook name.
-e : Edit the specified phonebook entry.
"""
def main():
import getopt
try:
opts, args = getopt.getopt(sys.argv[1:], "slc:d:e:")
except getopt.error as why:
print(why)
print(usage % (os.path.basename(sys.argv[0],)))
return
bCallback = 0
if args or not opts:
print(usage % (os.path.basename(sys.argv[0],)))
return
for opt, val in opts:
if opt=="-s":
bCallback = 1
if opt=="-l":
ShowConnections()
if opt=="-c":
hras, rc = Connect(val, bCallback)
if hras != None:
print("hras: 0x%8lx, rc: 0x%04x" % ( hras, rc ))
if opt=="-d":
Disconnect(val)
if opt=="-e":
EditEntry(val)
if __name__=='__main__':
main()

@ -0,0 +1,73 @@
""" Lists various types of information about current user's access token,
including UAC status on Vista
"""
import pywintypes, win32api, win32security
import win32con, winerror
from security_enums import TOKEN_GROUP_ATTRIBUTES, TOKEN_PRIVILEGE_ATTRIBUTES, \
SECURITY_IMPERSONATION_LEVEL, TOKEN_TYPE, TOKEN_ELEVATION_TYPE
def dump_token(th):
token_type=win32security.GetTokenInformation(th, win32security.TokenType)
print('TokenType:', token_type, TOKEN_TYPE.lookup_name(token_type))
if token_type==win32security.TokenImpersonation:
imp_lvl=win32security.GetTokenInformation(th, win32security.TokenImpersonationLevel)
print('TokenImpersonationLevel:', imp_lvl, SECURITY_IMPERSONATION_LEVEL.lookup_name(imp_lvl))
print('TokenSessionId:', win32security.GetTokenInformation(th, win32security.TokenSessionId))
privs=win32security.GetTokenInformation(th,win32security.TokenPrivileges)
print('TokenPrivileges:')
for priv_luid, priv_flags in privs:
flag_names, unk=TOKEN_PRIVILEGE_ATTRIBUTES.lookup_flags(priv_flags)
flag_desc = ' '.join(flag_names)
if (unk):
flag_desc += '(' + str(unk) + ')'
priv_name=win32security.LookupPrivilegeName('',priv_luid)
priv_desc=win32security.LookupPrivilegeDisplayName('',priv_name)
print('\t', priv_name, priv_desc, priv_flags, flag_desc)
print('TokenGroups:')
groups=win32security.GetTokenInformation(th,win32security.TokenGroups)
for group_sid, group_attr in groups:
flag_names, unk=TOKEN_GROUP_ATTRIBUTES.lookup_flags(group_attr)
flag_desc = ' '.join(flag_names)
if (unk):
flag_desc += '(' + str(unk) + ')'
if group_attr & TOKEN_GROUP_ATTRIBUTES.SE_GROUP_LOGON_ID:
sid_desc = 'Logon sid'
else:
sid_desc=win32security.LookupAccountSid('',group_sid)
print('\t',group_sid, sid_desc, group_attr, flag_desc)
## Vista token information types, will throw (87, 'GetTokenInformation', 'The parameter is incorrect.') on earier OS
try:
is_elevated=win32security.GetTokenInformation(th, win32security.TokenElevation)
print('TokenElevation:', is_elevated)
except pywintypes.error as details:
if details.winerror != winerror.ERROR_INVALID_PARAMETER:
raise
return None
print('TokenHasRestrictions:', win32security.GetTokenInformation(th, win32security.TokenHasRestrictions))
print('TokenMandatoryPolicy', win32security.GetTokenInformation(th, win32security.TokenMandatoryPolicy))
print('TokenVirtualizationAllowed:', win32security.GetTokenInformation(th, win32security.TokenVirtualizationAllowed))
print('TokenVirtualizationEnabled:', win32security.GetTokenInformation(th, win32security.TokenVirtualizationEnabled))
elevation_type = win32security.GetTokenInformation(th, win32security.TokenElevationType)
print('TokenElevationType:', elevation_type, TOKEN_ELEVATION_TYPE.lookup_name(elevation_type))
if elevation_type!=win32security.TokenElevationTypeDefault:
lt=win32security.GetTokenInformation(th, win32security.TokenLinkedToken)
print('TokenLinkedToken:', lt)
else:
lt=None
return lt
ph = win32api.GetCurrentProcess()
th = win32security.OpenProcessToken(ph,win32con.MAXIMUM_ALLOWED)
lt = dump_token(th)
if lt:
print('\n\nlinked token info:')
dump_token(lt)

@ -0,0 +1,31 @@
import win32security,win32file,win32api,ntsecuritycon,win32con
from security_enums import TRUSTEE_TYPE,TRUSTEE_FORM,ACE_FLAGS,ACCESS_MODE
new_privs = ((win32security.LookupPrivilegeValue('',ntsecuritycon.SE_SECURITY_NAME),win32con.SE_PRIVILEGE_ENABLED),
(win32security.LookupPrivilegeValue('',ntsecuritycon.SE_CREATE_PERMANENT_NAME),win32con.SE_PRIVILEGE_ENABLED),
(win32security.LookupPrivilegeValue('','SeEnableDelegationPrivilege'),win32con.SE_PRIVILEGE_ENABLED) ##doesn't seem to be in ntsecuritycon.py ?
)
ph = win32api.GetCurrentProcess()
th = win32security.OpenProcessToken(ph,win32security.TOKEN_ALL_ACCESS) ##win32con.TOKEN_ADJUST_PRIVILEGES)
win32security.AdjustTokenPrivileges(th,0,new_privs)
policy_handle = win32security.GetPolicyHandle('',win32security.POLICY_ALL_ACCESS)
tmp_sid = win32security.LookupAccountName('','tmp')[0]
privs=[ntsecuritycon.SE_DEBUG_NAME,ntsecuritycon.SE_TCB_NAME,ntsecuritycon.SE_RESTORE_NAME,ntsecuritycon.SE_REMOTE_SHUTDOWN_NAME]
win32security.LsaAddAccountRights(policy_handle,tmp_sid,privs)
privlist=win32security.LsaEnumerateAccountRights(policy_handle,tmp_sid)
for priv in privlist:
print(priv)
privs=[ntsecuritycon.SE_DEBUG_NAME,ntsecuritycon.SE_TCB_NAME]
win32security.LsaRemoveAccountRights(policy_handle,tmp_sid,0,privs)
privlist=win32security.LsaEnumerateAccountRights(policy_handle,tmp_sid)
for priv in privlist:
print(priv)
win32security.LsaClose(policy_handle)

@ -0,0 +1,144 @@
import os
import win32security,win32file,win32api,ntsecuritycon,win32con
from security_enums import TRUSTEE_TYPE,TRUSTEE_FORM,ACE_FLAGS,ACCESS_MODE
fname = os.path.join(win32api.GetTempPath(), "win32security_test.txt")
f=open(fname, "w")
f.write("Hello from Python\n");
f.close()
print("Testing on file", fname)
new_privs = ((win32security.LookupPrivilegeValue('',ntsecuritycon.SE_SECURITY_NAME),win32con.SE_PRIVILEGE_ENABLED),
(win32security.LookupPrivilegeValue('',ntsecuritycon.SE_SHUTDOWN_NAME),win32con.SE_PRIVILEGE_ENABLED),
(win32security.LookupPrivilegeValue('',ntsecuritycon.SE_RESTORE_NAME),win32con.SE_PRIVILEGE_ENABLED),
(win32security.LookupPrivilegeValue('',ntsecuritycon.SE_TAKE_OWNERSHIP_NAME),win32con.SE_PRIVILEGE_ENABLED),
(win32security.LookupPrivilegeValue('',ntsecuritycon.SE_CREATE_PERMANENT_NAME),win32con.SE_PRIVILEGE_ENABLED),
(win32security.LookupPrivilegeValue('','SeEnableDelegationPrivilege'),win32con.SE_PRIVILEGE_ENABLED) ##doesn't seem to be in ntsecuritycon.py ?
)
ph = win32api.GetCurrentProcess()
th = win32security.OpenProcessToken(ph,win32security.TOKEN_ALL_ACCESS) ##win32con.TOKEN_ADJUST_PRIVILEGES)
win32security.AdjustTokenPrivileges(th,0,new_privs)
all_security_info = \
win32security.OWNER_SECURITY_INFORMATION|win32security.GROUP_SECURITY_INFORMATION| \
win32security.DACL_SECURITY_INFORMATION|win32security.SACL_SECURITY_INFORMATION
sd=win32security.GetFileSecurity(fname,all_security_info)
old_sacl=sd.GetSecurityDescriptorSacl()
if old_sacl==None:
old_sacl=win32security.ACL()
old_dacl=sd.GetSecurityDescriptorDacl()
if old_dacl==None:
old_dacl=win32security.ACL()
my_sid = win32security.GetTokenInformation(th,ntsecuritycon.TokenUser)[0]
tmp_sid = win32security.LookupAccountName('','tmp')[0]
pwr_sid = win32security.LookupAccountName('','Power Users')[0]
## MultipleTrustee,MultipleTrusteeOperation,TrusteeForm,TrusteeType,Identifier
## first two are ignored
my_trustee = {}
my_trustee['MultipleTrustee']=None
my_trustee['MultipleTrusteeOperation']=0
my_trustee['TrusteeForm']=TRUSTEE_FORM.TRUSTEE_IS_SID
my_trustee['TrusteeType']=TRUSTEE_TYPE.TRUSTEE_IS_USER
my_trustee['Identifier']=my_sid
tmp_trustee = {}
tmp_trustee['MultipleTrustee']=None
tmp_trustee['MultipleTrusteeOperation']=0
tmp_trustee['TrusteeForm']=TRUSTEE_FORM.TRUSTEE_IS_NAME
tmp_trustee['TrusteeType']=TRUSTEE_TYPE.TRUSTEE_IS_USER
tmp_trustee['Identifier']='rupole\\tmp'
pwr_trustee = {}
pwr_trustee['MultipleTrustee']=None
pwr_trustee['MultipleTrusteeOperation']=0
pwr_trustee['TrusteeForm']=TRUSTEE_FORM.TRUSTEE_IS_SID
pwr_trustee['TrusteeType']=TRUSTEE_TYPE.TRUSTEE_IS_USER
pwr_trustee['Identifier']=pwr_sid
expl_list=[]
expl_list.append(
{
'Trustee':my_trustee,
'Inheritance':ACE_FLAGS.NO_INHERITANCE,
'AccessMode':ACCESS_MODE.SET_AUDIT_SUCCESS, ##|ACCESS_MODE.SET_AUDIT_FAILURE,
'AccessPermissions':win32con.GENERIC_ALL
}
)
expl_list.append(
{
'Trustee':my_trustee,
'Inheritance':ACE_FLAGS.NO_INHERITANCE,
'AccessMode':ACCESS_MODE.SET_AUDIT_FAILURE,
'AccessPermissions':win32con.GENERIC_ALL
}
)
expl_list.append(
{
'Trustee':tmp_trustee,
'Inheritance':ACE_FLAGS.NO_INHERITANCE,
'AccessMode':ACCESS_MODE.SET_AUDIT_SUCCESS,
'AccessPermissions':win32con.GENERIC_ALL
}
)
expl_list.append(
{
'Trustee':tmp_trustee,
'Inheritance':ACE_FLAGS.NO_INHERITANCE,
'AccessMode':ACCESS_MODE.SET_AUDIT_FAILURE,
'AccessPermissions':win32con.GENERIC_ALL
}
)
old_sacl.SetEntriesInAcl(expl_list)
expl_list=[]
expl_list.append(
{
'Trustee':tmp_trustee,
'Inheritance':ACE_FLAGS.NO_INHERITANCE,
'AccessMode':ACCESS_MODE.DENY_ACCESS,
'AccessPermissions':win32con.DELETE
}
)
expl_list.append(
{
'Trustee':tmp_trustee,
'Inheritance':ACE_FLAGS.NO_INHERITANCE,
'AccessMode':ACCESS_MODE.GRANT_ACCESS,
'AccessPermissions':win32con.WRITE_OWNER
}
)
expl_list.append(
{
'Trustee':pwr_trustee,
'Inheritance':ACE_FLAGS.NO_INHERITANCE,
'AccessMode':ACCESS_MODE.GRANT_ACCESS,
'AccessPermissions':win32con.GENERIC_READ
}
)
expl_list.append(
{
'Trustee':my_trustee,
'Inheritance':ACE_FLAGS.NO_INHERITANCE,
'AccessMode':ACCESS_MODE.GRANT_ACCESS,
'AccessPermissions':win32con.GENERIC_ALL
}
)
old_dacl.SetEntriesInAcl(expl_list)
sd.SetSecurityDescriptorSacl(1,old_sacl,1)
sd.SetSecurityDescriptorDacl(1,old_dacl,1)
sd.SetSecurityDescriptorOwner(pwr_sid,1)
win32security.SetFileSecurity(fname,
all_security_info,
sd)

@ -0,0 +1,23 @@
import win32security,win32file,win32api,ntsecuritycon,win32con
policy_handle = win32security.GetPolicyHandle('rupole',win32security.POLICY_ALL_ACCESS)
## mod_nbr, mod_time = win32security.LsaQueryInformationPolicy(policy_handle,win32security.PolicyModificationInformation)
## print mod_nbr, mod_time
domain_name,dns_domain_name, dns_forest_name, domain_guid, domain_sid = \
win32security.LsaQueryInformationPolicy(policy_handle,win32security.PolicyDnsDomainInformation)
print(domain_name, dns_domain_name, dns_forest_name, domain_guid, domain_sid)
event_audit_info=win32security.LsaQueryInformationPolicy(policy_handle,win32security.PolicyAuditEventsInformation)
print(event_audit_info)
domain_name,sid =win32security.LsaQueryInformationPolicy(policy_handle,win32security.PolicyPrimaryDomainInformation)
print(domain_name, sid)
domain_name,sid =win32security.LsaQueryInformationPolicy(policy_handle,win32security.PolicyAccountDomainInformation)
print(domain_name, sid)
server_role = win32security.LsaQueryInformationPolicy(policy_handle,win32security.PolicyLsaServerRoleInformation)
print('server role: ',server_role)
win32security.LsaClose(policy_handle)

@ -0,0 +1,20 @@
import win32security,win32file,win32api,ntsecuritycon,win32con
from security_enums import TRUSTEE_TYPE,TRUSTEE_FORM,ACE_FLAGS,ACCESS_MODE
new_privs = ((win32security.LookupPrivilegeValue('',ntsecuritycon.SE_SECURITY_NAME),win32con.SE_PRIVILEGE_ENABLED),
(win32security.LookupPrivilegeValue('',ntsecuritycon.SE_CREATE_PERMANENT_NAME),win32con.SE_PRIVILEGE_ENABLED),
(win32security.LookupPrivilegeValue('','SeEnableDelegationPrivilege'),win32con.SE_PRIVILEGE_ENABLED) ##doesn't seem to be in ntsecuritycon.py ?
)
ph = win32api.GetCurrentProcess()
th = win32security.OpenProcessToken(ph,win32security.TOKEN_ALL_ACCESS) ##win32con.TOKEN_ADJUST_PRIVILEGES)
win32security.AdjustTokenPrivileges(th,0,new_privs)
policy_handle = win32security.GetPolicyHandle('',win32security.POLICY_ALL_ACCESS)
sidlist=win32security.LsaEnumerateAccountsWithUserRight(policy_handle,ntsecuritycon.SE_RESTORE_NAME)
for sid in sidlist:
print(win32security.LookupAccountSid('',sid))
win32security.LsaClose(policy_handle)

@ -0,0 +1,61 @@
# A Python port of the MS knowledge base article Q157234
# "How to deal with localized and renamed user and group names"
# http://support.microsoft.com/default.aspx?kbid=157234
import sys
from win32net import NetUserModalsGet
from win32security import LookupAccountSid
import pywintypes
from ntsecuritycon import *
def LookupAliasFromRid(TargetComputer, Rid):
# Sid is the same regardless of machine, since the well-known
# BUILTIN domain is referenced.
sid = pywintypes.SID()
sid.Initialize(SECURITY_NT_AUTHORITY, 2)
for i, r in enumerate((SECURITY_BUILTIN_DOMAIN_RID, Rid)):
sid.SetSubAuthority(i, r)
name, domain, typ = LookupAccountSid(TargetComputer, sid)
return name
def LookupUserGroupFromRid(TargetComputer, Rid):
# get the account domain Sid on the target machine
# note: if you were looking up multiple sids based on the same
# account domain, only need to call this once.
umi2 = NetUserModalsGet(TargetComputer, 2)
domain_sid = umi2['domain_id']
SubAuthorityCount = domain_sid.GetSubAuthorityCount()
# create and init new sid with acct domain Sid + acct Rid
sid = pywintypes.SID()
sid.Initialize(domain_sid.GetSidIdentifierAuthority(),
SubAuthorityCount+1)
# copy existing subauthorities from account domain Sid into
# new Sid
for i in range(SubAuthorityCount):
sid.SetSubAuthority(i, domain_sid.GetSubAuthority(i))
# append Rid to new Sid
sid.SetSubAuthority(SubAuthorityCount, Rid)
name, domain, typ = LookupAccountSid(TargetComputer, sid)
return name
def main():
if len(sys.argv) == 2:
targetComputer = sys.argv[1]
else:
targetComputer = None
name = LookupUserGroupFromRid(targetComputer, DOMAIN_USER_RID_ADMIN)
print("'Administrator' user name = %s" % (name,))
name = LookupAliasFromRid(targetComputer, DOMAIN_ALIAS_RID_ADMINS)
print("'Administrators' local group/alias name = %s" % (name,))
if __name__=='__main__':
main()

@ -0,0 +1,8 @@
import win32security, win32event
evt = win32event.CreateEvent(None,0,0,None)
win32security.LsaRegisterPolicyChangeNotification(win32security.PolicyNotifyAuditEventsInformation, evt)
print("Waiting for you change Audit policy in Management console ...")
ret_code=win32event.WaitForSingleObject(evt,1000000000)
## should come back when you change Audit policy in Management console ...
print(ret_code)
win32security.LsaUnregisterPolicyChangeNotification(win32security.PolicyNotifyAuditEventsInformation, evt)

@ -0,0 +1,11 @@
import win32security
policy_handle = win32security.GetPolicyHandle('',win32security.POLICY_ALL_ACCESS)
privatedata='some sensitive data'
keyname='tmp'
win32security.LsaStorePrivateData(policy_handle,keyname,privatedata)
retrieveddata=win32security.LsaRetrievePrivateData(policy_handle,keyname)
assert retrieveddata==privatedata
## passing None deletes key
win32security.LsaStorePrivateData(policy_handle,keyname,None)
win32security.LsaClose(policy_handle)

@ -0,0 +1,25 @@
from ntsecuritycon import *
import win32api, win32security, winerror
# This is a Python implementation of win32api.GetDomainName()
def GetDomainName():
try:
tok = win32security.OpenThreadToken(win32api.GetCurrentThread(),
TOKEN_QUERY, 1)
except win32api.error as details:
if details[0] != winerror.ERROR_NO_TOKEN:
raise
# attempt to open the process token, since no thread token
# exists
tok = win32security.OpenProcessToken(win32api.GetCurrentProcess(),
TOKEN_QUERY)
sid, attr = win32security.GetTokenInformation(tok, TokenUser)
win32api.CloseHandle(tok)
name, dom, typ = win32security.LookupAccountSid(None, sid)
return dom
if __name__=='__main__':
print("Domain name is", GetDomainName())

@ -0,0 +1,37 @@
fname='h:\\tmp.reg'
import win32api, win32con, win32security, ntsecuritycon, pywintypes,os
## regsave will not overwrite a file
if os.path.isfile(fname):
os.remove(fname)
new_privs = ((win32security.LookupPrivilegeValue('',ntsecuritycon.SE_SECURITY_NAME),win32con.SE_PRIVILEGE_ENABLED),
(win32security.LookupPrivilegeValue('',ntsecuritycon.SE_TCB_NAME),win32con.SE_PRIVILEGE_ENABLED),
(win32security.LookupPrivilegeValue('',ntsecuritycon.SE_BACKUP_NAME),win32con.SE_PRIVILEGE_ENABLED),
(win32security.LookupPrivilegeValue('',ntsecuritycon.SE_RESTORE_NAME),win32con.SE_PRIVILEGE_ENABLED)
)
ph = win32api.GetCurrentProcess()
th = win32security.OpenProcessToken(ph,win32security.TOKEN_ALL_ACCESS|win32con.TOKEN_ADJUST_PRIVILEGES)
win32security.AdjustTokenPrivileges(th,0,new_privs)
my_sid = win32security.GetTokenInformation(th,ntsecuritycon.TokenUser)[0]
hklm=win32api.RegOpenKey(win32con.HKEY_LOCAL_MACHINE,None,0,win32con.KEY_ALL_ACCESS)
skey=win32api.RegOpenKey(hklm,'SYSTEM',0,win32con.KEY_ALL_ACCESS)
sa=pywintypes.SECURITY_ATTRIBUTES()
sd=pywintypes.SECURITY_DESCRIPTOR()
sa.SECURITY_DESCRIPTOR=sd
acl=pywintypes.ACL()
pwr_sid = win32security.LookupAccountName('','Power Users')[0]
acl.AddAccessAllowedAce(win32con.ACL_REVISION,win32con.GENERIC_READ|win32con.ACCESS_SYSTEM_SECURITY,my_sid)
sd.SetSecurityDescriptorDacl(1,acl,0)
sd.SetSecurityDescriptorOwner(pwr_sid,0)
sa.bInheritHandle=1
assert sa.SECURITY_DESCRIPTOR is sd
win32api.RegSaveKey(skey,fname,sa)

@ -0,0 +1,20 @@
import win32api, win32con, win32security, ntsecuritycon
new_privs = ((win32security.LookupPrivilegeValue('',ntsecuritycon.SE_SECURITY_NAME),win32con.SE_PRIVILEGE_ENABLED),
(win32security.LookupPrivilegeValue('',ntsecuritycon.SE_TCB_NAME),win32con.SE_PRIVILEGE_ENABLED)
)
ph = win32api.GetCurrentProcess()
th = win32security.OpenProcessToken(ph,win32security.TOKEN_ALL_ACCESS|win32con.TOKEN_ADJUST_PRIVILEGES)
win32security.AdjustTokenPrivileges(th,0,new_privs)
hkey=win32api.RegOpenKey(win32con.HKEY_LOCAL_MACHINE,None,0,win32con.KEY_ALL_ACCESS)
win32api.RegCreateKey(hkey,'SYSTEM\\NOTMP')
notmpkey=win32api.RegOpenKey(hkey,'SYSTEM\\notmp',0,win32con.ACCESS_SYSTEM_SECURITY)
tmp_sid = win32security.LookupAccountName('','tmp')[0]
sacl=win32security.ACL()
sacl.AddAuditAccessAce(win32security.ACL_REVISION,win32con.GENERIC_ALL,tmp_sid,1,1)
sd=win32security.SECURITY_DESCRIPTOR()
sd.SetSecurityDescriptorSacl(1,sacl,1)
win32api.RegSetKeySecurity(notmpkey,win32con.SACL_SECURITY_INFORMATION,sd)

@ -0,0 +1,7 @@
import pywintypes, win32security
sa=pywintypes.SECURITY_ATTRIBUTES()
tmp_sid=win32security.LookupAccountName('','tmp')[0]
sa.SetSecurityDescriptorOwner(tmp_sid,0)
sid=sa.SECURITY_DESCRIPTOR.GetSecurityDescriptorOwner()
print(win32security.LookupAccountSid('',sid))

@ -0,0 +1,316 @@
import win32security, ntsecuritycon, winnt
class Enum:
def __init__(self, *const_names):
"""Accepts variable number of constant names that can be found in either
win32security, ntsecuritycon, or winnt."""
for const_name in const_names:
try:
const_val=getattr(win32security,const_name)
except AttributeError:
try:
const_val=getattr(ntsecuritycon, const_name)
except AttributeError:
try:
const_val=getattr(winnt, const_name)
except AttributeError:
raise AttributeError('Constant "%s" not found in win32security, ntsecuritycon, or winnt.' %const_name)
setattr(self, const_name, const_val)
def lookup_name(self, const_val):
"""Looks up the name of a particular value."""
for k,v in self.__dict__.items():
if v==const_val:
return k
raise AttributeError('Value %s not found in enum' %const_val)
def lookup_flags(self, flags):
"""Returns the names of all recognized flags in input, and any flags not found in the enum."""
flag_names=[]
unknown_flags=flags
for k,v in self.__dict__.items():
if flags & v == v:
flag_names.append(k)
unknown_flags = unknown_flags & ~v
return flag_names, unknown_flags
TOKEN_INFORMATION_CLASS = Enum(
'TokenUser',
'TokenGroups',
'TokenPrivileges',
'TokenOwner',
'TokenPrimaryGroup',
'TokenDefaultDacl',
'TokenSource',
'TokenType',
'TokenImpersonationLevel',
'TokenStatistics',
'TokenRestrictedSids',
'TokenSessionId',
'TokenGroupsAndPrivileges',
'TokenSessionReference',
'TokenSandBoxInert',
'TokenAuditPolicy',
'TokenOrigin',
'TokenElevationType',
'TokenLinkedToken',
'TokenElevation',
'TokenHasRestrictions',
'TokenAccessInformation',
'TokenVirtualizationAllowed',
'TokenVirtualizationEnabled',
'TokenIntegrityLevel',
'TokenUIAccess',
'TokenMandatoryPolicy',
'TokenLogonSid')
TOKEN_TYPE = Enum(
'TokenPrimary',
'TokenImpersonation')
TOKEN_ELEVATION_TYPE = Enum(
'TokenElevationTypeDefault',
'TokenElevationTypeFull',
'TokenElevationTypeLimited')
POLICY_AUDIT_EVENT_TYPE = Enum(
'AuditCategorySystem',
'AuditCategoryLogon',
'AuditCategoryObjectAccess',
'AuditCategoryPrivilegeUse',
'AuditCategoryDetailedTracking',
'AuditCategoryPolicyChange',
'AuditCategoryAccountManagement',
'AuditCategoryDirectoryServiceAccess',
'AuditCategoryAccountLogon')
POLICY_INFORMATION_CLASS = Enum(
'PolicyAuditLogInformation',
'PolicyAuditEventsInformation',
'PolicyPrimaryDomainInformation',
'PolicyPdAccountInformation',
'PolicyAccountDomainInformation',
'PolicyLsaServerRoleInformation',
'PolicyReplicaSourceInformation',
'PolicyDefaultQuotaInformation',
'PolicyModificationInformation',
'PolicyAuditFullSetInformation',
'PolicyAuditFullQueryInformation',
'PolicyDnsDomainInformation')
POLICY_LSA_SERVER_ROLE = Enum(
'PolicyServerRoleBackup',
'PolicyServerRolePrimary')
## access modes for opening a policy handle - this is not a real enum
POLICY_ACCESS_MODES = Enum(
'POLICY_VIEW_LOCAL_INFORMATION',
'POLICY_VIEW_AUDIT_INFORMATION',
'POLICY_GET_PRIVATE_INFORMATION',
'POLICY_TRUST_ADMIN',
'POLICY_CREATE_ACCOUNT',
'POLICY_CREATE_SECRET',
'POLICY_CREATE_PRIVILEGE',
'POLICY_SET_DEFAULT_QUOTA_LIMITS',
'POLICY_SET_AUDIT_REQUIREMENTS',
'POLICY_AUDIT_LOG_ADMIN',
'POLICY_SERVER_ADMIN',
'POLICY_LOOKUP_NAMES',
'POLICY_NOTIFICATION',
'POLICY_ALL_ACCESS',
'POLICY_READ',
'POLICY_WRITE',
'POLICY_EXECUTE')
## EventAuditingOptions flags - not a real enum
POLICY_AUDIT_EVENT_OPTIONS_FLAGS = Enum(
'POLICY_AUDIT_EVENT_UNCHANGED',
'POLICY_AUDIT_EVENT_SUCCESS',
'POLICY_AUDIT_EVENT_FAILURE',
'POLICY_AUDIT_EVENT_NONE')
# AceType in ACE_HEADER - not a real enum
ACE_TYPE = Enum(
'ACCESS_MIN_MS_ACE_TYPE',
'ACCESS_ALLOWED_ACE_TYPE',
'ACCESS_DENIED_ACE_TYPE',
'SYSTEM_AUDIT_ACE_TYPE',
'SYSTEM_ALARM_ACE_TYPE',
'ACCESS_MAX_MS_V2_ACE_TYPE',
'ACCESS_ALLOWED_COMPOUND_ACE_TYPE',
'ACCESS_MAX_MS_V3_ACE_TYPE',
'ACCESS_MIN_MS_OBJECT_ACE_TYPE',
'ACCESS_ALLOWED_OBJECT_ACE_TYPE',
'ACCESS_DENIED_OBJECT_ACE_TYPE',
'SYSTEM_AUDIT_OBJECT_ACE_TYPE',
'SYSTEM_ALARM_OBJECT_ACE_TYPE',
'ACCESS_MAX_MS_OBJECT_ACE_TYPE',
'ACCESS_MAX_MS_V4_ACE_TYPE',
'ACCESS_MAX_MS_ACE_TYPE',
'ACCESS_ALLOWED_CALLBACK_ACE_TYPE',
'ACCESS_DENIED_CALLBACK_ACE_TYPE',
'ACCESS_ALLOWED_CALLBACK_OBJECT_ACE_TYPE',
'ACCESS_DENIED_CALLBACK_OBJECT_ACE_TYPE',
'SYSTEM_AUDIT_CALLBACK_ACE_TYPE',
'SYSTEM_ALARM_CALLBACK_ACE_TYPE',
'SYSTEM_AUDIT_CALLBACK_OBJECT_ACE_TYPE',
'SYSTEM_ALARM_CALLBACK_OBJECT_ACE_TYPE',
'SYSTEM_MANDATORY_LABEL_ACE_TYPE',
'ACCESS_MAX_MS_V5_ACE_TYPE')
#bit flags for AceFlags - not a real enum
ACE_FLAGS = Enum(
'CONTAINER_INHERIT_ACE',
'FAILED_ACCESS_ACE_FLAG',
'INHERIT_ONLY_ACE',
'INHERITED_ACE',
'NO_PROPAGATE_INHERIT_ACE',
'OBJECT_INHERIT_ACE',
'SUCCESSFUL_ACCESS_ACE_FLAG',
'NO_INHERITANCE',
'SUB_CONTAINERS_AND_OBJECTS_INHERIT',
'SUB_CONTAINERS_ONLY_INHERIT',
'SUB_OBJECTS_ONLY_INHERIT')
# used in SetEntriesInAcl - very similar to ACE_TYPE
ACCESS_MODE = Enum(
'NOT_USED_ACCESS',
'GRANT_ACCESS',
'SET_ACCESS',
'DENY_ACCESS',
'REVOKE_ACCESS',
'SET_AUDIT_SUCCESS',
'SET_AUDIT_FAILURE')
# Bit flags in PSECURITY_DESCRIPTOR->Control - not a real enum
SECURITY_DESCRIPTOR_CONTROL_FLAGS = Enum(
'SE_DACL_AUTO_INHERITED', ## win2k and up
'SE_SACL_AUTO_INHERITED', ## win2k and up
'SE_DACL_PROTECTED', ## win2k and up
'SE_SACL_PROTECTED', ## win2k and up
'SE_DACL_DEFAULTED',
'SE_DACL_PRESENT',
'SE_GROUP_DEFAULTED',
'SE_OWNER_DEFAULTED',
'SE_SACL_PRESENT',
'SE_SELF_RELATIVE',
'SE_SACL_DEFAULTED')
# types of SID
SID_NAME_USE = Enum(
'SidTypeUser',
'SidTypeGroup',
'SidTypeDomain',
'SidTypeAlias',
'SidTypeWellKnownGroup',
'SidTypeDeletedAccount',
'SidTypeInvalid',
'SidTypeUnknown',
'SidTypeComputer',
'SidTypeLabel')
## bit flags, not a real enum
TOKEN_ACCESS_PRIVILEGES = Enum(
'TOKEN_ADJUST_DEFAULT',
'TOKEN_ADJUST_GROUPS',
'TOKEN_ADJUST_PRIVILEGES',
'TOKEN_ALL_ACCESS',
'TOKEN_ASSIGN_PRIMARY',
'TOKEN_DUPLICATE',
'TOKEN_EXECUTE',
'TOKEN_IMPERSONATE',
'TOKEN_QUERY',
'TOKEN_QUERY_SOURCE',
'TOKEN_READ',
'TOKEN_WRITE')
SECURITY_IMPERSONATION_LEVEL = Enum(
'SecurityAnonymous',
'SecurityIdentification',
'SecurityImpersonation',
'SecurityDelegation')
POLICY_SERVER_ENABLE_STATE = Enum(
'PolicyServerEnabled',
'PolicyServerDisabled')
POLICY_NOTIFICATION_INFORMATION_CLASS = Enum(
'PolicyNotifyAuditEventsInformation',
'PolicyNotifyAccountDomainInformation',
'PolicyNotifyServerRoleInformation',
'PolicyNotifyDnsDomainInformation',
'PolicyNotifyDomainEfsInformation',
'PolicyNotifyDomainKerberosTicketInformation',
'PolicyNotifyMachineAccountPasswordInformation')
TRUSTED_INFORMATION_CLASS = Enum(
'TrustedDomainNameInformation',
'TrustedControllersInformation',
'TrustedPosixOffsetInformation',
'TrustedPasswordInformation',
'TrustedDomainInformationBasic',
'TrustedDomainInformationEx',
'TrustedDomainAuthInformation',
'TrustedDomainFullInformation',
'TrustedDomainAuthInformationInternal',
'TrustedDomainFullInformationInternal',
'TrustedDomainInformationEx2Internal',
'TrustedDomainFullInformation2Internal')
TRUSTEE_FORM = Enum(
'TRUSTEE_IS_SID',
'TRUSTEE_IS_NAME',
'TRUSTEE_BAD_FORM',
'TRUSTEE_IS_OBJECTS_AND_SID',
'TRUSTEE_IS_OBJECTS_AND_NAME')
TRUSTEE_TYPE = Enum(
'TRUSTEE_IS_UNKNOWN',
'TRUSTEE_IS_USER',
'TRUSTEE_IS_GROUP',
'TRUSTEE_IS_DOMAIN',
'TRUSTEE_IS_ALIAS',
'TRUSTEE_IS_WELL_KNOWN_GROUP',
'TRUSTEE_IS_DELETED',
'TRUSTEE_IS_INVALID',
'TRUSTEE_IS_COMPUTER')
## SE_OBJECT_TYPE - securable objects
SE_OBJECT_TYPE = Enum(
'SE_UNKNOWN_OBJECT_TYPE',
'SE_FILE_OBJECT',
'SE_SERVICE',
'SE_PRINTER',
'SE_REGISTRY_KEY',
'SE_LMSHARE',
'SE_KERNEL_OBJECT',
'SE_WINDOW_OBJECT',
'SE_DS_OBJECT',
'SE_DS_OBJECT_ALL',
'SE_PROVIDER_DEFINED_OBJECT',
'SE_WMIGUID_OBJECT',
'SE_REGISTRY_WOW64_32KEY')
PRIVILEGE_FLAGS = Enum(
'SE_PRIVILEGE_ENABLED_BY_DEFAULT',
'SE_PRIVILEGE_ENABLED',
'SE_PRIVILEGE_USED_FOR_ACCESS')
# Group flags used with TokenGroups
TOKEN_GROUP_ATTRIBUTES = Enum(
'SE_GROUP_MANDATORY',
'SE_GROUP_ENABLED_BY_DEFAULT',
'SE_GROUP_ENABLED',
'SE_GROUP_OWNER',
'SE_GROUP_USE_FOR_DENY_ONLY',
'SE_GROUP_INTEGRITY',
'SE_GROUP_INTEGRITY_ENABLED',
'SE_GROUP_LOGON_ID',
'SE_GROUP_RESOURCE')
# Privilege flags returned by TokenPrivileges
TOKEN_PRIVILEGE_ATTRIBUTES = Enum(
'SE_PRIVILEGE_ENABLED_BY_DEFAULT',
'SE_PRIVILEGE_ENABLED',
'SE_PRIVILEGE_REMOVED',
'SE_PRIVILEGE_USED_FOR_ACCESS')

@ -0,0 +1,58 @@
import win32security,win32file,win32api,ntsecuritycon,win32con, os
from win32security import ACL_REVISION_DS, CONTAINER_INHERIT_ACE, OBJECT_INHERIT_ACE, \
PROTECTED_DACL_SECURITY_INFORMATION, DACL_SECURITY_INFORMATION, SACL_SECURITY_INFORMATION, \
OWNER_SECURITY_INFORMATION, GROUP_SECURITY_INFORMATION, SE_FILE_OBJECT
## SE_SECURITY_NAME needed to access SACL, SE_RESTORE_NAME needed to change owner to someone other than yourself
new_privs = ((win32security.LookupPrivilegeValue('',ntsecuritycon.SE_SECURITY_NAME),win32con.SE_PRIVILEGE_ENABLED),
(win32security.LookupPrivilegeValue('',ntsecuritycon.SE_RESTORE_NAME),win32con.SE_PRIVILEGE_ENABLED),
)
ph = win32api.GetCurrentProcess()
th = win32security.OpenProcessToken(ph,win32security.TOKEN_ALL_ACCESS|win32con.TOKEN_ADJUST_PRIVILEGES)
modified_privs=win32security.AdjustTokenPrivileges(th,0,new_privs)
## look up a few sids that should be available on most systems
my_sid = win32security.GetTokenInformation(th,ntsecuritycon.TokenUser)[0]
pwr_sid = win32security.LookupAccountName('','Power Users')[0]
admin_sid = win32security.LookupAccountName('','Administrators')[0]
everyone_sid=win32security.LookupAccountName('','EveryOne')[0]
## create a dir and set security so Everyone has read permissions, and all files and subdirs inherit its ACLs
temp_dir=win32api.GetTempPath()
dir_name=win32api.GetTempFileName(temp_dir,'sfa')[0]
os.remove(dir_name)
os.mkdir(dir_name)
dir_dacl=win32security.ACL()
dir_dacl.AddAccessAllowedAceEx(ACL_REVISION_DS, CONTAINER_INHERIT_ACE|OBJECT_INHERIT_ACE, win32con.GENERIC_READ, everyone_sid)
## make sure current user has permissions on dir
dir_dacl.AddAccessAllowedAceEx(ACL_REVISION_DS, CONTAINER_INHERIT_ACE|OBJECT_INHERIT_ACE, win32con.GENERIC_ALL, my_sid)
## keep dir from inheriting any permissions so it only has ACEs explicitely set here
win32security.SetNamedSecurityInfo(dir_name, SE_FILE_OBJECT,
OWNER_SECURITY_INFORMATION|GROUP_SECURITY_INFORMATION|DACL_SECURITY_INFORMATION|PROTECTED_DACL_SECURITY_INFORMATION,
pwr_sid, pwr_sid, dir_dacl, None)
## Create a file in the dir and add some specific permissions to it
fname=win32api.GetTempFileName(dir_name,'sfa')[0]
print(fname)
file_sd=win32security.GetNamedSecurityInfo(fname, SE_FILE_OBJECT, DACL_SECURITY_INFORMATION|SACL_SECURITY_INFORMATION)
file_dacl=file_sd.GetSecurityDescriptorDacl()
file_sacl=file_sd.GetSecurityDescriptorSacl()
if file_dacl is None:
file_dacl=win32security.ACL()
if file_sacl is None:
file_sacl=win32security.ACL()
file_dacl.AddAccessDeniedAce(file_dacl.GetAclRevision(),win32con.DELETE,admin_sid)
file_dacl.AddAccessDeniedAce(file_dacl.GetAclRevision(),win32con.DELETE,my_sid)
file_dacl.AddAccessAllowedAce(file_dacl.GetAclRevision(),win32con.GENERIC_ALL,pwr_sid)
file_sacl.AddAuditAccessAce(file_dacl.GetAclRevision(),win32con.GENERIC_ALL,my_sid,True,True)
win32security.SetNamedSecurityInfo(fname, SE_FILE_OBJECT,
DACL_SECURITY_INFORMATION|SACL_SECURITY_INFORMATION,
None, None, file_dacl, file_sacl)
win32security.AdjustTokenPrivileges(th, 0, modified_privs)

@ -0,0 +1,38 @@
fname=r'h:\tmp.txt'
import win32security,win32file,win32api,ntsecuritycon,win32con
new_privs = ((win32security.LookupPrivilegeValue('',ntsecuritycon.SE_SECURITY_NAME),win32con.SE_PRIVILEGE_ENABLED),
(win32security.LookupPrivilegeValue('',ntsecuritycon.SE_SHUTDOWN_NAME),win32con.SE_PRIVILEGE_ENABLED),
(win32security.LookupPrivilegeValue('',ntsecuritycon.SE_TCB_NAME),win32con.SE_PRIVILEGE_ENABLED),
(win32security.LookupPrivilegeValue('',ntsecuritycon.SE_RESTORE_NAME),win32con.SE_PRIVILEGE_ENABLED),
(win32security.LookupPrivilegeValue('',ntsecuritycon.SE_TAKE_OWNERSHIP_NAME),win32con.SE_PRIVILEGE_ENABLED),
(win32security.LookupPrivilegeValue('',ntsecuritycon.SE_CREATE_PERMANENT_NAME),win32con.SE_PRIVILEGE_ENABLED),
(win32security.LookupPrivilegeValue('','SeEnableDelegationPrivilege'),win32con.SE_PRIVILEGE_ENABLED) ##doesn't seem to be in ntsecuritycon.py ?
)
ph = win32api.GetCurrentProcess()
th = win32security.OpenProcessToken(ph,win32security.TOKEN_ALL_ACCESS|win32con.TOKEN_ADJUST_PRIVILEGES)
win32security.AdjustTokenPrivileges(th,0,new_privs)
all_security_info = \
win32security.OWNER_SECURITY_INFORMATION|win32security.GROUP_SECURITY_INFORMATION| \
win32security.DACL_SECURITY_INFORMATION|win32security.SACL_SECURITY_INFORMATION
sd=win32security.GetFileSecurity(fname,all_security_info)
old_dacl=sd.GetSecurityDescriptorDacl()
old_sacl=sd.GetSecurityDescriptorSacl()
old_group=sd.GetSecurityDescriptorGroup()
new_sd=win32security.SECURITY_DESCRIPTOR()
print("relative, valid, size: ",new_sd.IsSelfRelative(), new_sd.IsValid(), new_sd.GetLength())
my_sid = win32security.GetTokenInformation(th,ntsecuritycon.TokenUser)[0]
tmp_sid = win32security.LookupAccountName('','tmp')[0]
new_sd.SetSecurityDescriptorSacl(1,old_sacl,1)
new_sd.SetSecurityDescriptorDacl(1,old_dacl,1)
new_sd.SetSecurityDescriptorOwner(tmp_sid,0)
new_sd.SetSecurityDescriptorGroup(old_group,0)
win32security.SetFileSecurity(fname,all_security_info,new_sd)

@ -0,0 +1,17 @@
import win32security,win32file,win32api,ntsecuritycon,win32con
policy_handle = win32security.GetPolicyHandle('rupole',win32security.POLICY_ALL_ACCESS)
event_audit_info=win32security.LsaQueryInformationPolicy(policy_handle,win32security.PolicyAuditEventsInformation)
print(event_audit_info)
new_audit_info=list(event_audit_info[1])
new_audit_info[win32security.AuditCategoryPolicyChange]= \
win32security.POLICY_AUDIT_EVENT_SUCCESS|win32security.POLICY_AUDIT_EVENT_FAILURE
new_audit_info[win32security.AuditCategoryAccountLogon]= \
win32security.POLICY_AUDIT_EVENT_SUCCESS|win32security.POLICY_AUDIT_EVENT_FAILURE
new_audit_info[win32security.AuditCategoryLogon]= \
win32security.POLICY_AUDIT_EVENT_SUCCESS|win32security.POLICY_AUDIT_EVENT_FAILURE
win32security.LsaSetInformationPolicy(policy_handle, win32security.PolicyAuditEventsInformation, (1,new_audit_info))
win32security.LsaClose(policy_handle)

@ -0,0 +1,67 @@
import win32security,win32api,win32con, win32process
## You need SE_RESTORE_NAME to be able to set the owner of a security descriptor to anybody
## other than yourself or your primary group. Most admin logins don't have it by default, so
## enabling it may fail
new_privs = ((win32security.LookupPrivilegeValue('',win32security.SE_SECURITY_NAME),win32con.SE_PRIVILEGE_ENABLED),
(win32security.LookupPrivilegeValue('',win32security.SE_TCB_NAME),win32con.SE_PRIVILEGE_ENABLED),
(win32security.LookupPrivilegeValue('',win32security.SE_SHUTDOWN_NAME),win32con.SE_PRIVILEGE_ENABLED),
(win32security.LookupPrivilegeValue('',win32security.SE_RESTORE_NAME),win32con.SE_PRIVILEGE_ENABLED),
(win32security.LookupPrivilegeValue('',win32security.SE_TAKE_OWNERSHIP_NAME),win32con.SE_PRIVILEGE_ENABLED),
(win32security.LookupPrivilegeValue('',win32security.SE_CREATE_PERMANENT_NAME),win32con.SE_PRIVILEGE_ENABLED),
(win32security.LookupPrivilegeValue('',win32security.SE_ENABLE_DELEGATION_NAME),win32con.SE_PRIVILEGE_ENABLED),
(win32security.LookupPrivilegeValue('',win32security.SE_CHANGE_NOTIFY_NAME),win32con.SE_PRIVILEGE_ENABLED),
(win32security.LookupPrivilegeValue('',win32security.SE_DEBUG_NAME),win32con.SE_PRIVILEGE_ENABLED),
(win32security.LookupPrivilegeValue('',win32security.SE_PROF_SINGLE_PROCESS_NAME),win32con.SE_PRIVILEGE_ENABLED),
(win32security.LookupPrivilegeValue('',win32security.SE_SYSTEM_PROFILE_NAME),win32con.SE_PRIVILEGE_ENABLED),
(win32security.LookupPrivilegeValue('',win32security.SE_LOCK_MEMORY_NAME),win32con.SE_PRIVILEGE_ENABLED)
)
all_info=win32security.OWNER_SECURITY_INFORMATION|win32security.GROUP_SECURITY_INFORMATION| \
win32security.DACL_SECURITY_INFORMATION|win32security.SACL_SECURITY_INFORMATION
pid=win32api.GetCurrentProcessId()
ph=win32api.OpenProcess(win32con.PROCESS_ALL_ACCESS,0,pid)
## PROCESS_ALL_ACCESS does not contain ACCESS_SYSTEM_SECURITY (neccessy to do SACLs)
th = win32security.OpenProcessToken(ph,win32security.TOKEN_ALL_ACCESS) ##win32con.TOKEN_ADJUST_PRIVILEGES)
old_privs=win32security.AdjustTokenPrivileges(th,0,new_privs)
my_sid = win32security.GetTokenInformation(th,win32security.TokenUser)[0]
pwr_sid=win32security.LookupAccountName('','Power Users')[0]
## reopen process with ACCESS_SYSTEM_SECURITY now that sufficent privs are enabled
ph=win32api.OpenProcess(win32con.PROCESS_ALL_ACCESS|win32con.ACCESS_SYSTEM_SECURITY,0,pid)
sd=win32security.GetKernelObjectSecurity(ph,all_info)
dacl=sd.GetSecurityDescriptorDacl()
if dacl is None:
dacl=win32security.ACL()
sacl=sd.GetSecurityDescriptorSacl()
if sacl is None:
sacl=win32security.ACL()
dacl_ace_cnt=dacl.GetAceCount()
sacl_ace_cnt=sacl.GetAceCount()
dacl.AddAccessAllowedAce(dacl.GetAclRevision(),win32con.ACCESS_SYSTEM_SECURITY|win32con.WRITE_DAC,my_sid)
sacl.AddAuditAccessAce(sacl.GetAclRevision(),win32con.GENERIC_ALL,my_sid,1,1)
sd.SetSecurityDescriptorDacl(1,dacl,0)
sd.SetSecurityDescriptorSacl(1,sacl,0)
sd.SetSecurityDescriptorGroup(pwr_sid,0)
sd.SetSecurityDescriptorOwner(pwr_sid,0)
win32security.SetKernelObjectSecurity(ph,all_info,sd)
new_sd=win32security.GetKernelObjectSecurity(ph,all_info)
if new_sd.GetSecurityDescriptorDacl().GetAceCount()!=dacl_ace_cnt+1:
print('New dacl doesn''t contain extra ace ????')
if new_sd.GetSecurityDescriptorSacl().GetAceCount()!=sacl_ace_cnt+1:
print('New Sacl doesn''t contain extra ace ????')
if win32security.LookupAccountSid('',new_sd.GetSecurityDescriptorOwner())[0]!='Power Users':
print('Owner not successfully set to Power Users !!!!!')
if win32security.LookupAccountSid('',new_sd.GetSecurityDescriptorGroup())[0]!='Power Users':
print('Group not successfully set to Power Users !!!!!')
sd.SetSecurityDescriptorSacl(0,None,0)
win32security.SetKernelObjectSecurity(ph, win32security.SACL_SECURITY_INFORMATION, sd)
new_sd_1=win32security.GetKernelObjectSecurity(ph,win32security.SACL_SECURITY_INFORMATION)
if new_sd_1.GetSecurityDescriptorSacl() is not None:
print('Unable to set Sacl to NULL !!!!!!!!')

@ -0,0 +1,60 @@
import win32security,win32api,win32con, win32process
fname, tmp = win32api.GetTempFileName(win32api.GetTempPath(),'tmp')
print(fname)
## You need SE_RESTORE_NAME to be able to set the owner of a security descriptor to anybody
## other than yourself or your primary group. Most admin logins don't have it by default, so
## enabling it may fail
new_privs = ((win32security.LookupPrivilegeValue('',win32security.SE_SECURITY_NAME),win32con.SE_PRIVILEGE_ENABLED),
(win32security.LookupPrivilegeValue('',win32security.SE_TCB_NAME),win32con.SE_PRIVILEGE_ENABLED),
(win32security.LookupPrivilegeValue('',win32security.SE_SHUTDOWN_NAME),win32con.SE_PRIVILEGE_ENABLED),
(win32security.LookupPrivilegeValue('',win32security.SE_RESTORE_NAME),win32con.SE_PRIVILEGE_ENABLED),
(win32security.LookupPrivilegeValue('',win32security.SE_TAKE_OWNERSHIP_NAME),win32con.SE_PRIVILEGE_ENABLED),
(win32security.LookupPrivilegeValue('',win32security.SE_CREATE_PERMANENT_NAME),win32con.SE_PRIVILEGE_ENABLED),
(win32security.LookupPrivilegeValue('',win32security.SE_ENABLE_DELEGATION_NAME),win32con.SE_PRIVILEGE_ENABLED),
(win32security.LookupPrivilegeValue('',win32security.SE_CHANGE_NOTIFY_NAME),win32con.SE_PRIVILEGE_ENABLED),
(win32security.LookupPrivilegeValue('',win32security.SE_DEBUG_NAME),win32con.SE_PRIVILEGE_ENABLED),
(win32security.LookupPrivilegeValue('',win32security.SE_PROF_SINGLE_PROCESS_NAME),win32con.SE_PRIVILEGE_ENABLED),
(win32security.LookupPrivilegeValue('',win32security.SE_SYSTEM_PROFILE_NAME),win32con.SE_PRIVILEGE_ENABLED),
(win32security.LookupPrivilegeValue('',win32security.SE_LOCK_MEMORY_NAME),win32con.SE_PRIVILEGE_ENABLED)
)
all_info=win32security.OWNER_SECURITY_INFORMATION|win32security.GROUP_SECURITY_INFORMATION| \
win32security.DACL_SECURITY_INFORMATION|win32security.SACL_SECURITY_INFORMATION
ph=win32process.GetCurrentProcess()
th = win32security.OpenProcessToken(ph,win32security.TOKEN_ALL_ACCESS) ##win32con.TOKEN_ADJUST_PRIVILEGES)
win32security.AdjustTokenPrivileges(th,0,new_privs)
my_sid = win32security.GetTokenInformation(th,win32security.TokenUser)[0]
pwr_sid=win32security.LookupAccountName('','Power Users')[0]
sd=win32security.GetNamedSecurityInfo(fname,win32security.SE_FILE_OBJECT,all_info)
dacl=sd.GetSecurityDescriptorDacl()
if dacl is None:
dacl=win32security.ACL()
sacl=sd.GetSecurityDescriptorSacl()
if sacl is None:
sacl=win32security.ACL()
dacl_ace_cnt=dacl.GetAceCount()
sacl_ace_cnt=sacl.GetAceCount()
dacl.AddAccessAllowedAce(dacl.GetAclRevision(),win32con.ACCESS_SYSTEM_SECURITY|win32con.WRITE_DAC,my_sid)
sacl.AddAuditAccessAce(sacl.GetAclRevision(),win32con.GENERIC_ALL,my_sid,1,1)
win32security.SetNamedSecurityInfo(fname,win32security.SE_FILE_OBJECT,all_info,pwr_sid, pwr_sid, dacl, sacl)
new_sd=win32security.GetNamedSecurityInfo(fname,win32security.SE_FILE_OBJECT,all_info)
## could do additional checking to make sure added ACE contains expected info
if new_sd.GetSecurityDescriptorDacl().GetAceCount()!=dacl_ace_cnt+1:
print('New dacl doesn''t contain extra ace ????')
if new_sd.GetSecurityDescriptorSacl().GetAceCount()!=sacl_ace_cnt+1:
print('New Sacl doesn''t contain extra ace ????')
if win32security.LookupAccountSid('',new_sd.GetSecurityDescriptorOwner())[0]!='Power Users':
print('Owner not successfully set to Power Users !!!!!')
if win32security.LookupAccountSid('',new_sd.GetSecurityDescriptorGroup())[0]!='Power Users':
print('Group not successfully set to Power Users !!!!!')
win32security.SetNamedSecurityInfo(fname,win32security.SE_FILE_OBJECT,win32security.SACL_SECURITY_INFORMATION, None, None, None, None)
new_sd_1=win32security.GetNamedSecurityInfo(fname,win32security.SE_FILE_OBJECT,win32security.SACL_SECURITY_INFORMATION)
if new_sd_1.GetSecurityDescriptorSacl() is not None:
print('Unable to set Sacl to NULL !!!!!!!!')

@ -0,0 +1,61 @@
import win32security,win32api,win32con, win32process
## You need SE_RESTORE_NAME to be able to set the owner of a security descriptor to anybody
## other than yourself or your primary group. Most admin logins don't have it by default, so
## enabling it may fail
new_privs = ((win32security.LookupPrivilegeValue('',win32security.SE_SECURITY_NAME),win32con.SE_PRIVILEGE_ENABLED),
(win32security.LookupPrivilegeValue('',win32security.SE_TCB_NAME),win32con.SE_PRIVILEGE_ENABLED),
(win32security.LookupPrivilegeValue('',win32security.SE_SHUTDOWN_NAME),win32con.SE_PRIVILEGE_ENABLED),
(win32security.LookupPrivilegeValue('',win32security.SE_RESTORE_NAME),win32con.SE_PRIVILEGE_ENABLED),
(win32security.LookupPrivilegeValue('',win32security.SE_TAKE_OWNERSHIP_NAME),win32con.SE_PRIVILEGE_ENABLED),
(win32security.LookupPrivilegeValue('',win32security.SE_CREATE_PERMANENT_NAME),win32con.SE_PRIVILEGE_ENABLED),
(win32security.LookupPrivilegeValue('',win32security.SE_ENABLE_DELEGATION_NAME),win32con.SE_PRIVILEGE_ENABLED),
(win32security.LookupPrivilegeValue('',win32security.SE_CHANGE_NOTIFY_NAME),win32con.SE_PRIVILEGE_ENABLED),
(win32security.LookupPrivilegeValue('',win32security.SE_DEBUG_NAME),win32con.SE_PRIVILEGE_ENABLED),
(win32security.LookupPrivilegeValue('',win32security.SE_PROF_SINGLE_PROCESS_NAME),win32con.SE_PRIVILEGE_ENABLED),
(win32security.LookupPrivilegeValue('',win32security.SE_SYSTEM_PROFILE_NAME),win32con.SE_PRIVILEGE_ENABLED),
(win32security.LookupPrivilegeValue('',win32security.SE_LOCK_MEMORY_NAME),win32con.SE_PRIVILEGE_ENABLED)
)
all_info=win32security.OWNER_SECURITY_INFORMATION|win32security.GROUP_SECURITY_INFORMATION| \
win32security.DACL_SECURITY_INFORMATION|win32security.SACL_SECURITY_INFORMATION
pid=win32api.GetCurrentProcessId()
ph=win32api.OpenProcess(win32con.PROCESS_ALL_ACCESS,0,pid)
## PROCESS_ALL_ACCESS does not contain ACCESS_SYSTEM_SECURITY (neccessy to do SACLs)
th = win32security.OpenProcessToken(ph,win32security.TOKEN_ALL_ACCESS) ##win32con.TOKEN_ADJUST_PRIVILEGES)
old_privs=win32security.AdjustTokenPrivileges(th,0,new_privs)
my_sid = win32security.GetTokenInformation(th,win32security.TokenUser)[0]
pwr_sid=win32security.LookupAccountName('','Power Users')[0]
## reopen process with ACCESS_SYSTEM_SECURITY now that sufficent privs are enabled
ph=win32api.OpenProcess(win32con.PROCESS_ALL_ACCESS|win32con.ACCESS_SYSTEM_SECURITY,0,pid)
sd=win32security.GetSecurityInfo(ph,win32security.SE_KERNEL_OBJECT,all_info)
dacl=sd.GetSecurityDescriptorDacl()
if dacl is None:
dacl=win32security.ACL()
sacl=sd.GetSecurityDescriptorSacl()
if sacl is None:
sacl=win32security.ACL()
dacl_ace_cnt=dacl.GetAceCount()
sacl_ace_cnt=sacl.GetAceCount()
dacl.AddAccessAllowedAce(dacl.GetAclRevision(),win32con.ACCESS_SYSTEM_SECURITY|win32con.WRITE_DAC,my_sid)
sacl.AddAuditAccessAce(sacl.GetAclRevision(),win32con.GENERIC_ALL,my_sid,1,1)
win32security.SetSecurityInfo(ph,win32security.SE_KERNEL_OBJECT,all_info,pwr_sid, pwr_sid, dacl, sacl)
new_sd=win32security.GetSecurityInfo(ph,win32security.SE_KERNEL_OBJECT,all_info)
if new_sd.GetSecurityDescriptorDacl().GetAceCount()!=dacl_ace_cnt+1:
print('New dacl doesn''t contain extra ace ????')
if new_sd.GetSecurityDescriptorSacl().GetAceCount()!=sacl_ace_cnt+1:
print('New Sacl doesn''t contain extra ace ????')
if win32security.LookupAccountSid('',new_sd.GetSecurityDescriptorOwner())[0]!='Power Users':
print('Owner not successfully set to Power Users !!!!!')
if win32security.LookupAccountSid('',new_sd.GetSecurityDescriptorGroup())[0]!='Power Users':
print('Group not successfully set to Power Users !!!!!')
win32security.SetSecurityInfo(ph,win32security.SE_KERNEL_OBJECT,win32security.SACL_SECURITY_INFORMATION, None, None, None, None)
new_sd_1=win32security.GetSecurityInfo(ph,win32security.SE_KERNEL_OBJECT,win32security.SACL_SECURITY_INFORMATION)
if new_sd_1.GetSecurityDescriptorSacl() is not None:
print('Unable to set Sacl to NULL !!!!!!!!')

@ -0,0 +1,42 @@
import win32security,win32api,win32con, win32process
new_privs = ((win32security.LookupPrivilegeValue('',win32security.SE_SECURITY_NAME),win32con.SE_PRIVILEGE_ENABLED),
(win32security.LookupPrivilegeValue('',win32security.SE_TCB_NAME),win32con.SE_PRIVILEGE_ENABLED),
(win32security.LookupPrivilegeValue('',win32security.SE_SHUTDOWN_NAME),win32con.SE_PRIVILEGE_ENABLED),
(win32security.LookupPrivilegeValue('',win32security.SE_RESTORE_NAME),win32con.SE_PRIVILEGE_ENABLED),
(win32security.LookupPrivilegeValue('',win32security.SE_TAKE_OWNERSHIP_NAME),win32con.SE_PRIVILEGE_ENABLED),
(win32security.LookupPrivilegeValue('',win32security.SE_CREATE_PERMANENT_NAME),win32con.SE_PRIVILEGE_ENABLED),
(win32security.LookupPrivilegeValue('',win32security.SE_ENABLE_DELEGATION_NAME),win32con.SE_PRIVILEGE_ENABLED),
(win32security.LookupPrivilegeValue('',win32security.SE_CHANGE_NOTIFY_NAME),win32con.SE_PRIVILEGE_ENABLED),
(win32security.LookupPrivilegeValue('',win32security.SE_DEBUG_NAME),win32con.SE_PRIVILEGE_ENABLED),
(win32security.LookupPrivilegeValue('',win32security.SE_PROF_SINGLE_PROCESS_NAME),win32con.SE_PRIVILEGE_ENABLED),
(win32security.LookupPrivilegeValue('',win32security.SE_SYSTEM_PROFILE_NAME),win32con.SE_PRIVILEGE_ENABLED),
(win32security.LookupPrivilegeValue('',win32security.SE_LOCK_MEMORY_NAME),win32con.SE_PRIVILEGE_ENABLED)
)
all_info=win32security.OWNER_SECURITY_INFORMATION|win32security.GROUP_SECURITY_INFORMATION| \
win32security.DACL_SECURITY_INFORMATION|win32security.SACL_SECURITY_INFORMATION
info=win32security.OWNER_SECURITY_INFORMATION|win32security.GROUP_SECURITY_INFORMATION|win32security.DACL_SECURITY_INFORMATION
ph=win32process.GetCurrentProcess()
th = win32security.OpenProcessToken(ph,win32security.TOKEN_ALL_ACCESS) ##win32con.TOKEN_ADJUST_PRIVILEGES)
win32security.AdjustTokenPrivileges(th,0,new_privs)
my_sid = win32security.GetTokenInformation(th,win32security.TokenUser)[0]
pwr_sid=win32security.LookupAccountName('','Power Users')[0]
h=win32process.GetProcessWindowStation()
sd=win32security.GetUserObjectSecurity(h,info)
dacl=sd.GetSecurityDescriptorDacl()
ace_cnt=dacl.GetAceCount()
dacl.AddAccessAllowedAce(dacl.GetAclRevision(),win32con.ACCESS_SYSTEM_SECURITY|win32con.WRITE_DAC,my_sid)
sd.SetSecurityDescriptorDacl(1,dacl,0)
sd.SetSecurityDescriptorGroup(pwr_sid,0)
sd.SetSecurityDescriptorOwner(pwr_sid,0)
win32security.SetUserObjectSecurity(h,info,sd)
new_sd=win32security.GetUserObjectSecurity(h,info)
assert new_sd.GetSecurityDescriptorDacl().GetAceCount()==ace_cnt+1,'Did not add an ace to the Dacl !!!!!!'
assert win32security.LookupAccountSid('',new_sd.GetSecurityDescriptorOwner())[0]=='Power Users','Owner not successfully set to Power Users !!!!!'
assert win32security.LookupAccountSid('',new_sd.GetSecurityDescriptorGroup())[0]=='Power Users','Group not successfully set to Power Users !!!!!'

@ -0,0 +1,146 @@
"""
Fetches a URL from a web-server supporting NTLM authentication
eg, IIS.
If no arguments are specified, a default of http://localhost/localstart.asp
is used. This script does follow simple 302 redirections, so pointing at the
root of an IIS server is should work.
"""
import sys
import urllib.request, urllib.parse, urllib.error
import http.client
import urllib.parse
from base64 import encodestring, decodestring
from sspi import ClientAuth
import optparse # sorry, this demo needs 2.3+
options = None # set to optparse options object
def open_url(host, url):
h = http.client.HTTPConnection(host)
# h.set_debuglevel(9)
h.putrequest('GET', url)
h.endheaders()
resp = h.getresponse()
print("Initial response is", resp.status, resp.reason)
body = resp.read()
if resp.status == 302: # object moved
url = "/" + resp.msg["location"]
resp.close()
h.putrequest('GET', url)
h.endheaders()
resp = h.getresponse()
print("After redirect response is", resp.status, resp.reason)
if options.show_headers:
print("Initial response headers:")
for name, val in list(resp.msg.items()):
print(" %s: %s" % (name, val))
if options.show_body:
print(body)
if resp.status == 401:
# 401: Unauthorized - here is where the real work starts
auth_info = None
if options.user or options.domain or options.password:
auth_info = options.user, options.domain, options.password
ca = ClientAuth("NTLM", auth_info=auth_info)
auth_scheme = ca.pkg_info['Name']
data = None
while 1:
err, out_buf = ca.authorize(data)
data = out_buf[0].Buffer
# Encode it as base64 as required by HTTP
auth = encodestring(data).replace("\012", "")
h.putrequest('GET', url)
h.putheader('Authorization', auth_scheme + ' ' + auth)
h.putheader('Content-Length', '0')
h.endheaders()
resp = h.getresponse()
if options.show_headers:
print("Token dance headers:")
for name, val in list(resp.msg.items()):
print(" %s: %s" % (name, val))
if err==0:
break
else:
if resp.status != 401:
print("Eeek - got response", resp.status)
cl = resp.msg.get("content-length")
if cl:
print(repr(resp.read(int(cl))))
else:
print("no content!")
assert resp.status == 401, resp.status
assert not resp.will_close, "NTLM is per-connection - must not close"
schemes = [s.strip() for s in resp.msg.get("WWW-Authenticate", "").split(",")]
for scheme in schemes:
if scheme.startswith(auth_scheme):
data = decodestring(scheme[len(auth_scheme)+1:])
break
else:
print("Could not find scheme '%s' in schemes %r" % (auth_scheme, schemes))
break
resp.read()
print("Final response status is", resp.status, resp.reason)
if resp.status == 200:
# Worked!
# Check we can read it again without re-authenticating.
if resp.will_close:
print("EEEK - response will close, but NTLM is per connection - it must stay open")
body = resp.read()
if options.show_body:
print("Final response body:")
print(body)
h.putrequest('GET', url)
h.endheaders()
resp = h.getresponse()
print("Second fetch response is", resp.status, resp.reason)
if options.show_headers:
print("Second response headers:")
for name, val in list(resp.msg.items()):
print(" %s: %s" % (name, val))
resp.read(int(resp.msg.get("content-length", 0)))
elif resp.status == 500:
print("Error text")
print(resp.read())
else:
if options.show_body:
cl = resp.msg.get("content-length")
print(resp.read(int(cl)))
if __name__=='__main__':
parser = optparse.OptionParser(description=__doc__)
parser.add_option("", "--show-body", action="store_true",
help="print the body of each response as it is received")
parser.add_option("", "--show-headers", action="store_true",
help="print the headers of each response as it is received")
parser.add_option("", "--user", action="store",
help="The username to login with")
parser.add_option("", "--password", action="store",
help="The password to login with")
parser.add_option("", "--domain", action="store",
help="The domain to login to")
options, args = parser.parse_args()
if not args:
print("Run with --help for usage details")
args = ["http://localhost/localstart.asp"]
for url in args:
scheme, netloc, path, params, query, fragment = urllib.parse.urlparse(url)
if (scheme != "http") or params or query or fragment:
parser.error("Scheme must be http, URL must be simple")
print("Opening '%s' from '%s'" % (path, netloc))
r = open_url(netloc, path)

@ -0,0 +1,71 @@
# A demo of basic SSPI authentication.
# There is a 'client' context and a 'server' context - typically these will
# be on different machines (here they are in the same process, but the same
# concepts apply)
import sspi
import win32security, sspicon, win32api
def lookup_ret_code(err):
for k,v in list(sspicon.__dict__.items()):
if k[0:6] in ('SEC_I_','SEC_E_') and v==err:
return k
"""
pkg_name='Kerberos'
sspiclient=SSPIClient(pkg_name, win32api.GetUserName(), ## target spn is ourself
None, None, ## use none for client name and authentication information for current context
## u'username', (u'username',u'domain.com',u'passwd'),
sspicon.ISC_REQ_INTEGRITY|sspicon.ISC_REQ_SEQUENCE_DETECT|sspicon.ISC_REQ_REPLAY_DETECT| \
sspicon.ISC_REQ_DELEGATE|sspicon.ISC_REQ_CONFIDENTIALITY|sspicon.ISC_REQ_USE_SESSION_KEY)
sspiserver=SSPIServer(pkg_name, None,
sspicon.ASC_REQ_INTEGRITY|sspicon.ASC_REQ_SEQUENCE_DETECT|sspicon.ASC_REQ_REPLAY_DETECT| \
sspicon.ASC_REQ_DELEGATE|sspicon.ASC_REQ_CONFIDENTIALITY|sspicon.ASC_REQ_STREAM|sspicon.ASC_REQ_USE_SESSION_KEY)
"""
pkg_name='NTLM'
# Setup the 2 contexts.
sspiclient=sspi.ClientAuth(pkg_name)
sspiserver=sspi.ServerAuth(pkg_name)
# Perform the authentication dance, each loop exchanging more information
# on the way to completing authentication.
sec_buffer=None
while 1:
err, sec_buffer = sspiclient.authorize(sec_buffer)
err, sec_buffer = sspiserver.authorize(sec_buffer)
if err==0:
break
# The server can now impersonate the client. In this demo the 2 users will
# always be the same.
sspiserver.ctxt.ImpersonateSecurityContext()
print('Impersonated user: ',win32api.GetUserNameEx(win32api.NameSamCompatible))
sspiserver.ctxt.RevertSecurityContext()
print('Reverted to self: ',win32api.GetUserName())
pkg_size_info=sspiclient.ctxt.QueryContextAttributes(sspicon.SECPKG_ATTR_SIZES)
# Now sign some data
msg='some data to be encrypted ......'
sigsize=pkg_size_info['MaxSignature']
sigbuf=win32security.PySecBufferDescType()
sigbuf.append(win32security.PySecBufferType(len(msg), sspicon.SECBUFFER_DATA))
sigbuf.append(win32security.PySecBufferType(sigsize, sspicon.SECBUFFER_TOKEN))
sigbuf[0].Buffer=msg
sspiclient.ctxt.MakeSignature(0,sigbuf,1)
sspiserver.ctxt.VerifySignature(sigbuf,1)
# And finally encrypt some.
trailersize=pkg_size_info['SecurityTrailer']
encbuf=win32security.PySecBufferDescType()
encbuf.append(win32security.PySecBufferType(len(msg), sspicon.SECBUFFER_DATA))
encbuf.append(win32security.PySecBufferType(trailersize, sspicon.SECBUFFER_TOKEN))
encbuf[0].Buffer=msg
sspiclient.ctxt.EncryptMessage(0,encbuf,1)
print('Encrypted data:',repr(encbuf[0].Buffer))
sspiserver.ctxt.DecryptMessage(encbuf,1)
print('Unencrypted data:',encbuf[0].Buffer)

@ -0,0 +1,178 @@
"""A sample socket server and client using SSPI authentication and encryption.
You must run with either 'client' or 'server' as arguments. A server must be
running before a client can connect.
To use with Kerberos you should include in the client options
--target-spn=username, where 'username' is the user under which the server is
being run.
Running either the client or server as a different user can be informative.
A command-line such as the following may be useful:
`runas /user:{user} {fqp}\python.exe {fqp}\socket_server.py --wait client|server`
{fqp} should specify the relevant fully-qualified path names.
To use 'runas' with Kerberos, the client program will need to
specify --target-spn with the username under which the *server* is running.
See the SSPI documentation for more details.
"""
import sys
import struct
import socketserver
import win32api
import http.client
import traceback
import win32security
import sspi, sspicon
import optparse # sorry, this demo needs 2.3+
options = None # set to optparse object.
def GetUserName():
try:
return win32api.GetUserName()
except win32api.error as details:
# Seeing 'access denied' errors here for non-local users (presumably
# without permission to login locally). Get the fully-qualified
# username, although a side-effect of these permission-denied errors
# is a lack of Python codecs - so printing the Unicode value fails.
# So just return the repr(), and avoid codecs completely.
return repr(win32api.GetUserNameEx(win32api.NameSamCompatible))
# Send a simple "message" over a socket - send the number of bytes first,
# then the string. Ditto for receive.
def _send_msg(s, m):
s.send(struct.pack("i", len(m)))
s.send(m)
def _get_msg(s):
size_data = s.recv(struct.calcsize("i"))
if not size_data:
return None
cb = struct.unpack("i", size_data)[0]
return s.recv(cb)
class SSPISocketServer(socketserver.TCPServer):
def __init__(self, *args, **kw):
socketserver.TCPServer.__init__(self, *args, **kw)
self.sa = sspi.ServerAuth(options.package)
def verify_request(self, sock, ca):
# Do the sspi auth dance
self.sa.reset()
while 1:
data = _get_msg(sock)
if data is None:
return False
try:
err, sec_buffer = self.sa.authorize(data)
except sspi.error as details:
print("FAILED to authorize client:", details)
return False
if err==0:
break
_send_msg(sock, sec_buffer[0].Buffer)
return True
def process_request(self, request, client_address):
# An example using the connection once it is established.
print("The server is running as user", GetUserName())
self.sa.ctxt.ImpersonateSecurityContext()
try:
print("Having conversation with client as user", GetUserName())
while 1:
# we need to grab 2 bits of data - the encrypted data, and the
# 'key'
data = _get_msg(request)
key = _get_msg(request)
if data is None or key is None:
break
data = self.sa.decrypt(data, key)
print("Client sent:", repr(data))
finally:
self.sa.ctxt.RevertSecurityContext()
self.close_request(request)
print("The server is back to user", GetUserName())
def serve():
s = SSPISocketServer(("localhost", options.port), None)
print("Running test server...")
s.serve_forever()
def sspi_client():
c = http.client.HTTPConnection("localhost", options.port)
c.connect()
# Do the auth dance.
ca = sspi.ClientAuth(options.package, targetspn=options.target_spn)
data = None
while 1:
err, out_buf = ca.authorize(data)
_send_msg(c.sock, out_buf[0].Buffer)
if err==0:
break
data = _get_msg(c.sock)
print("Auth dance complete - sending a few encryted messages")
# Assume out data is sensitive - encrypt the message.
for data in "Hello from the client".split():
blob, key = ca.encrypt(data)
_send_msg(c.sock, blob)
_send_msg(c.sock, key)
c.sock.close()
print("Client completed.")
if __name__=='__main__':
parser = optparse.OptionParser("%prog [options] client|server",
description=__doc__)
parser.add_option("", "--package", action="store", default="NTLM",
help="The SSPI package to use (eg, Kerberos) - default is NTLM")
parser.add_option("", "--target-spn", action="store",
help="""The target security provider name to use. The
string contents are security-package specific. For
example, 'Kerberos' or 'Negotiate' require the server
principal name (SPN) (ie, the username) of the remote
process. For NTLM this must be blank.""")
parser.add_option("", "--port", action="store", default="8181",
help="The port number to use (default=8181)")
parser.add_option("", "--wait", action="store_true",
help="""Cause the program to wait for input just before
terminating. Useful when using via runas to see
any error messages before termination.
""")
options, args = parser.parse_args()
try:
options.port = int(options.port)
except (ValueError, TypeError):
parser.error("--port must be an integer")
try:
try:
if not args:
args = ['']
if args[0]=="client":
sspi_client()
elif args[0]=="server":
serve()
else:
parser.error("You must supply 'client' or 'server' - " \
"use --help for details")
except KeyboardInterrupt:
pass
except SystemExit:
pass
except:
traceback.print_exc()
finally:
if options.wait:
input("Press enter to continue")

@ -0,0 +1,38 @@
# Demonstrates how to validate a password.
# See also MSKB article Q180548
#
# To use with Kerberos you need to jump through the 'targetspn' hoops.
import win32security
import sys
from sspi import ClientAuth, ServerAuth
def validate(username, password, domain = ""):
auth_info = username, domain, password
ca = ClientAuth("NTLM", auth_info = auth_info)
sa = ServerAuth("NTLM")
data = err = None
while err != 0:
err, data = ca.authorize(data)
err, data = sa.authorize(data)
# If we get here without exception, we worked!
if __name__=='__main__':
if len(sys.argv) not in [2,3,4]:
print("Usage: %s username [password [domain]]" % (__file__,))
sys.exit(1)
# password and domain are optional!
password = None
if len(sys.argv)>=3:
password = sys.argv[2]
domain = ""
if len(sys.argv)>=4:
domain = sys.argv[3]
try:
validate(sys.argv[1], password, domain)
print("Validated OK")
except win32security.error as details:
hr, func, msg = details
print("Validation failed: %s (%d)" % (msg, hr))

@ -0,0 +1,56 @@
# This is an example of a service hosted by python.exe rather than
# pythonservice.exe.
# Note that it is very rare that using python.exe is a better option
# than the default pythonservice.exe - the latter has better error handling
# so that if Python itself can't be initialized or there are very early
# import errors, you will get error details written to the event log. When
# using python.exe instead, you are forced to wait for the interpreter startup
# and imports to succeed before you are able to effectively setup your own
# error handling.
# So in short, please make sure you *really* want to do this, otherwise just
# stick with the default.
import sys
import os
import win32serviceutil
import servicemanager
from pipeTestService import TestPipeService
class NativeTestPipeService(TestPipeService):
_svc_name_ = "PyNativePipeTestService"
_svc_display_name_ = "Python Native Pipe Test Service"
_svc_description_ = "Tests Python.exe hosted services"
# tell win32serviceutil we have a custom executable and custom args
# so registration does the right thing.
_exe_name_ = sys.executable
_exe_args_ = '"' + os.path.abspath(sys.argv[0]) + '"'
def main():
if len(sys.argv)==1:
# service must be starting...
# for the sake of debugging etc, we use win32traceutil to see
# any unhandled exceptions and print statements.
import win32traceutil
print("service is starting...")
print("(execute this script with '--help' if that isn't what you want)")
servicemanager.Initialize()
servicemanager.PrepareToHostSingle(NativeTestPipeService)
# Now ask the service manager to fire things up for us...
servicemanager.StartServiceCtrlDispatcher()
print("service done!")
else:
win32serviceutil.HandleCommandLine(NativeTestPipeService)
if __name__=='__main__':
try:
main()
except (SystemExit, KeyboardInterrupt):
raise
except:
print("Something went bad!")
import traceback
traceback.print_exc()

@ -0,0 +1,163 @@
# A Demo of services and named pipes.
# A multi-threaded service that simply echos back its input.
# * Install as a service using "pipeTestService.py install"
# * Use Control Panel to change the user name of the service
# to a real user name (ie, NOT the SystemAccount)
# * Start the service.
# * Run the "pipeTestServiceClient.py" program as the client pipe side.
import win32serviceutil, win32service
import pywintypes, win32con, winerror
# Use "import *" to keep this looking as much as a "normal" service
# as possible. Real code shouldn't do this.
from win32event import *
from win32file import *
from win32pipe import *
from win32api import *
from ntsecuritycon import *
# Old versions of the service framework would not let you import this
# module at the top-level. Now you can, and can check 'Debugging()' and
# 'RunningAsService()' to check your context.
import servicemanager
import traceback
import _thread
def ApplyIgnoreError(fn, args):
try:
return fn(*args)
except error: # Ignore win32api errors.
return None
class TestPipeService(win32serviceutil.ServiceFramework):
_svc_name_ = "PyPipeTestService"
_svc_display_name_ = "Python Pipe Test Service"
_svc_description_ = "Tests Python service framework by receiving and echoing messages over a named pipe"
def __init__(self, args):
win32serviceutil.ServiceFramework.__init__(self, args)
self.hWaitStop = CreateEvent(None, 0, 0, None)
self.overlapped = pywintypes.OVERLAPPED()
self.overlapped.hEvent = CreateEvent(None,0,0,None)
self.thread_handles = []
def CreatePipeSecurityObject(self):
# Create a security object giving World read/write access,
# but only "Owner" modify access.
sa = pywintypes.SECURITY_ATTRIBUTES()
sidEveryone = pywintypes.SID()
sidEveryone.Initialize(SECURITY_WORLD_SID_AUTHORITY,1)
sidEveryone.SetSubAuthority(0, SECURITY_WORLD_RID)
sidCreator = pywintypes.SID()
sidCreator.Initialize(SECURITY_CREATOR_SID_AUTHORITY,1)
sidCreator.SetSubAuthority(0, SECURITY_CREATOR_OWNER_RID)
acl = pywintypes.ACL()
acl.AddAccessAllowedAce(FILE_GENERIC_READ|FILE_GENERIC_WRITE, sidEveryone)
acl.AddAccessAllowedAce(FILE_ALL_ACCESS, sidCreator)
sa.SetSecurityDescriptorDacl(1, acl, 0)
return sa
# The functions executed in their own thread to process a client request.
def DoProcessClient(self, pipeHandle, tid):
try:
try:
# Create a loop, reading large data. If we knew the data stream was
# was small, a simple ReadFile would do.
d = ''.encode('ascii') # ensure bytes on py2k and py3k...
hr = winerror.ERROR_MORE_DATA
while hr==winerror.ERROR_MORE_DATA:
hr, thisd = ReadFile(pipeHandle, 256)
d = d + thisd
print("Read", d)
ok = 1
except error:
# Client disconnection - do nothing
ok = 0
# A secure service would handle (and ignore!) errors writing to the
# pipe, but for the sake of this demo we dont (if only to see what errors
# we can get when our clients break at strange times :-)
if ok:
msg = ("%s (on thread %d) sent me %s" % (GetNamedPipeHandleState(pipeHandle)[4],tid, d)).encode('ascii')
WriteFile(pipeHandle, msg)
finally:
ApplyIgnoreError( DisconnectNamedPipe, (pipeHandle,) )
ApplyIgnoreError( CloseHandle, (pipeHandle,) )
def ProcessClient(self, pipeHandle):
try:
procHandle = GetCurrentProcess()
th = DuplicateHandle(procHandle, GetCurrentThread(), procHandle, 0, 0, win32con.DUPLICATE_SAME_ACCESS)
try:
self.thread_handles.append(th)
try:
return self.DoProcessClient(pipeHandle, th)
except:
traceback.print_exc()
finally:
self.thread_handles.remove(th)
except:
traceback.print_exc()
def SvcStop(self):
self.ReportServiceStatus(win32service.SERVICE_STOP_PENDING)
SetEvent(self.hWaitStop)
def SvcDoRun(self):
# Write an event log record - in debug mode we will also
# see this message printed.
servicemanager.LogMsg(
servicemanager.EVENTLOG_INFORMATION_TYPE,
servicemanager.PYS_SERVICE_STARTED,
(self._svc_name_, '')
)
num_connections = 0
while 1:
pipeHandle = CreateNamedPipe("\\\\.\\pipe\\PyPipeTest",
PIPE_ACCESS_DUPLEX| FILE_FLAG_OVERLAPPED,
PIPE_TYPE_MESSAGE | PIPE_READMODE_BYTE,
PIPE_UNLIMITED_INSTANCES, # max instances
0, 0, 6000,
self.CreatePipeSecurityObject())
try:
hr = ConnectNamedPipe(pipeHandle, self.overlapped)
except error as details:
print("Error connecting pipe!", details)
CloseHandle(pipeHandle)
break
if hr==winerror.ERROR_PIPE_CONNECTED:
# Client is already connected - signal event
SetEvent(self.overlapped.hEvent)
rc = WaitForMultipleObjects((self.hWaitStop, self.overlapped.hEvent), 0, INFINITE)
if rc==WAIT_OBJECT_0:
# Stop event
break
else:
# Pipe event - spawn thread to deal with it.
_thread.start_new_thread(self.ProcessClient, (pipeHandle,))
num_connections = num_connections + 1
# Sleep to ensure that any new threads are in the list, and then
# wait for all current threads to finish.
# What is a better way?
Sleep(500)
while self.thread_handles:
self.ReportServiceStatus(win32service.SERVICE_STOP_PENDING, 5000)
print("Waiting for %d threads to finish..." % (len(self.thread_handles)))
WaitForMultipleObjects(self.thread_handles, 1, 3000)
# Write another event log record.
servicemanager.LogMsg(
servicemanager.EVENTLOG_INFORMATION_TYPE,
servicemanager.PYS_SERVICE_STOPPED,
(self._svc_name_, " after processing %d connections" % (num_connections,))
)
if __name__=='__main__':
win32serviceutil.HandleCommandLine(TestPipeService)

@ -0,0 +1,121 @@
# A Test Program for pipeTestService.py
#
# Install and start the Pipe Test service, then run this test
# either from the same machine, or from another using the "-s" param.
#
# Eg: pipeTestServiceClient.py -s server_name Hi There
# Should work.
from win32pipe import *
from win32file import *
from win32event import *
import pywintypes
import win32api
import winerror
import sys, os, traceback
verbose = 0
#def ReadFromPipe(pipeName):
# Could (Should?) use CallNamedPipe, but this technique allows variable size
# messages (whereas you must supply a buffer size for CallNamedPipe!
# hPipe = CreateFile(pipeName, GENERIC_WRITE, 0, None, OPEN_EXISTING, FILE_ATTRIBUTE_NORMAL, 0)
# more = 1
# while more:
# hr = ReadFile(hPipe, 256)
# if hr==0:
# more = 0
# except win32api.error (hr, fn, desc):
# if hr==winerror.ERROR_MORE_DATA:
# data = dat
#
def CallPipe(fn, args):
ret = None
retryCount = 0
while retryCount < 8: # Keep looping until user cancels.
retryCount = retryCount + 1
try:
return fn(*args)
except win32api.error as exc:
if exc.winerror==winerror.ERROR_PIPE_BUSY:
win32api.Sleep(5000)
continue
else:
raise
raise RuntimeError("Could not make a connection to the server")
def testClient(server,msg):
if verbose:
print("Sending", msg)
data = CallPipe(CallNamedPipe, ("\\\\%s\\pipe\\PyPipeTest" % server, msg, 256, NMPWAIT_WAIT_FOREVER))
if verbose:
print("Server sent back '%s'" % data)
print("Sent and received a message!")
def testLargeMessage(server, size = 4096):
if verbose:
print("Sending message of size %d" % (size))
msg = "*" * size
data = CallPipe(CallNamedPipe, ("\\\\%s\\pipe\\PyPipeTest" % server, msg, 512, NMPWAIT_WAIT_FOREVER))
if len(data)-size:
print("Sizes are all wrong - send %d, got back %d" % (size, len(data)))
def stressThread(server, numMessages, wait):
try:
try:
for i in range(numMessages):
r = CallPipe(CallNamedPipe, ("\\\\%s\\pipe\\PyPipeTest" % server, "#" * 512, 1024, NMPWAIT_WAIT_FOREVER))
except:
traceback.print_exc()
print("Failed after %d messages" % i)
finally:
SetEvent(wait)
def stressTestClient(server, numThreads, numMessages):
import _thread
thread_waits = []
for t_num in range(numThreads):
# Note I could just wait on thread handles (after calling DuplicateHandle)
# See the service itself for an example of waiting for the clients...
wait = CreateEvent(None, 0, 0, None)
thread_waits.append(wait)
_thread.start_new_thread(stressThread, (server,numMessages, wait))
# Wait for all threads to finish.
WaitForMultipleObjects(thread_waits, 1, INFINITE)
def main():
import sys, getopt
server = "."
thread_count = 0
msg_count = 500
try:
opts, args = getopt.getopt(sys.argv[1:], 's:t:m:vl')
for o,a in opts:
if o=='-s':
server = a
if o=='-m':
msg_count = int(a)
if o=='-t':
thread_count = int(a)
if o=='-v':
global verbose
verbose = 1
if o=='-l':
testLargeMessage(server)
msg = " ".join(args).encode("mbcs")
except getopt.error as msg:
print(msg)
my_name = os.path.split(sys.argv[0])[1]
print("Usage: %s [-v] [-s server] [-t thread_count=0] [-m msg_count=500] msg ..." % my_name)
print(" -v = verbose")
print(" Specifying a value for -t will stress test using that many threads.")
return
testClient(server, msg)
if thread_count > 0:
print("Spawning %d threads each sending %d messages..." % (thread_count, msg_count))
stressTestClient(server, thread_count, msg_count)
if __name__=='__main__':
main()

@ -0,0 +1,88 @@
# A Demo of a service that takes advantage of the additional notifications
# available in later Windows versions.
# Note that all output is written as event log entries - so you must install
# and start the service, then look at the event log for messages as events
# are generated.
# Events are generated for USB device insertion and removal, power state
# changes and hardware profile events - so try putting your computer to
# sleep and waking it, inserting a memory stick, etc then check the event log
import win32serviceutil, win32service
import win32event
import servicemanager
# Most event notification support lives around win32gui
import win32gui, win32gui_struct, win32con
GUID_DEVINTERFACE_USB_DEVICE = "{A5DCBF10-6530-11D2-901F-00C04FB951ED}"
class EventDemoService(win32serviceutil.ServiceFramework):
_svc_name_ = "PyServiceEventDemo"
_svc_display_name_ = "Python Service Event Demo"
_svc_description_ = "Demonstrates a Python service which takes advantage of the extra notifications"
def __init__(self, args):
win32serviceutil.ServiceFramework.__init__(self, args)
self.hWaitStop = win32event.CreateEvent(None, 0, 0, None)
# register for a device notification - we pass our service handle
# instead of a window handle.
filter = win32gui_struct.PackDEV_BROADCAST_DEVICEINTERFACE(
GUID_DEVINTERFACE_USB_DEVICE)
self.hdn = win32gui.RegisterDeviceNotification(self.ssh, filter,
win32con.DEVICE_NOTIFY_SERVICE_HANDLE)
# Override the base class so we can accept additional events.
def GetAcceptedControls(self):
# say we accept them all.
rc = win32serviceutil.ServiceFramework.GetAcceptedControls(self)
rc |= win32service.SERVICE_ACCEPT_PARAMCHANGE \
| win32service.SERVICE_ACCEPT_NETBINDCHANGE \
| win32service.SERVICE_CONTROL_DEVICEEVENT \
| win32service.SERVICE_ACCEPT_HARDWAREPROFILECHANGE \
| win32service.SERVICE_ACCEPT_POWEREVENT \
| win32service.SERVICE_ACCEPT_SESSIONCHANGE
return rc
# All extra events are sent via SvcOtherEx (SvcOther remains as a
# function taking only the first args for backwards compat)
def SvcOtherEx(self, control, event_type, data):
# This is only showing a few of the extra events - see the MSDN
# docs for "HandlerEx callback" for more info.
if control == win32service.SERVICE_CONTROL_DEVICEEVENT:
info = win32gui_struct.UnpackDEV_BROADCAST(data)
msg = "A device event occurred: %x - %s" % (event_type, info)
elif control == win32service.SERVICE_CONTROL_HARDWAREPROFILECHANGE:
msg = "A hardware profile changed: type=%s, data=%s" % (event_type, data)
elif control == win32service.SERVICE_CONTROL_POWEREVENT:
msg = "A power event: setting %s" % data
elif control == win32service.SERVICE_CONTROL_SESSIONCHANGE:
# data is a single elt tuple, but this could potentially grow
# in the future if the win32 struct does
msg = "Session event: type=%s, data=%s" % (event_type, data)
else:
msg = "Other event: code=%d, type=%s, data=%s" \
% (control, event_type, data)
servicemanager.LogMsg(
servicemanager.EVENTLOG_INFORMATION_TYPE,
0xF000, # generic message
(msg, '')
)
def SvcStop(self):
self.ReportServiceStatus(win32service.SERVICE_STOP_PENDING)
win32event.SetEvent(self.hWaitStop)
def SvcDoRun(self):
# do nothing at all - just wait to be stopped
win32event.WaitForSingleObject(self.hWaitStop, win32event.INFINITE)
# Write a stop message.
servicemanager.LogMsg(
servicemanager.EVENTLOG_INFORMATION_TYPE,
servicemanager.PYS_SERVICE_STOPPED,
(self._svc_name_, '')
)
if __name__=='__main__':
win32serviceutil.HandleCommandLine(EventDemoService)

@ -0,0 +1,68 @@
# -*- Mode: Python; tab-width: 4 -*-
#
# This module, and the timer.pyd core timer support, were written by
# Sam Rushing (rushing@nightmare.com)
import timer
import time
# Timers are based on Windows messages. So we need
# to do the event-loop thing!
import win32event, win32gui
# glork holds a simple counter for us.
class glork:
def __init__ (self, delay=1000, max=10):
self.x = 0
self.max = max
self.id = timer.set_timer (delay, self.increment)
# Could use the threading module, but this is
# a win32 extension test after all! :-)
self.event = win32event.CreateEvent(None, 0, 0, None)
def increment (self, id, time):
print('x = %d' % self.x)
self.x = self.x + 1
# if we've reached the max count,
# kill off the timer.
if self.x > self.max:
# we could have used 'self.id' here, too
timer.kill_timer (id)
win32event.SetEvent(self.event)
# create a counter that will count from '1' thru '10', incrementing
# once a second, and then stop.
def demo (delay=1000, stop=10):
g = glork(delay, stop)
# Timers are message based - so we need
# To run a message loop while waiting for our timers
# to expire.
start_time = time.time()
while 1:
# We can't simply give a timeout of 30 seconds, as
# we may continouusly be recieving other input messages,
# and therefore never expire.
rc = win32event.MsgWaitForMultipleObjects(
(g.event,), # list of objects
0, # wait all
500, # timeout
win32event.QS_ALLEVENTS, # type of input
)
if rc == win32event.WAIT_OBJECT_0:
# Event signalled.
break
elif rc == win32event.WAIT_OBJECT_0+1:
# Message waiting.
if win32gui.PumpWaitingMessages():
raise RuntimeError("We got an unexpected WM_QUIT message!")
else:
# This wait timed-out.
if time.time()-start_time > 30:
raise RuntimeError("We timed out waiting for the timers to expire!")
if __name__=='__main__':
demo()

@ -0,0 +1,134 @@
# win32clipboardDemo.py
#
# Demo/test of the win32clipboard module.
from win32clipboard import *
from pywin32_testutil import str2bytes # py3k-friendly helper
import win32con
import types
if not __debug__:
print("WARNING: The test code in this module uses assert")
print("This instance of Python has asserts disabled, so many tests will be skipped")
cf_names = {}
# Build map of CF_* constants to names.
for name, val in list(win32con.__dict__.items()):
if name[:3]=="CF_" and name != "CF_SCREENFONTS": # CF_SCREEN_FONTS==CF_TEXT!?!?
cf_names[val] = name
def TestEmptyClipboard():
OpenClipboard()
try:
EmptyClipboard()
assert EnumClipboardFormats(0)==0, "Clipboard formats were available after emptying it!"
finally:
CloseClipboard()
def TestText():
OpenClipboard()
try:
text = "Hello from Python"
text_bytes = str2bytes(text)
SetClipboardText(text)
got = GetClipboardData(win32con.CF_TEXT)
# CF_TEXT always gives us 'bytes' back .
assert got == text_bytes, "Didnt get the correct result back - '%r'." % (got,)
finally:
CloseClipboard()
OpenClipboard()
try:
# CF_UNICODE text always gives unicode objects back.
got = GetClipboardData(win32con.CF_UNICODETEXT)
assert got == text, "Didnt get the correct result back - '%r'." % (got,)
assert type(got)==str, "Didnt get the correct result back - '%r'." % (got,)
# CF_OEMTEXT is a bytes-based format.
got = GetClipboardData(win32con.CF_OEMTEXT)
assert got == text_bytes, "Didnt get the correct result back - '%r'." % (got,)
# Unicode tests
EmptyClipboard()
text = "Hello from Python unicode"
text_bytes = str2bytes(text)
# Now set the Unicode value
SetClipboardData(win32con.CF_UNICODETEXT, text)
# Get it in Unicode.
got = GetClipboardData(win32con.CF_UNICODETEXT)
assert got == text, "Didnt get the correct result back - '%r'." % (got,)
assert type(got)==str, "Didnt get the correct result back - '%r'." % (got,)
# Close and open the clipboard to ensure auto-conversions take place.
finally:
CloseClipboard()
OpenClipboard()
try:
# Make sure I can still get the text as bytes
got = GetClipboardData(win32con.CF_TEXT)
assert got == text_bytes, "Didnt get the correct result back - '%r'." % (got,)
# Make sure we get back the correct types.
got = GetClipboardData(win32con.CF_UNICODETEXT)
assert type(got)==str, "Didnt get the correct result back - '%r'." % (got,)
got = GetClipboardData(win32con.CF_OEMTEXT)
assert got == text_bytes, "Didnt get the correct result back - '%r'." % (got,)
print("Clipboard text tests worked correctly")
finally:
CloseClipboard()
def TestClipboardEnum():
OpenClipboard()
try:
# Enumerate over the clipboard types
enum = 0
while 1:
enum = EnumClipboardFormats(enum)
if enum==0:
break
assert IsClipboardFormatAvailable(enum), "Have format, but clipboard says it is not available!"
n = cf_names.get(enum,"")
if not n:
try:
n = GetClipboardFormatName(enum)
except error:
n = "unknown (%s)" % (enum,)
print("Have format", n)
print("Clipboard enumerator tests worked correctly")
finally:
CloseClipboard()
class Foo:
def __init__(self, **kw):
self.__dict__.update(kw)
def __cmp__(self, other):
return cmp(self.__dict__, other.__dict__)
def __eq__(self, other):
return self.__dict__==other.__dict__
def TestCustomFormat():
OpenClipboard()
try:
# Just for the fun of it pickle Python objects through the clipboard
fmt = RegisterClipboardFormat("Python Pickle Format")
import pickle
pickled_object = Foo(a=1, b=2, Hi=3)
SetClipboardData(fmt, pickle.dumps( pickled_object ) )
# Now read it back.
data = GetClipboardData(fmt)
loaded_object = pickle.loads(data)
assert pickle.loads(data) == pickled_object, "Didnt get the correct data!"
print("Clipboard custom format tests worked correctly")
finally:
CloseClipboard()
if __name__=='__main__':
TestEmptyClipboard()
TestText()
TestCustomFormat()
TestClipboardEnum()
# And leave it empty at the end!
TestEmptyClipboard()

@ -0,0 +1,85 @@
import win32gui
import win32api
import win32clipboard
import win32con
import time
class ViewerWindow:
def __init__(self):
self.hwndNextViewer = None
def OnPaint(self, hwnd, msg, wp, lp):
dc, ps=win32gui.BeginPaint(hwnd)
wndrect = win32gui.GetClientRect(hwnd)
wndwidth = wndrect[2]-wndrect[0]
wndheight = wndrect[3]-wndrect[1]
win32clipboard.OpenClipboard()
try:
try:
hbitmap = win32clipboard.GetClipboardData(win32clipboard.CF_BITMAP)
except TypeError:
font=win32gui.LOGFONT()
font.lfHeight=15 #int(wndheight/20)
font.lfWidth=15 #font.lfHeight
# font.lfWeight=150
hf=win32gui.CreateFontIndirect(font)
win32gui.SelectObject(dc,hf)
win32gui.SetBkMode(dc, win32con.TRANSPARENT)
win32gui.SetTextColor(dc,win32api.RGB(0,0,0))
win32gui.DrawText(dc,'No bitmaps are in the clipboard\n(try pressing the PrtScn button)', -1,
(0,0, wndwidth, wndheight),
win32con.DT_CENTER)
else:
bminfo = win32gui.GetObject(hbitmap)
dcDC = win32gui.CreateCompatibleDC(None)
win32gui.SelectObject(dcDC, hbitmap)
win32gui.StretchBlt(dc, 0, 0, wndwidth, wndheight, dcDC, 0, 0, bminfo.bmWidth, bminfo.bmHeight, win32con.SRCCOPY)
win32gui.DeleteDC(dcDC)
win32gui.EndPaint(hwnd, ps)
finally:
win32clipboard.CloseClipboard()
return 0
def OnDrawClipboard(self, hwnd, msg, wp, lp):
win32gui.InvalidateRect(hwnd,None,True)
def OnChangeCBChain(self, hwnd, msg, wp, lp):
# If the next window is closing, repair the chain.
if wp == self.hwndNextViewer:
self.hwndNextViewer = lp
# Otherwise, pass the message to the next link.
elif self.hwndNextViewer:
win32gui.SendMessage(self.hwndNextViewer, msg, wp, lp)
def OnCreate(self, hwnd, msg, wp, lp):
self.hwndNextViewer = win32gui.SetClipboardViewer(hwnd);
def OnClose(self, hwnd, msg, wp, lp):
win32clipboard.ChangeClipboardChain(hwnd, self.hwndNextViewer)
win32gui.DestroyWindow(hwnd)
win32gui.PostQuitMessage(0)
def go(self):
wndproc={win32con.WM_PAINT: self.OnPaint,
win32con.WM_CLOSE: self.OnClose,
win32con.WM_CREATE: self.OnCreate,
win32con.WM_DRAWCLIPBOARD: self.OnDrawClipboard,
win32con.WM_CHANGECBCHAIN: self.OnChangeCBChain,
}
wc = win32gui.WNDCLASS()
wc.lpszClassName = 'test_win32clipboard_bmp'
wc.style = win32con.CS_GLOBALCLASS|win32con.CS_VREDRAW | win32con.CS_HREDRAW
wc.hbrBackground = win32con.COLOR_WINDOW+1
wc.lpfnWndProc=wndproc
class_atom=win32gui.RegisterClass(wc)
hwnd = win32gui.CreateWindowEx(0, class_atom,'ClipboardViewer',
win32con.WS_CAPTION|win32con.WS_VISIBLE|win32con.WS_THICKFRAME|win32con.WS_SYSMENU,
100,100,900,900, 0, 0, 0, None)
win32clipboard.SetClipboardViewer(hwnd)
win32gui.PumpMessages()
win32gui.UnregisterClass(class_atom,None)
if __name__=='__main__':
w = ViewerWindow()
w.go()

@ -0,0 +1,133 @@
# This is a simple serial port terminal demo.
#
# Its primary purpose is to demonstrate the native serial port access offered via
# win32file.
# It uses 3 threads:
# - The main thread, which cranks up the other 2 threads, then simply waits for them to exit.
# - The user-input thread - blocks waiting for a keyboard character, and when found sends it
# out the COM port. If the character is Ctrl+C, it stops, signalling the COM port thread to stop.
# - The COM port thread is simply listening for input on the COM port, and prints it to the screen.
# This demo uses userlapped IO, so that none of the read or write operations actually block (however,
# in this sample, the very next thing we do _is_ block - so it shows off the concepts even though it
# doesnt exploit them.
from win32file import * # The base COM port and file IO functions.
from win32event import * # We use events and the WaitFor[Multiple]Objects functions.
import win32con # constants.
import msvcrt # For the getch() function.
import threading
import sys
def FindModem():
# Snoop over the comports, seeing if it is likely we have a modem.
for i in range(1,5):
port = "COM%d" % (i,)
try:
handle = CreateFile(port,
win32con.GENERIC_READ | win32con.GENERIC_WRITE,
0, # exclusive access
None, # no security
win32con.OPEN_EXISTING,
win32con.FILE_ATTRIBUTE_NORMAL,
None)
# It appears that an available COM port will always success here,
# just return 0 for the status flags. We only care that it has _any_ status
# flags (and therefore probably a real modem)
if GetCommModemStatus(handle) != 0:
return port
except error:
pass # No port, or modem status failed.
return None
# A basic synchronous COM port file-like object
class SerialTTY:
def __init__(self, port):
if type(port)==type(0):
port = "COM%d" % (port,)
self.handle = CreateFile(port,
win32con.GENERIC_READ | win32con.GENERIC_WRITE,
0, # exclusive access
None, # no security
win32con.OPEN_EXISTING,
win32con.FILE_ATTRIBUTE_NORMAL | win32con.FILE_FLAG_OVERLAPPED,
None)
# Tell the port we want a notification on each char.
SetCommMask(self.handle, EV_RXCHAR)
# Setup a 4k buffer
SetupComm(self.handle, 4096, 4096)
# Remove anything that was there
PurgeComm(self.handle, PURGE_TXABORT | PURGE_RXABORT | PURGE_TXCLEAR | PURGE_RXCLEAR )
# Setup for overlapped IO.
timeouts = 0xFFFFFFFF, 0, 1000, 0, 1000
SetCommTimeouts(self.handle, timeouts)
# Setup the connection info.
dcb = GetCommState( self.handle )
dcb.BaudRate = CBR_115200
dcb.ByteSize = 8
dcb.Parity = NOPARITY
dcb.StopBits = ONESTOPBIT
SetCommState(self.handle, dcb)
print("Connected to %s at %s baud" % (port, dcb.BaudRate))
def _UserInputReaderThread(self):
overlapped = OVERLAPPED()
overlapped.hEvent = CreateEvent(None, 1, 0, None)
try:
while 1:
ch = msvcrt.getch()
if ord(ch)==3:
break
WriteFile(self.handle, ch, overlapped)
# Wait for the write to complete.
WaitForSingleObject(overlapped.hEvent, INFINITE)
finally:
SetEvent(self.eventStop)
def _ComPortThread(self):
overlapped = OVERLAPPED()
overlapped.hEvent = CreateEvent(None, 1, 0, None)
while 1:
# XXX - note we could _probably_ just use overlapped IO on the win32file.ReadFile() statement
# XXX but this tests the COM stuff!
rc, mask = WaitCommEvent(self.handle, overlapped)
if rc == 0: # Character already ready!
SetEvent(overlapped.hEvent)
rc = WaitForMultipleObjects([overlapped.hEvent, self.eventStop], 0, INFINITE)
if rc == WAIT_OBJECT_0:
# Some input - read and print it
flags, comstat = ClearCommError( self.handle )
rc, data = ReadFile(self.handle, comstat.cbInQue, overlapped)
WaitForSingleObject(overlapped.hEvent, INFINITE)
sys.stdout.write(data)
else:
# Stop the thread!
# Just incase the user input thread uis still going, close it
sys.stdout.close()
break
def Run(self):
self.eventStop = CreateEvent(None, 0, 0, None)
# Start the reader and writer threads.
user_thread = threading.Thread(target = self._UserInputReaderThread)
user_thread.start()
com_thread = threading.Thread(target = self._ComPortThread)
com_thread.start()
user_thread.join()
com_thread.join()
if __name__=='__main__':
print("Serial port terminal demo - press Ctrl+C to exit")
if len(sys.argv)<=1:
port = FindModem()
if port is None:
print("No COM port specified, and no modem could be found")
print("Please re-run this script with the name of a COM port (eg COM3)")
sys.exit(1)
else:
port = sys.argv[1]
tty = SerialTTY(port)
tty.Run()

@ -0,0 +1,103 @@
import win32console, win32con
import traceback, time
virtual_keys={}
for k,v in list(win32con.__dict__.items()):
if k.startswith('VK_'):
virtual_keys[v]=k
free_console=True
try:
win32console.AllocConsole()
except win32console.error as exc:
if exc.winerror!=5:
raise
## only free console if one was created successfully
free_console=False
stdout=win32console.GetStdHandle(win32console.STD_OUTPUT_HANDLE)
stdin=win32console.GetStdHandle(win32console.STD_INPUT_HANDLE)
newbuffer=win32console.CreateConsoleScreenBuffer()
newbuffer.SetConsoleActiveScreenBuffer()
newbuffer.SetConsoleTextAttribute(win32console.FOREGROUND_RED|win32console.FOREGROUND_INTENSITY
|win32console.BACKGROUND_GREEN|win32console.BACKGROUND_INTENSITY)
newbuffer.WriteConsole('This is a new screen buffer\n')
## test setting screen buffer and window size
## screen buffer size cannot be smaller than window size
window_size=newbuffer.GetConsoleScreenBufferInfo()['Window']
coord=win32console.PyCOORDType(X=window_size.Right+20, Y=window_size.Bottom+20)
newbuffer.SetConsoleScreenBufferSize(coord)
window_size.Right+=10
window_size.Bottom+=10
newbuffer.SetConsoleWindowInfo(Absolute=True,ConsoleWindow=window_size)
## write some records to the input queue
x=win32console.PyINPUT_RECORDType(win32console.KEY_EVENT)
x.Char='X'
x.KeyDown=True
x.RepeatCount=1
x.VirtualKeyCode=0x58
x.ControlKeyState=win32con.SHIFT_PRESSED
z=win32console.PyINPUT_RECORDType(win32console.KEY_EVENT)
z.Char='Z'
z.KeyDown=True
z.RepeatCount=1
z.VirtualKeyCode=0x5a
z.ControlKeyState=win32con.SHIFT_PRESSED
stdin.WriteConsoleInput([x,z,x])
newbuffer.SetConsoleTextAttribute(win32console.FOREGROUND_RED|win32console.FOREGROUND_INTENSITY
|win32console.BACKGROUND_GREEN|win32console.BACKGROUND_INTENSITY)
newbuffer.WriteConsole('Press some keys, click some characters with the mouse\n')
newbuffer.SetConsoleTextAttribute(win32console.FOREGROUND_BLUE|win32console.FOREGROUND_INTENSITY
|win32console.BACKGROUND_RED|win32console.BACKGROUND_INTENSITY)
newbuffer.WriteConsole('Hit "End" key to quit\n')
breakout=False
while not breakout:
input_records=stdin.ReadConsoleInput(10)
for input_record in input_records:
if input_record.EventType==win32console.KEY_EVENT:
if input_record.KeyDown:
if input_record.Char=='\0':
newbuffer.WriteConsole(virtual_keys.get(input_record.VirtualKeyCode, 'VirtualKeyCode: %s' %input_record.VirtualKeyCode))
else:
newbuffer.WriteConsole(input_record.Char)
if input_record.VirtualKeyCode==win32con.VK_END:
breakout=True
break
elif input_record.EventType==win32console.MOUSE_EVENT:
if input_record.EventFlags==0: ## 0 indicates a button event
if input_record.ButtonState!=0: ## exclude button releases
pos=input_record.MousePosition
# switch the foreground and background colors of the character that was clicked
attr=newbuffer.ReadConsoleOutputAttribute(Length=1, ReadCoord=pos)[0]
new_attr=attr
if attr&win32console.FOREGROUND_BLUE:
new_attr=(new_attr&~win32console.FOREGROUND_BLUE)|win32console.BACKGROUND_BLUE
if attr&win32console.FOREGROUND_RED:
new_attr=(new_attr&~win32console.FOREGROUND_RED)|win32console.BACKGROUND_RED
if attr&win32console.FOREGROUND_GREEN:
new_attr=(new_attr&~win32console.FOREGROUND_GREEN)|win32console.BACKGROUND_GREEN
if attr&win32console.BACKGROUND_BLUE:
new_attr=(new_attr&~win32console.BACKGROUND_BLUE)|win32console.FOREGROUND_BLUE
if attr&win32console.BACKGROUND_RED:
new_attr=(new_attr&~win32console.BACKGROUND_RED)|win32console.FOREGROUND_RED
if attr&win32console.BACKGROUND_GREEN:
new_attr=(new_attr&~win32console.BACKGROUND_GREEN)|win32console.FOREGROUND_GREEN
newbuffer.WriteConsoleOutputAttribute((new_attr,),pos)
else:
newbuffer.WriteConsole(str(input_record))
time.sleep(0.1)
stdout.SetConsoleActiveScreenBuffer()
newbuffer.Close()
if free_console:
win32console.FreeConsole()

Some files were not shown because too many files have changed in this diff Show More

Loading…
Cancel
Save