File Manager

✏️ ites/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 datetime
import json
import logging
import pwd
import os
from dataclasses import dataclass, field
from enum import Enum
from typing import Optional, Tuple, Dict, Union, Iterable, Any

from clcommon.clcagefs import _remount_cagefs
from clcommon.cpapi import cpusers
from clwpos.constants import (
    CLWPOS_VAR_DIR,
    CLWPOS_UIDS_PATH,
    ALLOWED_MODULES_JSON
)
from clwpos.utils import (
    uid_by_name,
    acquire_lock,
    get_server_wide_options,
    ServerWideOptions,
    ExtendedJSONEncoder,
    name_by_uid,
    is_user_owned_by_reseller
 )
from clwpos.optimization_features import Feature

from .suites import (
    ALL_SUITES,
    AWPSuite,
    PremiumSuite,
    CDNSuite,
    CDNSuitePro,
    OLD_NEW_SUITE_NAME_PAIRS
)

from .cache import get_cached_visible_any_user

ALLOWED_SUITES_CONFIG_VERSION = 3
ALLOWED_SUITES_JSON = 'suites_allowed.json'

class FeatureStatusEnum(Enum):
    # means that user does not have
    # any custom option set
    DEFAULT = 'default'
    # means that user is specifically forbidden
    # to use accelerate wp
    DISABLED = 'disabled'
    # means that user can see the feature,
    # but cannot install the plugin
    VISIBLE = 'visible'
    # user can both see and use feature
    ALLOWED = 'allowed'

@dataclass
class AdminSuitesConfig:
    version: int
    suites: Dict[str, FeatureStatusEnum]
    attributes: Dict[str, Any]
    unique_id: Optional[str]

purchase_dates: Dict[str, datetime.date]
    whmcs_suites: Optional[Dict[str, FeatureStatusEnum]] = field(default_factory=dict)

def __post_init__(self):
        """
        Remove unknown suites from resulting structure.
        Actual for downgrade cases, see AWP-272 for details
        """
        for suite in set(self.suites).difference(set(ALL_SUITES)):
            self.suites.pop(suite)

@property
    def suites_list(self):
        all_suites = sorted(set(list(self.suites.keys()) + list(self.whmcs_suites.keys())))
        return all_suites

class StatusSource(Enum):
    DEFAULT = 'default'
    COMMAND_LINE = 'manual'
    BILLING_OVERRIDE = 'billing'

@dataclass
class FeatureStatus:
    status: FeatureStatusEnum

def _choose_feature_status(target_statuses) -> FeatureStatus:
    """
    Choose final feature status according to priority
    CDN is included to multiple suites, so need to decide is it allowed/visible/etc

e.g if at least one suite with cdn feature is allowed -> feature is allowed
    """
    sort_by = [feature_status for feature_status in target_statuses
               if feature_status and feature_status.status is not None]
    status_priority = [FeatureStatusEnum.ALLOWED, FeatureStatusEnum.VISIBLE, FeatureStatusEnum.DISABLED]
    return sorted(sort_by, key=lambda item: status_priority.index(item.status))[0]

def extract_features(
        uid: int,
        admin_config: AdminSuitesConfig,
        server_wide_options: ServerWideOptions,
        allowed_state: FeatureStatusEnum = FeatureStatusEnum.DEFAULT,
    ) -> Dict[str, FeatureStatus]:
    """
    Construct feature allowance dict based on given suites.

We require both primary_features and feature_set
    iterables. Consider the following corner cases:
    feature_set iterable:
    1. All suites are disallowed
    2. We want to allow only accelerate_wp_cdn suite
    3. We aim to allow both cdn and site_optimization features,
       thus we need feature_set iterable

primary_features iterable:
    1. All suites are allowed
    2. We want to disallow only accelerate_wp_cdn suite
    3. We aim to disallow only cdn feature, not site_optimization,
       thus the need for a primary_features iterable
    """
    def _feature_supported_for_user(uid: int, feature: str):
        """
        All free features are available for users
        Addon ownership check for paid features
        """
        if not Feature(feature).IS_BILLABLE:
            return True
        return not is_user_owned_by_reseller(name_by_uid(uid))

