socket_utils.py

#!/opt/cloudlinux/venv/bin/python3 -bb
# 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/LICENCE.TXT
#

from __future__ import absolute_import

import json
import logging
import socket
import struct
import time
from typing import Optional

logger = logging.getLogger(__name__)

# uint32_t, big endian
_format = '>I'

# Socket read timeout, seconds
_WPOS_SOCKET_READ_TIMEOUT_SEC = 10

# Maximum allowed message size, bytes
_MAX_MSG_SIZE = 1 * 1024 * 1024

def get_uid_from_socket(sock_object: socket.socket) -> int:
    """
    Retrieve credentials from SO_PEERCRED option
    :param sock_object: Socket object
    :return: uid of user, which connects to this socket.
    """
    _format_string = '3I'
    creds = sock_object.getsockopt(socket.SOL_SOCKET, socket.SO_PEERCRED, struct.calcsize(_format_string))
    # creds contains _pid, _uid, _gid - 3 uint32_t numbers
    _, _uid, _ = struct.unpack(_format_string, creds)
    return _uid

def pack_data_for_socket(data_dict: dict) -> bytes:
    """
    Prefix message with a 4-byte length
    :param data_dict: Data dict for send
    :return: byte array for send to socket
    """
    msg_bytes = json.dumps(data_dict).encode('utf-8')
    # Output data format:
    # 4 bytes unsigned int, big-endian - data_len
    # data_len                         - data_bytes
    return struct.pack(_format, len(msg_bytes)) + msg_bytes

def _read_bytes_from_socket_with_timeout(sock_object: socket.socket, num_bytes: int,
                                         timeout_sec: int) -> Optional[bytes]:
    """
    Read amount data from socket
    :param sock_object: Socket object to read data from
    :param num_bytes: Bytes number to read
    :param timeout_sec: Read timeout, None - timeout expired, data not received
    """
    # Save the caller's timeout so a near-zero per-iteration budget on the
    # last recv doesn't leak into a subsequent sendall on the same socket
    # (bugbot ref1_cd536b1e on MR !38).
    prev_timeout = sock_object.gettimeout()
    deadline = time.monotonic() + timeout_sec
    msg = bytes()
    try:
        while len(msg) < num_bytes:
            remaining = deadline - time.monotonic()
            if remaining <= 0:
                return None
            sock_object.settimeout(remaining)
            chunk = sock_object.recv(num_bytes - len(msg))
            if not chunk:
                return None
            msg += chunk
    except socket.timeout:
        return None
    finally:
        sock_object.settimeout(prev_timeout)
    return msg

def read_unpack_response_from_socket_daemon(sock_object: socket.socket) -> Optional[dict]:
    """
    Read length-prefixed amount of data from socket
    :param sock_object: Socket object to read data
    :return: Data received from socket dictionary. None - socket data format error
    """
    # Socket Input data format:
    # 4 bytes unsigned int, big-endian - data_len
    # data_len                         - data_bytes
    # Get data length (4 bytes)
    raw_msglen = _read_bytes_from_socket_with_timeout(sock_object, 4, _WPOS_SOCKET_READ_TIMEOUT_SEC)
    if raw_msglen is None:
        return None
    msglen = struct.unpack(_format, raw_msglen)[0]
    if msglen > _MAX_MSG_SIZE:
        return None
    msg = _read_bytes_from_socket_with_timeout(sock_object, msglen, _WPOS_SOCKET_READ_TIMEOUT_SEC)
    if msg is None:
        return None
    return json.loads(msg.decode('utf-8'))

def read_unpack_response_from_socket_client(sock_object: socket.socket) -> Optional[dict]:
    """
    Read length-prefixed amount of data from socket
    :param sock_object: Socket object to read data
    :return: Data received from socket dictionary. None - socket data format error
    """
    # Socket Input data format:
    # 4 bytes unsigned int, big-endian - data_len
    # data_len                         - data_bytes
    try:
        raw_msglen = sock_object.recv(4)
        msglen = struct.unpack(_format, raw_msglen)[0]
        if msglen > _MAX_MSG_SIZE:
            return None
        msg = bytes()
        while len(msg) != msglen:
            msg += sock_object.recv(4096)
    except socket.timeout:
        return None
    return json.loads(msg.decode('utf-8'))

