firewall.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

# Educational Stateful Packet-Filtering Firewall
# A Python implementation of a stateful packet filtering firewall for educational purposes.
# This tool demonstrates fundamental concepts in network security and packet analysis.

import argparse
import json
import logging
import os
import sys
import time
from collections import defaultdict, namedtuple
from enum import Enum, auto
from ipaddress import IPv4Network, ip_address
from typing import Dict, List, Optional, Set, Tuple, Union

# Import Scapy - the essential packet manipulation library
try:
    from scapy.all import IP, TCP, UDP, ICMP, Ether, conf, sniff, wrpcap, raw
    from scapy.packet import Packet
    from scapy.utils import PcapReader
    HAS_SCAPY = True
except ImportError:
    HAS_SCAPY = False
    print("Error: scapy not installed. Install with 'pip install scapy'")
    sys.exit(1)

# Optional NetfilterQueue import for Linux systems - enables direct packet manipulation
try:
    from netfilterqueue import NetfilterQueue
    HAS_NFQ = True
except ImportError:
    HAS_NFQ = False

# Set up logging to track firewall activities
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger('firewall')

# Define common TCP flags for rule matching and logging
TCP_FLAGS = {
    'F': 'FIN',
    'S': 'SYN',
    'R': 'RST',
    'P': 'PSH',
    'A': 'ACK',
    'U': 'URG'
}

# Enumerations for firewall states and actions
class Direction(Enum):
    INBOUND = auto()  # Traffic coming into the system
    OUTBOUND = auto() # Traffic leaving the system

class Protocol(Enum):
    TCP = auto()  # Transmission Control Protocol
    UDP = auto()  # User Datagram Protocol
    ICMP = auto() # Internet Control Message Protocol
    ANY = auto()  # Any protocol (wildcard)

class Action(Enum):
    ACCEPT = auto() # Allow the packet to pass
    DROP = auto()   # Silently discard the packet
    REJECT = auto() # Discard and send error response

class TCPState(Enum):
    NEW = auto()         # New connection request
    ESTABLISHED = auto() # Connection established
    RELATED = auto()     # Related to an existing connection
    FIN_WAIT = auto()    # Connection closing (FIN sent)
    CLOSE_WAIT = auto()  # Waiting for connection close
    CLOSED = auto()      # Connection fully closed

# Data structures for rule management and connection tracking
Rule = namedtuple('Rule', [
    'id', 'direction', 'protocol',
    'src_ip', 'src_port', 'dst_ip', 'dst_port',
    'tcp_flags', 'action'
])

ConnectionTuple = namedtuple('ConnectionTuple', [
    'src_ip', 'src_port', 'dst_ip', 'dst_port', 'protocol'
])

class ConnectionState:
    """Tracks the state of a connection and associated statistics."""
    
    def __init__(self, state: TCPState = TCPState.NEW):
        self.state = state
        self.last_seen = time.time()
        self.packets_forward = 0
        self.packets_reverse = 0
        self.bytes_forward = 0
        self.bytes_reverse = 0
    
    def update(self, packet, is_forward: bool = True):
        """Updates connection statistics based on new packet data."""
        self.last_seen = time.time()
        
        if is_forward:
            self.packets_forward += 1
            if IP in packet:
                self.bytes_forward += packet[IP].len
        else:
            self.packets_reverse += 1
            if IP in packet:
                self.bytes_reverse += packet[IP].len

