166 lines
No EOL
5.3 KiB
Python
166 lines
No EOL
5.3 KiB
Python
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
|
|
import concurrent.futures
|
|
from datetime import datetime
|
|
import platform
|
|
|
|
def scan(subnet, manual_trigger=False):
|
|
"""
|
|
Scan a subnet for active hosts
|
|
|
|
Args:
|
|
subnet: The subnet object to scan
|
|
manual_trigger: If False, only scan if the subnet hasn't been scanned recently
|
|
"""
|
|
# Skip if not auto scan and not manually triggered
|
|
if not subnet.auto_scan and not manual_trigger:
|
|
return False
|
|
|
|
active_hosts = []
|
|
|
|
try:
|
|
# Parse the CIDR notation
|
|
network = ipaddress.ip_network(subnet.cidr, strict=False)
|
|
|
|
# For each address in this network, ping it
|
|
for ip in network.hosts():
|
|
if ping(str(ip)):
|
|
active_hosts.append(str(ip))
|
|
|
|
# Update subnet with scan results
|
|
subnet.active_hosts = json.dumps(active_hosts)
|
|
subnet.last_scanned = datetime.utcnow()
|
|
|
|
return True
|
|
except Exception as e:
|
|
print(f"Error scanning subnet {subnet.cidr}: {str(e)}")
|
|
return False
|
|
|
|
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(host):
|
|
"""
|
|
Returns True if host responds to a ping request
|
|
"""
|
|
# Ping parameters based on OS
|
|
param = '-n' if platform.system().lower() == 'windows' else '-c'
|
|
# Build the command
|
|
command = ['ping', param, '1', '-w', '1', host]
|
|
|
|
try:
|
|
# Run the command and capture output
|
|
output = subprocess.run(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=2)
|
|
# Return True if ping was successful
|
|
return output.returncode == 0
|
|
except subprocess.TimeoutExpired:
|
|
return False
|
|
except Exception:
|
|
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,),
|
|
daemon=True
|
|
)
|
|
thread.start()
|
|
|
|
# Sleep briefly to avoid overloading
|
|
time.sleep(1)
|
|
|
|
except Exception as e:
|
|
print(f"Error scheduling subnet scans: {e}") |