← Back
Editing: 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 import base64 import logging import urllib.request import urllib.error import urllib.parse import ssl import os from lxml import etree from lvestats.lib.commons.fileutil import open_nofollow from lvestats.lib.commons.func import get_all_user_domains, normalize_domain log = logging.getLogger('litespeed') 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 _open_nofollow(path): """Thin wrapper around :func:`fileutil.open_nofollow` kept for backward-compatible mocking in tests.""" return open_nofollow(path) @staticmethod def _read_pid_file(): """ Read and validate PID from litespeed's pidfile. :return: str pid value :raises: OSError, ValueError on any validation failure """ fd = LiteSpeed._open_nofollow(LiteSpeed.PID_FILE_PATH) try: if os.fstat(fd).st_uid != 0: raise ValueError("pid file not owned by root") with os.fdopen(fd, encoding='utf-8') as f: fd = -1 pid_str = f.readline().rstrip(os.linesep) finally: if fd != -1: os.close(fd) pid = int(pid_str) if pid <= 0: raise ValueError("invalid pid value") if os.stat(f'/proc/{pid}/status').st_uid != 0: raise ValueError("process not owned by root") return pid_str @staticmethod def _get_litespeed_pid(): """ Returns pid that is stored in litespeed's pidfile :return: str """ if not os.path.isfile(LiteSpeed.PID_FILE_PATH) or not os.path.isfile(LiteSpeed.HTPASSWD_PATH): return None try: return LiteSpeed._read_pid_file() except (OSError, ValueError): 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> with open(self.LS_ADMIN_CONFIG, 'r', encoding='utf-8') as f: ls_adm_cfg = etree.parse(f).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 " f"port from config {self.LS_ADMIN_CONFIG}: {e}" ) from e def _get_requests(self): """ Get info about connections from litespeed and returns array of rows with data :return: list :raise: [LiteSpeedInvalidCredentials, LiteSpeedDisabledException] """ port = self._get_litespeed_webadmin_port() # Try HTTP first for compatibility with older LiteSpeed versions # that may serve the admin console on plain HTTP. If HTTP fails, # retry with HTTPS. LiteSpeed 6.x forces HTTPS even when # <secure>0</secure> is set in admin_config.xml — HTTP requests # get 301-redirected to HTTPS, and urllib strips the Authorization # header on the cross-scheme redirect, resulting in a 401. So we # treat both connection errors AND auth errors on HTTP as reasons # to retry with HTTPS. data = self._fetch_status(f'http://localhost:{port}/status?rpt=details', raise_on_auth_error=False) if data is None: data = self._fetch_status(f'https://localhost:{port}/status?rpt=details', raise_on_auth_error=True) if data is None: raise LiteSpeedDisabledException( "Cannot connect to LiteSpeed WebAdmin on either HTTP or HTTPS" ) # remove empty lines result = [row for row in data.split(os.linesep.encode()) if row.strip() != b''] return result def _fetch_status(self, status_url, raise_on_auth_error=True): """ Fetch LiteSpeed status page at the given URL. :param raise_on_auth_error: if True, raise LiteSpeedInvalidCredentials on 401/403; if False, return None (allows caller to retry with a different URL scheme) :return: response bytes on success, None on failure :raise: LiteSpeedInvalidCredentials on 401/403 (when raise_on_auth_error=True) :raise: LiteSpeedDisabledException on non-auth HTTP errors """ 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) try: context = ssl._create_unverified_context() # pylint: disable=protected-access with urllib.request.urlopen( request, timeout=self.HTTP_TIMEOUT, context=context, ) as response: return response.read() except urllib.error.HTTPError as e: if e.code in [401, 403]: if raise_on_auth_error: raise LiteSpeedInvalidCredentials( "Litespeed login / password invalid. " "Please, try restart lvestats service." ) from e log.debug('Auth error on %s (HTTP %s), will retry', status_url, e.code) return None # Non-auth HTTP errors (4xx, 5xx) are server-side responses — # the server received and processed the request. Retrying with # a different URL scheme won't help, so raise immediately. raise LiteSpeedDisabledException(str(e)) from e except (urllib.error.URLError, OSError) as e: # Connection refused, SSL handshake failure, timeout, socket error. # Return None so the caller can retry with a different URL scheme. log.debug('Connection error on %s: %s, will retry', status_url, e) return None 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: 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.
Save File
Cancel