golden hour

/opt/imunify360/venv/versions/imunify-core-8.14.0-1/defence360agent/utils

File/Folder	Size	Actions
__init__.py	68.07 KB
__pycache__	-
_shutil.py	3.92 KB
antivirus_mode.py	497 B
async_utils.py	718 B
benchmark.py	538 B
buffer.py	1.9 KB
check_db.py	7.8 KB
check_lock.py	856 B
cli.py	9.06 KB
common.py	14.41 KB
completions.py	9.75 KB
config.py	1.66 KB
cronjob.py	902 B
doctor.py	5.37 KB
fd_ops.py	7.4 KB
hyperscan.py	149 B
importer.py	2.66 KB
ipecho.py	3.17 KB
json.py	953 B
kwconfig.py	1.71 KB
net.py	595 B
net_transport.py	15.91 KB
parsers.py	11.6 KB
resource_limits.py	5.04 KB
safe_fileops.py	13.35 KB
safe_sequence.py	363 B
serialization.py	2.35 KB
sshutil.py	14.82 KB
subprocess.py	1.53 KB
support.py	5.2 KB
tenant_path.py	4.35 KB
threads.py	1005 B
validate.py	4.27 KB
whmcs.py	7.6 KB
wordpress_mu_plugin.py	1.41 KB
zipsafe.py	720 B

Edit: net_transport.py

"""Networking transport helpers for urllib.

This module provides a small abstraction on top of urllib.request so that
callers can keep using urllib.request.Request, but routing of connections
can be customized:

- hostname resolution is handled in user code;
- selected IP may be randomized or chosen using any complex logic;
- for HTTPS: connects to a chosen IP but keeps correct SNI and certificate
  hostname validation for the original hostname (NOT the IP).

Examples:

Default behavior (plain urllib):

from defence360agent.utils.net_transport import UrlTransport

transport = UrlTransport()
    req = urllib.request.Request(
        "https://files.imunify360.com/static/sigs/v1/description.json"
    )
    with transport.open(req, timeout=10) as resp:
        body = resp.read()

Randomize target IP on each connection (A/AAAA -> random choice):

from defence360agent.utils.net_transport import UrlTransport, RandomIpChooser

chooser = RandomIpChooser()
    transport = UrlTransport(ip_chooser=chooser)

req = urllib.request.Request(
        "https://files.imunify360.com/static/sigs/v1/description.json"
    )
    with transport.open(req, timeout=10) as resp:
        body = resp.read()

Notes:

- HTTPS: connects to the chosen IP but keeps SNI/cert checks against original
  hostname.
- HTTP: Host header stays original hostname because urllib builds it from the
  URL.

"""

import http.client
import ipaddress
import pathlib
import random
import socket
import threading
import time
import urllib.request
from abc import ABC, abstractmethod
from logging import getLogger
from typing import Dict, Optional, Tuple, TYPE_CHECKING

if TYPE_CHECKING:
    import ssl

logger = getLogger(__name__)

#: default cache TTL for DNS responses
_DNS_DEFAULT_TTL_SECONDS = 300.0

_IPV6_DISABLED_STATE = pathlib.Path("/var/imunify360/.ipv6_disabled")
_SYSCTL_DISABLE_IPV6 = "/proc/sys/net/ipv6/conf/all/disable_ipv6"
_MOD_PAR_PATH = "/sys/module/{mod}/parameters/{parameter}"

def _is_ipv4(ip: str) -> bool:
    """Return True if *ip* is an IPv4 address string.

Implementation relies solely on ipaddress.ip_address for correctness.
    """
    try:
        return isinstance(ipaddress.ip_address(ip), ipaddress.IPv4Address)
    except ValueError:
        return False

class IpChooser(ABC):
    """Select an IP address to connect to for a given hostname and port.

Implementations may be stateful and can keep caches/metrics inside.
    """

@abstractmethod
    def choose(self, hostname: str, port: int) -> str:
        """Return an IP address (v4 or v6) for *hostname*:*port*."""
        raise NotImplementedError

def __call__(self, hostname: str, port: int) -> str:
        return self.choose(hostname, port)

class DnsCacheResolver:
    """DNS cache for socket.getaddrinfo() results.

It caches per (hostname, port, family). This is intentionally small and
    local: it is meant only to avoid excessive getaddrinfo() calls.
    """

