PHP Dosya Yöneticisi

# -*- coding: utf-8 -*-

# Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2021 All Rights Reserved
#
# Licensed under CLOUD LINUX LICENSE AGREEMENT
# http://cloudlinux.com/docs/LICENSE.TXT

import logging
import json
import os
import pwd

from typing import Any
from schema import Schema, SchemaError
from functools import wraps
from typing import Callable
from xray.internal.clwpos_safe_imports import get_user_auth_key
from clcommon.cpapi import userdomains, cpusers
from clcommon.clwpos_lib import find_wp_paths

from xray import gettext as _
from ..apiclient import get_client
from ..apiclient.schemas import pullzone_info_schema
from ..internal.exceptions import XRayAPIError, XRayError
from ..internal.utils import timestamp
from ..internal.user_plugin_utils import get_xray_exec_user

def validate_users_data(func: Callable):
    def validate(account_id: str, domain: str, website: str, username: str):
        """
        If user owns domain and website
        """
        proxy_user = get_xray_exec_user()

# run cl-smart-advice-user (as user)
        if proxy_user:
            username = proxy_user

# run cl-smart-advice (as root)
        else:
            if not username:
                for user in list(cpusers()):
                    auth_key = get_user_auth_key(user)
                    if auth_key == account_id:
                        username = user
                        break
                else:
                    raise XRayError(_('Passed account_id does not belong to any user on the server'))

domains_docroot_list = userdomains(username)
        for domain_item in domains_docroot_list:
            if domain_item[0] == domain.replace('https://', '').replace('http://', ''):
                wordpress_list = list(find_wp_paths(domain_item[1]))
                wp_site = website.replace('/', '')
                if wp_site not in wordpress_list:
                    raise XRayError(f'Passed website is not found, all sites: {wordpress_list}')
                break
        else:
            raise XRayError(_('Passed domain cannot be found in list of user domains'))

@wraps(func)
    def wrapper(*args, **kwargs):
        account_id = kwargs.get('account_id')
        username = kwargs.get('username')
        validate(account_id, kwargs['domain'], kwargs['website'], username)

response = func(*args, **kwargs)
        return response

return wrapper

class AWPProvisionAPI:
    """Awp provision service handlers"""

def __init__(self):
        self.logger = logging.getLogger('awp_provision_api')
        # initialize Adviser API client: awp provision uses same host as adviser
        awp_client_object = get_client('awp_provision')
        self.awp_client_api = awp_client_object()

def _account_id(self, account_id=None, username=None):
        if not account_id:
            try:
                proxy_user = os.environ.get('XRAYEXEC_UID')
                if proxy_user:
                    username = pwd.getpwuid(int(proxy_user)).pw_name
                return get_user_auth_key(username)
            except Exception as e:
                raise XRayAPIError(_('Auth file was not found!')) from e
        return account_id

@staticmethod
    def response(**kwargs) -> str:
        initial = {'result': 'success', 'timestamp': timestamp()}
        if kwargs:
            initial.update(kwargs)
        return json.dumps(dict(sorted(initial.items())))

def _validate(self, data: Any, schema: Schema) -> Any:
        """Validate given data using given schema"""
        try:
            return schema.validate(data)
        except SchemaError as e:
            self.logger.error('Failed to validate API response: %s', data)
            msg = e.errors[-1] or e.autos[-1]
            raise XRayAPIError(_('Malformed API response: %s') % str(msg))

@validate_users_data
    def get_pullzone(self, account_id, domain, website, username=None):
        account_id = self._account_id(account_id, username)
        data = self.awp_client_api.get_create_pullzone(account_id, domain, website)
        response = self._validate(data=data,
                                  schema=pullzone_info_schema)
        return self.response(data=response)

@validate_users_data
    def remove_pullzone(self, account_id, domain, website, username=None):
        account_id = self._account_id(account_id, username)
        response = self.awp_client_api.remove_pullzone(account_id, domain, website)
        return self.response(data=response)

@validate_users_data
    def purge_cdn_cache(self, account_id, domain, website, username=None):
        account_id = self._account_id(account_id, username)
        response = self.awp_client_api.purge_cdn_cache(account_id, domain, website)
        return self.response(data=response)

def sync_account(self, account_id, username=None):
        account_id = account_id or [self._account_id(account_id, username)]
        response = self.awp_client_api.sync_account(account_id)
        return self.response(data=response)

def get_usage(self, account_id, username=None):
        account_id = self._account_id(account_id, username)
        response = self.awp_client_api.get_usage(account_id)
        return self.response(data=response)

İptal

📁 PHP Dosya Yöneticisi

📝 awp_provision_api.py

📝 awp_provision_api.py

📝 Yeniden Adlandır

🔐 Dosya İzinleri (chmod)