"""Identity-anchored directory primitives for profile custody storage."""
from __future__ import annotations
import os
import stat
import sys
from collections.abc import Generator
from contextlib import ExitStack, contextmanager
from contextvars import ContextVar
from dataclasses import dataclass, field
from functools import cache
from pathlib import Path
from typing import Any, Final, Literal
from .errors import ProfileCustodyRecordError
PROFILE_CUSTODY_COMMIT_FILENAME: Final = "profile.commit.v1.json"
[docs]
@dataclass(frozen=True, slots=True)
class WindowsDirectoryAnchorErrors:
"""Operator-facing refusals emitted while anchoring a Windows directory."""
cannot_open: str
cannot_verify: str
invalid_entry: str
empty_path: str
_PROFILE_CUSTODY_DIRECTORY_ANCHOR_ERRORS = WindowsDirectoryAnchorErrors(
cannot_open="profile capsule directory cannot be identity-anchored",
cannot_verify="profile capsule directory identity cannot be verified",
invalid_entry="profile capsule directory must not be a reparse point or non-directory",
empty_path="profile capsule directory cannot be identity-anchored",
)
[docs]
@dataclass(frozen=True, slots=True)
class ProfileCustodyPasswordReadOperation:
"""One actual filesystem operation in the normal-password read set."""
operation: Literal["stat", "open", "read"]
path: Path
[docs]
def is_real_directory(path: Path) -> bool:
"""Return whether ``path`` is a real directory, never a symlink or reparse point."""
try:
metadata = path.lstat()
except OSError:
return False
attributes = getattr(metadata, "st_file_attributes", 0)
reparse = getattr(stat, "FILE_ATTRIBUTE_REPARSE_POINT", 0)
return stat.S_ISDIR(metadata.st_mode) and not stat.S_ISLNK(metadata.st_mode) and not bool(attributes & reparse)
[docs]
def ensure_real_directory(path: Path) -> None:
"""Create ``path`` as a real directory, refusing an existing link or non-directory."""
if os.path.lexists(path) and not is_real_directory(path):
raise ProfileCustodyRecordError("profile capsule root must not be a link or non-directory")
try:
path.mkdir(parents=True, exist_ok=True)
except OSError as exc:
raise ProfileCustodyRecordError("profile capsule root cannot be created") from exc
if not is_real_directory(path):
raise ProfileCustodyRecordError("profile capsule root was not created as a real directory")
[docs]
def anchor_directory(stack: ExitStack, path: Path, *, final_access: int = 0) -> int | None:
"""Keep a verified directory identity non-deletable for a path operation."""
if os.name == "nt":
return stack.enter_context(windows_directory_anchor(path, final_access=final_access))
return stack.enter_context(posix_directory_fd(path))
[docs]
def ensure_profile_custody_local_directory(path: Path) -> None:
"""Create one custody-owned directory while its parent identity is pinned.
The caller supplies a single child of an existing storage root. This is
intentionally not a recursive convenience function: recursive path walks
reintroduce a check-then-create window on Windows and POSIX alike.
"""
if path.name in {"", ".", ".."}:
raise ProfileCustodyRecordError("local custody directory must name one child")
if os.name != "nt":
with posix_directory_fd(path.parent) as parent_fd:
try:
os.mkdir(path.name, 0o700, dir_fd=parent_fd)
except FileExistsError:
pass
except OSError as exc:
raise ProfileCustodyRecordError("local custody directory cannot be created") from exc
descriptor = posix_open_child_directory(parent_fd, path.name)
os.close(descriptor)
return
with ExitStack() as anchors:
anchor_directory(anchors, path.parent, final_access=0x80000000)
try:
path.mkdir(mode=0o700)
except FileExistsError:
pass
except OSError as exc:
raise ProfileCustodyRecordError("local custody directory cannot be created") from exc
anchor_directory(anchors, path, final_access=0x80000000)
[docs]
@cache
def windows_create_file_api() -> tuple[Any, Any, Any, Any]:
"""Return configured ``CreateFileW`` bindings for identity-anchored calls.
Built once per process and shared. The binding is pure configuration, and
every consumer that configures a further function on the returned library
sets the same signature, so sharing cannot change a call's conversion.
The ``sys.platform == "win32"`` block, rather than a guard at the call
site or an early return, is what establishes the platform for the Windows
API below: it is the only guard shape every checker this project runs
narrows on, so those references resolve when the tree is analysed for a
platform that does not ship them.
"""
if sys.platform == "win32":
import ctypes
from ctypes import wintypes
kernel32 = ctypes.WinDLL("kernel32", use_last_error=True)
create_file = kernel32.CreateFileW
create_file.argtypes = [
ctypes.c_wchar_p,
wintypes.DWORD,
wintypes.DWORD,
ctypes.c_void_p,
wintypes.DWORD,
wintypes.DWORD,
wintypes.HANDLE,
]
create_file.restype = wintypes.HANDLE
return ctypes, wintypes, kernel32, create_file
raise ProfileCustodyRecordError("the CreateFileW bindings are not available on this platform")
@dataclass(slots=True)
class _HeldComponent:
"""One verified component handle and the number of live anchors relying on it."""
handle: int
access: int
users: int = 0
@dataclass(slots=True)
class _AnchorSession:
"""Component handles shared by the anchors of one custody operation."""
held: dict[str, list[_HeldComponent]] = field(default_factory=dict)
_ANCHOR_SESSION: ContextVar[_AnchorSession | None] = ContextVar("profile_custody_anchor_session", default=None)
[docs]
@contextmanager
def shared_directory_anchors() -> Generator[None]:
"""Let the anchors of one operation share the component handles they hold.
An operation such as one capsule scan anchors the same ancestor chain again
for every member it inspects -- every component of the storage root, then
the capsules directory -- and each anchor re-opened all of them. Inside this
scope an anchor reuses a component that another live anchor already holds
with at least the access it needs, and opens only what is missing. A
component stays held while ANY anchor relying on it is live, so no read runs
under an ancestor that was released; a handle closes when its last user
leaves. The scope holds nothing of its own, so nothing outlives the
operation. Nested scopes share the outermost session.
"""
if _ANCHOR_SESSION.get() is not None:
yield
return
session = _AnchorSession()
token = _ANCHOR_SESSION.set(session)
try:
yield
finally:
_ANCHOR_SESSION.reset(token)
leftover = [component for components in session.held.values() for component in components]
if leftover:
_ctypes, _wintypes, kernel32, _create_file = windows_create_file_api()
for component in leftover:
kernel32.CloseHandle(component.handle)
[docs]
@contextmanager
def windows_directory_anchor(
path: Path,
*,
final_access: int = 0,
errors: WindowsDirectoryAnchorErrors = _PROFILE_CUSTODY_DIRECTORY_ANCHOR_ERRORS,
) -> Generator[int]:
"""Lock every real component against reparse and delete substitution."""
session = _ANCHOR_SESSION.get()
if session is None:
with _windows_directory_anchor_owned(path, final_access=final_access, errors=errors) as handle:
yield handle
return
with _windows_directory_anchor_shared(session, path, final_access=final_access, errors=errors) as handle:
yield handle
@contextmanager
def _windows_directory_anchor_owned(
path: Path,
*,
final_access: int,
errors: WindowsDirectoryAnchorErrors,
) -> Generator[int]:
_ctypes, _wintypes, kernel32, _create_file = windows_create_file_api()
handles: list[int] = []
try:
component_paths = windows_component_paths(path)
for index, current in enumerate(component_paths):
access = final_access if index == len(component_paths) - 1 else 0
handles.append(_open_verified_component(current, access, errors=errors))
if not handles:
raise ProfileCustodyRecordError(errors.empty_path)
yield handles[-1]
finally:
for handle in reversed(handles):
kernel32.CloseHandle(handle)
@contextmanager
def _windows_directory_anchor_shared(
session: _AnchorSession,
path: Path,
*,
final_access: int,
errors: WindowsDirectoryAnchorErrors,
) -> Generator[int]:
_ctypes, _wintypes, kernel32, _create_file = windows_create_file_api()
relied_on: list[tuple[str, _HeldComponent]] = []
try:
component_paths = windows_component_paths(path)
for index, current in enumerate(component_paths):
access = final_access if index == len(component_paths) - 1 else 0
key = os.path.normcase(current)
held = session.held.setdefault(key, [])
component = next((candidate for candidate in held if candidate.access & access == access), None)
if component is None:
opened = _open_verified_component(current, access, errors=errors)
component = _HeldComponent(handle=opened, access=access)
held.append(component)
component.users += 1
relied_on.append((key, component))
if not relied_on:
raise ProfileCustodyRecordError(errors.empty_path)
yield relied_on[-1][1].handle
finally:
for key, component in reversed(relied_on):
component.users -= 1
if component.users == 0:
kernel32.CloseHandle(component.handle)
session.held[key].remove(component)
def _open_verified_component(current: str, access: int, *, errors: WindowsDirectoryAnchorErrors) -> int:
"""Open one component without delete sharing and prove it is a real directory."""
ctypes, wintypes, kernel32, create_file = windows_create_file_api()
handle = create_file(
current,
access,
0x00000001 | 0x00000002,
None,
3,
0x02000000 | 0x00200000,
None,
)
if handle == wintypes.HANDLE(-1).value:
raise ProfileCustodyRecordError(errors.cannot_open)
try:
info = windows_file_information_type()()
if not kernel32.GetFileInformationByHandle(handle, ctypes.byref(info)):
raise ProfileCustodyRecordError(errors.cannot_verify)
if not info.dwFileAttributes & 0x10 or info.dwFileAttributes & 0x400:
raise ProfileCustodyRecordError(errors.invalid_entry)
except BaseException:
kernel32.CloseHandle(handle)
raise
return int(handle)
[docs]
def windows_component_paths(path: Path) -> tuple[str, ...]:
"""Return the path of every component below the anchor, outermost first.
String-joined rather than built as successive ``Path`` objects: an anchor
walks every component of a deep storage path, and rendering each
intermediate ``Path`` was the largest cost of anchoring after the Win32
calls themselves.
"""
current = path.anchor
components = path.parts[1:] if path.anchor else path.parts
paths: list[str] = []
for component in components:
current = os.path.join(current, component) if current else component
paths.append(current)
return tuple(paths)
[docs]
@contextmanager
def posix_directory_fd(path: Path) -> Generator[int]:
"""Walk an absolute directory a component at a time without following links."""
flags = os.O_RDONLY | getattr(os, "O_DIRECTORY", 0) | getattr(os, "O_NOFOLLOW", 0)
try:
descriptor = os.open(path.anchor or "/", flags)
except OSError as exc:
raise ProfileCustodyRecordError("profile capsule root cannot be no-follow opened") from exc
try:
components = path.parts[1:] if path.anchor else path.parts
for component in components:
try:
next_descriptor = os.open(component, flags, dir_fd=descriptor)
except OSError as exc:
raise ProfileCustodyRecordError("profile capsule directory component is unsafe") from exc
os.close(descriptor)
descriptor = next_descriptor
yield descriptor
finally:
os.close(descriptor)
[docs]
def posix_open_child_directory(parent_fd: int, name: str) -> int:
"""Open ``name`` below ``parent_fd`` without following links, or raise."""
try:
return os.open(
name,
os.O_RDONLY | getattr(os, "O_DIRECTORY", 0) | getattr(os, "O_NOFOLLOW", 0),
dir_fd=parent_fd,
)
except OSError as exc:
raise ProfileCustodyRecordError("profile capsule directory component is unsafe") from exc
[docs]
def posix_mkdir_child_directory(parent_fd: int, name: str) -> int:
"""Create directory ``name`` below ``parent_fd`` and return its open descriptor."""
try:
os.mkdir(name, mode=0o700, dir_fd=parent_fd)
except OSError as exc:
raise ProfileCustodyRecordError("profile capsule staging directory cannot be created") from exc
return posix_open_child_directory(parent_fd, name)
[docs]
def write_exclusive_fsynced(path: Path, payload: bytes) -> None:
"""Create and fsync one capsule staging record at an anchored path."""
if os.name != "nt" and not is_real_directory(path.parent):
raise ProfileCustodyRecordError("profile capsule staging parent must not be a link or reparse directory")
try:
descriptor = os.open(
path,
os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_BINARY", 0),
0o600,
)
except OSError as exc:
raise ProfileCustodyRecordError("profile capsule staging record cannot be exclusively created") from exc
try:
_write_exclusive_descriptor_fsynced(descriptor, payload)
finally:
os.close(descriptor)
[docs]
def write_exclusive_fsynced_fd(parent_fd: int, name: str, payload: bytes) -> None:
"""Create and fsync one capsule staging record below an anchored descriptor."""
try:
descriptor = os.open(
name,
os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_NOFOLLOW", 0),
0o600,
dir_fd=parent_fd,
)
except OSError as exc:
raise ProfileCustodyRecordError("profile capsule staging record cannot be exclusively created") from exc
try:
_write_exclusive_descriptor_fsynced(descriptor, payload)
finally:
os.close(descriptor)
def _write_exclusive_descriptor_fsynced(descriptor: int, payload: bytes) -> None:
try:
offset = 0
while offset < len(payload):
written = os.write(descriptor, payload[offset:])
if written <= 0:
raise OSError("profile capsule staging short write")
offset += written
os.fsync(descriptor)
except OSError as exc:
raise ProfileCustodyRecordError("profile capsule staging record could not be fsynced") from exc
__all__ = [
"PROFILE_CUSTODY_COMMIT_FILENAME",
"ProfileCustodyPasswordReadOperation",
"WindowsDirectoryAnchorErrors",
"anchor_directory",
"ensure_profile_custody_local_directory",
"ensure_real_directory",
"is_real_directory",
"is_reparse_metadata",
"posix_directory_fd",
"posix_mkdir_child_directory",
"posix_open_child_directory",
"shared_directory_anchors",
"windows_create_file_api",
"windows_directory_anchor",
"windows_file_information_type",
"write_exclusive_fsynced",
"write_exclusive_fsynced_fd",
]