def __init__(
        self,
        *,
        family: int = socket.AF_UNSPEC,
        ttl_seconds: float = _DNS_DEFAULT_TTL_SECONDS,
    ):
        self._family = family
        self._ttl_seconds = ttl_seconds
        self._cache: Dict[
            Tuple[str, int, int], Tuple[float, Tuple[str, ...]]
        ] = {}
        self._lock = threading.Lock()

def get_ips(self, hostname: str, port: int) -> Tuple[str, ...]:
        key = (hostname, port, self._family)
        now = time.time()

with self._lock:
            cached = self._cache.get(key)
            if cached is not None:
                expires_at, ips = cached
                if now < expires_at:
                    logger.debug(
                        "DnsCacheResolver cache hit for %s:%s (family=%s)",
                        hostname,
                        port,
                        self._family,
                    )
                    return ips

logger.debug(
            "DnsCacheResolver cache miss/expired for %s:%s (family=%s)",
            hostname,
            port,
            self._family,
        )

infos = socket.getaddrinfo(
            hostname,
            port,
            self._family,
            socket.SOCK_STREAM,
        )

ips = []
        for _, _, _, _, sockaddr in infos:
            ip = sockaddr[0]
            if ip not in ips:
                ips.append(ip)

if not ips:
            raise OSError("No IPs resolved for {}:{}".format(hostname, port))

ips_t = tuple(ips)
        with self._lock:
            self._cache[key] = (now + self._ttl_seconds, ips_t)

logger.debug(
            "DnsCacheResolver resolved %s:%s (family=%s) to %s",
            hostname,
            port,
            self._family,
            ips_t,
        )
        return ips_t

class RandomIpChooserWithIPv6Toggle(IpChooser):
    """Resolve hostname and select a random IP.

IPv6 selection can be enabled/disabled at runtime:
    - when IPv6 is enabled: choose from IPv4 + IPv6 candidates
    - when IPv6 is disabled: choose from IPv4-only candidates
    """

def __init__(
        self,
        *,
        resolver: Optional[DnsCacheResolver] = None,
        rng: Optional[random.Random] = None,
        ipv6_enabled: bool = True,
    ):
        self._resolver = resolver or DnsCacheResolver()
        self._rng = rng or random.Random()
        self._ipv6_enabled = ipv6_enabled
        self._last_ip: Optional[str] = None

def enable_ipv6(self) -> None:
        self._ipv6_enabled = True

def disable_ipv6(self) -> None:
        self._ipv6_enabled = False

def is_ipv6_enabled(self) -> bool:
        return self._ipv6_enabled

def last_ip(self) -> Optional[str]:
        return self._last_ip

def choose(self, hostname: str, port: int) -> str:
        ips = self._resolver.get_ips(hostname, port)
        if self._ipv6_enabled:
            chosen = self._rng.choice(ips)
            self._last_ip = chosen
            logger.debug(
                "RandomIpChooserWithIPv6Toggle selected IP %s for %s:%s "
                "(IPv6 enabled)",
                chosen,
                hostname,
                port,
            )
            return chosen

ipv4_ips = tuple(ip for ip in ips if _is_ipv4(ip))
        if not ipv4_ips:
            raise OSError(
                "No IPv4 IPs resolved for {}:{}".format(hostname, port)
            )

chosen = self._rng.choice(ipv4_ips)
        self._last_ip = chosen
        logger.debug(
            "RandomIpChooserWithIPv6Toggle selected IPv4 IP %s for %s:%s "
            "(IPv6 disabled)",
            chosen,
            hostname,
            port,
        )
        return chosen

class RandomIpChooser(IpChooser):
    """Resolve hostname and select a random IP from resolved candidates."""

def __init__(
        self,
        *,
        resolver: Optional[DnsCacheResolver] = None,
        rng: Optional[random.Random] = None,
    ):
        self._resolver = resolver or DnsCacheResolver()
        self._rng = rng or random.Random()

def choose(self, hostname: str, port: int) -> str:
        ips = self._resolver.get_ips(hostname, port)
        chosen = self._rng.choice(ips)
        logger.debug(
            "RandomIpChooser selected IP %s for %s:%s",
            chosen,
            hostname,
            port,
        )
        return chosen