def send_dict_to_socket_connection_and_close(connection: socket.socket, data_to_send: dict):
    """
    Send a dictionary to a socket connection and close it.

This is a fire-and-forget terminal send. It never re-raises socket
    I/O errors to the caller  callers in daemon.py and
    daemon_subscription_handler.py already treat it that way (no
    try/except around the call). Delivery failures are logged at
    WARNING level only; the socket is always closed on the way out
    (best-effort).

The function does still propagate programming errors (e.g.,
    TypeError from a non-serializable dict) so they surface during
    development instead of being silently swallowed.
    """
    # Defensive: a socket that is already closed (e.g., a prior call in
    # the same handler chain failed mid-sendall and our finally closed
    # it, then daemon.py's error handler retried with the same socket
    # ref) must short-circuit. Calling settimeout on a closed FD would
    # raise OSError(EBADF). fileno() returns -1 on closed sockets, and
    # itself raises OSError on a fully-detached socket  treat both as
    # already-closed.
    try:
        if connection.fileno() < 0:
            return
    except OSError:
        return

# pack runs outside the OSError swallow zone: a TypeError here
    # indicates a non-serializable response dict (programming error).
    bytes_to_send = pack_data_for_socket(data_to_send)
    try:
        # Bound sendall  a peer that connects, sends a valid request,
        # and never reads the response must not hold a daemon thread.
        connection.settimeout(_WPOS_SOCKET_READ_TIMEOUT_SEC)
        connection.sendall(bytes_to_send)
    except OSError as exc:
        # OSError covers the realistic socket-failure modes:
        #   - BrokenPipeError, ConnectionResetError, ConnectionAbortedError
        #     (peer-side disconnects)
        #   - socket.timeout (== TimeoutError in Py3.10+, subclass of OSError)
        #     from the explicit send-side timeout
        #   - OSError(EBADF) from settimeout if the FD races to closed
        # Log and continue to close  the response is undeliverable,
        # but the daemon thread must move on.
        logger.warning('Failed to deliver response to socket peer: %s', exc)
    finally:
        # close() on an already-closed socket is normally a no-op, but
        # in pathological states (double-close after dup, kernel error)
        # it can raise. Swallow to keep the never-raises contract.
        try:
            connection.close()
        except OSError:
            pass

docroot: /home/induvedam/public_html

curdir: /opt/cloudlinux/venv/lib/python3.11/site-packages/clwpos

	.	4096
	..	12288
X	__init__.py	928
	__pycache__	4096
X	billing.py	6392
	bin	4096
X	cl_wpos_exceptions.py	4020
	cli_versions	102
X	clsop_repair.py	22674
X	constants.py	5695
X	create_user_uid_dirs.py	754
X	cron.py	2189
X	daemon.py	40729
X	daemon_base.py	2912
X	daemon_config.py	621
X	daemon_redis_lib.py	15864
X	daemon_subscription_handler.py	32735
X	data_collector_utils.py	9644
	feature_suites	102
	hooks	152
X	logsetup.py	4142
	migrations	62
	object_cache	148
	optimization_features	88
X	papi.py	10104
X	parse.py	2154
	php	77
X	redis_configuration_pid_file_cleaner.py	1037
X	report_generator.py	26597
X	scoped_cache.py	1372
X	socket_utils.py	6938
X	stats.py	12304
	user	163
X	utils.py	77973
X	whmcs_utils.py	11533
X	wp_config.py	3636
X	wp_utils.py	18807
X	wpos_admin.py	76414
X	wpos_hooks.py	8466
X	wpos_req_scanner.py	5779