__ __ __ __ _____ _ _ _____ _ _ _
| \/ | \ \ / / | __ \ (_) | | / ____| | | | |
| \ / |_ __\ V / | |__) | __ ___ ____ _| |_ ___ | (___ | |__ ___| | |
| |\/| | '__|> < | ___/ '__| \ \ / / _` | __/ _ \ \___ \| '_ \ / _ \ | |
| | | | |_ / . \ | | | | | |\ V / (_| | || __/ ____) | | | | __/ | |
|_| |_|_(_)_/ \_\ |_| |_| |_| \_/ \__,_|\__\___| |_____/|_| |_|\___V 2.1
if you need WebShell for Seo everyday contact me on Telegram
Telegram Address : @jackleet
For_More_Tools:
from abc import ABCMeta, abstractmethod
from im360.utils import RulesLock
from defence360agent.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()