ipscanner

#!/usr/bin/env python3
"""
ipscanner — Colorful LAN device scanner with manufacturer + Bonjour info.
Uses arp-scan for L2 discovery, avahi-browse for Bonjour/mDNS, and a local
IEEE OUI database for more reliable manufacturer detection.
"""

import argparse
import csv as csv_mod
import json
import re
import signal
import socket
import subprocess
import sys
import time
from collections import defaultdict
from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import datetime, timezone
from pathlib import Path

# ── ANSI colors ──────────────────────────────────────────────────────────────
RESET = "\033[0m"
BOLD = "\033[1m"
DIM = "\033[2m"
RED = "\033[91m"
GREEN = "\033[92m"
YELLOW = "\033[93m"
BLUE = "\033[94m"
MAGENTA = "\033[95m"
CYAN = "\033[96m"
WHITE = "\033[97m"

VERSION = "1.0.0"
DATA_DIR = Path.home() / ".local" / "share" / "ipscanner"
DB_PATH = DATA_DIR / "vendors.json"
DB_STALE_DAYS = 45
API_CACHE_PATH = DATA_DIR / "api_cache.json"

# Friendly device-type guessing based on vendor + services
DEVICE_ICONS = {
    "apple": "🍎",
    "raspberry": "🍓",
    "amazon": "📦",
    "google": "🔍",
    "samsung": "📱",
    "huawei": "📡",
    "intelbras": "📹",
    "arcadyan": "📶",
    "espressif": "⚡",
    "tp-link": "📶",
    "intel": "💻",
    "dell": "🖥️",
    "lenovo": "💻",
    "microsoft": "🪟",
    "sonos": "🔊",
    "roku": "📺",
    "lg": "📺",
    "sony": "🎮",
    "nvidia": "🎮",
    "philips": "💡",
    "nest": "🏠",
    "ring": "🔔",
    "wyze": "📹",
    "ubiquiti": "📡",
    "netgear": "📶",
}

SERVICE_FRIENDLY = {
    "_http._tcp": "🌐 HTTP",
    "_https._tcp": "🔒 HTTPS",
    "_smb._tcp": "📁 SMB",
    "_afpovertcp._tcp": "📁 AFP",
    "_ssh._tcp": "🔑 SSH",
    "_sftp-ssh._tcp": "🔑 SFTP",
    "_airplay._tcp": "📺 AirPlay",
    "_raop._tcp": "🎵 AirPlay Audio",
    "_googlecast._tcp": "📺 Chromecast",
    "_spotify-connect._tcp": "🎵 Spotify",
    "_printer._tcp": "🖨️  Printer",
    "_ipp._tcp": "🖨️  IPP Print",
    "_scanner._tcp": "📄 Scanner",
    "_device-info._tcp": "ℹ️  Device Info",
    "_companion-link._tcp": "📱 Apple Companion",
    "_homekit._tcp": "🏠 HomeKit",
    "_hap._tcp": "🏠 HomeKit",
    "_sleep-proxy._udp": "😴 Sleep Proxy",
    "_touch-able._tcp": "📱 Apple Remote",
    "_rdlink._tcp": "🔗 Remote Desktop",
    "_nfs._tcp": "📁 NFS",
    "_ftp._tcp": "📁 FTP",
    "_daap._tcp": "🎵 iTunes/DAAP",
    "_dpap._tcp": "📷 iPhoto/DPAP",
    "_pulse-server._tcp": "🔊 PulseAudio",
    "_mqtt._tcp": "📡 MQTT",
    "_coap._udp": "📡 CoAP",
    "_workstation._tcp": "💻 Workstation",
    "_net-assistant._udp": "🔧 Net Assistant",
    "_udisks-ssh._tcp": "💾 Disk (SSH)",
    "_webdav._tcp": "📁 WebDAV",
    "_mediaremotetv._tcp": "📺 Apple TV Remote",
    "_airplay2._tcp": "📺 AirPlay 2",
    "_matter._tcp": "🏠 Matter",
    "_esphomelib._tcp": "⚡ ESPHome",
    "Microsoft Windows Network": "📁 SMB/CIFS",
    "Web Site": "🌐 Web",
    "Device Info": "ℹ️  Device Info",
}


