MAINHACK

Viewing / Editing File: billing.py

File is not writable. Editing disabled.
import datetime
import logging
import os
import pwd
import re
import uuid
from dataclasses import dataclass, asdict
from enum import Enum

from clcommon.clpwd import drop_privileges
from clcommon.cpapi import cpinfo
from typing import List, Dict
from typing_extensions import TypedDict

from clwpos.feature_suites import (
  get_allowed_modules, 
  get_allowed_suites, 
  get_admin_suites_config, 
  write_suites_allowed, 
  ALL_SUITES
)
from clwpos.feature_suites import CDNSuitePro
from clwpos.feature_suites.configurations import FeatureStatusEnum
from clwpos.optimization_features import (
    OBJECT_CACHE_FEATURE,
    CRITICAL_CSS_FEATURE,
    IMAGE_OPTIMIZATION_FEATURE,
    CDN_FEATURE,
    Feature
)
from clwpos.user.config import UserConfig


class BillingFeature(Enum):
    """
    Backwards-compatible list of features that we bill for.
    """
    ACCELERATE_WP_PREMIUM = 'AccelerateWP Premium'
    ACCELERATE_WP_CDN = 'AccelerateWP CDN Free'
    ACCELERATE_WP_CDN_PRO = 'AccelerateWP CDN Pro'


FEATURE_TO_BILLING_FEATURE = {
    OBJECT_CACHE_FEATURE: BillingFeature.ACCELERATE_WP_PREMIUM,
    CRITICAL_CSS_FEATURE: BillingFeature.ACCELERATE_WP_PREMIUM,
    IMAGE_OPTIMIZATION_FEATURE: BillingFeature.ACCELERATE_WP_PREMIUM,
    CDN_FEATURE: BillingFeature.ACCELERATE_WP_CDN,
}

def billing_feature_by_awp_feature(feature, allowed_suites):
    if feature != CDN_FEATURE:
        return FEATURE_TO_BILLING_FEATURE.get(feature)
    # because CND feature is included to multiple suites
    if CDNSuitePro.name in allowed_suites:
        return BillingFeature.ACCELERATE_WP_CDN_PRO
    return FEATURE_TO_BILLING_FEATURE.get(feature)


def is_valid_uuid(value):
    """
    Checks that string has uuid4 format
    """
    if value is None:
        return False
    return re.match('[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}', value) is not None


def get_or_create_unique_identifier(username):
    """
    We need some unique identifier which user can
    use as his token on our provisioning server.

    We use uuid4 to make it somehow hard to bruteforce
    and still unique (hopefully as we don't check that :P)
    """
    if os.geteuid():
        return get_unique_identifier_as_user()

    pw = pwd.getpwnam(username)
    config = get_admin_suites_config(pw.pw_uid)

    unique_id = config.unique_id
    if unique_id is None or not is_valid_uuid(unique_id):
        config.unique_id = str(uuid.uuid4())
        write_suites_allowed(pw.pw_uid, pw.pw_gid, config)
    return config.unique_id


@dataclass
class FeatureRecord:
    name: str
    purchase_date: str  # iso
    attributes: Dict

    active: bool


@dataclass
class UserRecord:
    username: str
    primary_domain: str
    id: str
    features: List[FeatureRecord]


def get_report(legacy=False):
    """
    Collect report about billable users.

    Look for all users with allowed feature and add them
    to list for the further processing on CLN side.

    legacy argument changes format so it
    is accepted by older CLN versions
    """
    report = []
    for user, domain in cpinfo(keyls=('cplogin', 'dns')):
        try:
            uid = pwd.getpwnam(user).pw_uid
        except KeyError:
            logging.warning('User %s does not have system uid; Malformed control panel configuration?')
            continue
        modules: List[Feature] = get_allowed_modules(uid)

        enabled_modules = _get_enabled_modules(user)

        suites_admin_config = get_admin_suites_config(uid)
        allowed_suites = [
            suite
            for suite, status in suites_admin_config.suites.items()
            if status == FeatureStatusEnum.ALLOWED
        ]

        # configuration has information about purchase date of suites
        # here we transform it into information about feature purchase dates
        feature_purchase_dates = {
            feature: suites_admin_config.purchase_dates.get(suite)
            for suite in allowed_suites
            for feature in ALL_SUITES[suite].features
        }

        feature_attributes = {
            feature: suites_admin_config.attributes.get(suite, dict())
            for suite in allowed_suites
            for feature in ALL_SUITES[suite].features
        }

        billable_features = []
        processed_features = set()
        for module in modules:
            billing_feature = billing_feature_by_awp_feature(module, allowed_suites)
            if billing_feature is None:
                continue

            if billing_feature.value not in processed_features:
                if legacy:
                    billable_features.append(
                        billing_feature.value
                    )
                else:
                    billable_features.append(FeatureRecord(
                        name=billing_feature.value,
                        purchase_date=(
                            str(feature_purchase_dates[module])
                            if feature_purchase_dates.get(module) else
                            # for the features that were allowed before this change
                            # the purchase date would be 1st day of current month
                            str(datetime.date.today().replace(day=1))
                        ),
                        attributes=feature_attributes.get(module, {}),
                        active=module in enabled_modules
                    ))
                processed_features.add(billing_feature.value)

        report.append(asdict(UserRecord(
            username=user,
            primary_domain=domain,
            id=get_or_create_unique_identifier(user),
            features=billable_features
        )))
    return report


def _get_enabled_modules(user):
    with drop_privileges(user):
        return {module for _, _, module in UserConfig(
            username=user).enabled_modules()}


def get_unique_identifier_as_user():
    """
    Get unique identifier for current end-user
    """
    from clwpos.utils import daemon_communicate
    from clwpos.daemon import WposDaemon

    unique_id = daemon_communicate({
        "command": WposDaemon.DAEMON_GET_UNIQUE_ID_COMMAND
    })["unique_id"]
    return unique_id
Cancel / Back
[ MAINHACK ]

Viewing / Editing File: billing.py

Server Info

System Features

User Info
