/
azathd
/
mutiagent
Обзор
Документация
Войти
/
azathd
/
mutiagent
Код
Запросы
0
Задачи
Вики
Пакеты
0
Релизы
0
CI/CD
Аналитика
Безопасность
master
src/agents/executor/nftables.py
38 строк
1 KB
Your Name
init
15 май 2026, 17:47
15 май 2026, 17:47
359c81e
Код
Авторство
О чём код?
"""nftables IP block helper (executor container).""" from __future__ import annotations import asyncio import os class NftablesManager: def __init__(self, table: str = "inet", chain: str = "xray_block") -> None: self.table = os.environ.get("NFT_TABLE_NAME", table) self.chain = chain async def block_ip(self, ip: str, ttl_sec: int = 3600) -> bool: if os.environ.get("EXECUTOR_DRY_RUN", "true").lower() == "true": return True cmd = [ "nft", "add", "element", f"{self.table}", "filter", self.chain, "{", ip, "}", ] proc = await asyncio.create_subprocess_exec(*cmd) await proc.wait() if ttl_sec > 0: asyncio.create_task(self._unblock_later(ip, ttl_sec)) return proc.returncode == 0 async def _unblock_later(self, ip: str, ttl: int) -> None: await asyncio.sleep(ttl) cmd = ["nft", "delete", "element", f"{self.table}", "filter", self.chain, "{", ip, "}"] proc = await asyncio.create_subprocess_exec(*cmd) await proc.wait()