#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ NetScan — console réseau AV, en UN SEUL fichier, ZÉRO dépendance. Lancer localement : python3 netscan.py Depuis le site : python3 <(curl -s https://TON-SITE/tools/netscan.py) Accès multi-appareil : python3 netscan.py --lan (puis http://:5000) Changer une IP : sudo python3 netscan.py À utiliser uniquement sur un réseau que tu possèdes ou administres. """ from __future__ import annotations import concurrent.futures, ipaddress, json, platform, re, socket import subprocess, sys, threading, time, webbrowser from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from typing import Dict, List, Optional """ Network discovery, device classification, and port-scanning engine. Standard-library (+ optional psutil for nicer interface enumeration). Cross-platform: Linux / macOS / Windows. Discovers and inspects hosts on a network you administer, and classifies them for AV / stage-control use (projectors, sound consoles, switches, media servers, cameras...), surfacing each device's web remote page and control capabilities. It does not authenticate, exploit, or exfiltrate. """ import concurrent.futures import ipaddress import platform import re import socket import subprocess import time from typing import Dict, List, Optional SYSTEM = platform.system() # 'Linux', 'Darwin', 'Windows' # --------------------------------------------------------------------------- # Well-known ports -> service name (includes AV / stage-control protocols) # --------------------------------------------------------------------------- COMMON_PORTS: Dict[int, str] = { 21: "ftp", 22: "ssh", 23: "telnet", 25: "smtp", 53: "dns", 67: "dhcp", 80: "http", 110: "pop3", 111: "rpcbind", 123: "ntp", 135: "msrpc", 137: "netbios", 139: "netbios-ssn", 143: "imap", 161: "snmp", 389: "ldap", 443: "https", 445: "smb", 465: "smtps", 500: "isakmp", 515: "printer", 548: "afp", 554: "rtsp", 587: "submission", 631: "ipp", 993: "imaps", 995: "pop3s", 1080: "socks", 1433: "mssql", 1521: "oracle", 1723: "pptp", 1883: "mqtt", 1900: "upnp", 2049: "nfs", 2082: "cpanel", 2375: "docker", 3000: "http-alt", 3306: "mysql", 3389: "rdp", 5000: "upnp/http", 5060: "sip", 5353: "mdns", 5432: "postgres", 5555: "adb", 5900: "vnc", 5985: "winrm", 6379: "redis", 7547: "cwmp", 8000: "http-alt", 8080: "http-proxy", 8443: "https-alt", 8888: "http-alt", 9000: "http-alt", 9100: "printer-raw", 27017: "mongodb", 32400: "plex", 49152: "upnp", # --- AV / stage / broadcast control --- 4352: "pjlink", 6454: "art-net", 5568: "sacn", 1710: "q-sys", 41794: "crestron-cip", 41800: "crestron", 5960: "ndi", 5961: "ndi", 9090: "web-control", 3702: "ws-discovery", 8899: "onvif", 16000: "atem", 5250: "vmix", 7000: "airplay", 7100: "font/av", 5004: "rtp/dante", 4321: "dante", 8751: "dante-arc", } DEFAULT_SCAN_PORTS: List[int] = sorted(COMMON_PORTS.keys()) # Fast profiling probe run during discovery so the table shows a type + a # control link without a manual per-host scan. QUICK_PORTS: List[int] = [80, 443, 8080, 8000, 8443, 22, 23, 4352, 6454, 3389, 5900, 445, 9100, 554, 1710, 8899] # --------------------------------------------------------------------------- # OUI (MAC vendor prefix) table — general + AV brands. Convenience subset; # load_oui_file() plugs in the full IEEE registry for complete coverage. # --------------------------------------------------------------------------- OUI_PREFIXES: Dict[str, str] = { # general "FCFBFB": "Apple", "F0189E": "Apple", "3C0754": "Apple", "A85C2C": "Apple", "DCA904": "Apple", "001451": "Apple", "0026BB": "Apple", "F0D1A9": "Apple", "B827EB": "Raspberry Pi", "DCA632": "Raspberry Pi", "E45F01": "Raspberry Pi", "2CCF67": "Raspberry Pi", "001A11": "Google", "F4F5D8": "Google", "A47733": "Google", "F0EF86": "Google", "000C29": "VMware", "005056": "VMware", "080027": "VirtualBox", "525400": "QEMU/KVM", "0050F2": "Microsoft", "001DD8": "Microsoft", "7C1E52": "Microsoft", "F01FAF": "Dell", "00188B": "Dell", "D89EF3": "Dell", "001132": "Synology", "0011D8": "ASUSTek", "D850E6": "ASUSTek", "AC220B": "ASUSTek", "F832E4": "ASUSTek", "18AF61": "Samsung", "5CF6DC": "Samsung", "E8508B": "Samsung", "F008F1": "Samsung", "1C232C": "Samsung", "B0A737": "Roku", "DC3A5E": "Roku", "CC6DA0": "Roku", "BCDDC2": "Espressif(ESP)", "240AC4": "Espressif(ESP)", "A020A6": "Espressif(ESP)", "3C71BF": "Espressif(ESP)", "7CDFA1": "Espressif(ESP)", "D8A01D": "Espressif(ESP)", "18FE34": "Espressif(ESP)", "5CCF7F": "Espressif(ESP)", # networking gear "C46E1F": "TP-Link", "50C7BF": "TP-Link", "9C5322": "TP-Link", "AC84C6": "TP-Link", "68FF7B": "TP-Link", "001A70": "Cisco-Linksys", "0025B4": "Cisco", "00000C": "Cisco", "F4CFE2": "Cisco", "001E2A": "Netgear", "A040A0": "Netgear", "204E7F": "Netgear", "C03F0E": "Netgear", "000FB5": "Netgear", "001560": "Belkin", "94103E": "Belkin", "EC1A59": "Belkin", "F09FC2": "Ubiquiti", "788A20": "Ubiquiti", "245A4C": "Ubiquiti", "802AA8": "Ubiquiti", "FCECDA": "Ubiquiti", "B4FBE4": "Ubiquiti", "0418D6": "Ubiquiti", "001B54": "Cisco", "3C2C30": "Aruba/HPE", "000B86": "Aruba", "94B40F": "Aruba", "6CF37F": "Aruba", "001871": "HP", "70106F": "HP", "9457A5": "HP", # ---- AUDIO consoles / gear ---- "0009B6": "Yamaha", "00A0DE": "Yamaha", "AC442A": "Yamaha", "00074B": "DigiTech", "001CF0": "Allen & Heath", "0004B7": "Allen & Heath", "002196": "DiGiCo", "001CB3": "Midas/Behringer", "0013FD": "Behringer", "F44EFD": "Behringer/Music Tribe", "D0BD01": "Music Tribe", "001E2D": "Focusrite", "001B66": "Sennheiser", "000B4A": "Shure", "0C8CDC": "Shure", "001F79": "L-Acoustics", "0090D6": "QSC", "002438": "QSC", "0060EB": "Biamp", "000FD9": "Biamp", "0004F2": "Polycom", "000A27": "Apple", "001DBE": "Focusrite/Novation", # ---- Dante / audio networking ---- "001DC1": "Audinate(Dante)", "48A6B8": "Audinate(Dante)", "0C8E39": "Audinate(Dante)", "60C5A8": "Audinate(Dante)", # ---- VIDEO / projectors ---- "00000E": "Fujitsu", "3C0518": "Epson", "001125": "Epson", "6C626D": "Epson", "005056E": "Christie", "008077": "Brother", "0080A3": "Lantronix", "00A06F": "Barco", "00184D": "Barco", "0011A0": "NEC", "006097": "NEC", "008046": "Panasonic", "F84A73": "Panasonic", "3428C6": "Panasonic", "F0271C": "Panasonic", "00036F": "Sony", "FCE998": "Sony", "8CE748": "Sony", "001A80": "Sony", "00E091": "LG", "001E75": "LG", "40B0FA": "LG", "0022A1": "Christie", "001CC4": "HP/Video", "00040D": "Avaya", # ---- VIDEO switching / streaming ---- "38C986": "Blackmagic", "7CC70D": "Blackmagic", "0016E6": "GigaByte", "001E4F": "Dell", "00907F": "WatchGuard", "902B34": "GigaByte", "D0577B": "AJA", "0007F8": "Extron", "002480": "Extron", "000E1E": "Crestron", "001C57": "Crestron", "C442023": "Crestron", # ---- LIGHTING / DMX ---- "00A051": "ETC", "0080A0": "ETC", "001D42": "MA Lighting", "0050C2E": "ChamSys", "70B3D5": "IEEE-Registered(DMX/IoT)", # ---- cameras ---- "00408C": "Axis", "ACCC8E": "Axis", "B8A44F": "Axis", "E0507B": "Hikvision", "4419B6": "Hikvision", "C056E3": "Hikvision", "BCAD28": "Dahua", "3C1B7E": "Dahua", # printers / NAS "0011328": "Synology", "245EBE": "QNAP", "00089B": "ICP/QNAP", "9C5C8E": "ASRock", "AABBCC": "Locally-administered", } # Vendor -> device category (used by classify_device as a strong hint) VENDOR_CATEGORY: Dict[str, str] = { "Yamaha": "console", "Allen & Heath": "console", "DiGiCo": "console", "Midas/Behringer": "console", "Behringer": "console", "Behringer/Music Tribe": "console", "Music Tribe": "console", "QSC": "console", "Biamp": "console", "Focusrite": "audio", "Focusrite/Novation": "audio", "Sennheiser": "audio", "Shure": "audio", "L-Acoustics": "audio", "Audinate(Dante)": "audio", "DigiTech": "audio", "Epson": "projector", "Christie": "projector", "Barco": "projector", "NEC": "projector", "Panasonic": "projector", "Sony": "projector", "Blackmagic": "video", "AJA": "video", "Extron": "video", "Crestron": "control", "ETC": "lighting", "MA Lighting": "lighting", "ChamSys": "lighting", "Axis": "camera", "Hikvision": "camera", "Dahua": "camera", "Cisco": "network", "Cisco-Linksys": "network", "Netgear": "network", "TP-Link": "network", "Ubiquiti": "network", "Aruba": "network", "Aruba/HPE": "network", "Belkin": "network", "Synology": "nas", "QNAP": "nas", "ICP/QNAP": "nas", "Raspberry Pi": "computer", "Apple": "computer", "Dell": "computer", "Microsoft": "computer", "VMware": "computer", "VirtualBox": "computer", "QEMU/KVM": "computer", "Espressif(ESP)": "iot", "Roku": "media", "Google": "media", "Brother": "printer", "HP": "printer", } # category -> (label FR, emoji) CATEGORY_LABELS: Dict[str, tuple] = { "projector": ("Projecteur vidéo", "📽️"), "console": ("Console son", "🎛️"), "audio": ("Audio / Dante", "🔊"), "video": ("Vidéo / streaming", "🎬"), "lighting": ("Éclairage / DMX", "💡"), "control": ("Contrôle / automation", "🕹️"), "camera": ("Caméra IP", "📹"), "network": ("Switch / réseau", "🔀"), "nas": ("NAS / stockage", "🗄️"), "printer": ("Imprimante", "🖨️"), "media": ("Média / TV", "📺"), "computer": ("Ordinateur", "💻"), "iot": ("IoT / capteur", "📡"), "unknown": ("Inconnu", "❓"), } # port -> capability {label, action, scheme} # action: 'web' | 'link' | 'info' PORT_CAPABILITY: Dict[int, dict] = { 80: {"label": "Interface web", "action": "web", "scheme": "http"}, 443: {"label": "Interface web (HTTPS)", "action": "web", "scheme": "https"}, 8080: {"label": "Interface web", "action": "web", "scheme": "http"}, 8000: {"label": "Interface web", "action": "web", "scheme": "http"}, 8443: {"label": "Interface web (HTTPS)", "action": "web", "scheme": "https"}, 8888: {"label": "Interface web", "action": "web", "scheme": "http"}, 9090: {"label": "Interface web", "action": "web", "scheme": "http"}, 631: {"label": "Interface impression (IPP)", "action": "web", "scheme": "http"}, 4352: {"label": "PJLink (contrôle projecteur)", "action": "info"}, 6454: {"label": "Art-Net (DMX/lumière)", "action": "info"}, 5568: {"label": "sACN / E1.31 (lumière)", "action": "info"}, 1710: {"label": "Q-SYS (contrôle externe)", "action": "info"}, 41794:{"label": "Crestron CIP", "action": "info"}, 8899: {"label": "ONVIF (caméra)", "action": "info"}, 554: {"label": "Flux RTSP (caméra)", "action": "link", "scheme": "rtsp"}, 5960: {"label": "NDI (vidéo réseau)", "action": "info"}, 22: {"label": "SSH", "action": "link", "scheme": "ssh"}, 23: {"label": "Telnet (contrôle série)", "action": "link", "scheme": "telnet"}, 5900: {"label": "VNC (bureau distant)", "action": "link", "scheme": "vnc"}, 3389: {"label": "RDP (bureau distant)", "action": "link", "scheme": "rdp"}, 445: {"label": "Partage de fichiers (SMB)", "action": "link", "scheme": "smb"}, 548: {"label": "Partage de fichiers (AFP)", "action": "link", "scheme": "afp"}, 9100: {"label": "Impression brute (RAW)", "action": "info"}, 5353: {"label": "Bonjour/mDNS (Dante, AirPlay…)", "action": "info"}, 161: {"label": "SNMP (supervision)", "action": "info"}, 32400:{"label": "Serveur Plex", "action": "web", "scheme": "http"}, } def load_oui_file(path: str) -> int: """Load a full IEEE OUI file (oui.txt or a 'PREFIX,Vendor' CSV) at runtime.""" added = 0 try: with open(path, "r", encoding="utf-8", errors="ignore") as fh: for line in fh: line = line.strip() m = re.match(r"([0-9A-Fa-f]{2})-([0-9A-Fa-f]{2})-([0-9A-Fa-f]{2})\s+\(hex\)\s+(.+)", line) if m: pref = (m.group(1) + m.group(2) + m.group(3)).upper() OUI_PREFIXES[pref] = m.group(4).strip() added += 1 continue if "," in line: pref, _, vendor = line.partition(",") pref = re.sub(r"[^0-9A-Fa-f]", "", pref).upper()[:6] if len(pref) == 6 and vendor.strip(): OUI_PREFIXES[pref] = vendor.strip() added += 1 except OSError: pass return added def vendor_from_mac(mac: Optional[str]) -> str: if not mac: return "" clean = re.sub(r"[^0-9A-Fa-f]", "", mac).upper() if len(clean) < 6: return "" try: second = int(clean[1], 16) local_hint = "Randomized/Local" if (second & 0x2) else "" except ValueError: local_hint = "" return OUI_PREFIXES.get(clean[:6], local_hint) # --------------------------------------------------------------------------- # Classification # --------------------------------------------------------------------------- def _best_web_remote(ip: str, open_port_nums) -> Optional[str]: """Return the best URL to the device's own control page, if any.""" for p in (443, 8443, 80, 8080, 8000, 8888, 9090, 631, 32400): if p in open_port_nums: cap = PORT_CAPABILITY.get(p, {}) scheme = cap.get("scheme", "http") suffix = "" if p in (80, 443) else f":{p}" return f"{scheme}://{ip}{suffix}/" return None def classify_device(vendor: str, ports: Optional[List[dict]], hostname: str = "") -> dict: """Return {category, type_label, icon, web_remote, capabilities[]}. `ports` is the list produced by scan_ports (may be None/partial). """ port_nums = {p["port"] for p in ports} if ports else set() host_l = (hostname or "").lower() # Port-driven category (strong signals first) category = None if 4352 in port_nums: category = "projector" elif 6454 in port_nums or 5568 in port_nums: category = "lighting" elif 1710 in port_nums: category = "console" elif 8899 in port_nums or 554 in port_nums: category = "camera" elif 9100 in port_nums or 631 in port_nums or 515 in port_nums: category = "printer" elif 32400 in port_nums: category = "media" # Vendor hint if category is None and vendor: category = VENDOR_CATEGORY.get(vendor) # Hostname hint if category is None: for kw, cat in (("proj", "projector"), ("epson", "projector"), ("console", "console"), ("mix", "console"), ("cam", "camera"), ("switch", "network"), ("ap-", "network"), ("nas", "nas"), ("print", "printer"), ("pi", "computer")): if kw in host_l: category = cat break if category is None: category = "unknown" label, icon = CATEGORY_LABELS.get(category, CATEGORY_LABELS["unknown"]) # Capabilities from open ports caps: List[dict] = [] if ports: for p in ports: cap = PORT_CAPABILITY.get(p["port"]) if cap: entry = {"port": p["port"], "label": cap["label"], "action": cap["action"]} if cap.get("scheme") and cap["action"] in ("web", "link"): sfx = "" if p["port"] in (80, 443) else f":{p['port']}" entry["url"] = f"{cap['scheme']}://__IP__{sfx}/" caps.append(entry) web_remote = _best_web_remote("__IP__", port_nums) return {"category": category, "type_label": label, "icon": icon, "web_remote": web_remote, "capabilities": caps} # --------------------------------------------------------------------------- # Interface enumeration # --------------------------------------------------------------------------- def _primary_ip() -> Optional[str]: s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) try: s.connect(("8.8.8.8", 80)) return s.getsockname()[0] except Exception: return None finally: s.close() def _hexmask_to_dotted(m: str) -> str: """macOS gives netmask as hex, e.g. '0xffffff00' -> '255.255.255.0'.""" m = m.lower().replace("0x", "") if len(m) == 8: try: return ".".join(str(int(m[i:i + 2], 16)) for i in (0, 2, 4, 6)) except ValueError: return "255.255.255.0" return m def _parse_ifconfig(text: str) -> List[dict]: """Parse `ifconfig` output (macOS / Linux net-tools).""" ifaces: List[dict] = [] cur = None for line in text.splitlines(): if line and not line[0].isspace(): name = line.split(":", 1)[0].split()[0].strip() cur = {"name": name, "ip": None, "netmask": None, "mac": ""} ifaces.append(cur) elif cur is not None: l = line.strip() if l.startswith("ether "): cur["mac"] = l.split()[1].lower() elif l.startswith("inet ") and "inet6" not in l: parts = l.split() cur["ip"] = parts[1] if "netmask" in parts: nm = parts[parts.index("netmask") + 1] cur["netmask"] = _hexmask_to_dotted(nm) if nm.lower().startswith("0x") else nm return ifaces def _parse_ip_linux() -> List[dict]: """Parse iproute2 `ip` output when ifconfig isn't available.""" ifaces: Dict[str, dict] = {} try: out = subprocess.run(["ip", "-o", "-4", "addr", "show"], capture_output=True, text=True, timeout=6).stdout for line in out.splitlines(): parts = line.split() if len(parts) < 4 or "inet" not in parts: continue name = parts[1] cidr = parts[parts.index("inet") + 1] ip, _, pfx = cidr.partition("/") try: nm = str(ipaddress.IPv4Network(f"0.0.0.0/{pfx}", strict=False).netmask) except Exception: nm = "255.255.255.0" ifaces.setdefault(name, {"name": name, "ip": None, "netmask": None, "mac": ""}) ifaces[name]["ip"] = ip ifaces[name]["netmask"] = nm outl = subprocess.run(["ip", "-o", "link", "show"], capture_output=True, text=True, timeout=6).stdout for line in outl.splitlines(): parts = line.split() if len(parts) < 2 or "link/ether" not in parts: continue name = parts[1].rstrip(":") mac = parts[parts.index("link/ether") + 1] if name in ifaces: ifaces[name]["mac"] = mac.lower() except Exception: pass return list(ifaces.values()) def _parse_windows() -> List[dict]: """Best-effort parse of `ipconfig /all`.""" ifaces: List[dict] = [] cur = None try: out = subprocess.run(["ipconfig", "/all"], capture_output=True, text=True, timeout=8).stdout except Exception: return ifaces for raw in out.splitlines(): if raw and not raw[0].isspace() and raw.rstrip().endswith(":"): cur = {"name": raw.strip().rstrip(":"), "ip": None, "netmask": None, "mac": ""} ifaces.append(cur) elif cur is not None: l = raw.strip() if "Physical Address" in l and ":" in l: cur["mac"] = l.split(":", 1)[1].strip().replace("-", ":").lower() elif "IPv4 Address" in l: m = re.search(r"(\d+\.\d+\.\d+\.\d+)", l) if m: cur["ip"] = m.group(1) elif "Subnet Mask" in l: m = re.search(r"(\d+\.\d+\.\d+\.\d+)", l) if m: cur["netmask"] = m.group(1) return ifaces def get_interfaces() -> List[dict]: """Return [{name, ip, netmask, cidr, mac, is_primary}] — pure stdlib.""" primary = _primary_ip() raw: List[dict] = [] try: if SYSTEM == "Windows": raw = _parse_windows() else: out = subprocess.run(["ifconfig", "-a"], capture_output=True, text=True, timeout=6).stdout if out.strip(): raw = _parse_ifconfig(out) elif SYSTEM == "Linux": raw = _parse_ip_linux() except Exception: if SYSTEM == "Linux": raw = _parse_ip_linux() result: List[dict] = [] for i in raw: ip = i.get("ip") nm = i.get("netmask") or "255.255.255.0" if not ip or ip.startswith("127."): continue try: cidr = str(ipaddress.IPv4Network(f"{ip}/{nm}", strict=False)) except Exception: cidr = None result.append({"name": i["name"], "ip": ip, "netmask": nm, "cidr": cidr, "mac": i.get("mac", ""), "is_primary": ip == primary}) if not result and primary: net = ipaddress.IPv4Network(f"{primary}/24", strict=False) result.append({"name": "primary", "ip": primary, "netmask": "255.255.255.0", "cidr": str(net), "mac": "", "is_primary": True}) result.sort(key=lambda r: (not r["is_primary"], r["name"])) return result def suggest_cidr() -> Optional[str]: ifaces = get_interfaces() for iface in ifaces: if iface["is_primary"] and iface.get("cidr"): return iface["cidr"] for iface in ifaces: if iface.get("cidr"): return iface["cidr"] return None # --------------------------------------------------------------------------- # Host discovery # --------------------------------------------------------------------------- def _ping(ip: str, timeout_ms: int = 700) -> bool: if SYSTEM == "Windows": cmd = ["ping", "-n", "1", "-w", str(timeout_ms), ip] elif SYSTEM == "Darwin": cmd = ["ping", "-c", "1", "-W", str(timeout_ms), ip] else: cmd = ["ping", "-c", "1", "-W", str(max(1, timeout_ms // 1000)), ip] try: r = subprocess.run(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=timeout_ms / 1000 + 1.5) return r.returncode == 0 except Exception: return False def get_arp_table() -> Dict[str, str]: table: Dict[str, str] = {} try: out = subprocess.run(["arp", "-a"], capture_output=True, text=True, timeout=8).stdout if SYSTEM == "Windows": for line in out.splitlines(): m = re.search(r"(\d+\.\d+\.\d+\.\d+)\s+([0-9A-Fa-f]{2}(?:[-:][0-9A-Fa-f]{2}){5})", line) if m: table[m.group(1)] = m.group(2).replace("-", ":").lower() else: for line in out.splitlines(): m = re.search(r"\((\d+\.\d+\.\d+\.\d+)\)\s+at\s+([0-9A-Fa-f:]{11,17})", line) if m: table[m.group(1)] = m.group(2).lower() except Exception: pass if SYSTEM == "Linux": try: with open("/proc/net/arp") as fh: next(fh, None) for line in fh: parts = line.split() if len(parts) >= 4 and parts[3] != "00:00:00:00:00:00": table[parts[0]] = parts[3].lower() except Exception: pass return table def _reverse_dns(ip: str) -> str: try: return socket.gethostbyaddr(ip)[0] except Exception: return "" def _quick_ports(ip: str) -> List[dict]: """Fast profile probe (QUICK_PORTS) so the table can classify at a glance.""" found: List[dict] = [] with concurrent.futures.ThreadPoolExecutor(max_workers=len(QUICK_PORTS)) as ex: futs = [ex.submit(_check_port, ip, p, 0.6) for p in QUICK_PORTS] for fut in concurrent.futures.as_completed(futs): res = fut.result() if res: found.append(res) found.sort(key=lambda p: p["port"]) return found def scan_network(cidr: str, max_workers: int = 100, progress=None) -> List[dict]: net = ipaddress.IPv4Network(cidr, strict=False) hosts = [str(h) for h in net.hosts()] if net.num_addresses > 2 else [str(net.network_address)] total = len(hosts) alive: List[str] = [] done = 0 with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as ex: futures = {ex.submit(_ping, ip): ip for ip in hosts} for fut in concurrent.futures.as_completed(futures): ip = futures[fut] done += 1 try: if fut.result(): alive.append(ip) except Exception: pass if progress: progress(done, total) arp = get_arp_table() for ip in arp: try: if ipaddress.IPv4Address(ip) in net and ip not in alive: alive.append(ip) except Exception: pass # reverse DNS + quick profile in parallel names: Dict[str, str] = {} quick: Dict[str, List[dict]] = {} with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as ex: dfuts = {ex.submit(_reverse_dns, ip): ip for ip in alive} qfuts = {ex.submit(_quick_ports, ip): ip for ip in alive} for fut in concurrent.futures.as_completed(dfuts): names[dfuts[fut]] = fut.result() for fut in concurrent.futures.as_completed(qfuts): quick[qfuts[fut]] = fut.result() devices: List[dict] = [] for ip in alive: mac = arp.get(ip, "") vendor = vendor_from_mac(mac) qports = quick.get(ip, []) cls = classify_device(vendor, qports, names.get(ip, "")) devices.append({ "ip": ip, "mac": mac, "vendor": vendor, "hostname": names.get(ip, ""), "ports": None, # full scan filled on demand "quick_ports": qports, # from fast profile "category": cls["category"], "type_label": cls["type_label"], "icon": cls["icon"], "web_remote": _resolve_ip(cls["web_remote"], ip), "capabilities": _resolve_caps(cls["capabilities"], ip), "last_seen": time.time(), }) devices.sort(key=lambda d: ipaddress.IPv4Address(d["ip"])) return devices def _resolve_ip(template: Optional[str], ip: str) -> Optional[str]: return template.replace("__IP__", ip) if template else None def _resolve_caps(caps: List[dict], ip: str) -> List[dict]: out = [] for c in caps: c = dict(c) if "url" in c: c["url"] = c["url"].replace("__IP__", ip) out.append(c) return out # --------------------------------------------------------------------------- # Port / service scanning # --------------------------------------------------------------------------- def _grab_banner(ip: str, port: int, timeout: float = 1.0) -> str: try: with socket.create_connection((ip, port), timeout=timeout) as s: s.settimeout(timeout) if port in (80, 8080, 8000, 8888, 3000, 9000, 9090): s.sendall(b"HEAD / HTTP/1.0\r\n\r\n") data = s.recv(256) text = data.decode("latin-1", "ignore").strip() return (text.splitlines()[0] if text else "")[:120] except Exception: return "" def _check_port(ip: str, port: int, timeout: float = 0.8) -> Optional[dict]: try: with socket.create_connection((ip, port), timeout=timeout): pass except Exception: return None return {"port": port, "service": COMMON_PORTS.get(port, "unknown"), "banner": _grab_banner(ip, port)} def scan_ports(ip: str, ports: Optional[List[int]] = None, max_workers: int = 100) -> List[dict]: ports = ports or DEFAULT_SCAN_PORTS open_ports: List[dict] = [] with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as ex: futs = [ex.submit(_check_port, ip, p) for p in ports] for fut in concurrent.futures.as_completed(futs): res = fut.result() if res: open_ports.append(res) open_ports.sort(key=lambda p: p["port"]) return open_ports def parse_ports(spec: str) -> List[int]: if not spec or not spec.strip(): return DEFAULT_SCAN_PORTS ports: set = set() for chunk in spec.split(","): chunk = chunk.strip() if not chunk: continue if "-" in chunk: a, _, b = chunk.partition("-") try: lo, hi = int(a), int(b) for p in range(min(lo, hi), max(lo, hi) + 1): if 0 < p < 65536: ports.add(p) except ValueError: continue else: try: p = int(chunk) if 0 < p < 65536: ports.add(p) except ValueError: continue return sorted(ports) or DEFAULT_SCAN_PORTS """ Change the IP configuration of *this* machine's own interface. This is deliberately explicit and safe: - `build_commands()` returns the exact shell command(s) WITHOUT running them, so the UI can show a preview and require confirmation. - `apply()` runs them, but only when confirm=True. - Changing your own IP can drop your current connection; the UI warns first. Requires administrator / root privileges to actually apply changes. """ import ipaddress import platform import subprocess from typing import List, Optional SYSTEM = platform.system() def _mask_to_prefix(netmask: str) -> int: return ipaddress.IPv4Network(f"0.0.0.0/{netmask}", strict=False).prefixlen def _prefix_to_mask(prefix: int) -> str: return str(ipaddress.IPv4Network(f"0.0.0.0/{prefix}", strict=False).netmask) def validate(ip: str, netmask: str, gateway: Optional[str]) -> Optional[str]: """Return an error string if the config is invalid, else None.""" try: ipaddress.IPv4Address(ip) except Exception: return f"Adresse IP invalide : {ip}" try: prefix = _mask_to_prefix(netmask) except Exception: return f"Masque invalide : {netmask}" if gateway: try: gw = ipaddress.IPv4Address(gateway) except Exception: return f"Passerelle invalide : {gateway}" net = ipaddress.IPv4Network(f"{ip}/{prefix}", strict=False) if gw not in net: return "La passerelle n'est pas dans le même sous-réseau que l'IP." return None def build_commands(interface: str, ip: str, netmask: str, gateway: Optional[str] = None, mode: str = "static") -> List[str]: """Return the exact command(s) that would be run, as display strings. mode: 'static' or 'dhcp' """ if mode == "dhcp": if SYSTEM == "Windows": return [ f'netsh interface ip set address name="{interface}" source=dhcp', f'netsh interface ip set dns name="{interface}" source=dhcp', ] elif SYSTEM == "Darwin": return [f'networksetup -setdhcp "{interface}"'] else: # Linux return [ f"sudo ip addr flush dev {interface}", f"sudo dhclient {interface}", ] prefix = _mask_to_prefix(netmask) if SYSTEM == "Windows": cmds = [ f'netsh interface ip set address name="{interface}" static {ip} {netmask}' + (f" {gateway} 1" if gateway else "") ] return cmds elif SYSTEM == "Darwin": gw = gateway or "empty" return [f'networksetup -setmanual "{interface}" {ip} {netmask} {gw}'] else: # Linux cmds = [ f"sudo ip addr flush dev {interface}", f"sudo ip addr add {ip}/{prefix} dev {interface}", f"sudo ip link set {interface} up", ] if gateway: cmds.append(f"sudo ip route replace default via {gateway} dev {interface}") return cmds def apply(interface: str, ip: str, netmask: str, gateway: Optional[str] = None, mode: str = "static", confirm: bool = False) -> dict: """Apply the config. Returns {ok, commands, output, error}. Nothing runs unless confirm=True. """ if mode == "static": err = validate(ip, netmask, gateway) if err: return {"ok": False, "commands": [], "output": "", "error": err} commands = build_commands(interface, ip, netmask, gateway, mode) if not confirm: return {"ok": True, "commands": commands, "output": "", "error": "", "dry_run": True} outputs: List[str] = [] for cmd in commands: try: # shell=True so the exact preview command runs as shown proc = subprocess.run(cmd, shell=True, capture_output=True, text=True, timeout=30) tag = "OK" if proc.returncode == 0 else f"EXIT {proc.returncode}" piece = f"$ {cmd}\n[{tag}] {proc.stdout}{proc.stderr}".strip() outputs.append(piece) if proc.returncode != 0: return {"ok": False, "commands": commands, "output": "\n\n".join(outputs), "error": f"Échec de la commande (voir sortie). " f"Astuce : lance l'app avec les droits admin/root."} except Exception as e: outputs.append(f"$ {cmd}\n[ERREUR] {e}") return {"ok": False, "commands": commands, "output": "\n\n".join(outputs), "error": str(e)} return {"ok": True, "commands": commands, "output": "\n\n".join(outputs), "error": ""} PAGE = r""" NetScan — Console réseau AV

