netguard_app.py

import customtkinter as ctk
import subprocess
import threading
import re
import json
from datetime import datetime
import tkinter.messagebox as messagebox
from PIL import Image

class NetGuardApp:
    def __init__(self):
        # Set theme
        ctk.set_appearance_mode("dark")
        ctk.set_default_color_theme("blue")
        
        self.root = ctk.CTk()
        self.root.title("🛡️ NetGuard Pro")
    
        
        self.root.geometry("645x 700")
        self.root.minsize(645, 600)
        self.root.maxsize(645, 800)        
        # Data storage
        self.internal_devices = []
        self.external_connections = []
        self.blocked_ips = set()
        
        self.setup_ui()
        self.load_blocked_ips()
        self.refresh_data()
        
    def setup_ui(self):
        # Main container
        main_frame = ctk.CTkFrame(self.root)
        main_frame.pack(fill="both", expand=True, padx=10, pady=10)

        # Title
        title_frame = ctk.CTkFrame(main_frame, fg_color="transparent")
        title_frame.pack(pady=(10, 20))

        try:
            icon_image = ctk.CTkImage(Image.open("icon.png"), size=(62, 62))
            icon_label = ctk.CTkLabel(title_frame, image=icon_image, text="")
            icon_label.pack(side="left", padx=(0, 10), pady=10)
        except (FileNotFoundError, ImportError):
            print("icon.png not found or PIL not installed, skipping icon.")

        title_label = ctk.CTkLabel(
            title_frame,
            text="NetGuard Pro - Network Monitor & Blocker",
            font=ctk.CTkFont(size=24, weight="bold")
        )
        title_label.pack(side="left")

        # Author
        author_label = ctk.CTkLabel(
            main_frame,
            text="Author: Wontfallo | GitHub: github.com/wontfallo | Twitter: @wontfallo",
            font=ctk.CTkFont(size=12)
        )
        author_label.pack(pady=(0, 20))

        # Control buttons
        control_frame = ctk.CTkFrame(main_frame)
        control_frame.pack(fill="x", padx=10, pady=(0, 10))

        refresh_btn = ctk.CTkButton(
            control_frame,
            text="🔄 Refresh",
            command=self.refresh_data,
            width=100
        )
        refresh_btn.pack(side="left", padx=10, pady=10)

        clear_blocks_btn = ctk.CTkButton(
            control_frame,
            text="🗑️ Clear All Blocks",
            command=self.clear_all_blocks,
            fg_color="red",
            hover_color="darkred",
            width=150
        )
        clear_blocks_btn.pack(side="left", padx=10, pady=10)

        # Status label
        self.status_label = ctk.CTkLabel(
            control_frame,
            text="Ready",
            font=ctk.CTkFont(size=12)
        )
        self.status_label.pack(side="right", padx=20, pady=10)

        # Manual entry section
        manual_frame = ctk.CTkFrame(main_frame)
        manual_frame.pack(fill="x", padx=10, pady=(0, 10))

        # Manual entry label
        manual_label = ctk.CTkLabel(
            manual_frame,
            text="📝 Manual IP Entry & Filters",
            font=ctk.CTkFont(size=16, weight="bold")
        )
        manual_label.pack(pady=(10, 5))

        # Filters section
        filters_frame = ctk.CTkFrame(manual_frame)
        filters_frame.pack(fill="x", padx=10, pady=(0, 10))

        # Multicast filter toggle
        self.filter_multicast = ctk.BooleanVar(value=True)  # Default to filtering multicast
        multicast_filter_checkbox = ctk.CTkCheckBox(
            filters_frame,
            text="🚫 Filter Out Multicast IPs (224.0.0.0 - 239.255.255.255)",
            variable=self.filter_multicast,
            command=self.refresh_data
        )
        multicast_filter_checkbox.pack(side="top", padx=10, pady=5)

        # Manual entry content
        entry_content_frame = ctk.CTkFrame(manual_frame)
        entry_content_frame.pack(fill="x", padx=10, pady=(0, 10))

        # IP address entry
        ip_entry_label = ctk.CTkLabel(entry_content_frame, text="IP Address:")
        ip_entry_label.pack(side="left", padx=(10, 5), pady=5)

        self.manual_ip_entry = ctk.CTkEntry(
            entry_content_frame,
            placeholder_text="192.168.1.100",
            width=200
        )
        self.manual_ip_entry.pack(side="left", padx=(0, 20), pady=5)

        # Block custom IP button
        block_custom_btn = ctk.CTkButton(
            entry_content_frame,
            text="✅ Block Custom IP",
            command=self.block_custom_ip,
            fg_color="red",
            hover_color="darkred",
            width=150
        )
        block_custom_btn.pack(side="left", padx=(0, 10), pady=5)

        # Clear entry button
        clear_entry_btn = ctk.CTkButton(
            entry_content_frame,
            text="🗑️ Clear",
            command=self.clear_custom_entry,
            width=80
        )
        clear_entry_btn.pack(side="left", padx=5, pady=5)
        
        # Tabbed interface
        self.tabview = ctk.CTkTabview(main_frame)
        self.tabview.pack(fill="both", expand=True, padx=10, pady=10)
        
        # Internal devices tab
        self.tabview.add("🏠 Internal Devices")
        self.setup_internal_tab()
        
        # External connections tab
        self.tabview.add("🌐 External Connections")
        self.setup_external_tab()
        
        # Blocked IPs tab
        self.tabview.add("🚫 Blocked IPs")
        self.setup_blocked_tab()
        
    def setup_internal_tab(self):
        internal_frame = self.tabview.tab("🏠 Internal Devices")

        # Header
        header_frame = ctk.CTkFrame(internal_frame)
        header_frame.pack(fill="x", pady=(0, 10))

        ctk.CTkLabel(header_frame, text="IP Address", width=140).pack(side="left", padx=10, pady=5)
        ctk.CTkLabel(header_frame, text="MAC Address", width=140).pack(side="left", padx=10, pady=5)
        ctk.CTkLabel(header_frame, text="Status", width=120).pack(side="left", padx=1, pady=5)
        ctk.CTkLabel(header_frame, text="Action", width=120).pack(side="top", padx=1, pady=5)

        # Scrollable frame for table rows
        self.internal_scroll_frame = ctk.CTkScrollableFrame(internal_frame)
        self.internal_scroll_frame.pack(fill="both", expand=True, padx=10, pady=(0, 10))
        
    def setup_external_tab(self):
        external_frame = self.tabview.tab("🌐 External Connections")

        # Header
        header_frame = ctk.CTkFrame(external_frame)
        header_frame.pack(fill="x", pady=(0, 10))

        ctk.CTkLabel(header_frame, text="External IP", width=150).pack(side="left", padx=10, pady=5)
        ctk.CTkLabel(header_frame, text="Local Port", width=90).pack(side="left", padx=10, pady=5)
        ctk.CTkLabel(header_frame, text="Remote Port", width=90).pack(side="left", padx=10, pady=5)
        ctk.CTkLabel(header_frame, text="Status", width=120).pack(side="left", padx=10, pady=5)
        ctk.CTkLabel(header_frame, text="Action", width=100).pack(side="left", padx=10, pady=5)

        # Scrollable frame for table rows
        self.external_scroll_frame = ctk.CTkScrollableFrame(external_frame)
        self.external_scroll_frame.pack(fill="both", expand=True, padx=10, pady=(0, 10))
        
    def setup_blocked_tab(self):
        blocked_frame = self.tabview.tab("🚫 Blocked IPs")

        # Header
        header_frame = ctk.CTkFrame(blocked_frame)
        header_frame.pack(fill="x", pady=(0, 10))

        ctk.CTkLabel(header_frame, text="Blocked IP Address", width=200).pack(side="left", padx=10, pady=5)
        ctk.CTkLabel(header_frame, text="Time Blocked", width=200).pack(side="left", padx=10, pady=5)
        ctk.CTkLabel(header_frame, text="Action", width=120).pack(side="left", padx=10, pady=5)

        # Scrollable frame for table rows
        self.blocked_scroll_frame = ctk.CTkScrollableFrame(blocked_frame)
        self.blocked_scroll_frame.pack(fill="both", expand=True, padx=10, pady=(0, 10))

        # Store blocked times for display
        self.blocked_times = {}
        
    def is_multicast_ip(self, ip):
        """Check if an IP address is in the multicast range (224.0.0.0 - 239.255.255.255)"""
        try:
            first_octet = int(ip.split('.')[0])
            return 224 <= first_octet <= 239
        except:
            return False

    def is_broadcast_ip(self, ip):
        """Check if an IP address is a broadcast address (ends with .255)"""
        try:
            last_octet = ip.split('.')[3]
            return last_octet == '255'
        except:
            return False

    def is_local_ip(self, ip):
        """Check if an IP address is in the local network ranges"""
        try:
            first_octet = int(ip.split('.')[0])
            second_octet = int(ip.split('.')[1])

            # Private IP ranges
            private_ranges = [
                (10, None),         # 10.0.0.0 - 10.255.255.255
                (172, 16, 31),      # 172.16.0.0 - 172.31.255.255
                (192, 168)          # 192.168.0.0 - 192.168.255.255
            ]

            if first_octet == 10:
                return True
            elif first_octet == 172 and 16 <= second_octet <= 31:
                return True
            elif first_octet == 192 and second_octet == 168:
                return True

            return False
        except:
            return False

    def get_arp_table(self):
        """Get internal devices from ARP table"""
        try:
            result = subprocess.run(['arp', '-a'], capture_output=True, text=True)
            devices = []

            for line in result.stdout.split('\n'):
                # Parse ARP entries like: 192.168.1.1 00-1a-2b-3c-4d-5e dynamic
                match = re.search(r'(\d+\.\d+\.\d+\.\d+)\s+([0-9a-fA-F-]{17})', line)
                if match:
                    ip = match.group(1)
                    mac = match.group(2).replace('-', ':').upper()

                    # Filter out loopback and unspecified addresses
                    if ip.startswith('127.') or ip == '0.0.0.0':
                        continue

                    # Filter out broadcast addresses (x.x.x.255)
                    if self.is_broadcast_ip(ip):
                        continue

                    # Filter out multicast IPs if filter is enabled
                    if self.filter_multicast.get() and self.is_multicast_ip(ip):
                        continue

                    # Only include local network IPs for internal devices
                    if not self.is_local_ip(ip):
                        continue

                    # Skip invalid MAC addresses
                    if mac == 'ff:ff:ff:ff:ff:ff'.upper() or mac == '00:00:00:00:00:00':
                        continue

                    devices.append({'ip': ip, 'mac': mac, 'type': 'Internal'})
            return devices
        except Exception as e:
            print(f"Error getting ARP table: {e}")
            return []
    
    def get_external_connections(self):
        """Get external connections from netstat"""
        try:
            result = subprocess.run(['netstat', '-n'], capture_output=True, text=True)
            connections = []

            for line in result.stdout.split('\n'):
                # Parse netstat entries like: TCP 192.168.1.15:58072 104.208.16.95:443 ESTABLISHED
                match = re.search(r'TCP\s+\d+\.\d+\.\d+\.\d+:(\d+)\s+(\d+\.\d+\.\d+\.\d+):(\d+)\s+ESTABLISHED', line)
                if match:
                    local_port = match.group(1)
                    remote_ip = match.group(2)
                    remote_port = match.group(3)

                    # Only show external IPs (not local network)
                    is_local = (remote_ip.startswith('192.168.') or
                              remote_ip.startswith('10.') or
                              remote_ip.startswith('172.'))

                    if not is_local:
                        # Filter out multicast IPs if filter is enabled
                        if self.filter_multicast.get() and self.is_multicast_ip(remote_ip):
                            continue

                        connections.append({
                            'ip': remote_ip,
                            'local_port': local_port,
                            'remote_port': remote_port,
                            'type': 'External'
                        })

            return connections
        except Exception as e:
            print(f"Error getting connections: {e}")
            return []
    
    def block_ip(self, ip_address):
        """Block an IP using Windows Firewall"""
        try:
            rule_name = f"NetGuard_Block_{ip_address.replace('.', '_')}"
            
            # Create outbound block rule
            cmd = [
                'netsh', 'advfirewall', 'firewall', 'add', 'rule',
                f'name={rule_name}',
                'dir=out',
                'action=block',
                f'remoteip={ip_address}'
            ]
            
            result = subprocess.run(cmd, capture_output=True, text=True)
            
            if result.returncode == 0:
                self.blocked_ips.add(ip_address)
                # Store block time for display
                if not hasattr(self, 'blocked_times'):
                    self.blocked_times = {}
                self.blocked_times[ip_address] = datetime.now().strftime("%m/%d/%Y %H:%M")

                self.save_blocked_ips()
                self.update_status(f"✅ Blocked {ip_address}")
                self.refresh_blocked_display()
                return True
            else:
                self.update_status(f"❌ Failed to block {ip_address}")
                return False
                
        except Exception as e:
            print(f"Error blocking IP {ip_address}: {e}")
            self.update_status(f"❌ Error blocking {ip_address}")
            return False
    
    def unblock_ip(self, ip_address):
        """Unblock an IP by removing firewall rule"""
        try:
            rule_name = f"NetGuard_Block_{ip_address.replace('.', '_')}"
            
            # Delete the firewall rule
            cmd = [
                'netsh', 'advfirewall', 'firewall', 'delete', 'rule',
                f'name={rule_name}'
            ]
            
            result = subprocess.run(cmd, capture_output=True, text=True)
            
            if result.returncode == 0:
                self.blocked_ips.discard(ip_address)
                self.save_blocked_ips()
                self.update_status(f"✅ Unblocked {ip_address}")
                self.refresh_blocked_display()
                return True
            else:
                self.update_status(f"❌ Failed to unblock {ip_address}")
                return False
                
        except Exception as e:
            print(f"Error unblocking IP {ip_address}: {e}")
            self.update_status(f"❌ Error unblocking {ip_address}")
            return False
    
    def clear_all_blocks(self):
        """Clear all blocked IPs"""
        if not self.blocked_ips:
            messagebox.showinfo("Info", "No IPs are currently blocked.")
            return
            
        result = messagebox.askyesno(
            "Confirm", 
            f"Are you sure you want to unblock all {len(self.blocked_ips)} IPs?"
        )
        
        if result:
            blocked_copy = self.blocked_ips.copy()
            for ip in blocked_copy:
                self.unblock_ip(ip)
            
            self.update_status("✅ All blocks cleared")
    
    def save_blocked_ips(self):
        """Save blocked IPs to file"""
        try:
            data = {
                'blocked_ips': list(self.blocked_ips),
                'last_updated': datetime.now().isoformat()
            }
            with open('netguard_config.json', 'w') as f:
                json.dump(data, f, indent=2)
        except Exception as e:
            print(f"Error saving config: {e}")
    
    def load_blocked_ips(self):
        """Load blocked IPs from file"""
        try:
            with open('netguard_config.json', 'r') as f:
                data = json.load(f)
                self.blocked_ips = set(data.get('blocked_ips', []))
        except FileNotFoundError:
            self.blocked_ips = set()
        except Exception as e:
            print(f"Error loading config: {e}")
            self.blocked_ips = set()
    
    def refresh_data(self):
        """Refresh all data in background thread"""
        self.update_status("🔄 Refreshing data...")
        
        def refresh_thread():
            # Get data
            self.internal_devices = self.get_arp_table()
            self.external_connections = self.get_external_connections()
            
            # Update UI in main thread
            self.root.after(0, self.update_displays)
        
        threading.Thread(target=refresh_thread, daemon=True).start()
    
    def update_displays(self):
        """Update all display tabs"""
        self.update_internal_display()
        self.update_external_display()
        self.refresh_blocked_display()
        self.update_status("✅ Data refreshed")
    
    def update_internal_display(self):
        """Update internal devices display"""
        # Clear all existing children
        for widget in self.internal_scroll_frame.winfo_children():
            widget.destroy()

        # Add all device info as table rows
        for device in self.internal_devices:
            status = "🚫 Blocked" if device['ip'] in self.blocked_ips else "✅ Allowed"
            labels_list = [
                (device['ip'], 140),          # IP Address - slightly less for better spacing
                (device['mac'], 170),         # MAC Address
                (status, 110)                 # Status
            ]
            button_text = "Unblock" if device['ip'] in self.blocked_ips else "Block"
            button_color = "green" if device['ip'] in self.blocked_ips else "red"
            hover_color = "darkgreen" if device['ip'] in self.blocked_ips else "darkred"
            button_command = lambda ip=device['ip'], action=button_text.lower(): getattr(self, f"{action}_ip")(ip)

            self.create_table_row(
                self.internal_scroll_frame,
                labels_list,
                button_text,
                button_command,
                button_color,
                hover_color
            )
    
    def update_external_display(self):
        """Update external connections display"""
        # Clear all existing children
        for widget in self.external_scroll_frame.winfo_children():
            widget.destroy()

        # Add all connection info as table rows
        for conn in self.external_connections:
            status = "🚫 Blocked" if conn['ip'] in self.blocked_ips else "✅ Connected"
            labels_list = [
                (conn['ip'], 140),        # External IP - slightly less for better spacing
                (conn['local_port'], 80), # Local Port
                (conn['remote_port'], 80), # Remote Port
                (status, 110)             # Status
            ]
            button_text = "Unblock" if conn['ip'] in self.blocked_ips else "Block"
            button_color = "green" if conn['ip'] in self.blocked_ips else "red"
            hover_color = "darkgreen" if conn['ip'] in self.blocked_ips else "darkred"
            button_command = lambda ip=conn['ip'], action=button_text.lower(): getattr(self, f"{action}_ip")(ip)

            self.create_table_row(
                self.external_scroll_frame,
                labels_list,
                button_text,
                button_command,
                button_color,
                hover_color
            )
    
    def refresh_blocked_display(self):
        """Update blocked IPs display"""
        # Clear all existing children
        for widget in self.blocked_scroll_frame.winfo_children():
            widget.destroy()

        # Add all blocked IP info as table rows
        for ip in self.blocked_ips:
            blockade_time = self.blocked_times.get(ip, "Unknown")
            labels_list = [
                (ip, 190),                  # Blocked IP Address - slightly less for better spacing
                (blockade_time, 190)        # Time Blocked
            ]
            button_text = "Unblock"
            button_color = "green"
            hover_color = "darkgreen"
            button_command = lambda ip=ip: self.unblock_ip(ip)

            self.create_table_row(
                self.blocked_scroll_frame,
                labels_list,
                button_text,
                button_command,
                button_color,
                hover_color
            )
    
    def create_table_row(self, parent, labels_list, button_text, button_command, button_color, hover_color):
        """Create a table row with labels and a button that line up properly"""
        row_frame = ctk.CTkFrame(parent)
        row_frame.pack(fill="x", padx=5, pady=2)

        # Create labels with fixed widths
        for i, (text, width) in enumerate(labels_list):
            # Add some padding between columns
            padx_left = 10 if i == 0 else 5
            padx_right = 5
            ctk.CTkLabel(row_frame, text=text, width=width, anchor="w").pack(
                side="left", padx=(padx_left, padx_right), pady=5
            )

        # Add button last
        btn = ctk.CTkButton(
            row_frame,
            text=button_text,
            command=button_command,
            width=90,
            fg_color=button_color,
            hover_color=hover_color
        )
        btn.pack(side="right", padx=(5, 10), pady=5)

    def block_custom_ip(self):
        """Block a custom IP address entered manually"""
        ip_address = self.manual_ip_entry.get().strip()

        if not ip_address:
            messagebox.showwarning("Warning", "Please enter an IP address.")
            return

        # Basic IP address validation
        ip_pattern = r'^(\d{1,3}\.){3}\d{1,3}$'
        if not re.match(ip_pattern, ip_address):
            messagebox.showerror("Error", "Please enter a valid IP address (e.g., 192.168.1.100).")
            return

        # Check if IP is in multicast range (224.0.0.0 - 239.255.255.255)
        if self.is_multicast_ip(ip_address):
            result = messagebox.askyesno("⚠️ Multicast IP Detected",
                                       f"{ip_address} is a multicast IP address.\n\n"
                                       "Multicast IPs are normally used for special network communication\n"
                                       "and may not function as intended if blocked.\n\n"
                                       "Do you want to block it anyway?")
            if not result:
                return

        # Check if IP is already blocked
        if ip_address in self.blocked_ips:
            messagebox.showinfo("Info", f"{ip_address} is already blocked.")
            return

        # Block the IP
        if self.block_ip(ip_address):
            self.manual_ip_entry.delete(0, 'end')  # Clear the entry after successful block
            messagebox.showinfo("Success", f"Successfully blocked custom IP: {ip_address}")

    def clear_custom_entry(self):
        """Clear the manual IP entry field"""
        self.manual_ip_entry.delete(0, 'end')

    def update_status(self, message):
        """Update status label"""
        self.status_label.configure(text=message)
        print(f"Status: {message}")
    
    def run(self):
        """Start the application"""
        self.root.mainloop()

if __name__ == "__main__":
    app = NetGuardApp()
    app.run()