"""An RPC interface to SMTP blocking subsystem. It is needed only for manual testing purposes. """ import logging from defence360agent.rpc_tools import lookup, ValidationError from defence360agent.utils import Scope, check_disabled_firewall from im360.subsys.smtp_blocking import ( sync_rules_for_all_versions, reset_rules_for_all_versions, is_SMTP_blocking_supported, read_SMTP_settings, conflicts_exist, ) __all__ = ["SMTPEndpoints"] logger = logging.getLogger(__name__) class SMTPEndpoints(lookup.RootEndpoints): SCOPE = Scope.IM360 @lookup.bind("smtp-blocking", "sync") @check_disabled_firewall async def sync_rules(self): """Create SMTP blocking iptables rules according to settings.""" if await is_SMTP_blocking_supported(): if await conflicts_exist(): raise ValidationError( "SMTP blocking conflicts with third party features" ) smtp_settings = read_SMTP_settings() await sync_rules_for_all_versions(smtp_settings) else: raise ValidationError( "SMTP blocking is not supported by current kernel" ) @lookup.bind("smtp-blocking", "reset") @check_disabled_firewall async def reset_rules(self): """Remove all rules installed by this plugin.""" if await is_SMTP_blocking_supported(): if await conflicts_exist(): raise ValidationError( "SMTP blocking conflicts with third party features" ) await reset_rules_for_all_versions() else: raise ValidationError( "SMTP blocking is not supported by current kernel" )