80 lines
2.7 KiB
Python
80 lines
2.7 KiB
Python
"""Fail-fast root and filesystem capability probes."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import tempfile
|
|
import fcntl
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
|
|
|
|
class ProbeError(RuntimeError):
|
|
pass
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class FilesystemProbe:
|
|
root: Path
|
|
readable: bool
|
|
writable: bool
|
|
hard_link: bool
|
|
reflink: bool
|
|
sparse_files: bool
|
|
|
|
|
|
def probe_writable_directory(path: Path) -> None:
|
|
if not path.is_absolute() or not path.is_dir():
|
|
raise ProbeError(f"configured directory is not existing and absolute: {path}")
|
|
try:
|
|
descriptor, raw_path = tempfile.mkstemp(
|
|
prefix=".archive-control-write-probe-", dir=path
|
|
)
|
|
os.close(descriptor)
|
|
Path(raw_path).unlink()
|
|
except OSError as exc:
|
|
raise ProbeError(f"configured directory is not writable: {path}") from exc
|
|
|
|
|
|
def probe_root(root: Path) -> FilesystemProbe:
|
|
if not root.is_absolute() or not root.is_dir():
|
|
raise ProbeError(f"configured root is not an existing absolute directory: {root}")
|
|
if not os.access(root, os.R_OK | os.X_OK | os.W_OK):
|
|
raise ProbeError(f"configured root permissions are insufficient: {root}")
|
|
source: Path | None = None
|
|
linked: Path | None = None
|
|
cloned: Path | None = None
|
|
try:
|
|
descriptor, raw_path = tempfile.mkstemp(prefix=".archive-control-probe-", dir=root)
|
|
source = Path(raw_path)
|
|
with os.fdopen(descriptor, "wb") as probe:
|
|
probe.seek(1024 * 1024)
|
|
probe.write(b"x")
|
|
probe.flush()
|
|
os.fsync(probe.fileno())
|
|
allocated = source.stat().st_blocks * 512
|
|
sparse = allocated < source.stat().st_size
|
|
linked = source.with_name(f"{source.name}.link")
|
|
try:
|
|
os.link(source, linked)
|
|
hard_link = linked.stat().st_ino == source.stat().st_ino
|
|
except OSError:
|
|
hard_link = False
|
|
cloned = source.with_name(f"{source.name}.clone")
|
|
try:
|
|
with source.open("rb") as source_file, cloned.open("xb") as clone_file:
|
|
fcntl.ioctl(clone_file.fileno(), 0x40049409, source_file.fileno())
|
|
reflink = cloned.stat().st_size == source.stat().st_size
|
|
except OSError:
|
|
reflink = False
|
|
return FilesystemProbe(root, True, True, hard_link, reflink, sparse)
|
|
except OSError as exc:
|
|
raise ProbeError(f"filesystem capability probe failed for {root}") from exc
|
|
finally:
|
|
if linked is not None:
|
|
linked.unlink(missing_ok=True)
|
|
if cloned is not None:
|
|
cloned.unlink(missing_ok=True)
|
|
if source is not None:
|
|
source.unlink(missing_ok=True)
|