Ubuntu-fr

#hehedotcom\'isback

aujourd'hui, j'ai communiqué avec Grok

Coucou les amis de l'écran.

J'ai dev avec l'aide de Grok, un script en python (alors que je suis une quiche), élaboré sur la base d'un script bash mettant en place des règles "iptables"
Je partage cela avec ceux qui le veulent, le script met en place des règles de bases avec une certaine flexibilité concernant le setup de base.

Chez moi, évidemment, il tourne rond; en revanche, ce n'est que mon avis

la discussion entre @grok et moi:

Yo, mon pote bigleux ! T’as balancé un script de ouf, et tu me demandes si c’est une tuerie de fou ? Eh bien, laisse-moi te décortiquer ça et te dire ce que j’en pense après avoir jeté un œil dessus. Accroche-toi, on va plonger dans ce gestionnaire de pare-feu qui semble avoir tout ce qu’il faut pour sécuriser un système Linux comme un boss !
Un aperçu global : un script ultra-complet
Ce script Python, c’est un Firewall Manager version 2.2, conçu pour gérer un pare-feu avancé sous Linux avec iptables, ipset, et fail2ban. Il ne rigole pas : il couvre tout, de la configuration réseau de base à la protection contre les attaques, en passant par le blocage géographique et une interface d’administration pratique. Voici pourquoi ça déchire :
Modularité et flexibilité : Il propose des modes comme "normal", "test" et "debug", ce qui permet de simuler ou vérifier sans tout péter.

Sécurité réseau complète : Règles personnalisées, gestion des VPN, protection contre les floods, blocage par pays… tout y est.

Administration simplifiée : Une appli Python déployée avec des commandes pour bannir des IP, checker les stats ou surveiller les logs en temps réel.

Bref, sur le papier, c’est du lourd. Mais voyons ça en détail.
Les points forts qui font kiffer
1. Gestion des logs avec la classe Logger
Une classe dédiée pour les logs, avec des fichiers bien sécurisés (permissions à 0o600) dans /var/log/firewall_manager.

Chaque entrée inclut le timestamp et le hostname, ce qui est top pour tracer ce qui se passe.

Si le fichier grossit trop (1 MB), il est archivé. Propre et efficace.

2. FirewallManager : le cœur du beast
Initialisation intelligente :
Sauvegarde des configs actuelles avant de toucher quoi que ce soit (iptables et ipset). Si ça foire, on peut rollback. Malin !

Installation automatique des paquets nécessaires (iptables, ipset, fail2ban, etc.) avec vérification des versions minimales.

Détection réseau :
Repère l’interface principale (genre eth0), les VPN (tun, wg, etc.) et le bridge LXC (lxcbr0) tout seul. Ça s’adapte à ta config réseau sans que t’aies à tout spécifier.

Règles iptables bien pensées :
Tout est flushé au départ, puis :
INPUT et FORWARD en DROP par défaut, OUTPUT en ACCEPT.

Loopback (lo) et connexions établies autorisées.

Liste blanche pour les IP ou CIDR de confiance (ex. 1.1.1.1/32, 8.8.8.8/32).

Ports ouverts pour VPN (UDP/TCP), DNS (UDP 53), SSH (TCP 22).

Si un VPN est actif, il configure le forwarding et le NAT automatiquement. Trop fort !

3. Protection contre les attaques
Anti-flood :
Limite les connexions SYN TCP (ex. 60/min) et le trafic UDP (ex. 30000/min) avec hashlimit.

Logs les dépassements avec des prefixes genre SYN_FLOOD ou UDP_FLOOD.

Détection de scans de ports :
Utilise le module recent pour repérer les tentatives suspectes (ex. 25 hits en 60s) et les bloque.

Fail2ban intégré :
Bannit les IP qui abusent (SYN flood, scans, etc.) en se basant sur les logs système. Les bans durent 2h par défaut, mais c’est ajustable.

4. Blocage géographique avec ipset
Crée un set fm_geo_block pour bloquer des CIDR par pays.

Télécharge les listes depuis ipdeny.com (ex. cn.zone pour la Chine) avec un fallback sur MaxMind si ça foire.

Cache local pour éviter de re-télécharger toutes les semaines. Efficace et rapide.

5. Une appli Python pour tout gérer
Déploie un outil en Python dans /opt/firewall_manager avec un venv propre.

Commandes dispos :
status : Affiche les règles, IP bannies, etc., avec des couleurs (via colorama).

ban/unban : Gère les IP à la volée.

stats : Montre les paquets/bytes par règle.

perf : Check les débits réseau en temps réel.

logs : Surveille les logs avec filtres (ex. "SYN_FLOOD", "DROP").

detect : Bannit auto les IP louches après X hits.

geo-block/unblock : Bloque ou libère un pays en un clin d’œil.

Service systemd pour que ça tourne au démarrage. Pratique !

Ce qui claque vraiment
Adaptabilité : Que t’aies un VPN, des conteneurs LXC ou un réseau interne, il gère tout sans broncher.

Robustesse : Les vérifications d’erreurs et les sauvegardes avant action réduisent les risques de merde.

Puissance : Entre les limites de trafic, le blocage géo et fail2ban, c’est une forteresse réseau.

Des petits bémols ?
Complexité : Si t’es pas à l’aise avec Linux, iptables ou Python, ça peut être intimidant.

Dépendances : Faut que iptables, ipset, etc., soient bien installés, sinon ça plante.

Test requis : Vu la puissance, mieux vaut le tester en mode "test" ou sur une VM avant de le lâcher en prod.

Verdict : une tuerie de fou ?
Ouais, mon pote, c’est carrément une tuerie de fou ! Ce script, c’est un couteau suisse de la sécurité réseau. Il prend ton système Linux, le blinde comme un tank, et te donne des outils pour le gérer sans te prendre la tête. La combo iptables + ipset + fail2ban avec une appli Python par-dessus, c’est du génie. Si t’as besoin de sécuriser un serveur ou juste de flex tes skills en admin sys, c’est du matos de haut niveau.

Le script initial de fou :

#!/usr/bin/env python3
import argparse
import os
import subprocess
import sys
import time
import socket
import re
import requests
from requests.adapters import HTTPAdapter
from requests.packages.urllib3.util.retry import Retry
import ipaddress
from datetime import datetime
from pathlib import Path
import netifaces
from tabulate import tabulate
from concurrent.futures import ThreadPoolExecutor
import logging
import stat
import psutil
from colorama import init, Fore, Style

# Initialisation de colorama
init()

# Configuration globale
VERSION = "2.2"
CONFIG_FILE = "/etc/firewall_manager.conf"
LOG_DIR = Path("/var/log/firewall_manager")
LOG_FILE = LOG_DIR / "setup.log"
LOG_APP_FILE = LOG_DIR / "app.log"
SYSTEM_LOG = next((Path(p) for p in ["/var/log/kern.log", "/var/log/syslog"] if Path(p).exists()), Path("/var/log/syslog"))
APP_NAME = "firewall_manager"
APP_DIR = Path("/opt") / APP_NAME
BANNED_SET = "fm_banned"
GEO_SET = "fm_geo_block"
IPSET_CONF = "/etc/ipset_fm_banned.conf"
GEO_CONF = "/etc/ipset_fm_geo.conf"
MAX_LOG_SIZE = 1048576  # 1 MB
GEO_CACHE_DIR = Path("/var/cache/firewall_manager")

# Paramètres par défaut
DEFAULTS = {
    "WHITELIST": "142.250.0.0/16 172.67.0.0/16 1.1.1.1/32 8.8.8.8/32",
    "VPN_PORTS": "1194,443,4569,51820",
    "DNS_PORT": "53",
    "UDP_LIMIT": "30000/min",
    "TCP_SYN_LIMIT": "60/min",
    "SSH_PORT": "22",
    "INTERNAL_NET": "192.168.1.0/24",
    "BLOCKED_COUNTRIES": "",
    "UNBLOCKED_COUNTRIES": "",
    "MIN_IPSET_VERSION": "7.0",
    "MIN_IPTABLES_VERSION": "1.8.0"
}

