"""Root-only mount primitives for preservation-compatible C01 role views. Only trusted launcher code calls these operations, before credential drop. They do not migrate data, grant writer exclusion, select private/shared directories, or install a service. The launcher must own those decisions and the whole child cgroup through setup failure. Source capabilities are pinned before covering any paths, so a pathname replacement cannot redirect a prepared mount. """ import ctypes from dataclasses import dataclass import os from pathlib import Path import stat import subprocess import signal LIBC = ctypes.CDLL(None, use_errno=True) LIBC.mount.argtypes = [ctypes.c_char_p, ctypes.c_char_p, ctypes.c_char_p, ctypes.c_ulong, ctypes.c_char_p] LIBC.mount.restype = ctypes.c_int MS_NOSUID, MS_NODEV, MS_NOEXEC, MS_REC, MS_PRIVATE = 2, 4, 8, 16384, 1 << 18 PROTECTIONS = os.ST_RDONLY | os.ST_NOSUID | os.ST_NODEV | os.ST_NOEXEC def require(condition, message): if not condition: raise ValueError(message) def identity(fd): value = os.fstat(fd) return value.st_dev, value.st_ino, value.st_uid, value.st_gid, stat.S_IMODE(value.st_mode) def namespace_identity(): value = os.stat('/proc/self/ns/mnt') return value.st_dev, value.st_ino def _open_locator(path, directory): """Walk the absolute locator without following any component symlinks.""" path = Path(path) require(path.anchor == '/' and '..' not in path.parts, 'canonical mount source required') flags = os.O_PATH | os.O_NOFOLLOW | os.O_CLOEXEC parent = os.open('/', flags | os.O_DIRECTORY) try: if len(path.parts) == 1: require(directory, 'regular file locator required') result, parent = parent, None return result for component in path.parts[1:-1]: child = os.open(component, flags | os.O_DIRECTORY, dir_fd=parent) os.close(parent) parent = child fd = os.open(path.name, flags | (os.O_DIRECTORY if directory else 0), dir_fd=parent) try: meta = os.fstat(fd) require(stat.S_ISDIR(meta.st_mode) if directory else stat.S_ISREG(meta.st_mode), 'mount locator object type refused') return fd except BaseException: os.close(fd) raise finally: if parent is not None: os.close(parent) def pin_directory(path): return _open_locator(path, directory=True) @dataclass(frozen=True) class PinnedSource: """Caller-owned FD plus its authenticated PRE-unshare canonical locator. The FD is the authority, not the locator. Never derive a pathname from a foreign-mount /proc/self/fd symlink: its rendering can lose the mount prefix. Renamed-away, covered, unlinked or replaced locators refuse; there is no search/fallback to the old inode's new name. Prepare a new admission before unshare if a root operator intentionally changes the layout. """ fd: int locator: Path expected: tuple protections: int directory: bool namespace: tuple @classmethod def capture(cls, fd, locator): meta = os.fstat(fd) require(stat.S_ISDIR(meta.st_mode) or stat.S_ISREG(meta.st_mode), 'unsupported source capability') source = cls(fd, Path(locator), identity(fd), os.fstatvfs(fd).f_flag & PROTECTIONS, stat.S_ISDIR(meta.st_mode), namespace_identity()) # Authenticate the supplied canonical lookup against the actual FD while # it is still in the preparing namespace. No source contents are read. check = source.reopen() os.close(check) return source def checked_protections(self): require(identity(self.fd) == self.expected and os.fstat(self.fd).st_nlink > 0, 'pinned source identity changed or unlinked') protections = os.fstatvfs(self.fd).f_flag & PROTECTIONS require(protections & self.protections == self.protections, 'pinned source mount protections weakened') return protections def reopen(self): protections = self.checked_protections() current = _open_locator(self.locator, self.directory) try: require(identity(current) == self.expected, 'source locator no longer names pinned object') require(os.fstatvfs(current).f_flag & protections == protections, 'source mount protections weakened across namespace setup') return current except BaseException: os.close(current) raise def require_private_backing(path): """Backing root may retain old ownership; all ancestry must exclude writers.""" path = Path(path) require(path.is_absolute() and path.resolve(strict=True) == path, 'canonical backing required') for parent in path.parents: metadata = parent.lstat() require(stat.S_ISDIR(metadata.st_mode) and metadata.st_uid == 0 and not metadata.st_mode & 0o022, 'unprotected backing ancestry') parent = path.parent.stat() require(stat.S_IMODE(parent.st_mode) == 0o700, 'backing requires a root-private parent') return pin_directory(path) def swap_map(original, target, extents=((0, 2**32 - 1),)): require(type(original) is int and type(target) is int and 0 < original < 2**32 - 1 and 0 < target < 2**32 - 1 and original != target, 'distinct nonroot mount identities required') # A nested owner may map only the IDs actually available in its parent # namespace. Asking mount(8) to map all UINT32 IDs in a rootless PID1 realm # fails even though both appointed principals are valid. Preserve EVERY # available identity except the exact two-way swap; never add unmapped IDs. extents = sorted(extents) end = 0 for start, count in extents: require(type(start) is int and type(count) is int and start >= end and count > 0 and start + count <= 2**32 - 1, 'invalid parent identity extents') end = start + count require(all(any(start <= uid < start + count for start, count in extents) for uid in (original, target)), 'mount principal outside parent identity map') rows = [] for start, count in extents: stops = sorted({start, start + count} | {n for n in (original, original+1, target, target+1) if start < n < start + count}) for a, b in zip(stops, stops[1:]): rows.append((a, target if a == original else original if a == target else a, b-a)) return rows class PrivateMounts: def __init__(self): require(os.getresuid() == (0, 0, 0), 'role views require root') require(len(list(Path('/proc/self/task').iterdir())) == 1, 'role view setup must be single-threaded') before = os.stat('/proc/self/ns/mnt') require(LIBC.unshare(0x00020000) == 0, 'private mount namespace unavailable') require(LIBC.mount(None, b'/', None, MS_REC | MS_PRIVATE, None) == 0, 'private mount propagation unavailable') after = os.stat('/proc/self/ns/mnt') require((before.st_dev, before.st_ino) != (after.st_dev, after.st_ino), 'mount namespace did not change') self.namespace = (after.st_dev, after.st_ino) def check(self): require(os.geteuid() == 0, 'mount setup after credential drop refused') current = os.stat('/proc/self/ns/mnt') require((current.st_dev, current.st_ino) == self.namespace, 'mount namespace changed during setup') def cover(self, target, mode=0o755, size=16 * 1024 * 1024): self.check() require(mode in (0o700, 0o755, 0o1777) and type(size) is int and 0 < size <= 512 * 1024 * 1024, 'invalid private tmpfs bounds') fd = pin_directory(target) try: destination = f'/proc/self/fd/{fd}'.encode() options = f'mode={mode:o},size={size}'.encode() require(LIBC.mount(b'tmpfs', destination, b'tmpfs', MS_NOSUID | MS_NODEV | MS_NOEXEC, options) == 0, 'private directory cover failed') require(os.statvfs(target).f_flag & os.ST_NOSUID, 'private cover flags missing') finally: os.close(fd) def idmapped_directory(self, source, target, original_uid, target_uid, original_gid=None, target_gid=None, readonly=False): self.check() require(isinstance(source, PinnedSource) and source.directory and source.namespace != self.namespace, 'pre-namespace pinned directory locator required') before = source.expected source_protections = source.checked_protections() original_gid = original_uid if original_gid is None else original_gid target_gid = target_uid if target_gid is None else target_gid # Reopen the authenticated original locator in this namespace. Foreign # proc-fd renderings may be mount-relative (e.g. /lnd), not usable paths. # The retained FD/snapshot still authenticate object and protections. current_source = source.reopen() destination = None try: source_protections |= os.fstatvfs(current_source).f_flag & PROTECTIONS destination = pin_directory(target) command = ['/usr/bin/mount', '--internal-only', '--no-mtab', '--no-canonicalize', '--bind'] if readonly: command.append('--read-only') for option, old, new in (('--map-users', original_uid, target_uid), ('--map-groups', original_gid, target_gid)): map_name = 'uid_map' if option == '--map-users' else 'gid_map' extents = [(int(row[0]), int(row[2])) for row in (line.split() for line in Path('/proc/self', map_name).read_text().splitlines())] for inside, outside, count in swap_map(old, new, extents): command.extend((option, f'{inside}:{outside}:{count}')) command.extend(('--', f'/proc/self/fd/{current_source}/.', f'/proc/self/fd/{destination}/.')) parent=os.getpid() def child_guard(): if LIBC.prctl(1,signal.SIGKILL,0,0,0) or os.getppid()!=parent:os._exit(125) result = subprocess.run(command, stdin=subprocess.DEVNULL, capture_output=True, timeout=10, pass_fds=(current_source, destination), close_fds=True,preexec_fn=child_guard, env={'PATH': '/usr/bin:/bin', 'LC_ALL': 'C'}) if result.returncode: os.write(2,('C01_IDMAP_REFUSED exit='+str(result.returncode)+' '+ result.stderr.decode(errors='replace')[:1024]+'\n').encode()) require(result.returncode == 0, 'idmapped role view refused') source.checked_protections() require(identity(current_source) == before, 'source ownership/mode/inode changed during mount') visible = os.stat(target) translate = lambda value, old, new: new if value == old else old if value == new else value require((visible.st_dev, visible.st_ino, visible.st_uid, visible.st_gid, stat.S_IMODE(visible.st_mode)) == (before[0], before[1], translate(before[2], original_uid, target_uid), translate(before[3], original_gid, target_gid), before[4]), 'idmapped role view identity mismatch') require(os.statvfs(target).f_flag & source_protections == source_protections, 'role view weakened source mount protections') if readonly: require(os.statvfs(target).f_flag & os.ST_RDONLY, 'read-only role view not enforced') finally: if destination is not None: os.close(destination) os.close(current_source)