241 lines
12 KiB
Python
241 lines
12 KiB
Python
"""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)
|