aboutsummaryrefslogtreecommitdiffstats
path: root/library/python/windows/__init__.py
blob: 94083026e39c3f6eb2cad56c7f647091b0aebb6f (plain) (blame)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
# coding: utf-8

import os
import stat
import sys
import shutil
import logging

from six import reraise

import library.python.func
import library.python.strings

logger = logging.getLogger(__name__)


ERRORS = {
    'SUCCESS': 0,
    'PATH_NOT_FOUND': 3,
    'ACCESS_DENIED': 5,
    'SHARING_VIOLATION': 32,
    'INSUFFICIENT_BUFFER': 122,
    'DIR_NOT_EMPTY': 145,
}

RETRIABLE_FILE_ERRORS = (ERRORS['ACCESS_DENIED'], ERRORS['SHARING_VIOLATION'])
RETRIABLE_DIR_ERRORS = (ERRORS['ACCESS_DENIED'], ERRORS['DIR_NOT_EMPTY'], ERRORS['SHARING_VIOLATION'])


@library.python.func.lazy
def on_win():
    """Check if code run on Windows"""
    return os.name == 'nt'


class NotOnWindowsError(RuntimeError):
    def __init__(self, message):
        super(NotOnWindowsError, self).__init__(message)


class DisabledOnWindowsError(RuntimeError):
    def __init__(self, message):
        super(DisabledOnWindowsError, self).__init__(message)


class NoCTypesError(RuntimeError):
    def __init__(self, message):
        super(NoCTypesError, self).__init__(message)


def win_only(f):
    """Decorator for Windows-only functions"""

    def f_wrapped(*args, **kwargs):
        if not on_win():
            raise NotOnWindowsError('Windows-only function is called, but platform is not Windows')
        return f(*args, **kwargs)

    return f_wrapped


def win_disabled(f):
    """Decorator for functions disabled on Windows"""

    def f_wrapped(*args, **kwargs):
        if on_win():
            run_disabled()
        return f(*args, **kwargs)

    return f_wrapped


def errorfix(f):
    if not on_win():
        return f

    def f_wrapped(*args, **kwargs):
        try:
            return f(*args, **kwargs)
        except WindowsError:
            tp, value, tb = sys.exc_info()
            fix_error(value)
            reraise(tp, value, tb)

    return f_wrapped


def diehard(winerrors, tries=100, delay=1):
    """
    Decorator for diehard wrapper

    On Windows platform retries to run function while specific WindowsError is thrown

    On non-Windows platforms fallbacks to function itself
    """

    def wrap(f):
        if not on_win():
            return f

        return lambda *args, **kwargs: run_diehard(f, winerrors, tries, delay, *args, **kwargs)

    return wrap


def win_path_fix(path):
    """Fix slashes in paths on windows"""
    return path if sys.platform != 'win32' else path.replace('\\', '/')


