import asyncio import socket import struct from typing import List, Dict, Optional, Tuple import ipaddress import logging try: from pysnmp.hlapi.v3arch.asyncio import * except: # Fallback for different pysnmp versions pass logger = logging.getLogger(__name__) # OUI Database (sample - top vendors) OUI_DATABASE = { "00:0C:29": "VMware", "00:50:56": "VMware", "00:1B:21": "Intel", "00:15:5D": "Microsoft", "08:00:27": "VirtualBox", "00:0D:3A": "MikroTik", "00:03:7F": "Atheros", "00:1A:79": "Cisco", "00:23:89": "Cisco", "F4:CF:E2": "Huawei", "00:E0:FC": "Huawei", "48:2C:A0": "VSOL", "00:11:22": "VSOL", "84:B8:02": "TP-Link", "F4:F2:6D": "TP-Link", "00:1F:3C": "D-Link", "00:05:CD": "D-Link", "FC:EC:DA": "Ubiquiti", "00:27:22": "Ubiquiti", "00:90:7F": "Juniper", "F0:1C:2D": "ZTE", "00:1D:0F": "FiberHome", } class NetworkScanner: def __init__(self): self.timeout = 1 self.default_ports = [21, 22, 23, 25, 53, 80, 110, 143, 161, 443, 445, 993, 995, 1433, 1521, 3306, 3389, 5432, 5900, 6379, 8080, 8443, 8888, 9090, 27017] self.well_known_services = { 21: "FTP", 22: "SSH", 23: "Telnet", 25: "SMTP", 53: "DNS", 80: "HTTP", 110: "POP3", 143: "IMAP", 161: "SNMP", 162: "SNMP-Trap", 443: "HTTPS", 445: "SMB", 993: "IMAPS", 995: "POP3S", 1433: "MSSQL", 1521: "Oracle", 3306: "MySQL", 3389: "RDP", 5432: "PostgreSQL", 5900: "VNC", 6379: "Redis", 8080: "HTTP-Alt", 8443: "HTTPS-Alt", 8888: "HTTP-Proxy", 9090: "Prometheus", 27017: "MongoDB", 179: "BGP", 520: "RIP", 830: "NETCONF", 49: "TACACS", 1812: "RADIUS", 514: "Syslog", 2049: "NFS", 67: "DHCP-Server", 68: "DHCP-Client", 69: "TFTP", 123: "NTP", 500: "IKE", 4500: "IPSec-NAT-T", 1701: "L2TP", 1723: "PPTP", 47: "GRE", } def get_vendor_from_mac(self, mac: str) -> Optional[str]: """Get vendor from MAC address using OUI database""" if not mac: return None # Extract OUI (first 3 octets) oui = ":".join(mac.split(":")[:3]).upper() return OUI_DATABASE.get(oui, "Unknown") async def ping_host(self, ip: str, timeout: int = 1) -> bool: """Check if host is alive using socket connection""" try: # Try to connect to common ports for port in [80, 443, 22, 23]: try: reader, writer = await asyncio.wait_for( asyncio.open_connection(ip, port), timeout=timeout ) writer.close() await writer.wait_closed() return True except: continue return False except: return False async def get_hostname(self, ip: str) -> Optional[str]: """Get hostname from IP""" try: loop = asyncio.get_event_loop() hostname = await loop.run_in_executor(None, socket.gethostbyaddr, ip) return hostname[0] except: return None async def scan_ports(self, ip: str, ports: List[int] = None, timeout: int = 1) -> List[Dict]: """Scan ports and return detailed port info with service names""" if ports is None: ports = self.default_ports open_ports = [] async def check_port(port: int): try: reader, writer = await asyncio.wait_for( asyncio.open_connection(ip, port), timeout=timeout ) banner = "" try: data = await asyncio.wait_for(reader.read(256), timeout=0.5) banner = data.decode('utf-8', errors='ignore').strip() except: pass writer.close() await writer.wait_closed() return { "port": port, "state": "open", "service": self.well_known_services.get(port, "unknown"), "banner": banner[:128] if banner else None } except: return None tasks = [check_port(port) for port in ports] results = await asyncio.gather(*tasks) open_ports = [p for p in results if p is not None] return open_ports def parse_port_range(self, port_range: str) -> List[int]: """Parse port range string like '80,443,8000-8100'""" ports = [] for part in port_range.split(','): part = part.strip() if '-' in part: start, end = part.split('-', 1) ports.extend(range(int(start.strip()), int(end.strip()) + 1)) else: ports.append(int(part)) return sorted(set(ports)) async def scan_single_host(self, ip: str, port: int = None, scan_ports_flag: bool = True, check_snmp_flag: bool = True, snmp_community: str = "public", custom_ports: List[int] = None, timeout: int = 2) -> Optional[Dict]: """Scan a single host with optional specific port targeting""" if port: try: reader, writer = await asyncio.wait_for( asyncio.open_connection(ip, port), timeout=timeout ) banner = "" try: data = await asyncio.wait_for(reader.read(256), timeout=0.5) banner = data.decode('utf-8', errors='ignore').strip() except: pass writer.close() await writer.wait_closed() device_info = { "ip": ip, "status": "online", "hostname": await self.get_hostname(ip), "mac": None, "vendor": None, "open_ports": [{"port": port, "state": "open", "service": self.well_known_services.get(port, "unknown"), "banner": banner[:128] if banner else None}], "snmp_enabled": False, "snmp_data": None } if check_snmp_flag: snmp_enabled, snmp_data = await self.check_snmp(ip, snmp_community) device_info["snmp_enabled"] = snmp_enabled if snmp_data: device_info["snmp_data"] = snmp_data device_info["sys_descr"] = snmp_data.get("sys_descr") self._identify_vendor(device_info, snmp_data) return device_info except: return None else: return await self.scan_host(ip, scan_ports_flag, check_snmp_flag, snmp_community, custom_ports, timeout) def _identify_vendor(self, device_info: Dict, snmp_data: Dict): """Identify vendor/device type from SNMP data""" sys_descr_lower = snmp_data.get("sys_descr", "").lower() if "mikrotik" in sys_descr_lower or "routeros" in sys_descr_lower: device_info["vendor"] = "MikroTik" device_info["device_type"] = "Router" elif "cisco" in sys_descr_lower or "ios" in sys_descr_lower: device_info["vendor"] = "Cisco" if "switch" in sys_descr_lower: device_info["device_type"] = "Switch" elif "firewall" in sys_descr_lower or "asa" in sys_descr_lower: device_info["device_type"] = "Firewall" else: device_info["device_type"] = "Router" elif "huawei" in sys_descr_lower: device_info["vendor"] = "Huawei" if "olt" in sys_descr_lower: device_info["device_type"] = "OLT" else: device_info["device_type"] = "Router" elif "vsol" in sys_descr_lower: device_info["vendor"] = "VSOL" device_info["device_type"] = "OLT" elif "juniper" in sys_descr_lower or "junos" in sys_descr_lower: device_info["vendor"] = "Juniper" device_info["device_type"] = "Router" elif "fortinet" in sys_descr_lower or "fortigate" in sys_descr_lower: device_info["vendor"] = "Fortinet" device_info["device_type"] = "Firewall" elif "pfsense" in sys_descr_lower: device_info["vendor"] = "pfSense" device_info["device_type"] = "Firewall" elif "ubiquiti" in sys_descr_lower or "edgeos" in sys_descr_lower or "unifi" in sys_descr_lower: device_info["vendor"] = "Ubiquiti" device_info["device_type"] = "Router" elif "zte" in sys_descr_lower: device_info["vendor"] = "ZTE" device_info["device_type"] = "OLT" elif "fiberhome" in sys_descr_lower: device_info["vendor"] = "FiberHome" device_info["device_type"] = "OLT" elif "linux" in sys_descr_lower: device_info["vendor"] = "Linux" device_info["device_type"] = "Server" elif "windows" in sys_descr_lower: device_info["vendor"] = "Microsoft" device_info["device_type"] = "Server" async def check_snmp(self, ip: str, community: str = "public", version: str = "v2c") -> Tuple[bool, Optional[Dict]]: """Check if SNMP is enabled and get basic info""" try: from pysnmp.hlapi.v3arch.asyncio import ( SnmpEngine, CommunityData, UdpTransportTarget, ContextData, ObjectType, ObjectIdentity, getCmd ) # System Description OID errorIndication, errorStatus, errorIndex, varBinds = await getCmd( SnmpEngine(), CommunityData(community), await UdpTransportTarget.create((ip, 161), timeout=2, retries=0), ContextData(), ObjectType(ObjectIdentity('SNMPv2-MIB', 'sysDescr', 0)) ) if errorIndication or errorStatus: return False, None # Get system info sys_descr = str(varBinds[0][1]) # Get more info snmp_data = { "sys_descr": sys_descr, "community": community, "version": version } return True, snmp_data except Exception as e: logger.debug(f"SNMP check failed for {ip}: {e}") return False, None async def scan_host(self, ip: str, scan_ports_flag: bool = True, check_snmp_flag: bool = True, snmp_community: str = "public", custom_ports: List[int] = None, timeout: int = 2) -> Optional[Dict]: """Scan a single host and gather information""" is_alive = await self.ping_host(ip, timeout=timeout) if not is_alive: return None device_info = { "ip": ip, "status": "online", "hostname": None, "mac": None, "vendor": None, "open_ports": [], "snmp_enabled": False, "snmp_data": None } hostname = await self.get_hostname(ip) device_info["hostname"] = hostname if scan_ports_flag: port_results = await self.scan_ports(ip, ports=custom_ports, timeout=timeout) device_info["open_ports"] = port_results snmp_port_open = any( (p["port"] if isinstance(p, dict) else p) == 161 for p in device_info["open_ports"] ) if device_info["open_ports"] else False if check_snmp_flag and (snmp_port_open or not scan_ports_flag): snmp_enabled, snmp_data = await self.check_snmp(ip, snmp_community) device_info["snmp_enabled"] = snmp_enabled if snmp_data: device_info["snmp_data"] = snmp_data device_info["sys_descr"] = snmp_data.get("sys_descr") self._identify_vendor(device_info, snmp_data) return device_info async def scan_subnet(self, subnet: str, scan_ports: bool = True, check_snmp: bool = True, snmp_community: str = "public", custom_ports: List[int] = None, timeout: int = 2) -> List[Dict]: """Scan entire subnet or IP range""" try: network = ipaddress.ip_network(subnet, strict=False) hosts = list(network.hosts()) if len(hosts) > 254: hosts = hosts[:254] logger.info(f"Scanning {len(hosts)} hosts in {subnet}") semaphore = asyncio.Semaphore(50) async def scan_with_semaphore(ip): async with semaphore: return await self.scan_host(str(ip), scan_ports, check_snmp, snmp_community, custom_ports, timeout) tasks = [scan_with_semaphore(ip) for ip in hosts] results = await asyncio.gather(*tasks) devices = [device for device in results if device is not None] logger.info(f"Found {len(devices)} devices") return devices except Exception as e: logger.error(f"Subnet scan error: {e}") raise async def scan_target(self, target: str, port: int = None, port_range: str = None, scan_ports: bool = True, check_snmp: bool = True, snmp_community: str = "public", custom_ports: List[int] = None, timeout: int = 2) -> List[Dict]: """Scan a target - can be IP, IP:port, CIDR, hostname, or range""" devices = [] if ':' in target and not '/' in target: parts = target.rsplit(':', 1) ip = parts[0] try: target_port = int(parts[1]) except ValueError: target_port = None result = await self.scan_single_host(ip, port=target_port, scan_ports_flag=scan_ports, check_snmp_flag=check_snmp, snmp_community=snmp_community, custom_ports=custom_ports, timeout=timeout) if result: devices.append(result) elif '/' in target: devices = await self.scan_subnet(target, scan_ports, check_snmp, snmp_community, custom_ports, timeout) elif '-' in target.split('.')[-1] if '.' in target else False: parts = target.rsplit('.', 1) base = parts[0] range_part = parts[1] start, end = range_part.split('-') for i in range(int(start), int(end) + 1): ip = f"{base}.{i}" result = await self.scan_single_host(ip, port=port, scan_ports_flag=scan_ports, check_snmp_flag=check_snmp, snmp_community=snmp_community, custom_ports=custom_ports, timeout=timeout) if result: devices.append(result) else: try: ip = socket.gethostbyname(target) except socket.gaierror: ip = target ports_to_scan = custom_ports if port_range: ports_to_scan = self.parse_port_range(port_range) result = await self.scan_single_host(ip, port=port, scan_ports_flag=scan_ports, check_snmp_flag=check_snmp, snmp_community=snmp_community, custom_ports=ports_to_scan, timeout=timeout) if result: result["hostname"] = result.get("hostname") or target devices.append(result) return devices