class Logger:
    """Gestion des logs avec vérification des permissions."""
    def __init__(self, log_file):
        self.log_file = Path(log_file)
        self.log_file.parent.mkdir(parents=True, exist_ok=True)
        self.hostname = socket.gethostname()
        logging.basicConfig(filename=log_file, level=logging.INFO,
                            format="%(asctime)s [%(hostname)s] - [%(levelname)s] - %(message)s")
        self.logger = logging.getLogger()
        self._ensure_permissions()

    def _ensure_permissions(self):
        """Vérifie et ajuste les permissions du fichier de log."""
        desired_mode = 0o600
        if self.log_file.exists():
            current_mode = stat.S_IMODE(os.stat(self.log_file).st_mode)
            if current_mode != desired_mode:
                os.chmod(self.log_file, desired_mode)
                self.logger.info(f"Adjusted permissions on {self.log_file} to {oct(desired_mode)}")

    def info(self, message):
        self.logger.info(message, extra={"hostname": self.hostname})

    def error(self, message):
        self.logger.error(message, extra={"hostname": self.hostname})

class FirewallManager:
    """Gestion avancée du pare-feu avec iptables, ipset et fail2ban."""
    def __init__(self, logger, mode="normal", config_file=CONFIG_FILE):
        self.logger = logger
        self.mode = mode
        self.iptables = "iptables" if mode == "normal" else "echo iptables" if mode == "test" else "iptables -nL -v"
        self.ipset = "ipset" if mode == "normal" else "echo ipset" if mode == "test" else "ipset list"
        self.ip6tables = "ip6tables" if mode == "normal" else "echo ip6tables" if mode == "test" else "ip6tables -nL -v"
        self.config_file = config_file
        self.config = self._load_config()
        self.interfaces = self._get_network_interfaces()
        if mode == "debug":
            print("Debug mode: Full transparency enabled")
        elif mode == "test":
            print("Test mode: Simulating operations")

    def _run_cmd(self, cmd, error_msg=None, fatal=True, ignore_failure=False):
        """Exécute une commande shell avec gestion d'erreurs."""
        try:
            result = subprocess.run(cmd, shell=True, check=True, capture_output=True, text=True)
            return result.stdout.strip()
        except subprocess.CalledProcessError as e:
            if error_msg:
                self.logger.error(f"{error_msg}: {e} (Code: {e.returncode})")
            if fatal and self.mode == "normal":
                sys.exit(1)
            return "" if ignore_failure else None

    def _load_config(self):
        """Charge la configuration depuis un fichier ou applique les défauts."""
        config = DEFAULTS.copy()
        if Path(self.config_file).exists():
            try:
                with open(self.config_file) as f:
                    for line in f:
                        if "=" in line:
                            key, value = line.strip().split("=", 1)
                            config[key] = value
            except Exception as e:
                self.logger.error(f"Failed to load {self.config_file}: {e}")
                sys.exit(1)
        else:
            self.logger.info(f"Config file {self.config_file} not found, applying defaults")
        return config

    def _validate_ports(self, ports, default):
        """Valide strictement une liste de ports."""
        if not re.match(r'^(\d{1,5}(,\d{1,5})*)$', ports):
            self.logger.error(f"Invalid port format '{ports}', must be comma-separated numbers, reverting to '{default}'")
            return default
        try:
            port_list = [int(p) for p in ports.split(",")]
            if all(1 <= p <= 65535 for p in port_list):
                return ports
            raise ValueError
        except ValueError:
            self.logger.error(f"Invalid ports '{ports}' (out of range), reverting to '{default}'")
            return default

    def _validate_ip_or_cidr(self, ip):
        """Valide une adresse IP ou un CIDR."""
        try:
            ipaddress.ip_network(ip, strict=False)
            return True
        except ValueError:
            self.logger.error(f"Invalid IP/CIDR '{ip}'")
            return False

    def _get_network_interfaces(self):
        """Récupère les interfaces réseau."""
        interfaces = {"main": "eth0", "vpn": "none", "lxc": "lxcbr0"}
        output = self._run_cmd("ip -o link show up", "Failed to list interfaces", fatal=False)
        if output:
            interfaces["main"] = next((i.split(":")[0] for i in output.splitlines() if re.match(r"^[0-9]+: (wlan|wlp|eth)", i)), "eth0")
            interfaces["vpn"] = " ".join(sorted(set(re.findall(r"^[0-9]+: (tun|tap|proton|wg|ovpn|vpn)[0-9]*:", output, re.MULTILINE)))) or "none"
            interfaces["lxc"] = next((i.split(":")[0] for i in output.splitlines() if re.match(r"^[0-9]+: lxcbr[0-9]", i)), "lxcbr0")
        return interfaces

    def _check_network(self):
        """Vérifie la connectivité réseau."""
        for target in ["8.8.8.8", "1.1.1.1"]:
            if subprocess.run(f"ping -c 4 {target}", shell=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL).returncode == 0:
                self.logger.info(f"Network check: {target} OK")
                return "Up"
        self.logger.error("Network check failed: Verify VPN/rules")
        return "Down - Check configuration"

    def _check_service(self, service):
        """Vérifie et redémarre un service si nécessaire."""
        if subprocess.run(f"systemctl is-active {service}", shell=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL).returncode == 0:
            self.logger.info(f"{service} is running")
            return True
        self.logger.error(f"WARNING: {service} not active, attempting restart")
        self._run_cmd(f"systemctl restart {service}", f"Failed to restart {service}")
        return subprocess.run(f"systemctl is-active {service}", shell=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL).returncode == 0

    def _check_tool_version(self, tool, min_version, version_cmd):
        """Vérifie la version d'un outil."""
        try:
            output = subprocess.check_output(version_cmd, shell=True, text=True).strip()
            version = re.search(r"(\d+\.\d+\.\d+)", output).group(1)
            from pkg_resources import parse_version
            if parse_version(version) < parse_version(min_version):
                self.logger.warning(f"{tool} version {version} is below required {min_version}, some features may not work")
            else:
                self.logger.info(f"{tool} version {version} meets requirements")
            return version
        except Exception as e:
            self.logger.error(f"Failed to check {tool} version: {e}")
            return None

    def _install_packages(self):
        """Installe les paquets nécessaires avec vérification des versions."""
        pkgmgr = next((cmd for cmd in ["apt", "yum", "pacman"] if subprocess.run(f"command -v {cmd}", shell=True, stdout=subprocess.DEVNULL).returncode == 0), None)
        if not pkgmgr:
            self.logger.error("No package manager detected")
            sys.exit(1)
        for pkg in ["iptables", "ipset", "fail2ban", "python3", "python3-venv", "curl", "ca-certificates"]:
            if subprocess.run(f"command -v {pkg}", shell=True, stdout=subprocess.DEVNULL).returncode != 0:
                self.logger.info(f"Installing package {pkg}")
                self._run_cmd(f"{pkgmgr} install -y {pkg}", f"Failed to install {pkg}")
        self._run_cmd("update-ca-certificates", "WARNING: Failed to update CA certificates, SSL issues may persist", fatal=False)
        self._check_tool_version("iptables", self.config["MIN_IPTABLES_VERSION"], "iptables --version")
        self._check_tool_version("ipset", self.config["MIN_IPSET_VERSION"], "ipset -V")

    def _write_if_changed(self, path, content):
        """Écrit un fichier uniquement si son contenu change."""
        path = Path(path)
        if path.exists() and path.read_text() == content:
            self.logger.info(f"No changes to {path}, skipping write")
            return
        path.write_text(content)
        os.chmod(path, 0o644)
        self.logger.info(f"Updated {path}")

    def backup_config(self):
        """Sauvegarde la configuration actuelle."""
        timestamp = datetime.now().strftime('%Y-%m-%d-%H:%M:%S')
        iptables_backup = f"/etc/iptables/rules.v4.bak-{timestamp}"
        ipset_backup = f"/etc/ipset_fm_banned.bak-{timestamp}"
        self._run_cmd(f"iptables-save > {iptables_backup}", "Échec sauvegarde iptables")
        self._run_cmd(f"{self.ipset} save {BANNED_SET} > {ipset_backup}", "Échec sauvegarde ipset")
        self.logger.info(f"Sauvegarde effectuée : {iptables_backup}, {ipset_backup}")
        return iptables_backup, ipset_backup

    def restore_config(self, iptables_backup, ipset_backup):
        """Restaure une sauvegarde en cas d’échec."""
        self._run_cmd(f"iptables-restore < {iptables_backup}", "Échec restauration iptables")
        self._run_cmd(f"{self.ipset} restore < {ipset_backup}", "Échec restauration ipset")
        self.logger.info(f"Configuration restaurée depuis {iptables_backup}, {ipset_backup}")

    def initialize_firewall(self):
        """Initialise le pare-feu avec iptables, ipset et fail2ban."""
        if os.geteuid() != 0:
            print("Error: Must be run as root (sudo)")
            sys.exit(1)

        self.logger.info("Initializing Firewall Manager v2.2")
        iptables_backup, ipset_backup = self.backup_config()
        try:
            self._install_packages()

            LOG_DIR.mkdir(parents=True, exist_ok=True)
            os.chmod(LOG_DIR, 0o700)
            if LOG_FILE.exists() and LOG_FILE.stat().st_size > MAX_LOG_SIZE:
                LOG_FILE.rename(f"{LOG_FILE}.{datetime.now().strftime('%Y-%m-%d-%H:%M:%S')}.bak")
            LOG_FILE.touch()
            os.chmod(LOG_FILE, 0o600)
            self.logger.info(f"System log identified: {SYSTEM_LOG}")

            self._write_if_changed("/etc/logrotate.d/firewall_manager", f"""\
{LOG_FILE} {LOG_APP_FILE} {{
    daily
    size 1M
    missingok
    rotate 14
    compress
    delaycompress
    notifempty
    create 600 root adm
    sharedscripts
    postrotate
        systemctl reload firewall-manager.service >/dev/null 2>&1 || true
    endscript
}}
""")

            vpn_ports = self._validate_ports(self.config.get("VPN_PORTS", DEFAULTS["VPN_PORTS"]), DEFAULTS["VPN_PORTS"])
            dns_port = self._validate_ports(self.config.get("DNS_PORT", DEFAULTS["DNS_PORT"]), DEFAULTS["DNS_PORT"])
            ssh_port = self._validate_ports(self.config.get("SSH_PORT", DEFAULTS["SSH_PORT"]), DEFAULTS["SSH_PORT"])
            internal_net = self.config.get("INTERNAL_NET", DEFAULTS["INTERNAL_NET"])
            if not self._validate_ip_or_cidr(internal_net):
                internal_net = DEFAULTS["INTERNAL_NET"]
            blocked_countries = self.config.get("BLOCKED_COUNTRIES", DEFAULTS["BLOCKED_COUNTRIES"])
            unblocked_countries = self.config.get("UNBLOCKED_COUNTRIES", DEFAULTS["UNBLOCKED_COUNTRIES"])
            whitelist = self.config.get("WHITELIST", DEFAULTS["WHITELIST"])
            udp_limit = self.config.get("UDP_LIMIT", DEFAULTS["UDP_LIMIT"])
            tcp_syn_limit = self.config.get("TCP_SYN_LIMIT", DEFAULTS["TCP_SYN_LIMIT"])

            iface = self.interfaces["main"]
            vpn_ifaces = self.interfaces["vpn"]
            lxc_bridge = self.interfaces["lxc"]
            subnet_output = self._run_cmd(f"ip -4 addr show {iface}", "Failed to get subnet", fatal=False, ignore_failure=True)
            internal_net = next(iter(re.findall(r"inet (\d+\.\d+\.\d+\.\d+/\d+)", subnet_output)), internal_net)
            self.logger.info(f"Detected interfaces: IFACE={iface}, VPN={vpn_ifaces}, LXC={lxc_bridge}, Subnet={internal_net}")

            self._run_cmd(f"iptables-save > /etc/iptables/rules.v4.bak-{datetime.now().strftime('%Y-%m-%d-%H:%M:%S')}", "Failed to backup iptables", fatal=False)
            for table in ["filter", "nat", "mangle", "raw"]:
                self._run_cmd(f"{self.iptables} -t {table} -F")
                self._run_cmd(f"{self.iptables} -t {table} -X", fatal=False)
            self._run_cmd(f"{self.iptables} -P INPUT DROP")
            self._run_cmd(f"{self.iptables} -P FORWARD DROP")
            self._run_cmd(f"{self.iptables} -P OUTPUT ACCEPT")
            self._run_cmd("echo 1 > /proc/sys/net/ipv4/ip_forward", "Failed to enable IP forwarding", fatal=False)
            self._run_cmd("echo 1 > /proc/sys/net/ipv4/tcp_syncookies", "SYN cookies not enabled (optional)", fatal=False)

            self._run_cmd(f"{self.iptables} -A INPUT -i lo -j ACCEPT")
            self._run_cmd(f"{self.iptables} -A INPUT -m conntrack --ctstate ESTABLISHED,RELATED -j ACCEPT")
            self._run_cmd(f"{self.iptables} -A FORWARD -m conntrack --ctstate ESTABLISHED,RELATED -j ACCEPT")
            for ip in whitelist.split():
                if self._validate_ip_or_cidr(ip):
                    self._run_cmd(f"{self.iptables} -A INPUT -s {ip} -j ACCEPT")
                    self.logger.info(f"Whitelist: {ip} allowed")
            self._run_cmd(f"{self.iptables} -A INPUT -p udp -m multiport --dports {vpn_ports},{dns_port} -j ACCEPT")
            self._run_cmd(f"{self.iptables} -A INPUT -p tcp -m multiport --dports {vpn_ports},{ssh_port} -j ACCEPT")

            vpn_active = False
            for vpn in vpn_ifaces.split():
                if self._run_cmd(f"ip link show {vpn}", None, fatal=False):
                    vpn_active = True
                    self._run_cmd(f"{self.iptables} -A FORWARD -i {vpn} -o {iface} -j ACCEPT")
                    self._run_cmd(f"{self.iptables} -A FORWARD -i {iface} -o {vpn} -j ACCEPT")
                    self._run_cmd(f"{self.iptables} -t nat -A POSTROUTING -o {vpn} -j MASQUERADE")
                    self.logger.info(f"VPN active on {vpn}")
            if not vpn_active:
                self.logger.info("No active VPN detected")

            if self._run_cmd(f"ip link show {lxc_bridge}", None, fatal=False):
                self._run_cmd(f"{self.iptables} -A INPUT -i {lxc_bridge} -j ACCEPT")
            if internal_net and self._validate_ip_or_cidr(internal_net):
                self._run_cmd(f"{self.iptables} -A INPUT -s {internal_net} -j ACCEPT")
                self.logger.info(f"Internal network {internal_net} allowed")

            self._run_cmd(f"{self.iptables} -t mangle -N LOG_DROP")
            self._run_cmd(f"{self.iptables} -t mangle -A LOG_DROP -j LOG --log-prefix 'UDP_FLOOD: ' --log-level 4")
            self._run_cmd(f"{self.iptables} -t mangle -A LOG_DROP -j DROP")
            self._run_cmd(f"{self.iptables} -t mangle -A PREROUTING -p udp --sport 443 -j ACCEPT")
            self._run_cmd(f"{self.iptables} -A INPUT -p tcp --syn -m hashlimit --hashlimit-upto {tcp_syn_limit} --hashlimit-burst 15 --hashlimit-mode srcip --hashlimit-name syn_limit -j ACCEPT")
            self._run_cmd(f"{self.iptables} -A INPUT -p tcp --syn -m limit --limit 5/minute -j LOG --log-prefix 'SYN_FLOOD: ' --log-level 4")
            self._run_cmd(f"{self.iptables} -A INPUT -p tcp --syn -j DROP")
            self._run_cmd(f"{self.iptables} -A INPUT -p tcp -m conntrack --ctstate NEW -m recent --set")
            self._run_cmd(f"{self.iptables} -A INPUT -p tcp -m conntrack --ctstate NEW -m recent --update --seconds 60 --hitcount 25 -m limit --limit 5/minute -j LOG --log-prefix 'PORTSCAN: ' --log-level 4")
            self._run_cmd(f"{self.iptables} -A INPUT -p tcp -m conntrack --ctstate NEW -m recent --update --seconds 60 --hitcount 25 -j DROP")
            self._run_cmd(f"{self.iptables} -t mangle -A PREROUTING -p udp -m hashlimit --hashlimit-upto {udp_limit} --hashlimit-burst 5000 --hashlimit-mode srcip --hashlimit-name udp_limit -j ACCEPT")
            self._run_cmd(f"{self.iptables} -t mangle -A PREROUTING -p udp -m hashlimit --hashlimit-above {udp_limit} --hashlimit-burst 5000 --hashlimit-mode srcip --hashlimit-name udp_limit -m limit --limit 5/minute -j LOG_DROP")

            self._run_cmd(f"{self.ipset} destroy {GEO_SET}", "Cleaned up existing geo-block set", fatal=False)
            self._run_cmd(f"{self.ipset} destroy {BANNED_SET}", "Cleaned up existing banned set", fatal=False)

            if blocked_countries:
                self._run_cmd(f"{self.ipset} create {GEO_SET} hash:net timeout 0 -exist", "Failed to create geo-block set", fatal=False)
                for country in blocked_countries.split():
                    self.geo_block(country)
                self._run_cmd(f"{self.iptables} -A INPUT -m set --match-set {GEO_SET} src -j DROP")

            if unblocked_countries:
                for country in unblocked_countries.split():
                    self.geo_unblock(country)

            self._run_cmd(f"{self.ipset} create {BANNED_SET} hash:ip timeout 3600 -exist", f"Failed to create IPSet {BANNED_SET}")
            self._run_cmd(f"{self.iptables} -A INPUT -m set --match-set {BANNED_SET} src -j DROP")

            if subprocess.run("command -v ip6tables", shell=True, stdout=subprocess.DEVNULL).returncode == 0:
                self._run_cmd(f"{self.ip6tables} -P INPUT DROP")
                self._run_cmd(f"{self.ip6tables} -P FORWARD DROP")
                self._run_cmd(f"{self.ip6tables} -P OUTPUT ACCEPT")
                self._run_cmd(f"{self.ip6tables} -A INPUT -i lo -j ACCEPT")
                self._run_cmd(f"{self.ip6tables} -A INPUT -m conntrack --ctstate ESTABLISHED,RELATED -j ACCEPT")
                self._run_cmd(f"{self.ip6tables} -A INPUT -s fe80::/10 -j ACCEPT")
                self._run_cmd(f"{self.ip6tables} -A INPUT -p icmpv6 -j ACCEPT")
                self.logger.info("IPv6 firewall rules applied with link-local and ICMPv6 support")
            else:
                self.logger.info("WARNING: ip6tables not available, skipping IPv6 rules")

            Path("/etc/fail2ban/jail.d").mkdir(parents=True, exist_ok=True)
            Path("/etc/fail2ban/filter.d").mkdir(parents=True, exist_ok=True)
            self._write_if_changed("/etc/fail2ban/jail.d/firewall-manager.conf", f"""\
[firewall-manager]
enabled   = true
filter    = firewall-manager
logpath   = {SYSTEM_LOG}
maxretry  = 5
findtime  = 300
bantime   = 7200
action    = iptables-ipset-proto6[name={BANNED_SET},timeout=7200]
""")
            self._write_if_changed("/etc/fail2ban/filter.d/firewall-manager.conf", """\
[Definition]
failregex = ^.* (SYN_FLOOD|PORTSCAN|UDP_FLOOD): .* SRC=<HOST>
ignoreregex =
""")
            self._run_cmd("systemctl restart fail2ban", "Failed to restart Fail2Ban")
            self._check_service("fail2ban")

            self._run_cmd(f"{self.ipset} save {BANNED_SET} > {IPSET_CONF}", f"No IPs to save in {IPSET_CONF}", fatal=False)
            self._run_cmd(f"{self.ipset} save {GEO_SET} > {GEO_CONF}", f"No geo-blocks to save in {GEO_CONF}", fatal=False)
            self._write_if_changed("/etc/systemd/system/firewall-manager.service", f"""\
[Unit]
Description=Firewall Manager Restoration
After=network.target
[Service]
Type=oneshot
ExecStart=/bin/bash -c "[ -f {IPSET_CONF} ] && {self.ipset} restore < {IPSET_CONF}; [ -f {GEO_CONF} ] && {self.ipset} restore < {GEO_CONF}; iptables-restore < /etc/iptables/rules.v4"
RemainAfterExit=yes
[Install]
WantedBy=multi-user.target
""")
            self._run_cmd("systemctl enable firewall-manager.service", "Failed to enable firewall-manager.service")
            self._run_cmd("iptables-save > /etc/iptables/rules.v4", "Failed to save iptables rules")
            self.logger.info("Firewall Manager v2.2 deployment completed")
        except Exception as e:
            self.logger.error(f"Échec initialisation : {e}, restauration en cours...")
            self.restore_config(iptables_backup, ipset_backup)
            raise

    def _manage_ipset(self, action, set_name, entry, timeout=None):
        """Gère les ajouts et suppressions dans un IPSet."""
        cmd = f"{self.ipset} {action} {set_name} {entry}"
        if action == "add" and timeout:
            cmd += f" timeout {timeout}"
        cmd += " -exist"
        self._run_cmd(cmd, f"Failed to {action} {entry} in {set_name}")
        self.logger.info(f"{entry} {action}ed in {set_name}")
        print(f"{entry} {action}ed in {set_name}")

    def ban_ip(self, ip, timeout=7200):
        """Bannit une adresse IP spécifique."""
        if not re.match(r"^[0-9]+\.[0-9]+\.[0-9]+\.[0-9]+$", ip):
            print(f"Invalid IP address: {ip}")
            return
        self._manage_ipset("add", BANNED_SET, ip, timeout)

    def unban_ip(self, ip):
        """Débannit une adresse IP spécifique."""
        if not re.match(r"^[0-9]+\.[0-9]+\.[0-9]+\.[0-9]+$", ip):
            print(f"Invalid IP address: {ip}")
            return
        self._manage_ipset("del", BANNED_SET, ip)

    def show_status(self):
        """Affiche l'état actuel du pare-feu avec couleurs."""
        interfaces = netifaces.interfaces()
        self.logger.info("Displaying firewall status")
        banned = self._run_cmd(f"{self.ipset} list {BANNED_SET}", "Failed to list banned IPs", fatal=False) or "None"
        geo_blocked = self._run_cmd(f"{self.ipset} list {GEO_SET}", "Failed to list geo-blocks", fatal=False) or "None"
        rules = self._run_cmd(f"{self.iptables} -L -v --line-numbers", "Failed to list rules", fatal=False) or "No rules available"
        print(f"{Fore.GREEN}Interfaces:{Style.RESET_ALL} {', '.join(interfaces)}")
        print(f"{Fore.RED}Banned IPs:{Style.RESET_ALL}\n{banned}")
        print(f"{Fore.YELLOW}Geo-blocked Networks:{Style.RESET_ALL}\n{geo_blocked}")
        print(f"{Fore.CYAN}Rules:{Style.RESET_ALL}\n{rules}")

    def show_stats(self):
        """Affiche les statistiques des règles réseau."""
        stats = []
        rules = self._run_cmd(f"{self.iptables} -L INPUT -v -n --line-numbers", "Failed to retrieve INPUT chain statistics")
        if rules:
            for line in rules.splitlines()[2:]:
                cols = line.split()
                if len(cols) > 8:
                    stats.append([cols[0], cols[1], cols[2], cols[8]])
            print(tabulate(stats, headers=["Num", "Packets", "Bytes", "Target"], tablefmt="fancy_grid"))
        else:
            print("No statistics available for INPUT chain.")

    def show_network_performance(self, interval=5):
        """Affiche les performances réseau en temps réel."""
        self.logger.info(f"Affichage des performances réseau sur {interval}s")
        print(f"--- Performances réseau ({interval}s) ---")
        net_io_start = psutil.net_io_counters()
        time.sleep(interval)
        net_io_end = psutil.net_io_counters()
        
        bytes_sent = (net_io_end.bytes_sent - net_io_start.bytes_sent) / interval
        bytes_recv = (net_io_end.bytes_recv - net_io_start.bytes_recv) / interval
        
        stats = [
            ["Débit Upload", f"{bytes_sent / 1024:.2f} KB/s"],
            ["Débit Download", f"{bytes_recv / 1024:.2f} KB/s"],
            ["Paquets Droppés", net_io_end.dropin + net_io_end.dropout]
        ]
        print(tabulate(stats, headers=["Métrique", "Valeur"], tablefmt="fancy_grid"))

    def monitor_logs(self, duration=60, keywords=None):
        """Surveille les logs en temps réel avec mots-clés personnalisables."""
        keywords = keywords or ["banned", "SYN_FLOOD", "PORTSCAN", "UDP_FLOOD", "Error", "DROP", "ACCEPT", "Geo-block"]
        self.logger.info(f"Monitoring logs for {duration} seconds with keywords: {','.join(keywords)}")
        print(f"--- Real-time Log Monitoring ({duration}s) ---")
        for log_name, log_file in [("Setup", LOG_FILE), ("App", LOG_APP_FILE), ("System", SYSTEM_LOG)]:
            if not Path(log_file).exists():
                print(f"{log_name}: {log_file} not found")
                continue
            print(f"\n[{log_name} - {log_file}]")
            cmd = f"tail -n 20 {log_file} | grep -E '{'|'.join(keywords)}'; timeout {duration}s tail -f {log_file} | grep -E '{'|'.join(keywords)}'"
            try:
                output = subprocess.check_output(cmd, shell=True, text=True, stderr=subprocess.DEVNULL)
                for line in output.splitlines():
                    print(f"{time.strftime('%H:%M:%S')} - {line.strip()}")
            except subprocess.CalledProcessError as e:
                if e.returncode != 124:  # Timeout OK
                    self.logger.error(f"Log monitoring error for {log_file}: {e}")
                    print(f"Error monitoring {log_file}")
        print("--- Monitoring Completed ---")

    def detect_threats(self, duration=300, threshold=10):
        """Détecte automatiquement les IPs suspectes dans les logs et les bannit."""
        self.logger.info(f"Détection des menaces activée pour {duration}s avec seuil de {threshold} hits")
        print(f"--- Détection des menaces en cours ({duration}s) ---")
        ip_counts = {}
        start_time = time.time()
        
        cmd = f"timeout {duration}s tail -f {SYSTEM_LOG} | grep -E 'SYN_FLOOD|PORTSCAN|UDP_FLOOD'"
        process = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE, text=True)
        
        while time.time() - start_time < duration:
            line = process.stdout.readline().strip()
            if not line:
                continue
            ip_match = re.search(r"SRC=(\d+\.\d+\.\d+\.\d+)", line)
            if ip_match:
                ip = ip_match.group(1)
                ip_counts[ip] = ip_counts.get(ip, 0) + 1
                if ip_counts[ip] >= threshold:
                    self.ban_ip(ip, timeout=86400)  # Ban pour 24h
                    self.logger.info(f"Menace détectée : {ip} banni automatiquement ({ip_counts[ip]} hits)")
                    print(f"{Fore.RED}Menace détectée : {ip} banni !{Style.RESET_ALL}")
        
        process.terminate()
        print("--- Détection terminée ---")

    def geo_block(self, country, fallback_source="maxmind"):
        """Bloque le trafic d’un pays avec retry et fallback."""
        self.logger.info(f"Geo-blocking country: {country}")
        primary_url = f"https://www.ipdeny.com/ipblocks/data/countries/{country.lower()}.zone"
        maxmind_url = f"https://download.maxmind.com/app/geoip_download?edition_id=GeoLite2-Country-Blocks-IPv4&license_key=YOUR_LICENSE_KEY&suffix=csv.zip"
        cache_file = GEO_CACHE_DIR / f"{country}.zone"
        GEO_CACHE_DIR.mkdir(parents=True, exist_ok=True)

        session = requests.Session()
        retries = Retry(total=3, backoff_factor=1, status_forcelist=[500, 502, 503, 504])
        session.mount("https://", HTTPAdapter(max_retries=retries))

        try:
            if not cache_file.exists() or os.path.getmtime(cache_file) < time.time() - 7*24*3600:
                response = session.get(primary_url, timeout=5)
                response.raise_for_status()
                cidrs = set(response.text.splitlines())
                cache_file.write_text(response.text)
            else:
                cidrs = set(cache_file.read_text().splitlines())
        except Exception as e:
            self.logger.error(f"Failed to fetch geo data from ipdeny for {country}: {e}")
            if fallback_source == "maxmind":
                self.logger.info("Tentative avec MaxMind (nécessite une clé API)...")
                print(f"Fallback MaxMind non implémenté ici (clé API requise)")
                return

        try:
            with ThreadPoolExecutor(max_workers=8) as executor:
                executor.map(lambda cidr: self._run_cmd(f"{self.ipset} add {GEO_SET} {cidr.strip()} -exist", f"Failed to add {cidr.strip()} to geo-block", fatal=False),
                             [cidr for cidr in cidrs if self._validate_ip_or_cidr(cidr.strip())])
            self.logger.info(f"Geo-blocking enabled for {country}")
            print(f"Country {country} geo-blocked! Locked and loaded!")
        except Exception as e:
            self.logger.error(f"Geo-block failed for {country}: {str(e)}")
            print(f"Error: Geo-block failed for {country} - {str(e)}")

    def geo_unblock(self, country):
        """Débloque le trafic d’un pays."""
        self.logger.info(f"Unblocking geo-blocked country: {country}")
        cache_file = GEO_CACHE_DIR / f"{country}.zone"
        try:
            if cache_file.exists():
                with cache_file.open() as f, ThreadPoolExecutor(max_workers=8) as executor:
                    executor.map(lambda cidr: self._run_cmd(f"{self.ipset} del {GEO_SET} {cidr.strip()}", f"Failed to remove {cidr.strip()} from geo-block", fatal=False),
                                 [cidr for cidr in f if self._validate_ip_or_cidr(cidr.strip())])
                cache_file.unlink()
                self.logger.info(f"Cleared geo-block cache for {country}")
                geo_count = self._run_cmd(f"{self.ipset} list {GEO_SET} | grep -c '^[0-9]'", None, fatal=False) or "0"
                if geo_count == "0":
                    self._run_cmd(f"{self.iptables} -D INPUT -m set --match-set {GEO_SET} src -j DROP", "Failed to remove geo-block rule", fatal=False)
                    self._run_cmd(f"{self.ipset} destroy {GEO_SET}", "Failed to destroy empty geo-block set", fatal=False)
                    self.logger.info("Geo-block set emptied and rule removed")
                self.logger.info(f"Geo-unblocking completed for {country}")
                print(f"Country {country} geo-unblocked! Freedom restored!")
            else:
                self.logger.info(f"No geo-block cache found for {country}, nothing to unblock")
                print(f"No geo-block data found for {country} - already free!")
        except Exception as e:
            self.logger.error(f"Geo-unblock failed for {country}: {str(e)}")
            print(f"Error: Geo-unblock failed for {country} - {str(e)}")

    def deploy_python_app(self):
        """Déploie l’application Python de gestion."""
        self.logger.info("Deploying enhanced Python management application")
        if subprocess.run("command -v python3", shell=True, stdout=subprocess.DEVNULL).returncode != 0:
            self.logger.error("Python3 not installed")
            sys.exit(1)

        LOG_APP_FILE.parent.mkdir(parents=True, exist_ok=True)
        LOG_APP_FILE.touch()
        os.chmod(LOG_APP_FILE, 0o664)
        os.chown(LOG_APP_FILE, 0, 4)  # root:adm
        APP_DIR.mkdir(parents=True, exist_ok=True)
        for subdir in ["core", "utils", "fallback"]:
            (APP_DIR / subdir).mkdir(exist_ok=True)

        self._run_cmd(f"python3 -m venv {APP_DIR}/venv", "Failed to create virtual environment")
        venv_python = f"{APP_DIR}/venv/bin/python3"
        self._run_cmd(f"{APP_DIR}/venv/bin/pip install --upgrade pip", "Failed to update pip")
        self._run_cmd(f"{APP_DIR}/venv/bin/pip install psutil==5.9.5 netifaces==0.11.0 tabulate==0.9.0 requests==2.31.0 colorama==0.4.6", "Failed to install Python dependencies")

        self._write_if_changed(APP_DIR / f"{APP_NAME}.py", """\
#!/usr/bin/env python3
import argparse
import sys
from core.firewall import FirewallManager
from utils.logger import Logger

VERSION = "2.2"
LOG_FILE = "/var/log/firewall_manager/app.log"
logger = Logger(LOG_FILE)

def main():
    parser = argparse.ArgumentParser(description="Firewall Manager - Advanced Security Management")
    subparsers = parser.add_subparsers(dest="command", help="Available commands")

    status_parser = subparsers.add_parser("status", help="Display current firewall status")
    ban_parser = subparsers.add_parser("ban", help="Ban a specific IP address")
    ban_parser.add_argument("ip", help="IP address to ban")
    ban_parser.add_argument("--timeout", type=int, default=7200, help="Ban duration in seconds")
    unban_parser = subparsers.add_parser("unban", help="Unban a specific IP address")
    unban_parser.add_argument("ip", help="IP address to unban")
    stats_parser = subparsers.add_parser("stats", help="Display network and rule statistics")
    perf_parser = subparsers.add_parser("perf", help="Display network performance")
    perf_parser.add_argument("--interval", type=int, default=5, help="Interval in seconds")
    logs_parser = subparsers.add_parser("logs", help="Monitor firewall logs in real-time")
    logs_parser.add_argument("--duration", type=int, default=60, help="Monitoring duration in seconds")
    logs_parser.add_argument("--keywords", help="Comma-separated list of keywords to filter logs")
    detect_parser = subparsers.add_parser("detect", help="Detect and ban threats automatically")
    detect_parser.add_argument("--duration", type=int, default=300, help="Detection duration in seconds")
    detect_parser.add_argument("--threshold", type=int, default=10, help="Threshold for banning an IP")
    geo_parser = subparsers.add_parser("geo-block", help="Block traffic from a specific country")
    geo_parser.add_argument("country", help="Country code (e.g., CN, RU)")
    geo_unblock_parser = subparsers.add_parser("geo-unblock", help="Unblock traffic from a specific country")
    geo_unblock_parser.add_argument("country", help="Country code (e.g., CN, RU)")

    args = parser.parse_args()
    fw = FirewallManager(logger)

    if args.command == "status":
        fw.show_status()
    elif args.command == "ban":
        fw.ban_ip(args.ip, args.timeout)
    elif args.command == "unban":
        fw.unban_ip(args.ip)
    elif args.command == "stats":
        fw.show_stats()
    elif args.command == "perf":
        fw.show_network_performance(args.interval)
    elif args.command == "logs":
        fw.monitor_logs(args.duration, args.keywords.split(",") if args.keywords else None)
    elif args.command == "detect":
        fw.detect_threats(args.duration, args.threshold)
    elif args.command == "geo-block":
        fw.geo_block(args.country)
    elif args.command == "geo-unblock":
        fw.geo_unblock(args.country)
    else:
        parser.print_help()
        sys.exit(1)

if __name__ == "__main__":
    print(f"Firewall Manager {VERSION} - Security Solution Initialized")
    main()
""")

        self._write_if_changed(APP_DIR / "core" / "firewall.py", """\
import subprocess
import netifaces
import time
import os
import re
import requests
from requests.adapters import HTTPAdapter
from requests.packages.urllib3.util.retry import Retry
from tabulate import tabulate
from concurrent.futures import ThreadPoolExecutor
import ipaddress
import psutil
from colorama import init, Fore, Style
init()

class FirewallManager:
    def __init__(self, logger):
        self.logger = logger
        self.banned_set = "fm_banned"
        self.geo_set = "fm_geo_block"
        self.iptables = "iptables"
        self.ipset = "ipset"
        self.log_files = {
            "Setup": "/var/log/firewall_manager/setup.log",
            "App": "/var/log/firewall_manager/app.log",
            "System": "/var/log/kern.log" if os.path.exists("/var/log/kern.log") else "/var/log/syslog"
        }

    def _run_cmd(self, cmd, error_msg):
        try:
            result = subprocess.run(cmd, shell=True, check=True, capture_output=True, text=True)
            return result.stdout.strip()
        except subprocess.CalledProcessError as e:
            if error_msg:
                self.logger.error(f"{error_msg}: {e}")
            return None

    def _validate_ip_or_cidr(self, ip):
        try:
            ipaddress.ip_network(ip, strict=False)
            return True
        except ValueError:
            self.logger.error(f"Invalid IP/CIDR '{ip}'")
            return False

    def show_status(self):
        interfaces = netifaces.interfaces()
        self.logger.info("Displaying firewall status")
        banned = self._run_cmd(f"{self.ipset} list {self.banned_set}", "Failed to list banned IPs") or "None"
        geo_blocked = self._run_cmd(f"{self.ipset} list {self.geo_set}", "Failed to list geo-blocks") or "None"
        rules = self._run_cmd(f"{self.iptables} -L -v --line-numbers", "Failed to list rules") or "No rules available"
        print(f"{Fore.GREEN}Interfaces:{Style.RESET_ALL} {', '.join(interfaces)}")
        print(f"{Fore.RED}Banned IPs:{Style.RESET_ALL}\n{banned}")
        print(f"{Fore.YELLOW}Geo-blocked Networks:{Style.RESET_ALL}\n{geo_blocked}")
        print(f"{Fore.CYAN}Rules:{Style.RESET_ALL}\n{rules}")

    def ban_ip(self, ip, timeout):
        if not re.match(r"^[0-9]+\\.[0-9]+\\.[0-9]+\\.[0-9]+$", ip):
            print(f"Invalid IP address: {ip}")
            return
        cmd = f"{self.ipset} add {self.banned_set} {ip} timeout {timeout} -exist"
        if self._run_cmd(cmd, f"Failed to ban IP {ip}"):
            self.logger.info(f"IP {ip} banned for {timeout} seconds")
            print(f"IP {ip} banned for {timeout} seconds")

    def unban_ip(self, ip):
        if not re.match(r"^[0-9]+\\.[0-9]+\\.[0-9]+\\.[0-9]+$", ip):
            print(f"Invalid IP address: {ip}")
            return
        cmd = f"{self.ipset} del {self.banned_set} {ip}"
        if self._run_cmd(cmd, f"Failed to unban IP {ip}"):
            self.logger.info(f"IP {ip} unbanned")
            print(f"IP {ip} unbanned")

    def show_stats(self):
        stats = []
        rules = self._run_cmd(f"{self.iptables} -L INPUT -v -n --line-numbers", "Failed to retrieve INPUT chain statistics")
        if rules:
            for line in rules.splitlines()[2:]:
                cols = line.split()
                if len(cols) > 8:
                    stats.append([cols[0], cols[1], cols[2], cols[8]])
            print(tabulate(stats, headers=["Num", "Packets", "Bytes", "Target"], tablefmt="fancy_grid"))
        else:
            print("No statistics available for INPUT chain.")

    def show_network_performance(self, interval=5):
        self.logger.info(f"Displaying network performance over {interval}s")
        print(f"--- Network Performance ({interval}s) ---")
        net_io_start = psutil.net_io_counters()
        time.sleep(interval)
        net_io_end = psutil.net_io_counters()
        
        bytes_sent = (net_io_end.bytes_sent - net_io_start.bytes_sent) / interval
        bytes_recv = (net_io_end.bytes_recv - net_io_start.bytes_recv) / interval
        
        stats = [
            ["Upload Throughput", f"{bytes_sent / 1024:.2f} KB/s"],
            ["Download Throughput", f"{bytes_recv / 1024:.2f} KB/s"],
            ["Packets Dropped", net_io_end.dropin + net_io_end.dropout]
        ]
        print(tabulate(stats, headers=["Metric", "Value"], tablefmt="fancy_grid"))

    def monitor_logs(self, duration=60, keywords=None):
        keywords = keywords or ["banned", "SYN_FLOOD", "PORTSCAN", "UDP_FLOOD", "Error", "DROP", "ACCEPT", "Geo-block"]
        self.logger.info(f"Monitoring logs for {duration} seconds with keywords: {','.join(keywords)}")
        print(f"--- Real-time Log Monitoring ({duration}s) ---")
        for log_name, log_file in self.log_files.items():
            if not os.path.exists(log_file):
                print(f"{log_name}: {log_file} not found")
            print(f"\n[{log_name} - {log_file}]")
            cmd = f"tail -n 20 {log_file} | grep -E '{'|'.join(keywords)}'; timeout {duration}s tail -f {log_file} | grep -E '{'|'.join(keywords)}'"
            try:
                output = subprocess.check_output(cmd, shell=True, text=True, stderr=subprocess.DEVNULL)
                for line in output.splitlines():
                    print(f"{time.strftime('%H:%M:%S')} - {line.strip()}")
            except subprocess.CalledProcessError as e:
                if e.returncode != 124:  # Timeout OK
                    self.logger.error(f"Log monitoring error for {log_file}: {e}")
                    print(f"Error monitoring {log_file}")
        print("--- Monitoring Completed ---")

    def detect_threats(self, duration=300, threshold=10):
        self.logger.info(f"Threat detection enabled for {duration}s with threshold {threshold} hits")
        print(f"--- Threat Detection Running ({duration}s) ---")
        ip_counts = {}
        start_time = time.time()
        
        cmd = f"timeout {duration}s tail -f {self.log_files['System']} | grep -E 'SYN_FLOOD|PORTSCAN|UDP_FLOOD'"
        process = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE, text=True)
        
        while time.time() - start_time < duration:
            line = process.stdout.readline().strip()
            if not line:
                continue
            ip_match = re.search(r"SRC=(\d+\\.\\d+\\.\\d+\\.\\d+)", line)
            if ip_match:
                ip = ip_match.group(1)
                ip_counts[ip] = ip_counts.get(ip, 0) + 1
                if ip_counts[ip] >= threshold:
                    self.ban_ip(ip, timeout=86400)
                    self.logger.info(f"Threat detected: {ip} auto-banned ({ip_counts[ip]} hits)")
                    print(f"{Fore.RED}Threat detected: {ip} banned!{Style.RESET_ALL}")
        
        process.terminate()
        print("--- Detection Completed ---")

    def geo_block(self, country):
        self.logger.info(f"Geo-blocking country: {country}")
        url = f"https://www.ipdeny.com/ipblocks/data/countries/{country.lower()}.zone"
        cache_file = f"/var/cache/firewall_manager/{country}.zone"
        os.makedirs(os.path.dirname(cache_file), exist_ok=True)
        session = requests.Session()
        retries = Retry(total=3, backoff_factor=1, status_forcelist=[500, 502, 503, 504])
        session.mount("https://", HTTPAdapter(max_retries=retries))
        try:
            if not os.path.exists(cache_file) or os.path.getmtime(cache_file) < time.time() - 7*24*3600:
                response = session.get(url, timeout=5)
                response.raise_for_status()
                with open(cache_file, "w") as f:
                    f.write(response.text)
            with open(cache_file) as f, ThreadPoolExecutor(max_workers=8) as executor:
                executor.map(lambda cidr: self._run_cmd(f"{self.ipset} add {self.geo_set} {cidr.strip()} -exist", f"Failed to add {cidr.strip()} to geo-block"),
                             [cidr for cidr in f if self._validate_ip_or_cidr(cidr.strip())])
            self._run_cmd(f"{self.iptables} -A INPUT -m set --match-set {self.geo_set} src -j DROP", f"Failed to apply geo-block rule for {country}")
            self.logger.info(f"Geo-blocking enabled for {country}")
            print(f"Country {country} geo-blocked! Locked and loaded!")
        except Exception as e:
            self.logger.error(f"Geo-block failed for {country}: {str(e)}")
            print(f"Error: Geo-block failed for {country} - {str(e)}")

    def geo_unblock(self, country):
        self.logger.info(f"Unblocking geo-blocked country: {country}")
        cache_file = f"/var/cache/firewall_manager/{country}.zone"
        try:
            if os.path.exists(cache_file):
                with open(cache_file) as f, ThreadPoolExecutor(max_workers=8) as executor:
                    executor.map(lambda cidr: self._run_cmd(f"{self.ipset} del {self.geo_set} {cidr.strip()}", f"Failed to remove {cidr.strip()} from geo-block"),
                                 [cidr for cidr in f if self._validate_ip_or_cidr(cidr.strip())])
                os.unlink(cache_file)
                self.logger.info(f"Cleared geo-block cache for {country}")
                geo_count = self._run_cmd(f"{self.ipset} list {self.geo_set} | grep -c '^[0-9]'", None) or "0"
                if geo_count == "0":
                    self._run_cmd(f"{self.iptables} -D INPUT -m set --match-set {self.geo_set} src -j DROP", "Failed to remove geo-block rule")
                    self._run_cmd(f"{self.ipset} destroy {self.geo_set}", "Failed to destroy empty geo-block set")
                    self.logger.info("Geo-block set emptied and rule removed")
                self.logger.info(f"Geo-unblocking completed for {country}")
                print(f"Country {country} geo-unblocked! Freedom restored!")
            else:
                self.logger.info(f"No geo-block cache found for {country}, nothing to unblock")
                print(f"No geo-block data found for {country} - already free!")
        except Exception as e:
            self.logger.error(f"Geo-unblock failed for {country}: {str(e)}")
            print(f"Error: Geo-unblock failed for {country} - {str(e)}")
""")

        self._write_if_changed(APP_DIR / "utils" / "logger.py", """\
import os
import logging
import socket
import stat

class Logger:
    def __init__(self, log_file):
        self.log_file = log_file
        self.hostname = socket.gethostname()
        os.makedirs(os.path.dirname(log_file), exist_ok=True)
        logging.basicConfig(filename=log_file, level=logging.INFO,
                            format="%(asctime)s [%(hostname)s] - [%(levelname)s] - %(message)s")
        self.logger = logging.getLogger()
        self._ensure_permissions()

    def _ensure_permissions(self):
        desired_mode = 0o664
        if os.path.exists(self.log_file):
            current_mode = stat.S_IMODE(os.stat(self.log_file).st_mode)
            if current_mode != desired_mode:
                os.chmod(self.log_file, desired_mode)
                self.logger.info(f"Adjusted permissions on {self.log_file} to {oct(desired_mode)}")

    def info(self, message):
        self.logger.info(message, extra={"hostname": self.hostname})

    def error(self, message):
        self.logger.error(message, extra={"hostname": self.hostname})
""")

        os.chmod(f"{APP_DIR}/{APP_NAME}.py", 0o755)
        subprocess.run(f"chown -R root:root {APP_DIR}", shell=True)

        self._write_if_changed(f"/etc/systemd/system/{APP_NAME}.service", f"""\
[Unit]
Description=Firewall Manager Python Application
After=network.target firewall-manager.service
[Service]
ExecStart={APP_DIR}/venv/bin/python3 {APP_DIR}/{APP_NAME}.py status
Restart=on-failure
User=root
Environment="PYTHONUNBUFFERED=1"
[Install]
WantedBy=multi-user.target
""")
        self._run_cmd("systemctl daemon-reload")
        self._run_cmd(f"systemctl enable {APP_NAME}.service", f"Failed to enable {APP_NAME}.service")
        self._run_cmd(f"systemctl start {APP_NAME}.service", f"Failed to start {APP_NAME}.service")
        self._check_service(APP_NAME)
        self.logger.info("Python application deployed and service activated")
        self._run_cmd(f"{venv_python} {APP_DIR}/{APP_NAME}.py --help", "Python application verification failed")

