Skip to content

PRODUCTION_READINESS_PLAN.md

Latest commit

 

History

History
925 lines (705 loc) · 26.2 KB

PRODUCTION_READINESS_PLAN.md

File metadata and controls

925 lines (705 loc) · 26.2 KB
Raw

NetworkMonitor Production Readiness Analysis & Improvement Plan

Analysis Date: February 2, 2026
Project: NetworkMonitor - Network Control Tool (Similar to NetCut)
Status: Development Stage - Requires Critical Fixes Before Production

Executive Summary

After a thorough analysis of the NetworkMonitor codebase, I've identified 18 critical issues, 12 broken workflows, 8 working features, and 25+ improvements needed for production readiness. The application has a solid foundation but requires significant work in core functionality, error handling, security, and testing.

Table of Contents

  1. Critical Missing Methods (Broken Core)
  2. Broken Workflows
  3. Working Workflows
  4. Security Vulnerabilities
  5. Code Quality Issues
  6. Production Readiness Checklist
  7. Detailed Implementation Plan
  8. Priority Matrix

1. Critical Missing Methods (Broken Core)

🔴 CRITICAL: Missing get_connected_devices Method

Impact: Application cannot discover devices on the network
Location: networkmonitor/monitor.py - NetworkController class
Status: NOT IMPLEMENTED

The server.py calls monitor.get_connected_devices() on line 215, but this method does not exist in the NetworkController class.

# server.py line 215 - Broken
devices = monitor.get_connected_devices(interface)

Required Implementation:

def get_connected_devices(self, interface: str = None) -> List[Device]:
    """
    Scan network and return list of connected devices
    
    Args:
        interface: Optional network interface to scan
        
    Returns:
        List[Device]: List of discovered devices
    """
    # Implementation needed:
    # 1. Use ARP scanning with Scapy
    # 2. Parse ARP table for existing entries
    # 3. Resolve hostnames
    # 4. Look up MAC vendors
    # 5. Determine device types
    # 6. Update self.devices dictionary
    pass

🔴 CRITICAL: Missing get_default_interface Method

Impact: Application cannot determine which network interface to use
Location: networkmonitor/monitor.py - NetworkController class
Status: NOT IMPLEMENTED

Called in multiple places:

  • server.py line 190
  • monitor.py line 271 (inside cut_device)
# monitor.py line 271 - Broken
iface = self.get_default_interface()

Required Implementation:

def get_default_interface(self) -> Optional[str]:
    """
    Get the default network interface for packet operations
    
    Returns:
        str: Interface name (e.g., 'eth0', 'Wi-Fi')
    """
    # Implementation needed:
    # 1. Check platform (Windows/Linux/macOS)
    # 2. Find interface with default gateway
    # 3. Return Scapy-compatible interface name
    pass

🔴 CRITICAL: Missing restore_device Method

Impact: Cannot restore network access after cutting device
Location: networkmonitor/monitor.py - NetworkController class
Status: NOT IMPLEMENTED

Called in server.py line 443:

result = getattr(monitor, 'restore_device', lambda x: False)(ip)

This is currently using stop_cut but API calls non-existent restore_device.

🔴 CRITICAL: Missing get_protection_status Method

Impact: Cannot retrieve device protection/attack status
Location: networkmonitor/monitor.py - NetworkController class
Status: NOT IMPLEMENTED

Called in server.py line 455:

status = getattr(monitor, 'get_protection_status', lambda: {})(request.args.get('ip'))

2. Broken Workflows

2.1 Device Discovery Workflow ❌

Status: BROKEN
Reason: get_connected_devices method not implemented

Step Expected Actual
1. API call /api/devices Returns device list Error: Method not found
2. Frontend displays devices Shows network devices Empty/Error state
3. Periodic refresh Updates device list Fails silently

2.2 Network Cut Workflow ❌

Status: PARTIALLY BROKEN
Reason: get_default_interface not implemented

Step Expected Actual
1. User clicks "Cut" Sends cut request Request sent
2. Get interface Returns network interface Error: Method not found
3. ARP spoofing starts Cuts device connection Fails
4. Status updates Shows "cutting" May show incorrect status

2.3 Speed Limiting Workflow ❌