class StateTable:
    """Maintains a state table of all tracked network connections for stateful filtering."""
    
    def __init__(self, timeout: int = 300):
        self.connections: Dict[ConnectionTuple, ConnectionState] = {}
        self.timeout = timeout
        self.last_cleanup = time.time()
    
    def get_connection_tuple(self, packet) -> Optional[ConnectionTuple]:
        """Extracts unique connection identifiers from a packet."""
        if IP not in packet:
            return None
        
        src_ip = packet[IP].src
        dst_ip = packet[IP].dst
        
        if TCP in packet:
            protocol = Protocol.TCP
            src_port = packet[TCP].sport
            dst_port = packet[TCP].dport
        elif UDP in packet:
            protocol = Protocol.UDP
            src_port = packet[UDP].sport
            dst_port = packet[UDP].dport
        elif ICMP in packet:
            protocol = Protocol.ICMP
            src_port = 0  # ICMP doesn't use ports
            dst_port = 0
        else:
            return None
        
        return ConnectionTuple(src_ip, src_port, dst_ip, dst_port, protocol)
    
    def get_connection_status(self, packet) -> Tuple[Optional[ConnectionTuple], Optional[ConnectionState], bool]:
        """
        Determines if a packet belongs to an existing connection.
        Returns connection identifier, state object, and direction flag.
        """
        if IP not in packet:
            return None, None, False
        
        conn_tuple = self.get_connection_tuple(packet)
        if conn_tuple in self.connections:
            return conn_tuple, self.connections[conn_tuple], True
        
        # Check if this is return traffic for an existing connection
        if conn_tuple:
            rev_conn_tuple = ConnectionTuple(
                conn_tuple.dst_ip, conn_tuple.dst_port,
                conn_tuple.src_ip, conn_tuple.src_port,
                conn_tuple.protocol
            )
            if rev_conn_tuple in self.connections:
                return rev_conn_tuple, self.connections[rev_conn_tuple], False
        
        return conn_tuple, None, True
    
    def update_tcp_state(self, packet, conn_tuple: ConnectionTuple, is_forward: bool) -> None:
        """Updates TCP connection state machine based on packet flags."""
        if conn_tuple not in self.connections:
            if TCP in packet and packet[TCP].flags & 0x02:  # SYN flag
                self.connections[conn_tuple] = ConnectionState(TCPState.NEW)
                logger.debug(f"New TCP connection: {conn_tuple}")
            else:
                # For existing connections we discover mid-stream, treat as ESTABLISHED
                self.connections[conn_tuple] = ConnectionState(TCPState.ESTABLISHED)
                logger.debug(f"Discovered mid-stream connection: {conn_tuple}")
        
        state = self.connections[conn_tuple]
        state.update(packet, is_forward)
        
        if TCP in packet:
            flags = packet[TCP].flags
            
            # Track TCP state transitions using the standard TCP state machine
            if state.state == TCPState.NEW:
                if flags & 0x12 == 0x12:  # SYN+ACK
                    state.state = TCPState.ESTABLISHED
                    logger.debug(f"Connection established: {conn_tuple}")
            
            elif state.state == TCPState.ESTABLISHED:
                if flags & 0x01:  # FIN
                    state.state = TCPState.FIN_WAIT
                    logger.debug(f"Connection FIN received: {conn_tuple}")
                elif flags & 0x04:  # RST
                    state.state = TCPState.CLOSED
                    logger.debug(f"Connection reset: {conn_tuple}")
            
            elif state.state == TCPState.FIN_WAIT:
                if flags & 0x01:  # FIN
                    state.state = TCPState.CLOSE_WAIT
                    logger.debug(f"Connection closing: {conn_tuple}")
                elif flags & 0x04:  # RST
                    state.state = TCPState.CLOSED
                    logger.debug(f"Connection reset during close: {conn_tuple}")
            
            elif state.state == TCPState.CLOSE_WAIT:
                if flags & 0x04:  # RST
                    state.state = TCPState.CLOSED
                    logger.debug(f"Connection reset during wait: {conn_tuple}")
    
    def update_udp_state(self, packet, conn_tuple: ConnectionTuple, is_forward: bool) -> None:
        """Tracks UDP flows (simpler than TCP since UDP is connectionless)."""
        if conn_tuple not in self.connections:
            self.connections[conn_tuple] = ConnectionState(TCPState.ESTABLISHED)  # For UDP we use ESTABLISHED directly
            logger.debug(f"New UDP flow: {conn_tuple}")
        else:
            self.connections[conn_tuple].update(packet, is_forward)
    
    def update_icmp_state(self, packet, conn_tuple: ConnectionTuple, is_forward: bool) -> None:
        """Tracks ICMP message exchanges."""
        if conn_tuple not in self.connections:
            self.connections[conn_tuple] = ConnectionState(TCPState.ESTABLISHED)
            logger.debug(f"New ICMP flow: {conn_tuple}")
        else:
            self.connections[conn_tuple].update(packet, is_forward)
    
    def update_state(self, packet) -> Tuple[Optional[ConnectionTuple], bool]:
        """Updates the state table based on a new packet."""
        conn_tuple, state, is_forward = self.get_connection_status(packet)
        if not conn_tuple:
            return None, False
        
        if TCP in packet:
            self.update_tcp_state(packet, conn_tuple, is_forward)
        elif UDP in packet:
            self.update_udp_state(packet, conn_tuple, is_forward)
        elif ICMP in packet:
            self.update_icmp_state(packet, conn_tuple, is_forward)
        
        return conn_tuple, is_forward
    
    def cleanup(self) -> None:
        """Removes expired connections from the state table."""
        now = time.time()
        # Only cleanup periodically to reduce overhead
        if now - self.last_cleanup < 10:  # cleanup every 10 seconds
            return
        
        expired = []
        for conn_tuple, state in self.connections.items():
            if now - state.last_seen > self.timeout:
                expired.append(conn_tuple)
        
        for conn_tuple in expired:
            logger.debug(f"Connection expired: {conn_tuple}")
            del self.connections[conn_tuple]
        
        self.last_cleanup = now

