MAINHACK

Viewing / Editing File: clselectctlphp.py

File is not writable. Editing disabled.
#!/opt/cloudlinux/venv/bin/python3 -bb
# -*- coding: utf-8 -*-

# Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2019 All Rights Reserved
#
# Licensed under CLOUD LINUX LICENSE AGREEMENT
# http://cloudlinux.com/docs/LICENSE.TXT

from __future__ import print_function
from __future__ import absolute_import
from __future__ import division

import re
import subprocess
import sys
import getopt
import os
import clcommon
import traceback
import simplejson as json
import shutil
import pwd

from builtins import map
from .clselect import ClSelect
from .clextselect import ClExtSelect, depend_modules_dict
from .cluserselect import ClUserSelect
from .cluserextselect import ClUserExtSelect
from .cluseroptselect import ClUserOptSelect
from .clselectprint import clprint
from clcommon import clcagefs, ClPwd
from clcommon.const import Feature
from clcommon.cpapi import is_panel_feature_supported
from clcommon.sysctl import SysCtlConf, SYSCTL_CL_CONF_FILE
from clcommon.utils import is_ubuntu, run_command, ExternalProgramFailed
from .utils import make_symlink, safely_resolve_username_and_doc_root
from .clselectexcept import ClSelectExcept, BaseClSelectException
from .models import Website


# This is the oldest version with all bad things
API_0 = 0

# This version replaces hardcoded interpreter name in utility output with more
# generic key names that will be correctly understood by cloudlinux-selector
API_1 = 1

# Path to cagefs command
CAGEFSCTL_COMMAND = "/usr/sbin/cagefsctl"


def usage():
    print(" -v | --version                   : Specify alternative version")
    print(" -u | --user                      : Username")
    print("      --domain                    : Domain name (for per-domain configuration)")
    print(" -l | --list                      : List alternatives for interpreter")
    print(" -G | --list-extensions           : List global set of extensions for alternative")
    print(" -S | --summary                   : List summary of alternatives")
    print(" -s | --user-summary              : List user summary of alternatives")
    print(" -C | --current                   : Show currently selected alternative")
    print(" -c | --user-current              : Show currently selected alternative for a user")
    print(" -B | --set-current               : Set alternative as global default")
    print(" -b | --set-user-current          : Set alternative as user default")
    print(" -Y | --enable-alternative        : Enable alternative globally")
    print(" -N | --disable-alternative       : Disable alternative globally")
    print(" -E | --enable-extensions         : Enable comma-separated list of extensions globally for a version")
    print(" -D | --disable-extensions        : Disable comma-separated list of extensions globally for a version")
    print(" -R | --replace-extensions        : Replace extensions with comma-separated list of extensions ")
    print("                                    for a version globally")
    print(" -e | --enable-user-extensions    : Enable comma-separated list of extensions for a user")
    print(" -d | --disable-user-extensions   : Disable comma-separated list of extensions for a user")
    print(" -r | --replace-user-extensions   : Replace user extensions with comma-separated list of extensions")
    print(" -t | --reset-user-extensions     : Replace user extensions with version default extensions")
    print(" -g | --list-user-extensions      : List enabled extensions for a user. With key --all shows all extensions")
    print(" -a | --all                       : Show all extensions")
    print(" -p | --print-summary             : If specified along with setting an alternative prints user summary")
    print(" -V | --show-native-version       : Shows native version while showing summary or selected version")
    print(" -L | --list-users                : List users who use a specified alternative")
    print(" -T | --change-to-version         : Changes to a specified version all users who have a certain version")
    print(" -k | --add-options               : Add comma-separated list options for a user")
    print(" -m | --replace-options           : Replace user options with comma-separated list of options")
    print(" -x | --delete-options            : Delete comma-separated list options for a user")
    print(" -Q | --base64                    : Expects data as comma-separated base64-encoded string")
    print(" -q | --quiet                     : Suppress errors messages for wrong input")
    print(" -P | --print-options             : Prints user options. By default prints as plain text")
    print(" --print-options-safe             : Prints user options. By default prints as plain text (safe strings)")
    print(" --apply-symlinks-rules           : Recreate symlinks to php extensions for all users ")
    print("                                    based on /etc/cl.selector/symlinks.rules file")
    print(" --exclude-pid-list               : Specify list of PIDs of processes that should not be signaled by SIGHUP")
    print(" -j | --json                      : Print data as JSON")
    print(" -w | --csv                       : Print data as CSV")
    print(" -W | --perl                      : Print data as perl structure")
    print(" --api-version                    : Integer, representing specific api version to use. ")
    print("                                    Defaults to {}".format(API_0))
    print(" -z | --reset-options             : Deletes all user custom options. Range can be narrowed with user ")
    print("                                    or version options")
    print(" --update-backup                  : write settings to backup")
    print(" --apply-global-php-ini           : use with 0, 1 or 2 arguments from the list: error_log, date.timezone")
    print("                                    without arguments applies all global php options including two above")
    print(" --setup-without-cagefs           : setup PHP Selector without CageFS")
    print(" --revert-to-cagefs               : revert to default setup of PHP Selector (with CageFS)")
    print(" --for-all-users                  : applies specified action for all users in CageFS. ")
    print("                                    Available only for enable/disable user extensions")


def print_error_and_exit(message, prefix=None):
    """
    Prints to stderr
    @param message: string
    """
    fmt = "%s\n"
    if prefix:
        fmt = "%s:%s\n" % (prefix, "%s")
    sys.stderr.write(fmt % message)
    sys.exit(1)


