mirror of
https://github.com/morpheus65535/bazarr
synced 2024-12-23 00:03:33 +00:00
175 lines
7.6 KiB
Python
175 lines
7.6 KiB
Python
|
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
|