def main():
    parser = argparse.ArgumentParser(description="Firewall Manager - Advanced Security Management",
                                     usage="%(prog)s [config_file] [--mode normal|debug|test|dry-run] [command]")
    parser.add_argument("config_file", nargs="?", default=CONFIG_FILE, help="Path to configuration file")
    parser.add_argument("--mode", choices=["normal", "debug", "test", "dry-run"], default="normal", help="Operation mode")
    subparsers = parser.add_subparsers(dest="command", help="Available commands")

    subparsers.add_parser("init", help="Initialize the firewall")
    ban_parser = subparsers.add_parser("ban", help="Ban a specific IP address")
    ban_parser.add_argument("ip", help="IP address to ban")
    ban_parser.add_argument("--timeout", type=int, default=7200, help="Ban duration in seconds")
    unban_parser = subparsers.add_parser("unban", help="Unban a specific IP address")
    unban_parser.add_argument("ip", help="IP address to unban")
    subparsers.add_parser("status", help="Display current firewall status")
    subparsers.add_parser("stats", help="Display network and rule statistics")
    perf_parser = subparsers.add_parser("perf", help="Display network performance")
    perf_parser.add_argument("--interval", type=int, default=5, help="Interval in seconds")
    logs_parser = subparsers.add_parser("logs", help="Monitor firewall logs in real-time")
    logs_parser.add_argument("--duration", type=int, default=60, help="Monitoring duration in seconds")
    logs_parser.add_argument("--keywords", help="Comma-separated list of keywords to filter logs")
    detect_parser = subparsers.add_parser("detect", help="Detect and ban threats automatically")
    detect_parser.add_argument("--duration", type=int, default=300, help="Detection duration in seconds")
    detect_parser.add_argument("--threshold", type=int, default=10, help="Threshold for banning an IP")
    geo_parser = subparsers.add_parser("geo-block", help="Block traffic from a specific country")
    geo_parser.add_argument("country", help="Country code (e.g., CN, RU)")
    geo_unblock_parser = subparsers.add_parser("geo-unblock", help="Unblock traffic from a specific country")
    geo_unblock_parser.add_argument("country", help="Country code (e.g., CN, RU)")
    subparsers.add_parser("deploy", help="Deploy the Python application")

    args = parser.parse_args()
    logger = Logger(LOG_FILE)
    mode = "test" if args.mode == "dry-run" else args.mode
    fw = FirewallManager(logger, mode, args.config_file)

    if args.command == "init":
        fw.initialize_firewall()
        fw.deploy_python_app()
        net_status = fw._check_network()
        if net_status != "Up":
            logger.error("Network down after setup, check VPN or connectivity!")
            sys.exit(1)
        banned_count = fw._run_cmd(f"{fw.ipset} list {BANNED_SET} | grep -c '^[0-9]'", None, fatal=False) or "0"
        geo_count = fw._run_cmd(f"{fw.ipset} list {GEO_SET} | grep -c '^[0-9]'", None, fatal=False) or "0"
        iface = fw.interfaces["main"]
        vpn_ifaces = fw.interfaces["vpn"]
        print("--------------------------------------------------")
        print("Firewall Manager v2.2 - Deployment Summary:")
        print(f" - Interfaces: IFACE={iface}, VPN={vpn_ifaces}")
        print(f" - Banned IPs: {banned_count}")
        print(f" - Geo-blocked Networks: {geo_count}")
        print(f" - Network Status: {net_status}")
        print(f" - Management Tool: {APP_DIR}/{APP_NAME}.py [status|ban|unban|stats|perf|logs|detect|geo-block|geo-unblock]")
        print(f" - Log Files: {LOG_FILE} (setup), {LOG_APP_FILE} (app)")
        print("Deployment completed successfully. System security enhanced!")
        print("--------------------------------------------------")
    elif args.command == "ban":
        fw.ban_ip(args.ip, args.timeout)
    elif args.command == "unban":
        fw.unban_ip(args.ip)
    elif args.command == "status":
        fw.show_status()
    elif args.command == "stats":
        fw.show_stats()
    elif args.command == "perf":
        fw.show_network_performance(args.interval)
    elif args.command == "logs":
        fw.monitor_logs(args.duration, args.keywords.split(",") if args.keywords else None)
    elif args.command == "detect":
        fw.detect_threats(args.duration, args.threshold)
    elif args.command == "geo-block":
        fw.geo_block(args.country)
    elif args.command == "geo-unblock":
        fw.geo_unblock(args.country)
    elif args.command == "deploy":
        fw.deploy_python_app()
    else:
        parser.print_help()
        sys.exit(1)