# POSIX-ish username allow-list. Rejects shell metacharacters so any caller
# concatenating this string into a command line cannot inject (F-27), and
# prevents stored garbage in USER_CONF from re-detonating later.
_USERNAME_RE = re.compile(r"\A[a-z_][a-z0-9_-]{0,31}\Z")


def _validate_cpanel_username(user):
    """Reject usernames that are not safe to embed in argv or persist."""
    if not isinstance(user, str) or not _USERNAME_RE.match(user):
        print_error_and_exit("Error: invalid cPanel username")


def check_args_presence():
    """
    Checks presence of command line arguments
    and exits with usage info if missing
    """
    if len(sys.argv) == 1:
        print_error_and_exit("Command line arguments expected. For help use '-h' or '--help' options")


def get_name_modifier(version):
    """ """
    BASE_ALT_DIR = "/opt/alt/php"
    ver = version.replace(".", "")
    name_modifier_file = BASE_ALT_DIR + ver + "/name_modifier"
    name_modifier = ""
    if os.path.isfile(name_modifier_file):
        try:
            name_modifier = open(name_modifier_file, "r").readline().strip()
        except (OSError, IOError):
            return ""
    return name_modifier


def letter_to_status(letter):
    if letter == "-":
        return "disabled"
    elif letter == "e":
        return "enabled"


def format_summary(data, format="text", api_version=API_0):
    if api_version == API_0:
        available_versions_key = "PHPConfiguration"
        default_version_key = "defaultPHPversion"
    else:
        available_versions_key = "available_versions"
        default_version_key = "default_version"

    states = ["e", "d", "s"]
    text_lines = []
    json_dict = {
        available_versions_key: [],
    }
    for alt in data:
        row_length = len(alt[1])
        fmt = " ".join(["%s"] * (row_length + 1))
        row_data = [alt[0]]
        row_data.extend(list(map((lambda i: ((alt[1][i] and states[i]) or "-")), range(row_length))))  # pylint: disable=range-builtin-not-iterating
        name_modifier = get_name_modifier(alt[0])
        if name_modifier != "":
            fmt = " ".join(["%s"] * (row_length + 2))
            row_data.append(name_modifier)
        if format == "text":
            text_lines.append(fmt % tuple(row_data))
        if "d" in row_data:
            json_dict[default_version_key] = row_data[0]
        json_dict[available_versions_key].append(
            {
                "version": row_data[0],
                "status": letter_to_status(row_data[1]),
                "name_modifier": name_modifier,
            }
        )
    if format == "json":
        return json.dumps(json_dict)
    elif format == "text":
        return "\n".join(text_lines)


def print_summary(data, format="text", api_version=API_0):
    """
    Prints alternatives summary
    """
    data = format_summary(data, format, api_version)
    print(data)


def check_params(config, param_list):
    """
    Check that config has param_list and this params not None
    """
    for param in param_list:
        if param not in config or config[param] == None:
            print_error_and_exit("Error: %s must be specified" % param)


def ext_letter_to_status(letter):
    if letter == "~":
        return "build-in"
    elif letter == "+":
        return "enabled"
    else:
        return "disabled"


def fill_descriptions(tmp_list):
    descr_file = "/etc/cl.selector.conf.d/phpextdesc.txt" if clcagefs.in_cagefs() else "/etc/cl.selector/phpextdesc.txt"
    with open(descr_file) as f:
        desct_content = f.readlines()
    full_desct_dict = {}
    for line in desct_content:
        line_parts = line.split("=")
        full_desct_dict[line_parts[0]] = line_parts[1].strip()
    for item in tmp_list:
        try:
            item["description"] = full_desct_dict[item["name"]]
        except KeyError:  # skip extention without description
            pass
    return tmp_list


def print_json_status_ok():
    result_dict = {"status": "ok"}
    print(json.dumps(result_dict))


def get_cpanel_user():
    """
    Return user (name of cpanel account) for PHP Selector without CageFS feature
    """
    if os.path.isfile(ClSelect.USER_CONF):
        with open(ClSelect.USER_CONF, "r") as f:
            return f.read().strip()
    for user in ClPwd().get_user_dict():
        if os.path.exists("/var/cpanel/users/" + user):
            return user
    return None


def set_cpanel_user(user):
    """
    Set user (name of cpanel account) for PHP Selector without CageFS feature
    """
    # F-27: validate at the persistence boundary so attacker-controlled bytes
    # cannot land in USER_CONF and re-detonate when read back by revert.
    _validate_cpanel_username(user)
    with open(ClSelect.USER_CONF, "w") as f:
        f.write(user)
    os.chmod(ClSelect.USER_CONF, 0o644)


def switch_linksafe(enable=False):
    if not is_panel_feature_supported(Feature.LVE):
        return
    new_conf = "/etc/sysctl.d/cloudlinux-linksafe.conf"
    if os.path.isfile(new_conf):
        conf = new_conf
    else:
        conf = SYSCTL_CL_CONF_FILE
    sysctl_cfg = SysCtlConf(config_file=conf)
    sysctl_cfg.set("fs.protected_symlinks_create", str(int(enable)))
    sysctl_cfg.set("fs.protected_hardlinks_create", str(int(enable)))
    if subprocess.call("sysctl --system &>/dev/null", shell=True, executable="/bin/bash") != 0:
        print("Error while executing: sysctl --system")
        sys.exit(1)


