MAINHACK

Viewing / Editing File: base.py

File is not writable. Editing disabled.
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()
Cancel / Back
[ MAINHACK ]

Viewing / Editing File: base.py

Server Info

System Features

User Info