class ForcedIPHTTPConnection(http.client.HTTPConnection):
    """HTTPConnection that connects to a chosen IP.

Important: urllib builds the request URL with the original hostname,
    therefore the Host header stays correct.
    """

def __init__(
        self,
        hostname: str,
        port: Optional[int] = None,
        *,
        ip_chooser: IpChooser,
        timeout=socket._GLOBAL_DEFAULT_TIMEOUT,
        source_address=None,
    ):
        super().__init__(
            hostname,
            port=port,
            timeout=timeout,
            source_address=source_address,
        )
        self._ip_chooser = ip_chooser

def connect(self) -> None:
        # not "or 80": http.client always resolves a port, and a falsy
        # one (:0) must stay the port the caller asked for
        port = self.port
        ip = self._ip_chooser.choose(self.host, port)
        logger.debug(
            "ForcedIPHTTPConnection connecting to %s:%s for hostname %s",
            ip,
            port,
            self.host,
        )

self.sock = _connect_to(
            self._ip_chooser, ip, port, self.timeout, self.source_address
        )

class ForcedIPHTTPSConnection(http.client.HTTPSConnection):
    """HTTPSConnection that connects to a chosen IP.

TLS details:
    - Uses original hostname for SNI (server_hostname in wrap_socket), or
      the tunnelled hostname when going through a proxy
    - Certificate hostname validation is performed for that same hostname
    """

def __init__(
        self,
        hostname: str,
        port: Optional[int] = None,
        *,
        ip_chooser: IpChooser,
        context: "ssl.SSLContext",
        timeout=socket._GLOBAL_DEFAULT_TIMEOUT,
        source_address=None,
    ):
        super().__init__(
            hostname,
            port=port,
            context=context,
            timeout=timeout,
            source_address=source_address,
        )
        self._ip_chooser = ip_chooser

def connect(self) -> None:
        port = self.port
        ip = self._ip_chooser.choose(self.host, port)
        logger.debug(
            "ForcedIPHTTPSConnection connecting to %s:%s for hostname %s",
            ip,
            port,
            self.host,
        )

raw_sock = _connect_to(
            self._ip_chooser, ip, port, self.timeout, self.source_address
        )

if self._tunnel_host:
            self.sock = raw_sock
            self._tunnel()
            raw_sock = self.sock

# through a CONNECT tunnel the certificate belongs to the origin
        # host, not to the proxy we opened the socket to
        self.sock = self._context.wrap_socket(
            raw_sock,
            server_hostname=self._tunnel_host or self.host,
        )

class ForcedIPHTTPHandler(urllib.request.HTTPHandler):
    """urllib handler that creates ForcedIPHTTPConnection."""

def __init__(self, *, ip_chooser: IpChooser):
        super().__init__()
        self._ip_chooser = ip_chooser

def http_open(self, req) -> http.client.HTTPResponse:
        def factory(host, **kwargs):
            return ForcedIPHTTPConnection(
                host,
                ip_chooser=self._ip_chooser,
                timeout=kwargs.get("timeout"),
            )

return self.do_open(factory, req)

class ForcedIPHTTPSHandler(urllib.request.HTTPSHandler):
    """urllib handler that creates ForcedIPHTTPSConnection."""

def __init__(self, *, ip_chooser: IpChooser, context: "ssl.SSLContext"):
        super().__init__(context=context)
        self._ip_chooser = ip_chooser
        self._context = context

def https_open(self, req) -> http.client.HTTPResponse:
        def factory(host, **kwargs):
            return ForcedIPHTTPSConnection(
                host,
                ip_chooser=self._ip_chooser,
                context=self._context,
                timeout=kwargs.get("timeout"),
            )

return self.do_open(factory, req)

class UrlTransport:
    """Single entrypoint for opening urllib requests.

If *ip_chooser* is provided, the transport will connect to the selected IP
    address, while keeping correct Host/SNI/cert validation for the original
    hostname. If *ip_chooser* is not provided, it behaves like plain urllib.
    """

def __init__(
        self,
        *,
        ip_chooser: Optional[IpChooser] = None,
        ssl_context: Optional["ssl.SSLContext"] = None,
        use_proxies: bool = True,
    ):
        import ssl as _ssl