# ── CLI args ──────────────────────────────────────────────────────────────────

def parse_args():
    parser = argparse.ArgumentParser(
        description="ipscanner — LAN device scanner with vendor + Bonjour info",
        add_help=False,
    )
    parser.add_argument("--db-info", action="store_true",
                        help="show local vendor DB info and exit")
    parser.add_argument("--version", action="store_true",
                        help="show version and exit")
    parser.add_argument("--subnet", metavar="CIDR",
                        help="subnet to scan (e.g. 192.168.1.0/24); detected automatically if omitted")
    parser.add_argument("--no-bonjour", action="store_true",
                        help="skip Bonjour/mDNS service discovery")
    parser.add_argument("--json", action="store_true",
                        help="output results as JSON")
    parser.add_argument("--csv", action="store_true",
                        help="output results as CSV")
    parser.add_argument("--timeout", type=int, default=15, metavar="S",
                        help="arp-scan timeout in seconds (default: 15)")
    parser.add_argument("--api", action="store_true",
                        help="query macvendors.com API for unknown MACs (requires internet)")
    parser.add_argument("-h", "--help", action="help",
                        help="show this help message and exit")
    return parser.parse_args()


# ── Vendor DB ─────────────────────────────────────────────────────────────────

def normalize_mac(mac):
    return re.sub(r"[^0-9A-Fa-f]", "", mac or "").upper()


def is_unknown_vendor(vendor):
    if not vendor:
        return True
    text = vendor.strip().lower()
    return text in {"?", "n/a", "private"} or "unknown" in text


def is_locally_administered_mac(mac):
    normalized = normalize_mac(mac)
    if len(normalized) < 2:
        return False
    return bool(int(normalized[:2], 16) & 0x02)


def load_vendor_db():
    if not DB_PATH.exists():
        return None
    try:
        data = json.loads(DB_PATH.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError):
        return None

    prefixes = data.get("prefixes", {})
    return {
        "generated_at": data.get("generated_at"),
        "source_name": data.get("source_name", "IEEE Registration Authority"),
        "sources": data.get("sources", []),
        "counts": data.get("counts", {}),
        "prefixes": {
            9: prefixes.get("9", {}),
            7: prefixes.get("7", {}),
            6: prefixes.get("6", {}),
        },
    }


def lookup_vendor(mac, vendor_db):
    if not vendor_db:
        return None
    normalized = normalize_mac(mac)
    if len(normalized) < 12:
        return None
    for prefix_len in (9, 7, 6):
        vendor = vendor_db["prefixes"][prefix_len].get(normalized[:prefix_len])
        if vendor:
            return vendor
    return None


# ── API fallback (macvendors.com) ─────────────────────────────────────────────

_api_cache: dict = {}


def _load_api_cache():
    global _api_cache
    if API_CACHE_PATH.exists():
        try:
            _api_cache = json.loads(API_CACHE_PATH.read_text())
        except Exception:
            _api_cache = {}


def _save_api_cache():
    try:
        DATA_DIR.mkdir(parents=True, exist_ok=True)
        API_CACHE_PATH.write_text(json.dumps(_api_cache))
    except Exception:
        pass


def lookup_vendor_api(mac: str) -> str | None:
    """Query macvendors.com for unknown OUIs, with persistent local cache."""
    prefix = normalize_mac(mac)[:6]
    if prefix in _api_cache:
        return _api_cache[prefix] or None

    try:
        import urllib.request
        url = f"https://api.macvendors.com/{mac}"
        req = urllib.request.Request(url, headers={"User-Agent": f"ipscanner/{VERSION}"})
        with urllib.request.urlopen(req, timeout=3) as resp:
            vendor = resp.read().decode().strip()
            _api_cache[prefix] = vendor
            return vendor
    except Exception:
        _api_cache[prefix] = ""  # negative cache
        return None


