opto_scan.py

#!/usr/bin/env python3
from __future__ import annotations
# https://documents.opto22.com/1465_OptoMMP_Protocol_Guide.pdf
# down round pg 123 starts being useful
import argparse
import concurrent.futures
import ipaddress
import platform
import re
import socket
import subprocess
import sys
import time
from typing import Iterable, List, Optional, Set, Tuple

DISCOVERY_PORTS = (2002, 2001)
DISCOVERY_PAYLOAD = b"\x00\x00\x04\x50\x00\x00\xff\xff\xf0\x30\x00\x20\x01\x30\x00\x00"

DEFAULT_MAC_PREFIXES = [
    "00:a0:3d",
    "6c:bf:b5",
    "b8:27:eb",
]


def parse_target(value: str) -> tuple[ipaddress.IPv4Network, str]:
    """
    Returns:
        network, broadcast_ip

    Behavior:
    - CIDR input: use exactly that network
    - Single IP input: use /24 around it
    - x.x.x.255 input: treat as broadcast, infer /24
    """
    try:
        if "/" in value:
            net = ipaddress.ip_network(value, strict=False)
            if not isinstance(net, ipaddress.IPv4Network):
                raise argparse.ArgumentTypeError("Only IPv4 is supported")
            return net, str(net.broadcast_address)

        ip = ipaddress.ip_address(value)
        if not isinstance(ip, ipaddress.IPv4Address):
            raise argparse.ArgumentTypeError("Only IPv4 is supported")

        if str(ip).endswith(".255"):
            net = ipaddress.ip_network(f"{ip}/24", strict=False)
            return net, str(ip)

        net = ipaddress.ip_network(f"{ip}/24", strict=False)
        return net, str(net.broadcast_address)

    except ValueError as e:
        raise argparse.ArgumentTypeError(str(e)) from e


def discover_targets(broadcast_ip: str) -> None:
    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    try:
        sock.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1)
        for port in DISCOVERY_PORTS:
            sock.sendto(DISCOVERY_PAYLOAD, (broadcast_ip, port))
    finally:
        sock.close()


def build_ping_command(ip: str) -> List[str]:
    system = platform.system().lower()
    if system == "windows":
        return ["ping", "-n", "1", "-w", "700", ip]
    return ["ping", "-c", "1", "-W", "1", ip]


def ping_host(ip: str) -> None:
    try:
        subprocess.run(
            build_ping_command(ip),
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
            check=False,
        )
    except Exception:
        pass


def ping_subset(hosts: List[ipaddress.IPv4Address], workers: int = 64) -> None:
    """
    Run one ping pass. Handles Ctrl+C more cleanly than executor.map().
    """
    executor = concurrent.futures.ThreadPoolExecutor(max_workers=workers)
    futures = []
    try:
        for host in hosts:
            futures.append(executor.submit(ping_host, str(host)))

        for fut in concurrent.futures.as_completed(futures):
            fut.result()

    except KeyboardInterrupt:
        for fut in futures:
            fut.cancel()
        executor.shutdown(wait=False, cancel_futures=True)
        raise

    except Exception:
        executor.shutdown(wait=False, cancel_futures=True)
        raise

    else:
        executor.shutdown(wait=True)


def read_arp_table() -> List[Tuple[str, str]]:
    system = platform.system().lower()
    commands = [["arp", "-a"]] if system == "windows" else [["ip", "neigh"], ["arp", "-n"]]

    output = ""
    for cmd in commands:
        try:
            output = subprocess.check_output(cmd, stderr=subprocess.DEVNULL).decode(errors="ignore")
            if output.strip():
                break
        except Exception:
            continue

    if not output.strip():
        return []

    entries: List[Tuple[str, str]] = []
    seen: Set[Tuple[str, str]] = set()

    unix_re = re.compile(r"(\d+\.\d+\.\d+\.\d+).*?([0-9a-fA-F]{2}(?::[0-9a-fA-F]{2}){5})")
    win_re = re.compile(r"(\d+\.\d+\.\d+\.\d+)\s+([0-9a-fA-F]{2}(?:-[0-9a-fA-F]{2}){5})")

    for ip, mac in unix_re.findall(output):
        key = (ip, mac.lower())
        if key not in seen:
            seen.add(key)
            entries.append(key)

    for ip, mac in win_re.findall(output):
        key = (ip, mac.lower().replace("-", ":"))
        if key not in seen:
            seen.add(key)
            entries.append(key)

    return entries


def mac_matches(mac: str, prefixes: List[str]) -> bool:
    mac = mac.lower()
    return any(mac.startswith(p.lower()) for p in prefixes)


def opto_name_from_mac(mac: str) -> str:
    parts = mac.split(":")
    if len(parts) != 6:
        return "opto-unknown"
    return f"opto-{parts[-3]}-{parts[-2]}-{parts[-1]}"


def in_network(ip: str, network: ipaddress.IPv4Network) -> bool:
    try:
        addr = ipaddress.ip_address(ip)
        return addr in network and addr != network.network_address and addr != network.broadcast_address
    except ValueError:
        return False


