1#@!#!123s

: / opt / imunify360 / venv / lib / python3.11 / site-packages / im360 / internals / core / firewall /
Filename : base.py
back
from abc import ABCMeta, abstractmethod

from im360.utils import RulesLock
from im360.utils.validate import IP, IPVersion


class FirewallError(Exception):
    """Root exception class for firewall errors"""

    def __init__(self, ip_version: IPVersion, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.ip_version = ip_version


class FirewallTemporaryError(FirewallError):
    """Root exception class for temporary (transient) errors"""

    pass


class FirewallCommandNotFoundError(FirewallError):
    """Raised if underlying command execution fails with FileNotFoundError"""

    pass


class FirewallBatchCommandError(FirewallError):
    """Raised when a batch of command fails."""

    def __init__(self, ip_version: IPVersion, command: str, *args, **kwargs):
        super().__init__(ip_version, command, *args, **kwargs)
        self.command = command


class AbstractFirewall(metaclass=ABCMeta):
    """
    Abstract class that defines required interface
    for iptables ruleset editing classes
    """

    _lock = RulesLock()

    def __init__(self, ip_version=IP.V4):
        self.ip_version = ip_version

    @abstractmethod
    async def has_rule(self, *args, **kwargs):
        pass

    @abstractmethod
    async def has_chain(self, *args, **kwargs):
        pass

    @abstractmethod
    async def append_rule(self, *args, **kwargs):
        pass

    @abstractmethod
    async def insert_rule(self, *args, **kwargs):
        pass

    @abstractmethod
    async def delete_rule(self, *args, **kwargs):
        pass

    @abstractmethod
    async def flush_chain(self, *args, **kwargs):
        pass

    @abstractmethod
    async def create_chain(self, *args, **kwargs):
        pass

    @abstractmethod
    async def delete_chain(self, *args, **kwargs):
        pass

    @abstractmethod
    async def commit(self, records):
        pass

    async def __aenter__(self):
        await self._lock.acquire()
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        self._lock.release()