def enrich_vendors_via_api(devices: list) -> list:
    """Fill in missing vendors via macvendors.com API, in parallel."""
    unknown = [(i, mac) for i, (ip, mac, vendor) in enumerate(devices)
               if is_unknown_vendor(vendor)]
    if not unknown:
        return devices

    def fetch(args):
        idx, mac = args
        return idx, lookup_vendor_api(mac)

    devices = list(devices)
    with ThreadPoolExecutor(max_workers=5) as pool:
        for idx, vendor in pool.map(fetch, unknown):
            ip, mac, _ = devices[idx]
            devices[idx] = (ip, mac, vendor or "Unknown")

    _save_api_cache()
    return devices


# ── Context-based vendor guessing ─────────────────────────────────────────────

def guess_vendor_from_context(hostname, services_list):
    haystack = " ".join(
        filter(None, [hostname] + [item for svc in services_list for item in svc[:3]])
    ).lower()

    if any(m in haystack for m in ["macbook", "mac mini", "mac-mini", "iphone", "ipad",
                                    "imac", "airplay", "raop", "companion-link",
                                    "apple-mobdev", "remotepairing"]):
        return "Apple (provável, MAC privado)"
    if any(m in haystack for m in ["googlecast", "chromecast", "nest"]):
        return "Google (provável, MAC privado)"
    if any(m in haystack for m in ["amazon fire tv", "fire tv", "alexa", "echo"]):
        return "Amazon (provável, MAC privado)"
    return None


def resolve_vendor(mac, arp_vendor, vendor_db, hostname, services_list):
    db_vendor = lookup_vendor(mac, vendor_db)
    if db_vendor and db_vendor.strip().lower() != "private":
        return db_vendor
    if arp_vendor and not is_unknown_vendor(arp_vendor):
        return arp_vendor.strip()
    if is_locally_administered_mac(mac):
        guessed = guess_vendor_from_context(hostname, services_list)
        if guessed:
            return guessed
    if db_vendor:
        return db_vendor
    return arp_vendor.strip() if arp_vendor else "Unknown"


# ── Vendor DB status / info ───────────────────────────────────────────────────

def vendor_db_status(vendor_db):
    if not vendor_db:
        return "Fabricantes: fallback do arp-scan (sem base IEEE local)"
    generated_at = vendor_db.get("generated_at")
    if not generated_at:
        return "Fabricantes: IEEE local"
    try:
        dt = datetime.fromisoformat(generated_at.replace("Z", "+00:00"))
        age_days = (datetime.now(timezone.utc) - dt).days
    except ValueError:
        return "Fabricantes: IEEE local"
    if age_days > DB_STALE_DAYS:
        return f"Fabricantes: IEEE local ({age_days} dias, rode ipscanner-update-db)"
    return f"Fabricantes: IEEE local ({age_days} dias)"


def print_db_info(vendor_db):
    print(f"DB path: {DB_PATH}")
    if not vendor_db:
        print("Status: ausente")
        print("Ação: rode ipscanner-update-db")
        return
    print("Status: presente")
    print(f"Fonte: {vendor_db.get('source_name', 'IEEE Registration Authority')}")
    print(f"Gerado em: {vendor_db.get('generated_at', '?')}")
    counts = vendor_db.get("counts", {})
    print(
        "Entradas: "
        f"MA-L={counts.get('6', 0)} "
        f"MA-M={counts.get('7', 0)} "
        f"MA-S={counts.get('9', 0)}"
    )
    print(vendor_db_status(vendor_db))


# ── Network interface + gateway ───────────────────────────────────────────────

def get_interface_and_gateway():
    """Return (iface, gateway_ip) from the default route."""
    iface = "eth0"
    gateway_ip = None
    try:
        out = subprocess.check_output(
            ["ip", "route", "show", "default"], text=True
        )
        for line in out.splitlines():
            # "default via 192.168.1.1 dev eth0 ..."
            m = re.search(r"default via (\S+) dev (\S+)", line)
            if m:
                gateway_ip = m.group(1)
                candidate = m.group(2)
                if not candidate.startswith(("docker", "br-", "veth", "nord", "tun", "wg")):
                    iface = candidate
                break
    except Exception:
        pass
    return iface, gateway_ip


