homedocs/app/scripts/ip_scanner.py

import socket
import threading
import ipaddress
import time
from concurrent.futures import ThreadPoolExecutor
from app.core.extensions import db
from app.core.models import Subnet, Server
import json
import subprocess

def scan(cidr, max_threads=10, save_results=False):
    """
    Scan a subnet for active hosts
    
    Args:
        cidr: The subnet in CIDR notation (e.g. "192.168.1.0/24")
        max_threads: Maximum number of threads to use
        save_results: Whether to save results to the database
        
    Returns:
        A list of dictionaries with IP, hostname, and status
    """
    print(f"Starting scan of {cidr}")
    network = ipaddress.ip_network(cidr)
    
    # Skip network and broadcast addresses for IPv4
    if network.version == 4:
        hosts = list(network.hosts())
    else:
        # For IPv6, just take the first 100 addresses to avoid scanning too many
        hosts = list(network.hosts())[:100]
    
    # Split the hosts into chunks for multithreading
    chunks = [[] for _ in range(max_threads)]
    for i, host in enumerate(hosts):
        chunks[i % max_threads].append(host)
    
    # Initialize results
    results = [[] for _ in range(max_threads)]
    
    # Create and start threads
    threads = []
    for i in range(max_threads):
        if chunks[i]:  # Only start a thread if there are IPs to scan
            t = threading.Thread(target=scan_worker, args=(chunks[i], results, i))
            threads.append(t)
            t.start()
    
    # Wait for all threads to complete
    for t in threads:
        t.join()
    
    # Combine results
    all_results = []
    for r in results:
        all_results.extend(r)
    
    # Save results to database if requested
    if save_results:
        try:
            save_scan_results(cidr, all_results)
        except Exception as e:
            print(f"Error saving scan results: {e}")
    
    print(f"Scan completed. Found {len(all_results)} active hosts.")
    return all_results

def scan_worker(ip_list, results, index):
    """Worker function for threading"""
    for ip in ip_list:
        if ping(ip):
            hostname = get_hostname(ip)
            results[index].append({
                'ip': str(ip),
                'hostname': hostname if hostname else str(ip),
                'status': 'up'
            })

def ping(ip):
    """Ping an IP address and return True if it responds"""
    try:
        # Faster timeout (1 second)
        subprocess.check_output(['ping', '-c', '1', '-W', '1', str(ip)], stderr=subprocess.STDOUT)
        return True
    except subprocess.CalledProcessError:
        return False

def get_hostname(ip):
    """Try to get the hostname for an IP address"""
    try:
        hostname = socket.gethostbyaddr(str(ip))[0]
        return hostname
    except (socket.herror, socket.gaierror):
        return None

def is_host_active_ping(ip):
    """Simple ICMP ping test (platform dependent)"""
    import platform
    import subprocess
    
    param = '-n' if platform.system().lower() == 'windows' else '-c'
    command = ['ping', param, '1', '-w', '1', ip]
    
    try:
        return subprocess.call(command, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) == 0
    except:
        return False

def save_scan_results(cidr, results):
    """Save scan results to the database"""
    from flask import current_app
    
    # Need to be in application context
    if not hasattr(current_app, 'app_context'):
        print("Not in Flask application context, cannot save results")
        return
    
    try:
        # Find subnet by CIDR
        subnet = Subnet.query.filter_by(cidr=cidr).first()
        if not subnet:
            print(f"Subnet {cidr} not found in database")
            return
        
        # Get existing servers in this subnet
        existing_servers = {server.ip_address: server for server in Server.query.filter_by(subnet_id=subnet.id).all()}
        
        # Process scan results
        for host in results:
            ip = host['ip']
            hostname = host['hostname']
            
            # Check if server already exists
            if ip in existing_servers:
                # Update hostname if it was previously unknown
                if existing_servers[ip].hostname == ip and hostname != ip:
                    existing_servers[ip].hostname = hostname
                    db.session.add(existing_servers[ip])
            else:
                # Create new server
                server = Server(
                    hostname=hostname,
                    ip_address=ip,
                    subnet_id=subnet.id,
                    documentation=f"# {hostname}\n\nAutomatically discovered by network scan on {time.strftime('%Y-%m-%d %H:%M:%S')}"
                )
                db.session.add(server)
        
        db.session.commit()
        print(f"Saved scan results for {cidr}")
        
    except Exception as e:
        db.session.rollback()
        print(f"Error saving scan results: {e}")

def schedule_subnet_scans():
    """Schedule automatic scans for subnets marked as auto_scan"""
    from flask import current_app
    
    with current_app.app_context():
        try:
            # Find all subnets with auto_scan enabled
            subnets = Subnet.query.filter_by(auto_scan=True).all()
            
            for subnet in subnets:
                # Start a thread for each subnet
                thread = threading.Thread(
                    target=scan,
                    args=(subnet.cidr,),
                    daemon=True
                )
                thread.start()
                
                # Sleep briefly to avoid overloading
                time.sleep(1)
                
        except Exception as e:
            print(f"Error scheduling subnet scans: {e}")