import asyncio import functools import json import logging import os from abc import ABC, abstractmethod from contextvars import ContextVar from itertools import chain from pathlib import Path from typing import Dict, List, Optional, Set, Tuple from defence360agent.contracts.config import Core from defence360agent.subsys.panels.base import ( AbstractPanel, ModsecVendorsError, PanelException, ) from defence360agent.subsys.web_server import ( apache_modules, apache_running, litespeed_running, ) from defence360agent.utils import async_lru_cache, finally_happened from im360 import files from im360.contracts.config import RBL_WHITELIST_FILE, Modsec from im360.subsys.shared_disabled_rules import ( get_shared_disabled_modsec_rules_ids, ) from im360.utils import ModSecLock logger = logging.getLogger(__name__) APACHE = "apache" NGINX = "nginx" LITESPEED = "litespeed" OPENLITESPEED = "openlitespeed" # imunify360-(full|minimal)-WEBSERVER-PANEL MODSEC_NAME_TEMPLATE = "imunify360-{ruleset_suffix}-{webserver}-{panel}" class ModsecImunifyVendorNotInstalled(ModsecVendorsError): """ Raises when there is no imunify vendor installed """ pass class ModsecNotInstalledVendors(ModsecVendorsError): """ Raises when there is no vendors installed at all """ pass class _ModSecLocker: LOCK = ModSecLock() def __call__(self, coro): @functools.wraps(coro) async def wrapper(*args, **kwargs): async with self.LOCK: return await coro(*args, **kwargs) return wrapper def is_modsec_locked(): return _ModSecLocker.LOCK.locked() use_modsec_lock = _ModSecLocker() def skip_if_not_installed_modsec(func): """Skip the call if ModSecurity is not installed.""" async def wrapper(cls, *args, **kwargs): if not await cls.installed_modsec(): logger.warning("ModSecurity is not installed") return return await func(cls, *args, **kwargs) return wrapper async def _get_web_server() -> Optional[str]: async def get_web_server(): if litespeed_running(): return LITESPEED elif apache_running(): return APACHE return None result = await finally_happened(get_web_server, max_tries=15, delay=1) if result is None: logger.warning("Couldn't detect any Web Server running") return result class PanelInterface(AbstractPanel): @abstractmethod async def _get_all_admin_emails(self) -> list: pass async def get_admin_emails(self) -> list: """ Return admin contact emails """ try: return await self._get_all_admin_emails() except asyncio.CancelledError: raise except Exception as e: logger.warning( "Something went wrong while getting admin email: {}".format( str(e) ) ) return [] def http_ports(self) -> Set[int]: """ Return panel's http ports """ return set() def https_ports(self) -> Set[int]: """ Return panel's https ports """ return set() def remoteip_supported(self) -> bool: return False async def get_web_server(self) -> Optional[str]: return await _get_web_server() def get_webshield_protected_ports(self): return dict() class ModSecurityInterface(ABC): REBUILD_HTTPDCONF_CMD = None APP_BASED_EXCLUDE_CONF_NAME = "i360-app-based-excludes.conf" #: whether ModSecurity vendors are being installed installing_settings_var: ContextVar[bool] = ContextVar( "installing_settings", default=False ) @classmethod @abstractmethod async def installed_modsec(cls): """ Check if ModSecurity installed and enabled :return: bool """ pass @abstractmethod async def _install_settings(self, reload_wafd=True): pass @skip_if_not_installed_modsec async def install_settings(self, reload_wafd=True): """ Install ModSecurity vendors and patch ModSecurity config """ token = self.installing_settings_var.set(True) try: await self._install_settings(reload_wafd=reload_wafd) finally: self.installing_settings_var.reset(token) # clean cache to avoid duplicate install/update attempts await self.invalidate_installed_vendors_cache() @abstractmethod async def reset_modsec_directives(self): """ Reset ModSecurity settings to values chosen by Imunify360 """ @abstractmethod async def reset_modsec_rulesets(self): """ Reset ModSecurity rulesets to values chosen by Imunify360 """ @abstractmethod async def revert_settings(self, reload_wafd=True): """ Uninstall previously installedModSecurity vendors and revert ModSecurity config """ pass @classmethod @abstractmethod def detect_cwaf(cls): """ Detects Comodo ModSecurity Rule Set installed as Plugin :return: bool: """ pass @classmethod @abstractmethod async def modsec_vendor_list(cls) -> list: """Return a list of installed ModSecurity vendors.""" pass @classmethod @abstractmethod async def enabled_modsec_vendor_list(cls) -> list: """Return a list of enabled ModSecurity vendors.""" pass @classmethod @abstractmethod async def modsec_get_directive(cls, directive_name, default=None): """ Example: >>> modsec_interface.modsec_get_directive("SecRuleEngine") 'Off' """ @classmethod @abstractmethod async def build_vendor_file_path(cls, vendor: str, filename: str) -> Path: """Return path to a specified vendor file""" raise NotImplementedError @classmethod async def build_version_file_path(cls, vendor: str) -> Path: """Return path to Imunify360 vendor VERSION file""" return await cls.build_vendor_file_path(vendor, "VERSION") @classmethod async def get_i360_vendor_name(cls) -> str: """ Return a name of Imunify360 ModSecurity vendor. """ try: installed_vendors = await cls.modsec_vendor_list() except PanelException as e: raise ModsecVendorsError(str(e)) if not installed_vendors: raise ModsecNotInstalledVendors("No vendors installed") name = next( (v for v in installed_vendors if v.startswith(Core.PRODUCT)), None ) if name is None: raise ModsecImunifyVendorNotInstalled( "Imunify360 vendor is not installed, all vendors are :%s", " ".join(installed_vendors), ) return name @classmethod async def get_i360_vendor_version(cls) -> str: """ Return a version of the Imunify360 ModSecurity vendor. """ vendor = await cls.get_i360_vendor_name() version_file = await cls.build_version_file_path(vendor) try: with version_file.open() as f: return f.read().strip() except OSError as err: raise ModsecVendorsError( "Cannot read Imunify360 vendor version: {}".format(err) ) @classmethod async def invalidate_installed_vendors_cache(cls): if hasattr(cls.modsec_vendor_list, "cache_clear"): cls.modsec_vendor_list.cache_clear() # NOSONAR Pylint:E1101 if hasattr(cls._get_release_info_from_file, "cache_clear"): cls._get_release_info_from_file.cache_clear() # noqa NOSONAR Pylint:E1101 if hasattr(cls.enabled_modsec_vendor_list, "cache_clear"): cls.enabled_modsec_vendor_list.cache_clear() # noqa NOSONAR Pylint:E1101 @classmethod @abstractmethod async def _apply_modsec_files_update(cls): """ :param list updates: [ { "groups": [ "cpanel", "litespeed" ], ... "name": "imunify360-litespeed-meta", "url": "https://files.imunify360.com/.../meta_imunify360_litespeed.yaml" }, { "groups": [ "cpanel", "apache", "litespeed" ], ... "name": "imunify360-rules-meta", "url": "https://files.imunify360.com/.../meta_imunify360_rules.yaml" }] """ # noqa: E501 pass @classmethod async def apply_modsec_files_update(cls): await cls._apply_modsec_files_update() await cls.invalidate_installed_vendors_cache() @classmethod @abstractmethod def get_audit_log_path(cls): """ Returns path to ModSecurity audit log file :return: srt: """ pass @classmethod @abstractmethod def get_audit_logdir_path(cls): """ Returns path to ModSecurity audit log dir for concurrent mode :return: srt: """ pass @classmethod @abstractmethod def write_global_disabled_rules(cls, rule_list) -> bool: """ Disable mod_security rules on global level. Rebuild httpd conf and restart httpd server is caller method responsibility. :param list rule_list: list of rules to disable :return: True if config was changed, False otherwise """ @classmethod @abstractmethod async def sync_global_disabled_rules(cls, rule_list) -> bool: """ Disable mod_security rules on global level This method should be idempotent :param list rule_list: list of rules to disable :return: True if config was changed, False otherwise """ @classmethod @abstractmethod async def sync_disabled_rules_for_domains( cls, domain_rules_map: Dict[str, list] ): """ Disable mod_security rules on domain level for each domain specified in a map. This method should be idempotent. """ @classmethod def generate_disabled_rules_config(cls, rule_list): tpl = """ \tSecRuleRemoveById {rules_list} \tmodsecurity_rules 'SecRuleRemoveById {rules_list}' """ content = "" rules_ids = { str(id_) for id_ in chain( rule_list, get_shared_disabled_modsec_rules_ids(), ) } if rules_ids: content = tpl.format(rules_list=" ".join(sorted(rules_ids))) return content @classmethod async def get_rbl_whitelist_path(cls) -> Optional[Path]: """Return RBL whitelist path. Return None on modsec vendor errors. """ try: vendor = await cls.get_i360_vendor_name() return await cls.build_vendor_file_path(vendor, RBL_WHITELIST_FILE) except ModsecVendorsError as e: logger.warning( "Can't get RBL whitelist path. ModSecurity ruleset error: %s", e, ) return None @classmethod @async_lru_cache(maxsize=1) async def _get_release_info_from_file(cls) -> Optional[dict]: vendor = await cls.get_i360_vendor_name() release_file = await cls.build_vendor_file_path(vendor, "RELEASE") try: with release_file.open() as release_f: json_data = json.load(release_f) return json_data except (OSError, json.JSONDecodeError): return None @classmethod async def get_modsec_vendor_from_release_file(cls): release_dict = await cls._get_release_info_from_file() if release_dict: try: return "{vendor}-{modsec3}{ruleset_type}-{webserver}".format( vendor=release_dict["vendor"], ruleset_type=release_dict["ruleset_type"], webserver=release_dict["webserver"], modsec3="modsec3-" * (release_dict["modsec_version"] == 3), ).lower() except KeyError as err: logger.warning( "Release file with info about ruleset is broken:" " %s is not found", err, ) return None else: return None @classmethod def get_modsec_active_conf_files(cls) -> List[str]: pass @classmethod def get_modsec_engine_mode(cls) -> str: pass @classmethod def get_modsec_vendor_updates(cls) -> List[str]: pass @classmethod @abstractmethod def _get_conf_dir(cls): pass @classmethod def get_app_specific_waf_config(cls) -> str: """ Return rules config path if WAF Rules Set configurator is supported for this hosting panel, raise NotImplementedError otherwise """ conf_dir = cls._get_conf_dir() return os.path.join(conf_dir, cls.APP_BASED_EXCLUDE_CONF_NAME) @classmethod async def apply_myimunify_modsec_rules_for_domains( cls, *, enabled_users_domains: dict, disabled_users_domains: dict ) -> set: """ Disable/enable extra modsecurity rules for specific domains. Return set of updated domains. """ return set() class ModSecSettingInterface(ABC): @abstractmethod async def apply(self): """ Installs and enabled ModSecurity Vendor """ pass @abstractmethod async def revert(self, **kwargs): """ Removes and disables ModSecurity Vendor :param kwargs: previous_value: values that was applied before """ pass class FilesVendor(ModSecSettingInterface): modsec_interface = ModSecurityInterface def __init__(self, item): """ :param dict item: e.g. { "groups": [ "cpanel", "apache" "configserver" ], "obsoletes": [ "imunify360_rules", "comodo_apache", ], "url": "https://files.imunify360.com/.../meta_imunify360_apache.yaml" } """ # noqa: E501 self._item = item self.vendor_id = self._vendor_id() @classmethod async def _get_web_server(cls) -> Optional[str]: return await _get_web_server() @abstractmethod async def apply(self): pass @abstractmethod async def _remove_vendor(): pass @abstractmethod def _vendor_id(self): pass async def revert(self): """ Removes and disables ModSecurity vendor """ if await self._is_installed(): await self._remove_vendor() logger.info("Successfully removed vendor %r.", self.vendor_id) async def _is_installed(self): installed_vendors = await self.modsec_interface.modsec_vendor_list() return self.vendor_id in installed_vendors class FilesVendorList(ModSecSettingInterface): class CompatiblityCheckFailed(RuntimeError): """ >>> e.args[0] {conflicting items} """ pass """ role: facade to handle FilesVendor list """ config_key = None files_vendor = FilesVendor modsec_interface = ModSecurityInterface @classmethod async def apply(cls): await cls.modsec_interface.invalidate_installed_vendors_cache() await cls.install_or_update() @classmethod @abstractmethod def vendor_fit_panel(cls, item): return @classmethod @abstractmethod async def _get_compatible_name(cls, installed_vendors): return @classmethod async def _get_web_server(cls) -> Optional[str]: return await _get_web_server() @classmethod async def install_or_update(cls) -> bool: """ Installs and enabled ModSecurity vendor list. Returns True if ModSecurity vendor was installed, False otherwise. """ index = files.Index(files.MODSEC) installed_vendors = await cls.modsec_interface.modsec_vendor_list() try: compatible_name = await cls._get_compatible_name(installed_vendors) except cls.CompatiblityCheckFailed as e: logger.warning("No vendor can be installed: %s", e) return False try: item = next( i for i in index.items() if i["name"] == compatible_name ) except StopIteration: logger.warning("Vendor %s not found in index", compatible_name) return False item["local_path"] = index.localfilepath(item["url"]) vendor_to_install = cls.files_vendor(item) # we do not want to have more than one imunify360 ruleset installed for i in index.items(): v = cls.files_vendor(i) if ( v.vendor_id in installed_vendors and v.vendor_id != vendor_to_install.vendor_id ): await v.revert() await vendor_to_install.apply() return True @classmethod async def revert(cls, *_): """ Removes and disables ModSecurity vendor list """ index = files.Index(files.MODSEC) for item in index.items(): if cls.vendor_fit_panel(item): await cls.files_vendor(item).revert() @staticmethod def get_ruleset_suffix(): return Modsec.RULESET.lower() class RemoteIPInterface(ABC): """This class implements panel-specific methods for activating and deactivating mod_remoteip or similar functionality. Concrete panel implementation may or may not derive from this class as a means of signaling of support for this functionality.""" _REMOTEIP_MODULE_NAMES: Tuple[bytes, ...] = ( b"mod_remoteip", b"remoteip_module", ) @abstractmethod async def remoteip_activated(self) -> bool: """Checks if remoteip feature is already activated""" return False @abstractmethod async def remoteip_install(self) -> Optional[str]: """Activates/ installs remoteip feature. Returns a path to installation log file or None if no log file is produced.""" return None async def remoteip_customize_logging(self): return None async def _is_loaded_to_apache(self): modules = await apache_modules() if not modules: return True return any(name in modules for name in self._REMOTEIP_MODULE_NAMES)