sashatrask
1441 lines
53 KiB
Python
1441 lines
53 KiB
Python
from contextlib import contextmanager
|
|
import ctypes
|
|
from dataclasses import dataclass, field
|
|
from enum import Enum
|
|
import errno
|
|
import hashlib
|
|
import hmac
|
|
import os
|
|
import posixpath
|
|
import re
|
|
import secrets
|
|
import stat
|
|
import sys
|
|
import tempfile
|
|
import threading
|
|
|
|
try:
|
|
import fcntl
|
|
except ImportError: # pragma: no cover - non-Linux import portability
|
|
fcntl = None
|
|
|
|
|
|
MAX_MANAGED_ROOTS = 8
|
|
MAX_MANAGED_ROOT_ID_BYTES = 64
|
|
MAX_MANAGED_ROOT_PATH_BYTES = 4096
|
|
MAX_RELATIVE_PATH_BYTES = 4096
|
|
MAX_COMPONENT_BYTES = 255
|
|
MAX_PATH_DEPTH = 32
|
|
MAX_LISTING_ENTRIES = 1000
|
|
MAX_LISTING_BYTES = 1024 * 1024
|
|
MAX_FILE_BYTES = 64 * 1024 * 1024
|
|
MAX_RESULT_FILE_BYTES = 256 * 1024 * 1024
|
|
|
|
RUNTIME_LOG_ROOT_ID = 'runtime-logs'
|
|
RUNTIME_LOG_ROOT_PATH = '/data/runtime-linux/logs'
|
|
RUNTIME_KEYCHECK_ROOT_ID = 'runtime-keychecks'
|
|
RUNTIME_KEYCHECK_ROOT_PATH = '/data/runtime-linux/keychecks'
|
|
RUNTIME_RESULT_ROOT_ID = 'runtime-results'
|
|
RUNTIME_RESULT_ROOT_PATH = '/data/runtime-linux/results'
|
|
MANAGED_DATA_ROOT = '/data/managed-files'
|
|
|
|
_PREDEFINED_READ_ONLY_ROOTS = {
|
|
RUNTIME_KEYCHECK_ROOT_ID: RUNTIME_KEYCHECK_ROOT_PATH,
|
|
RUNTIME_LOG_ROOT_ID: RUNTIME_LOG_ROOT_PATH,
|
|
RUNTIME_RESULT_ROOT_ID: RUNTIME_RESULT_ROOT_PATH,
|
|
}
|
|
|
|
_ROOT_ID = re.compile(r'^[a-z][a-z0-9-]{0,63}$')
|
|
_DRIVE_PATH = re.compile(r'^/?[A-Za-z]:')
|
|
_DRIVE_COMPONENT = re.compile(r'^[A-Za-z]:')
|
|
_HASH = re.compile(r'^[0-9a-f]{64}$')
|
|
_RESULT_PROJECTION_FILE = re.compile(
|
|
r'^(?:found_secrets|scan_results)(?:\.g[0-9]{6})?\.jsonl$',
|
|
)
|
|
_TEMPORARY_PREFIX = '.truf-managed-file-'
|
|
_TEMPORARY_ATTEMPTS = 16
|
|
_READ_CHUNK_BYTES = 64 * 1024
|
|
_RENAME_NOREPLACE = 1
|
|
_PERMISSION_KEYS = frozenset({'list', 'read', 'create_replace', 'delete'})
|
|
_LIMIT_BOUNDS = {
|
|
'max_relative_path_bytes': MAX_RELATIVE_PATH_BYTES,
|
|
'max_component_bytes': MAX_COMPONENT_BYTES,
|
|
'max_path_depth': MAX_PATH_DEPTH,
|
|
'max_listing_entries': MAX_LISTING_ENTRIES,
|
|
'max_listing_bytes': MAX_LISTING_BYTES,
|
|
'max_file_bytes': MAX_FILE_BYTES,
|
|
}
|
|
|
|
|
|
class ManagedFileConfigurationError(ValueError):
|
|
def __init__(self, category, field):
|
|
self.category = category
|
|
self.field = tuple(field)
|
|
super().__init__('managed file root configuration is invalid')
|
|
|
|
|
|
class ManagedFileAccessError(RuntimeError):
|
|
def __init__(self, category):
|
|
self.category = category
|
|
super().__init__('managed file access failed')
|
|
|
|
|
|
class ManagedFileOperation(str, Enum):
|
|
LIST = 'list'
|
|
READ = 'read'
|
|
CREATE_REPLACE = 'create-replace'
|
|
DELETE = 'delete'
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ManagedFilePermissions:
|
|
allow_list: bool
|
|
allow_read: bool
|
|
allow_create_replace: bool
|
|
allow_delete: bool
|
|
|
|
def allows(self, operation):
|
|
if operation is ManagedFileOperation.LIST:
|
|
return self.allow_list
|
|
if operation is ManagedFileOperation.READ:
|
|
return self.allow_read
|
|
if operation is ManagedFileOperation.CREATE_REPLACE:
|
|
return self.allow_create_replace
|
|
if operation is ManagedFileOperation.DELETE:
|
|
return self.allow_delete
|
|
return False
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ManagedFileLimits:
|
|
max_relative_path_bytes: int
|
|
max_component_bytes: int
|
|
max_path_depth: int
|
|
max_listing_entries: int
|
|
max_listing_bytes: int
|
|
max_file_bytes: int
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ManagedFileRoot:
|
|
root_id: str
|
|
absolute_path: str = field(repr=False)
|
|
permissions: ManagedFilePermissions
|
|
limits: ManagedFileLimits
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ManagedFileRootRegistry:
|
|
roots: tuple[ManagedFileRoot, ...] = ()
|
|
|
|
def get(self, root_id):
|
|
if type(root_id) is not str:
|
|
return None
|
|
return next((root for root in self.roots if root.root_id == root_id), None)
|
|
|
|
def root_ids(self):
|
|
return tuple(root.root_id for root in self.roots)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ManagedFileIdentity:
|
|
sha256: str
|
|
byte_count: int
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ManagedFileDirectoryEntry:
|
|
name: str
|
|
kind: str
|
|
byte_count: int | None
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ManagedFileListing:
|
|
entries: tuple[ManagedFileDirectoryEntry, ...]
|
|
name_bytes: int
|
|
|
|
|
|
class ManagedFileSnapshot:
|
|
__slots__ = ('_handle', '_release', '_lock')
|
|
|
|
def __init__(self, handle, release):
|
|
self._handle = handle
|
|
self._release = release
|
|
self._lock = threading.Lock()
|
|
|
|
def chunks(self):
|
|
try:
|
|
while True:
|
|
with self._lock:
|
|
handle = self._handle
|
|
chunk = None if handle is None else handle.read(_READ_CHUNK_BYTES)
|
|
if not chunk:
|
|
break
|
|
yield chunk
|
|
finally:
|
|
self.close()
|
|
|
|
def close(self):
|
|
with self._lock:
|
|
handle = self._handle
|
|
release = self._release
|
|
self._handle = None
|
|
self._release = None
|
|
if handle is not None:
|
|
try:
|
|
handle.close()
|
|
except OSError:
|
|
pass
|
|
if release is not None:
|
|
release(self)
|
|
|
|
def __repr__(self):
|
|
return '<ManagedFileSnapshot>'
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ManagedFileDownload:
|
|
identity: ManagedFileIdentity
|
|
content: bytes | None = field(repr=False, default=None)
|
|
snapshot: ManagedFileSnapshot | None = field(repr=False, default=None)
|
|
|
|
def __post_init__(self):
|
|
if (self.content is None) == (self.snapshot is None):
|
|
raise ValueError('managed file download requires one body source')
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ManagedFileMutation:
|
|
before: ManagedFileIdentity | None
|
|
after: ManagedFileIdentity | None
|
|
written: bool
|
|
|
|
|
|
class ManagedFileOpenedTarget:
|
|
__slots__ = ('_descriptor', '_details')
|
|
|
|
def __init__(self, descriptor, details):
|
|
self._descriptor = descriptor
|
|
self._details = details
|
|
|
|
def fileno(self):
|
|
if self._descriptor is None:
|
|
raise ManagedFileAccessError('closed')
|
|
return self._descriptor
|
|
|
|
@property
|
|
def details(self):
|
|
return self._details
|
|
|
|
def _close(self):
|
|
descriptor = self._descriptor
|
|
self._descriptor = None
|
|
if descriptor is not None:
|
|
try:
|
|
os.close(descriptor)
|
|
except OSError:
|
|
pass
|
|
|
|
def __repr__(self):
|
|
return '<ManagedFileOpenedTarget>'
|
|
|
|
|
|
def _error(category, *field):
|
|
raise ManagedFileConfigurationError(category, ('root',) + field)
|
|
|
|
|
|
def _absolute_root_path(value):
|
|
if type(value) is not str:
|
|
_error('type', 'path')
|
|
try:
|
|
encoded = value.encode('utf-8')
|
|
except UnicodeEncodeError:
|
|
_error('bounds', 'path')
|
|
if (
|
|
not value or len(encoded) > MAX_MANAGED_ROOT_PATH_BYTES
|
|
or '\\' in value or '\x00' in value or _DRIVE_PATH.match(value)
|
|
or not value.startswith('/') or value == '/' or value.endswith('/')
|
|
or '//' in value
|
|
):
|
|
_error('deployment_path', 'path')
|
|
components = value.split('/')[1:]
|
|
if any(
|
|
component in ('', '.', '..')
|
|
or len(component.encode('utf-8')) > MAX_COMPONENT_BYTES
|
|
for component in components
|
|
) or posixpath.normpath(value) != value:
|
|
_error('deployment_path', 'path')
|
|
return value
|
|
|
|
|
|
def _permissions(value):
|
|
if type(value) is not dict:
|
|
_error('type', 'permissions')
|
|
if set(value) != _PERMISSION_KEYS:
|
|
category = 'unknown_key' if set(value) - _PERMISSION_KEYS else 'schema'
|
|
_error(category, 'permissions')
|
|
if any(type(value[name]) is not bool for name in _PERMISSION_KEYS):
|
|
_error('type', 'permissions')
|
|
return ManagedFilePermissions(
|
|
allow_list=value['list'],
|
|
allow_read=value['read'],
|
|
allow_create_replace=value['create_replace'],
|
|
allow_delete=value['delete'],
|
|
)
|
|
|
|
|
|
def _limits(value, root_id):
|
|
if type(value) is not dict:
|
|
_error('type', 'limits')
|
|
expected = set(_LIMIT_BOUNDS)
|
|
if set(value) != expected:
|
|
category = 'unknown_key' if set(value) - expected else 'schema'
|
|
_error(category, 'limits')
|
|
for name, maximum in _LIMIT_BOUNDS.items():
|
|
if name == 'max_file_bytes' and root_id == RUNTIME_RESULT_ROOT_ID:
|
|
maximum = MAX_RESULT_FILE_BYTES
|
|
if type(value[name]) is not int or not 1 <= value[name] <= maximum:
|
|
_error('bounds', 'limits', name)
|
|
if (
|
|
value['max_component_bytes'] > value['max_relative_path_bytes']
|
|
or value['max_listing_bytes'] < value['max_component_bytes']
|
|
):
|
|
_error('bounds', 'limits')
|
|
return ManagedFileLimits(**{name: value[name] for name in _LIMIT_BOUNDS})
|
|
|
|
|
|
def _allowed_root(root_id, path, permissions):
|
|
predefined_path = _PREDEFINED_READ_ONLY_ROOTS.get(root_id)
|
|
predefined_id = next((
|
|
candidate_id for candidate_id, candidate_path
|
|
in _PREDEFINED_READ_ONLY_ROOTS.items()
|
|
if candidate_path == path
|
|
), None)
|
|
if predefined_path is not None or predefined_id is not None:
|
|
if predefined_path != path or permissions != ManagedFilePermissions(
|
|
allow_list=True,
|
|
allow_read=True,
|
|
allow_create_replace=False,
|
|
allow_delete=False,
|
|
):
|
|
_error('deployment_path', 'path')
|
|
return
|
|
parent, name = posixpath.split(path)
|
|
if parent != MANAGED_DATA_ROOT or not name:
|
|
_error('deployment_path', 'path')
|
|
|
|
|
|
def normalize_managed_file_roots(value):
|
|
if type(value) is not dict:
|
|
raise ManagedFileConfigurationError('type', ('root',))
|
|
if len(value) > MAX_MANAGED_ROOTS:
|
|
raise ManagedFileConfigurationError('bounds', ('root',))
|
|
|
|
if any(type(root_id) is not str for root_id in value):
|
|
_error('bounds', 'id')
|
|
normalized = {}
|
|
roots = []
|
|
paths = set()
|
|
for root_id in sorted(value):
|
|
if (
|
|
type(root_id) is not str or not _ROOT_ID.fullmatch(root_id)
|
|
or len(root_id.encode('utf-8')) > MAX_MANAGED_ROOT_ID_BYTES
|
|
):
|
|
_error('bounds', 'id')
|
|
raw = value[root_id]
|
|
if type(raw) is not dict:
|
|
_error('type')
|
|
expected = {'path', 'permissions', 'limits'}
|
|
if set(raw) != expected:
|
|
category = 'unknown_key' if set(raw) - expected else 'schema'
|
|
_error(category)
|
|
path = _absolute_root_path(raw['path'])
|
|
permissions = _permissions(raw['permissions'])
|
|
limits = _limits(raw['limits'], root_id)
|
|
_allowed_root(root_id, path, permissions)
|
|
if path in paths:
|
|
_error('deployment_path', 'path')
|
|
paths.add(path)
|
|
root = ManagedFileRoot(root_id, path, permissions, limits)
|
|
roots.append(root)
|
|
normalized[root_id] = {
|
|
'path': path,
|
|
'permissions': {
|
|
'list': permissions.allow_list,
|
|
'read': permissions.allow_read,
|
|
'create_replace': permissions.allow_create_replace,
|
|
'delete': permissions.allow_delete,
|
|
},
|
|
'limits': {name: getattr(limits, name) for name in _LIMIT_BOUNDS},
|
|
}
|
|
return normalized, ManagedFileRootRegistry(tuple(roots))
|
|
|
|
|
|
def managed_file_root_registry_from_config(config):
|
|
if type(config) is not dict:
|
|
raise ManagedFileConfigurationError('type', ('root',))
|
|
supervisor = config.get('supervisor', {})
|
|
if type(supervisor) is not dict:
|
|
raise ManagedFileConfigurationError('type', ('root',))
|
|
worker_api = supervisor.get('worker_api', {})
|
|
if type(worker_api) is not dict:
|
|
raise ManagedFileConfigurationError('type', ('root',))
|
|
admin = worker_api.get('admin', {})
|
|
if admin is None:
|
|
admin = {}
|
|
if type(admin) is not dict:
|
|
raise ManagedFileConfigurationError('type', ('root',))
|
|
return normalize_managed_file_roots(admin.get('managed_file_roots', {}))[1]
|
|
|
|
|
|
def parse_managed_relative_path(value, limits):
|
|
if type(value) is not str or not isinstance(limits, ManagedFileLimits):
|
|
raise ManagedFileAccessError('invalid_path')
|
|
try:
|
|
encoded = value.encode('utf-8')
|
|
except UnicodeEncodeError:
|
|
raise ManagedFileAccessError('invalid_path') from None
|
|
if (
|
|
not value or len(encoded) > limits.max_relative_path_bytes
|
|
or value.startswith('/') or '\\' in value or '\x00' in value
|
|
or _DRIVE_PATH.match(value)
|
|
):
|
|
raise ManagedFileAccessError('invalid_path')
|
|
components = value.split('/')
|
|
if (
|
|
len(components) > limits.max_path_depth
|
|
or any(
|
|
component in ('', '.', '..')
|
|
or _DRIVE_COMPONENT.match(component)
|
|
or component.startswith(_TEMPORARY_PREFIX)
|
|
or len(component.encode('utf-8')) > limits.max_component_bytes
|
|
for component in components
|
|
)
|
|
):
|
|
raise ManagedFileAccessError('invalid_path')
|
|
return tuple(components)
|
|
|
|
|
|
def _close_descriptor(descriptor):
|
|
if descriptor is None:
|
|
return
|
|
try:
|
|
os.close(descriptor)
|
|
except OSError:
|
|
pass
|
|
|
|
|
|
def _descriptor_flags():
|
|
required = (
|
|
'O_PATH', 'O_DIRECTORY', 'O_NOFOLLOW', 'O_CLOEXEC',
|
|
'O_NONBLOCK', 'O_NOCTTY', 'O_CREAT', 'O_EXCL',
|
|
)
|
|
if (
|
|
not sys.platform.startswith('linux')
|
|
or os.open not in getattr(os, 'supports_dir_fd', ())
|
|
or any(type(getattr(os, name, None)) is not int for name in required)
|
|
):
|
|
raise ManagedFileAccessError('filesystem_unavailable')
|
|
return {
|
|
'directory': (
|
|
os.O_PATH | os.O_DIRECTORY | os.O_NOFOLLOW | os.O_CLOEXEC
|
|
),
|
|
'list': os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | os.O_CLOEXEC,
|
|
'inspect': os.O_PATH | os.O_NOFOLLOW | os.O_CLOEXEC,
|
|
'read': os.O_RDONLY | os.O_NONBLOCK | os.O_NOCTTY | os.O_CLOEXEC,
|
|
'temporary': (
|
|
os.O_RDWR | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW | os.O_CLOEXEC
|
|
),
|
|
}
|
|
|
|
|
|
def _filesystem_error_category(exc):
|
|
if exc.errno == errno.ENOENT:
|
|
return 'not_found'
|
|
if exc.errno in {
|
|
errno.ELOOP, errno.ENOTDIR, errno.EISDIR, errno.ENXIO,
|
|
errno.ENODEV, errno.EMLINK,
|
|
getattr(errno, 'EOPNOTSUPP', -1),
|
|
getattr(errno, 'ENOTSUP', -1),
|
|
}:
|
|
return 'unsafe_target'
|
|
return 'filesystem_unavailable'
|
|
|
|
|
|
def _file_revision(details):
|
|
return (
|
|
details.st_dev,
|
|
details.st_ino,
|
|
stat.S_IFMT(details.st_mode),
|
|
details.st_nlink,
|
|
details.st_size,
|
|
getattr(details, 'st_mtime_ns', None),
|
|
getattr(details, 'st_ctime_ns', None),
|
|
)
|
|
|
|
|
|
def _regular_single_link(details):
|
|
return stat.S_ISREG(details.st_mode) and details.st_nlink == 1
|
|
|
|
|
|
def _valid_hash(value):
|
|
if type(value) is not str or not _HASH.fullmatch(value):
|
|
raise ManagedFileAccessError('invalid_hash')
|
|
return value
|
|
|
|
|
|
def _rename_noreplace(source, target, directory_descriptor):
|
|
try:
|
|
function = ctypes.CDLL(None, use_errno=True).renameat2
|
|
function.argtypes = (
|
|
ctypes.c_int, ctypes.c_char_p, ctypes.c_int, ctypes.c_char_p,
|
|
ctypes.c_uint,
|
|
)
|
|
function.restype = ctypes.c_int
|
|
ctypes.set_errno(0)
|
|
result = function(
|
|
directory_descriptor, source.encode('utf-8'),
|
|
directory_descriptor, target.encode('utf-8'),
|
|
_RENAME_NOREPLACE,
|
|
)
|
|
except (AttributeError, OSError, TypeError, ValueError):
|
|
raise ManagedFileAccessError('filesystem_unavailable') from None
|
|
if result == 0:
|
|
return
|
|
error_number = ctypes.get_errno()
|
|
if error_number == errno.EEXIST:
|
|
raise ManagedFileAccessError('hash_conflict')
|
|
raise ManagedFileAccessError('filesystem_unavailable')
|
|
|
|
|
|
class ManagedFileTraversal:
|
|
def __init__(self, registry):
|
|
if not isinstance(registry, ManagedFileRootRegistry):
|
|
raise ValueError('managed file root registry is invalid')
|
|
if any(not isinstance(root, ManagedFileRoot) for root in registry.roots):
|
|
raise ValueError('managed file root registry is invalid')
|
|
root_ids = tuple(root.root_id for root in registry.roots)
|
|
try:
|
|
roots_are_valid = all(
|
|
_ROOT_ID.fullmatch(root.root_id)
|
|
and _absolute_root_path(root.absolute_path) == root.absolute_path
|
|
and isinstance(root.permissions, ManagedFilePermissions)
|
|
and isinstance(root.limits, ManagedFileLimits)
|
|
for root in registry.roots
|
|
)
|
|
except (AttributeError, ManagedFileConfigurationError, TypeError):
|
|
roots_are_valid = False
|
|
if not roots_are_valid or len(root_ids) != len(set(root_ids)):
|
|
raise ValueError('managed file root registry is invalid')
|
|
|
|
self._registry = registry
|
|
self._lock = threading.Lock()
|
|
self._mutation_lock = threading.Lock()
|
|
self._snapshot_gate = threading.BoundedSemaphore(1)
|
|
self._snapshots = set()
|
|
self._closed = False
|
|
self._roots = {}
|
|
self._flags = None
|
|
if not registry.roots:
|
|
return
|
|
|
|
self._flags = _descriptor_flags()
|
|
try:
|
|
for root in registry.roots:
|
|
self._roots[root.root_id] = self._open_root(root.absolute_path)
|
|
except BaseException:
|
|
descriptors = tuple(self._roots.values())
|
|
self._roots.clear()
|
|
self._closed = True
|
|
for descriptor in descriptors:
|
|
_close_descriptor(descriptor)
|
|
raise
|
|
|
|
def _open_root(self, absolute_path):
|
|
current = None
|
|
try:
|
|
current = os.open('/', self._flags['directory'])
|
|
if not stat.S_ISDIR(os.fstat(current).st_mode):
|
|
raise ManagedFileAccessError('root_unavailable')
|
|
for component in absolute_path.split('/')[1:]:
|
|
child = None
|
|
try:
|
|
child = os.open(
|
|
component, self._flags['directory'], dir_fd=current,
|
|
)
|
|
if not stat.S_ISDIR(os.fstat(child).st_mode):
|
|
raise ManagedFileAccessError('root_unavailable')
|
|
except BaseException:
|
|
_close_descriptor(child)
|
|
raise
|
|
previous = current
|
|
current = child
|
|
child = None
|
|
_close_descriptor(previous)
|
|
descriptor = current
|
|
current = None
|
|
return descriptor
|
|
except OSError:
|
|
raise ManagedFileAccessError('root_unavailable') from None
|
|
finally:
|
|
_close_descriptor(current)
|
|
|
|
def close(self):
|
|
with self._lock:
|
|
if self._closed:
|
|
return
|
|
self._closed = True
|
|
descriptors = tuple(self._roots.values())
|
|
snapshots = tuple(self._snapshots)
|
|
self._roots.clear()
|
|
self._snapshots.clear()
|
|
for descriptor in descriptors:
|
|
_close_descriptor(descriptor)
|
|
for snapshot in snapshots:
|
|
snapshot.close()
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, exc_type, exc_value, traceback):
|
|
self.close()
|
|
|
|
def _root_for_operation(self, root_id, operation):
|
|
root = self._registry.get(root_id)
|
|
if root is None:
|
|
raise ManagedFileAccessError('unknown_root')
|
|
if not root.permissions.allows(operation):
|
|
raise ManagedFileAccessError('operation_not_allowed')
|
|
return root
|
|
|
|
def _duplicate_root(self, root_id):
|
|
with self._lock:
|
|
if self._closed:
|
|
raise ManagedFileAccessError('closed')
|
|
descriptor = self._roots.get(root_id)
|
|
if descriptor is None:
|
|
raise ManagedFileAccessError('root_unavailable')
|
|
try:
|
|
return os.dup(descriptor)
|
|
except OSError:
|
|
raise ManagedFileAccessError('root_unavailable') from None
|
|
|
|
@staticmethod
|
|
def _relative_target_allowed(root_id, components, target_kind):
|
|
if root_id != RUNTIME_RESULT_ROOT_ID:
|
|
return True
|
|
if target_kind == 'directory':
|
|
return not components
|
|
return (
|
|
len(components) == 1
|
|
and _RESULT_PROJECTION_FILE.fullmatch(components[0]) is not None
|
|
)
|
|
|
|
def _require_relative_target(self, root_id, components, target_kind):
|
|
if not self._relative_target_allowed(root_id, components, target_kind):
|
|
raise ManagedFileAccessError('not_found')
|
|
|
|
def _snapshot_closed(self, snapshot):
|
|
with self._lock:
|
|
self._snapshots.discard(snapshot)
|
|
self._snapshot_gate.release()
|
|
|
|
def _new_snapshot_file(self, root_id):
|
|
root_descriptor = self._duplicate_root(root_id)
|
|
try:
|
|
return tempfile.TemporaryFile(
|
|
mode='w+b', dir=f'/proc/self/fd/{root_descriptor}',
|
|
)
|
|
except OSError:
|
|
raise ManagedFileAccessError('filesystem_unavailable') from None
|
|
finally:
|
|
_close_descriptor(root_descriptor)
|
|
|
|
@staticmethod
|
|
def _write_snapshot(handle, payload):
|
|
view = memoryview(payload)
|
|
written = 0
|
|
while written < len(view):
|
|
try:
|
|
count = handle.write(view[written:])
|
|
except InterruptedError:
|
|
continue
|
|
except OSError:
|
|
raise ManagedFileAccessError('filesystem_unavailable') from None
|
|
if not count:
|
|
raise ManagedFileAccessError('filesystem_unavailable')
|
|
written += count
|
|
|
|
def _snapshot_descriptor(self, root, descriptor):
|
|
handle = None
|
|
try:
|
|
before = os.fstat(descriptor)
|
|
if not _regular_single_link(before):
|
|
raise ManagedFileAccessError('unsafe_target')
|
|
if before.st_size > root.limits.max_file_bytes:
|
|
raise ManagedFileAccessError('limit_exceeded')
|
|
handle = self._new_snapshot_file(root.root_id)
|
|
os.lseek(descriptor, 0, os.SEEK_SET)
|
|
digest = hashlib.sha256()
|
|
byte_count = 0
|
|
while True:
|
|
remaining = root.limits.max_file_bytes - byte_count
|
|
if remaining < 0:
|
|
raise ManagedFileAccessError('limit_exceeded')
|
|
try:
|
|
chunk = os.read(
|
|
descriptor, min(_READ_CHUNK_BYTES, remaining + 1),
|
|
)
|
|
except InterruptedError:
|
|
continue
|
|
except OSError:
|
|
raise ManagedFileAccessError('filesystem_unavailable') from None
|
|
if not chunk:
|
|
break
|
|
byte_count += len(chunk)
|
|
if byte_count > root.limits.max_file_bytes:
|
|
raise ManagedFileAccessError('limit_exceeded')
|
|
digest.update(chunk)
|
|
self._write_snapshot(handle, chunk)
|
|
after = os.fstat(descriptor)
|
|
if (
|
|
not _regular_single_link(after)
|
|
or _file_revision(before) != _file_revision(after)
|
|
or byte_count != before.st_size
|
|
):
|
|
raise ManagedFileAccessError('concurrent_change')
|
|
handle.flush()
|
|
handle.seek(0)
|
|
with self._lock:
|
|
if self._closed:
|
|
raise ManagedFileAccessError('closed')
|
|
snapshot = ManagedFileSnapshot(handle, self._snapshot_closed)
|
|
self._snapshots.add(snapshot)
|
|
handle = None
|
|
return ManagedFileIdentity(digest.hexdigest(), byte_count), snapshot
|
|
except OSError:
|
|
raise ManagedFileAccessError('filesystem_unavailable') from None
|
|
finally:
|
|
if handle is not None:
|
|
try:
|
|
handle.close()
|
|
except OSError:
|
|
pass
|
|
|
|
def _open_relative(self, root, components, target_kind):
|
|
current = self._duplicate_root(root.root_id)
|
|
leaf = None
|
|
try:
|
|
for component in components[:-1]:
|
|
child = None
|
|
try:
|
|
child = os.open(
|
|
component, self._flags['directory'], dir_fd=current,
|
|
)
|
|
if not stat.S_ISDIR(os.fstat(child).st_mode):
|
|
raise ManagedFileAccessError('unsafe_target')
|
|
except BaseException:
|
|
_close_descriptor(child)
|
|
raise
|
|
previous = current
|
|
current = child
|
|
child = None
|
|
_close_descriptor(previous)
|
|
|
|
leaf_name = components[-1] if components else '.'
|
|
if target_kind == 'directory':
|
|
leaf = os.open(leaf_name, self._flags['list'], dir_fd=current)
|
|
details = os.fstat(leaf)
|
|
safe = stat.S_ISDIR(details.st_mode)
|
|
else:
|
|
leaf, details = self._open_file_at(current, leaf_name)
|
|
safe = True
|
|
if not safe:
|
|
raise ManagedFileAccessError('unsafe_target')
|
|
target = ManagedFileOpenedTarget(leaf, details)
|
|
leaf = None
|
|
return target
|
|
except OSError as exc:
|
|
raise ManagedFileAccessError(_filesystem_error_category(exc)) from None
|
|
finally:
|
|
_close_descriptor(leaf)
|
|
_close_descriptor(current)
|
|
|
|
def _open_file_at(self, directory_descriptor, leaf_name):
|
|
inspected = readable = None
|
|
try:
|
|
inspected = os.open(
|
|
leaf_name, self._flags['inspect'], dir_fd=directory_descriptor,
|
|
)
|
|
inspected_details = os.fstat(inspected)
|
|
if not _regular_single_link(inspected_details):
|
|
raise ManagedFileAccessError('unsafe_target')
|
|
try:
|
|
readable = os.open(
|
|
f'/proc/self/fd/{inspected}', self._flags['read'],
|
|
)
|
|
readable_details = os.fstat(readable)
|
|
except OSError:
|
|
raise ManagedFileAccessError('filesystem_unavailable') from None
|
|
if (
|
|
not _regular_single_link(readable_details)
|
|
or readable_details.st_dev != inspected_details.st_dev
|
|
or readable_details.st_ino != inspected_details.st_ino
|
|
):
|
|
raise ManagedFileAccessError('unsafe_target')
|
|
result = readable
|
|
readable = None
|
|
return result, readable_details
|
|
finally:
|
|
_close_descriptor(readable)
|
|
_close_descriptor(inspected)
|
|
|
|
@contextmanager
|
|
def _opened_parent(self, root, components):
|
|
current = operational = None
|
|
try:
|
|
current = self._duplicate_root(root.root_id)
|
|
for component in components[:-1]:
|
|
child = None
|
|
try:
|
|
child = os.open(
|
|
component, self._flags['directory'], dir_fd=current,
|
|
)
|
|
if not stat.S_ISDIR(os.fstat(child).st_mode):
|
|
raise ManagedFileAccessError('unsafe_target')
|
|
except BaseException:
|
|
_close_descriptor(child)
|
|
raise
|
|
previous = current
|
|
current = child
|
|
child = None
|
|
_close_descriptor(previous)
|
|
anchored = os.fstat(current)
|
|
operational = os.open('.', self._flags['list'], dir_fd=current)
|
|
opened = os.fstat(operational)
|
|
if (
|
|
not stat.S_ISDIR(opened.st_mode)
|
|
or opened.st_dev != anchored.st_dev
|
|
or opened.st_ino != anchored.st_ino
|
|
):
|
|
raise ManagedFileAccessError('unsafe_target')
|
|
descriptor = operational
|
|
operational = None
|
|
try:
|
|
yield descriptor, components[-1]
|
|
finally:
|
|
_close_descriptor(descriptor)
|
|
except OSError as exc:
|
|
raise ManagedFileAccessError(_filesystem_error_category(exc)) from None
|
|
finally:
|
|
_close_descriptor(operational)
|
|
_close_descriptor(current)
|
|
|
|
def _read_descriptor(self, descriptor, max_bytes, *, include_content):
|
|
payload = bytearray() if include_content else None
|
|
try:
|
|
before = os.fstat(descriptor)
|
|
if not _regular_single_link(before):
|
|
raise ManagedFileAccessError('unsafe_target')
|
|
if before.st_size > max_bytes:
|
|
raise ManagedFileAccessError('limit_exceeded')
|
|
try:
|
|
os.lseek(descriptor, 0, os.SEEK_SET)
|
|
except OSError:
|
|
raise ManagedFileAccessError('filesystem_unavailable') from None
|
|
digest = hashlib.sha256()
|
|
byte_count = 0
|
|
while True:
|
|
remaining = max_bytes - byte_count
|
|
if remaining < 0:
|
|
raise ManagedFileAccessError('limit_exceeded')
|
|
try:
|
|
chunk = os.read(
|
|
descriptor, min(_READ_CHUNK_BYTES, remaining + 1),
|
|
)
|
|
except InterruptedError:
|
|
continue
|
|
except OSError:
|
|
raise ManagedFileAccessError('filesystem_unavailable') from None
|
|
if not chunk:
|
|
break
|
|
byte_count += len(chunk)
|
|
if byte_count > max_bytes:
|
|
raise ManagedFileAccessError('limit_exceeded')
|
|
digest.update(chunk)
|
|
if payload is not None:
|
|
payload.extend(chunk)
|
|
after = os.fstat(descriptor)
|
|
if (
|
|
not _regular_single_link(after)
|
|
or _file_revision(before) != _file_revision(after)
|
|
or byte_count != before.st_size
|
|
):
|
|
raise ManagedFileAccessError('concurrent_change')
|
|
identity = ManagedFileIdentity(digest.hexdigest(), byte_count)
|
|
content = bytes(payload) if payload is not None else None
|
|
if payload is not None:
|
|
payload.clear()
|
|
return identity, before, content
|
|
except OSError:
|
|
raise ManagedFileAccessError('filesystem_unavailable') from None
|
|
except BaseException:
|
|
if payload is not None:
|
|
payload.clear()
|
|
raise
|
|
|
|
def _read_file_at(self, directory_descriptor, leaf_name, max_bytes, *, content):
|
|
descriptor = None
|
|
try:
|
|
descriptor, details = self._open_file_at(
|
|
directory_descriptor, leaf_name,
|
|
)
|
|
identity, stable, payload = self._read_descriptor(
|
|
descriptor, max_bytes, include_content=content,
|
|
)
|
|
if _file_revision(details) != _file_revision(stable):
|
|
raise ManagedFileAccessError('concurrent_change')
|
|
return identity, stable, payload
|
|
except OSError as exc:
|
|
raise ManagedFileAccessError(_filesystem_error_category(exc)) from None
|
|
finally:
|
|
_close_descriptor(descriptor)
|
|
|
|
def _revalidate_named_file(self, directory_descriptor, leaf_name, expected):
|
|
descriptor = None
|
|
try:
|
|
descriptor = os.open(
|
|
leaf_name, self._flags['inspect'], dir_fd=directory_descriptor,
|
|
)
|
|
current = os.fstat(descriptor)
|
|
if (
|
|
not _regular_single_link(current)
|
|
or _file_revision(current) != _file_revision(expected)
|
|
):
|
|
raise ManagedFileAccessError('concurrent_change')
|
|
return current
|
|
except ManagedFileAccessError:
|
|
raise
|
|
except OSError:
|
|
raise ManagedFileAccessError('concurrent_change') from None
|
|
finally:
|
|
_close_descriptor(descriptor)
|
|
|
|
def _new_temporary_file(self, directory_descriptor):
|
|
for _ in range(_TEMPORARY_ATTEMPTS):
|
|
name = f'{_TEMPORARY_PREFIX}{secrets.token_hex(12)}.tmp'
|
|
try:
|
|
descriptor = os.open(
|
|
name, self._flags['temporary'], 0o600,
|
|
dir_fd=directory_descriptor,
|
|
)
|
|
return name, descriptor
|
|
except FileExistsError:
|
|
continue
|
|
except OSError:
|
|
raise ManagedFileAccessError('filesystem_unavailable') from None
|
|
raise ManagedFileAccessError('filesystem_unavailable')
|
|
|
|
def _stage_temporary_file(self, directory_descriptor, payload, max_bytes):
|
|
name = descriptor = view = None
|
|
try:
|
|
name, descriptor = self._new_temporary_file(directory_descriptor)
|
|
os.fchmod(descriptor, 0o600)
|
|
details = os.fstat(descriptor)
|
|
if (
|
|
not _regular_single_link(details)
|
|
or details.st_uid != os.geteuid()
|
|
or stat.S_IMODE(details.st_mode) != 0o600
|
|
):
|
|
raise ManagedFileAccessError('unsafe_target')
|
|
view = memoryview(payload)
|
|
written = 0
|
|
while written < len(view):
|
|
try:
|
|
count = os.write(descriptor, view[written:])
|
|
except InterruptedError:
|
|
continue
|
|
except OSError:
|
|
raise ManagedFileAccessError('filesystem_unavailable') from None
|
|
if count <= 0:
|
|
raise ManagedFileAccessError('filesystem_unavailable')
|
|
written += count
|
|
try:
|
|
os.fsync(descriptor)
|
|
except OSError:
|
|
raise ManagedFileAccessError('filesystem_unavailable') from None
|
|
identity, stable, _ = self._read_descriptor(
|
|
descriptor, max_bytes, include_content=False,
|
|
)
|
|
if (
|
|
identity.byte_count != len(payload)
|
|
or stable.st_uid != os.geteuid()
|
|
or stat.S_IMODE(stable.st_mode) != 0o600
|
|
):
|
|
raise ManagedFileAccessError('unsafe_target')
|
|
return name, descriptor, identity, stable
|
|
except OSError:
|
|
failure = ManagedFileAccessError('filesystem_unavailable')
|
|
self._release_unpublished_temporary(
|
|
directory_descriptor, name, descriptor, failure,
|
|
)
|
|
descriptor = None
|
|
raise failure from None
|
|
except BaseException as failure:
|
|
self._release_unpublished_temporary(
|
|
directory_descriptor, name, descriptor, failure,
|
|
)
|
|
descriptor = None
|
|
raise
|
|
finally:
|
|
if view is not None:
|
|
view.release()
|
|
payload = None
|
|
|
|
def _release_unpublished_temporary(
|
|
self, directory_descriptor, name, descriptor, failure=None):
|
|
cleanup_failure = close_failure = None
|
|
if name is not None:
|
|
try:
|
|
self._cleanup_temporary_file(directory_descriptor, name)
|
|
except BaseException as exc:
|
|
cleanup_failure = exc
|
|
try:
|
|
_close_descriptor(descriptor)
|
|
except BaseException as exc:
|
|
close_failure = exc
|
|
if failure is not None:
|
|
if isinstance(failure, Exception):
|
|
if cleanup_failure is not None:
|
|
raise cleanup_failure
|
|
if close_failure is not None:
|
|
raise close_failure
|
|
return
|
|
if cleanup_failure is not None:
|
|
raise cleanup_failure
|
|
if close_failure is not None:
|
|
raise close_failure
|
|
|
|
@staticmethod
|
|
def _cleanup_temporary_file(directory_descriptor, name):
|
|
try:
|
|
os.unlink(name, dir_fd=directory_descriptor)
|
|
except FileNotFoundError:
|
|
pass
|
|
except OSError:
|
|
raise ManagedFileAccessError('durability_uncertain') from None
|
|
try:
|
|
os.fsync(directory_descriptor)
|
|
except OSError:
|
|
raise ManagedFileAccessError('durability_uncertain') from None
|
|
|
|
@staticmethod
|
|
def _fsync_directory(directory_descriptor):
|
|
try:
|
|
os.fsync(directory_descriptor)
|
|
except OSError:
|
|
raise ManagedFileAccessError('durability_uncertain') from None
|
|
|
|
@staticmethod
|
|
def _lock_mutation_directory(directory_descriptor):
|
|
if fcntl is None:
|
|
raise ManagedFileAccessError('filesystem_unavailable')
|
|
while True:
|
|
try:
|
|
fcntl.flock(directory_descriptor, fcntl.LOCK_EX)
|
|
return
|
|
except InterruptedError:
|
|
continue
|
|
except OSError:
|
|
raise ManagedFileAccessError(
|
|
'filesystem_unavailable',
|
|
) from None
|
|
|
|
@staticmethod
|
|
def _unlock_mutation_directory(directory_descriptor):
|
|
if fcntl is None:
|
|
return
|
|
try:
|
|
fcntl.flock(directory_descriptor, fcntl.LOCK_UN)
|
|
except OSError:
|
|
pass
|
|
|
|
@contextmanager
|
|
def _locked_parent(self, root, components):
|
|
with self._opened_parent(root, components) as parent:
|
|
directory_descriptor, _ = parent
|
|
self._lock_mutation_directory(directory_descriptor)
|
|
try:
|
|
yield parent
|
|
finally:
|
|
self._unlock_mutation_directory(directory_descriptor)
|
|
|
|
def _publish_temporary(
|
|
self, directory_descriptor, temporary_name, leaf_name, *, create):
|
|
failure = None
|
|
try:
|
|
if create:
|
|
_rename_noreplace(
|
|
temporary_name, leaf_name, directory_descriptor,
|
|
)
|
|
else:
|
|
try:
|
|
os.replace(
|
|
temporary_name, leaf_name,
|
|
src_dir_fd=directory_descriptor,
|
|
dst_dir_fd=directory_descriptor,
|
|
)
|
|
except OSError:
|
|
raise ManagedFileAccessError(
|
|
'concurrent_change',
|
|
) from None
|
|
except BaseException as exc:
|
|
failure = exc
|
|
raise
|
|
finally:
|
|
try:
|
|
self._fsync_directory(directory_descriptor)
|
|
except BaseException as sync_failure:
|
|
if (
|
|
failure is None
|
|
or isinstance(failure, Exception)
|
|
and not isinstance(sync_failure, Exception)
|
|
):
|
|
raise
|
|
|
|
def _verify_private_file(
|
|
self, directory_descriptor, leaf_name, max_bytes,
|
|
expected_identity, expected_inode):
|
|
identity, details, _ = self._read_file_at(
|
|
directory_descriptor, leaf_name, max_bytes, content=False,
|
|
)
|
|
if (
|
|
identity != expected_identity
|
|
or details.st_dev != expected_inode.st_dev
|
|
or details.st_ino != expected_inode.st_ino
|
|
or details.st_uid != os.geteuid()
|
|
or stat.S_IMODE(details.st_mode) != 0o600
|
|
):
|
|
raise ManagedFileAccessError('concurrent_change')
|
|
return details
|
|
|
|
@contextmanager
|
|
def _opened_target(self, root, components, target_kind):
|
|
target = self._open_relative(root, components, target_kind)
|
|
try:
|
|
yield target
|
|
finally:
|
|
target._close()
|
|
|
|
def open_list_directory(self, root_id, relative_path=None):
|
|
root = self._root_for_operation(root_id, ManagedFileOperation.LIST)
|
|
components = (
|
|
() if relative_path is None
|
|
else parse_managed_relative_path(relative_path, root.limits)
|
|
)
|
|
self._require_relative_target(root.root_id, components, 'directory')
|
|
return self._opened_target(root, components, 'directory')
|
|
|
|
def open_read_file(self, root_id, relative_path):
|
|
root = self._root_for_operation(root_id, ManagedFileOperation.READ)
|
|
components = parse_managed_relative_path(relative_path, root.limits)
|
|
self._require_relative_target(root.root_id, components, 'file')
|
|
return self._opened_target(root, components, 'file')
|
|
|
|
def list_directory(self, root_id, relative_path=None):
|
|
root = self._root_for_operation(root_id, ManagedFileOperation.LIST)
|
|
components = (
|
|
() if relative_path is None
|
|
else parse_managed_relative_path(relative_path, root.limits)
|
|
)
|
|
self._require_relative_target(root.root_id, components, 'directory')
|
|
entries = []
|
|
name_bytes = scanned = 0
|
|
try:
|
|
with self._opened_target(root, components, 'directory') as opened:
|
|
try:
|
|
iterator = os.scandir(opened.fileno())
|
|
with iterator:
|
|
for item in iterator:
|
|
scanned += 1
|
|
if scanned > root.limits.max_listing_entries:
|
|
raise ManagedFileAccessError('limit_exceeded')
|
|
name = item.name
|
|
if type(name) is not str or name.startswith(
|
|
_TEMPORARY_PREFIX):
|
|
continue
|
|
if not self._relative_target_allowed(
|
|
root.root_id, (*components, name), 'file'):
|
|
continue
|
|
try:
|
|
parse_managed_relative_path(
|
|
'/'.join((*components, name)), root.limits,
|
|
)
|
|
encoded = name.encode('utf-8')
|
|
details = os.stat(
|
|
name, dir_fd=opened.fileno(),
|
|
follow_symlinks=False,
|
|
)
|
|
except ManagedFileAccessError:
|
|
continue
|
|
except FileNotFoundError:
|
|
continue
|
|
except OSError:
|
|
raise ManagedFileAccessError(
|
|
'filesystem_unavailable',
|
|
) from None
|
|
if stat.S_ISDIR(details.st_mode):
|
|
kind = 'directory'
|
|
byte_count = None
|
|
elif (
|
|
_regular_single_link(details)
|
|
and details.st_size <= root.limits.max_file_bytes
|
|
):
|
|
kind = 'file'
|
|
byte_count = details.st_size
|
|
else:
|
|
continue
|
|
if not self._relative_target_allowed(
|
|
root.root_id, (*components, name), kind):
|
|
continue
|
|
name_bytes += len(encoded)
|
|
if name_bytes > root.limits.max_listing_bytes:
|
|
raise ManagedFileAccessError('limit_exceeded')
|
|
entries.append(ManagedFileDirectoryEntry(
|
|
name, kind, byte_count,
|
|
))
|
|
except OSError:
|
|
raise ManagedFileAccessError(
|
|
'filesystem_unavailable',
|
|
) from None
|
|
entries.sort(key=lambda entry: entry.name.encode('utf-8'))
|
|
return ManagedFileListing(tuple(entries), name_bytes)
|
|
except BaseException:
|
|
entries.clear()
|
|
raise
|
|
|
|
def download_file(self, root_id, relative_path):
|
|
root = self._root_for_operation(root_id, ManagedFileOperation.READ)
|
|
components = parse_managed_relative_path(relative_path, root.limits)
|
|
self._require_relative_target(root.root_id, components, 'file')
|
|
if root.root_id == RUNTIME_RESULT_ROOT_ID:
|
|
if not self._snapshot_gate.acquire(blocking=False):
|
|
raise ManagedFileAccessError('download_busy')
|
|
try:
|
|
for attempt in range(2):
|
|
try:
|
|
with self._opened_target(root, components, 'file') as opened:
|
|
identity, snapshot = self._snapshot_descriptor(
|
|
root, opened.fileno(),
|
|
)
|
|
return ManagedFileDownload(
|
|
identity, snapshot=snapshot,
|
|
)
|
|
except ManagedFileAccessError as exc:
|
|
if attempt or exc.category not in (
|
|
'concurrent_change', 'unsafe_target'):
|
|
raise
|
|
except BaseException:
|
|
self._snapshot_gate.release()
|
|
raise
|
|
raise ManagedFileAccessError('concurrent_change')
|
|
for attempt in range(2):
|
|
try:
|
|
with self._opened_target(root, components, 'file') as opened:
|
|
identity, _, content = self._read_descriptor(
|
|
opened.fileno(), root.limits.max_file_bytes,
|
|
include_content=True,
|
|
)
|
|
return ManagedFileDownload(identity, content)
|
|
except ManagedFileAccessError as exc:
|
|
if attempt or exc.category not in ('concurrent_change', 'unsafe_target'):
|
|
raise
|
|
raise ManagedFileAccessError('concurrent_change')
|
|
|
|
def mutation_file_identity(
|
|
self, root_id, relative_path, operation, *, require_private_sha256=None):
|
|
if operation not in (
|
|
ManagedFileOperation.CREATE_REPLACE, ManagedFileOperation.DELETE,
|
|
):
|
|
raise ManagedFileAccessError('operation_not_allowed')
|
|
root = self._root_for_operation(root_id, operation)
|
|
components = parse_managed_relative_path(relative_path, root.limits)
|
|
with self._opened_parent(root, components) as parent:
|
|
directory_descriptor, leaf_name = parent
|
|
try:
|
|
try:
|
|
identity, details, _ = self._read_file_at(
|
|
directory_descriptor, leaf_name,
|
|
root.limits.max_file_bytes, content=False,
|
|
)
|
|
except ManagedFileAccessError as exc:
|
|
if exc.category != 'not_found':
|
|
raise
|
|
self._fsync_directory(directory_descriptor)
|
|
try:
|
|
self._read_file_at(
|
|
directory_descriptor, leaf_name,
|
|
root.limits.max_file_bytes, content=False,
|
|
)
|
|
except ManagedFileAccessError as repeated:
|
|
if repeated.category == 'not_found':
|
|
raise exc
|
|
raise
|
|
raise ManagedFileAccessError('concurrent_change')
|
|
# Replay may follow a crash after namespace publication. A
|
|
# successful directory fsync turns the observed state into
|
|
# durable evidence before terminal operation reconciliation.
|
|
self._fsync_directory(directory_descriptor)
|
|
self._revalidate_named_file(
|
|
directory_descriptor, leaf_name, details,
|
|
)
|
|
if (
|
|
require_private_sha256 is not None
|
|
and hmac.compare_digest(identity.sha256, require_private_sha256)
|
|
and (
|
|
details.st_uid != os.geteuid()
|
|
or stat.S_IMODE(details.st_mode) != 0o600
|
|
)
|
|
):
|
|
raise ManagedFileAccessError('unsafe_target')
|
|
except ManagedFileAccessError:
|
|
raise
|
|
return identity
|
|
|
|
def _create_replace_file_payload(
|
|
self, root_id, relative_path, payload_box, *, expected_sha256):
|
|
root = self._root_for_operation(
|
|
root_id, ManagedFileOperation.CREATE_REPLACE,
|
|
)
|
|
components = parse_managed_relative_path(relative_path, root.limits)
|
|
if len(payload_box) != 1 or type(payload_box[0]) is not bytes:
|
|
raise ManagedFileAccessError('invalid_content')
|
|
if len(payload_box[0]) > root.limits.max_file_bytes:
|
|
raise ManagedFileAccessError('limit_exceeded')
|
|
if expected_sha256 is not None:
|
|
expected_sha256 = _valid_hash(expected_sha256)
|
|
proposed = ManagedFileIdentity(
|
|
hashlib.sha256(payload_box[0]).hexdigest(), len(payload_box[0]),
|
|
)
|
|
|
|
with self._mutation_lock, self._locked_parent(root, components) as parent:
|
|
directory_descriptor, leaf_name = parent
|
|
before = before_details = None
|
|
if expected_sha256 is not None:
|
|
before, before_details, _ = self._read_file_at(
|
|
directory_descriptor, leaf_name,
|
|
root.limits.max_file_bytes, content=False,
|
|
)
|
|
if not hmac.compare_digest(before.sha256, expected_sha256):
|
|
raise ManagedFileAccessError('hash_conflict')
|
|
if before == proposed:
|
|
self._revalidate_named_file(
|
|
directory_descriptor, leaf_name, before_details,
|
|
)
|
|
return ManagedFileMutation(before, before, False)
|
|
|
|
temporary_name = temporary_descriptor = None
|
|
published = False
|
|
failure = None
|
|
try:
|
|
(
|
|
temporary_name, temporary_descriptor,
|
|
staged_identity, staged_details,
|
|
) = self._stage_temporary_file(
|
|
directory_descriptor, payload_box[0],
|
|
root.limits.max_file_bytes,
|
|
)
|
|
if staged_identity != proposed:
|
|
raise ManagedFileAccessError('concurrent_change')
|
|
if expected_sha256 is not None:
|
|
self._revalidate_named_file(
|
|
directory_descriptor, leaf_name, before_details,
|
|
)
|
|
self._publish_temporary(
|
|
directory_descriptor, temporary_name, leaf_name,
|
|
create=expected_sha256 is None,
|
|
)
|
|
published = True
|
|
try:
|
|
verified = self._verify_private_file(
|
|
directory_descriptor, leaf_name,
|
|
root.limits.max_file_bytes, proposed, staged_details,
|
|
)
|
|
self._revalidate_named_file(
|
|
directory_descriptor, leaf_name, verified,
|
|
)
|
|
except ManagedFileAccessError as exc:
|
|
if exc.category == 'concurrent_change':
|
|
raise
|
|
raise ManagedFileAccessError(
|
|
'durability_uncertain',
|
|
) from None
|
|
return ManagedFileMutation(before, proposed, True)
|
|
except BaseException as exc:
|
|
failure = exc
|
|
raise
|
|
finally:
|
|
if temporary_name is not None and not published:
|
|
self._release_unpublished_temporary(
|
|
directory_descriptor, temporary_name,
|
|
temporary_descriptor, failure,
|
|
)
|
|
temporary_descriptor = None
|
|
try:
|
|
_close_descriptor(temporary_descriptor)
|
|
except BaseException:
|
|
if failure is None:
|
|
raise
|
|
|
|
def create_replace_file(
|
|
self, root_id, relative_path, payload, *, expected_sha256):
|
|
payload_box = [payload]
|
|
payload = None
|
|
try:
|
|
return self._create_replace_file_payload(
|
|
root_id, relative_path, payload_box,
|
|
expected_sha256=expected_sha256,
|
|
)
|
|
finally:
|
|
payload_box.clear()
|
|
payload = payload_box = None
|
|
|
|
def delete_file(self, root_id, relative_path, *, expected_sha256):
|
|
root = self._root_for_operation(root_id, ManagedFileOperation.DELETE)
|
|
components = parse_managed_relative_path(relative_path, root.limits)
|
|
expected_sha256 = _valid_hash(expected_sha256)
|
|
with self._mutation_lock, self._locked_parent(root, components) as parent:
|
|
directory_descriptor, leaf_name = parent
|
|
before, before_details, _ = self._read_file_at(
|
|
directory_descriptor, leaf_name,
|
|
root.limits.max_file_bytes, content=False,
|
|
)
|
|
if not hmac.compare_digest(before.sha256, expected_sha256):
|
|
raise ManagedFileAccessError('hash_conflict')
|
|
self._revalidate_named_file(
|
|
directory_descriptor, leaf_name, before_details,
|
|
)
|
|
failure = None
|
|
try:
|
|
try:
|
|
os.unlink(leaf_name, dir_fd=directory_descriptor)
|
|
except OSError:
|
|
raise ManagedFileAccessError(
|
|
'concurrent_change',
|
|
) from None
|
|
except BaseException as exc:
|
|
failure = exc
|
|
raise
|
|
finally:
|
|
try:
|
|
self._fsync_directory(directory_descriptor)
|
|
except BaseException as sync_failure:
|
|
if (
|
|
failure is None
|
|
or isinstance(failure, Exception)
|
|
and not isinstance(sync_failure, Exception)
|
|
):
|
|
raise
|
|
return ManagedFileMutation(before, None, True)
|