def choose_ping_hosts(network: ipaddress.IPv4Network, limit: int = 256) -> List[ipaddress.IPv4Address]:
    hosts = list(network.hosts())
    if len(hosts) <= limit:
        return hosts
    return hosts[:limit]


def initial_probe(network: ipaddress.IPv4Network) -> None:
    """
    One active pass only. No repeated rescanning during the wait loop.
    """
    hosts = choose_ping_hosts(network)
    ping_subset(hosts)


def scan_for_targets(
    network: ipaddress.IPv4Network,
    broadcast_ip: str,
    wait_seconds: int,
    interval: float,
    mac_prefixes: List[str],
    show_all: bool,
    resend: bool,
) -> int:
    print(f"[+] Network:   {network}")
    print(f"[+] Broadcast: {broadcast_ip}")
    print(f"[+] Waiting up to {wait_seconds} seconds")
    print()

    seen: Set[Tuple[str, str]] = set()

    print("[+] Sending discovery packet")
    discover_targets(broadcast_ip)

    print("[+] Doing one initial ARP warm-up ping pass")
    initial_probe(network)

    resend_at: Optional[float] = time.time() + 20 if resend else None
    start = time.time()

    try:
        while True:
            now = time.time()
            if now - start >= wait_seconds:
                break

            if resend and resend_at is not None and now >= resend_at:
                print("[+] Re-sending discovery packet")
                discover_targets(broadcast_ip)
                resend_at = None

            for ip, mac in read_arp_table():
                if not in_network(ip, network):
                    continue

                is_opto = mac_matches(mac, mac_prefixes)
                if not show_all and not is_opto:
                    continue

                key = (ip, mac)
                if key in seen:
                    continue
                seen.add(key)

                if is_opto:
                    print(f"[FOUND] {ip:15}  {mac:17}  {opto_name_from_mac(mac)}")
                else:
                    print(f"[FOUND] {ip:15}  {mac:17}")

            time.sleep(interval)

    except KeyboardInterrupt:
        print("\n[!] Interrupted by user.")
        print(f"[+] Found {len(seen)} device(s) before exit.")
        return 130

    print()
    if not seen:
        print("[-] No matching targets found.")
        return 1

    print(f"[+] Found {len(seen)} device(s).")
    return 0


def main() -> int:
    parser = argparse.ArgumentParser(description="Scan for Opto targets that may appear late.")
    parser.add_argument(
        "target",
        help="IPv4 subnet, host IP, or broadcast IP",
    )
    parser.add_argument(
        "--wait",
        type=int,
        default=60,
        help="Seconds to wait for late-appearing devices (default: 60)",
    )
    parser.add_argument(
        "--interval",
        type=float,
        default=1.0,
        help="ARP poll interval in seconds (default: 1.0)",
    )
    parser.add_argument(
        "--prefix",
        action="append",
        default=[],
        help="Additional MAC prefix to treat as Opto",
    )
    parser.add_argument(
        "--all",
        action="store_true",
        help="Show all devices seen in ARP, not just Opto-like MACs",
    )
    parser.add_argument(
        "--resend",
        action="store_true",
        help="Re-send the discovery packet once during the wait window",
    )

    args = parser.parse_args()
    network, broadcast_ip = parse_target(args.target)
    mac_prefixes = DEFAULT_MAC_PREFIXES + args.prefix

    return scan_for_targets(
        network=network,
        broadcast_ip=broadcast_ip,
        wait_seconds=args.wait,
        interval=args.interval,
        mac_prefixes=mac_prefixes,
        show_all=args.all,
        resend=args.resend,
    )


if __name__ == "__main__":
    sys.exit(main())
'''
# EXAMPLES::
# no optos respond:
$ ./opto_scan.py 192.168.1.69 --wait 60
[+] Network:   192.168.1.0/24
[+] Broadcast: 192.168.1.255
[+] Waiting up to 60 seconds

[+] Sending discovery packet (1/3)
[+] Sending discovery packet (2/3)
[+] Sending discovery packet (3/3)

[-] No matching targets found.

# An opto response:
$ ./opto_scan.py 192.168.1.69 --wait 60
[+] Network:   192.168.1.0/24
[+] Broadcast: 192.168.1.255
[+] Waiting up to 60 seconds

[+] Sending discovery packet (1/3)
[FOUND] 192.168.1.114    00:a0:3d:12:34:56    opto-12-34-56

[+] Found 1 target(s).

# use --all flag to list even non opto devices: #ok thats a bug lol
$ ./opto_scan.py 192.168.1.69 --all
[FOUND] 192.168.1.1      3c:52:82:aa:bb:cc    opto-aa-bb-cc
[FOUND] 192.168.1.10     dc:a6:32:11:22:33    opto-11-22-33
[FOUND] 192.168.1.114    00:a0:3d:12:34:56    opto-12-34-56
'''