def _feature_state(uid: int, suite: str, feature: str, status_for_user: FeatureStatusEnum):
        if feature not in server_wide_options.supported_features:
            return FeatureStatus(status=FeatureStatusEnum.DISABLED)

if not _feature_supported_for_user(uid, feature):
            return FeatureStatus(status=FeatureStatusEnum.DISABLED)

if status_for_user == FeatureStatusEnum.DEFAULT:
            if feature in server_wide_options.allowed_features:
                return FeatureStatus(status=FeatureStatusEnum.ALLOWED)
            if feature in server_wide_options.visible_features:
                return FeatureStatus(status=FeatureStatusEnum.VISIBLE)
            else:
                return FeatureStatus(status=FeatureStatusEnum.DISABLED)
        else:
            return FeatureStatus(status=status_for_user)

features_state = {}
    for suite in admin_config.suites_list:
        is_disallowed_by_default = (
            ALL_SUITES[suite].primary_features not in server_wide_options.visible_features
        )

if allowed_state == FeatureStatusEnum.DISABLED or \
           allowed_state == FeatureStatusEnum.DEFAULT and is_disallowed_by_default:
            iterable = 'primary_features'
        else:
            iterable = 'feature_set'
        manual_status = admin_config.suites.get(suite)
        whmcs_suite_status = admin_config.whmcs_suites.get(suite)
        for feature in getattr(ALL_SUITES[suite], iterable):
            features_state[feature] = _choose_feature_status([
                features_state.get(feature),
                _feature_state(uid, suite, feature, manual_status),
                _feature_state(uid, suite, feature, whmcs_suite_status)
            ])

return features_state

def extract_suites(
        admin_config: AdminSuitesConfig,
        server_wide_options: ServerWideOptions) -> Dict[str, FeatureStatus]:
    """
    Construct feature dict based on given suites
    """
    def _suite_state(suite: str, status_for_user: FeatureStatusEnum):
        if status_for_user == FeatureStatusEnum.DEFAULT:
            if suite in server_wide_options.allowed_suites_list:
                return FeatureStatus(status=FeatureStatusEnum.ALLOWED)
            if suite in server_wide_options.visible_suites_list:
                return FeatureStatus(status=FeatureStatusEnum.VISIBLE)
            else:
                return FeatureStatus(status=FeatureStatusEnum.DISABLED)
        else:
            return FeatureStatus(status=status_for_user)

suites_state = {}
    for suite in admin_config.suites_list:
        whmcs_suite_status = admin_config.whmcs_suites.get(suite)
        manual_status = admin_config.suites.get(suite)
        final_state = _choose_feature_status([
            _suite_state(suite, manual_status),
            _suite_state(suite, whmcs_suite_status)
        ])
        suites_state[suite] = final_state
    return suites_state

