-
-
Save dpk/53fa5a1c36568534d2a4 to your computer and use it in GitHub Desktop.
| import os | |
| import os.path | |
| import fcntl | |
| import ctypes | |
| from ctypes.util import find_library | |
| import re | |
| libc = None | |
| F_CLOSEM = None | |
| strategy = None | |
| __all__ = ('closefrom', 'closefrom_impl') | |
| def closefrom(fd): | |
| """ | |
| A portable Python implementation-shim for the closefrom(2) system call, | |
| using various other OS's equivalents of the same wherever available. | |
| Should work on most BSDs, Linux, and OS X at least. Raises a | |
| NotImplementedError if no implementation strategy works. | |
| """ | |
| fd = int(fd) | |
| global strategy | |
| if not strategy: strategy = closefrom_impl() | |
| if strategy == 'closefrom': | |
| return libc.closefrom(fd) | |
| elif strategy == 'fclosem': | |
| return fcntl.fcntl(fd, F_CLOSEM, 0) | |
| elif strategy == 'devfd': | |
| return closefrom_dir('/dev/fd', fd) | |
| elif strategy == 'procfs': | |
| return closefrom_dir('/proc/%d/fd' % os.getpid(), fd) | |
| else: | |
| raise NotImplementedError('closefrom is not supported on this platform') | |
| def closefrom_dir(dir, fd): | |
| fds = [int(fd) for fd in os.listdir(dir)] | |
| for open_fd in fds: | |
| try: | |
| if open_fd >= fd: os.close(open_fd) | |
| except OSError: pass | |
| def closefrom_impl(): | |
| """ | |
| Detects available methods for implementing closefrom(2), and returns | |
| a string value representing the one that closefrom will use. | |
| Return values, in order of preference: | |
| - closefrom | |
| closefrom(2) directly (FreeBSD) | |
| - fclosem | |
| The F_CLOSEM fcntl, functionally equivalent to closefrom (NetBSD) | |
| - devfd | |
| The /dev/fd/ directory will be listed for the necessary file | |
| descriptor numbers (OS X) | |
| - procfs | |
| Same as devfd except that the /proc/<pid>/fd/ directory is used | |
| instead (Linux) | |
| - None | |
| No implementation available | |
| """ | |
| global libc | |
| load_libc() | |
| if libc: | |
| try: | |
| if callable(libc.closefrom): | |
| return 'closefrom' | |
| except AttributeError: pass | |
| global F_CLOSEM | |
| F_CLOSEM = fclosem_constant() | |
| if F_CLOSEM: | |
| return 'fclosem' | |
| if os.path.isdir('/dev/fd'): | |
| return 'devfd' | |
| elif os.path.isdir('/proc/%d/fd' % os.getpid()): | |
| return 'procfs' | |
| return None | |
| def fclosem_constant(): | |
| try: | |
| return fcntl.F_CLOSEM | |
| except AttributeError: pass | |
| fclosem_re = re.compile(rb"^#define\s+F_CLOSEM\s+(\d)+\s") | |
| if os.path.isfile('/usr/include/fcntl.h'): | |
| with open('/usr/include/fcntl.h', 'rb') as header_file: | |
| for line in header_file: | |
| match = fclosem_re.match(line) | |
| if match: | |
| return int(match.group(1)) | |
| return None | |
| def load_libc(): | |
| global libc | |
| libc_name = find_library('c') | |
| if libc_name: | |
| try: | |
| libc = ctypes.cdll.LoadLibrary(libc_name) | |
| except OSError: return None | |
| else: | |
| return None |
| # basic test script for closefrom.py | |
| # passes on: | |
| # - Mac OS X 10.10.2 (devfd) | |
| # - FreeBSD 10.1 (closefrom) | |
| # - Linux 3.13.0 (Ubuntu 14.04, devfd) | |
| # - NetBSD 6.1.5 (closefrom) | |
| import closefrom | |
| import os | |
| strategy = closefrom.closefrom_impl() | |
| if strategy: | |
| print('using %r closefrom implementation' % strategy) | |
| else: | |
| print('no closefrom implementation supported on this platform') | |
| try: | |
| os.closefrom(3) | |
| except NotImplementedError: | |
| print('correctly raised NotImplementedError') | |
| os._exit(0) | |
| print("something happened and NotImplementedError didn't get raised, though?") | |
| os._exit(1) | |
| pipe = os.pipe() | |
| print('created pipe with fds %r' % (pipe,)) | |
| lowerfd = min(pipe) | |
| print("calling closefrom(%d)" % lowerfd) | |
| try: | |
| closefrom.closefrom(lowerfd) | |
| except NotImplementedError: | |
| print("hmm, we detected an implementation strategy but still got a NotImplementedError") | |
| except OSError: | |
| print("hmm, OSError was raised") | |
| passed = True | |
| for fd in pipe: | |
| print("checking %d" % fd) | |
| try: | |
| os.fstat(fd) | |
| except OSError: | |
| print("%d seems to have been closed properly" % fd) | |
| continue | |
| print("hmm, it looks like %d wasn't closed" % fd) | |
| passed = False | |
| if passed: | |
| print("closefrom tests passed!") | |
| os._exit(0) | |
| else: | |
| print("closefrom tests failed") | |
| os._exit(1) |
@46bit Well, you’d hope so. The ‘POSIX way’ to do this is woefully inefficient, though: you do sysconf(_SC_OPEN_MAX) to get the largest file descriptor possible (the default on my shell is 2560, to give you an idea — server processes often have it set much higher) and call close on all of them individually.
Quite a lot of Unixes now support closefrom or the fcntl, with OS X and Linux being the major exception. (NetBSD actually supports closefrom now too, and as far as I know AIX is the only OS that has the fcntl and no closefrom.) Hopefully one or the other will get standardized in a future version of POSIX.
(Not that POSIX standardization necessarily represents cross-platform support: Have you seen all the bugs in Apple’s semaphore implementation?)
Isn't this shit what POSIX was meant to help with?