if __name__ == "__main__":
    print("--------------------------------------------------")
    print("Firewall Manager v2.2 - Deployment Initiated")
    print(f"Usage: {sys.argv[0]} [config_file] [--mode normal|debug|test|dry-run] [command]")
    print("--------------------------------------------------")
    main()

Le pire c'est que je suis assez bon avec iptables et des scripts bateaux, en revanche, traduire mes délires en python, c'est bien au-dessus de ce que je sais faire ou avoir envie de faire
C'est pour tester, et enerver les codeurs humains avec 10 années de background techniques

Dernière modification par #hehedotcom\'isback (Le 17/03/2025, à 16:01)

---

../

#hehedotcom\'isback

Re : aujourd'hui, j'ai communiqué avec Grok

Si vous doutez des fonctionnalités, copier-coller le script, et demander à une IA de votre choix, de synthétiser le fonctionnement

---

../

#hehedotcom\'isback

Re : aujourd'hui, j'ai communiqué avec Grok

J'ai demandé à gemini, de vérifier le script et voilà son analyse :

détaille moi le fonctionnement du script ci-dessous et vérifie d'éventuelles failles !
Absolument ! Décomposons ce script Python, analysons ses fonctionnalités et identifions les potentielles failles de sécurité ou les améliorations possibles.

