| """subprocess.run replacement for running untrusted model programs on Windows. |
| |
| Two Windows traps make subprocess.run(..., timeout=) hang forever on a stuck program: |
| * in a venv, python.exe is a launcher that starts the real interpreter as a child; the timeout kills only the |
| launcher and the child keeps the pipes open; |
| * Popen.communicate writes `input` synchronously before it starts waiting, so a program that stops reading a large |
| input blocks the write and the timeout is never checked. |
| run() feeds stdin from a temporary file and kills the whole process tree on timeout, then raises |
| subprocess.TimeoutExpired like subprocess.run.""" |
| import os |
| import subprocess |
| import tempfile |
|
|
|
|
| def run(cmd, input=None, timeout=None, capture_output=True, text=False, encoding=None, errors=None, **kw): |
| kw.setdefault("creationflags", getattr(subprocess, "CREATE_NO_WINDOW", 0)) |
| fin, path = subprocess.DEVNULL, None |
| if input is not None: |
| fd, path = tempfile.mkstemp(suffix=".in") |
| with os.fdopen(fd, "wb") as f: |
| f.write(input if isinstance(input, bytes) else input.encode(encoding or "utf-8")) |
| fin = open(path, "rb") |
| try: |
| p = subprocess.Popen(cmd, stdin=fin, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=text, |
| encoding=encoding, errors=errors, **kw) |
| try: |
| out, err = p.communicate(timeout=timeout) |
| except subprocess.TimeoutExpired: |
| subprocess.run(["taskkill", "/F", "/T", "/PID", str(p.pid)], capture_output=True) |
| try: |
| p.communicate(timeout=10) |
| except subprocess.TimeoutExpired: |
| pass |
| raise subprocess.TimeoutExpired(cmd, timeout) |
| return subprocess.CompletedProcess(cmd, p.returncode, out, err) |
| finally: |
| if path: |
| fin.close() |
| try: |
| os.remove(path) |
| except OSError: |
| pass |
|
|