def modify_search_path_in_bashrc(user, homedir, obj=None, add=True):
    """
    Add path to PATH variable in ~/.bashrc
    :param user: name of user
    :type user: string
    :param homedir: path to home directory
    :type homedir: string
    :param obj: instance of ClUserSelect class
    :type obj: instance of ClUserSelect class
    :param add: add path to .bashrc when True, remove otherwise
    :type obj: bool
    """
    if obj is None:
        obj = ClUserSelect("php")
    cur_user = obj._change_uid(user)
    # On Ubuntu we are using ~/.profile
    # On CL - ~/.bashrc
    if is_ubuntu():
        bashrc = homedir + "/.profile"
    else:
        bashrc = homedir + "/.bashrc"
    line = "PATH=$HOME/" + ClUserSelect.SELECTOR2_DIR + ":$HOME/.cl.selector:$PATH"
    with open(bashrc, "r") as f:
        found = line + "\n" in f
    if add:
        if not found:
            with open(bashrc, "a") as f:
                f.write("\n" + line + "\n")
    else:
        if found:
            clcommon.utils.delete_line_from_file(bashrc, line)
    obj._restore_uid(cur_user)


def restore_settings_from_backup(user, homedir, uid, alt_dirs, obj=None):
    """
    Restore (apply) settings for PHP Selector from backup
    :param user: name of user
    :type user: string
    :param homedir: path to user's home directory
    :type homedir: string
    :param uid: user's uid
    :type uid: int
    :param alt_dirs: list of alt-php directories like ['php51', 'php52']
    :type alt_dirs: list
    :param obj: instance of ClUserSelect class
    :type obj: ClUserSelect object
    """

    def cleanup():
        # Delete unneeded files
        shutil.rmtree(os.path.join("/var/cagefs", str(uid)[-2:]), True)
        shutil.rmtree("/usr/share/cagefs/etc", True)
        shutil.rmtree("/usr/share/cagefs/etc.new", True)

    if obj is None:
        obj = ClUserSelect("php")
    base_dest_path = homedir + "/.cl.selector"
    cleanup()
    # Generate alt_php.ini for all versions using cagefsctl
    # F-27: argv list, no shell — `user` was attacker-tainted via args[0].
    if (
        subprocess.call(["/usr/sbin/cagefsctl", "--silent", "--force-update-etc", user])
        != 0
    ):
        sys.exit(1)
    cur_user = obj._change_uid(user)
    # Copy generated alt_php.ini files to new location in user's home directory
    base_src_path = os.path.join("/var/cagefs", str(uid)[-2:], user, "etc", "cl.php.d")
    for alt_dir in alt_dirs:
        src_path = base_src_path + "/alt-" + alt_dir + "/alt_php.ini"
        dest_path = base_dest_path + "/alt_" + alt_dir + ".ini"
        shutil.copy(src_path, dest_path)
    # Select php version from backup (or default when backup does not exist)
    obj.set_version_from_backup(user)
    shutil.rmtree(homedir + "/.cagefs", True)
    obj._restore_uid(cur_user)
    cleanup()


def disable_cagefs_service():
    if os.path.isfile("/usr/bin/systemctl"):
        subprocess.run("/usr/bin/systemctl disable cagefs", shell=True, executable="/bin/bash")
        subprocess.run("/usr/bin/systemctl stop cagefs", shell=True, executable="/bin/bash")
        subprocess.run("/usr/bin/systemctl mask cagefs", shell=True, executable="/bin/bash")
    else:
        subprocess.run("/sbin/service cagefs stop &> /dev/null", shell=True, executable="/bin/bash")
        subprocess.run("/sbin/chkconfig cagefs off", shell=True, executable="/bin/bash")


def enable_cagefs_service():
    if os.path.isfile("/usr/bin/systemctl"):
        subprocess.run("/usr/bin/systemctl unmask cagefs", shell=True, executable="/bin/bash")
        subprocess.run("/usr/bin/systemctl enable cagefs", shell=True, executable="/bin/bash")
        subprocess.run("/usr/bin/systemctl start cagefs", shell=True, executable="/bin/bash")
    else:
        subprocess.run("/sbin/chkconfig cagefs on", shell=True, executable="/bin/bash")
        subprocess.run("/sbin/service cagefs start &> /dev/null", shell=True, executable="/bin/bash")


def setup_without_cagefs(args):
    """
    Setup PHP Selector without CageFS
    """
    sys.path.append("/usr/share/cagefs")
    try:
        import cagefslib
    except ImportError:
        print("Error: CageFS is not installed")
        sys.exit(1)
    # alt-php versions are installed ?
    alt_dirs = cagefslib.get_alt_dirs()
    if not alt_dirs:
        print("alt-php not found")
        sys.exit(1)
    # detect cpanel user
    if args:
        user = args[0]
        set_cpanel_user(user)
    else:
        user = get_cpanel_user()
        if not user:
            print("Error: failed to detect cpanel account. Please specify name of an account as argument:")
            print("selectorctl --setup-without-cagefs USER")
            sys.exit(1)
        if not os.path.exists(ClSelect.USER_CONF):
            set_cpanel_user(user)
    # F-27: re-validate after either branch (USER_CONF may pre-exist).
    _validate_cpanel_username(user)
    # disable linksafe protection
    switch_linksafe()
    pw = pwd.getpwnam(user)
    homedir = pw.pw_dir
    # create symlinks to user's alt_php.ini files
    for alt_dir in alt_dirs:
        alt_path = "/opt/alt/" + alt_dir + "/link/conf/alt_php.ini"
        user_path = homedir + "/.cl.selector/alt_" + alt_dir + ".ini"
        make_symlink(user_path, alt_path)
    obj = ClUserSelect("php")
    restore_settings_from_backup(user, homedir, pw.pw_uid, alt_dirs, obj)
    obj.create_selector_symlinks(user)
    modify_search_path_in_bashrc(user, homedir, obj)
    disable_cagefs_service()
    # kill user's processes in LVE
    subprocess.run(
        "/usr/sbin/lvectl destroy "
        + str(pw.pw_uid)
        + " &>/dev/null; /usr/sbin/lvectl apply "
        + str(pw.pw_uid)
        + " &>/dev/null",
        shell=True,
        executable="/bin/bash",
    )


