Na}{

File Content: `litespeed.py`

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

from __future__ import absolute_import
import base64
import urllib.request
import urllib.error
import urllib.parse
import ssl
import os

from lxml import etree
from lvestats.lib.commons.func import get_all_user_domains, normalize_domain


class LiteSpeedException(Exception):
    pass


class LiteSpeedDisabledException(LiteSpeedException):
    pass


class LiteSpeedInvalidCredentials(LiteSpeedException):
    pass


class LiteSpeedDataMapping(object):
    TIME = 3
    HOST = 8
    REQUEST = 14

    TOTAL_LEN = 15


class LiteSpeed(object):
    IGNORE_HOSTS = [b'_AdminVHost']
    PID_FILE_PATH = '/tmp/lshttpd/lshttpd.pid'
    HTPASSWD_PATH = '/usr/local/lsws/admin/htpasswds/status'
    HTTP_TIMEOUT = 2
    LS_ADMIN_CONFIG = "/usr/local/lsws/admin/conf/admin_config.xml"

    def __init__(self, login, password):
        self.login = login
        self.password = password

    @staticmethod
    def _get_litespeed_pid():
        """
        Returns pid that is stored in litespeed's pidfile
        :return: str
        """
        if os.path.isfile(LiteSpeed.PID_FILE_PATH) and os.path.isfile(LiteSpeed.HTPASSWD_PATH):
            with open(LiteSpeed.PID_FILE_PATH) as f:
                return f.readline().rstrip(os.linesep)
        else:
            return None

    @staticmethod
    def is_litespeed_running():
        """
        Checks whether pid is not None.
        :return: bool
        """
        return LiteSpeed._get_litespeed_pid() is not None

    def _get_litespeed_webadmin_port(self):
        """
        Retrives current LiteSpeed webadmin console port
        :return: LiteSpeed webadmin console port as string
        """
        try:
            # Part of Litespeed config, containing console port:
            # <?xml version="1.0" encoding="UTF-8"?>
            # <adminConfig>
            #     <listenerList>
            #         <listener>
            #             <name>adminListener</name>
            #             <address>*:7080</address>
            #             <secure>0</secure>
            #         </listener>
            #     </listenerList>
            ls_adm_cfg = etree.parse(open(self.LS_ADMIN_CONFIG)).getroot()
            data = ls_adm_cfg.xpath("listenerList/listener/address")[0]
            return data.text.split(':')[1]
        except (AttributeError, IndexError, ValueError, OSError, IOError) as e:
            raise LiteSpeedException("Can't determine current LiteSpeed webadmin console port from config {}: {}".
                                     format(self.LS_ADMIN_CONFIG, str(e)))

    def _get_requests(self):
        """
        Get info about connections from litespeed
        and returns array of rows with data
        :return: list
        :raise: [LiteSpeedInvalidCredentials, LiteSpeedDisabledException]
        """
        status_url = 'http://localhost:{}/status?rpt=details'.format(self._get_litespeed_webadmin_port())
        request = urllib.request.Request(status_url)

        base64string = base64.b64encode(b'%s:%s' % (self.login.encode(), self.password.encode()))
        request.add_header(b"Authorization", b"Basic %s" % base64string)
        # get data from litespeed, check whether http code is 200
        try:
            response = urllib.request.urlopen(request, timeout=self.HTTP_TIMEOUT,
                                              context=ssl._create_unverified_context()).read()
        except urllib.error.HTTPError as e:
            if e.code in [401, 403]:
                raise LiteSpeedInvalidCredentials("Litespeed login / password invalid. "
                                                  "Please, try restart lvestats service.")
            raise LiteSpeedDisabledException(str(e))
        except Exception as e:  # not good, but urllib raises lot of exceptions
            raise LiteSpeedDisabledException(str(e))

        # remove empty lines
        result = [row for row in response.split(os.linesep.encode()) if row.strip() != b'']
        return result

    def __is_host_valid(self, host):
        """
        Check whether host is not empty.
        :type host: str
        :return: bool
        """
        host = host.strip()
        if host and host not in self.IGNORE_HOSTS:
            return True
        return False

    def _parse_request_info(self, request: bytes):
        """
        :return: method, url, http_version
        """
        request_info = request.strip(b'"').split()
        if len(request_info) == 3:
            method, url, http_version = request_info
        elif len(request_info) == 2:
            method, url = request_info
            http_version = b''
        else:
            return None
        return method, url, http_version

    def get_user_data(self, username):
        """
        Returns information about processed by user pages.
        :param username:
        :return list[list]:
        list of the lists
        [[Pid, Domain, Http type, Path, Http version, Time],...]
        :raises: LiteSpeedDownException
        """
        data_delimiter = b'\t'

        pid = self._get_litespeed_pid()
        all_domains = get_all_user_domains(username)
        normalized_domains = set(map(normalize_domain, all_domains))

        requests = self._get_requests()
        litespeed_requests = []
        for request in requests:
            request_info = request.split(data_delimiter)
            if len(request_info) < LiteSpeedDataMapping.TOTAL_LEN:
                # that is not valid request info, skip it...
                continue

            host = request_info[LiteSpeedDataMapping.HOST]
            request = request_info[LiteSpeedDataMapping.REQUEST]

            # time since first request, seconds
            request_time = self.to_float(request_info[LiteSpeedDataMapping.TIME])

            if self.__is_host_valid(host) and \
                    normalize_domain(host.decode()) in normalized_domains:
                request_data = self._parse_request_info(request)
                if request_data is not None:
                    # pylint: disable=unpacking-non-sequence
                    method, url, http_version = request_data
                    litespeed_requests.append((pid, host, method, url, http_version, request_time))

        return litespeed_requests

    @staticmethod
    def to_float(string):
        """
        Converts str to float, if can't return -1.
        :type string: str
        :rtype: float
        """
        try:
            return float(string)
        except ValueError:
            return -1.
Name	Size	Permission
__pycache__	---	0755
__init__.py	219 bytes	0644
argparse_utils.py	11241 bytes	0644
dateutil.py	6037 bytes	0644
decorators.py	931 bytes	0644
func.py	15282 bytes	0644
htpasswd.py	2197 bytes	0644
litespeed.py	6721 bytes	0644
logsetup.py	4557 bytes	0644
proctitle.py	2878 bytes	0644
profiler.py	651 bytes	0644
progress.py	1112 bytes	0644
sentry.py	6462 bytes	0644
server_status.py	1366 bytes	0644
sizeutil.py	2762 bytes	0644
users_manager.py	3086 bytes	0644
File Content: litespeed.py

File Content: `litespeed.py`