class RuleEngine:
    """Processes and applies firewall rules to network traffic."""
    
    def __init__(self, rules_file: str, default_policy: Action = Action.DROP):
        self.rules: List[Rule] = []
        self.default_policy = default_policy
        self.load_rules(rules_file)
    
    def load_rules(self, rules_file: str) -> None:
        """Loads firewall rules from a JSON configuration file."""
        try:
            with open(rules_file, 'r') as f:
                config = json.load(f)
            
            # Set default policy if specified
            if 'default_policy' in config:
                policy = config['default_policy'].upper()
                if hasattr(Action, policy):
                    self.default_policy = getattr(Action, policy)
            
            # Load rules
            for rule_data in config.get('rules', []):
                rule = self._parse_rule(rule_data)
                if rule:
                    self.rules.append(rule)
            
            logger.info(f"Loaded {len(self.rules)} rules with default policy: {self.default_policy.name}")
        
        except FileNotFoundError:
            logger.error(f"Rules file not found: {rules_file}")
            sys.exit(1)
        except json.JSONDecodeError:
            logger.error(f"Invalid JSON in rules file: {rules_file}")
            sys.exit(1)
    
    def _parse_rule(self, rule_data: dict) -> Optional[Rule]:
        """Parses a single rule from its JSON representation."""
        try:
            rule_id = rule_data.get('id', 'unnamed')
            
            # Parse direction (inbound or outbound)
            direction_str = rule_data.get('direction', '').upper()
            if direction_str == 'IN':
                direction = Direction.INBOUND
            elif direction_str == 'OUT':
                direction = Direction.OUTBOUND
            else:
                logger.warning(f"Invalid direction in rule {rule_id}: {direction_str}")
                return None
            
            # Parse protocol (TCP, UDP, ICMP, ANY)
            protocol_str = rule_data.get('protocol', '').upper()
            if hasattr(Protocol, protocol_str):
                protocol = getattr(Protocol, protocol_str)
            else:
                logger.warning(f"Invalid protocol in rule {rule_id}: {protocol_str}")
                return None
            
            # Parse IP addresses (supports CIDR notation)
            try:
                src_ip = IPv4Network(rule_data.get('src_ip', '0.0.0.0/0'))
                dst_ip = IPv4Network(rule_data.get('dst_ip', '0.0.0.0/0'))
            except ValueError as e:
                logger.warning(f"Invalid IP address in rule {rule_id}: {e}")
                return None
            
            # Parse port specifications
            src_port = rule_data.get('src_port', 0)
            dst_port = rule_data.get('dst_port', 0)
            
            # Parse TCP flags for advanced filtering
            tcp_flags = rule_data.get('tcp_flags', '')
            
            # Parse action (ACCEPT, DROP, REJECT)
            action_str = rule_data.get('action', '').upper()
            if hasattr(Action, action_str):
                action = getattr(Action, action_str)
            else:
                logger.warning(f"Invalid action in rule {rule_id}: {action_str}")
                return None
            
            return Rule(
                rule_id, direction, protocol,
                src_ip, src_port, dst_ip, dst_port,
                tcp_flags, action
            )
        
        except Exception as e:
            logger.warning(f"Error parsing rule: {e}")
            return None
    
    def match_rule(self, packet, direction: Direction) -> Optional[Rule]:
        """Matches a packet against the rule set and returns the first matching rule."""
        if IP not in packet:
            return None
        
        src_ip = packet[IP].src
        dst_ip = packet[IP].dst
        
        # Extract protocol-specific information
        if TCP in packet:
            protocol = Protocol.TCP
            src_port = packet[TCP].sport
            dst_port = packet[TCP].dport
            tcp_flags = packet[TCP].flags
        elif UDP in packet:
            protocol = Protocol.UDP
            src_port = packet[UDP].sport
            dst_port = packet[UDP].dport
            tcp_flags = 0
        elif ICMP in packet:
            protocol = Protocol.ICMP
            src_port = 0
            dst_port = 0
            tcp_flags = 0
        else:
            # Unsupported protocol
            return None
        
        # Match rules in priority order (first match wins)
        for rule in self.rules:
            # Check direction
            if rule.direction != direction:
                continue
            
            # Check protocol
            if rule.protocol != Protocol.ANY and rule.protocol != protocol:
                continue
            
            # Check IP addresses
            src_ip_addr = ip_address(src_ip)
            dst_ip_addr = ip_address(dst_ip)
            if src_ip_addr not in rule.src_ip or dst_ip_addr not in rule.dst_ip:
                continue
            
            # Check ports
            if protocol in (Protocol.TCP, Protocol.UDP):
                if rule.src_port != 0 and rule.src_port != src_port:
                    continue
                if rule.dst_port != 0 and rule.dst_port != dst_port:
                    continue
            
            # Check TCP flags if specified
            if protocol == Protocol.TCP and rule.tcp_flags:
                match_flags = True
                for flag_char in rule.tcp_flags:
                    if flag_char in TCP_FLAGS:
                        flag_name = TCP_FLAGS[flag_char]
                        if flag_char == 'S' and not (tcp_flags & 0x02):  # SYN
                            match_flags = False
                            break
                        elif flag_char == 'A' and not (tcp_flags & 0x10):  # ACK
                            match_flags = False
                            break
                        elif flag_char == 'F' and not (tcp_flags & 0x01):  # FIN
                            match_flags = False
                            break
                        elif flag_char == 'R' and not (tcp_flags & 0x04):  # RST
                            match_flags = False
                            break
                
                if not match_flags:
                    continue
            
            # All conditions matched, return this rule
            return rule
        
        # No matching rule found
        return None

