1#@!#!123s

: / opt / cloudlinux / venv / lib64 / python3.11 / site-packages / clwpos / feature_suites /
Filename : configurations.py
back
# -*- 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
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 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, get_server_wide_options, ServerWideOptions, ExtendedJSONEncoder

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

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]
    sources: Dict[str, str]
    attributes: Dict[str, Any]
    unique_id: Optional[str]

    purchase_dates: Dict[str, datetime.date]

    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)


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


@dataclass
class FeatureStatus:
    status: FeatureStatusEnum
    source: StatusSource = StatusSource.DEFAULT


def extract_features(
        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_state(suite: str, feature: str, status_for_user: FeatureStatusEnum):
        if status_for_user == FeatureStatusEnum.DEFAULT:
            if feature in server_wide_options.allowed_features:
                return FeatureStatus(status=FeatureStatusEnum.ALLOWED, source=StatusSource.DEFAULT)
            if feature in server_wide_options.visible_features:
                return FeatureStatus(status=FeatureStatusEnum.VISIBLE, source=StatusSource.DEFAULT)
            else:
                return FeatureStatus(status=FeatureStatusEnum.DISABLED, source=StatusSource.DEFAULT)
        else:
            return FeatureStatus(status=status_for_user,
                                 source=admin_config.sources.get(
                                     suite, StatusSource.COMMAND_LINE))

    def _choose_feature_status(s1: FeatureStatus, s2: FeatureStatus) -> 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
        """
        if s1 is None:
            return s2
        if s2 is None:
            return s1
        status_priority = [FeatureStatusEnum.ALLOWED, FeatureStatusEnum.VISIBLE, FeatureStatusEnum.DISABLED]
        return sorted([s1, s2], key=lambda item: status_priority.index(item.status))[0]

    features_state = {}

    for suite, status_for_user in admin_config.suites.items():
        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'

        for feature in getattr(ALL_SUITES[suite], iterable):
            features_state[feature] = _choose_feature_status(
                features_state.get(feature),
                _feature_state(suite, feature, status_for_user))

    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:
                return FeatureStatus(status=FeatureStatusEnum.ALLOWED, source=StatusSource.DEFAULT)
            if suite in server_wide_options.visible_suites:
                return FeatureStatus(status=FeatureStatusEnum.VISIBLE, source=StatusSource.DEFAULT)
            else:
                return FeatureStatus(status=FeatureStatusEnum.DISABLED, source=StatusSource.DEFAULT)
        else:
            return FeatureStatus(status=status_for_user,
                                 source=admin_config.sources.get(
                                     suite, StatusSource.COMMAND_LINE))

    return {
        suite: _suite_state(suite, status_for_user=state)
        for suite, state in admin_config.suites.items()
    }


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, status in suites_admin_config.suites.items()
        if status == FeatureStatusEnum.ALLOWED
    ]


def _is_suite_in_states_for_any_user(suite: str, states: Iterable[FeatureStatusEnum]):
    """
    Checks whether <suite> is in one of the passed states for any server user.
    """
    if is_cl_solo_edition(skip_jwt_check=True):
        data = get_admin_suites_config(uid=None).suites
        return data.get(suite) == FeatureStatusEnum.ALLOWED
    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) in states:
                return True
        return False


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) -> bool:
    """
    Checks whether <suite> visible for at least one user
    """

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


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_admin_suites_config(uid=None) -> AdminSuitesConfig:
    """
    Reads suites statuses from .json.
    In case if config does not exist returns defaults.
    """
    suites = {
        suite.name: (
            FeatureStatusEnum.ALLOWED
            if suite.is_allowed_by_default
            else FeatureStatusEnum.DEFAULT
        )
        for suite in ALL_SUITES.values()
    }

    defaults = AdminSuitesConfig(
        version=ALLOWED_SUITES_CONFIG_VERSION,
        suites=suites,
        sources={},
        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"
        },
        "sources": {
            "accelerate_wp": "ssh",
            "accelerate_wp_premium": "billing"
        }
    }
    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()
                },
                'sources': {},
                '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()
                },
                'sources': {},
                'purchase_dates': {},
                'attributes': {},
                'unique_id': None
            }
        else:
            sources = {}
            # sources actually should always be here and it's more
            # fault-tolerance then real need to use get() instead of []
            for k, source in suites_from_file.get('sources', {}).items():
                try:
                    sources[k] = StatusSource(source)
                except ValueError:
                    logging.warning('Unknown source %s, skipping' % source)

            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()
                },
                'sources': sources,
                '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
    if is_cl_solo_edition(skip_jwt_check=True):
        # root:root 644 - CL Solo
        owner, group, mode = os.getuid(), os.getuid(), 0o644
    else:
        # 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(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()
    if is_cl_solo_edition(skip_jwt_check=True):
        data = extract_features(get_admin_suites_config(uid=None), server_config)
        return data.get(module).status in statuses
    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),
                                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) -> bool:
    """
    Checks whether <module> enabled for at least one user
    """
    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