# -*- coding: utf-8 -*- # Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2022 All Rights Reserved # # Licensed under CLOUD LINUX LICENSE AGREEMENT # http://cloudlinux.com/docs/LICENSE.TXT # import glob import importlib.util import logging import os import stat import sys from .helpers import LISTENERS_DIRECTORY logger = logging.getLogger(__name__) from .modify_domain_hook import ModifyDomainHook from .modify_admin_hook import ModifyAdminHook from .modify_user_hook import ModifyUserHook from .modify_package_hook import ModifyPackageHook def _is_root_owned_unwritable(filepath): """ Return True only if `filepath` is a regular file owned by root (uid 0) and not group- or world-writable. Used to gate a listener before its top-level code is exec_module()'d as root: a *.py (or a symlink whose real target) a non-root principal can write is refused rather than executed. os.stat follows symlinks, so it stats the real inode whose code would run. """ try: st = os.stat(filepath) except OSError: return False return (stat.S_ISREG(st.st_mode) and st.st_uid == 0 and not (st.st_mode & (stat.S_IWGRP | stat.S_IWOTH))) def import_file(filepath): # Fast path: see if the module has already been imported. name = os.path.basename(filepath)[:-3] if name in sys.modules: return sys.modules[name] # Defense-in-depth: exec_module() below runs the listener's top-level code # in this root process. Refuse a listener that is not root-owned or is # group/world-writable, so an attacker-influenceable file is not executed. if not _is_root_owned_unwritable(filepath): logger.warning('refusing to load listener %s: not root-owned or ' 'group/world-writable', filepath) return None # If any of the following calls raises an exception, # there's a problem we can't handle -- let the caller handle it. module_spec = importlib.util.spec_from_file_location(name, filepath) if module_spec is None: raise ImportError(f"Can't find module {name} at {filepath}") # Load the module, potentially raising ImportError if the module cannot be loaded module = importlib.util.module_from_spec(module_spec) module_spec.loader.exec_module(module) # Add the module to sys.modules sys.modules[name] = module return module # automatically scan directory for universal hooks for hook_py in glob.glob(os.path.join(LISTENERS_DIRECTORY, '*.py')): import_file(hook_py) __all__ = ( 'ModifyDomainHook', 'ModifyAdminHook', 'ModifyUserHook', 'ModifyPackageHook', )