#!/usr/bin/env python3 """Live home network sensor: Find, for a house. Runs on the machine that sits on the network (for the demo, Greg's MacBook on his home Wi-Fi). Standard library only. It discovers the devices on the local subnet the same way the platform discovers agents: count everything, name what you can, and flag the rest as UNACCOUNTED until a human registers or removes it. What it does, in plain terms: 1. Finds the local subnet from this machine's own address. 2. Sweeps the subnet with one ping per address so the ARP table fills (user level, no administrator rights, no packet capture). 3. Reads the ARP table: IP, MAC, and a vendor guess from the MAC prefix (small built in table, extend it in OUI below). 4. Browses mDNS and Bonjour service announcements for named things (TVs, printers, speakers, HomeKit devices) where available. 5. Writes home-census.json and appends discovery events to live-events.jsonl in the platform's sniffable shape. Watch mode rescans on an interval and emits a NEW_DEVICE event the moment an address appears that was not in the first census: the live demo beat. A phone joining the Wi-Fi becomes the undeclared agent. Honest limits: this reads the same discovery evidence any device on the network can see. It captures no packets, reads no traffic content, and blocks nothing. Devices that are asleep or firewalled against ping can be missed; the census says what was found, not what exists. Usage: python3 sensor.py [--out DIR] [--watch SECONDS] [--subnet A.B.C.0/24] """ import argparse import json import re import socket import subprocess import sys import time from datetime import datetime, timezone from pathlib import Path # Small MAC prefix (OUI) table: enough to name common home gear. # First three octets, uppercase, colon separated. OUI = { "00:17:88": "Philips Hue", "18:B4:30": "Google Nest", "F0:9F:C2": "Ubiquiti", "DC:A6:32": "Raspberry Pi", "E4:5F:01": "Raspberry Pi", "00:1B:63": "Apple", "3C:22:FB": "Apple", "8C:85:90": "Apple", "A4:83:E7": "Apple", "F0:18:98": "Apple", "AC:BC:32": "Apple", "00:04:4B": "NVIDIA", "48:B0:2D": "NVIDIA", "94:EA:32": "Sonos", "78:28:CA": "Sonos", "5C:AA:FD": "Sonos", "18:B1:DE": "Amazon (Echo/Ring)", "FC:A1:83": "Amazon (Echo/Ring)", "68:54:FD": "Amazon (Echo/Ring)", "00:FC:8B": "Amazon (Echo/Ring)", "D8:A0:1D": "Amazon (Echo/Ring)", "7C:2F:80": "Ring", "90:48:6C": "Ring", "00:11:32": "Synology", "24:5E:BE": "QNAP", "B8:27:EB": "Raspberry Pi", "28:CD:C1": "Raspberry Pi", "04:CF:8C": "TP-Link", "50:C7:BF": "TP-Link", "AC:84:C6": "TP-Link", "00:50:F1": "Denon/Marantz", "8C:F5:A3": "LG", "A8:23:FE": "LG", "7C:1C:4E": "LG", "00:1E:B5": "LG", "8C:EA:48": "Samsung", "5C:3C:27": "Samsung", "84:25:DB": "Samsung", "40:16:3B": "Samsung", "E8:BB:3D": "Samsung", "00:25:90": "Super Micro", "18:03:73": "Dell", "D8:CB:8A": "Dell", "98:90:96": "Dell", "3C:F8:62": "Intel", "8C:8C:AA": "Intel", "34:E6:AD": "Intel", "00:15:5D": "Microsoft (Hyper-V)", "00:0C:29": "VMware", "00:50:56": "VMware", "52:54:00": "QEMU/KVM (virtual)", "02:00:4C": "Virtual (locally administered)", } def now_iso(): return datetime.now(timezone.utc).isoformat() def vendor_for(mac): if not mac: return None prefix = ":".join(mac.upper().split(":")[:3]) if mac.upper().startswith(("02:", "06:", "0A:", "0E:")): return "Randomized MAC (phone privacy addressing)" return OUI.get(prefix) def local_ip_and_subnet(): """This machine's LAN address and its /24, the standard home case.""" sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) try: sock.connect(("8.8.8.8", 80)) ip = sock.getsockname()[0] except OSError: ip = socket.gethostbyname(socket.gethostname()) finally: sock.close() base = ".".join(ip.split(".")[:3]) return ip, f"{base}.0/24", base def ping_sweep(base, timeout_s=900): """One ping per address so neighbours answer and ARP fills. macOS ping uses -t for timeout seconds; Linux uses -W.""" is_mac = sys.platform == "darwin" procs = [] for host in range(1, 255): addr = f"{base}.{host}" cmd = ["ping", "-c", "1", "-t", "1", addr] if is_mac \ else ["ping", "-c", "1", "-W", "1", addr] try: procs.append(subprocess.Popen( cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)) except FileNotFoundError: return False deadline = time.time() + timeout_s / 1000 * 0 + 20 for proc in procs: try: proc.wait(timeout=max(1, int(deadline - time.time()))) except subprocess.TimeoutExpired: proc.kill() return True def read_arp(): """Parse arp -a output on macOS and Linux into device dicts.""" try: out = subprocess.run(["arp", "-a"], capture_output=True, text=True, timeout=30).stdout except (FileNotFoundError, subprocess.TimeoutExpired): return [] devices = [] # hostname (192.168.1.5) at aa:bb:cc:dd:ee:ff on en0 ifscope [ethernet] pattern = re.compile( r"(?P\S+) \((?P\d+\.\d+\.\d+\.\d+)\) at " r"(?P[0-9a-fA-F:]+|\(incomplete\))") for match in pattern.finditer(out): mac = match.group("mac") if mac == "(incomplete)": continue mac = ":".join(part.zfill(2) for part in mac.split(":")).lower() hostname = match.group("host") devices.append({ "ip": match.group("ip"), "mac": mac, "hostname": None if hostname == "?" else hostname, "vendor": vendor_for(mac), }) devices.sort(key=lambda d: tuple(int(p) for p in d["ip"].split("."))) return devices def mdns_services(timeout=4): """Browse Bonjour service types and instances, where dns-sd exists. Best effort: returns {service_type: [instance names]}.""" services = {} if sys.platform != "darwin": return services try: out = subprocess.run( ["dns-sd", "-B", "_services._dns-sd._udp", "local"], capture_output=True, text=True, timeout=timeout).stdout except (FileNotFoundError, subprocess.TimeoutExpired) as exc: out = getattr(exc, "stdout", "") or "" if isinstance(out, bytes): out = out.decode(errors="replace") for line in out.splitlines(): parts = line.split() if len(parts) >= 7 and parts[3] == "Add": services.setdefault(parts[6], []) return services def next_sequence(out_dir): """Monotonic census sequence with a heartbeat timestamp, the sensor health pattern borrowed from the ARIA comparison: a consumer can tell a live sensor from a stale one by the sequence advancing and the heartbeat staying fresh.""" state_path = Path(out_dir) / "sensor-state.json" seq = 1 if state_path.exists(): try: seq = int(json.loads(state_path.read_text()).get("seq", 0)) + 1 except (json.JSONDecodeError, ValueError, TypeError): seq = 1 state = {"seq": seq, "heartbeat": now_iso()} Path(out_dir).mkdir(parents=True, exist_ok=True) state_path.write_text(json.dumps(state, indent=2) + "\n") return state def census(out_dir, previous=None): sensor = next_sequence(out_dir) my_ip, subnet, base = local_ip_and_subnet() ping_sweep(base) devices = read_arp() services = mdns_services() for device in devices: device["is_this_machine"] = device["ip"] == my_ip device["named"] = bool(device["hostname"] or device["vendor"]) device["status"] = "NAMED" if device["named"] else "UNACCOUNTED" result = { "generated_at": now_iso(), "sensor": sensor, "subnet": subnet, "sensor_ip": my_ip, "device_count": len(devices), "named": sum(1 for d in devices if d["named"]), "unaccounted": sum(1 for d in devices if not d["named"]), "bonjour_service_types": sorted(services.keys()), "devices": devices, } out_dir = Path(out_dir) out_dir.mkdir(parents=True, exist_ok=True) (out_dir / "home-census.json").write_text(json.dumps(result, indent=2) + "\n") events = [] known = {d["mac"] for d in previous} if previous else set() for device in devices: if previous is not None and device["mac"] not in known: events.append({ "timestamp": now_iso(), "kind": "NEW_DEVICE", "ip": device["ip"], "mac": device["mac"], "hostname": device["hostname"], "vendor": device["vendor"], "status": "UNDECLARED", "detail": "New device on the network that was not in the " "previous census. Register it or remove it."}) if events: with (out_dir / "live-events.jsonl").open("a") as fh: for event in events: fh.write(json.dumps(event) + "\n") return result, events def main(): parser = argparse.ArgumentParser(description="Live home network sensor") parser.add_argument("--out", default="live-out") parser.add_argument("--watch", type=int, default=0, help="Rescan every N seconds and emit NEW_DEVICE " "events. 0 means a single census.") args = parser.parse_args() result, _ = census(args.out, previous=None) print(f"Census of {result['subnet']}: {result['device_count']} device(s), " f"{result['named']} named, {result['unaccounted']} UNACCOUNTED.") for d in result["devices"]: label = d["hostname"] or d["vendor"] or "unknown" me = " (this machine)" if d["is_this_machine"] else "" print(f" {d['ip']:<15} {d['mac']} {label}{me} [{d['status']}]") if result["bonjour_service_types"]: print("Bonjour service types seen: " + ", ".join(result["bonjour_service_types"])) print(f"Written to {Path(args.out) / 'home-census.json'}") if args.watch: print(f"Watching every {args.watch}s. Ctrl+C to stop. " "Join the network with a new device to see it caught live.") previous = result["devices"] try: while True: time.sleep(args.watch) result, events = census(args.out, previous=previous) for event in events: print(f"NEW_DEVICE: {event['ip']} {event['mac']} " f"{event['hostname'] or event['vendor'] or 'unknown'} " "is UNDECLARED. Register it or remove it.") previous = result["devices"] except KeyboardInterrupt: print("Watch stopped.") if __name__ == "__main__": main()