52 lines
2.4 KiB
Python
52 lines
2.4 KiB
Python
"""Capability verifier for mutating workspace children.
|
|
|
|
The backend passes writer.lock as fd 3 and the retained workspace directory as fd 4.
|
|
This module intentionally has no path fallback: callers either run with the capability
|
|
or fail closed before touching artifacts.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import fcntl
|
|
import os
|
|
import stat
|
|
from dataclasses import dataclass
|
|
|
|
|
|
class WorkspaceWriterConflict(RuntimeError):
|
|
"preprocessing_conflict"
|
|
def __init__(self, message: str = "preprocessing_conflict") -> None:
|
|
super().__init__(message)
|
|
|
|
@dataclass(frozen=True)
|
|
class WorkspaceCapability:
|
|
workspace_id: str
|
|
revision: str
|
|
device: int
|
|
inode: int
|
|
writer_device: int
|
|
writer_inode: int
|
|
|
|
def _identity(env: dict[str, str]) -> tuple[str, str, int, int]:
|
|
wid, rev = env.get("THOTH_WORKSPACE_ID"), env.get("THOTH_WORKSPACE_REVISION")
|
|
if not wid or not rev or not __import__("re").fullmatch(r"[a-z][a-z0-9-]{2,62}", wid) or not __import__("re").fullmatch(r"[0-9a-f]{40}", rev):
|
|
raise WorkspaceWriterConflict()
|
|
try: device, inode = int(env["THOTH_WORKSPACE_DEVICE"]), int(env["THOTH_WORKSPACE_INODE"])
|
|
except (KeyError, ValueError): raise WorkspaceWriterConflict()
|
|
return wid, rev, device, inode
|
|
|
|
def verify_workspace_writer_fds(*, writer_fd: int = 3, root_fd: int = 4, env: dict[str, str] | None = None) -> WorkspaceCapability:
|
|
env = dict(os.environ if env is None else env)
|
|
wid, rev, device, inode = _identity(env)
|
|
try: root = os.fstat(root_fd); writer = os.fstat(writer_fd)
|
|
except OSError as exc: raise WorkspaceWriterConflict() from exc
|
|
if not stat.S_ISDIR(root.st_mode) or root.st_uid != os.getuid() or (root.st_mode & 0o777) != 0o700 or (root.st_dev, root.st_ino) != (device, inode): raise WorkspaceWriterConflict()
|
|
if not stat.S_ISREG(writer.st_mode) or writer.st_uid != os.getuid() or (writer.st_mode & 0o777) != 0o600: raise WorkspaceWriterConflict()
|
|
try: fcntl.flock(writer_fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
|
except OSError as exc: raise WorkspaceWriterConflict() from exc
|
|
# Keep the OFD locked. A lock check is necessarily best effort on some BSDs; identity and
|
|
# descriptor ownership remain mandatory and no path-based lock is accepted.
|
|
return WorkspaceCapability(wid, rev, root.st_dev, root.st_ino, writer.st_dev, writer.st_ino)
|
|
|
|
def require_workspace_writer_capability() -> WorkspaceCapability:
|
|
return verify_workspace_writer_fds()
|