self._ssl_context = ssl_context or _ssl.create_default_context()

handlers: list = []
        if not use_proxies:
            # empty mapping disables urllib's default env-based proxy
            handlers.append(urllib.request.ProxyHandler({}))
        if ip_chooser is not None:
            handlers += [
                ForcedIPHTTPHandler(ip_chooser=ip_chooser),
                ForcedIPHTTPSHandler(
                    ip_chooser=ip_chooser,
                    context=self._ssl_context,
                ),
            ]
        self._opener = urllib.request.build_opener(*handlers)

def open(
        self,
        req: urllib.request.Request,
        *,
        timeout: Optional[float] = None,
    ) -> http.client.HTTPResponse:
        if timeout is None:
            return self._opener.open(req)
        return self._opener.open(req, timeout=timeout)

def _is_kernel_ipv6_disabled() -> bool:
    """Check whether IPv6 is disabled at the kernel level.

Reads the module parameter and the runtime sysctl without
    depending on the im360 package.
    """
    param_file = _MOD_PAR_PATH.format(mod="ipv6", parameter="disable")
    try:
        with open(param_file) as f:
            if f.read().strip() != "0":
                return True
    except OSError:
        # ipv6 module is absent
        return True

try:
        with open(_SYSCTL_DISABLE_IPV6) as f:
            if f.read().strip() == "1":
                return True
    except OSError:
        pass

return False

def _should_disable_ipv6() -> bool:
    """True when either the kernel has disabled IPv6 or a previous agent
    run persisted the disabled state after a runtime network failure."""
    return _is_kernel_ipv6_disabled() or _IPV6_DISABLED_STATE.exists()

def _persist_ipv6_disabled() -> None:
    """Persist IPv6 disabled state so it survives agent restarts."""
    try:
        _IPV6_DISABLED_STATE.touch()
    except OSError:
        logger.debug("Could not persist IPv6 disabled state", exc_info=True)

# lazy-initialized on first use to keep the SSL CA store (~1.5-2 MB)
# out of import time
_IP_CHOOSER: Optional[RandomIpChooserWithIPv6Toggle] = None
_TRANSPORT: Optional[UrlTransport] = None

_INIT_LOCK = threading.Lock()

def _shared_ip_chooser() -> RandomIpChooserWithIPv6Toggle:
    """The agent's single address-family decision.

One instance for every outbound subsystem on purpose: broken IPv6
    transit is a property of the route, not of the endpoint, so a
    subsystem that has not tried yet should not have to stall to learn
    what another one already found out.
    """
    global _IP_CHOOSER
    with _INIT_LOCK:
        if _IP_CHOOSER is None:
            ipv6_enabled = not _should_disable_ipv6()
            _IP_CHOOSER = RandomIpChooserWithIPv6Toggle(
                ipv6_enabled=ipv6_enabled
            )
            if not ipv6_enabled:
                logger.info(
                    "IPv6 disabled at startup (kernel flag or persisted state)"
                )
        return _IP_CHOOSER

def shared_transport() -> UrlTransport:
    global _TRANSPORT
    # resolved before the lock is taken: _shared_ip_chooser() takes the
    # same non-reentrant lock
    chooser = _shared_ip_chooser()
    with _INIT_LOCK:
        if _TRANSPORT is None:
            _TRANSPORT = UrlTransport(ip_chooser=chooser)
        return _TRANSPORT

def _connect_to(
    ip_chooser: IpChooser,
    ip: str,
    port: int,
    timeout,
    source_address,
) -> socket.socket:
    """Open a socket to *ip*, judging the address family on failure.

The verdict has to be reached here: this is the only point that knows
    both which address was tried and that it did not answer. Deriving it
    afterwards from state on the chooser cannot work, because the chooser
    is shared and a concurrent request overwrites that state.
    """
    try:
        return socket.create_connection((ip, port), timeout, source_address)
    except OSError as e:
        if isinstance(ip_chooser, RandomIpChooserWithIPv6Toggle) and (
            ip_chooser.is_ipv6_enabled() and not _is_ipv4(ip)
        ):
            logger.warning(
                "Connect to IPv6 address %r failed, disabling IPv6. Error: %r",
                ip,
                e,
            )
            ip_chooser.disable_ipv6()
            _persist_ipv6_disabled()
        raise