/opt/cloudlinux/venv/lib/python3.11/site-packages/clwpos
Edit: /opt/cloudlinux/venv/lib/python3.11/site-packages/clwpos/billing.py (6392B)
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