132 lines
7.9 KiB
Python
132 lines
7.9 KiB
Python
"""Private Windows upload handle lease. No filenames or contents in diagnostics."""
|
|
import base64, ctypes as c, ctypes.wintypes as w, hashlib, json, ntpath, os, re, sys
|
|
|
|
VERSION = 'safetynet-upload-lease-v1'
|
|
LIMIT = 10 * 1024 * 1024
|
|
|
|
def run():
|
|
if os.name != 'nt':
|
|
raise ValueError('platform')
|
|
k = c.WinDLL('kernel32', use_last_error=True)
|
|
k.CreateFileW.argtypes = [w.LPCWSTR,w.DWORD,w.DWORD,c.c_void_p,w.DWORD,w.DWORD,w.HANDLE]
|
|
k.CreateFileW.restype = w.HANDLE
|
|
k.CloseHandle.argtypes = [w.HANDLE]
|
|
k.GetFinalPathNameByHandleW.argtypes = [w.HANDLE,w.LPWSTR,w.DWORD,w.DWORD]
|
|
k.GetFileType.argtypes = [w.HANDLE]
|
|
k.GetDriveTypeW.argtypes = [w.LPCWSTR]
|
|
k.GetVolumeInformationW.argtypes = [w.LPCWSTR,w.LPWSTR,w.DWORD,c.POINTER(w.DWORD),c.POINTER(w.DWORD),c.POINTER(w.DWORD),w.LPWSTR,w.DWORD]
|
|
k.ReadFile.argtypes = [w.HANDLE,c.c_void_p,w.DWORD,c.POINTER(w.DWORD),c.c_void_p]
|
|
class Info(c.Structure):
|
|
_fields_ = [('attrs',w.DWORD),('created',w.FILETIME),('accessed',w.FILETIME),('written',w.FILETIME),('volume',w.DWORD),('sizeHi',w.DWORD),('sizeLo',w.DWORD),('links',w.DWORD),('indexHi',w.DWORD),('indexLo',w.DWORD)]
|
|
k.GetFileInformationByHandle.argtypes = [w.HANDLE,c.POINTER(Info)]
|
|
handles = []
|
|
copies=[]
|
|
parent=None
|
|
announced=False
|
|
data=bytearray()
|
|
buf=None
|
|
try:
|
|
line = sys.stdin.buffer.readline(65537)
|
|
if len(line) > 65536 or not line.endswith(b'\n'): raise ValueError('config')
|
|
cfg = json.loads(line)
|
|
if type(cfg) == dict and set(cfg) == {'version','parentPid','closeHandles'}:
|
|
if cfg['version'] != VERSION or type(cfg['parentPid']) != int or cfg['parentPid'] != os.getppid() or type(cfg['closeHandles']) != list or not 1 <= len(cfg['closeHandles']) <= 4096 or not all(type(v)==str and re.fullmatch(r'[1-9][0-9]{0,19}',v) and int(v)<2**64 for v in cfg['closeHandles']): raise ValueError('cleanup')
|
|
k.OpenProcess.argtypes=[w.DWORD,w.BOOL,w.DWORD]; k.OpenProcess.restype=w.HANDLE
|
|
k.DuplicateHandle.argtypes=[w.HANDLE,w.HANDLE,w.HANDLE,c.POINTER(w.HANDLE),w.DWORD,w.BOOL,w.DWORD]
|
|
k.GetCurrentProcess.restype=w.HANDLE
|
|
parent=k.OpenProcess(0x40,False,cfg['parentPid'])
|
|
if not parent: raise ValueError('parent')
|
|
handles.append(parent)
|
|
for value in cfg['closeHandles']:
|
|
duplicate=w.HANDLE()
|
|
if not k.DuplicateHandle(parent,w.HANDLE(int(value)),k.GetCurrentProcess(),c.byref(duplicate),0,False,3): raise ValueError('cleanup')
|
|
k.CloseHandle(duplicate)
|
|
print(json.dumps({'version':VERSION,'released':True}),flush=True)
|
|
return
|
|
if set(cfg) != {'version','paths','maxBytes','parentPid'} or cfg['version'] != VERSION: raise ValueError('config')
|
|
maximum = cfg['maxBytes']
|
|
if type(maximum) != int or not 1 <= maximum <= LIMIT: raise ValueError('config')
|
|
if type(cfg['paths']) != list or not 1 <= len(cfg['paths']) <= 16: raise ValueError('config')
|
|
k.OpenProcess.argtypes=[w.DWORD,w.BOOL,w.DWORD]; k.OpenProcess.restype=w.HANDLE
|
|
k.DuplicateHandle.argtypes=[w.HANDLE,w.HANDLE,w.HANDLE,c.POINTER(w.HANDLE),w.DWORD,w.BOOL,w.DWORD]
|
|
k.GetCurrentProcess.restype=w.HANDLE
|
|
if type(cfg['parentPid']) != int or cfg['parentPid'] != os.getppid(): raise ValueError('parent')
|
|
parent=k.OpenProcess(0x40,False,cfg['parentPid'])
|
|
if not parent: raise ValueError('parent')
|
|
handles.append(parent)
|
|
copies=[]
|
|
files, total = [], 0
|
|
for path in cfg['paths']:
|
|
if type(path) != str or len(path) > 240 or not re.fullmatch(r'[A-Za-z]:\\[^\x00-\x1f:]*', path): raise ValueError('path')
|
|
canonical = ntpath.normpath(path)
|
|
if canonical != path or any(part.endswith((' ','.')) for part in path[3:].split('\\')): raise ValueError('path')
|
|
root = path[:3]
|
|
if k.GetDriveTypeW(root) != 3: raise ValueError('volume')
|
|
filesystem=c.create_unicode_buffer(32)
|
|
if not k.GetVolumeInformationW(root,None,0,None,None,None,filesystem,len(filesystem)) or filesystem.value != 'NTFS': raise ValueError('filesystem')
|
|
parts = path[3:].split('\\')
|
|
chain = [root] + [root + '\\'.join(parts[:i]) for i in range(1,len(parts)+1)]
|
|
volume = None
|
|
for index, name in enumerate(chain):
|
|
directory = index != len(chain)-1
|
|
h = k.CreateFileW(name, 0x80000000, 1, None, 3, 0x00200000 | (0x02000000 if directory else 0), None)
|
|
if h == c.c_void_p(-1).value: raise ValueError('locked')
|
|
handles.append(h)
|
|
duplicate=w.HANDLE()
|
|
if not k.DuplicateHandle(k.GetCurrentProcess(),h,parent,c.byref(duplicate),0,False,2): raise ValueError('duplicate')
|
|
copies.append(str(duplicate.value))
|
|
info = Info()
|
|
if not k.GetFileInformationByHandle(h,c.byref(info)): raise ValueError('identity')
|
|
if info.attrs & 0x400 or bool(info.attrs & 0x10) != directory or k.GetFileType(h) != 1: raise ValueError('type')
|
|
if volume is None: volume = info.volume
|
|
if info.volume != volume: raise ValueError('volume')
|
|
out = c.create_unicode_buffer(32768)
|
|
n = k.GetFinalPathNameByHandleW(h,out,len(out),0)
|
|
if not n or n >= len(out) or out.value.removeprefix('\\\\?\\').casefold().rstrip('\\') != name.casefold().rstrip('\\'): raise ValueError('identity')
|
|
size = (info.sizeHi << 32) | info.sizeLo
|
|
total += size
|
|
if total > maximum: raise ValueError('size')
|
|
data = bytearray()
|
|
while len(data) < size:
|
|
buf = c.create_string_buffer(min(65536,size-len(data))); n = w.DWORD()
|
|
try:
|
|
if not k.ReadFile(h,buf,len(buf),c.byref(n),None) or not n.value: raise ValueError('read')
|
|
data.extend(buf.raw[:n.value])
|
|
finally:
|
|
c.memset(c.addressof(buf),0,len(buf))
|
|
buf=None
|
|
after = Info()
|
|
if not k.GetFileInformationByHandle(h,c.byref(after)) or any(getattr(info,n) != getattr(after,n) for n in ('attrs','volume','sizeHi','sizeLo','links','indexHi','indexLo')) or bytes(info.written) != bytes(after.written): raise ValueError('changed')
|
|
files.append({'path':path,'size':size,'sha256':hashlib.sha256(data).hexdigest(),'volume':info.volume,'fileId':f'{info.indexHi:08x}{info.indexLo:08x}','bytes':base64.b64encode(data).decode('ascii')})
|
|
data[:] = b'\x00' * len(data)
|
|
data.clear()
|
|
print(json.dumps({'version':VERSION,'ok':True,'files':files,'handles':copies},separators=(',',':')),flush=True)
|
|
announced=True
|
|
files.clear()
|
|
command = sys.stdin.buffer.readline(129)
|
|
if command != b'{"release":true}\n': raise ValueError('release')
|
|
for value in copies:
|
|
duplicate=w.HANDLE()
|
|
if not k.DuplicateHandle(parent,w.HANDLE(int(value)),k.GetCurrentProcess(),c.byref(duplicate),0,False,3): raise ValueError('cleanup')
|
|
k.CloseHandle(duplicate)
|
|
for h in reversed(handles): k.CloseHandle(h)
|
|
handles.clear()
|
|
print(json.dumps({'version':VERSION,'released':True}),flush=True)
|
|
finally:
|
|
data[:] = b'\x00' * len(data)
|
|
data.clear()
|
|
if buf is not None: c.memset(c.addressof(buf),0,len(buf))
|
|
if not announced and copies and parent:
|
|
for value in copies:
|
|
duplicate=w.HANDLE()
|
|
if not k.DuplicateHandle(parent,w.HANDLE(int(value)),k.GetCurrentProcess(),c.byref(duplicate),0,False,3):
|
|
os._exit(2)
|
|
k.CloseHandle(duplicate)
|
|
for h in reversed(handles): k.CloseHandle(h)
|
|
|
|
if __name__ == '__main__':
|
|
try: run()
|
|
except BaseException:
|
|
print(json.dumps({'version':VERSION,'ok':False,'code':'upload-lease-refused'}),flush=True)
|
|
sys.exit(1)
|