Measure real throughput to an iperf3 peer on the LAN — the thing an internet speed test can't do (switch/cable/link validation between two nodes). `reverse` tests download (peer->here); `udp` reports jitter/loss.
(server, port=_IPERF3_PORT, duration=5, reverse=False, udp=False)
| 1217 | |
| 1218 | def _mac_watch_load(): |
| 1219 | try: |
| 1220 | with open(_MAC_WATCH_PATH) as f: |
| 1221 | d = json.load(f) |
| 1222 | return d if isinstance(d, dict) else {} |
| 1223 | except (OSError, ValueError): |
| 1224 | return {} |
| 1225 | |
| 1226 | |
| 1227 | def _mac_watch_save(d): |
| 1228 | try: |
| 1229 | os.makedirs(os.path.dirname(_MAC_WATCH_PATH), exist_ok=True) |
| 1230 | tmp = _MAC_WATCH_PATH + '.tmp' |
| 1231 | with open(tmp, 'w') as f: |
| 1232 | json.dump(d, f, indent=2) |
| 1233 | os.replace(tmp, _MAC_WATCH_PATH) |
| 1234 | except OSError: |
| 1235 | pass |
| 1236 | |
| 1237 | |
| 1238 | def _fmt_ago(seconds): |
| 1239 | seconds = max(0, int(seconds)) |
| 1240 | if seconds < 60: |
| 1241 | return f'{seconds}s ago' |
| 1242 | if seconds < 3600: |
| 1243 | return f'{seconds // 60}m ago' |
| 1244 | if seconds < 86400: |
| 1245 | return f'{seconds // 3600}h ago' |
| 1246 | return f'{seconds // 86400}d ago' |
| 1247 | |
| 1248 | |
| 1249 | def do_mac_watch_reset(): |
| 1250 | """Clear the MAC-watch history (first/last-seen, per-IP history, events).""" |
| 1251 | with _mac_watch_lock: |
| 1252 | _mac_watch_save({}) |
| 1253 | return {'success': True, 'reset': True} |
| 1254 | |
| 1255 | |
| 1256 | def do_mac_watch(scan=True, interface=None): |
| 1257 | """Detect current + past MAC spoofing/cloning and randomization, and track |
| 1258 | devices that rotate randomized MACs. Detection-only; nothing is spoofed. |
| 1259 | |
| 1260 | scan=True adds an arp-scan sweep (active but harmless) to widen coverage |
| 1261 | beyond whatever is already in the neighbour table; scan=False reads the |
| 1262 | neighbour table only (works unprivileged, no traffic generated).""" |
| 1263 | now = time.time() |
| 1264 | gw = _default_gateway() |
| 1265 | gw_mac = _neigh_mac(gw) if gw else None |
| 1266 | local = _local_macs() |
| 1267 | |
| 1268 | # Current (ip, mac) observations from the neighbour table, optionally |
| 1269 | # widened by an arp-scan sweep. Deduped; own-NIC MACs dropped. |
| 1270 | seen = {} # mac -> set(ips) |
| 1271 | for ip, mac, _state in _neigh_entries(): |
| 1272 | m = _mac_norm(mac) |
no test coverage detected