class Firewall:
    """Main firewall class that processes packets and applies rules."""
    
    def __init__(self, rules_file: str, log_file: str = None):
        # Initialize the state table and rule engine
        self.state_table = StateTable()
        self.rule_engine = RuleEngine(rules_file)
        
        # Set up additional logging if requested
        if log_file:
            file_handler = logging.FileHandler(log_file)
            file_handler.setFormatter(logging.Formatter(
                '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
            ))
            logger.addHandler(file_handler)
    
    def _get_protocol_name(self, packet) -> str:
        """Gets a human-readable protocol name from a packet."""
        if IP not in packet:
            return "UNKNOWN"
        
        proto_num = packet[IP].proto
        
        # Map common protocol numbers to names
        if proto_num == 1:
            return "ICMP"
        elif proto_num == 6:
            return "TCP"
        elif proto_num == 17:
            return "UDP"
        else:
            return f"PROTO:{proto_num}"
    
    def process_packet(self, packet) -> Action:
        """
        Core packet processing logic - determines if a packet should be allowed or blocked.
        This is the heart of the firewall's decision-making process.
        """
        # Skip non-IP packets
        if IP not in packet:
            return Action.ACCEPT
        
        # Determine packet direction (inbound vs outbound)
        # This is simplified for educational purposes
        direction = Direction.INBOUND
        
        # Try to determine direction based on local interfaces
        try:
            if hasattr(conf.route.routes, 'keys'):
                local_interfaces = conf.route.routes.keys()
                if packet[IP].src in local_interfaces:
                    direction = Direction.OUTBOUND
        except Exception:
            # If we can't determine direction, default to INBOUND
            pass
        
        # First, check if packet belongs to an established connection
        conn_tuple, is_forward = self.state_table.update_state(packet)
        
        # Allow established TCP connections (stateful filtering)
        if conn_tuple and conn_tuple.protocol == Protocol.TCP:
            state = self.state_table.connections[conn_tuple]
            if state.state == TCPState.ESTABLISHED:
                # Allow established connections
                logger.debug(f"ACCEPT: Established connection {conn_tuple}")
                return Action.ACCEPT
        
        # For connectionless protocols, use pseudo-stateful behavior
        if conn_tuple and not is_forward and conn_tuple.protocol in (Protocol.UDP, Protocol.ICMP):
            logger.debug(f"ACCEPT: Return traffic for {conn_tuple.protocol.name}")
            return Action.ACCEPT
        
        # For new connections or non-tracked packets, apply ruleset
        rule = self.rule_engine.match_rule(packet, direction)
        
        if rule:
            # Get protocol name for logging
            proto_name = self._get_protocol_name(packet)
            
            action_msg = f"{rule.action.name}: {rule.id} - {packet[IP].src}:{getattr(packet, 'sport', 0)} -> {packet[IP].dst}:{getattr(packet, 'dport', 0)} ({proto_name})"
            if rule.action == Action.ACCEPT:
                logger.info(action_msg)
            else:
                logger.warning(action_msg)
            
            return rule.action
        
        # Apply default policy if no rules matched
        proto_name = self._get_protocol_name(packet)
        default_msg = f"{self.rule_engine.default_policy.name}: Default - {packet[IP].src}:{getattr(packet, 'sport', 0)} -> {packet[IP].dst}:{getattr(packet, 'dport', 0)} ({proto_name})"
        
        if self.rule_engine.default_policy == Action.DROP:
            logger.warning(default_msg)
        else:
            logger.info(default_msg)
        return self.rule_engine.default_policy
        
    def packet_callback(self, packet) -> None:
        """Callback function for processing each captured packet."""
        try:
            # Process the packet
            action = self.process_packet(packet)
            
            # Clean up expired connections periodically
            self.state_table.cleanup()
        except Exception as e:
            logger.error(f"Error processing packet: {e}")
            # Continue processing other packets
    
    def run_on_interface(self, interface: str) -> None:
        """Runs the firewall on a live network interface."""
        logger.info(f"Starting firewall on interface: {interface}")
        try:
            # Check for admin privileges on Windows
            if os.name == 'nt':
                import ctypes
                if not ctypes.windll.shell32.IsUserAnAdmin():
                    logger.warning("On Windows, administrator privileges are required for packet capture.")
                    logger.warning("Please run this script as Administrator.")
            
            # Start capturing and processing packets
            logger.info("Starting packet capture (Press Ctrl+C to stop)...")
            sniff(iface=interface, prn=self.packet_callback, store=0)
        except KeyboardInterrupt:
            logger.info("Firewall stopped by user")
        except Exception as e:
            logger.error(f"Error sniffing packets: {e}")
            if "winpcap is not installed" in str(e).lower() or "npcap" in str(e).lower():
                logger.error("Packet capture requires WinPcap or Npcap on Windows.")
                logger.error("Please install Npcap from: https://npcap.com/#download")
                logger.error("Make sure to select 'Install Npcap in WinPcap API-compatible Mode' during installation.")
    
    def run_on_pcap(self, pcap_file: str) -> None:
        """Runs the firewall on a pre-captured PCAP file for analysis."""
        logger.info(f"Processing PCAP file: {pcap_file}")
        try:
            sniff(offline=pcap_file, prn=self.packet_callback, store=0)
            logger.info("Finished processing PCAP file")
        except Exception as e:
            logger.error(f"Error processing PCAP file: {e}")
    
    def run_with_nfqueue(self, queue_num: int = 0) -> None:
        """
        Runs the firewall using Linux's NetfilterQueue for inline packet processing.
        This allows actual blocking of traffic on Linux systems.
        """
        if not HAS_NFQ:
            logger.error("NetfilterQueue not available. Install with 'pip install netfilterqueue'")
            sys.exit(1)
        
        def nfqueue_callback(nfpacket):
            """Process packet from NetfilterQueue and apply verdict."""
            try:
                data = nfpacket.get_payload()
                scapy_packet = IP(data)
                
                # Process the packet
                action = self.process_packet(scapy_packet)
                
                # Apply verdict (actually accept or drop the packet)
                if action == Action.ACCEPT:
                    nfpacket.accept()
                else:
                    nfpacket.drop()
                
                # Clean up expired connections periodically
                self.state_table.cleanup()
                
            except Exception as e:
                logger.error(f"Error processing packet: {e}")
                nfpacket.accept()  # Failsafe - accept on error
        
        # Set up the queue
        queue = NetfilterQueue()
        try:
            logger.info(f"Binding to NetfilterQueue: {queue_num}")
            queue.bind(queue_num, nfqueue_callback)
            
            # Provide helpful instructions for iptables setup
            logger.info("Ensure iptables rules are set up to direct traffic to the queue:")
            logger.info(f"  iptables -A INPUT -j NFQUEUE --queue-num {queue_num}")
            logger.info(f"  iptables -A OUTPUT -j NFQUEUE --queue-num {queue_num}")
            logger.info("Starting packet processing...")
            
            queue.run()
        
        except KeyboardInterrupt:
            logger.info("Firewall stopped by user")
        except Exception as e:
            logger.error(f"Error with NetfilterQueue: {e}")
        finally:
            queue.unbind()