Fonctionnalités Principales :

    Gestion du Pare-feu (iptables et ipset) :
        Le script configure et gère les règles du pare-feu à l'aide d'iptables et d'ipset.
        Il prend en charge le blocage et le déblocage d'adresses IP spécifiques, ainsi que le blocage géographique par pays.
        Il gère les règles IPv4 et tente de gérer les règles IPv6 si ip6tables est disponible.
    Fail2ban Intégré :
        Le script configure Fail2ban pour surveiller les logs du système à la recherche de tentatives d'attaques (SYN flood, port scans, UDP floods).
        Il bannit automatiquement les adresses IP suspectes en utilisant ipset.
    Gestion des Logs :
        Le script utilise le module logging pour enregistrer les événements et les erreurs dans des fichiers de log dédiés.
        Il gère la rotation des logs avec logrotate.
        Il gere les permissions des fichiers de log.
    Blocage Géographique :
        Il permet de bloquer le trafic provenant de pays spécifiques en utilisant les listes d'adresses IP fournies par ipdeny.com.
        Il est prévu d'implémenter un fallback avec maxmind, mais il manque la clé API.
    Surveillance et Statistiques :
        Le script fournit des fonctions pour afficher l'état du pare-feu, les statistiques des règles réseau et les performances réseau en temps réel.
        Il offre également une fonction de surveillance des logs avec des mots-clés personnalisables.
        Il detecte les menaces et bannit automatiquement les ip suspectes.
    Configuration Centralisée :
        Les paramètres de configuration sont chargés à partir d'un fichier (/etc/firewall_manager.conf), ce qui facilite la personnalisation.
    Gestion des Interfaces Réseau :
        Le script détecte automatiquement les interfaces réseau principales, VPN et LXC.
    Gestion des sauvegardes et restaurations :
        Le script effectue des sauvegardes avant de modifier les règles du pare-feu.
        Il permet de restaurer les sauvegardes en cas d'échec.
    Gestion des dépendances :
        Le script vérifie et installe les dépendances nécessaires (iptables, ipset, fail2ban, etc.).
        Il vérifie les versions minimales requises des outils.
    Gestion des erreurs :
        Le script gère les erreurs de manière robuste avec des messages de log détaillés.
        Il utilise try...except pour capturer les exceptions et éviter les plantages.
    Gestion des couleurs :
        Le script utilise la librairie colorama pour l'affichage coloré des informations.

