1 """Windows-specific implementation of process utilities. 2 3 This file is only meant to be imported by process.py, not by end-users. 4 """ 5 6 #----------------------------------------------------------------------------- 7 # Copyright (C) 2010-2011 The IPython Development Team 8 # 9 # Distributed under the terms of the BSD License. The full license is in 10 # the file COPYING, distributed as part of this software. 11 #----------------------------------------------------------------------------- 12 13 #----------------------------------------------------------------------------- 14 # Imports 15 #----------------------------------------------------------------------------- 16 17 # stdlib 18 import os 19 import sys 20 import ctypes 21 import time 22 23 from ctypes import c_int, POINTER 24 from ctypes.wintypes import LPCWSTR, HLOCAL 25 from subprocess import STDOUT, TimeoutExpired 26 from threading import Thread 27 28 # our own imports 29 from ._process_common import read_no_interrupt, process_handler, arg_split as py_arg_split 30 from . import py3compat 31 from .encoding import DEFAULT_ENCODING 32 33 #----------------------------------------------------------------------------- 34 # Function definitions 35 #----------------------------------------------------------------------------- 36 37 class AvoidUNCPath(object): 38 """A context manager to protect command execution from UNC paths. 39 40 In the Win32 API, commands can't be invoked with the cwd being a UNC path. 41 This context manager temporarily changes directory to the 'C:' drive on 42 entering, and restores the original working directory on exit. 43 44 The context manager returns the starting working directory *if* it made a 45 change and None otherwise, so that users can apply the necessary adjustment 46 to their system calls in the event of a change. 47 48 Examples 49 -------- 50 :: 51 cmd = 'dir' 52 with AvoidUNCPath() as path: 53 if path is not None: 54 cmd = '"pushd %s &&"%s' % (path, cmd) 55 os.system(cmd) 56 """ 57 def __enter__(self): 58 self.path = os.getcwd() 59 self.is_unc_path = self.path.startswith(r"\\") 60 if self.is_unc_path: 61 # change to c drive (as cmd.exe cannot handle UNC addresses) 62 os.chdir("C:") 63 return self.path 64 else: 65 # We return None to signal that there was no change in the working 66 # directory 67 return None 68 69 def __exit__(self, exc_type, exc_value, traceback): 70 if self.is_unc_path: 71 os.chdir(self.path) 72 73 74 def _system_body(p): 75 """Callback for _system.""" 76 enc = DEFAULT_ENCODING 77 78 def stdout_read(): 79 for line in read_no_interrupt(p.stdout).splitlines(): 80 line = line.decode(enc, 'replace') 81 print(line, file=sys.stdout) 82 83 def stderr_read(): 84 for line in read_no_interrupt(p.stderr).splitlines(): 85 line = line.decode(enc, 'replace') 86 print(line, file=sys.stderr) 87 88 Thread(target=stdout_read).start() 89 Thread(target=stderr_read).start() 90 91 # Wait to finish for returncode. Unfortunately, Python has a bug where 92 # wait() isn't interruptible (https://bugs.python.org/issue28168) so poll in 93 # a loop instead of just doing `return p.wait()`. 94 while True: 95 result = p.poll() 96 if result is None: 97 time.sleep(0.01) 98 else: 99 return result 100 101 102 def system(cmd): 103 """Win32 version of os.system() that works with network shares. 104 105 Note that this implementation returns None, as meant for use in IPython. 106 107 Parameters 108 ---------- 109 cmd : str or list 110 A command to be executed in the system shell. 111 112 Returns 113 ------- 114 int : child process' exit code. 115 """ 116 # The controller provides interactivity with both 117 # stdin and stdout 118 #import _process_win32_controller 119 #_process_win32_controller.system(cmd) 120 121 with AvoidUNCPath() as path: 122 if path is not None: 123 cmd = '"pushd %s &&"%s' % (path, cmd) 124 return process_handler(cmd, _system_body) 125 126 def getoutput(cmd): 127 """Return standard output of executing cmd in a shell. 128 129 Accepts the same arguments as os.system(). 130 131 Parameters 132 ---------- 133 cmd : str or list 134 A command to be executed in the system shell. 135 136 Returns 137 ------- 138 stdout : str 139 """ 140 141 with AvoidUNCPath() as path: 142 if path is not None: 143 cmd = '"pushd %s &&"%s' % (path, cmd) 144 out = process_handler(cmd, lambda p: p.communicate()[0], STDOUT) 145 146 if out is None: 147 out = b'' 148 return py3compat.decode(out) 149 150 try: 151 CommandLineToArgvW = ctypes.windll.shell32.CommandLineToArgvW 152 CommandLineToArgvW.arg_types = [LPCWSTR, POINTER(c_int)] 153 CommandLineToArgvW.restype = POINTER(LPCWSTR) 154 LocalFree = ctypes.windll.kernel32.LocalFree 155 LocalFree.res_type = HLOCAL 156 LocalFree.arg_types = [HLOCAL] 157 158 def arg_split(commandline, posix=False, strict=True): 159 """Split a command line's arguments in a shell-like manner. 160 161 This is a special version for windows that use a ctypes call to CommandLineToArgvW 162 to do the argv splitting. The posix parameter is ignored. 163 164 If strict=False, process_common.arg_split(...strict=False) is used instead. 165 """ 166 #CommandLineToArgvW returns path to executable if called with empty string. 167 if commandline.strip() == "": 168 return [] 169 if not strict: 170 # not really a cl-arg, fallback on _process_common 171 return py_arg_split(commandline, posix=posix, strict=strict) 172 argvn = c_int() 173 result_pointer = CommandLineToArgvW(commandline.lstrip(), ctypes.byref(argvn)) 174 result_array_type = LPCWSTR * argvn.value 175 result = [arg for arg in result_array_type.from_address(ctypes.addressof(result_pointer.contents))] 176 retval = LocalFree(result_pointer) 177 return result 178 except AttributeError: 179 arg_split = py_arg_split 180 181 def check_pid(pid): 182 # OpenProcess returns 0 if no such process (of ours) exists 183 # positive int otherwise 184 return bool(ctypes.windll.kernel32.OpenProcess(1,0,pid)) 185