def list_interfaces():
    """Lists available network interfaces for user selection."""
    try:
        from scapy.arch import get_windows_if_list, IFACES
        
        print("\nAvailable Network Interfaces:")
        print("=" * 60)
        
        if os.name == 'nt':  # Windows
            ifaces = get_windows_if_list()
            for i, iface in enumerate(ifaces):
                print(f"{i+1}. {iface.get('name', 'Unknown')} - {iface.get('description', 'N/A')}")
                print(f"   IP Address: {iface.get('ips', ['N/A'])[0] if iface.get('ips') else 'N/A'}")
                print(f"   Interface Name: {iface.get('name', 'Unknown')}")
                print("-" * 60)
        else:  # Linux/Unix
            for name, iface in sorted(IFACES.items()):
                print(f"Interface: {name}")
                print(f"   IP: {iface.ip if hasattr(iface, 'ip') else 'N/A'}")
                print("-" * 60)
    except Exception as e:
        print(f"Error listing interfaces: {e}")
        print("You might need to install WinPcap/Npcap on Windows or run with administrator privileges.")

def main():
    """Entry point for the firewall application."""
    parser = argparse.ArgumentParser(description='Python Stateful Packet-Filtering Firewall')
    
    # Define command-line options for different operating modes
    mode_group = parser.add_mutually_exclusive_group(required=True)
    mode_group.add_argument('--interface', '-i', help='Network interface to monitor')
    mode_group.add_argument('--pcap', '-p', help='PCAP file to process')
    mode_group.add_argument('--nfqueue', '-q', type=int, help='Use NetfilterQueue with specified queue number (Linux only)')
    mode_group.add_argument('--list-interfaces', '-l', action='store_true', help='List available network interfaces')
    
    # Additional configuration options
    parser.add_argument('--rules', '-r', help='JSON rules configuration file')
    parser.add_argument('--log', help='Log file path')
    parser.add_argument('--verbose', '-v', action='store_true', help='Enable verbose logging')
    args = parser.parse_args()
    
    # Handle listing interfaces separately
    if args.list_interfaces:
        list_interfaces()
        return 0
    
    # Check if rules file is provided for all other modes
    if not args.rules:
        print("Error: --rules/-r parameter is required unless using --list-interfaces")
        return 1
        
    # Configure logging level
    if args.verbose:
        logger.setLevel(logging.DEBUG)
    
    # Check dependencies
    if not HAS_SCAPY:
        print("Error: scapy not installed. Install with 'pip install scapy'")
        return 1
    
    if args.nfqueue and not HAS_NFQ:
        print("Error: NetfilterQueue not installed. Install with 'pip install netfilterqueue'")
        print("(NetfilterQueue only works on Linux systems.)")
        return 1
    
    # Print disclaimer
    print("\n" + "="*80)
    print("EDUCATIONAL FIREWALL DISCLAIMER".center(80))
    print("="*80)
    print("This firewall is for EDUCATIONAL PURPOSES ONLY.")
    print("It is NOT designed for production use and provides NO SECURITY GUARANTEES.")
    print("Use at your own risk.\n")
    
    try:
        # Initialize the firewall
        firewall = Firewall(args.rules, args.log)
        
        # Run the firewall in the selected mode
        if args.interface:
            # On Windows, check for admin privileges
            if os.name == 'nt':
                import ctypes
                if not ctypes.windll.shell32.IsUserAnAdmin():
                    print("WARNING: Administrator privileges required for packet capture on Windows.")
                    print("Consider restarting this command from an Administrator PowerShell/Command Prompt.")
                    print("Continuing anyway, but packet capture may fail...\n")
            
            firewall.run_on_interface(args.interface)
        elif args.pcap:
            firewall.run_on_pcap(args.pcap)
        elif args.nfqueue is not None:
            firewall.run_with_nfqueue(args.nfqueue)
        
        return 0
    
    except Exception as e:
        logger.critical(f"Fatal error: {e}")
        return 1

if __name__ == "__main__":
    sys.exit(main())