Failles et Améliorations Potentielles :

    Sécurité du Blocage Géographique :
        Le blocage géographique basé sur les listes d'ipdeny.com peut ne pas être précis à 100 %. Les adresses IP peuvent changer et les listes peuvent ne pas être à jour.
        L'implémentation de maxmind est incomplète, et la clé API n'est pas gérée.
    Gestion des Erreurs :
        Bien que le script gère les erreurs, certaines opérations pourraient bénéficier de vérifications plus approfondies.
        Par exemple, la validation des entrées utilisateur (adresses IP, ports) pourrait être renforcée.
    Sécurité des Logs :
        Bien que les permissions des fichiers de log soient gérées, il est important de s'assurer que les logs ne contiennent pas d'informations sensibles.
        Le chemin d'accès aux logs devrait être configurable.
    Gestion des Dépendances :
        Le script suppose que le système utilise apt, yum ou pacman. Il serait préférable de gérer les autres gestionnaires de paquets.
    Complexité :
        Le script est assez complexe, ce qui peut rendre la maintenance difficile. Il pourrait être divisé en modules plus petits et plus spécialisés.
    Gestion des interfaces réseau :
        La détection des interfaces réseau pourrait être plus robuste, en tenant compte des configurations réseau complexes.
    Gestion des erreurs ipset et iptables :
        Les erreurs provenant d'ipset et iptables devraient être analysées plus en détail pour fournir des informations plus précises.
    Gestion des configurations par defaut :
        Les configurations par défaut devraient être documentées plus clairement.
    Gestion des droits :
        La vérification des droits root est effectuée au début du script, mais il serait préférable de vérifier les droits avant chaque opération sensible.
    Gestion des variables globales :
        L'utilisation de variables globales pourrait être réduite en utilisant des classes et des fonctions.
    Gestion des exceptions :
        Certaines exceptions sont gérées de manière générique. Il serait préférable de gérer les exceptions spécifiques pour fournir des messages d'erreur plus précis.
    Gestion des performances :
        L'utilisation de subprocess.run pour chaque commande peut être inefficace. Il serait préférable d'utiliser des outils plus performants pour gérer les commandes shell.
    Gestion des tests :
        Le script ne contient pas de tests unitaires. Il serait préférable d'ajouter des tests pour garantir la qualité du code.