Status: BROKEN - Windows Implementation Invalid
Reason: Uses non-existent netsh command syntax

Broken Code in monitor.py line 538:

# This command doesn't exist!
command = f"netsh interface set interface {ip} throttled {speed_limit}"

Also broken in windows.py: The qoslevel parameter doesn't exist in Windows Firewall:

# This parameter is invalid!
f"qoslevel={limit_bps}"

Correct Approach:

  • Use Windows QoS Policy (requires Group Policy)
  • Or use Traffic Control (tc) on Linux
  • Or implement own bandwidth shaping via packet manipulation

2.4 Device Blocking Workflow ⚠️

Status: PARTIALLY WORKING
Issues:

  • Windows: Works via firewall rules (local only, doesn't affect other network devices)
  • Linux: Works via iptables (local only)
  • macOS: Works via pfctl (local only)

Problem: These methods only block traffic on the HOST machine, not network-wide like NetCut.

NetCut-style blocking requires:

  • ARP spoofing to poison device's ARP cache
  • Continuous ARP packets to maintain block
  • This IS implemented in cut_device but broken due to missing methods

2.5 Test Suite ❌

Status: BROKEN
Location: tests/test_dependency_check.py

# This import doesn't match the actual function signature
from networkmonitor.dependency_check import check_system_requirements

def test_system_requirements():
    result = check_system_requirements()
    assert isinstance(result, tuple)
    assert len(result) == 2  # Actual returns 3-tuple!

The check_system_requirements returns (bool, str) but DependencyChecker.check_all_dependencies() returns (bool, List[str], List[str]) - inconsistent API.

2.6 Gateway Detection Workflow ⚠️

Status: PARTIALLY BROKEN
Location: monitor.py _get_gateway_info()

Issues:

  1. Linux/macOS: Uses psutil.net_if_stats() which doesn't return gateway info
  2. Hardcoded self.arp_path used on all platforms (fails on Linux/macOS)
# Line 186 - Breaks on non-Windows
arp_output = subprocess.check_output([self.arp_path, "-a"], ...)

2.7 WiFi Interface Detection ⚠️

Status: PARTIAL
Issues:

  • Windows: Works
  • Linux: Works
  • macOS: Limited (relies on deprecated airport command)

2.8 Monitoring Loop ⚠️

Status: INCOMPLETE
Location: monitor.py _monitor_loop()

def _monitor_loop(self):
    while not self._stop_event.is_set():
        try:
            self.get_connected_devices()  # BROKEN - method doesn't exist!
            self._update_device_speeds()
            time.sleep(5)

2.9 Device Speed Measurement ❌

Status: BROKEN
Location: monitor.py _update_device_speeds()

def _update_device_speeds(self):
    stats = psutil.net_io_counters(pernic=True)
    for ip, device in self.devices.items():
        if device.status == "active":
            # This is WRONG - measures total interface bytes, not per-device!
            total_bytes = sum(s.bytes_sent + s.bytes_recv for s in stats.values())
            device.current_speed = total_bytes / 1_000_000  # Wrong unit (bytes vs bits)

Problems:

  1. Measures total interface traffic, not per-device
  2. Shows cumulative bytes, not speed (rate)
  3. Wrong conversion (bytes to Mbps should be * 8 / 1_000_000)

2.10 MAC Vendor Lookup ❌

Status: NOT IMPLEMENTED
Location: monitor.py

The mac_vendor_cache is initialized but never used. No vendor lookup implementation exists.

2.11 Device Hostname Resolution ❌

Status: NOT IMPLEMENTED

No code exists to resolve IP addresses to hostnames via DNS/NetBIOS.

2.12 API Error Handling ⚠️

Status: INCOMPLETE

Many API endpoints don't properly handle edge cases:

  • Missing IP validation
  • No rate limiting
  • No authentication
  • Silent failures masked by try/except

3. Working Workflows

3.1 Dependency Checking ✅

Status: WORKING
Location: dependency_check.py

  • Checks Python version ✅
  • Checks Npcap (Windows) ✅
  • Checks iptables/tc (Linux) ✅
  • Checks pfctl (macOS) ✅
  • Checks Python packages ✅

3.2 Flask Server Startup ✅

Status: WORKING
Location: server.py

  • CORS enabled ✅
  • Proper error handlers ✅
  • Cleanup on exit ✅

3.3 Interface Listing ✅

Status: WORKING
Location: windows.py, linux.py, macos.py

All platforms can list network interfaces.

3.4 WiFi Interface Detection (Windows) ✅

Status: WORKING
Location: windows.py

WMI and netsh fallback both work.

3.5 ARP Table Reading (Windows) ✅

Status: WORKING
Location: windows.py get_arp_table()

Can read existing ARP entries.

3.6 Web UI Framework ✅

Status: WORKING
Location: networkmonitor/web/

  • Next.js setup ✅
  • Material-UI components ✅
  • Drag-and-drop support ✅
  • Responsive design ✅

3.7 Launcher/Splash Screen ✅

Status: WORKING
Location: launcher.py

  • Tkinter UI ✅
  • System tray support ✅
  • Browser opening ✅

3.8 CI/CD Pipeline ✅

Status: WORKING
Location: .github/workflows/ci.yml

  • Multi-platform builds ✅
  • Automated releases ✅
  • GitHub Pages deployment ✅

4. Security Vulnerabilities

4.1 🔴 No Authentication

Severity: CRITICAL
Location: server.py

The API has no authentication. Anyone on the network can:

  • View all devices
  • Cut any device's network
  • Block devices
  • Modify speed limits

Fix Required:

# Add API key or session-based authentication
@app.before_request
def authenticate():
    api_key = request.headers.get('X-API-Key')
    if not api_key or api_key != app.config['API_KEY']:
        return jsonify({'error': 'Unauthorized'}), 401

4.2 🔴 Command Injection Vulnerabilities

Severity: HIGH
Location: Multiple files

IP addresses are directly interpolated into shell commands:

# Vulnerable - monitor.py line 538
command = f"netsh interface set interface {ip} throttled {speed_limit}"

# Vulnerable - linux.py line 193
subprocess.run(["iptables", "-A", "INPUT", "-s", ip, "-j", "DROP"])

Fix Required:

  • Validate IP addresses with regex
  • Use parameterized commands where possible

4.3 🟡 No Input Validation

Severity: MEDIUM

API endpoints don't validate:

  • IP address format
  • Speed limit ranges
  • Device names (potential XSS in web UI)

4.4 🟡 Sensitive Data Exposure

Severity: MEDIUM

Logs may contain sensitive network information. No log rotation or secure storage.

4.5 🟡 No HTTPS Support

Severity: MEDIUM

All traffic is unencrypted (HTTP). Network credentials/data visible to sniffers.

5. Code Quality Issues

5.1 Inconsistent Error Handling

Many methods silently catch and log errors without propagating:

except Exception as e:
    logging.error(f"Error: {e}")
    return False  # Caller doesn't know what went wrong

5.2 Duplicate Code

  • Logging setup appears in multiple files
  • Command path setup duplicated
  • Similar subprocess patterns repeated

5.3 Type Hints Incomplete

Many methods lack type hints or have incorrect hints.

5.4 No Unit Tests

Only 1 test file exists with 1 test. Critical functions are untested.

5.5 Hardcoded Values

  • Port 5000 hardcoded in multiple places
  • Interface names hardcoded (e.g., en0 for macOS)
  • API URLs hardcoded in frontend

5.6 Missing Docstrings

Many functions lack proper documentation.

6. Production Readiness Checklist

Core Functionality

  • Implement get_connected_devices() method ✅ DONE (Feb 2, 2026)
  • Implement get_default_interface() method ✅ DONE (Feb 2, 2026)
  • Implement restore_device() method (alias to stop_cut) ✅ DONE (Feb 2, 2026)
  • Implement get_protection_status() method ✅ DONE (Feb 2, 2026)
  • Fix device speed measurement ✅ DONE (Feb 2, 2026)
  • Implement MAC vendor lookup ✅ DONE (Feb 2, 2026)
  • Implement hostname resolution ✅ DONE (Feb 2, 2026)
  • Fix cross-platform gateway detection ✅ DONE (Feb 2, 2026)
  • Add ARP table fallback for device discovery ✅ DONE (Feb 2, 2026)
  • Implement proper bandwidth limiting (placeholder implemented, platform-specific pending)

Security

  • Add API authentication
  • Validate all IP address inputs ✅ DONE (Feb 2, 2026)
  • Sanitize device names for XSS ✅ DONE (Feb 2, 2026)
  • Add HTTPS support option
  • Implement rate limiting
  • Secure log storage
  • Remove shell=True command injection risks ✅ DONE (Feb 2, 2026)

Testing

  • Write unit tests for NetworkController
  • Write unit tests for platform monitors
  • Write integration tests for API
  • Write end-to-end tests for UI
  • Fix existing broken test ✅ DONE (Feb 2, 2026)

Code Quality

  • Add comprehensive type hints
  • Add proper error types/exceptions
  • Reduce code duplication
  • Add all docstrings
  • Set up linting (flake8, mypy)

DevOps

  • Add health check endpoint ✅ (via /api/status)
  • Add metrics/monitoring
  • Improve logging (structured logs)
  • Add configuration file support
  • Create Docker containerization

Documentation

  • Update README with accurate status ✅ DONE (Feb 2, 2026)
  • Document all API endpoints ✅ DONE (Feb 2, 2026)
  • Add troubleshooting guide ✅ (in README)
  • Add contribution guidelines

Frontend

  • Auto-retry connection logic ✅ DONE (Feb 2, 2026)
  • Improved error handling UI ✅ DONE (Feb 2, 2026)
  • Service installation instructions ✅ DONE (Feb 2, 2026)
  • Environment variable support for API URL ✅ DONE (Feb 2, 2026)
  • CORS configuration for Vercel ✅ DONE (Feb 2, 2026)

7. Detailed Implementation Plan

Phase 1: Critical Fixes (Week 1-2)

Task 1.1: Implement get_connected_devices()

Priority: P0 - Blocker
Estimate: 4-6 hours
File: networkmonitor/monitor.py

def get_connected_devices(self, interface: str = None) -> List[Device]:
    """
    Scan network for connected devices using ARP
    """
    devices = []
    
    try:
        # Get network range from interface
        if not interface:
            interface = self.get_default_interface()
        
        iface_info = self.get_interfaces()
        target_range = None
        
        for iface in iface_info:
            if iface.get('name') == interface or iface.get('ip'):
                ip = iface.get('ip')
                mask = iface.get('network_mask', '255.255.255.0')
                # Calculate network range
                target_range = f"{ip}/24"  # Simplified
                break
        
        if not target_range:
            return list(self.devices.values())
        
        # Perform ARP scan using Scapy
        answered, unanswered = srp(
            Ether(dst="ff:ff:ff:ff:ff:ff") / ARP(pdst=target_range),
            timeout=3,
            verbose=False,
            iface=interface
        )
        
        for sent, received in answered:
            ip = received.psrc
            mac = received.hwsrc.upper()
            
            # Get or create device
            if ip in self.devices:
                device = self.devices[ip]
                device.last_seen = datetime.now()
                device.status = "active"
            else:
                hostname = self._resolve_hostname(ip)
                vendor = self._get_mac_vendor(mac)
                device_type = self.guess_device_type(hostname, vendor)
                
                device = Device(
                    ip=ip,
                    mac=mac,
                    hostname=hostname,
                    vendor=vendor,
                    device_type=device_type
                )
                self.devices[ip] = device
            
            devices.append(device)
        
        # Mark devices not seen as inactive
        for ip, device in self.devices.items():
            if device not in devices:
                if (datetime.now() - device.last_seen).seconds > 60:
                    device.status = "inactive"
        
        return devices
        
    except Exception as e:
        logging.error(f"Error scanning devices: {e}")
        return list(self.devices.values())

Task 1.2: Implement get_default_interface()

Priority: P0 - Blocker
Estimate: 2-3 hours
File: networkmonitor/monitor.py

def get_default_interface(self) -> Optional[str]:
    """Get the default network interface for operations"""
    try:
        if self.os_type == "Windows":
            # Use route to find default gateway interface
            output = subprocess.check_output(
                ['route', 'print', '0.0.0.0'],
                text=True,
                creationflags=subprocess.CREATE_NO_WINDOW
            )
            for line in output.splitlines():
                if '0.0.0.0' in line and 'On-link' not in line:
                    parts = line.split()
                    if len(parts) >= 5:
                        gateway_ip = parts[2]
                        # Find interface with this gateway
                        for iface in self.get_interfaces():
                            if iface.get('ip', '').rsplit('.', 1)[0] == gateway_ip.rsplit('.', 1)[0]:
                                return iface.get('name')
            
            # Fallback to first WiFi interface
            wifi_ifaces = self.get_wifi_interfaces()
            if wifi_ifaces:
                return wifi_ifaces[0]
            
            # Fallback to first active interface
            interfaces = self.get_interfaces()
            for iface in interfaces:
                if iface.get('ip') and not iface['ip'].startswith('127.'):
                    return iface.get('name')
        
        elif self.os_type == "Linux":
            output = subprocess.check_output(['ip', 'route'], text=True)
            for line in output.splitlines():
                if line.startswith('default'):
                    parts = line.split()
                    dev_idx = parts.index('dev') if 'dev' in parts else -1
                    if dev_idx != -1 and dev_idx + 1 < len(parts):
                        return parts[dev_idx + 1]
        
        elif self.os_type == "Darwin":  # macOS
            output = subprocess.check_output(['route', 'get', 'default'], text=True)
            for line in output.splitlines():
                if 'interface:' in line:
                    return line.split(':')[1].strip()
        
        return None
        
    except Exception as e:
        logging.error(f"Error getting default interface: {e}")
        return None

Task 1.3: Fix Gateway Detection Cross-Platform

Priority: P0 - Blocker
Estimate: 2 hours
File: networkmonitor/monitor.py

Fix _get_gateway_info() to work on all platforms.

Task 1.4: Add Missing API Methods

Priority: P0 - Blocker
Estimate: 1 hour
File: networkmonitor/monitor.py

def restore_device(self, ip: str) -> bool:
    """Alias for stop_cut - restores network access"""
    return self.stop_cut(ip)

def get_protection_status(self, ip: str = None) -> Dict:
    """Get protection/attack status for devices"""
    if ip:
        device = self.devices.get(ip)
        if device:
            return {
                'ip': ip,
                'is_protected': device.is_protected,
                'attack_status': device.attack_status,
                'is_blocked': device.is_blocked
            }
        return {}
    
    return {
        ip: {
            'is_protected': d.is_protected,
            'attack_status': d.attack_status,
            'is_blocked': d.is_blocked
        }
        for ip, d in self.devices.items()
    }

Phase 2: Core Improvements (Week 2-3)

Task 2.1: Fix Speed Measurement

Priority: P1 - High
Estimate: 3-4 hours

Implement proper per-device bandwidth measurement using packet capture.

Task 2.2: Implement MAC Vendor Lookup

Priority: P1 - High
Estimate: 2 hours

def _get_mac_vendor(self, mac: str) -> Optional[str]:
    """Look up vendor from MAC address"""
    if mac in self.mac_vendor_cache:
        return self.mac_vendor_cache[mac]
    
    try:
        # Use OUI (first 6 hex digits)
        oui = mac.replace(':', '').replace('-', '')[:6].upper()
        
        # Try local OUI database first
        vendor = self._local_oui_lookup(oui)
        
        if not vendor:
            # Fallback to API (with rate limiting)
            response = requests.get(
                f"https://api.macvendors.com/{oui}",
                timeout=2
            )
            if response.status_code == 200:
                vendor = response.text
        
        self.mac_vendor_cache[mac] = vendor
        return vendor
        
    except Exception as e:
        logging.debug(f"MAC vendor lookup failed: {e}")
        return None

Task 2.3: Implement Hostname Resolution

Priority: P1 - High
Estimate: 2 hours

def _resolve_hostname(self, ip: str) -> Optional[str]:
    """Resolve IP to hostname"""
    try:
        hostname, _, _ = socket.gethostbyaddr(ip)
        return hostname
    except (socket.herror, socket.gaierror):
        # Try NetBIOS on Windows
        if self.os_type == "Windows":
            try:
                output = subprocess.check_output(
                    ['nbtstat', '-A', ip],
                    text=True,
                    timeout=5,
                    creationflags=subprocess.CREATE_NO_WINDOW
                )
                for line in output.splitlines():
                    if '<00>' in line and 'UNIQUE' in line:
                        return line.split()[0].strip()
            except:
                pass
        return None

Task 2.4: Fix Bandwidth Limiting

Priority: P1 - High
Estimate: 4-6 hours

Implement proper traffic shaping for each platform.

Phase 3: Security (Week 3-4)

Task 3.1: Add Authentication

Priority: P0 - Critical
Estimate: 4 hours

Task 3.2: Input Validation

Priority: P0 - Critical
Estimate: 2 hours

import re
from functools import wraps

def validate_ip(ip: str) -> bool:
    """Validate IPv4 address format"""
    pattern = r'^(?:(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\.){3}(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)$'
    return bool(re.match(pattern, ip))

def require_valid_ip(f):
    @wraps(f)
    def decorated(*args, **kwargs):
        ip = kwargs.get('ip') or request.json.get('ip')
        if not ip or not validate_ip(ip):
            return jsonify({'error': 'Invalid IP address'}), 400
        return f(*args, **kwargs)
    return decorated

Task 3.3: Rate Limiting

Priority: P1 - High
Estimate: 2 hours

from flask_limiter import Limiter
from flask_limiter.util import get_remote_address

limiter = Limiter(
    get_remote_address,
    app=app,
    default_limits=["100 per hour"]
)

@app.route('/api/device/cut', methods=['POST'])
@limiter.limit("10 per minute")
def cut_device():
    # ...

Phase 4: Testing (Week 4-5)

Task 4.1: Fix Existing Test

Priority: P1 - High
Estimate: 30 minutes

Task 4.2: Unit Tests for NetworkController

Priority: P1 - High
Estimate: 8 hours

Task 4.3: API Integration Tests

Priority: P1 - High
Estimate: 6 hours

Phase 5: Polish & Documentation (Week 5-6)

Task 5.1: Configuration File Support

Priority: P2 - Medium
Estimate: 3 hours

Task 5.2: Structured Logging

Priority: P2 - Medium
Estimate: 2 hours

Task 5.3: API Documentation

Priority: P2 - Medium
Estimate: 4 hours

Task 5.4: Docker Support

Priority: P2 - Medium
Estimate: 4 hours

8. Priority Matrix

Priority Category Tasks Time Est.
P0 Blocker get_connected_devices, get_default_interface, Gateway fix, Missing methods, Authentication, Input validation 15-20 hrs
P1 High Speed measurement, MAC vendor, Hostname resolution, Bandwidth limiting, Rate limiting, Tests 25-30 hrs
P2 Medium Config file, Logging, Docker, Documentation 13-17 hrs
P3 Low UI improvements, Performance optimization 10-15 hrs

Recommended Next Steps

  1. Immediate (Today):

    • Implement get_connected_devices() - THE application is useless without it
    • Implement get_default_interface()

  2. This Week:

    • Add restore_device() and get_protection_status()
    • Fix gateway detection
    • Add basic IP validation

  3. Next Week:

    • Add authentication
    • Implement MAC vendor lookup
    • Add hostname resolution

  4. Following Weeks:

    • Fix speed measurement
    • Add comprehensive tests
    • Documentation and polish

Files to Modify

File Changes Required
networkmonitor/monitor.py Add 4+ methods, fix 3+ methods
networkmonitor/server.py Add authentication, validation
networkmonitor/windows.py Fix speed limiting
networkmonitor/linux.py Minor fixes
networkmonitor/macos.py Fix interface detection
tests/test_dependency_check.py Fix test
tests/test_monitor.py New file - comprehensive tests
tests/test_api.py New file - API tests

Conclusion

The NetworkMonitor application has a solid architecture and good UI foundation, but the core network monitoring functionality is not implemented. The most critical issue is the missing get_connected_devices() method - without it, the entire purpose of the application fails.

Estimated total effort for production readiness: 60-80 hours

Priority should be:

  1. Make it work (core functionality)
  2. Make it secure (authentication, validation)
  3. Make it reliable (tests, error handling)
  4. Make it polished (documentation, UX)