def get_admin_config_directory(uid: 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
    """
    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
    """
    server_defaults = get_server_wide_options()
    suites_statuses = extract_suites(get_admin_suites_config(uid), server_defaults)
    return [suite for suite, suite_status in suites_statuses.items()
            if suite_status.status == FeatureStatusEnum.ALLOWED]

def _is_suite_in_states_for_any_user(suite: str, states: Iterable[FeatureStatusEnum], username=None):
    """
    Checks whether <suite> is in one of the passed states for any server user.
    """
    server_defaults = get_server_wide_options()
    if username:
        uid = uid_by_name(username)
        return extract_suites(get_admin_suites_config(uid), server_defaults).get(suite).status in states
    else:
        users = list(cpusers())
        for username in users:
            uid = uid_by_name(username)
            if not uid:
                continue
            if extract_suites(get_admin_suites_config(uid), server_defaults).get(suite).status in states:
                return True
        return extract_suites(get_admin_suites_config(uid=None), server_defaults).get(suite).status in states

def is_suite_allowed_for_user(suite: str) -> bool:
    """
    Checks whether <suite> enabled for at least one user
    """
    return _is_suite_in_states_for_any_user(suite, (FeatureStatusEnum.ALLOWED, ))

def is_suite_visible_for_user(suite: str, username=None) -> bool:
    """
    Checks whether <suite> visible for at least one user
    """

return _is_suite_in_states_for_any_user(suite,
                                            (FeatureStatusEnum.ALLOWED, FeatureStatusEnum.VISIBLE),
                                            username)

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 any_suite_visible_on_server() -> bool:
    """
    Check if there are any feature suite allowed on server
    """
    return any(is_suite_visible_for_user(suite) for suite in ALL_SUITES)

def get_default_suites_states():
    suites = {
        suite.name: (
            FeatureStatusEnum.ALLOWED
            if suite.is_allowed_by_default
            else FeatureStatusEnum.DEFAULT
        )
        for suite in ALL_SUITES.values()
    }
    return suites

def get_admin_suites_config(uid=None) -> AdminSuitesConfig:
    """
    Reads suites statuses from .json.
    In case if config does not exist returns defaults.
    """
    defaults = AdminSuitesConfig(
        version=ALLOWED_SUITES_CONFIG_VERSION,
        suites=get_default_suites_states(),
        whmcs_suites={},
        purchase_dates={},
        attributes={},
        unique_id=None
    )
    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: AdminSuitesConfig,
        suites_json_path,
        old_config=False) -> AdminSuitesConfig:
    """
    Reads json with suites statuses
    for new awp version:
    {
        "version": "3",
        "suites": {
            "accelerate_wp": "allowed",
            "accelerate_wp_premium": "visible"
        }
    }
    for older awp version:
    {
        "version": "2",
        "suites": {
            "accelerate_wp": true,
            "accelerate_wp_premium": true
        }
    }
    for oldest 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 old_config:
            suites = {
                'version': ALLOWED_SUITES_CONFIG_VERSION,
                'suites': {
                    OLD_NEW_SUITE_NAME_PAIRS[k]: FeatureStatusEnum.ALLOWED if v else FeatureStatusEnum.DISABLED
                    for k, v in suites_from_file[suites_key].items()
                },
                'purchase_dates': {},
                'attributes': {},
                'unique_id': None
            }
        # both versions can be in config based on the time when it was created first time
        elif int(suites_from_file['version']) == 2 or int(suites_from_file['version']) == 1:
            suites = {
                'version': ALLOWED_SUITES_CONFIG_VERSION,
                'suites': {
                    k: FeatureStatusEnum.ALLOWED if v else FeatureStatusEnum.DISABLED
                    for k, v in suites_from_file[suites_key].items()
                },
                'purchase_dates': {},
                'attributes': {},
                'unique_id': None
            }
        else:
            attributes = {}
            for suite, suite_config in ALL_SUITES.items():
                suite_attrs = suites_from_file.get('attributes', {}).get(suite, {})
                suite_valid_attrs = {}
                for attribute in suite_config.allowed_attrubites:
                    attr_value = suite_attrs.get(attribute.name)
                    suite_valid_attrs[attribute.name] = attribute.type(attr_value) \
                        if attr_value else attribute.default
                if suite_valid_attrs:
                    attributes[suite] = suite_valid_attrs

suites = {
                'version': ALLOWED_SUITES_CONFIG_VERSION,
                'suites': {
                    k: FeatureStatusEnum(v)
                    for k, v in suites_from_file[suites_key].items()
                },
                'whmcs_suites': {
                    k: FeatureStatusEnum(v)
                    for k, v in suites_from_file.get('whmcs_suites', {}).items()
                },
                'purchase_dates': {
                    suite: datetime.datetime.strptime(date, '%Y-%m-%d').date()
                    for suite, date in suites_from_file.get('purchase_dates', {}).items()
                },
                'attributes': attributes,
                'unique_id': suites_from_file.get('unique_id')
            }
        # 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.items():
            suites['suites'].setdefault(suite, status)

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

return AdminSuitesConfig(**suites)

def write_suites_allowed(uid: int, gid: int,
                         data_dict_to_write: Union[Dict, AdminSuitesConfig],
                         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, cls=ExtendedJSONEncoder)

try:
        os.makedirs(os.path.dirname(modules_allowed_path), 0o755, exist_ok=False)
    except OSError:
        pass
    else:
        # this won't happen a lot of time because in --allowed-for-all
        # loop we create path manually
        # the purpose of this is to handle situation when we create config when
        # we are trying to get unique_id and it still does not exists
        _remount_cagefs(pwd.getpwuid(uid).pw_name)

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)
    """
    # usually os.getuid() will be root here
    # but for example in unit tests it could be mockbuild
    # and we should live with that
    # root:username 640 - CL Shared Pro
    owner, group, mode = os.getuid(), gid, 0o640
    return owner, group, mode

