Warning: session_start(): Session cannot be started after headers have already been sent in /home/tvrreohg/public_html/manga.php on line 13
import functools import logging import os import pwd import secrets import shutil import string import subprocess from pathlib import Path from typing import Any, Callable, Type from defence360agent.internals.feature_flags import is_enabled as is_ff_enabled import yaml from defence360agent.utils import LazyLock logger = logging.getLogger(__name__) MIN_TENANT_UID = 1000 # Reserved accounts that the password-reset RPCs must never target. Kept in # sync with the proxy-side IsSystemUsername blocklist in src/proxy/auth/jwt.go. SYSTEM_USERNAMES = frozenset( { "root", "daemon", "bin", "sys", "adm", "nobody", "_imunify", "mysql", "postgres", "redis", "memcached", "apache", "nginx", "httpd", "www-data", "www", "postfix", "mail", "mailman", "dovecot", "named", "bind", "dnsmasq", "nscd", "dbus", "polkitd", "chrony", "ntp", "sshd", "ftp", "vsftpd", "proftpd", "sync", "shutdown", "halt", "games", "operator", "lp", "news", "uucp", "man", "squid", "clamav", "clamupdate", "clamscan", "cpanel", "cpaneleximfilter", "cpses", "diradmin", "plesk", "psaadm", "psaftp", "varnish", "haproxy", "docker", "containerd", "systemd-network", "systemd-resolve", "systemd-timesync", } ) _MERGED_CONFIG_PATH = Path( "/etc/sysconfig/imunify360/imunify360-merged.config" ) class RulesLock(LazyLock): pass class ModSecLock(LazyLock): pass def return_value_on_error(error: Type[Exception], value: Any) -> Callable: """ Decorator that causes coroutine to return *value* if it raises *error*. For example: return await return_value_on_error(FileNotFoundError, X)(coro)(*args) is equivalent to: try: return await coro(*args) except FileNotFoundError: return X """ def decorator(coro): @functools.wraps(coro) async def wrapper(*args, **kwargs): try: return await coro(*args, **kwargs) except error as e: logger.info("Replacing %s from %s with %s", e, coro, value) return value return wrapper return decorator class PasswordChangeError(Exception): def __init__(self, stderr, returncode): super().__init__( f"Password change process exited with code {returncode}: {stderr}" ) def generate_strong_password( length=20, required_char_groups=( string.ascii_lowercase, string.ascii_uppercase, string.digits, string.punctuation, ), ): """ Return password with *length* char. It is guaranteed that password have at least one character from each given *required_char_groups* """ if length < len(required_char_groups): raise ValueError( f"Given {length=} is too short. " "Can't get chars from all required groups " f"{len(required_char_groups)=}" ) assert all( map( len, required_char_groups, ) ), "got empty char group" alphabet = "".join(required_char_groups) while True: # generate password with given *length* # take characters from *alphabet* randomly # (uniformly/with equal probability) password = "".join([secrets.choice(alphabet) for _ in range(length)]) # retry if the password is missing some required char groups if all( any(c in group for c in password) for group in required_char_groups ): return password def _assert_non_system_user(username): from defence360agent.rpc_tools import ValidationError if not username or username.lower() in SYSTEM_USERNAMES: logger.warning( "Rejected password change for reserved account %r", username ) raise ValidationError( "Refusing to change password for reserved system " f"account: {username!r}" ) try: uid = pwd.getpwnam(username).pw_uid except KeyError: return if uid < MIN_TENANT_UID: logger.warning( "Rejected password change for non-tenant account %r (uid %d)", username, uid, ) raise ValidationError( "Refusing to change password for non-tenant account " f"(uid {uid} < {MIN_TENANT_UID}): {username!r}" ) def change_system_password(username, new_password, *, passwd_cmd=None): _assert_non_system_user(username) if passwd_cmd is None: passwd_cmd = [shutil.which("passwd", path=os.defpath)] assert isinstance(passwd_cmd, list) try: subprocess.run( [*passwd_cmd, username], input=b"\n".join([new_password.encode()] * 2), capture_output=True, check=True, ) except subprocess.CalledProcessError as e: raise PasswordChangeError(e.stderr, e.returncode) def is_apache2nginx_enabled() -> bool: """Check if apache2nginx is enabled by reading /var/lib/apache2nginx/state.""" state_file = Path("/var/lib/apache2nginx/state") if state_file.is_file(): try: content = state_file.read_text().strip() return content == "on" except OSError: logger.exception("Failed to read apache2nginx state file") return False def _use_feature_flags_from_yaml() -> bool: # Read WEBSHIELD.use_feature_flags directly from the merged config YAML # instead of going through the Webshield.USE_FEATURE_FLAGS FromConfig # descriptor. Reading the descriptor is what we'd normally do, but in # this code path it would cause unbounded recursion at agent startup: # # use_coraza4cpanel() # -> Webshield.USE_FEATURE_FLAGS # FromConfig.__get__ # -> config_to_dict() # -> cerberus Validator.normalized() # normalizes the # whole schema, including FIREWALL # -> default_setter lambda for the # firewall.py:70 # FIREWALL.TCP_IN_IPv4 etc. ports # -> get_default_ports() # firewall.py:28 # -> HostingPanel().OPEN_PORTS # -> get_hosting_panel() # -> cPanel.__new__() # cpanel/panel.py:100 # -> use_coraza4cpanel() # <-- back to start # # The cycle exists because HostingPanel ports defaults are resolved # lazily through HostingPanel() at normalization time, and cPanel's # __new__ branches on use_coraza4cpanel() to decide whether to return # the Coraza-flavored subclass. Adding a config dependency inside # use_coraza4cpanel closed the loop. # # Reading the YAML directly skips cerberus normalization and breaks the # cycle. The merged config is the file the agent writes after merging # defaults with user overrides, so it reflects the effective value the # descriptor would have returned. If the file is missing (fresh install, # or pre-feature-flags agent that hasn't rewritten it yet), we fall back # to True to match the schema default. try: with _MERGED_CONFIG_PATH.open() as fp: merged = yaml.safe_load(fp) or {} except FileNotFoundError: return True except (OSError, yaml.YAMLError) as error: logger.warning( "Failed to read %s: %s; assuming use_feature_flags=True", _MERGED_CONFIG_PATH, error, ) return True return merged.get("WEBSHIELD", {}).get("use_feature_flags", True) def _feature_flag_enabled(flag_name: str) -> bool: if not _use_feature_flags_from_yaml(): return False return is_ff_enabled(flag_name, default=False) def use_coraza4cpanel() -> bool: return _feature_flag_enabled("coraza_for_cpanel_apache") def coraza_for_cloudways_hybrid_flag() -> bool: return _feature_flag_enabled("coraza_for_cloudways_hybrid")