D7net
Home
Console
Upload
information
Create File
Create Folder
About
Tools
:
/
opt
/
imunify360
/
venv
/
lib64
/
python3.11
/
site-packages
/
defence360agent
/
contracts
/
Filename :
hooks.py
back
Copy
import grp import os from defence360agent.contracts.config import Config, Core from defence360agent.contracts.config_provider import CachedConfigReader from defence360agent.utils import antivirus_mode class Schema: @staticmethod def dict(data): return { "type": "dict", "schema": data, "default": {}, } @staticmethod def list_of_strings(regex=None): return { "type": "list", "schema": { "type": "string", **({"regex": regex} if regex else {}), }, "nullable": False, "default": [], } @staticmethod def list_of_emails(default_enabled=True): regex = ( r"^.+@(.+\.)+.+|default$" if default_enabled else r"^.+@(.+\.)+.+$" ) return Schema.list_of_strings(regex) @staticmethod def period(): return { "period": { "type": "integer", "coerce": int, "min": 1, "default": 1, } } @staticmethod def string(nullable): return { "type": "string", "nullable": nullable, } @staticmethod def enabled(): return { "enabled": { "type": "boolean", "default": False, } } @staticmethod def admin(period): return { "ADMIN": Schema.dict( { **Schema.enabled(), "admin_emails": Schema.list_of_emails(), **(Schema.period() if period else {}), } ) } @staticmethod def script(period): return { "SCRIPT": Schema.dict( { **Schema.enabled(), "scripts": Schema.list_of_strings(r"^\/.+$"), **(Schema.period() if period else {}), } ) } @staticmethod def user(period): return { "USER": Schema.dict( { **Schema.enabled(), **(Schema.period() if period else {}), } ) } @staticmethod def target_script(period=False): return Schema.dict( { **Schema.script(period=period), } ) @staticmethod def target_admin_and_script(period=False): return Schema.dict( { **Schema.admin(period=period), **Schema.script(period=period), } ) @staticmethod def target_all(period=False): return Schema.dict( { **Schema.admin(period=period), # **Schema.user(period=period), # stage 2 **Schema.script(period=period), } ) class HooksConfigReader(CachedConfigReader): GROUP_NAME = "_imunify" def _post_write(self): os.chmod(self.path, 0o640) os.chown(self.path, 0, grp.getgrnam(self.GROUP_NAME).gr_gid) class HooksConfig(Config): def __init__( self, path=os.path.join(Core.GLOBAL_CONFDIR, Core.HOOKS_CONFIGFILENAME) ): validation_schema = ( { "admin": Schema.dict( { "default_emails": Schema.list_of_emails( default_enabled=False ), "notify_from_email": { "type": "string", "default": None, "nullable": True, }, "locale": Schema.string(nullable=True), } ), "users": { "type": "list", "schema": Schema.dict( { "username": Schema.string(nullable=False), "emails": Schema.list_of_emails(), "locale": Schema.string(nullable=True), } ), "nullable": True, "default": [], }, "rules": Schema.dict( { "REALTIME_MALWARE_FOUND": ( Schema.target_admin_and_script(period=True) ), "USER_SCAN_MALWARE_FOUND": Schema.target_all(), "SCRIPT_BLOCKED": Schema.target_admin_and_script( period=True ), "USER_SCAN_STARTED": Schema.target_script(), "CUSTOM_SCAN_STARTED": Schema.target_script(), "USER_SCAN_FINISHED": Schema.target_script(), "CUSTOM_SCAN_FINISHED": Schema.target_script(), "CUSTOM_SCAN_MALWARE_FOUND": ( Schema.target_admin_and_script() ), } ), "default": {}, } if antivirus_mode.disabled else { "rules": Schema.dict( { "USER_SCAN_MALWARE_FOUND": Schema.target_script(), "USER_SCAN_STARTED": Schema.target_script(), "CUSTOM_SCAN_STARTED": Schema.target_script(), "USER_SCAN_FINISHED": Schema.target_script(), "CUSTOM_SCAN_FINISHED": Schema.target_script(), "CUSTOM_SCAN_MALWARE_FOUND": Schema.target_script(), } ), "default": {}, } ) super().__init__( path=path, validation_schema=validation_schema, config_reader=HooksConfigReader(path), ) def get(self): data = self.config_to_dict() data.pop("users", None) return data def update(self, data): data.pop("users", None) self.dict_to_config(data)