Viewing file: utils.py (15.53 KB) -rw-r--r-- Select action/file-type: (+) | (+) | (+) | Code (+) | Session (+) | (+) | SDB (+) | (+) | (+) | (+) | (+) | (+) |
# -*- 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
""" This module contains helpful utility functions for SSA Agent """ import datetime import dbm import logging import os import platform import re import xml.etree.ElementTree as ET from collections import namedtuple from contextlib import contextmanager from datetime import date, timedelta from distutils.version import LooseVersion from socket import socket, fromfd, AF_UNIX, SOCK_STREAM, AF_INET, AF_INET6, \ SOCK_DGRAM from typing import Optional, Union from urllib.parse import urlparse
import sentry_sdk from sentry_sdk.integrations.atexit import AtexitIntegration from sentry_sdk.integrations.logging import LoggingIntegration
from clcommon.const import Feature from clcommon.cpapi import ( is_panel_feature_supported, get_cp_description, is_throttling_supported, is_wp2_environment ) from clcommon.utils import get_kmodlve_module_version, get_username from clcommon.lib.cledition import get_cl_edition_readable from clcommon.lib.network import get_hostname from clcommon.utils import get_rhn_systemid_value
from .constants import sentry_dsn from .exceptions import SSAError
logger = logging.getLogger('utils') URL = namedtuple('URL', ['domain_name', 'uri_path'])
# --------- FUNCTIONS ---------
def url_split(url: str) -> URL: """ Split URL into domain_name and uripath including query string :param url: URL of format protocol://domain/path;parameters?query#fragment :return: namedtuple URL(domain_name, uripath) """ fragments = urlparse(url) qs = f'?{fragments.query}' if fragments.query else '' uri = f'{fragments.path}{qs}' if fragments.path else '/' # logger.info('Parsed %s into %s:%s', url, fragments.netloc, uri) return URL(fragments.netloc.replace('www.', ''), uri)
def ssa_version() -> Optional[str]: """Get version of alt-php-ssa package""" return pkg_version('/usr/share/clos_ssa/version')
def xray_version() -> Optional[str]: """Get version of alt-php-xray package""" return pkg_version('/usr/share/alt-php-xray/version')
def sentry_init() -> None: """ Initialize Sentry client shutdown_timeout=0 disables Atexit integration as stated in docs: 'it’s easier to disable it by setting the shutdown_timeout to 0' https://docs.sentry.io/platforms/python/default-integrations/#atexit On the other hand, docs say, that 'Setting this value too low will most likely cause problems for sending events from command line applications' https://docs.sentry.io/error-reporting/configuration/?platform=python#shutdown-timeout """
def add_info(event: dict, hint: dict) -> dict: """ Add extra data into sentry event :param event: original event :param hint: additional data caught :return: updated event """ event['extra'].update({'ssa.version': '0.4-15.el8'}) return event
def try_get_ip(address_family, private_ip): """ address_family - we can choose constants represent the address (and protocol) families (AF_INET for ipv4 and AF_INET6 for ipv6) private_ip - specify some private ip address. For instance: ipv4 -> 10.255.255.255 or ipv6 -> fc00:: """ try: with socket(address_family, SOCK_DGRAM) as s: s.connect((private_ip, 1)) return s.getsockname()[0] except Exception: logger.info('Cannot retrieve IP address')
def get_ip(): """ We are trying to get an IPv4 or IPv6 address. In case of failure we'll return 127.0.0.1 """ ipversions = (AF_INET, '10.255.255.255'), (AF_INET6, 'fc00::') for addr_fam, priv_ip in ipversions: ip = try_get_ip(addr_fam, priv_ip) if ip: return ip return '127.0.0.1'
def set_tags(sentry_scope): cp_description = get_cp_description() cp_version = cp_description.get('version') if cp_description else None cp_name = cp_description.get('name') if cp_description else None cp_product = 'WP2' if is_wp2_environment() else None
tags = (('alt-php-xray', xray_version() or 'UNKNOWN'), ('Control Panel Name', cp_name), ('Control Panel Version', cp_version), ('Control Panel Product', cp_product), ('kernel', platform.release()), ('CloudLinux version', get_rhn_systemid_value("os_release")), ('Cloudlinux edition', get_cl_edition_readable()), ('Architecture', get_rhn_systemid_value("architecture")), ('ip_address', get_ip()), ('username', get_username()) ) # set_tags does not work in current version of sentry_sdk # https://github.com/getsentry/sentry-python/issues/1344 for tag in tags: sentry_scope.set_tag(*tag)
def nope(pending, timeout) -> None: pass
ssa_ver = ssa_version() or "alt-php-ssa@0.4-15.el8"
sentry_logging = LoggingIntegration(level=logging.INFO, event_level=logging.WARNING) silent_atexit = AtexitIntegration(callback=nope) sentry_sdk.init(dsn=sentry_dsn, before_send=add_info, release=ssa_ver, max_value_length=10000, integrations=[sentry_logging, silent_atexit]) with sentry_sdk.configure_scope() as scope: scope.user = { "id": get_rhn_systemid_value("system_id") or get_ip() or get_hostname() or get_username() } try: set_tags(scope) except Exception: pass
def set_logging_into_file(fname: str, as_error: bool = False) -> str: """ Try to configure logging into given fname If as_error True, log the exception as ERROR, otherwise -- as INFO """ try: logging.basicConfig(filename=fname, level=logging.INFO, format='%(asctime)s %(message)s', datefmt='%m/%d/%Y %I:%M:%S %p') try: os.chmod(fname, 0o666) except PermissionError: pass return fname except OSError as e: logger.log(logging.ERROR if as_error else logging.INFO, 'No logging configuration applied: %s', str(e))
def configure_logging(logname: str) -> str: """ Configure logging :param logname: path to log :return: logpath """ sentry_init() if set_logging_into_file(logname) is None: try: os.makedirs(os.path.dirname(logname)) except Exception as e: logger.warning('Failed to create logdir %s', str(e)) return '' logname = set_logging_into_file(logname, as_error=True) return logname
def create_socket(sock_location: str) -> 'socket object': """ Create world-writable socket in given sock_location or reuse existing one :param sock_location: socket address :return: socket object """ LISTEN_FDS = int(os.environ.get("LISTEN_FDS", 0)) if LISTEN_FDS == 0: with umask_0(): sockobj = socket(AF_UNIX) sockobj.bind(sock_location) sockobj.listen() else: sockobj = fromfd(3, AF_UNIX, SOCK_STREAM) sockobj.listen() return sockobj
def previous_day_date() -> str: """ Returns date of previous day in a format "day.month.year" """ yesterday = date.today() - timedelta(days=1) return yesterday.strftime('%d.%m.%Y')
def format_date(datestr: str, formatstr='%d.%m.%Y') -> str: """ Convert date to format YYYY-mm-dd """ _date = datetime.datetime.strptime(datestr, formatstr) return _date.strftime("%Y-%m-%d")
def read_sys_id() -> str: """ Obtain system ID from /etc/sysconfig/rhn/systemid :return: system ID without ID- prefix """ try: tree = ET.parse('/etc/sysconfig/rhn/systemid') root = tree.getroot() whole_id = root.find(".//member[name='system_id']/value/string").text with sentry_sdk.configure_scope() as scope: scope.set_tag("system_id", whole_id) return whole_id.lstrip('ID-') except (OSError, ET.ParseError) as e: logger.warning('Failed to retrieve system_id: %s', str(e))
def duration_cast(duration: int) -> float: """ Cast duration from microseconds to seconds leaving 2 digits after point """ return float(format(duration/1000000, '0.2f'))
def pkg_version(filepath: str) -> Optional[str]: """Get version of package from file. alt-php-ssa/alt-php-xray supported""" try: with open(filepath) as v_file: version = v_file.read().strip() except OSError: return # remove dist suffix return '.'.join(version.split('.')[:2]) or '0.0-0'
def is_xray_version_supported() -> bool: """ Check version of alt-php-xray package. Autotracing in X-Ray is supported since 0.4-1 """ version_number = xray_version() if version_number is None: # no xray installed return False try: return LooseVersion(version_number) >= LooseVersion('0.4-1') except (TypeError, AttributeError): return False
def is_xray_user_agent_active() -> bool: """Check if User Agent is listening""" user_agent_sock = '/opt/alt/php-xray/run/xray-user.sock' with socket(AF_UNIX, SOCK_STREAM) as s: try: s.connect(user_agent_sock) except (ConnectionError, OSError): return False return True
def no_xray_active_tasks() -> bool: """Check if there are no active X-Ray tasks (== empty task storage)""" xray_tasks_storage = '/usr/share/alt-php-xray/tasks' if not os.path.isfile(xray_tasks_storage): # no tasks file usually means no X-Ray at all, thus no tasks return True
try: with dbm.open(xray_tasks_storage, 'c') as xray_tasks: return len(xray_tasks.keys()) == 0 except dbm.error: # in case we somehow failed to open the file, but it exists, # suppose there are active tasks return False
def switch_schedstats(enabled: bool) -> None: """ Switch on/off throttle statistics gathering by kmodlve :param enabled: True or False """ if not is_panel_feature_supported(Feature.LVE): # do nothing if there is no LVE feature return
logger.info('Switching schedstats: %s', enabled) try: with open('/proc/sys/kernel/sched_schedstats', mode='wb', buffering=0) as f: f.write(b'1' if enabled else b'0') logger.info('Done OK') except OSError as e: logger.info('Failed to set sched_schedstats to %s: %s', enabled, str(e))
def is_io_throttling_detection_available() -> bool: """ Check kmodlve module version or kernel version in order to determine if it provides the detection of IO throttling """ kmod_min_version = LooseVersion('2.0-23') kernel_min_version = LooseVersion('1.5.58') kmod_current = get_kmodlve_module_version() if kmod_current: return compare_versions_ge(kmod_current, kmod_min_version) return compare_versions_ge(extract_kernel_version(), kernel_min_version)
def extract_kernel_version() -> str: """ Get kernel version in the form of "major-minor" from current platform """ pattern = re.compile(r'lve([0-9]+(\.[0-9]+)+)\.el') release = platform.release() try: version_num = pattern.search(release).group(1) except AttributeError: return '0.0.0' return version_num
def compare_versions_ge(first_version: str, second_version: Union[str, LooseVersion]) -> bool: """ Comparing two versions using the greater or equal operator. """ try: return LooseVersion(first_version) >= second_version except TypeError as e: raise SSAError( f'Unable to compare required versions: unexpected versions format "{first_version}" and "{second_version}"')
def is_kernel_version_supported(): """ General check of kernel support (IO throttling availability is required) """ if not is_throttling_supported(): # no throttling at all return True return is_io_throttling_detection_available()
# --------- CONTEXT MANAGERS ---------
@contextmanager def umask_0(mask: int = 0) -> None: """ Context manager for dropping umask """ prev = os.umask(mask) yield os.umask(prev)
@contextmanager def set_privileges(target_uid: int = None, target_gid: int = None, target_path='.', mask: int = None, with_check=True) -> None: """ Context manager to drop privileges during some operation and then restore them back. If target_uid or target_gid are given, use input values. Otherwise, stat target_uid and target_gid from given target_path. If no target_path given, use current directory. Use mask if given. :param target_uid: uid to set :param target_gid: gid to set :param target_path: directory or file to stat for privileges, default -- current directory :param mask: umask to use :param with_check: check the result of switching privileges """ prev_uid = os.getuid() prev_gid = os.getgid()
try: stat_info = os.stat(target_path) except OSError: stat_info = None
if target_uid is None: if stat_info is None: target_uid = prev_uid else: target_uid = stat_info.st_uid if target_gid is None: if stat_info is None: target_gid = prev_gid else: target_gid = stat_info.st_gid
if mask is not None: prev = os.umask(mask)
if prev_gid != target_gid: os.setegid(target_gid) logger.debug('Dropped GID privs to %s', target_gid) if with_check and os.getegid() != target_gid: # break operation if privileges dropping failed raise SSAError( 'Unable to execute required operation: permission issue') if prev_uid != target_uid: os.seteuid(target_uid) logger.debug('Dropped UID privs to %s', target_uid) if with_check and os.geteuid() != target_uid: if prev_gid != target_gid: # check if GID should be restored os.setegid(prev_gid) # break operation if privileges dropping failed raise SSAError( 'Unable to execute required operation: permission issue') yield if prev_uid != target_uid: os.seteuid(prev_uid) logger.debug('Restored UID privs to %s', prev_uid) if prev_gid != target_gid: os.setegid(prev_gid) logger.debug('Restored GID privs to %s', prev_gid)
if mask is not None: os.umask(prev)
# --------- DECORATORS ---------
# for easy mocking in tests where # we don't need singleton IS_SINGLETON_ENABLED = True
def singleton(some_cls): class __Singleton: """ A singleton wrapper class. Its instances would be created for each decorated class. """
def __init__(self, _cls): self._wrapped = _cls self._instance = None
def __call__(self, *args, **kwargs): """Returns a single instance of decorated class""" if self._instance is None or not IS_SINGLETON_ENABLED: self._instance = self._wrapped(*args, **kwargs) return self._instance
return __Singleton(some_cls)
|