bazarr/libs/py7zr/win32compat.py

import pathlib
import stat
import sys
from logging import getLogger
from typing import Union

if sys.platform == "win32":
    import ctypes
    from ctypes.wintypes import BOOL, DWORD, HANDLE, LPCWSTR, LPDWORD, LPVOID, LPWSTR

    _stdcall_libraries = {}
    _stdcall_libraries['kernel32'] = ctypes.WinDLL('kernel32')
    CloseHandle = _stdcall_libraries['kernel32'].CloseHandle
    CreateFileW = _stdcall_libraries['kernel32'].CreateFileW
    DeviceIoControl = _stdcall_libraries['kernel32'].DeviceIoControl
    GetFileAttributesW = _stdcall_libraries['kernel32'].GetFileAttributesW
    OPEN_EXISTING = 3
    GENERIC_READ = 2147483648
    FILE_FLAG_OPEN_REPARSE_POINT = 0x00200000
    FSCTL_GET_REPARSE_POINT = 0x000900A8
    FILE_FLAG_BACKUP_SEMANTICS = 0x02000000
    IO_REPARSE_TAG_MOUNT_POINT = 0xA0000003
    IO_REPARSE_TAG_SYMLINK = 0xA000000C
    MAXIMUM_REPARSE_DATA_BUFFER_SIZE = 16 * 1024

    def _check_bit(val: int, flag: int) -> bool:
        return bool(val & flag == flag)

    class SymbolicLinkReparseBuffer(ctypes.Structure):
        """ Implementing the below in Python:

        typedef struct _REPARSE_DATA_BUFFER {
            ULONG  ReparseTag;
            USHORT ReparseDataLength;
            USHORT Reserved;
            union {
                struct {
                    USHORT SubstituteNameOffset;
                    USHORT SubstituteNameLength;
                    USHORT PrintNameOffset;
                    USHORT PrintNameLength;
                    ULONG Flags;
                    WCHAR PathBuffer[1];
                } SymbolicLinkReparseBuffer;
                struct {
                    USHORT SubstituteNameOffset;
                    USHORT SubstituteNameLength;
                    USHORT PrintNameOffset;
                    USHORT PrintNameLength;
                    WCHAR PathBuffer[1];
                } MountPointReparseBuffer;
                struct {
                    UCHAR  DataBuffer[1];
                } GenericReparseBuffer;
            } DUMMYUNIONNAME;
        } REPARSE_DATA_BUFFER, *PREPARSE_DATA_BUFFER;
        """
        # See https://docs.microsoft.com/en-us/windows-hardware/drivers/ddi/content/ntifs/ns-ntifs-_reparse_data_buffer
        _fields_ = [
            ('flags', ctypes.c_ulong),
            ('path_buffer', ctypes.c_byte * (MAXIMUM_REPARSE_DATA_BUFFER_SIZE - 20))
        ]

    class MountReparseBuffer(ctypes.Structure):
        _fields_ = [
            ('path_buffer', ctypes.c_byte * (MAXIMUM_REPARSE_DATA_BUFFER_SIZE - 16)),
        ]

    class ReparseBufferField(ctypes.Union):
        _fields_ = [
            ('symlink', SymbolicLinkReparseBuffer),
            ('mount', MountReparseBuffer)
        ]

    class ReparseBuffer(ctypes.Structure):
        _anonymous_ = ("u",)
        _fields_ = [
            ('reparse_tag', ctypes.c_ulong),
            ('reparse_data_length', ctypes.c_ushort),
            ('reserved', ctypes.c_ushort),
            ('substitute_name_offset', ctypes.c_ushort),
            ('substitute_name_length', ctypes.c_ushort),
            ('print_name_offset', ctypes.c_ushort),
            ('print_name_length', ctypes.c_ushort),
            ('u', ReparseBufferField)
        ]

    def is_reparse_point(path: Union[str, pathlib.Path]) -> bool:
        GetFileAttributesW.argtypes = [LPCWSTR]
        GetFileAttributesW.restype = DWORD
        return _check_bit(GetFileAttributesW(str(path)), stat.FILE_ATTRIBUTE_REPARSE_POINT)

    def readlink(path: Union[str, pathlib.Path]) -> Union[str, pathlib.WindowsPath]:
        # FILE_FLAG_OPEN_REPARSE_POINT alone is not enough if 'path'
        # is a symbolic link to a directory or a NTFS junction.
        # We need to set FILE_FLAG_BACKUP_SEMANTICS as well.
        # See https://docs.microsoft.com/en-us/windows/desktop/api/fileapi/nf-fileapi-createfilea

        # description from _winapi.c:601
        # /* REPARSE_DATA_BUFFER usage is heavily under-documented, especially for
        #  junction points. Here's what I've learned along the way:
        #  - A junction point has two components: a print name and a substitute
        #  name. They both describe the link target, but the substitute name is
        #  the physical target and the print name is shown in directory listings.
        #  - The print name must be a native name, prefixed with "\??\".
        #  - Both names are stored after each other in the same buffer (the
        #  PathBuffer) and both must be NUL-terminated.
        #  - There are four members defining their respective offset and length
        #  inside PathBuffer: SubstituteNameOffset, SubstituteNameLength,
        #  PrintNameOffset and PrintNameLength.
        #  - The total size we need to allocate for the REPARSE_DATA_BUFFER, thus,
        #  is the sum of:
        #  - the fixed header size (REPARSE_DATA_BUFFER_HEADER_SIZE)
        #  - the size of the MountPointReparseBuffer member without the PathBuffer
        #  - the size of the prefix ("\??\") in bytes
        #  - the size of the print name in bytes
        #  - the size of the substitute name in bytes
        #  - the size of two NUL terminators in bytes */

        target_is_path = isinstance(path, pathlib.Path)
        if target_is_path:
            target = str(path)
        else:
            target = path
        CreateFileW.argtypes = [LPWSTR, DWORD, DWORD, LPVOID, DWORD, DWORD, HANDLE]
        CreateFileW.restype = HANDLE
        DeviceIoControl.argtypes = [HANDLE, DWORD, LPVOID, DWORD, LPVOID, DWORD, LPDWORD, LPVOID]
        DeviceIoControl.restype = BOOL
        handle = HANDLE(CreateFileW(target, GENERIC_READ, 0, None, OPEN_EXISTING,
                                    FILE_FLAG_BACKUP_SEMANTICS | FILE_FLAG_OPEN_REPARSE_POINT, 0))
        buf = ReparseBuffer()
        ret = DWORD(0)
        status = DeviceIoControl(handle, FSCTL_GET_REPARSE_POINT, None, 0, ctypes.byref(buf),
                                 MAXIMUM_REPARSE_DATA_BUFFER_SIZE, ctypes.byref(ret), None)
        CloseHandle(handle)
        if not status:
            logger = getLogger(__file__)
            logger.error("Failed IOCTL access to REPARSE_POINT {})".format(target))
            raise ValueError("not a symbolic link or access permission violation")

        if buf.reparse_tag == IO_REPARSE_TAG_SYMLINK:
            offset = buf.substitute_name_offset
            ending = offset + buf.substitute_name_length
            rpath = bytearray(buf.symlink.path_buffer)[offset:ending].decode('UTF-16-LE')
        elif buf.reparse_tag == IO_REPARSE_TAG_MOUNT_POINT:
            offset = buf.substitute_name_offset
            ending = offset + buf.substitute_name_length
            rpath = bytearray(buf.mount.path_buffer)[offset:ending].decode('UTF-16-LE')
        else:
            raise ValueError("not a symbolic link")
        # on posixmodule.c:7859 in py38, we do that
        # ```
        # else if (rdb->ReparseTag == IO_REPARSE_TAG_MOUNT_POINT)
        # {
        #    name = (wchar_t *)((char*)rdb->MountPointReparseBuffer.PathBuffer +
        #                       rdb->MountPointReparseBuffer.SubstituteNameOffset);
        #    nameLen = rdb->MountPointReparseBuffer.SubstituteNameLength / sizeof(wchar_t);
        # }
        # else
        # {
        #    PyErr_SetString(PyExc_ValueError, "not a symbolic link");
        # }
        # if (nameLen > 4 && wcsncmp(name, L"\\??\\", 4) == 0) {
        #             /* Our buffer is mutable, so this is okay */
        #             name[1] = L'\\';
        #         }
        # ```
        # so substitute prefix here.
        if rpath.startswith('\\??\\'):
            rpath = '\\\\' + rpath[2:]
        if target_is_path:
            return pathlib.WindowsPath(rpath)
        else:
            return rpath