0x1998 - MANAGER

Düzenlenen Dosya: remoteip.py

import ipaddress
import logging
from typing import Optional

from defence360agent.subsys.panels.base import PanelException

from defence360agent.subsys import web_server
from .panels.base import RemoteIPInterface
from .panels.hosting_panel import HostingPanel

logger = logging.getLogger(__name__)

class RemoteIPError(Exception):
    pass

class LiteSpeedRemoteIP(RemoteIPInterface):
    async def remoteip_activated(self) -> bool:
        content = open(web_server.LITESPEED_CONF_PATH, "rb").read()
        config = web_server.LiteSpeedConfig(content)
        proxyIP = config.client_ip_in_header()
        allowed = config.access_control_allowed_list()
        if proxyIP == config.CLIENT_IP_IN_HEADER_DISABLED:
            return False
        if proxyIP == config.CLIENT_IP_IN_HEADER_ENABLED:
            return True
        if proxyIP == config.CLIENT_IP_IN_HEADER_TRUSTED_IP_ONLY:
            localhost = ipaddress.IPv4Address("127.0.0.1")
            for item, trusted in allowed:
                if trusted:
                    if item == "ALL":
                        return True
                    try:
                        item_network = ipaddress.ip_network(item)
                    except ValueError:
                        continue
                    else:
                        if localhost in item_network:  # NOSONAR pylint:S134
                            return True
            return False
        raise PanelException("Unknown value for client ip in header option")

async def remoteip_install(self) -> Optional[str]:
        content = open(web_server.LITESPEED_CONF_PATH, "rb").read()
        config = web_server.LiteSpeedConfig(content)
        allowed = config.access_control_allowed_list()
        config.set_client_ip_in_header(
            config.CLIENT_IP_IN_HEADER_TRUSTED_IP_ONLY
        )
        allowed.add(("127.0.0.1", True))
        config.set_access_control_allowed_list(allowed)
        with open(web_server.LITESPEED_CONF_PATH, "wb") as config_file:
            config_file.write(config.tostring())
        await web_server.graceful_restart()
        return None

class ModRemoteIp:
    """Manages mod_remoteip or similar functionality"""

NAME = "mod_remoteip"

def __init__(self):
        self.__interface = None
        self._panel = HostingPanel()
        if self._panel.remoteip_supported():
            if web_server.litespeed_running():
                self.__interface = LiteSpeedRemoteIP()
            else:
                self.__interface = self._panel

def is_supported(self):
        return self.__interface is not None

async def is_installed(self) -> bool:
        try:
            if not self.is_supported():
                return False
            return await self.__interface.remoteip_activated()
        except PanelException as exc:
            raise RemoteIPError from exc

async def install(self):
        try:
            if self.is_supported():
                return await self.__interface.remoteip_install()
            else:
                logger.warning(
                    "skipping installation of remoteip, panel not"
                    " supported %s",
                    self._panel.NAME,
                )
        except PanelException as exc:
            raise RemoteIPError() from exc

geri dön