"""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)