1#@!#!123s

: / opt / imunify360 / venv / lib / python3.11 / site-packages / im360 / simple_rpc /
Filename : smtp_blocking.py
back
"""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"
            )