diff --git a/package/klish-plugin-infix/klish-plugin-infix.mk b/package/klish-plugin-infix/klish-plugin-infix.mk index 23c0c387..2b5f4e6d 100644 --- a/package/klish-plugin-infix/klish-plugin-infix.mk +++ b/package/klish-plugin-infix/klish-plugin-infix.mk @@ -28,7 +28,6 @@ KLISH_PLUGIN_INFIX_CONF_OPTS += --enable-shell else KLISH_PLUGIN_INFIX_CONF_OPTS += --disable-shell endif - ifeq ($(BR2_PACKAGE_BASH),y) KLISH_PLUGIN_INFIX_CONF_OPTS += --with-shell=/bin/bash else diff --git a/src/show/show.py b/src/show/show.py index 49d2a3e3..573ee160 100755 --- a/src/show/show.py +++ b/src/show/show.py @@ -164,6 +164,20 @@ def lldp(args: List[str]): return cli_pretty(data, "show-lldp") +def wifi(args: List[str]): + iface = args[0] + if len(args) == 0: + print("Illigal usage") + return + if is_valid_interface_name(iface): + if not os.path.exists(f"/sys/class/net/{iface}/wireless"): + print("Not a Wi-Fi interface") + return + data = run_sysrepocfg("/ietf-interfaces:interfaces") + cli_pretty(data, "show-wifi-scan", "-n", iface) + else: + print(f"Invalid interface name: {iface}") + def execute_command(command: str, args: List[str]): command_mapping = { 'dhcp': dhcp, @@ -174,6 +188,7 @@ def execute_command(command: str, args: List[str]): 'lldp': lldp, 'software' : software, 'stp': stp, + 'wifi': wifi } if command in command_mapping: diff --git a/src/statd/python/cli_pretty/cli_pretty.py b/src/statd/python/cli_pretty/cli_pretty.py index 19e64c54..232818ce 100755 --- a/src/statd/python/cli_pretty/cli_pretty.py +++ b/src/statd/python/cli_pretty/cli_pretty.py @@ -82,6 +82,20 @@ class PadNtpSource: poll = 14 +class PadWifiScan: + ssid = 40 + encryption = 30 + signal = 9 + + +class PadLldp: + interface = 16 + rem_idx = 10 + time = 12 + chassis_id = 20 + port_id = 20 + + class Decore(): @staticmethod def decorate(sgr, txt, restore="0"): @@ -103,6 +117,10 @@ class Decore(): def green(txt): return Decore.decorate("32", txt, "39") + @staticmethod + def bright_green(txt): + return Decore.decorate("1;32", txt, "39") + @staticmethod def yellow(txt): return Decore.decorate("33", txt, "39") @@ -116,12 +134,17 @@ class Decore(): return Decore.decorate("100", txt) -class PadLldp: - interface = 16 - rem_idx = 10 - time = 12 - chassis_id = 20 - port_id = 20 +def rssi_to_status(rssi): + if rssi <= -75: + status = Decore.bright_green("excellent") + elif rssi <= -65: + status = Decore.green("good") + elif rssi <= -50: + status = Decore.yellow("poor") + else: + status = Decore.red("bad") + + return status def datetime_now(): @@ -196,7 +219,6 @@ class Date(datetime): tz = tz.replace(":", "") return cls.strptime(f"{date}+{tz}", "%Y-%m-%dT%H:%M:%S%z") - class Route: def __init__(self, data, ip): self.data = data @@ -524,12 +546,16 @@ class Iface: self.gre = self.data.get('infix-interfaces:gre') self.vxlan = self.data.get('infix-interfaces:vxlan') + self.wifi = self.data.get('infix-interfaces:wifi') if self.data.get('infix-interfaces:vlan'): self.lower_if = self.data.get('infix-interfaces:vlan', None).get('lower-layer-if',None) else: self.lower_if = '' + def is_wifi(self): + return self.type == "infix-if-type:wifi" + def is_vlan(self): return self.type == "infix-if-type:vlan" @@ -554,7 +580,7 @@ class Iface: def is_gretap(self): return self.data['type'] == "infix-if-type:gretap" - + def oper(self, detail=False): """Remap in brief overview to fit column widths.""" if not detail and self.oper_status == "lower-layer-down": @@ -625,6 +651,47 @@ class Iface: row = self._pr_proto_common("loopback", False, pipe); print(row) + def pr_wifi_ssids(self): + hdr = (f"{'SSID':<{PadWifiScan.ssid}}" + f"{'ENCRYPTION':<{PadWifiScan.encryption}}" + f"{'SIGNAL':<{PadWifiScan.signal}}" + ) + + print(Decore.invert(hdr)) + results=self.wifi.get("scan-results", {}) + for result in results: + encstr = ",".join(result["encryption"]) + status=rssi_to_status(result["rssi"]) + row = f"{result['ssid']:<{PadWifiScan.ssid}}" + row += f"{encstr:<{PadWifiScan.encryption}}" + row += f"{status:<{PadWifiScan.signal}}" + + print(row) + + + def pr_proto_wifi(self, pipe=''): + row = self._pr_proto_common("ethernet", True, pipe); + print(row) + ssid = None + rssi = None + + if self.wifi: + rssi=self.wifi.get("rssi") + ssid=self.wifi.get("ssid") + if ssid is None: + ssid="------" + + if rssi is None: + signal="------" + else: + signal=rssi_to_status(rssi) + data_str = f"ssid: {ssid}, signal: {signal}" + + row = f"{'':<{Pad.iface}}" + row += f"{'wifi':<{Pad.proto}}" + row += f"{'':<{Pad.state}}{data_str}" + print(row) + def pr_proto_br(self, br_vlans): data_str = "" @@ -765,6 +832,12 @@ class Iface: self.pr_proto_ipv4() self.pr_proto_ipv6() + def pr_wifi(self): + self.pr_name(pipe="") + self.pr_proto_wifi() + self.pr_proto_ipv4() + self.pr_proto_ipv6() + def pr_vlan(self, _ifaces): self.pr_name(pipe="") self.pr_proto_eth() @@ -874,6 +947,14 @@ class Iface: else: print(f"{'ipv6 addresses':<{20}}:") + if self.wifi: + ssid=self.wifi.get('ssid', "----") + rssi=self.wifi.get('rssi', "----") + print(f"{'SSID':<{20}}: {ssid}") + print(f"{'Signal':<{20}}: {rssi}") + print("") + self.pr_wifi_ssids() + if self.gre: print(f"{'local address':<{20}}: {self.gre['local']}") print(f"{'remote address':<{20}}: {self.gre['remote']}") @@ -1045,6 +1126,10 @@ def pr_interface_list(json): iface.pr_vxlan() continue + if iface.is_wifi(): + iface.pr_wifi() + continue + if iface.is_vlan(): iface.pr_vlan(ifaces) continue @@ -1320,13 +1405,13 @@ def show_lldp(json): f"{'CHASSIS-ID':<{PadLldp.chassis_id}}" f"{'PORT-ID':<{PadLldp.port_id}}" ) - + print(Decore.invert(header)) for port_data in lldp_ports: port_name = port_data["name"] neighbors = port_data.get("remote-systems-data", []) - + for neighbor in neighbors: entry = LldpNeighbor(port_name, neighbor) entry.print() @@ -1394,6 +1479,7 @@ def main(): show_routing_table(json_data, args.ip) elif args.command == "show-software": show_software(json_data, args.name) + else: print(f"Error, unknown command '{args.command}'") sys.exit(1) diff --git a/src/statd/python/yanger/ietf_interfaces/link.py b/src/statd/python/yanger/ietf_interfaces/link.py index 5fff2a6f..fd7c5d8c 100644 --- a/src/statd/python/yanger/ietf_interfaces/link.py +++ b/src/statd/python/yanger/ietf_interfaces/link.py @@ -1,3 +1,4 @@ +from ..host import HOST from . import common from . import bridge @@ -7,6 +8,7 @@ from . import lag from . import tun from . import veth from . import vlan +from . import wifi def statistics(iplink): @@ -24,13 +26,16 @@ def statistics(iplink): def iplink2yang_type(iplink): + ifname=iplink["ifname"] match iplink["link_type"]: case "loopback": return "infix-if-type:loopback" case "gre"|"gre6": return "infix-if-type:gre" case "ether": - pass + data = HOST.run(tuple(f"ls /sys/class/net/{ifname}/wireless/".split()), default="no") + if data != "no": + return "infix-if-type:wifi" case _: return "infix-if-type:other" @@ -133,6 +138,9 @@ def interface(iplink, ipaddr): case "infix-if-type:vlan": if v := vlan.vlan(iplink): interface["infix-interfaces:vlan"] = v + case "infix-if-type:wifi": + if w := wifi.wifi(iplink["ifname"]): + interface["infix-interfaces:wifi"] = w match iplink2yang_lower(iplink): case "infix-interfaces:bridge-port": diff --git a/src/statd/python/yanger/ietf_interfaces/wifi.py b/src/statd/python/yanger/ietf_interfaces/wifi.py new file mode 100644 index 00000000..d373dd5c --- /dev/null +++ b/src/statd/python/yanger/ietf_interfaces/wifi.py @@ -0,0 +1,160 @@ +from ..host import HOST +import json +import re + +def wifi(ifname): + data=HOST.run(tuple(f"wpa_cli -i {ifname} status".split()), default="") + wifi_data={} + + if data != "": + for line in data.splitlines(): + k,v = line.split("=") + if k == "ssid": + wifi_data["ssid"] = v + if k == "wpa_state" and v == "DISCONNECTED": # wpa_suppicant has most likely restarted, restart scanning + HOST.run(tuple(f"wpa_cli -i {ifname} scan".split()), default="") + + data=HOST.run(tuple(f"wpa_cli -i {ifname} signal_poll".split()), default="FAIL") + + # signal_poll return FAIL not connected + if data.strip() != "FAIL": + for line in data.splitlines(): + k,v = line.strip().split("=") + if k == "RSSI": + wifi_data["rssi"]=int(v) + data=HOST.run(tuple(f"wpa_cli -i {ifname} scan_result".split()), default="FAIL") + + if data != "FAIL": + wifi_data["scan-results"] = parse_wpa_scan_result(data) + + return wifi_data + + +def parse_wpa_scan_result(scan_output): + networks = {} + lines = scan_output.strip().split('\n') + + # Skip header line and any empty lines + for line in lines: + line = line.strip() + if not line or 'bssid / frequency' in line.lower(): + continue + + # Split by tabs or multiple spaces + parts = re.split(r'\t+|\s{2,}', line) + + if len(parts) >= 5: + bssid = parts[0].strip() + frequency = int(parts[1].strip()) + rssi = int(parts[2].strip()) + flags = parts[3].strip() + ssid = parts[4].strip() if len(parts) > 4 else "" + + # Skip hidden SSIDs (empty or whitespace only) + if not ssid or ssid.isspace() or '\\x00' in ssid: + continue + + # Extract encryption information from flags + encryption = extract_encryption(flags) + + # Convert frequency to channel + channel = frequency_to_channel(frequency) + + # Keep only the network with best (highest) RSSI per SSID + if ssid not in networks or rssi < networks[ssid]['rssi']: + networks[ssid] = { + 'bssid': bssid, + 'ssid': ssid, + 'rssi': rssi, + 'encryption': encryption, + 'channel': channel + } + # Convert to list and sort by RSSI (best first) + result = list(networks.values()) + result.sort(key=lambda x: x['rssi'], reverse=False) + + return result + +def frequency_to_channel(frequency): + """Convert frequency (MHz) to WiFi channel number""" + freq = int(frequency) + + # 2.4 GHz band (channels 1-14) + if 2412 <= freq <= 2484: # Channel 14 is special + if freq == 2484: + return 14 + return (freq - 2412) // 5 + 1 + + # 5 GHz band (channels 36-165) + elif 5170 <= freq <= 5825: + return (freq - 5000) // 5 + + # 6 GHz band (channels 1-233) + elif 5955 <= freq <= 7115: + return (freq - 5950) // 5 + + else: + return f"Unknown ({freq} MHz)" + +def extract_encryption(flags): + """Extract detailed encryption information from flags string""" + flags = flags.upper() + encryption_info = { + 'protocols': [], + 'key_mgmt': [], + 'ciphers': [], + 'auth_type': 'Unknown' + } + + # Extract WPA protocols + if 'WPA3' in flags: + encryption_info['protocols'].append('WPA3') + if 'WPA2' in flags: + encryption_info['protocols'].append('WPA2') + if 'WPA-' in flags and 'WPA2' not in flags and 'WPA3' not in flags: + encryption_info['protocols'].append('WPA') + + # Extract key management methods + if 'PSK' in flags: + encryption_info['key_mgmt'].append('PSK') + encryption_info['auth_type'] = 'Personal' + if 'EAP' in flags: + encryption_info['key_mgmt'].append('EAP') + encryption_info['auth_type'] = 'Enterprise' + if 'SAE' in flags: # WPA3 Personal + encryption_info['key_mgmt'].append('SAE') + encryption_info['auth_type'] = 'Personal' + if 'OWE' in flags: # Enhanced Open (WPA3) + encryption_info['key_mgmt'].append('OWE') + encryption_info['auth_type'] = 'Enhanced Open' + if 'FT' in flags: + encryption_info['key_mgmt'].append('FT') + + # Extract cipher suites + if 'CCMP' in flags: + encryption_info['ciphers'].append('CCMP') + if 'TKIP' in flags: + encryption_info['ciphers'].append('TKIP') + if 'GCMP' in flags: + encryption_info['ciphers'].append('GCMP') + + # Handle special cases + if 'WEP' in flags: + return ['WEP'] + + if not encryption_info['protocols'] and 'ESS' in flags: + return ['Open'] + + # Return array of supported protocols with auth type + result = [] + for protocol in encryption_info['protocols']: + if encryption_info['auth_type'] == 'Enterprise': + result.append(f"{protocol}-Enterprise") + elif encryption_info['auth_type'] == 'Personal': + result.append(f"{protocol}-Personal") + elif encryption_info['auth_type'] == 'Enhanced Open': + result.append(f"{protocol}-Enhanced-Open") + else: + result.append(protocol) + + return result if result else ['Unknown']