def revert_to_cagefs():
    """
    Revert to default PHP Selector setup with CageFS
    """
    if not os.path.exists(ClSelect.USER_CONF):
        print('PHP Selector is in default mode already ("with CageFS" mode)')
        sys.exit(1)
    sys.path.append("/usr/share/cagefs")
    try:
        import cagefslib
    except ImportError:
        print("Error: CageFS is not installed")
        sys.exit(1)
    # alt-php versions are installed ?
    alt_dirs = cagefslib.get_alt_dirs()
    if not alt_dirs:
        print("alt-php not found")
        sys.exit(1)
    switch_linksafe(enable=True)
    # delete symlinks to user's alt_php.ini files
    for alt_dir in alt_dirs:
        alt_path = "/opt/alt/" + alt_dir + "/link/conf/alt_php.ini"
        if os.path.islink(alt_path):
            os.unlink(alt_path)
    user = get_cpanel_user()
    if not user:
        print("Error: failed to detect user")
        sys.exit(1)
    # F-27: reject malformed USER_CONF content before any downstream use.
    _validate_cpanel_username(user)
    pw = pwd.getpwnam(user)
    homedir = pw.pw_dir
    obj = ClUserSelect("php")
    modify_search_path_in_bashrc(user, homedir, obj, add=False)
    # remove config file (file-switch) for "without CageFS" mode
    os.unlink(ClSelect.USER_CONF)
    # Generate alt_php.ini for all versions using cagefsctl
    # F-27: argv list, no shell — `user` is now validated above.
    subprocess.run(["/usr/sbin/cagefsctl", "--silent", "--force-update-etc", user])
    enable_cagefs_service()
    # kill user's processes in LVE
    subprocess.run(
        "/usr/sbin/lvectl destroy "
        + str(pw.pw_uid)
        + " &>/dev/null; /usr/sbin/lvectl apply "
        + str(pw.pw_uid)
        + " &>/dev/null",
        shell=True,
        executable="/bin/bash",
    )


def apply_global_php_ini(args):
    """
    Apply "global" php.ini settings to all alt-php versions
    :param args: list of command line parameters (names of php.ini options)
    :type args: list
    """
    sys.path.append("/usr/share/cagefs")
    try:
        import cagefslib
        import cagefsreconfigure
    except ImportError:
        print("Error: CageFS is not installed")
        sys.exit(1)
    # alt-php versions are installed ?
    if cagefslib.get_alt_versions():
        cagefsreconfigure.replace_alt_settings(options=args)


def _check_depencies_and_print_message(print_format, print_message):
    if len(depend_modules_dict):
        # Warning - blocked modules present
        modules_list = list()
        for module_name, dep_module in depend_modules_dict.items():
            modules_list.append(" '%s' is required for '%s'" % (module_name, dep_module))
        clprint.print_diag(print_format, {"status": "WARN", "message": print_message + ",".join(modules_list)})


def get_extensions(interpreter, version, fmt="text"):
    ext_list = ClExtSelect(interpreter).list_extensions(version)
    return parse_extensions(ext_list, version, fmt)


def parse_extensions(ext_list, version, fmt):
    json_list = []
    for ext in ext_list:
        action = "~"
        if ext[1] is True:
            action = "+"
        elif ext[1] is False:
            action = "-"
        if fmt == "text":
            json_list.append((action, ext[0]))
        else:
            json_list.append({"name": ext[0], "description": "", "state": ext_letter_to_status(action)})
            json_list = fill_descriptions(json_list)
    if fmt == "json":
        result_dict = {"version": version, "extensions": json_list}
        return result_dict
    return json_list


def get_cagefs_users():
    """
    Return list of users that are in CageFS
    If CageFS is not installed or initialized
    throws exception and prints it
    :return:
    """
    not_installed_msg = "No such file or directory"
    not_initialized_msg = "CageFS is not initialized"
    try:
        users = run_command([CAGEFSCTL_COMMAND, "--list-enabled"]).strip()
        if users == "":
            return []
        return users.split("\n")[1:]  # First element shows number of users
    except ExternalProgramFailed as e:
        if not_installed_msg in str(e):
            print_error_and_exit("ERROR: CageFS not installed.")
        elif not_initialized_msg in str(e):
            print_error_and_exit(
                'Error: CageFS is not initialized. Use "/usr/sbin/cagefsctl --init" to initialize CageFS'
            )
        print_error_and_exit(e)