def _get_modules_by_status(uid: int, statues: Iterable[FeatureStatusEnum]):
    """
    Reads configuration file (which is manipulated by admin)
    and returns only that modules which are in any of given status
    :param uid: uid (used only for CL Shared, not used on solo)
    @return: list of module unique ids
    """
    default_config = get_server_wide_options()
    suites_admin_config = get_admin_suites_config(uid)

return [
        feature
        for feature, feature_status in extract_features(
            uid,
            suites_admin_config, default_config).items()
        if feature_status.status in statues
    ]

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
    """
    return _get_modules_by_status(uid, (FeatureStatusEnum.ALLOWED, ))

def get_visible_modules(uid: int) -> list:
    """
    Reads configuration file (which is manipulated by admin)
    and returns only that modules which are visible by endusers.
    :param uid: uid (used only for CL Shared, not used on solo)
    @return: list of module unique ids
    """
    return _get_modules_by_status(uid, (FeatureStatusEnum.ALLOWED, FeatureStatusEnum.VISIBLE))

def _get_features_dict_in_statuses(uid: int, statuses: Iterable[FeatureStatusEnum]):
    """
    Dict with features per feature-set which are in <statuses>
    """
    allowed_features = _get_modules_by_status(uid, statuses)
    allowed_suites = get_allowed_suites(uid)

free, premium, cdn_pro = [], [], []
    for feature in allowed_features:
        if feature in AWPSuite.feature_set or feature in CDNSuite.feature_set:
            free.append(feature)
        elif feature in PremiumSuite.feature_set:
            premium.append(feature)
        if feature in CDNSuitePro.primary_features and CDNSuitePro.name in allowed_suites:
            cdn_pro.append(feature)

return {
        'accelerate_wp': free,
        'accelerate_wp_premium': premium,
        'accelerate_wp_cdn_pro': cdn_pro,
    }

def get_allowed_features_dict(uid: int):
    """
    Dict with allowed features per feature-set
    """
    return _get_features_dict_in_statuses(uid, (FeatureStatusEnum.ALLOWED, ))

def get_visible_features_dict(uid: int):
    """
    Dict with visible features per feature-set
    """
    return _get_features_dict_in_statuses(uid, (FeatureStatusEnum.ALLOWED, FeatureStatusEnum.VISIBLE))

def _is_module_in_state_for_user(module: str, statuses: Iterable[FeatureStatusEnum]) -> bool:
    """
    Checks whether <module> is in any of <statuses> for at least one user
    """
    server_config = get_server_wide_options()
    users = list(cpusers())
    for username in users:
        uid = uid_by_name(username)
        if not uid:
            continue
        if extract_features(uid, get_admin_suites_config(uid),
                            server_wide_options=server_config).get(module).status in statuses:
            return True
    return False

def is_module_allowed_for_user(module: str) -> bool:
    """
    Checks whether <module> enabled for at least one user
    """
    return _is_module_in_state_for_user(module, (FeatureStatusEnum.ALLOWED, ))

def is_module_visible_for_user(module: str, use_cache=False) -> bool:
    """
    Checks whether <module> enabled for at least one user
    """
    if use_cache:
        cached_statuses = get_cached_visible_any_user()
        if cached_statuses:
            return module in cached_statuses
    return _is_module_in_state_for_user(module, (FeatureStatusEnum.ALLOWED, FeatureStatusEnum.VISIBLE))

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'] == FeatureStatusEnum.ALLOWED,
            'site_optimization': suites_admin_config['accelerate_wp'] == FeatureStatusEnum.ALLOWED
        }
    }
    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

Batal

⬅ clwpos 5 item

Nama	Tipe	Ukuran	Diubah	Aksi
📁__pycache__	dir	—	2026-03-18 20:54	📂 Buka
🐍__init__.py	py	933 B	2026-01-13 11:07	✏️ 👁️
🐍cache.py	py	787 B	2026-01-13 11:07	✏️ 👁️
🐍configurations.py	py	21.6 KB	2026-01-13 11:07	✏️ 👁️
🐍suites.py	py	3.9 KB	2026-01-13 11:07	✏️ 👁️

✏️ Rename