def get_my_ip(iface):
    try:
        out = subprocess.check_output(["ip", "-4", "addr", "show", iface], text=True)
        m = re.search(r"inet\s+(\d+\.\d+\.\d+\.\d+)", out)
        if m:
            return m.group(1)
    except Exception:
        pass
    return None


def get_my_mac(iface):
    try:
        out = subprocess.check_output(["ip", "link", "show", iface], text=True)
        m = re.search(r"link/ether\s+([0-9a-fA-F:]{17})", out)
        if m:
            return m.group(1).lower()
    except Exception:
        pass
    return "??:??:??:??:??:??"


# ── ARP scan ──────────────────────────────────────────────────────────────────

def run_arp_scan(iface, subnet=None, timeout=15):
    """Run arp-scan and return list of (ip, mac, vendor)."""
    devices = []
    seen = set()
    cmd = ["sudo", "arp-scan", f"--interface={iface}", f"--retry=2",
           f"--timeout={timeout * 1000}"]
    if subnet:
        cmd.append(subnet)
    else:
        cmd.append("--localnet")

    try:
        out = subprocess.check_output(
            cmd, text=True, stderr=subprocess.DEVNULL, timeout=timeout + 5
        )
        for line in out.splitlines():
            m = re.match(r"(\d+\.\d+\.\d+\.\d+)\s+([0-9a-fA-F:]{17})\s+(.*)", line.strip())
            if m:
                ip = m.group(1)
                mac = m.group(2).lower()
                vendor = m.group(3).strip()
                key = (ip, mac)
                if key not in seen:
                    seen.add(key)
                    devices.append((ip, mac, vendor))
    except (subprocess.CalledProcessError, subprocess.TimeoutExpired, FileNotFoundError) as exc:
        print(f"{RED}Erro ao executar arp-scan: {exc}{RESET}", file=sys.stderr)
    return devices


# ── Bonjour / mDNS ───────────────────────────────────────────────────────────

def run_avahi_browse():
    """Run avahi-browse and return dict: ip -> [(name, type, hostname, port)]."""
    services = defaultdict(list)
    try:
        raw = subprocess.check_output(
            ["avahi-browse", "-atrpc"],
            stderr=subprocess.DEVNULL,
            timeout=12,
        )
        out = raw.decode("utf-8", errors="replace")
        for line in out.splitlines():
            if not line.startswith("="):
                continue
            parts = line.split(";")
            if len(parts) < 8:
                continue
            iface_name = parts[1]
            if iface_name.startswith(("veth", "docker", "br-")):
                continue
            svc_name = parts[3].replace("\\032", " ").replace("\\046", "&")
            svc_type = parts[4]
            hostname = parts[6]
            ip = parts[7]
            port = parts[8] if len(parts) > 8 else ""
            if ip.startswith("fe80:") or ip.startswith("::"):
                continue
            services[ip].append((svc_name, svc_type, hostname, port))
    except (subprocess.CalledProcessError, subprocess.TimeoutExpired, FileNotFoundError):
        pass
    return services


# ── Parallel DNS resolution ───────────────────────────────────────────────────

def resolve_hostnames_parallel(ips: list) -> dict:
    """Resolve multiple IPs to hostnames concurrently."""
    results = {}

    def resolve_one(ip):
        try:
            return ip, socket.gethostbyaddr(ip)[0]
        except (socket.herror, socket.gaierror, OSError):
            return ip, None

    with ThreadPoolExecutor(max_workers=20) as pool:
        for ip, hostname in pool.map(resolve_one, ips):
            results[ip] = hostname

    return results


# ── Display helpers ───────────────────────────────────────────────────────────

def get_device_icon(vendor, services_list):
    vendor_lower = vendor.lower()
    for key, icon in DEVICE_ICONS.items():
        if key in vendor_lower:
            return icon
    if services_list:
        svc_types = " ".join(svc[1].lower() for svc in services_list)
        if "airplay" in svc_types or "companion" in svc_types:
            return "🍎"
        if "googlecast" in svc_types:
            return "📺"
        if "homekit" in svc_types or "hap" in svc_types:
            return "🏠"
        if "printer" in svc_types or "ipp" in svc_types:
            return "🖨️"
    return "🔌"


