← Back
Editing: safe_fileops.py
import asyncio import atexit import functools import logging import os import pathlib import pwd import shutil import stat from concurrent.futures import ProcessPoolExecutor from contextlib import contextmanager, suppress from itertools import chain from typing import Set, Tuple, Union from defence360agent import utils R_FLAGS = os.O_RDONLY W_FLAGS = os.O_TRUNC | os.O_CREAT | os.O_WRONLY logger = logging.getLogger(__name__) # Track active ProcessPoolExecutors so they can be cleaned up during shutdown. # Each call to drop() permanently changes the worker process identity, so we # must use a fresh executor per call. We track them to prevent orphaned worker # processes from blocking agent shutdown (causing systemd SIGKILL). _active_pools: Set[ProcessPoolExecutor] = set() async def _run_in_fresh_executor(loop: asyncio.AbstractEventLoop, *args): pool = ProcessPoolExecutor(max_workers=1) _active_pools.add(pool) try: return await loop.run_in_executor(pool, *args) finally: try: pool.shutdown(wait=False) finally: _active_pools.discard(pool) def shutdown_process_pools() -> None: """Shutdown all tracked ProcessPoolExecutors. Should be called during agent shutdown to ensure clean process termination. """ for pool in list(_active_pools): try: pool.shutdown(wait=False, cancel_futures=True) except Exception as e: logger.warning("Error shutting down ProcessPoolExecutor: %s", e) _active_pools.clear() # Register cleanup at exit as a fallback atexit.register(shutdown_process_pools) def drop(fun, uid, gid, *args): os.setgroups([]) os.setgid(gid) os.setuid(uid) return fun(*args) class UnsafeFileOperation(Exception): pass def ensure_regular_file(path: str) -> None: """Verify path is a regular file; remove and raise FileNotFoundError if not. Uses os.lstat() to avoid following symlinks. If the file is a FIFO, symlink, socket, device, etc., it is deleted so the caller can recreate it as a regular file. """ st = os.lstat(path) # raises FileNotFoundError if missing if not stat.S_ISREG(st.st_mode): logger.warning( "Identity file %s is not a regular file (mode=%s), removing", path, stat.filemode(st.st_mode), ) os.unlink(path) raise FileNotFoundError(f"Removed non-regular identity file: {path}") def check_non_admin_file(file): st = os.stat(str(file)) if st.st_uid < utils.get_min_uid(): raise UnsafeFileOperation( "The file belongs to admin user: " + str(file) ) return True def safe(missing_ok=False): def _safe(fun): @functools.wraps(fun) async def wrapper(filename, *args, loop=None): if not os.path.exists(filename) and not missing_ok: raise FileNotFoundError( "No such file or directory: " + filename ) path = pathlib.Path(filename) paths = chain(reversed(path.parents), [path]) if missing_ok: paths = reversed(path.parents) for p in paths: st = os.stat(str(p)) if st.st_uid != 0 and st.st_gid != 0: uid, gid = st.st_uid, st.st_gid break else: raise UnsafeFileOperation( "Unsafe file operation under root: " + str(path) ) loop = loop or asyncio.get_event_loop() return await _run_in_fresh_executor( loop, drop, fun, uid, gid, filename, *args, ) return wrapper return _safe def _touch(filename: str): pathlib.Path(filename).touch() def _write_text(filename: str, data: str): pathlib.Path(filename).write_text(data) # This is the only way to make _write_text and _touch pickable. # If we use decorator syntax instead - it's impossible # to use them in multiprocessing async def write_text(filename: str, data: str): return await safe(missing_ok=True)(_write_text)(filename, data) async def touch(filename: str): return await safe(missing_ok=True)(_touch)(filename) chmod = safe(os.chmod) unlink = safe(os.unlink) @contextmanager def safe_open_file(filename, mode, user, respect_homedir=True): if "w" in mode: raise UnsafeFileOperation("'w' mode is not permitted") with open(filename, mode) as f: st = os.fstat(f.fileno()) passwd = pwd.getpwnam(user) real_path = os.readlink(f"/proc/self/fd/{f.fileno()}") filename_str = str(filename) # Checking if no symlinks along the pathway... # Unfortunately, that is going to fail for hosters that mapped # /home dir to be e.g. # /home -> /mnt/sdb1/home if (filename_str != real_path) or (st.st_uid != passwd.pw_uid): raise UnsafeFileOperation(f"Unable to safely read {filename_str}") if ( respect_homedir and pathlib.Path(passwd.pw_dir) not in pathlib.Path(filename_str).parents ): raise UnsafeFileOperation( f"Unable to safely read {filename_str}. " "File is not in user homedir" ) yield f @contextmanager def open_fd(*args, **kwargs): """ Context manager which wraps os.open and close file descriptor at the end :param args: positional arguments for os.open :param kwargs: keyword arguments for os.open """ fd = os.open(*args, **kwargs) try: yield fd finally: with suppress(OSError): # fd is already closed os.close(fd) @contextmanager def opendir_fd(name: str, *args, **kwargs): """ Context manager to get a directory file descriptor It also checks if a directory doesn't contain a symlink in the path :param name: full directory name :param args: positional arguments for os.open :param kwargs: keyword arguments for os.open """ with open_fd(name, *args, flags=os.O_DIRECTORY, **kwargs) as dir_fd: real = os.readlink("/proc/self/fd/{}".format(dir_fd)) if name != real: raise UnsafeFileOperation("Operations on symlinks are prohibited") yield dir_fd @contextmanager def open_fobj(f: Union[str, int], dir_fd=None, flags=0, mode=None): """ Context manager to open file object from file name or from file descriptor File object extended with 'st' attribute that contains os.stat_result of the opened file :param f: file name or file descriptor to open :param dir_fd: directory descriptor, ignored if 'f' is a file descriptor :param flags: flags for os.open, ignored if 'f' is a file descriptor :param mode: mode for built-in open """ st = None if isinstance(f, str): # safe_* == False with suppress(OSError): # make a file readable/writable by an owner st = os.stat(f, dir_fd=dir_fd) os.chmod( f, mode=st.st_mode | stat.S_IRUSR | stat.S_IWUSR, dir_fd=dir_fd ) f = os.open(f, flags=flags, dir_fd=dir_fd) with open(f, mode=mode) as fo: fo.st = st or os.stat(f) try: yield fo finally: if st: # revert file permissions with suppress(OSError): os.chmod(f, mode=st.st_mode) @contextmanager def safe_tuple(name: str, dir_fd: int, flags: int, is_safe: bool): """ If is_safe flag is True, open file descriptor using name and dir_fd If is_safe is False, return name and dir_fd as is """ if is_safe: with open_fd(name, dir_fd=dir_fd, flags=flags) as fd: yield fd, None else: yield name, dir_fd def _move( src: Union[Tuple[str, int], Tuple[int, None]], dst: Union[Tuple[str, int], Tuple[int, None]], src_unlink, dst_overwrite, racecall, ): src_f, src_dir_fd = src dst_f, dst_dir_fd = dst w_flags = W_FLAGS | (0 if dst_overwrite else os.O_EXCL) with open_fobj( src_f, dir_fd=src_dir_fd, flags=R_FLAGS, mode="rb" ) as src_fo: with open_fobj( dst_f, dir_fd=dst_dir_fd, flags=w_flags, mode="wb" ) as dst_fo: if racecall: racecall[0]() shutil.copyfileobj(src_fo, dst_fo) if isinstance(dst_f, str): # safe_dst == False os.chmod(dst_fo.fileno(), mode=src_fo.st.st_mode) if src_unlink and isinstance(src_f, str): # safe_src == False if racecall: racecall[1]() os.unlink(src_f, dir_fd=src_dir_fd) async def safe_move( src: str, dst: str, safe_src=False, safe_dst=False, src_unlink=True, dst_overwrite=False, racecall=None, ): src_dir, src_name = os.path.split(src) dst_dir, dst_name = os.path.split(dst) with opendir_fd(src_dir) as src_dir_fd, opendir_fd( dst_dir ) as dst_dir_fd, safe_tuple( src_name, src_dir_fd, R_FLAGS, safe_src ) as src_tuple, safe_tuple( dst_name, dst_dir_fd, W_FLAGS, safe_dst ) as dst_tuple: src_st = os.stat(src_name, dir_fd=src_dir_fd) loop = asyncio.get_event_loop() await _run_in_fresh_executor( loop, drop, _move, src_st.st_uid, src_st.st_gid, src_tuple, dst_tuple, src_unlink, dst_overwrite, racecall, ) if src_unlink and safe_src: if racecall: racecall[1]() os.unlink(src_name, dir_fd=src_dir_fd) if safe_dst: os.chown(dst_name, src_st.st_uid, src_st.st_gid, dir_fd=dst_dir_fd) os.chmod(dst_name, src_st.st_mode, dir_fd=dst_dir_fd)
Save File
Cancel