def verify_options_in_cagefs(options, supported_options):
    """
    Verifies that the supplied options are admissible inside CageFS and
    rebinds --user/-u to the caller's identity.

    F-28: previously this used set-intersection — any single allowed option
    was enough to pass even when root-only dispatches (--setup-without-cagefs,
    --revert-to-cagefs, --apply-global-php-ini) were smuggled alongside. We
    now reject those dispatches explicitly when invoked from inside CageFS,
    so option smuggling cannot reach the privileged code paths.

    F-14: previously this only auto-appended --user when omitted, accepting
    any caller-supplied value verbatim. We now reject any --user/-u whose
    value does not match the caller's effective UID, closing the IDOR
    independently of CageFS /etc/passwd masking.
    """
    # F-28: dispatches that mutate global/admin state must never be reachable
    # from inside a cage, irrespective of which "allowed" option they were
    # smuggled with.
    root_only_in_cagefs = {
        "--setup-without-cagefs",
        "--revert-to-cagefs",
        "--apply-global-php-ini",
    }
    input_options = {opt for opt, _ in options}
    if input_options & root_only_in_cagefs:
        print_error_and_exit("Error: This command is not supported in CageFS")

    if not input_options & set(supported_options):
        print_error_and_exit("Error: This command is not supported in CageFS")

    # F-14: rebind --user/-u to caller identity. Any caller-supplied value
    # whose resolved username differs from the caller's euid name is refused.
    current_user_name = pwd.getpwuid(os.geteuid()).pw_name
    user_opts = {"--user", "-u"}
    for opt, value in options:
        if opt in user_opts and value != current_user_name:
            print_error_and_exit("Error: --user must match the current user in CageFS")
    if not (input_options & user_opts):
        options.append(("--user", current_user_name))


def _get_website_from_domain(user, domain):
    """
    Resolve domain to docroot and create a Website object.
    Returns None if domain is not provided.
    """
    if not domain:
        return None

    try:
        _, doc_root = safely_resolve_username_and_doc_root(user, domain)
        return Website(domain=domain, docroot=doc_root)
    except (ClSelectExcept, BaseClSelectException):
        # If domain resolution fails, return None (will use per-user config)
        return None