def friendly_service(svc_type):
    svc_key = svc_type.strip().rstrip(".")
    if svc_key in SERVICE_FRIENDLY:
        return SERVICE_FRIENDLY[svc_key]
    clean = svc_key.replace("_", "").replace("._tcp", "").replace("._udp", "")
    return f"🔹 {clean}"


def print_header():
    print()
    print(f"  {BOLD}{CYAN}╔══════════════════════════════════════════════════════════════════╗{RESET}")
    print(f"  {BOLD}{CYAN}║{RESET}  {BOLD}{WHITE}📡  IP Scanner — Dispositivos na Rede Local{RESET}                    {BOLD}{CYAN}║{RESET}")
    print(f"  {BOLD}{CYAN}╚══════════════════════════════════════════════════════════════════╝{RESET}")
    print()


def print_device(ip, mac, vendor, hostname, services_list, is_self, is_gateway):
    icon = get_device_icon(vendor, services_list)

    if is_self:
        label = f"{GREEN}{BOLD}★ ESTE DISPOSITIVO{RESET}"
    elif is_gateway:
        gw_vendor = f"  {YELLOW}{vendor}{RESET}" if vendor and "unknown" not in vendor.lower() else ""
        label = f"{YELLOW}{BOLD}⬆ GATEWAY{RESET}{gw_vendor}"
    elif "unknown" in vendor.lower():
        label = f"{DIM}{vendor}{RESET}"
    else:
        label = f"{YELLOW}{vendor}{RESET}"

    print(f"  {CYAN}{'─' * 66}{RESET}")
    print(f"  {icon}  {BOLD}{WHITE}{ip:<18}{RESET} {label}")
    print(f"      {DIM}MAC:{RESET} {MAGENTA}{mac}{RESET}", end="")

    if hostname:
        print(f"   {DIM}DNS:{RESET} {BLUE}{hostname}{RESET}")
    else:
        print()

    if services_list:
        seen = set()
        unique_svcs = []
        for svc_name, svc_type, svc_host, svc_port in services_list:
            key = (svc_type, svc_port)
            if key not in seen:
                seen.add(key)
                unique_svcs.append((svc_name, svc_type, svc_host, svc_port))

        bonjour_name = services_list[0][2]
        if bonjour_name and bonjour_name != hostname:
            print(f"      {DIM}mDNS:{RESET} {GREEN}{bonjour_name}{RESET}")

        svc_labels = []
        for svc_name, svc_type, svc_host, svc_port in unique_svcs:
            friendly = friendly_service(svc_type)
            if svc_port and svc_port != "0":
                friendly += f"{DIM}:{svc_port}{RESET}"
            svc_labels.append(friendly)

        if svc_labels:
            print(f"      {DIM}Serviços:{RESET} {', '.join(svc_labels)}")


# ── Output formatters ─────────────────────────────────────────────────────────

def output_json(resolved_devices, avahi_services, my_ip, gateway_ip):
    out = []
    for ip, mac, vendor, hostname, services in resolved_devices:
        out.append({
            "ip": ip,
            "mac": mac,
            "vendor": vendor,
            "hostname": hostname,
            "is_self": ip == my_ip,
            "is_gateway": ip == gateway_ip,
            "randomly_assigned_mac": is_locally_administered_mac(mac),
            "services": [
                {"name": s[0], "type": s[1], "host": s[2], "port": s[3]}
                for s in services
            ],
        })
    print(json.dumps(out, indent=2, ensure_ascii=False))


def output_csv(resolved_devices, my_ip, gateway_ip):
    writer = csv_mod.writer(sys.stdout)
    writer.writerow(["ip", "mac", "vendor", "hostname", "is_self", "is_gateway", "services"])
    for ip, mac, vendor, hostname, services in resolved_devices:
        svcs = "|".join(f"{s[1]}:{s[3]}" for s in services)
        writer.writerow([ip, mac, vendor, hostname or "", ip == my_ip, ip == gateway_ip, svcs])


