|
|
|
@ -1,6 +1,7 @@ |
|
|
|
"""Safe discovery of local source documents for the guided import workflow.""" |
|
|
|
"""Safe discovery of local source documents for the guided import workflow.""" |
|
|
|
|
|
|
|
|
|
|
|
import ctypes |
|
|
|
import ctypes |
|
|
|
|
|
|
|
import errno |
|
|
|
import os |
|
|
|
import os |
|
|
|
import stat |
|
|
|
import stat |
|
|
|
from contextlib import contextmanager |
|
|
|
from contextlib import contextmanager |
|
|
|
@ -159,19 +160,35 @@ def _validate_name(name: str) -> None: |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def _rename_without_replacement(name: str, source_fd: int, destination_fd: int) -> None: |
|
|
|
def _rename_without_replacement(name: str, source_fd: int, destination_fd: int) -> None: |
|
|
|
# The supported Docker runtime is Linux. Fail closed if atomic no-replace |
|
|
|
# Use the Linux atomic no-replace rename where the mounted filesystem supports |
|
|
|
# rename is unavailable instead of falling back to a check-then-replace race. |
|
|
|
# it. Docker Desktop bind mounts can reject that flag, so preserve the same |
|
|
|
|
|
|
|
# no-overwrite guarantee by linking before removing the source. |
|
|
|
try: |
|
|
|
try: |
|
|
|
rename = ctypes.CDLL(None).renameat2 |
|
|
|
library = ctypes.CDLL(None, use_errno=True) |
|
|
|
|
|
|
|
rename = library.renameat2 |
|
|
|
except AttributeError: |
|
|
|
except AttributeError: |
|
|
|
raise ParseError("UNREADABLE_FILE") from None |
|
|
|
rename = None |
|
|
|
|
|
|
|
if rename is None: |
|
|
|
|
|
|
|
_link_then_unlink(name, source_fd, destination_fd) |
|
|
|
|
|
|
|
return |
|
|
|
rename.argtypes = [ctypes.c_int, ctypes.c_char_p, ctypes.c_int, ctypes.c_char_p, ctypes.c_uint] |
|
|
|
rename.argtypes = [ctypes.c_int, ctypes.c_char_p, ctypes.c_int, ctypes.c_char_p, ctypes.c_uint] |
|
|
|
rename.restype = ctypes.c_int |
|
|
|
rename.restype = ctypes.c_int |
|
|
|
encoded_name = os.fsencode(name) |
|
|
|
encoded_name = os.fsencode(name) |
|
|
|
if rename(source_fd, encoded_name, destination_fd, encoded_name, 1) != 0: # RENAME_NOREPLACE |
|
|
|
if rename(source_fd, encoded_name, destination_fd, encoded_name, 1) == 0: # RENAME_NOREPLACE |
|
|
|
raise ParseError("UNREADABLE_FILE") |
|
|
|
return |
|
|
|
|
|
|
|
if ctypes.get_errno() in {errno.EINVAL, errno.ENOSYS, errno.EOPNOTSUPP}: |
|
|
|
|
|
|
|
_link_then_unlink(name, source_fd, destination_fd) |
|
|
|
|
|
|
|
return |
|
|
|
|
|
|
|
raise ParseError("UNREADABLE_FILE") |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def _link_then_unlink(name: str, source_fd: int, destination_fd: int) -> None: |
|
|
|
|
|
|
|
try: |
|
|
|
|
|
|
|
os.link(name, name, src_dir_fd=source_fd, dst_dir_fd=destination_fd, follow_symlinks=False) |
|
|
|
|
|
|
|
os.unlink(name, dir_fd=source_fd) |
|
|
|
|
|
|
|
except OSError: |
|
|
|
|
|
|
|
raise ParseError("UNREADABLE_FILE") from None |
|
|
|
|
|
|
|
|
|
|
|
def _move_file(name: str, source: str | None, destination: str | None, *, overwrite=False) -> None: |
|
|
|
def _move_file(name: str, source: str | None, destination: str | None, *, overwrite=False) -> None: |
|
|
|
_validate_name(name) |
|
|
|
_validate_name(name) |
|
|
|
with _directory(source) as source_fd, _directory(destination) as destination_fd: |
|
|
|
with _directory(source) as source_fd, _directory(destination) as destination_fd: |
|
|
|
|