Na}{

File Content: `configurations.py`

# -*- 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

# configurations.py - configuration helpers for AccelerateWP feature suites

import json
import logging
import pwd
import os
from typing import Optional, Tuple, Dict

from clcommon.cpapi import cpusers
from clcommon.lib.cledition import is_cl_solo_edition
from clwpos import gettext as _
from clwpos.cl_wpos_exceptions import WposError
from clwpos.constants import (
    CLWPOS_VAR_DIR,
    CLWPOS_UIDS_PATH,
    ALLOWED_MODULES_JSON
)
from clwpos.utils import uid_by_name, acquire_lock

from .suites import ALL_SUITES, AWPSuite, PremiumSuite, OLD_NEW_SUITE_NAME_PAIRS

ALLOWED_SUITES_CONFIG_VERSION = 2
ALLOWED_SUITES_JSON = 'suites_allowed.json'
IS_SUITE_ALLOWED_BY_DEFAULT = bool(is_cl_solo_edition())


def extract_features(suites: Dict[str, bool]) -> Dict[str, bool]:
    """
    Construct feature dict based on given suites
    """
    return {feature: state for suite, state in suites.items() for feature
            in ALL_SUITES[suite].feature_set}


def get_admin_config_directory(uid: Optional[int]) -> str:
    """
    Get directory path in which admin's config files are stored.
    Hides logic of detecting current OS edition environment.
    :param uid: uid
    :return: admin's config directory path
    """
    is_solo = is_cl_solo_edition()
    if is_solo:
        admin_config_dir = os.path.join(CLWPOS_VAR_DIR, 'solo')
    else:
        if uid is None:
            raise WposError(
                message=_('Internal error: obtaining config path without uid is only '
                          'available for CloudLinux OS Solo. '
                          'Please contact support for help: '
                          'https://cloudlinux.zendesk.com'))
        admin_config_dir = os.path.join(CLWPOS_UIDS_PATH, str(uid))
    return admin_config_dir


def get_suites_allowed_path(uid: Optional[int], old=False) -> str:
    """
    Get suites_allowed file path for user.
    :param uid: uid
    :param old: if "old" allowed modules config needed
    :return: suites_allowed file path
    """
    admin_config_dir = get_admin_config_directory(uid)
    if not old:
        suites_allowed_path = os.path.join(admin_config_dir, ALLOWED_SUITES_JSON)
    else:
        suites_allowed_path = os.path.join(admin_config_dir, ALLOWED_MODULES_JSON)
    return suites_allowed_path


def get_allowed_suites(uid: int) -> list:
    """
    Reads configuration file (which is manipulated by admin)
    and returns only that suites which are allowed
    to be enabled by endusers.
    :param uid: uid (used only for CL Shared, not used on solo)
    @return: list of module unique ids
    """
    suites_admin_config = get_admin_suites_config(uid)
    return [
        suite
        for suite, is_allowed in suites_admin_config['suites'].items()
        if is_allowed
    ]


def is_suite_allowed_for_user(suite: str) -> bool:
    """
    Checks whether <suite> enabled for at least one user
    """
    if is_cl_solo_edition(skip_jwt_check=True):
        data = get_admin_suites_config(uid=None)['suites']
        return data.get(suite)
    else:
        users = list(cpusers())
        for username in users:
            uid = uid_by_name(username)
            if not uid:
                continue
            if get_admin_suites_config(uid)['suites'].get(suite):
                return True
        return False


def any_suite_allowed_on_server() -> bool:
    """
    Check if there are any feature suite allowed on server
    """
    return any(is_suite_allowed_for_user(suite) for suite in ALL_SUITES)


def get_admin_suites_config(uid=None):
    """
    Reads suites statuses from .json.
    In case if config does not exist returns defaults.
    """
    suites = dict.fromkeys(ALL_SUITES, IS_SUITE_ALLOWED_BY_DEFAULT)

    if is_cl_solo_edition():
        suites[PremiumSuite.name] = os.path.exists('/var/lve/enable-wpos.flag')

    defaults = {
        "version": str(ALLOWED_SUITES_CONFIG_VERSION),
        "suites": suites,
    }
    suites_json_path = get_suites_allowed_path(uid)
    old_suited_json_path = get_suites_allowed_path(uid, old=True)
    if os.path.exists(suites_json_path):
        return read_json_with_allowed_suites(defaults, suites_json_path)
    elif os.path.exists(old_suited_json_path):
        return read_json_with_allowed_suites(defaults, old_suited_json_path, old_config=True)
    else:
        return defaults


def read_json_with_allowed_suites(defaults, suites_json_path, old_config=False):
    """
    Reads json with suites statuses
    for new awp version:
    {
        "version": "2",
        "suites": {
            "accelerate_wp": true,
            "accelerate_wp_premium": true
        }
    }
    for old awp version:
    {
        "version": "1",
        "modules": {
            "object_cache": true,
            "site_optimization": true
        }
    }
    """
    suites_key = 'suites' if not old_config else 'modules'
    # TODO: locking and tempfiles
    # https://cloudlinux.atlassian.net/browse/LU-2073
    try:
        with open(suites_json_path, "r") as f:
            suites_from_file = json.load(f)

        if not old_config:
            # update admin's config with modules that are not in it (values are taken from defaults)
            # case: new module was added in the lve-utils update and it is not in the config yet
            for suite, status in defaults[suites_key].items():
                suites_from_file[suites_key].setdefault(suite, status)

        if old_config:
            suites = {k: v for k, v in suites_from_file.items() if k != suites_key}
            suites['suites'] = {OLD_NEW_SUITE_NAME_PAIRS[k]: v for k, v in suites_from_file['modules'].items()}
        else:
            suites = suites_from_file

    except (json.JSONDecodeError, KeyError) as e:
        logging.warning('Config %s is malformed, using defaults instead, error: %s', suites_json_path, e)
        return defaults

    return suites


def write_suites_allowed(uid: int, gid: int, data_dict_to_write: dict, custom_allowed_path: str = None):
    """
    Writes modules_allowed file for user
    :param uid: User uid
    :param gid: User gid
    :param data_dict_to_write: Data to write
    :param custom_allowed_path: custom path of allowed config
    """
    modules_allowed_path = custom_allowed_path or get_suites_allowed_path(uid)
    json_data = json.dumps(data_dict_to_write, indent=4)

    with open(modules_allowed_path, "w") as f:
        f.write(json_data)

    owner, group, mode = get_admin_config_permissions(gid)
    os.chown(modules_allowed_path, owner, group)
    os.chmod(modules_allowed_path, mode)


def get_admin_config_permissions(gid: int) -> Tuple[int, int, int]:
    """
    Return owner, group and permission which files inside
    admin's config directory should have.
    User should have rights to read (not write) config,
    so we set owner root, group depends on CL edition (see comment above)
    """
    if is_cl_solo_edition(skip_jwt_check=True):
        # root:root 644 - CL Solo
        owner, group, mode = 0, 0, 0o644
    else:
        # root:username 640 - CL Shared Pro
        owner, group, mode = 0, gid, 0o640
    return owner, group, mode


def get_allowed_modules(uid: int) -> list:
    """
    Reads configuration file (which is manipulated by admin)
    and returns only that modules which are allowed
    to be enabled by endusers.
    :param uid: uid (used only for CL Shared, not used on solo)
    @return: list of module unique ids
    """
    suites_admin_config = get_admin_suites_config(uid)
    return [
        feature
        for feature, is_allowed in extract_features(suites_admin_config['suites']).items()
        if is_allowed
    ]


def get_allowed_features_dict(uid: int):
    """
    Dict with allowed features per feature-set
    """
    allowed_features = get_allowed_modules(uid)
    free, premium = [], []
    for feature in allowed_features:
        if feature in AWPSuite.feature_set:
            free.append(feature)
        elif feature in PremiumSuite.feature_set:
            premium.append(feature)
    return {
        'accelerate_wp': free,
        'accelerate_wp_premium': premium
    }


def is_module_allowed_for_user(module: str) -> bool:
    """
    Checks whether <module> enabled for at least one user
    """
    if is_cl_solo_edition(skip_jwt_check=True):
        data = extract_features(get_admin_suites_config(uid=None)['suites'])
        return data.get(module)
    else:
        users = list(cpusers())
        for username in users:
            uid = uid_by_name(username)
            if not uid:
                continue
            if extract_features(get_admin_suites_config(uid)['suites']).get(module):
                return True
        return False


def _sync_allowed_configs(username):
    """
    Syncing allowed configs (needed for downgrade)
    """
    uid = pwd.getpwnam(username).pw_uid
    gid = pwd.getpwnam(username).pw_gid
    suites_json_path = get_suites_allowed_path(uid)

    # means there is no custom settings in config
    if not os.path.exists(suites_json_path):
        return

    suites_admin_config = get_admin_suites_config(uid)['suites']
    config_to_sync = get_suites_allowed_path(uid, old=True)

    modules_states = {'version': 1,
                      'modules': {'object_cache': suites_admin_config['accelerate_wp_premium'],
                                  'site_optimization': suites_admin_config['accelerate_wp']}
                      }
    with acquire_lock(config_to_sync):
        write_suites_allowed(uid, gid, modules_states, custom_allowed_path=config_to_sync)


def sync_allowed_configs():
    users = list(cpusers())
    for user in users:
        try:
            _sync_allowed_configs(user)
        except Exception:
            logging.exception('Error while syncing the allowed configs for user %s', user)
            continue
Name	Size	Permission
__pycache__	---	0755
__init__.py	740 bytes	0644
configurations.py	10042 bytes	0644
suites.py	1090 bytes	0644
File Content: configurations.py

File Content: `configurations.py`
