#!/usr/bin/env python3 import argparse import ctypes import errno import fcntl import hashlib import json import os import shutil import stat as stat_module import subprocess import tempfile import uuid from contextlib import contextmanager from pathlib import Path SERVICES = ("mem-pressure-notify.service", "disk-fill-notify.service") RENAME_NOREPLACE = 1 MAX_NOTIFIER_BYTES = 16 * 1024 * 1024 LIBC = ctypes.CDLL(None, use_errno=True) NAMES = ( "mem-pressure-notify", "disk-fill-notify", "mem-pressure-notify.service", "disk-fill-notify.service", ) def run_systemctl(user, uid, *args): return subprocess.run( [ "runuser", "-u", user, "--", "env", f"XDG_RUNTIME_DIR=/run/user/{uid}", "systemctl", "--user", *args, ], check=True, capture_output=True, text=True, ).stdout.strip() def service_state(user, uid, service): load = run_systemctl(user, uid, "show", service, "-p", "LoadState", "--value") if load == "not-found": return {"load": load, "active": "inactive", "enabled": "not-found"} if load != "loaded": raise RuntimeError(f"invalid LoadState for {service}: {load or 'missing'}") active = run_systemctl(user, uid, "show", service, "-p", "ActiveState", "--value") enabled = run_systemctl(user, uid, "show", service, "-p", "UnitFileState", "--value") if active not in {"active", "inactive"}: raise RuntimeError(f"unsupported ActiveState for {service}: {active or 'missing'}") if enabled not in {"enabled", "disabled", "linked"}: raise RuntimeError(f"unsupported UnitFileState for {service}: {enabled or 'missing'}") return {"load": load, "active": active, "enabled": enabled} def validate_recorded_states(states): if not isinstance(states, dict) or set(states) != set(SERVICES): raise RuntimeError("invalid retired notifier service-state receipt") for service, state in states.items(): if not isinstance(state, dict) or set(state) != {"load", "active", "enabled"}: raise RuntimeError(f"invalid retired notifier state for {service}") if state["load"] not in {"loaded", "not-found"}: raise RuntimeError(f"invalid recorded LoadState for {service}") if state["active"] not in {"active", "inactive"}: raise RuntimeError(f"unsupported recorded ActiveState for {service}") if state["enabled"] not in {"enabled", "disabled", "linked", "not-found"}: raise RuntimeError(f"unsupported recorded UnitFileState for {service}") return states def service_states(user, uid): return {service: service_state(user, uid, service) for service in SERVICES} def apply_service_states(user, uid, states): for service, wanted in states.items(): current = service_state(user, uid, service) if wanted["enabled"] == "enabled" and current["enabled"] != "enabled": run_systemctl(user, uid, "enable", service) elif wanted["enabled"] in {"disabled", "linked"} and current["enabled"] == "enabled": run_systemctl(user, uid, "disable", service) current = service_state(user, uid, service) if wanted["active"] == "active" and current["active"] != "active": run_systemctl(user, uid, "start", service) elif wanted["active"] == "inactive" and current["active"] == "active": run_systemctl(user, uid, "stop", service) final = service_states(user, uid) if final != states: raise RuntimeError(f"service-state restoration mismatch: wanted {states}, got {final}") def notifier_paths(home): return { "mem-pressure-notify": home / ".local/bin/mem-pressure-notify", "disk-fill-notify": home / ".local/bin/disk-fill-notify", "mem-pressure-notify.service": home / ".config/systemd/user/mem-pressure-notify.service", "disk-fill-notify.service": home / ".config/systemd/user/disk-fill-notify.service", } def exists(path): return path.exists() or path.is_symlink() def fsync_file(path): descriptor = os.open(path, os.O_RDONLY | os.O_NOFOLLOW) try: os.fsync(descriptor) finally: os.close(descriptor) def fsync_directory(path): descriptor = os.open(path, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW) try: os.fsync(descriptor) finally: os.close(descriptor) def validate_directories(home, destinations, uid): if home.is_symlink() or not home.is_dir(): raise RuntimeError("target home must be an existing non-symlink directory") home_stat = home.stat() if home_stat.st_uid != uid: raise RuntimeError("target home has unexpected owner") home = home.resolve(strict=True) metadata = {} for destination in destinations.values(): parent = destination.parent relative = parent.relative_to(home) current = home missing = False for part in relative.parts: current = current / part if current.is_symlink(): raise RuntimeError(f"target directory path contains symlink: {current}") if not current.exists(): missing = True break if not current.is_dir(): raise RuntimeError(f"target directory path is not a directory: {current}") if missing: metadata[parent] = None else: stat = parent.stat() if stat.st_uid != uid: raise RuntimeError(f"target directory has unexpected owner: {parent}") metadata[parent] = ( stat.st_mode & 0o7777, stat.st_uid, stat.st_gid, stat.st_dev, stat.st_ino, ) return metadata def copy_object(source, destination, source_parent=None): destination.parent.mkdir(parents=True, exist_ok=True) temporary = destination.with_name(f".{destination.name}.tmp-{uuid.uuid4().hex}") name = source.name owned_parent = None if source_parent is None: owned_parent = os.open(source.parent, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW) source_parent = owned_parent try: stat = os.stat(name, dir_fd=source_parent, follow_symlinks=False) if stat_module.S_ISLNK(stat.st_mode): temporary.symlink_to(os.readlink(name, dir_fd=source_parent)) source_path = f"/proc/self/fd/{source_parent}/{name}" for attribute in os.listxattr(source_path, follow_symlinks=False): os.setxattr( temporary, attribute, os.getxattr(source_path, attribute, follow_symlinks=False), follow_symlinks=False, ) try: os.utime( temporary, ns=(stat.st_atime_ns, stat.st_mtime_ns), follow_symlinks=False, ) except OSError as error: if error.errno not in {errno.ENOTSUP, errno.EPERM}: raise elif stat_module.S_ISREG(stat.st_mode): source_descriptor = os.open( name, os.O_RDONLY | os.O_NOFOLLOW, dir_fd=source_parent, ) try: opened = os.fstat(source_descriptor) if (opened.st_dev, opened.st_ino) != (stat.st_dev, stat.st_ino): raise RuntimeError(f"notifier object changed during copy: {source}") if opened.st_size > MAX_NOTIFIER_BYTES: raise RuntimeError(f"notifier object exceeds backup limit: {source}") target_descriptor = os.open( temporary, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, stat.st_mode & 0o7777, ) try: remaining = opened.st_size while remaining: chunk = os.read(source_descriptor, min(1024 * 1024, remaining)) if not chunk: raise RuntimeError(f"notifier object shrank during copy: {source}") view = memoryview(chunk) while view: written = os.write(target_descriptor, view) view = view[written:] remaining -= len(chunk) if os.read(source_descriptor, 1): raise RuntimeError(f"notifier object grew during copy: {source}") if os.fstat(source_descriptor).st_size != opened.st_size: raise RuntimeError(f"notifier object size changed during copy: {source}") os.fchown(target_descriptor, stat.st_uid, stat.st_gid) os.fchmod(target_descriptor, stat.st_mode & 0o7777) target_path = f"/proc/self/fd/{target_descriptor}" for attribute in os.listxattr( f"/proc/self/fd/{source_descriptor}", follow_symlinks=False ): os.setxattr( target_path, attribute, os.getxattr( f"/proc/self/fd/{source_descriptor}", attribute, follow_symlinks=False, ), follow_symlinks=False, ) os.utime( target_descriptor, ns=(stat.st_atime_ns, stat.st_mtime_ns), ) os.fsync(target_descriptor) finally: os.close(target_descriptor) finally: os.close(source_descriptor) else: raise RuntimeError(f"unsupported notifier object type: {source}") if stat_module.S_ISLNK(stat.st_mode): os.chown(temporary, stat.st_uid, stat.st_gid, follow_symlinks=False) os.replace(temporary, destination) fsync_directory(destination.parent) finally: temporary.unlink(missing_ok=True) if owned_parent is not None: os.close(owned_parent) def directory_metadata(destinations): metadata = {} for destination in destinations.values(): parent = destination.parent if parent in metadata: continue if parent.is_dir() and not parent.is_symlink(): stat = parent.stat() metadata[parent] = ( stat.st_mode & 0o7777, stat.st_uid, stat.st_gid, stat.st_dev, stat.st_ino, ) else: metadata[parent] = None return metadata def encode_directory_metadata(metadata, home, portable=True): return { str(path.relative_to(home)): ( values[:3] if portable and values is not None else values ) for path, values in metadata.items() } def decode_directory_metadata(encoded, home): metadata = {} for relative, values in encoded.items(): path = Path(relative) if ( path.is_absolute() or not path.parts or any(part in {".", ".."} for part in path.parts) ): raise RuntimeError("invalid retired notifier directory metadata") if values is not None and len(values) not in {3, 5}: raise RuntimeError("invalid retired notifier directory metadata") metadata[home / path] = tuple(values) if values is not None else None return metadata def mkdir_as_identity(name, mode, descriptor, uid, gid): if os.geteuid() == uid and os.getegid() == gid: os.mkdir(name, mode=mode, dir_fd=descriptor) return if os.geteuid() != 0: raise RuntimeError("cannot create notifier directory as recorded owner") child = os.fork() if child == 0: try: os.setgroups([]) os.setgid(gid) os.setuid(uid) os.mkdir(name, mode=mode, dir_fd=descriptor) except BaseException: os._exit(1) os._exit(0) _, status = os.waitpid(child, 0) if status != 0: raise OSError("target-identity directory creation failed") def ensure_directories( home, metadata, created, expected_current=None, journal_created=None, ): if home.is_symlink() or not home.is_dir(): raise RuntimeError("target home must be an existing non-symlink directory") home = home.resolve(strict=True) for path, values in metadata.items(): if values is None: continue relative = path.relative_to(home) descriptor = os.open(home, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW) current = home try: for part in relative.parts: current = current / part expected_absent = ( expected_current is not None and current in expected_current and expected_current[current] is None ) if expected_absent: created[current] = None if journal_created is not None: journal_created(created) try: mode, uid, gid = values[:3] mkdir_as_identity(part, mode, descriptor, uid, gid) except FileExistsError as error: created.pop(current) if journal_created is not None: journal_created(created) raise RuntimeError( f"expected absent target directory appeared: {current}" ) from error try: child = os.open( part, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=descriptor, ) child_stat = os.fstat(child) except Exception: os.rmdir(part, dir_fd=descriptor) os.fsync(descriptor) raise created[current] = (child_stat.st_dev, child_stat.st_ino) if journal_created is not None: journal_created(created) os.fsync(descriptor) else: try: child = os.open( part, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=descriptor, ) except FileNotFoundError: created[current] = None if journal_created is not None: journal_created(created) try: mode, uid, gid = values[:3] mkdir_as_identity(part, mode, descriptor, uid, gid) except Exception: created.pop(current) if journal_created is not None: journal_created(created) raise try: child = os.open( part, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=descriptor, ) child_stat = os.fstat(child) except Exception: os.rmdir(part, dir_fd=descriptor) os.fsync(descriptor) raise created[current] = (child_stat.st_dev, child_stat.st_ino) if journal_created is not None: journal_created(created) os.fsync(descriptor) child_stat = os.fstat(child) expected = expected_current.get(current) if expected_current is not None else None if expected is not None and (child_stat.st_dev, child_stat.st_ino) != expected[3:5]: os.close(child) raise RuntimeError(f"target directory identity changed: {current}") os.close(descriptor) descriptor = child mode, uid, gid = values[:3] os.fchown(descriptor, uid, gid) os.fchmod(descriptor, mode) os.fsync(descriptor) finally: os.close(descriptor) def remove_created_directories(home, created): failures = [] root = os.open(home, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW) try: for path, identity in sorted( created.items(), key=lambda item: len(item[0].parts), reverse=True, ): descriptor = os.dup(root) try: for part in path.parent.relative_to(home).parts: child = os.open( part, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=descriptor, ) os.close(descriptor) descriptor = child stat = os.stat(path.name, dir_fd=descriptor, follow_symlinks=False) if identity is not None and ( (stat.st_dev, stat.st_ino) != identity or not stat_module.S_ISDIR(stat.st_mode) ): raise RuntimeError(f"created directory identity changed: {path}") if not stat_module.S_ISDIR(stat.st_mode): raise RuntimeError(f"created directory is not a directory: {path}") os.rmdir(path.name, dir_fd=descriptor) os.fsync(descriptor) except FileNotFoundError: continue except Exception as error: failures.append(f"{path}: {error}") finally: os.close(descriptor) finally: os.close(root) if failures: raise RuntimeError("created directory cleanup failed: " + "; ".join(failures)) def restore_directory_metadata(home, metadata): existing = {path: values for path, values in metadata.items() if values is not None} if not existing: return with locked_directories(home, existing) as descriptors: for path, values in existing.items(): descriptor = descriptors[path] mode, uid, gid = values[:3] os.fchown(descriptor, uid, gid) os.fchmod(descriptor, mode) os.fsync(descriptor) def restore_directories(metadata): for path, values in metadata.items(): if values is None: continue mode, uid, gid = values[:3] path.mkdir(parents=True, exist_ok=True) os.chown(path, uid, gid) os.chmod(path, mode) def update_manifest(archive, **changes): manifest = read_manifest(archive) manifest.update(changes) write_manifest(archive, manifest) def manifest_created(home, created): return { str(path.relative_to(home)): identity for path, identity in created.items() } def write_manifest(archive, manifest): path = archive / ".notifier-transaction.json" temporary = archive / f".manifest-{uuid.uuid4().hex}" try: temporary.write_text(json.dumps(manifest, sort_keys=True) + "\n") fsync_file(temporary) os.replace(temporary, path) fsync_directory(archive) finally: temporary.unlink(missing_ok=True) def clear_manifest(archive): path = archive / ".notifier-transaction.json" path.unlink(missing_ok=True) fsync_directory(archive) def read_manifest(archive): path = archive / ".notifier-transaction.json" if not path.exists(): return None manifest = json.loads(path.read_text()) if not isinstance(manifest, dict): raise RuntimeError("invalid notifier transaction manifest") return manifest def portable_object_identity(path): identity = object_identity(path) if identity is None: return None return identity[2:] def portable_bound_identity(descriptor, name): identity = bound_leaf_identity(descriptor, name) if identity is None: return None return identity[2:] def decode_created_paths(encoded, home): if not isinstance(encoded, dict): raise RuntimeError("invalid notifier created-directory journal") created = {} for relative, identity in encoded.items(): path = Path(relative) if ( not isinstance(relative, str) or path.is_absolute() or not path.parts or any(part in {".", ".."} for part in path.parts) ): raise RuntimeError("invalid notifier created-directory journal") if identity is not None and ( not isinstance(identity, list) or len(identity) != 2 ): raise RuntimeError("invalid notifier created-directory journal") created[home / path] = tuple(identity) if identity is not None else None return created def manifest_backup(archive, manifest): name = manifest.get("backup") if ( not isinstance(name, str) or not name or name in {".", ".."} or Path(name).name != name or os.sep in name ): raise RuntimeError("invalid notifier transaction backup") backup = archive / name if backup.is_symlink() or not backup.is_dir(): raise RuntimeError("notifier transaction backup is missing") archive_stat = archive.stat() backup_stat = backup.stat() if backup_stat.st_dev != archive_stat.st_dev or backup.parent.resolve() != archive.resolve(): raise RuntimeError("notifier transaction backup escaped archive") return backup def recover_transaction(user, uid, home, archive): manifest = read_manifest(archive) if manifest is None: return if manifest.get("version") != 1 or manifest.get("home") != str(home): raise RuntimeError("notifier transaction manifest identity mismatch") action = manifest.get("action") if action not in {"retire", "restore"}: raise RuntimeError("invalid notifier transaction action") destinations = notifier_paths(home) backup = manifest_backup(archive, manifest) captures = manifest.get("captures") if not isinstance(captures, dict) or set(captures) != set(destinations): raise RuntimeError("invalid notifier transaction captures") for capture in captures.values(): if capture is not None and ( not isinstance(capture, str) or not capture or Path(capture).name != capture or capture in {".", ".."} ): raise RuntimeError("invalid notifier transaction capture") directory_values = manifest.get("directories") directories = decode_directory_metadata(directory_values, home) existing = directory_metadata(destinations) bound = {path: values for path, values in existing.items() if values is not None} phase = manifest.get("phase") if phase == "publishing": generation_name = manifest.get("generation") if ( not isinstance(generation_name, str) or Path(generation_name).name != generation_name or generation_name in {"", ".", ".."} ): raise RuntimeError("invalid staged notifier generation") generation = archive / generation_name if current_generation(archive) == generation: update_manifest(archive, phase="committed") manifest = read_manifest(archive) phase = "committed" if phase == "committed": with locked_directories(home, bound) as cleanup_descriptors: pending = dict(captures) delete_pending_captures( pending, destinations, cleanup_descriptors, manifest.get("capture_identities", {}), archive, ) clear_manifest(archive) shutil.rmtree(backup) fsync_directory(archive) return with locked_directories(home, bound) as descriptors: for name, capture in captures.items(): destination = destinations[name] descriptor = descriptors.get(destination.parent) if descriptor is None: if capture is not None: raise RuntimeError(f"notifier capture parent is missing: {destination.parent}") continue capture_exists = capture is not None and bound_leaf_identity(descriptor, capture) is not None if capture_exists: expected_capture = manifest.get("capture_identities", {}).get(name) current_capture = capture_identity(descriptor, capture) if expected_capture is None or current_capture != tuple(expected_capture): raise RuntimeError(f"notifier capture identity changed: {destination}") if action == "restore": expected = manifest.get("expected", {}).get(name) current = portable_bound_identity(descriptor, destination.name) if expected is not None: if current == tuple(expected): unlink_destination(destination, descriptors) elif current is not None: raise RuntimeError( f"published notifier object identity changed: {destination}" ) elif current is not None: raise RuntimeError( f"unexpected notifier object during recovery: {destination}" ) if capture_exists: rename_noreplace(descriptor, capture, descriptor, destination.name) os.fsync(descriptor) states = validate_recorded_states(manifest["states"]) rollback(user, uid, backup, destinations, directories, states, restore_files=False) if action == "retire": generation_name = manifest.get("generation") if generation_name is not None: if ( not isinstance(generation_name, str) or Path(generation_name).name != generation_name or generation_name in {"", ".", ".."} ): raise RuntimeError("invalid staged notifier generation") generation = archive / generation_name if current_generation(archive) == generation: raise RuntimeError("uncommitted notifier generation is current") shutil.rmtree(generation) fsync_directory(archive) if action == "restore": restore_directory_metadata(home, directories) created = decode_created_paths(manifest.get("created", {}), home) remove_created_directories(home, created) clear_manifest(archive) shutil.rmtree(backup) fsync_directory(archive) @contextmanager def transaction_lock(archive): descriptor = os.open(archive / ".transaction.lock", os.O_RDWR | os.O_CREAT, 0o600) try: fcntl.flock(descriptor, fcntl.LOCK_EX) yield finally: fcntl.flock(descriptor, fcntl.LOCK_UN) os.close(descriptor) @contextmanager def locked_directories(home, metadata): descriptors = [] root = os.open(home, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW) try: for path, values in metadata.items(): descriptor = os.dup(root) try: for part in path.relative_to(home).parts: child = os.open( part, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=descriptor, ) os.close(descriptor) descriptor = child stat = os.fstat(descriptor) if (stat.st_dev, stat.st_ino) != values[3:5]: raise RuntimeError(f"target directory identity changed: {path}") descriptors.append((descriptor, path)) except Exception: os.close(descriptor) raise yield {path: descriptor for descriptor, path in descriptors} finally: for descriptor, _ in descriptors: os.close(descriptor) os.close(root) def destination_home(destinations): return Path(os.path.commonpath([str(path) for path in destinations.values()])) def destination_binding(destination, descriptors): return descriptors[destination.parent], destination.name def digest_descriptor(descriptor): digest = hashlib.sha256() os.lseek(descriptor, 0, os.SEEK_SET) with os.fdopen(os.dup(descriptor), "rb") as source: for chunk in iter(lambda: source.read(1024 * 1024), b""): digest.update(chunk) return digest.hexdigest() def object_digest(path): if not exists(path): return None if path.is_symlink(): return hashlib.sha256(os.fsencode(os.readlink(path))).hexdigest() digest = hashlib.sha256() descriptor = os.open(path, os.O_RDONLY | os.O_NOFOLLOW) try: return digest_descriptor(descriptor) finally: os.close(descriptor) def xattr_digest(path=None, descriptor=None): target = path if descriptor is None else f"/proc/self/fd/{descriptor}" digest = hashlib.sha256() try: names = sorted(os.listxattr(target, follow_symlinks=False)) for name in names: digest.update(os.fsencode(name)) digest.update(b"\0") digest.update(os.getxattr(target, name, follow_symlinks=False)) digest.update(b"\0") except OSError as error: if error.errno not in {errno.ENOTSUP, errno.EPERM}: raise return digest.hexdigest() def object_identity(path): try: stat = path.lstat() except FileNotFoundError: return None if stat_module.S_ISLNK(stat.st_mode): digest = hashlib.sha256(os.fsencode(os.readlink(path))).hexdigest() attributes = xattr_digest(path=path) else: if not stat_module.S_ISREG(stat.st_mode): raise RuntimeError(f"unsupported notifier object type: {path}") descriptor = os.open(path, os.O_RDONLY | os.O_NOFOLLOW) try: opened = os.fstat(descriptor) if (opened.st_dev, opened.st_ino) != (stat.st_dev, stat.st_ino): raise RuntimeError(f"notifier object changed during identity read: {path}") if opened.st_size > MAX_NOTIFIER_BYTES: raise RuntimeError(f"notifier object exceeds backup limit: {path}") digest = digest_descriptor(descriptor) if os.fstat(descriptor).st_size != opened.st_size: raise RuntimeError(f"notifier object size changed during identity read: {path}") attributes = xattr_digest(descriptor=descriptor) stat = opened finally: os.close(descriptor) return ( stat.st_dev, stat.st_ino, stat.st_mode, stat.st_uid, stat.st_gid, stat.st_size, stat.st_mtime_ns, digest, attributes, ) def assert_object_identities(destinations, identities, descriptors): for name, destination in destinations.items(): if destination.parent not in descriptors: if identities[name] is not None: raise RuntimeError(f"notifier parent directory disappeared: {destination.parent}") continue descriptor, leaf = destination_binding(destination, descriptors) try: stat = os.stat(leaf, dir_fd=descriptor, follow_symlinks=False) current = stat.st_dev, stat.st_ino, stat.st_mode except FileNotFoundError: current = None if current != identities[name]: raise RuntimeError(f"notifier object identity changed: {destination}") def unlink_destination(destination, descriptors): descriptor, name = destination_binding(destination, descriptors) try: os.unlink(name, dir_fd=descriptor) except FileNotFoundError: pass def rename_noreplace(first_dir_fd, first, second_dir_fd, second): result = LIBC.renameat2( first_dir_fd, os.fsencode(first), second_dir_fd, os.fsencode(second), RENAME_NOREPLACE, ) if result != 0: error = ctypes.get_errno() raise OSError(error, os.strerror(error)) def capture_leaf(destination, descriptor, capture): name = destination.name try: rename_noreplace(descriptor, name, descriptor, capture) except OSError as error: if error.errno != errno.ENOENT: raise return None return capture def bound_object_digest(descriptor, name): stat = os.stat(name, dir_fd=descriptor, follow_symlinks=False) if stat_module.S_ISLNK(stat.st_mode): return hashlib.sha256(os.fsencode(os.readlink(name, dir_fd=descriptor))).hexdigest() source = os.open(name, os.O_RDONLY | os.O_NOFOLLOW, dir_fd=descriptor) try: return digest_descriptor(source) finally: os.close(source) def bound_xattr_digest(descriptor, name): if stat_module.S_ISLNK( os.stat(name, dir_fd=descriptor, follow_symlinks=False).st_mode ): target = f"/proc/self/fd/{descriptor}/{name}" return xattr_digest(path=target) source = os.open(name, os.O_RDONLY | os.O_NOFOLLOW, dir_fd=descriptor) try: return xattr_digest(descriptor=source) finally: os.close(source) def capture_identity(descriptor, capture): if capture is None: return None stat = os.stat(capture, dir_fd=descriptor, follow_symlinks=False) return ( stat.st_dev, stat.st_ino, stat.st_mode, stat.st_uid, stat.st_gid, stat.st_size, stat.st_mtime_ns, bound_object_digest(descriptor, capture), bound_xattr_digest(descriptor, capture), ) def bound_leaf_identity(descriptor, name): try: stat = os.stat(name, dir_fd=descriptor, follow_symlinks=False) except FileNotFoundError: return None return ( stat.st_dev, stat.st_ino, stat.st_mode, stat.st_uid, stat.st_gid, stat.st_size, stat.st_mtime_ns, bound_object_digest(descriptor, name), bound_xattr_digest(descriptor, name), ) def restore_capture(destination, descriptor, capture): if capture is None: return try: rename_noreplace(descriptor, capture, descriptor, destination.name) except OSError as error: if error.errno != errno.EEXIST: raise os.fsync(descriptor) def delete_capture(descriptor, capture): if capture is None: return try: os.unlink(capture, dir_fd=descriptor) except FileNotFoundError: pass os.fsync(descriptor) def restore_pending_captures( captures, destinations, descriptors, unlink_installed=False, published_identities=None, ): failures = [] for name in list(captures): capture = captures[name] destination = destinations[name] descriptor = descriptors.get(destination.parent) try: if unlink_installed and published_identities is not None and name in published_identities: if descriptor is None: raise RuntimeError(f"capture parent directory is missing: {destination.parent}") expected = published_identities[name] current_identity = bound_leaf_identity(descriptor, destination.name) current = current_identity[2:] if current_identity is not None else None if current != expected: raise RuntimeError( f"published notifier object identity changed: {destination}" ) if current is not None: unlink_destination(destination, descriptors) if capture is None: captures.pop(name) if descriptor is not None: os.fsync(descriptor) continue if descriptor is None: raise RuntimeError(f"capture parent directory is missing: {destination.parent}") capture_present = bound_leaf_identity(descriptor, capture) is not None if not capture_present: if published_identities is not None and name in published_identities: captures.pop(name) os.fsync(descriptor) continue raise RuntimeError(f"notifier capture is missing: {destination}") rename_noreplace(descriptor, capture, descriptor, destination.name) captures.pop(name) os.fsync(descriptor) except Exception as error: failures.append(f"{destination}: {error}") if failures: raise RuntimeError("capture restoration failed: " + "; ".join(failures)) def delete_pending_captures( captures, destinations, descriptors, expected_identities=None, archive=None, ): failures = [] for name in list(captures): capture = captures[name] destination = destinations[name] descriptor = descriptors.get(destination.parent) try: if capture is not None: if descriptor is None: raise RuntimeError(f"capture parent directory is missing: {destination.parent}") expected = expected_identities.get(name) if expected_identities is not None else None trash = f"{capture}.cleanup" try: rename_noreplace(descriptor, capture, descriptor, trash) except OSError as error: if error.errno != errno.ENOENT or archive is None: raise try: current_trash = capture_identity(descriptor, trash) except FileNotFoundError: captures.pop(name) update_manifest(archive, captures=captures) continue if expected is None or current_trash != tuple(expected): raise RuntimeError(f"notifier cleanup identity changed: {destination}") os.fsync(descriptor) current = capture_identity(descriptor, trash) if expected_identities is not None and ( expected is None or current != tuple(expected) ): rename_noreplace(descriptor, trash, descriptor, capture) os.fsync(descriptor) raise RuntimeError(f"notifier capture identity changed: {destination}") os.unlink(trash, dir_fd=descriptor) os.fsync(descriptor) captures.pop(name) if archive is not None: update_manifest(archive, captures=captures) if descriptor is not None: os.fsync(descriptor) except Exception as error: failures.append(f"{destination}: {error}") if failures: raise RuntimeError("capture cleanup failed: " + "; ".join(failures)) def replace_from_source(source, destination, descriptors): descriptor, name = destination_binding(destination, descriptors) temporary = f".{name}.tmp-{uuid.uuid4().hex}" try: stat = source.lstat() if stat_module.S_ISLNK(stat.st_mode): os.symlink(os.readlink(source), temporary, dir_fd=descriptor) temporary_path = f"/proc/self/fd/{descriptor}/{temporary}" os.utime( temporary_path, ns=(stat.st_atime_ns, stat.st_mtime_ns), follow_symlinks=False, ) for attribute in os.listxattr(source, follow_symlinks=False): os.setxattr( temporary_path, attribute, os.getxattr(source, attribute, follow_symlinks=False), follow_symlinks=False, ) elif stat_module.S_ISREG(stat.st_mode): source_descriptor = os.open(source, os.O_RDONLY | os.O_NOFOLLOW) try: opened = os.fstat(source_descriptor) if (opened.st_dev, opened.st_ino) != (stat.st_dev, stat.st_ino): raise RuntimeError(f"notifier restore source changed: {source}") if opened.st_size > MAX_NOTIFIER_BYTES: raise RuntimeError(f"notifier object exceeds restore limit: {source}") target_descriptor = os.open( temporary, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, stat.st_mode & 0o7777, dir_fd=descriptor, ) try: remaining = opened.st_size while remaining: chunk = os.read(source_descriptor, min(1024 * 1024, remaining)) if not chunk: raise RuntimeError(f"notifier restore source shrank: {source}") view = memoryview(chunk) while view: written = os.write(target_descriptor, view) view = view[written:] remaining -= len(chunk) if os.read(source_descriptor, 1): raise RuntimeError(f"notifier restore source grew: {source}") if os.fstat(source_descriptor).st_size != opened.st_size: raise RuntimeError(f"notifier restore source size changed: {source}") os.fchown(target_descriptor, stat.st_uid, stat.st_gid) os.fchmod(target_descriptor, stat.st_mode & 0o7777) target_path = f"/proc/self/fd/{target_descriptor}" for attribute in os.listxattr( f"/proc/self/fd/{source_descriptor}", follow_symlinks=False ): os.setxattr( target_path, attribute, os.getxattr( f"/proc/self/fd/{source_descriptor}", attribute, follow_symlinks=False, ), follow_symlinks=False, ) os.utime( target_descriptor, ns=(stat.st_atime_ns, stat.st_mtime_ns), ) os.fsync(target_descriptor) finally: os.close(target_descriptor) finally: os.close(source_descriptor) else: raise RuntimeError(f"unsupported notifier restore object type: {source}") if stat_module.S_ISLNK(stat.st_mode): os.chown(temporary, stat.st_uid, stat.st_gid, dir_fd=descriptor, follow_symlinks=False) rename_noreplace(descriptor, temporary, descriptor, name) return bound_leaf_identity(descriptor, name) finally: try: os.unlink(temporary, dir_fd=descriptor) except FileNotFoundError: pass def restore_objects(backups, destinations, directories, descriptors=None): restore_directories(directories) if not any(exists(backups / name) for name in destinations): return context = ( locked_directories(destination_home(destinations), directory_metadata(destinations)) if descriptors is None else None ) if context is not None: with context as bound: restore_objects(backups, destinations, directories, bound) return for name, destination in destinations.items(): backup = backups / name if exists(backup): replace_from_source(backup, destination, descriptors) else: unlink_destination(destination, descriptors) def rollback(user, uid, backups, destinations, directories, states, restore_files=True): failures = [] if restore_files: try: restore_objects(backups, destinations, directories) except Exception as error: failures.append(f"files: {error}") try: run_systemctl(user, uid, "daemon-reload") except Exception as error: failures.append(f"daemon-reload: {error}") try: apply_service_states(user, uid, states) except Exception as error: failures.append(f"services: {error}") if failures: raise RuntimeError("rollback failed: " + "; ".join(failures)) def disable_services(user, uid, states): for service, state in states.items(): if state["load"] != "not-found": if state["enabled"] == "enabled": run_systemctl(user, uid, "disable", "--now", service) elif state["active"] == "active": run_systemctl(user, uid, "stop", service) final = service_state(user, uid, service) if final["active"] != "inactive" or final["enabled"] not in { "disabled", "linked", "not-found", }: raise RuntimeError(f"unsafe state after disabling {service}: {final}") def current_generation(archive): current = archive / "current" if not current.is_symlink(): return None target = Path(os.readlink(current)) if ( target.is_absolute() or len(target.parts) != 1 or target.name in {".", ".."} ): raise RuntimeError("invalid retired notifier generation link") generation = archive / target if generation.is_symlink() or not generation.is_dir(): raise RuntimeError("retired notifier generation is missing") return generation def stage_generation(archive, staged): for child in staged.iterdir(): if child.is_symlink(): continue fsync_file(child) fsync_directory(staged) generation = f"generation-{uuid.uuid4().hex}" final = archive / generation os.replace(staged, final) fsync_directory(archive) return final def publish_current(archive, generation): current = archive / "current" link = archive / f".current-{uuid.uuid4().hex}" descriptor = os.open(archive, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW) try: link.symlink_to(generation.name) rename_noreplace(descriptor, link.name, descriptor, current.name) os.fsync(descriptor) except Exception: link.unlink(missing_ok=True) raise finally: os.close(descriptor) def _retire(user, uid, home, state_dir, dry_run=False): destinations = notifier_paths(home) archive = state_dir / "retired" if dry_run: for service in SERVICES: print(f"DRY-RUN: systemctl disable {service}") for destination in destinations.values(): print(f"DRY-RUN: archive {destination}") print(f"DRY-RUN: rm {destination}") print("DRY-RUN: systemctl user-daemon-reload") return states = service_states(user, uid) state_dir.mkdir(parents=True, exist_ok=True) archive_existed = archive.is_dir() archive.mkdir(parents=True, exist_ok=True) if not archive_existed: fsync_directory(state_dir) directories = validate_directories(home, destinations, uid) existing_generation = current_generation(archive) backup = Path(tempfile.mkdtemp(prefix="notifier-backup-", dir=archive)) try: backups = backup identities = {} existing_directories = {path: value for path, value in directories.items() if value} with locked_directories(home, existing_directories) as source_directories: for name, source in destinations.items(): descriptor = source_directories.get(source.parent) before = object_identity(source) if before is not None: if descriptor is None: raise RuntimeError(f"notifier parent directory disappeared: {source.parent}") copy_object(source, backups / name, descriptor) after = object_identity(source) if after != before: raise RuntimeError(f"notifier object changed during backup: {source}") identities[name] = before staged_raw = None staged_generation = None files_mutated = False committed = False captures = { name: None for name in destinations } write_manifest( archive, { "version": 1, "action": "retire", "home": str(home), "backup": backup.name, "states": states, "directories": encode_directory_metadata(directories, home, portable=False), "captures": captures, "capture_identities": {}, "generation": None, "phase": "mutating", }, ) try: if existing_generation is None: staged_raw = tempfile.mkdtemp(prefix=".notifier-generation-", dir=archive) staged = Path(staged_raw) for name in NAMES: source = backups / name if exists(source): copy_object(source, staged / name) (staged / "notifier-state.json").write_text( json.dumps( { "services": states, "directories": encode_directory_metadata(directories, home), }, sort_keys=True, ) + "\n" ) staged_generation = stage_generation(archive, Path(staged_raw)) staged_raw = None update_manifest(archive, generation=staged_generation.name) disable_services(user, uid, states) existing_directories = {path: value for path, value in directories.items() if value} with locked_directories(home, existing_directories) as descriptors: try: for name, destination in destinations.items(): if destination.parent not in descriptors: continue descriptor = descriptors[destination.parent] capture = f".{destination.name}.retire-{uuid.uuid4().hex}" captures[name] = capture manifest = read_manifest(archive) manifest["captures"] = captures capture_identities = manifest["capture_identities"] capture_identities[name] = ( list(identities[name]) if identities[name] is not None else None ) manifest["capture_identities"] = capture_identities write_manifest(archive, manifest) capture = capture_leaf( destination, descriptor, capture, ) os.fsync(descriptor) captures[name] = capture captured_identity = capture_identity(descriptor, capture) manifest = read_manifest(archive) manifest["captures"] = captures capture_identities = manifest["capture_identities"] capture_identities[name] = ( list(captured_identity) if captured_identity is not None else None ) manifest["capture_identities"] = capture_identities write_manifest(archive, manifest) if captured_identity != identities[name]: raise RuntimeError(f"notifier object identity changed: {destination}") files_mutated = True except Exception: restore_pending_captures(captures, destinations, descriptors) raise run_systemctl(user, uid, "daemon-reload") for service in SERVICES: final = service_state(user, uid, service) if final != { "load": "not-found", "active": "inactive", "enabled": "not-found", }: raise RuntimeError( f"obsolete service remains after retirement: {service}: {final}" ) if staged_generation is not None: update_manifest(archive, phase="publishing") publish_current(archive, staged_generation) update_manifest(archive, phase="committed") committed = True with locked_directories(home, existing_directories) as descriptors: delete_pending_captures( captures, destinations, descriptors, read_manifest(archive)["capture_identities"], archive, ) clear_manifest(archive) shutil.rmtree(backup) fsync_directory(archive) except Exception as original: if committed: raise rollback_failures = [] if captures: try: existing_directories = { path: value for path, value in directories.items() if value } with locked_directories(home, existing_directories) as descriptors: restore_pending_captures(captures, destinations, descriptors) except Exception as error: rollback_failures.append(f"captures: {error}") try: rollback( user, uid, backups, destinations, directories, states, restore_files=False, ) except Exception as error: rollback_failures.append(str(error)) if rollback_failures: raise RuntimeError( f"retirement failed: {original}; rollback failed: " + "; ".join(rollback_failures) ) from original clear_manifest(archive) shutil.rmtree(backup) if staged_generation is not None: shutil.rmtree(staged_generation) fsync_directory(archive) raise finally: if staged_raw is not None: shutil.rmtree(staged_raw, ignore_errors=True) except Exception: if read_manifest(archive) is None: shutil.rmtree(backup, ignore_errors=True) raise def retire(user, uid, home, state_dir, dry_run=False): if dry_run: return _retire(user, uid, home, state_dir, True) state_dir.mkdir(parents=True, exist_ok=True) archive = state_dir / "retired" archive.mkdir(parents=True, exist_ok=True) with transaction_lock(archive): recover_transaction(user, uid, home, archive) return _retire(user, uid, home, state_dir, False) def restore(user, uid, home, state_dir): archive = state_dir / "retired" if not archive.is_dir(): raise RuntimeError("no retired notifier archive exists") with transaction_lock(archive): recover_transaction(user, uid, home, archive) return _restore(user, uid, home, state_dir) def _restore(user, uid, home, state_dir): destinations = notifier_paths(home) archive = state_dir / "retired" generation = current_generation(archive) if generation is None: raise RuntimeError("no retired notifier generation exists") receipt = json.loads((generation / "notifier-state.json").read_text()) if not isinstance(receipt, dict) or set(receipt) != {"services", "directories"}: raise RuntimeError("invalid retired notifier receipt") desired_states = validate_recorded_states(receipt["services"]) archived_directories = decode_directory_metadata(receipt["directories"], home) validate_directories(home, destinations, uid) current_states = service_states(user, uid) current_directories = directory_metadata(destinations) backup = Path(tempfile.mkdtemp(prefix="notifier-restore-", dir=archive)) try: backups = backup identities = {} captures = {name: None for name in destinations} published_identities = {} for name, destination in destinations.items(): before = object_identity(destination) if before is not None: copy_object(destination, backups / name) after = object_identity(destination) if after != before: raise RuntimeError(f"notifier object changed during restore backup: {destination}") identities[name] = before files_mutated = False committed = False created_directories = {} write_manifest( archive, { "version": 1, "action": "restore", "home": str(home), "backup": backup.name, "states": current_states, "directories": encode_directory_metadata(current_directories, home, portable=False), "captures": captures, "capture_identities": {}, "published": published_identities, "expected": {}, "created": {}, "phase": "mutating", }, ) try: missing = [ path for path, values in archived_directories.items() if values is not None and current_directories.get(path) is None ] if missing: raise RuntimeError( "restore target directory is missing: " + ", ".join(str(path) for path in missing) ) for path, values in current_directories.items(): archived = archived_directories.get(path) if values is not None and archived is not None: archived_directories[path] = (*archived[:3], *values[3:5]) bound_metadata = { path: values for path, values in directory_metadata(destinations).items() if values is not None } with locked_directories(home, bound_metadata) as descriptors: try: for name, destination in destinations.items(): source = generation / name if destination.parent not in descriptors: if identities[name] is not None or exists(source): raise RuntimeError( f"restore target directory is missing: {destination.parent}" ) expected_identity = None expected = manifest["expected"] expected[name] = expected_identity manifest["expected"] = expected published_identities[name] = None manifest["published"] = published_identities write_manifest(archive, manifest) continue descriptor = descriptors[destination.parent] capture = f".{destination.name}.retire-{uuid.uuid4().hex}" captures[name] = capture manifest = read_manifest(archive) manifest["captures"] = captures capture_identities = manifest["capture_identities"] capture_identities[name] = ( list(identities[name]) if identities[name] is not None else None ) manifest["capture_identities"] = capture_identities write_manifest(archive, manifest) capture = capture_leaf(destination, descriptor, capture) os.fsync(descriptor) captures[name] = capture captured_identity = capture_identity(descriptor, capture) manifest = read_manifest(archive) manifest["captures"] = captures capture_identities = manifest["capture_identities"] capture_identities[name] = ( list(captured_identity) if captured_identity is not None else None ) manifest["capture_identities"] = capture_identities write_manifest(archive, manifest) if captured_identity != identities[name]: raise RuntimeError(f"notifier object identity changed: {destination}") source = generation / name expected_identity = portable_object_identity(source) if exists(source) else None manifest = read_manifest(archive) expected = manifest["expected"] expected[name] = expected_identity manifest["expected"] = expected write_manifest(archive, manifest) if exists(source): published_identities[name] = replace_from_source( source, destination, descriptors, )[2:] else: published_identities[name] = None manifest = read_manifest(archive) manifest["published"] = published_identities write_manifest(archive, manifest) os.fsync(descriptor) files_mutated = True except Exception: restore_pending_captures( captures, destinations, descriptors, unlink_installed=True, published_identities=published_identities, ) raise run_systemctl(user, uid, "daemon-reload") apply_service_states(user, uid, desired_states) restore_directory_metadata(home, archived_directories) update_manifest(archive, phase="committed") committed = True with locked_directories(home, bound_metadata) as descriptors: delete_pending_captures( captures, destinations, descriptors, read_manifest(archive)["capture_identities"], archive, ) clear_manifest(archive) shutil.rmtree(backup) fsync_directory(archive) except Exception as original: if committed: raise rollback_failures = [] if captures: try: restored_directories = directory_metadata(destinations) with locked_directories(home, restored_directories) as descriptors: restore_pending_captures( captures, destinations, descriptors, unlink_installed=True, published_identities=published_identities, ) except Exception as error: rollback_failures.append(f"captures: {error}") try: rollback( user, uid, backups, destinations, current_directories, current_states, restore_files=False, ) except Exception as error: rollback_failures.append(str(error)) try: restore_directory_metadata(home, current_directories) except Exception as error: rollback_failures.append(f"directory metadata: {error}") if created_directories: try: remove_created_directories(home, created_directories) except Exception as error: rollback_failures.append(f"directories: {error}") if rollback_failures: raise RuntimeError( f"restore failed: {original}; rollback failed: " + "; ".join(rollback_failures) ) from original clear_manifest(archive) shutil.rmtree(backup) fsync_directory(archive) raise except Exception: if read_manifest(archive) is None: shutil.rmtree(backup, ignore_errors=True) raise def main(): parser = argparse.ArgumentParser() parser.add_argument("action", choices=("retire", "restore")) parser.add_argument("--user", required=True) parser.add_argument("--uid", required=True, type=int) parser.add_argument("--home", required=True, type=Path) parser.add_argument("--state-dir", required=True, type=Path) parser.add_argument("--dry-run", action="store_true") args = parser.parse_args() if args.uid < 0 or not args.home.is_absolute() or args.home == Path("/"): parser.error("invalid target identity") if args.action == "retire": retire(args.user, args.uid, args.home, args.state_dir, args.dry_run) else: if args.dry_run: parser.error("--dry-run is valid only for retire") restore(args.user, args.uid, args.home, args.state_dir) if __name__ == "__main__": main()