jaredandClaude Opus 5.5
aa7341af06
Lint / JS (eslint) (push) Successful in 21s
Lint / Python (flake8) (push) Successful in 1m20s
Lint / Notify on failure (push) Skipped
Lint / Deploy (push) Skipped
Security / Python Security (bandit) (push) Successful in 1m0s
Test / Python Tests (pytest) (push) Successful in 1m48s
Lint / Python (flake8) (pull_request) Successful in 1m44s
Lint / JS (eslint) (pull_request) Successful in 30s
Lint / Notify on failure (pull_request) Skipped
Lint / Deploy (pull_request) Skipped
Security / Python Security (bandit) (pull_request) Successful in 1m42s
Test / Python Tests (pytest) (pull_request) Successful in 1m16s
The dashboard's hand-drawn diagram was out of date. /topology replaces it with the network as the switches report it, every device drawn to scale from its faceplate or cutsheet front, and one cable per physical link leaving the port it is plugged into. - topology.py builds the graph: switch-to-switch cables seen from both ends are merged, servers get one cable per NIC, the gateway's internet port goes to the modem and on to the internet, unknown neighbours are stubs, and gear with no live cable is drawn where it was last seen. - Ports named after a different device than the one plugged in are flagged. - Cables show medium, speed, state and traffic; click one for its details, hover a device to trace its path to the internet; pan, zoom and filters. - The monitor now keeps host_macs (MAC to host NIC) in link_stats. - The old dashboard diagram and its JS and CSS are removed. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
412 lines
20 KiB
Python
412 lines
20 KiB
Python
"""The live network topology: every device and every physical cable between them.
|
|
|
|
Pure functions with no I/O. app.py passes in the monitor's cached state and the
|
|
cutsheet, and the /topology page draws what build_graph returns.
|
|
|
|
The graph comes from the switches' point of view: each UniFi port with a neighbour
|
|
is one end of a cable. The same switch-to-switch cable is seen from both ends, so
|
|
those are merged into one link.
|
|
"""
|
|
import re
|
|
from collections import deque
|
|
from typing import Dict, List, Optional, Tuple
|
|
|
|
import devices
|
|
|
|
INTERNET = 'internet'
|
|
|
|
# How UniFi devices are drawn and grouped. Power gear goes in its own rail.
|
|
_POWER_MODELS = {'USPPDUP': 'pdu', 'USPDA2B': 'ups', 'UP6': 'strip'}
|
|
POWER_KINDS = ('pdu', 'ups', 'strip')
|
|
|
|
_SEVERITY_RANK = {'critical': 3, 'warning': 2, 'info': 1}
|
|
_PORT_REF_RE = re.compile(r'(?:^|\s)port\s*(\d+)$', re.IGNORECASE)
|
|
_TOKEN_RE = re.compile(r'[a-z0-9]+')
|
|
|
|
|
|
def _norm(name: str) -> str:
|
|
"""A name with case, spaces, dashes and underscores ignored: "Large1" == "large-1"."""
|
|
return ''.join(_TOKEN_RE.findall((name or '').lower()))
|
|
|
|
|
|
def _node_kind(entry: dict) -> str:
|
|
if entry['kind'] == 'pdu':
|
|
return _POWER_MODELS.get(entry.get('model', ''), 'pdu')
|
|
return entry['kind']
|
|
|
|
|
|
def _medium(port: dict) -> str:
|
|
"""dac, fibre, sfp (a module that doesn't say which) or copper."""
|
|
sfp = port.get('sfp') or {}
|
|
if sfp.get('found') or sfp.get('part'):
|
|
kind = sfp.get('kind') or ''
|
|
if kind == 'dac':
|
|
return 'dac'
|
|
if kind == 'optic':
|
|
return 'fibre'
|
|
return 'sfp'
|
|
if str(port.get('media') or '').upper().startswith('SFP'):
|
|
return 'sfp' if port.get('up') else 'copper'
|
|
return 'copper'
|
|
|
|
|
|
def _rate(value) -> float:
|
|
try:
|
|
return max(float(value or 0), 0.0)
|
|
except (TypeError, ValueError):
|
|
return 0.0
|
|
|
|
|
|
def _utilisation(a_to_b: float, b_to_a: float, speed_mbps) -> Optional[float]:
|
|
try:
|
|
speed = float(speed_mbps or 0)
|
|
except (TypeError, ValueError):
|
|
speed = 0.0
|
|
if speed <= 0:
|
|
return None
|
|
return round(min(max(a_to_b, b_to_a) * 8 / (speed * 1e6), 1.0), 4)
|
|
|
|
|
|
def _port_stats(port: dict) -> dict:
|
|
sfp = port.get('sfp') or {}
|
|
return {
|
|
'speed_mbps': port.get('speed_mbps') or 0,
|
|
'up': bool(port.get('up')),
|
|
'medium': _medium(port),
|
|
'sfp_part': sfp.get('part') or '',
|
|
'sfp_vendor': sfp.get('vendor') or '',
|
|
'sfp_serial': sfp.get('serial') or '',
|
|
'errors': round(_rate(port.get('rx_errs_rate')) + _rate(port.get('tx_errs_rate')), 3),
|
|
'drops': round(_rate(port.get('rx_drops_rate')) + _rate(port.get('tx_drops_rate')), 3),
|
|
}
|
|
|
|
|
|
def _end(node: str, port_name: str = '', port_idx=None, iface: str = '') -> dict:
|
|
return {'node': node, 'port_idx': port_idx, 'port_name': port_name, 'iface': iface}
|
|
|
|
|
|
def _far_port(switches: dict, here: str, port: dict, neighbour: str) -> Optional[int]:
|
|
"""Which port on a neighbouring switch the cable from `port` lands on, if it can be told."""
|
|
peer = (port.get('sfp') or {}).get('cable_peer') or {}
|
|
if peer.get('switch') == neighbour and peer.get('port_idx') is not None:
|
|
return peer['port_idx']
|
|
far = switches.get(neighbour)
|
|
if not far:
|
|
return None
|
|
ports = [p for p in (far.get('ports') or {}).values() if p.get('port_idx') is not None]
|
|
# LLDP's port ID names the neighbour's own port ("Port 3", "local Port 7")
|
|
m = _PORT_REF_RE.search(str((port.get('lldp') or {}).get('port_id') or '').strip())
|
|
if m and any(p['port_idx'] == int(m.group(1)) for p in ports):
|
|
return int(m.group(1))
|
|
back = [p['port_idx'] for p in ports if (p.get('lldp') or {}).get('system_name') == here]
|
|
if len(back) == 1:
|
|
return back[0]
|
|
# Gear that sends no LLDP of its own (a PDU): its single uplink
|
|
uplinks = [p['port_idx'] for p in ports if p.get('is_uplink')]
|
|
return uplinks[0] if len(uplinks) == 1 else None
|
|
|
|
|
|
def _is_generic_port_name(name: str) -> bool:
|
|
return bool(re.fullmatch(r'(port|sfp\+?|uplink|downlink)\s*\d*', (name or '').strip(), re.IGNORECASE))
|
|
|
|
|
|
def name_mismatch(port_name: str, neighbour: str, own: str, known: Dict[str, str]) -> Optional[dict]:
|
|
"""A port named after a different device than the one plugged into it.
|
|
|
|
known: {normalised name: display name} of every device. Runs of up to four words of
|
|
the port name are compared with them, so "Large1 Management" names large1 and
|
|
"Downlink to 24RJ45-POE" names 24RJ45-POE. Ports named only "Port 4" never match.
|
|
"""
|
|
if not port_name or _is_generic_port_name(port_name):
|
|
return None
|
|
words = _TOKEN_RE.findall(port_name.lower())
|
|
skip = {_norm(neighbour), _norm(own)}
|
|
for size in range(min(4, len(words)), 0, -1):
|
|
for i in range(len(words) - size + 1):
|
|
run = ''.join(words[i:i + size])
|
|
if run in known and run not in skip:
|
|
return {'named': known[run], 'actual': neighbour}
|
|
return None
|
|
|
|
|
|
def _cable_labels(items: List[dict], cables: List[dict]) -> Dict[Tuple[str, str], dict]:
|
|
"""{(device, port): cable} for cutsheet cables still in place. A cable can name the
|
|
device itself or a cutsheet item linked to it."""
|
|
device_of = {i['name']: (i.get('device') or i['name']) for i in items or []}
|
|
out: Dict[Tuple[str, str], dict] = {}
|
|
for c in cables or []:
|
|
if c.get('status') == 'retired':
|
|
continue
|
|
for end in ('a', 'b'):
|
|
dev = c.get(f'{end}_device') or ''
|
|
if not dev:
|
|
continue
|
|
ref = {'id': c.get('id'), 'label': c.get('label') or '', 'type': c.get('cable_type') or '',
|
|
'color': c.get('color') or ''}
|
|
for name in {dev, device_of.get(dev, dev)}:
|
|
out[(name, str(c.get(f'{end}_port') or ''))] = ref
|
|
return out
|
|
|
|
|
|
def _alerts_by_target(events: List[dict]) -> Dict[str, List[dict]]:
|
|
out: Dict[str, List[dict]] = {}
|
|
for ev in events or []:
|
|
if ev.get('is_suppressed'):
|
|
continue
|
|
out.setdefault(ev.get('target_name') or '', []).append({
|
|
'id': ev.get('id'), 'severity': ev.get('severity') or 'info',
|
|
'type': ev.get('event_type') or '', 'detail': ev.get('target_detail') or '',
|
|
'description': ev.get('description') or '',
|
|
})
|
|
return out
|
|
|
|
|
|
def _worst(alerts: List[dict]) -> str:
|
|
return max((a['severity'] for a in alerts), key=lambda s: _SEVERITY_RANK.get(s, 0), default='')
|
|
|
|
|
|
def build_graph(snapshot: dict, link_stats: dict, modem: Optional[dict], events: List[dict],
|
|
cutsheet_items: List[dict], cutsheet_cables: List[dict],
|
|
known_uplinks: Optional[dict] = None) -> dict:
|
|
"""Every device and cable, as {nodes, links, warnings, updated}.
|
|
|
|
snapshot, link_stats, modem: the monitor's saved state.
|
|
events: active events, with is_suppressed set (suppressed ones are left out).
|
|
known_uplinks: the monitor's record of where each device was last plugged in,
|
|
drawn as "last seen" links for a device with no live link.
|
|
"""
|
|
snapshot = snapshot or {}
|
|
link_stats = link_stats or {}
|
|
switches = link_stats.get('unifi_switches') or {}
|
|
unifi_info = link_stats.get('unifi_devices') or {}
|
|
alerts = _alerts_by_target(events)
|
|
labels = _cable_labels(cutsheet_items, cutsheet_cables)
|
|
chassis = {}
|
|
for item in cutsheet_items or []:
|
|
chassis.setdefault(item.get('device') or item['name'], item.get('chassis') or '')
|
|
warnings: List[str] = []
|
|
|
|
# ── Nodes ────────────────────────────────────────────────────────────
|
|
nodes: Dict[str, dict] = {}
|
|
configured = set(snapshot.get('hosts') or {})
|
|
for entry in devices.build_index(snapshot, link_stats, modem):
|
|
name = entry['name']
|
|
if entry['kind'] == 'host' and name not in configured:
|
|
continue # IP-only Prometheus targets: shown only as a stub where a port sees them
|
|
kind = _node_kind(entry)
|
|
node = {
|
|
'id': name, 'name': name, 'kind': kind, 'status': entry['status'], 'model': entry.get('model', ''),
|
|
'ip': entry.get('ip', ''), 'url': f'/device/{name}', 'power': kind in POWER_KINDS,
|
|
}
|
|
if kind == 'host':
|
|
node['chassis'] = chassis.get(name, '')
|
|
node['interfaces'] = devices.physical_interfaces(name, snapshot)
|
|
for nic in node['interfaces']:
|
|
nic_alerts = [a for a in alerts.get(name, []) if a['detail'] == nic['name']]
|
|
nic['alert'] = _worst(nic_alerts)
|
|
elif name in switches or kind in ('router', 'switch'):
|
|
node['ports'] = {
|
|
str(p['port_idx']): {'name': pname, 'up': bool(p.get('up')), 'speed_mbps': p.get('speed_mbps') or 0,
|
|
'is_uplink': bool(p.get('is_uplink')), 'media': p.get('media') or '',
|
|
'poe_power': p.get('poe_power'), 'sfp': p.get('sfp') or None,
|
|
'tx_bytes_rate': _rate(p.get('tx_bytes_rate')),
|
|
'rx_bytes_rate': _rate(p.get('rx_bytes_rate')),
|
|
'neighbour': (p.get('lldp') or {}).get('system_name') or ''}
|
|
for pname, p in ((switches.get(name) or {}).get('ports') or {}).items()
|
|
if p.get('port_idx') is not None
|
|
}
|
|
if kind in POWER_KINDS:
|
|
info = unifi_info.get(name) or {}
|
|
node['power_w'] = info.get('power_w')
|
|
node['outlets'] = info.get('outlets') or []
|
|
node_alerts = alerts.get(name, [])
|
|
node['alerts'] = node_alerts[:10]
|
|
node['alert_count'] = len(node_alerts)
|
|
node['severity'] = _worst(node_alerts)
|
|
nodes[name] = node
|
|
|
|
router = next((n for n in nodes.values() if n['kind'] == 'router'), None)
|
|
modem_name = (modem or {}).get('name') if (modem or {}).get('name') in nodes else None
|
|
wan = ((unifi_info.get(router['name']) or {}).get('wan') or {}) if router else {}
|
|
wan_up = any(w.get('up') for w in wan.values()) if wan else None
|
|
if modem_name:
|
|
wan_up = (modem or {}).get('broadband_state', '').lower() == 'up' if wan_up is None else wan_up
|
|
nodes[INTERNET] = {
|
|
'id': INTERNET, 'name': 'Internet', 'kind': INTERNET, 'model': '', 'ip': '', 'url': '',
|
|
'status': 'unknown' if wan_up is None else ('up' if wan_up else 'down'),
|
|
'alerts': alerts.get('WAN', [])[:10], 'alert_count': len(alerts.get('WAN', [])),
|
|
'severity': _worst(alerts.get('WAN', [])), 'power': False,
|
|
}
|
|
known_names = {_norm(n['name']): n['name'] for n in nodes.values() if n['id'] != INTERNET}
|
|
|
|
# ── Links ────────────────────────────────────────────────────────────
|
|
links: List[dict] = []
|
|
claimed: Dict[Tuple[str, int], dict] = {}
|
|
|
|
def add_link(a: dict, b: dict, port: dict, **extra) -> dict:
|
|
link = {'id': '', 'a': a, 'b': b, 'ghost': False, 'name_mismatch': None,
|
|
'cable': labels.get((a['node'], str(a['port_idx'] if a['port_idx'] is not None else a['iface'])))
|
|
or labels.get((b['node'], str(b['port_idx'] if b['port_idx'] is not None else b['iface']))),
|
|
**_port_stats(port),
|
|
'_tx': _rate(port.get('tx_bytes_rate')), '_rx': _rate(port.get('rx_bytes_rate'))}
|
|
link.update(extra)
|
|
links.append(link)
|
|
for end in (a, b):
|
|
if end['port_idx'] is not None and end['node'] in switches:
|
|
claimed[(end['node'], end['port_idx'])] = link
|
|
return link
|
|
|
|
def ordered_ports(sw: dict):
|
|
return sorted(((pn, p) for pn, p in (sw.get('ports') or {}).items() if p.get('port_idx') is not None),
|
|
key=lambda kv: kv[1]['port_idx'])
|
|
|
|
# The router's internet port goes to the modem, then the modem to the internet
|
|
wan_port = None
|
|
if router and router['name'] in switches:
|
|
wan_port = next(((pn, p) for pn, p in ordered_ports(switches[router['name']]) if p.get('is_uplink')), None)
|
|
upstream = modem_name or INTERNET
|
|
if wan_port:
|
|
pn, p = wan_port
|
|
add_link(_end(router['name'], pn, p['port_idx']), _end(upstream, 'WAN' if modem_name else ''), p, wan=True)
|
|
elif router:
|
|
warnings.append(f'No internet port found on {router["name"]}')
|
|
if modem_name:
|
|
m = modem or {}
|
|
rates = m.get('rates') or {}
|
|
up = str(m.get('broadband_state') or '').lower() == 'up'
|
|
links.append({
|
|
'id': '', 'a': _end(INTERNET), 'b': _end(modem_name, str(m.get('source') or 'WAN').title()),
|
|
'ghost': False, 'name_mismatch': None, 'cable': None, 'wan': True,
|
|
'speed_mbps': m.get('line_speed_mbps') or 0, 'up': up,
|
|
'medium': 'fibre' if str(m.get('source') or '').upper() == 'FIBER' else 'copper',
|
|
'sfp_part': m.get('optics_part_no') or '', 'sfp_vendor': '', 'sfp_serial': '',
|
|
'errors': round(_rate(rates.get('rx_errors')) + _rate(rates.get('tx_errors')), 3),
|
|
'drops': round(_rate(rates.get('rx_drops')) + _rate(rates.get('tx_drops')), 3),
|
|
# Seen from the modem's broadband side: rx comes down from the internet
|
|
'_tx': _rate(rates.get('rx_bytes')), '_rx': _rate(rates.get('tx_bytes')),
|
|
})
|
|
elif not router:
|
|
warnings.append('No UniFi gateway reported yet, so the internet edge is not drawn')
|
|
|
|
# Device-to-device and device-to-host cables, from every switch port with a neighbour
|
|
for sw_name in sorted(switches):
|
|
for pname, port in ordered_ports(switches[sw_name]):
|
|
idx = port['port_idx']
|
|
if (sw_name, idx) in claimed:
|
|
continue
|
|
lldp = port.get('lldp') or {}
|
|
neighbour = lldp.get('system_name') or ''
|
|
here = _end(sw_name, pname, idx)
|
|
if neighbour in nodes and neighbour != sw_name:
|
|
nb = nodes[neighbour]
|
|
if nb['kind'] == 'host':
|
|
far = _end(neighbour, iface=lldp.get('port_id') or '')
|
|
else:
|
|
far_idx = _far_port(switches, sw_name, port, neighbour)
|
|
if far_idx is not None and (neighbour, far_idx) in claimed:
|
|
continue # the other end already drew this cable
|
|
far_port = next((pn for pn, p in ((switches.get(neighbour) or {}).get('ports') or {}).items()
|
|
if p.get('port_idx') == far_idx), '') if far_idx is not None else ''
|
|
far = _end(neighbour, far_port, far_idx)
|
|
link = add_link(here, far, port)
|
|
link['name_mismatch'] = name_mismatch(pname, neighbour, sw_name, known_names)
|
|
if link['name_mismatch'] is None and far['port_name']:
|
|
link['name_mismatch'] = name_mismatch(far['port_name'], sw_name, neighbour, known_names)
|
|
if link['name_mismatch']:
|
|
link['name_mismatch']['end'] = 'b'
|
|
elif link['name_mismatch']:
|
|
link['name_mismatch']['end'] = 'a'
|
|
continue
|
|
if port.get('is_uplink') and not neighbour:
|
|
continue # leads upstream to gear already drawn (a PDU's management port)
|
|
if not port.get('up') and not lldp:
|
|
continue
|
|
# Something is plugged in that Gandalf can't name: a grey stub
|
|
stub_id = f'stub:{sw_name}:{idx}'
|
|
mac = lldp.get('chassis_id') or ''
|
|
label = neighbour or ('' if _is_generic_port_name(pname) else pname) or mac or 'unknown device'
|
|
nodes[stub_id] = {
|
|
'id': stub_id, 'name': label, 'kind': 'stub', 'status': 'up' if port.get('up') else 'down',
|
|
'model': '', 'ip': neighbour if re.fullmatch(r'[\d.]+', neighbour) else '', 'mac': mac,
|
|
'url': f'/device/{sw_name}#port-{idx}', 'switch': sw_name, 'port_idx': idx, 'port_name': pname,
|
|
'iface': (lldp.get('port_id') or '') if neighbour else '', 'anonymous': not lldp,
|
|
'alerts': [], 'alert_count': 0, 'severity': '', 'power': False,
|
|
}
|
|
link = add_link(here, _end(stub_id, iface=nodes[stub_id]['iface']), port)
|
|
# A stub labelled by its port name would always "match" it, so compare with what was seen
|
|
link['name_mismatch'] = name_mismatch(pname, neighbour or mac or 'an unnamed device', sw_name, known_names)
|
|
if link['name_mismatch']:
|
|
link['name_mismatch']['end'] = 'a'
|
|
|
|
# Gear with no live cable: draw where it was last plugged in
|
|
linked = {e['node'] for link in links for e in (link['a'], link['b'])}
|
|
for name, entry in sorted((known_uplinks or {}).items()):
|
|
node = nodes.get(name)
|
|
if not node or name in linked or node['kind'] == 'stub':
|
|
continue
|
|
for up in entry.get('uplinks') or []:
|
|
sw, idx = up.get('switch'), up.get('port_idx')
|
|
if sw not in nodes or idx is None or (sw, idx) in claimed:
|
|
continue
|
|
port = next((p for p in ((switches.get(sw) or {}).get('ports') or {}).values()
|
|
if p.get('port_idx') == idx), {})
|
|
far = (_end(name, iface=up.get('host_interface') or '') if node['kind'] == 'host'
|
|
else _end(name, '', None))
|
|
link = add_link(_end(sw, up.get('port') or '', idx), far, port,
|
|
ghost=True, seen=entry.get('seen') or '')
|
|
link['up'] = False
|
|
|
|
# ── Depth: hops from the internet over device-to-device cables ───────
|
|
adjacency: Dict[str, set] = {}
|
|
for link in links:
|
|
a, b = link['a']['node'], link['b']['node']
|
|
adjacency.setdefault(a, set()).add(b)
|
|
adjacency.setdefault(b, set()).add(a)
|
|
depth: Dict[str, int] = {INTERNET: 0}
|
|
queue = deque([INTERNET])
|
|
while queue:
|
|
cur = queue.popleft()
|
|
for nxt in sorted(adjacency.get(cur, ())):
|
|
if nxt in depth:
|
|
continue
|
|
depth[nxt] = depth[cur] + 1
|
|
# Hosts and stubs are leaves: they never carry the path on to another device
|
|
if nodes[nxt]['kind'] not in ('host', 'stub'):
|
|
queue.append(nxt)
|
|
for node in nodes.values():
|
|
node['depth'] = depth.get(node['id'])
|
|
|
|
# Orient each cable from the internet side (a) to the far side (b), and count traffic that way
|
|
for i, link in enumerate(links):
|
|
da = depth.get(link['a']['node'], 1 << 20)
|
|
db_ = depth.get(link['b']['node'], 1 << 20)
|
|
# Rates were measured at end a as added: the reporting switch port, or the internet side
|
|
tx, rx = link.pop('_tx'), link.pop('_rx')
|
|
if db_ < da:
|
|
link['a'], link['b'] = link['b'], link['a']
|
|
if link.get('name_mismatch') and link['name_mismatch'].get('end'):
|
|
link['name_mismatch']['end'] = 'a' if link['name_mismatch']['end'] == 'b' else 'b'
|
|
tx, rx = rx, tx
|
|
link['a_to_b_bps'] = round(tx, 1)
|
|
link['b_to_a_bps'] = round(rx, 1)
|
|
link['utilisation'] = _utilisation(tx, rx, link['speed_mbps'])
|
|
link['id'] = f'l{i}'
|
|
|
|
# Which host NIC each cable lands on, so the drawing can light the right socket
|
|
for link in links:
|
|
for end in (link['a'], link['b']):
|
|
node = nodes[end['node']]
|
|
if node['kind'] == 'host' and end['iface']:
|
|
for nic in node.get('interfaces', []):
|
|
if nic['name'] == end['iface']:
|
|
nic.setdefault('links', []).append(link['id'])
|
|
|
|
if not switches:
|
|
warnings.append('No UniFi switch data yet: the monitor has not reported any ports')
|
|
return {
|
|
'nodes': sorted(nodes.values(), key=lambda n: (99 if n['depth'] is None else n['depth'], n['name'].lower())),
|
|
'links': links,
|
|
'warnings': warnings,
|
|
'updated': link_stats.get('updated'),
|
|
}
|