def main():
    config = {}
    config["interpreter"] = "php"
    config["version"] = None
    config["show-all"] = False
    config["print-summary"] = False
    config["show-native-version"] = False
    config["decoder"] = "plain"
    config["quiet"] = False
    config["format"] = "text"
    config["api-version"] = API_0
    config["domain"] = None

    actions = {}
    exclude_pid_list = []

    check_args_presence()

    try:
        opts, args = getopt.getopt(
            sys.argv[1:],
            "hi:lSsCcB:Y:N:E:D:R:v:Gu:b:ge:d:r:atpVLT:k:m:x:QqPjwWz",
            [
                "help",
                "setup-without-cagefs",
                "revert-to-cagefs",
                "interpreter=",
                "list",
                "summary",
                "user-summary",
                "current",
                "user-current",
                "set-current=",
                "enable-alternative=",
                "disable-alternative=",
                "enable-extensions=",
                "disable-extensions=",
                "replace-extensions=",
                "version=",
                "list-extensions",
                "user=",
                "domain=",
                "set-user-current=",
                "list-user-extensions",
                "enable-user-extensions=",
                "disable-user-extensions=",
                "replace-user-extensions=",
                "all",
                "reset-user-extensions",
                "print-summary",
                "show-native-version",
                "list-users",
                "change-to-version=",
                "add-options=",
                "replace-options=",
                "delete-options=",
                "base64",
                "apply-symlinks-rules",
                "quiet",
                "print-options",
                "print-options-safe",
                "json",
                "csv",
                "perl",
                "api-version=",
                "reset-options",
                "update-backup",
                "apply-global-php-ini",
                "exclude-pid-list=",
                "for-all-users",
            ],
        )
    except getopt.GetoptError:
        usage()
        sys.exit(1)

    opts_supported_in_cagefs = [
        "-l",
        "--list",
        "-s",
        "--user-summary",
        "-c",
        "--user-current",
        "-b",
        "--set-user-current",
        "-e",
        "--enable-user-extensions",
        "-d",
        "--disable-user-extensions",
        "-r",
        "--replace-user-extensions",
        "-t",
        "--reset-user-extensions",
        "-g",
        "--list-user-extensions",
        "-k",
        "--add-options",
        "-m",
        "--replace-options",
        "-x",
        "--delete-options",
        "-P",
        "--print-options",
        "--print-options-safe",
        "-z",
        "--reset-options",
    ]

    if clcagefs.in_cagefs():
        verify_options_in_cagefs(opts, opts_supported_in_cagefs)

    for o, a in opts:
        if o in ["-h", "--help"]:
            usage()
            sys.exit(0)
        elif o in ["--exclude-pid-list"]:
            for pid in a.split(","):
                try:
                    exclude_pid_list.append(int(pid))
                except ValueError:
                    continue
        elif o in ["--apply-symlinks-rules"]:
            actions["apply-symlinks-rules"] = True
        elif o in ["--setup-without-cagefs"]:
            setup_without_cagefs(args)
            sys.exit(0)
        elif o in ["--revert-to-cagefs"]:
            revert_to_cagefs()
            sys.exit(0)
        elif o in ("--apply-global-php-ini",):
            apply_global_php_ini(args)
            sys.exit(0)
        elif o in ["-l", "--list"]:
            actions["list-alternatives"] = True
        elif o in ["-S", "--summary"]:
            actions["show-summary"] = True
        elif o in ["-s", "--user-summary"]:
            actions["show-user-summary"] = True
        elif o in ["-C", "--current"]:
            actions["show-current"] = True
        elif o in ["-c", "--user-current"]:
            actions["show-user-current"] = True
        elif o in ["-a", "--all"]:
            config["show-all"] = True
        elif o in ["-v", "--version"]:
            config["version"] = a
        elif o in ["-d", "--domain"]:
            config["domain"] = a
        elif o in ["-u", "--user"]:
            config["user"] = a
            clpwd = ClPwd()
            if ClSelect.work_without_cagefs():
                uid = clpwd.get_uid(a)
            else:
                users = a.split(",")
                user_list = list()
                try:
                    if len(users) == 1:
                        uid = clpwd.get_uid(a)
                        if os.geteuid() == 0:
                            for user in clpwd.get_names(uid):
                                ClUserSelect().cagefs_copy_etc(user)
                    else:
                        for user in users:
                            if user not in user_list:
                                user_list += clpwd.get_names(clpwd.get_uid(user))
                        config["user"] = ",".join(user_list)
                except ClPwd.NoSuchUserException as e:
                    sys.stderr.write(str(e) + "\n")
                    sys.exit(1)
        elif o in ["--domain"]:
            config["domain"] = a
        elif o in ["-B", "--set-current"]:
            actions["set-current"] = a
        elif o in ["-b", "--set-user-current"]:
            actions["set-user-current"] = a
        elif o in ["-Y", "--enable-alternative"]:
            actions["enable-alternative"] = a
        elif o in ["-N", "--disable-alternative"]:
            actions["disable-alternative"] = a
        elif o in ["-G", "--list-extensions"]:
            actions["list-extensions"] = True
        elif o in ["-g", "--list-user-extensions"]:
            actions["list-user-extensions"] = True
        elif o in ["-E", "--enable-extensions"]:
            actions["enable-extensions"] = a
        elif o in ["-D", "--disable-extensions"]:
            actions["disable-extensions"] = a
        elif o in ["-R", "--replace-extensions"]:
            actions["replace-extensions"] = a
        elif o in ["-e", "--enable-user-extensions"]:
            actions["enable-user-extensions"] = a
        elif o in ["-d", "--disable-user-extensions"]:
            actions["disable-user-extensions"] = a
        elif o in ["-r", "--replace-user-extensions"]:
            actions["replace-user-extensions"] = a
        elif o in ["-t", "--reset-user-extensions"]:
            actions["reset-user-extensions"] = True
        elif o in ["-p", "--print-summary"]:
            config["print-summary"] = True
        elif o in ["-V", "--show-native-version"]:
            config["show-native-version"] = True
        elif o in ["-L", "--list-users"]:
            actions["list-users"] = True
        elif o in ["-T", "--change-to-version"]:
            actions["change-to-version"] = a
        elif o in ["-k", "--add-options"]:
            actions["add-options"] = a
        elif o in ["-m", "--replace-options"]:
            actions["replace-options"] = a
        elif o in ["-x", "--delete-options"]:
            actions["delete-options"] = a
        elif o in ["-Q", "--base64"]:
            config["decoder"] = "base64"
        elif o in ["-q", "--quiet"]:
            config["quiet"] = True
        elif o in ["-P", "--print-options"]:
            actions["print-options"] = True
        elif o in ["--print-options-safe"]:
            actions["print-options-safe"] = True
        elif o in ["-j", "--json"]:
            config["format"] = "json"
        elif o in ["-w", "--csv"]:
            config["format"] = "csv"
        elif o in ["--api-version"]:
            config["api-version"] = int(a)
        elif o in ["-W", "--perl"]:
            config["format"] = "perl"
        elif o in ["-z", "--reset-options"]:
            actions["reset-options"] = True
        elif o in ["--update-backup"]:
            actions["update-backup"] = True
        elif o in ["--for-all-users"]:
            if "user" in config:
                print_error_and_exit(
                    "--for-all-users and --user options are mutually"
                    " exclusive options and cannot be used simultaneously."
                    "\nUse --for-all-user OR --user instead."
                )
            users = get_cagefs_users()
            if not users:
                print_error_and_exit("No changes were made: there are no users with cagefs enabled ")
            clpwd = ClPwd()
            user_list = list()
            for user in users:
                if user not in user_list:
                    user_list += clpwd.get_names(clpwd.get_uid(user))
            config["user"] = ",".join(user_list)
    if len(actions) != 1:
        if len(actions) == 0 and config["show-native-version"]:
            try:
                print(ClSelect(config["interpreter"]).get_native_version()[0])
            except TypeError:
                pass
        else:
            print_error_and_exit("Wrong set of options", "ERROR")

    try:
        # check if we are able to do anything before actually parsing options

        # this two exceptions check for native version inside
        # this is done because selectorctl is called in alt-php spec in cycle
        # and we should avoid printing lot of messages there
        if "set-user-current" not in actions and "show-user-current" not in actions:
            ClSelect().check_requirements()

        if "list-alternatives" in actions:
            if config["format"] != "json":
                for alt in ClSelect(config["interpreter"]).list_alternatives():
                    print("%s\t%s\t%s" % (alt))
            else:
                alternatives_dict = {"status": "ok", "data": []}
                for alt in ClSelect(config["interpreter"]).list_alternatives():
                    alternatives_dict["data"].append({"short": alt[0], "full": alt[1], "path": alt[2]})
                print(json.dumps(alternatives_dict))
        elif "show-summary" in actions:
            data = ClSelect(config["interpreter"]).get_summary(config["show-native-version"])
            print_summary(data, config["format"], config["api-version"])
        elif "show-current" in actions:
            print("%s\t%s\t%s" % ClSelect(config["interpreter"]).get_version(config["show-native-version"]))
        elif "set-current" in actions:
            ClSelect(config["interpreter"]).set_version(actions["set-current"])
            if config["format"] == "json":
                print_json_status_ok()
        elif "enable-alternative" in actions:
            if "user" in config and config["user"]:
                raise ClSelectExcept.VersionModificationBlocked(
                    "The 'enable-alternative' option is used to enable configurations globally."
                    " The 'user' option can't be used with it."
                )
            else:
                ClSelect(config["interpreter"]).enable_version(actions["enable-alternative"])
                if config["format"] == "json":
                    print_json_status_ok()
        elif "disable-alternative" in actions:
            if "user" in config and config["user"]:
                raise ClSelectExcept.VersionModificationBlocked(
                    "The 'disable-alternative' option is used to disable configurations globally."
                    " The 'user' option can't be used with it."
                )
            else:
                ClSelect(config["interpreter"]).disable_version(actions["disable-alternative"])
                if config["format"] == "json":
                    print_json_status_ok()
        elif "list-extensions" in actions:
            check_params(config, ("interpreter", "version"))
            ext_list = get_extensions(config["interpreter"], config["version"], config["format"])
            if config["format"] == "text":
                for item in ext_list:
                    print("%s %s" % item)
            elif config["format"] == "json":
                print(json.dumps(ext_list))
        elif "enable-extensions" in actions:
            check_params(config, ("interpreter", "version"))
            ClExtSelect(config["interpreter"]).enable_extensions(
                config["version"], list(map((lambda i: i.strip()), actions["enable-extensions"].split(",")))
            )
        elif "disable-extensions" in actions:
            check_params(config, ("interpreter", "version"))
            ClExtSelect(config["interpreter"]).disable_extensions(
                config["version"], list(map((lambda i: i.strip()), actions["disable-extensions"].split(",")))
            )
            if len(depend_modules_dict):
                # Warning - blocked modules present
                _check_depencies_and_print_message(config["format"], "Modules left by dependencies:")
        elif "replace-extensions" in actions:
            check_params(config, ("interpreter", "version"))
            ClExtSelect(config["interpreter"]).replace_extensions(
                config["version"], list(map((lambda i: i.strip()), actions["replace-extensions"].split(",")))
            )
            if len(depend_modules_dict):
                # Warning - blocked modules present
                _check_depencies_and_print_message(config["format"], "Modules left/added by dependencies:")
            elif config["format"] == "json":
                print_json_status_ok()
        elif "show-user-summary" in actions:
            check_params(config, ("interpreter", "user"))
            data = ClUserSelect(config["interpreter"], exclude_pid_list).get_summary(
                config["user"], config["show-native-version"]
            )
            print_summary(data)
        elif "show-user-current" in actions:
            check_params(config, ("interpreter", "user"))
            print(
                "%s\t%s\t%s"
                % ClUserSelect(config["interpreter"], exclude_pid_list).get_version(
                    config["user"], config["domain"], config["show-native-version"]
                )
            )
        elif "apply-symlinks-rules" in actions:
            check_params(config, ("interpreter",))
            ClUserSelect(config["interpreter"], exclude_pid_list).apply_symlinks_rules()
        elif "set-user-current" in actions:
            check_params(config, ("interpreter", "user"))
            # hack for alt-php spec where we read and re-apply
            # php version for each user in system
            # in order not to bump deps, we just silently
            # ignore version set requests for 'native'
            # (which should be set for all users on server
            # because web ui does not work)
            try:
                ClSelect().check_requirements()
            except ClSelectExcept.NativeNotInstalled:
                if actions["set-user-current"] != "native":
                    raise
                exit(0)

            # we intentionally use first user cause set_version has workaround for multiple same uid users
            # LVEMAN-1670
            user = clpwd.get_names(uid)[0]
            c = ClUserSelect(config["interpreter"], exclude_pid_list)
            data = c.set_version(
                user,
                actions["set-user-current"],
                config["domain"],
                config["print-summary"],
                config["show-native-version"]
            )
            c.clean_crui_images(clpwd.get_names(uid))
            if config["print-summary"]:
                print_summary(data)
        elif "list-user-extensions" in actions:
            check_params(config, ("interpreter", "user"))
            website = _get_website_from_domain(config["user"], config["domain"])
            if config["show-all"]:
                for ext in ClUserExtSelect(config["interpreter"], exclude_pid_list).list_all_extensions(
                    config["user"], config["version"], website=website
                ):
                    action = "-"
                    if ext[1] is None:
                        action = "~"
                    elif ext[1] is True:
                        action = "+"
                    print("%s %s" % (action, ext[0]))
            else:
                for ext in ClUserExtSelect(config["interpreter"], exclude_pid_list).list_enabled_extensions(
                    config["user"], config["version"], website=website
                ):
                    print(ext[0])
        elif "enable-user-extensions" in actions:
            check_params(config, ("interpreter", "version", "user"))
            ClSelect.check_multiphp_system_default_version()
            users = config["user"].split(",")
            for user in users:
                c = ClUserExtSelect(config["interpreter"], exclude_pid_list)
                website = _get_website_from_domain(user, config["domain"])
                c.bulk_enable_extensions(
                    user=user,
                    version=config["version"],
                    ext_list=list(map((lambda i: i.strip()), actions["enable-user-extensions"].split(","))),
                    check_ext=True,
                    website=website,
                )
            c.clean_crui_images(users)
        elif "disable-user-extensions" in actions:
            check_params(config, ("interpreter", "version", "user"))
            ClSelect.check_multiphp_system_default_version()
            users = config["user"].split(",")
            for user in users:
                c = ClUserExtSelect(config["interpreter"], exclude_pid_list)
                website = _get_website_from_domain(user, config["domain"])
                c.bulk_disable_extensions(
                    user,
                    config["version"],
                    list(map((lambda i: i.strip()), actions["disable-user-extensions"].split(","))),
                    website=website,
                )
            c.clean_crui_images(users)
        elif "replace-user-extensions" in actions:
            check_params(config, ("interpreter", "version", "user"))
            ClSelect.check_multiphp_system_default_version()
            user = clpwd.get_names(uid)[0]
            c = ClUserExtSelect(config["interpreter"], exclude_pid_list)
            website = _get_website_from_domain(user, config["domain"])
            c.replace_extensions(
                user, config["version"], list(map((lambda i: i.strip()), actions["replace-user-extensions"].split(","))),
                website=website
            )
            c.clean_crui_images(clpwd.get_names(uid))
        elif "reset-user-extensions" in actions:
            check_params(config, ("interpreter", "version", "user"))
            ClSelect.check_multiphp_system_default_version()
            user = clpwd.get_names(uid)[0]
            c = ClUserExtSelect(config["interpreter"], exclude_pid_list)
            website = _get_website_from_domain(user, config["domain"])
            extensions = c.reset_extensions(user, config["version"], website=website)
            c.clean_crui_images(clpwd.get_names(uid))
            print(",".join(extensions))
        elif "list-users" in actions:
            check_params(config, ("interpreter", "version"))
            users = ClUserSelect(config["interpreter"], exclude_pid_list).list_users(config["version"])
            print(",".join(users))
        elif "change-to-version" in actions:
            check_params(config, ("interpreter", "version"))
            ClUserSelect(config["interpreter"], exclude_pid_list).change_to_version(
                actions["change-to-version"], config["version"]
            )
        elif "add-options" in actions:
            check_params(config, ("interpreter", "version", "user"))
            ClSelect.check_multiphp_system_default_version()
            user = clpwd.get_names(uid)[0]
            website = _get_website_from_domain(user, config["domain"])
            c = ClUserOptSelect(config["interpreter"], exclude_pid_list)
            c.insert_options(user, config["version"], actions["add-options"], config["decoder"], True, config["quiet"],
                             website=website)
            c.clean_crui_images(clpwd.get_names(uid))
            if config["format"] == "json":
                clprint.print_data(config["format"], {})
        elif "replace-options" in actions:
            check_params(config, ("interpreter", "version", "user"))
            ClSelect.check_multiphp_system_default_version()
            user = clpwd.get_names(uid)[0]
            website = _get_website_from_domain(user, config["domain"])
            c = ClUserOptSelect(config["interpreter"], exclude_pid_list)
            c.insert_options(
                user, config["version"], actions["replace-options"], config["decoder"], False, config["quiet"],
                website=website,
            )
            c.clean_crui_images(clpwd.get_names(uid))
            if config["format"] == "json":
                clprint.print_data(config["format"], {})
        elif "delete-options" in actions:
            check_params(config, ("interpreter", "version", "user"))
            ClSelect.check_multiphp_system_default_version()
            user = clpwd.get_names(uid)[0]
            website = _get_website_from_domain(user, config["domain"])
            c = ClUserOptSelect(config["interpreter"], exclude_pid_list)
            c.delete_options(user, config["version"], actions["delete-options"], config["decoder"], config["quiet"],
                             website=website)
            c.clean_crui_images(clpwd.get_names(uid))
            if config["format"] == "json":
                clprint.print_data(config["format"], {})
        elif "reset-options" in actions:
            user = None
            version = None
            website = _get_website_from_domain(config.get("user"), config["domain"])
            if "user" in config:
                user = config["user"].split(",")
            if config["version"]:
                version = config["version"].split(",")
            c = ClUserOptSelect(config["interpreter"], exclude_pid_list)
            c.reset_options(user, version, website=website)
            c.clean_crui_images(user)
            if config["format"] == "json":
                clprint.print_data(config["format"], {})
        elif "print-options" in actions:
            check_params(config, ("interpreter", "user"))
            website = _get_website_from_domain(config["user"], config["domain"])
            clprint.print_data(
                config["format"],
                ClUserOptSelect(config["interpreter"], exclude_pid_list).get_options(
                    config["user"], config["version"], website=website),
            )
        elif "print-options-safe" in actions:
            check_params(config, ("interpreter", "user"))
            website = _get_website_from_domain(config["user"], config["domain"])
            clprint.print_data(
                config["format"],
                ClUserOptSelect(config["interpreter"], exclude_pid_list).get_options(
                    config["user"], config["version"], website=website),
                escape=True,
            )
        elif "update-backup" in actions:
            clpwd = ClPwd()
            for user in clpwd.get_user_dict().keys():
                try:
                    ClUserSelect()._check_user_in_cagefs(user)
                    ClUserSelect()._backup_settings(user)
                    ClUserOptSelect().backup_php_options(user)
                except ClSelectExcept.NotCageFSUser:
                    pass  # SKIP user with disabled cagefs
                except ClSelectExcept.UnableToSaveData as e:
                    if not config["quiet"]:
                        clprint.print_diag(config["format"], {"status": "ERROR", "message": str(e)})
                    pass  # SKIP user with errors
    except ClSelectExcept.NativeNotInstalled as e:
        clprint.print_diag(
            config["format"], {"status": "WARNING", "message": str(e), "details": e.details, "context": e.context}
        )
        sys.exit(1)
    except BaseClSelectException as e:
        clprint.print_diag(
            config["format"], {"status": "ERROR", "message": str(e), "details": e.details, "context": e.context}
        )
        sys.exit(1)
    except (KeyError, UnboundLocalError):
        print_error_and_exit("Incomplete or incorrect set of arguments")
    except Exception as e:
        msg = traceback.format_exc()
        clprint.print_diag(config["format"], {"status": "ERROR", "message": msg})


if __name__ == "__main__":
    main()
Cancel / Back
[ MAINHACK ]

Viewing / Editing File: clselectctlphp.py

Server Info

System Features

User Info