NetScan · Console réseau AV

…

Découverte du réseau

Prêt.

Appareils détectés

En ligne0
AV identifiés0
Avec page web0
Sous-réseau—
TypeÉtatAdresse IPNom d'hôteFabricantContrôle
Lance un scan pour cartographier le réseau.

Configuration IP de cette machine

— clique sur « Prévisualiser » —
NetScan · outil de régie · à utiliser uniquement sur un réseau que tu administres
""" # --- single-file namespace shim: let app code call scanner.X / ipconfig.X --- class _SelfModule: def __getattr__(self, name): return globals()[name] scanner = ipconfig = _SelfModule() """ NetScan — LAN inventory & AV admin console. Zero-dependency: uses only the Python standard library (http.server). Run it, and it opens http://127.0.0.1:5000 in your browser automatically. Use only on networks you own or are authorized to administer. """ import json import threading import webbrowser from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer # --------------------------------------------------------------------------- # Background scan state (one scan at a time is plenty for a LAN) # --------------------------------------------------------------------------- _scan_lock = threading.Lock() _scan = { "running": False, "done": 0, "total": 0, "cidr": None, "devices": [], "error": "", } def _run_scan(cidr: str) -> None: def progress(done, total): with _scan_lock: _scan["done"] = done _scan["total"] = total try: devices = scanner.scan_network(cidr, progress=progress) with _scan_lock: _scan["devices"] = devices _scan["error"] = "" except Exception as e: # pragma: no cover with _scan_lock: _scan["error"] = str(e) finally: with _scan_lock: _scan["running"] = False # --------------------------------------------------------------------------- # API handlers (return a dict, or (status, dict)) # --------------------------------------------------------------------------- def _h_interfaces(): return {"interfaces": scanner.get_interfaces(), "suggested_cidr": scanner.suggest_cidr(), "system": scanner.SYSTEM} def _h_scan_status(): with _scan_lock: return {"running": _scan["running"], "done": _scan["done"], "total": _scan["total"], "cidr": _scan["cidr"], "devices": _scan["devices"], "error": _scan["error"]} def _h_scan_start(d): cidr = (d.get("cidr") or "").strip() or scanner.suggest_cidr() if not cidr: return 400, {"ok": False, "error": "Aucun sous-réseau détecté."} with _scan_lock: if _scan["running"]: return 409, {"ok": False, "error": "Un scan est déjà en cours."} _scan.update(running=True, done=0, total=0, cidr=cidr, devices=[], error="") threading.Thread(target=_run_scan, args=(cidr,), daemon=True).start() return 200, {"ok": True, "cidr": cidr} def _h_ports(d): ip = (d.get("ip") or "").strip() if not ip: return 400, {"ok": False, "error": "IP manquante."} ports = scanner.parse_ports(d.get("ports", "")) open_ports = scanner.scan_ports(ip, ports) vendor, hostname = "", "" with _scan_lock: for dev in _scan["devices"]: if dev["ip"] == ip: vendor, hostname = dev.get("vendor", ""), dev.get("hostname", "") break cls = scanner.classify_device(vendor, open_ports, hostname) web_remote = scanner._resolve_ip(cls["web_remote"], ip) capabilities = scanner._resolve_caps(cls["capabilities"], ip) with _scan_lock: for dev in _scan["devices"]: if dev["ip"] == ip: dev.update(ports=open_ports, category=cls["category"], type_label=cls["type_label"], icon=cls["icon"], web_remote=web_remote, capabilities=capabilities) break return 200, {"ok": True, "ip": ip, "ports": open_ports, "scanned": len(ports), "type_label": cls["type_label"], "icon": cls["icon"], "category": cls["category"], "web_remote": web_remote, "capabilities": capabilities} def _h_ip_preview(d): cmds = ipconfig.build_commands(d.get("interface", ""), d.get("ip", ""), d.get("netmask", ""), d.get("gateway") or None, d.get("mode", "static")) err = None if d.get("mode", "static") == "static": err = ipconfig.validate(d.get("ip", ""), d.get("netmask", ""), d.get("gateway") or None) return 200, {"ok": err is None, "commands": cmds, "error": err or ""} def _h_ip_apply(d): res = ipconfig.apply(d.get("interface", ""), d.get("ip", ""), d.get("netmask", ""), d.get("gateway") or None, d.get("mode", "static"), confirm=bool(d.get("confirm"))) return 200, res # --------------------------------------------------------------------------- # HTTP server # --------------------------------------------------------------------------- class Handler(BaseHTTPRequestHandler): def log_message(self, *args): # keep the console quiet pass def _send(self, code, body, ctype="application/json"): data = body if isinstance(body, bytes) else body.encode("utf-8") self.send_response(code) self.send_header("Content-Type", ctype + "; charset=utf-8") self.send_header("Content-Length", str(len(data))) self.end_headers() try: self.wfile.write(data) except (BrokenPipeError, ConnectionResetError): pass def _json(self, code, obj): self._send(code, json.dumps(obj), "application/json") def _read_json(self): try: n = int(self.headers.get("Content-Length", 0) or 0) raw = self.rfile.read(n) if n else b"{}" return json.loads(raw or b"{}") except Exception: return {} def do_GET(self): p = self.path.split("?", 1)[0] if p == "/": return self._send(200, PAGE, "text/html") if p == "/api/interfaces": return self._json(200, _h_interfaces()) if p == "/api/scan/status": return self._json(200, _h_scan_status()) return self._send(404, "not found", "text/plain") def do_POST(self): p = self.path.split("?", 1)[0] d = self._read_json() routes = { "/api/scan/start": _h_scan_start, "/api/ports": _h_ports, "/api/ipconfig/preview": _h_ip_preview, "/api/ipconfig/apply": _h_ip_apply, } fn = routes.get(p) if not fn: return self._send(404, "not found", "text/plain") code, obj = fn(d) self._json(code, obj) def main(): host = "0.0.0.0" if "--lan" in sys.argv else "127.0.0.1" port = 5000 url = f"http://127.0.0.1:{port}" server = ThreadingHTTPServer((host, port), Handler) print(f"\n NetScan lancé sur {url}") if host == "0.0.0.0": print(f" Accessible sur le réseau : http://:{port}") print(" (ferme cette fenêtre ou Ctrl+C pour arrêter)\n") print(" Astuce : pour CHANGER une IP, lance avec les droits admin/root.\n") try: threading.Timer(1.0, lambda: webbrowser.open(url)).start() except Exception: pass try: server.serve_forever() except KeyboardInterrupt: print("\n Arrêt de NetScan.") server.shutdown() if __name__ == "__main__": main()