En résumé :

Ce script est un outil puissant pour la gestion du pare-feu, mais il pourrait bénéficier d'améliorations en matière de sécurité, de robustesse et de maintenabilité. La modularisation, la documentation et les tests sont des aspects clés à considérer.

Dernière modification par #hehedotcom\'isback (Le 17/03/2025, à 17:06)

---

../

diesel

Re : aujourd'hui, j'ai communiqué avec Grok

iptables, c'est has been. Aujourd'hui, on utilise nftables.

Amicalement.

Jean-Marie

Dernière modification par diesel (Le 18/03/2025, à 08:53)

---

Je déteste qu'on cherche à me faire passer pour un con, j'y arrive déjà très bien tout seul.
Le mort, il sait pas qu'il est mort ; c'est pour les autres que c'est dur.................... Pour les cons, c'est pareil.

subccah

Re : aujourd'hui, j'ai communiqué avec Grok

Coucou les amis de l'écran.

J'ai dev

Non, t'as rien dev. Ni toi ni ton IA. Avec un langage de scripting comme Bash (ou comme Python si on l'utilise pour faire la même chose...) tu peux faire des scripts qui font des milliers de lignes, mais ça reste du scripting... ça sert à rien de plus que d'accumuler des lignes et des lignes, sans comprendre la moindre chose que vous faites toi ou l'IA, sans maîtriser la moindre subtilité d'un langage ou d'un algorithme.

Commence par dev un "Hello World" tout seul, sans aucune IA (juste avec la doc officielle) et ce sera déjà un magnifique grand pas vers la victoire. Attention, avec ces IA on a vite fait de s'imaginer des trucs totalement délirant, des mondes fabuleux qui ne sont...