if on_win():
    import msvcrt
    import time

    import library.python.strings

    _has_ctypes = True
    try:
        import ctypes
        from ctypes import wintypes
    except ImportError:
        _has_ctypes = False

    _INVALID_HANDLE_VALUE = -1

    _MOVEFILE_REPLACE_EXISTING = 0x1
    _MOVEFILE_WRITE_THROUGH = 0x8

    _SEM_FAILCRITICALERRORS = 0x1
    _SEM_NOGPFAULTERRORBOX = 0x2
    _SEM_NOALIGNMENTFAULTEXCEPT = 0x4
    _SEM_NOOPENFILEERRORBOX = 0x8

    _SYMBOLIC_LINK_FLAG_DIRECTORY = 0x1

    _CREATE_NO_WINDOW = 0x8000000

    _ATOMIC_RENAME_FILE_TRANSACTION_DEFAULT_TIMEOUT = 1000

    _HANDLE_FLAG_INHERIT = 0x1

    @win_only
    def require_ctypes(f):
        def f_wrapped(*args, **kwargs):
            if not _has_ctypes:
                raise NoCTypesError('No ctypes found')
            return f(*args, **kwargs)

        return f_wrapped

    # Run function in diehard mode (see diehard decorator commentary)
    @win_only
    def run_diehard(f, winerrors, tries, delay, *args, **kwargs):
        if isinstance(winerrors, int):
            winerrors = (winerrors,)

        ei = None
        for t in range(tries):
            if t:
                logger.debug('Diehard [errs %s]: try #%d in %s', ','.join(str(x) for x in winerrors), t, f)
            try:
                return f(*args, **kwargs)
            except WindowsError as e:
                if e.winerror not in winerrors:
                    raise
                ei = sys.exc_info()
                time.sleep(delay)
        reraise(ei[0], ei[1], ei[2])

    @win_only
    def run_disabled(*args, **kwargs):
        """Placeholder for disabled functions"""
        raise DisabledOnWindowsError('Function called is disabled on Windows')

    class CustomWinError(WindowsError):
        def __init__(self, winerror, message='', filename=None):
            super(CustomWinError, self).__init__(winerror, message)
            self.message = message
            self.strerror = self.message if self.message else format_error(self.windows_error)
            self.filename = filename
            self.utf8 = True

    @win_only
    def unicode_path(path):
        return library.python.strings.to_unicode(path, library.python.strings.fs_encoding())

    @win_only
    @require_ctypes
    def format_error(error):
        if isinstance(error, WindowsError):
            error = error.winerror
        if not isinstance(error, int):
            return 'Unknown'
        return ctypes.FormatError(error)

    @win_only
    def fix_error(windows_error):
        if not windows_error.strerror:
            windows_error.strerror = format_error(windows_error)
        transcode_error(windows_error)

    @win_only
    def transcode_error(windows_error, to_enc='utf-8'):
        from_enc = 'utf-8' if getattr(windows_error, 'utf8', False) else library.python.strings.guess_default_encoding()
        if from_enc != to_enc:
            windows_error.strerror = library.python.strings.to_str(
                windows_error.strerror, to_enc=to_enc, from_enc=from_enc
            )
        setattr(windows_error, 'utf8', to_enc == 'utf-8')

    class Transaction(object):
        def __init__(self, timeout=None, description=''):
            self.timeout = timeout
            self.description = description

        @require_ctypes
        def __enter__(self):
            self._handle = ctypes.windll.ktmw32.CreateTransaction(None, 0, 0, 0, 0, self.timeout, self.description)
            if self._handle == _INVALID_HANDLE_VALUE:
                raise ctypes.WinError()
            return self._handle

        @require_ctypes
        def __exit__(self, t, v, tb):
            try:
                if not ctypes.windll.ktmw32.CommitTransaction(self._handle):
                    raise ctypes.WinError()
            finally:
                ctypes.windll.kernel32.CloseHandle(self._handle)

    @win_only
    def file_handle(f):
        return msvcrt.get_osfhandle(f.fileno())

    # https://www.python.org/dev/peps/pep-0446/
    # http://mihalop.blogspot.ru/2014/05/python-subprocess-and-file-descriptors.html
    @require_ctypes
    @win_only
    def open_file(*args, **kwargs):
        f = open(*args, **kwargs)
        ctypes.windll.kernel32.SetHandleInformation(file_handle(f), _HANDLE_FLAG_INHERIT, 0)
        return f

    @win_only
    @require_ctypes
    def replace_file(src, dst):
        if not ctypes.windll.kernel32.MoveFileExW(
            unicode_path(src), unicode_path(dst), _MOVEFILE_REPLACE_EXISTING | _MOVEFILE_WRITE_THROUGH
        ):
            raise ctypes.WinError()

    @win_only
    @require_ctypes
    def replace_file_across_devices(src, dst):
        with Transaction(
            timeout=_ATOMIC_RENAME_FILE_TRANSACTION_DEFAULT_TIMEOUT,
            description='ya library.python.windows replace_file_across_devices',
        ) as transaction:
            if not ctypes.windll.kernel32.MoveFileTransactedW(
                unicode_path(src),
                unicode_path(dst),
                None,
                None,
                _MOVEFILE_REPLACE_EXISTING | _MOVEFILE_WRITE_THROUGH,
                transaction,
            ):
                raise ctypes.WinError()

    @win_only
    @require_ctypes
    def hardlink(src, lnk):
        if not ctypes.windll.kernel32.CreateHardLinkW(unicode_path(lnk), unicode_path(src), None):
            raise ctypes.WinError()

    # Requires SE_CREATE_SYMBOLIC_LINK_NAME privilege
    @win_only
    @win_disabled
    @require_ctypes
    def symlink_file(src, lnk):
        if not ctypes.windll.kernel32.CreateSymbolicLinkW(unicode_path(lnk), unicode_path(src), 0):
            raise ctypes.WinError()

    # Requires SE_CREATE_SYMBOLIC_LINK_NAME privilege
    @win_only
    @win_disabled
    @require_ctypes
    def symlink_dir(src, lnk):
        if not ctypes.windll.kernel32.CreateSymbolicLinkW(
            unicode_path(lnk), unicode_path(src), _SYMBOLIC_LINK_FLAG_DIRECTORY
        ):
            raise ctypes.WinError()

    @win_only
    @require_ctypes
    def lock_file(f, offset, length, raises=True):
        locked = ctypes.windll.kernel32.LockFile(
            file_handle(f), _low_dword(offset), _high_dword(offset), _low_dword(length), _high_dword(length)
        )
        if not raises:
            return bool(locked)
        if not locked:
            raise ctypes.WinError()

    @win_only
    @require_ctypes
    def unlock_file(f, offset, length, raises=True):
        unlocked = ctypes.windll.kernel32.UnlockFile(
            file_handle(f), _low_dword(offset), _high_dword(offset), _low_dword(length), _high_dword(length)
        )
        if not raises:
            return bool(unlocked)
        if not unlocked:
            raise ctypes.WinError()

    @win_only
    @require_ctypes
    def set_error_mode(mode):
        return ctypes.windll.kernel32.SetErrorMode(mode)

    @win_only
    def rmtree(path):
        def error_handler(func, handling_path, execinfo):
            e = execinfo[1]
            if e.winerror == ERRORS['PATH_NOT_FOUND']:
                handling_path = "\\\\?\\" + handling_path  # handle path over 256 symbols
                if os.path.exists(path):
                    return func(handling_path)
            if e.winerror == ERRORS['ACCESS_DENIED']:
                try:
                    # removing of r/w directory with read-only files in it yields ACCESS_DENIED
                    # which is not an insuperable obstacle https://bugs.python.org/issue19643
                    os.chmod(handling_path, stat.S_IWRITE)
                except OSError:
                    pass
                else:
                    # propagate true last error if this attempt fails
                    return func(handling_path)
            raise e

        shutil.rmtree(path, onerror=error_handler)

    # Don't display the Windows GPF dialog if the invoked program dies.
    # http://msdn.microsoft.com/en-us/library/windows/desktop/ms680621.aspx
    @win_only
    def disable_error_dialogs():
        set_error_mode(_SEM_NOGPFAULTERRORBOX | _SEM_FAILCRITICALERRORS)

    @win_only
    def default_process_creation_flags():
        return 0

    @require_ctypes
    def _low_dword(x):
        return ctypes.c_ulong(x & ((1 << 32) - 1))

    @require_ctypes
    def _high_dword(x):
        return ctypes.c_ulong((x >> 32) & ((1 << 32) - 1))

    @win_only
    @require_ctypes
    def get_current_process():
        handle = ctypes.windll.kernel32.GetCurrentProcess()
        if not handle:
            raise ctypes.WinError()
        return wintypes.HANDLE(handle)

    @win_only
    @require_ctypes
    def get_process_handle_count(proc_handle):
        assert isinstance(proc_handle, wintypes.HANDLE)

        GetProcessHandleCount = ctypes.WINFUNCTYPE(wintypes.BOOL, wintypes.HANDLE, wintypes.POINTER(wintypes.DWORD))(
            ("GetProcessHandleCount", ctypes.windll.kernel32)
        )
        hndcnt = wintypes.DWORD()
        if not GetProcessHandleCount(proc_handle, ctypes.byref(hndcnt)):
            raise ctypes.WinError()
        return hndcnt.value

    @win_only
    @require_ctypes
    def set_handle_information(file, inherit=None, protect_from_close=None):
        for flag, value in [(inherit, 1), (protect_from_close, 2)]:
            if flag is not None:
                assert isinstance(flag, bool)
                if not ctypes.windll.kernel32.SetHandleInformation(
                    file_handle(file), _low_dword(value), _low_dword(int(flag))
                ):
                    raise ctypes.WinError()

    @win_only
    @require_ctypes
    def get_windows_directory():
        buf = ctypes.create_unicode_buffer(ctypes.wintypes.MAX_PATH)
        size = ctypes.windll.kernel32.GetWindowsDirectoryW(buf, ctypes.wintypes.MAX_PATH)
        if not size:
            raise ctypes.WinError()
        if size > ctypes.wintypes.MAX_PATH - 1:
            raise CustomWinError(ERRORS['INSUFFICIENT_BUFFER'])
        return ctypes.wstring_at(buf, size)