✏️ Editing: billing.py Read Only

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 clwpos.feature_suites import (
  get_allowed_modules,
  get_admin_suites_config, 
  write_suites_allowed, 
  ALL_SUITES,
  get_allowed_suites
)
from clwpos.feature_suites import CDNSuitePro
from clwpos.optimization_features import (
    OBJECT_CACHE_FEATURE,
    CRITICAL_CSS_FEATURE,
    IMAGE_OPTIMIZATION_FEATURE,
    CDN_FEATURE,
    Feature
)
from clwpos.user.config import UserConfig
from clwpos.utils import is_wp2_environment_safe

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 = {
    IMAGE_OPTIMIZATION_FEATURE: BillingFeature.ACCELERATE_WP_PREMIUM,
    CDN_FEATURE: BillingFeature.ACCELERATE_WP_CDN,
}

if not is_wp2_environment_safe():
    FEATURE_TO_BILLING_FEATURE[CRITICAL_CSS_FEATURE] = BillingFeature.ACCELERATE_WP_PREMIUM

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():
    """
    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?', user)
            continue
        try:
            user_record = _single_user_report(uid, user, domain)
        except Exception:
            logging.exception('CLN billing report for user %s failed', user)
            continue
        report.append(asdict(user_record))
    return report

def _single_user_report(uid, username, domain):
    features: List[Feature] = get_allowed_modules(uid)

enabled_features = _get_enabled_modules(username)

suites_admin_config = get_admin_suites_config(uid)
    allowed_suites = get_allowed_suites(uid)

# 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 = dict()
    for feature in features:
        billing_feature = billing_feature_by_awp_feature(feature, allowed_suites)
        if billing_feature is None:
            continue

if billing_feature.value not in billable_features:
            billable_features[billing_feature.value] = FeatureRecord(
                name=billing_feature.value,
                purchase_date=(
                    str(feature_purchase_dates[feature])
                    if feature_purchase_dates.get(feature) 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(feature, {}),
                active=feature in enabled_features
            )
        else:
            # if suite was already processed, we still need to check other features
            # because e.g. suite Premium has 3(!) features included and ANY
            # of those features enabled should set it to active=true
            # previously we had a bad logic hare where we skipped all the rest of the data
            billable_features[billing_feature.value].active = \
                billable_features[billing_feature.value].active or (feature in enabled_features)
    return UserRecord(
            username=username,
            primary_domain=domain,
            id=get_or_create_unique_identifier(username),
            features=list(billable_features.values())
        )

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

🔐 Sid Gifari File Manager Pro