# ── Main ──────────────────────────────────────────────────────────────────────

def main():
    args = parse_args()
    quiet = args.json or args.csv

    def _handle_exit(sig, frame):
        if not quiet:
            print(f"\n\n  {YELLOW}Interrompido pelo usuário.{RESET}\n")
        sys.exit(0)

    signal.signal(signal.SIGINT, _handle_exit)
    signal.signal(signal.SIGTERM, _handle_exit)

    vendor_db = load_vendor_db()

    if args.db_info:
        print_db_info(vendor_db)
        return
    if args.version:
        print(VERSION)
        return

    if args.api:
        _load_api_cache()

    if not quiet:
        print_header()

    iface, gateway_ip = get_interface_and_gateway()
    my_ip = get_my_ip(iface)

    if not quiet:
        gw_str = f"  |  Gateway: {gateway_ip}" if gateway_ip else ""
        print(f"  {DIM}Interface: {iface}  |  Meu IP: {my_ip or '?'}{gw_str}{RESET}")
        print(f"  {DIM}{vendor_db_status(vendor_db)}{RESET}")
        print(f"  {DIM}Escaneando...{RESET}", end="", flush=True)

    devices = run_arp_scan(iface, subnet=args.subnet, timeout=args.timeout)

    if my_ip:
        found_ips = {d[0] for d in devices}
        if my_ip not in found_ips:
            devices.append((my_ip, get_my_mac(iface), "Este dispositivo"))

    if args.api:
        if not quiet:
            print(f"\r  {DIM}Consultando API para MACs desconhecidos...{RESET}  ", end="", flush=True)
        devices = enrich_vendors_via_api(devices)

    avahi_services: dict = {}
    if not args.no_bonjour:
        if not quiet:
            print(f"\r  {DIM}Buscando serviços Bonjour/mDNS...{RESET}  ", end="", flush=True)
        avahi_services = run_avahi_browse()

    if not quiet:
        print(f"\r  {GREEN}{BOLD}{len(devices)}{RESET}{DIM} dispositivos encontrados{RESET}          ")

    # Resolve all hostnames in parallel before printing
    all_ips = [d[0] for d in devices]
    hostnames = resolve_hostnames_parallel(all_ips)

    # Sort: gateway first (real IP, not assumed .1), then self, then by IP
    def sort_key(device):
        ip = device[0]
        parts = list(map(int, ip.split(".")))
        if ip == gateway_ip:
            return (0,) + tuple(parts)
        if ip == my_ip:
            return (1,) + tuple(parts)
        return (2,) + tuple(parts)

    devices.sort(key=sort_key)

    # Build resolved device list
    resolved_devices = []
    for ip, mac, arp_vendor in devices:
        hostname = hostnames.get(ip)
        services = avahi_services.get(ip, [])
        vendor = resolve_vendor(mac, arp_vendor, vendor_db, hostname, services)
        resolved_devices.append((ip, mac, vendor, hostname, services))

    if args.json:
        output_json(resolved_devices, avahi_services, my_ip, gateway_ip)
        return

    if args.csv:
        output_csv(resolved_devices, my_ip, gateway_ip)
        return

    for ip, mac, vendor, hostname, services in resolved_devices:
        print_device(ip, mac, vendor, hostname, services,
                     is_self=ip == my_ip, is_gateway=ip == gateway_ip)

    print(f"  {CYAN}{'─' * 66}{RESET}")
    print()

    vendors = defaultdict(int)
    for _, _, vendor, _, _ in resolved_devices:
        key = vendor if "unknown" not in vendor.lower() else "Desconhecido"
        vendors[key] += 1

    if vendors:
        print(f"  {BOLD}Resumo por fabricante:{RESET}")
        for vendor, count in sorted(vendors.items(), key=lambda item: (-item[1], item[0].lower())):
            bar = f"{CYAN}{'█' * count}{RESET}"
            print(f"    {bar} {vendor} ({count})")
        print()


if __name__ == "__main__":
    main()