✏️ Editing: client.py Read Only

#!/opt/cloudlinux/venv/bin/python3 -bb
# 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 platform
import threading

from requests.exceptions import RequestException
from clcommon import cpapi, get_lve_version
from clcommon.lib.cledition import get_cl_edition_readable
from clcommon.utils import get_rhn_systemid_value, get_username
from clcommon.lib.network import get_ip_addr, get_hostname
from clcommon.cpapi.const import UNKNOWN_CP_NAME
from raven import Client
from raven.transport import RequestsHTTPTransport
from sentry_sdk.transport import HttpTransport
from clsentry.utils import get_pkg_version, SENSITIVE_FIELDS

class SafeRequestsHTTPTransport(RequestsHTTPTransport):

def send(self, url, data, headers):
        try:
            super().send(url, data, headers)
        except RequestException:
            # We hide errors while sending message to Sentry in varied cases:
            # problems with network, Sentry server is down, incorrect firewall's rules, etc
            pass

class SafeRequestsHTTPTransportSentrySdk(HttpTransport):
    def _send_event(self, event):
        try:
            super()._send_event(event)
        except RequestException:
            # We hide errors while sending message to Sentry in varied cases:
            # problems with network, Sentry server is down, incorrect firewall's rules, etc
            pass

class ThreadedHttpTransport(HttpTransport):
    def send_event(self, event):
        thread = threading.Thread(target=super().send_event, args=(event,))
        thread.daemon = True
        thread.start()

class UserlandClient(Client):
    """
    Userland's sentry client with some common settings.
    """

def __init__(self, dsn, **options):
        options.update({
            'tags': self._get_user_tags(),
            'ignore_exceptions': [
                KeyboardInterrupt,
            ],
            'exclude_paths': [
                'sentry',
                'raven',
            ],
            'auto_log_stacks': True,
            'string_max_length': 1000,
            'list_max_length': 100,
            'processors': ('clsentry.processors.UserlandSanitize',),
            # async transport is noisy and prints messages like
            # 'Sentry is attempting to send 1 pending error messages'
            'transport': SafeRequestsHTTPTransport,
        })
        super().__init__(dsn, **options)

@classmethod
    def _get_user_tags(cls):
        """
        Get tags for easy search
        :rtype: dict
        """
        return get_user_tags()

def get_user_tags():
    """
    Get tags for easy search
    :rtype: dict
    """
    cp_description = cpapi.get_cp_description()
    cp_version = cp_description.get('version') if cp_description else None
    cp_name = cp_description.get('name') if cp_description else UNKNOWN_CP_NAME
    cp_product = 'WP2' if cpapi.is_wp2_environment() else None

return {
        'Control Panel Name': cp_name,
        'Control Panel Version': cp_version,
        'Control Panel Product': cp_product,
        'CloudLinux version': get_rhn_systemid_value("os_release"),
        'Cloudlinux edition': get_cl_edition_readable(),
        'Architecture': get_rhn_systemid_value("architecture"),
        'lve_version': get_lve_version()[0],
        # TODO: use 'include_paths' instead
        'lvemanager': get_pkg_version('lvemanager'),
        'alt-python27-cllib': get_pkg_version('alt-python27-cllib'),
        'lve-stats': get_pkg_version('lve-stats'),
        'lve-utils': get_pkg_version('lve-utils'),
        'cl-end-server-tools': get_pkg_version('cl-end-server-tools'),
        'cl-node-exporter': get_pkg_version('cl-node-exporter'),
        'cagefs': get_pkg_version('cagefs'),
        'kernel': platform.release(),
        'username': get_username(),
        'ip_address': get_ip_addr(get_hostname()),
        'hostname': get_hostname(),
        'system_id': get_rhn_systemid_value("system_id"),
    }

def sanitize_event(event, hint):
    def sanitize_dict(data):
        for key, value in data.items():
            if key.lower() in SENSITIVE_FIELDS:
                data[key] = '*********'
            elif isinstance(value, dict):
                sanitize_dict(value)
        return data

if "request" in event and isinstance(event["request"], dict):
        event["request"] = sanitize_dict(event["request"])

return event

def before_send(event, hint):
    exception_values = event.get("exception", {}).get("values", [])
    if exception_values:
        stacktrace = exception_values[0].get("stacktrace", {})
        frames = stacktrace.get("frames", [])
        if frames and "sentry" in frames[-1].get("module", ""):
            return None

log_record = hint.get("log_record")
    if log_record:
        fingerprint = getattr(log_record, "fingerprint", None)
        if fingerprint:
            event["fingerprint"] = [fingerprint]

return sanitize_event(event, hint)

🔐 Sid Gifari File Manager Pro