#!/usr/bin/env python """ WIFICANARY - flags rogue APs from a periodic passive WiFi scan. Scope (see GitHub issue #1789): only the 6 heuristics that a plain `iw scan` snapshot can see are implemented here, plus the addendum's "known device turned rogue" escalation (cross-referencing a detection's own BSSID against NetAlertX's Devices table - not the deauth/probe/beacon source-MAC version of that idea, which still needs monitor-mode data this plugin doesn't have). Deauth/probe-flood/beacon-flood themselves need real monitor-mode frame capture (rate over time, not a point-in-time scan) and are out of scope for this plugin - see a dedicated monitor-mode tool (e.g. ESP32 WiFi Canary, https://github.com/simeononsecurity/esp32-wifi-canary) for those. """ import json import os import re import subprocess import sys from pytz import timezone INSTALL_PATH = os.getenv('NETALERTX_APP', '/app') sys.path.extend([f"{INSTALL_PATH}/server/plugins", f"{INSTALL_PATH}/server"]) from const import logPath # noqa: E402, E261 from plugin_helper import Plugin_Objects, normalize_mac, decode_settings_base64 # noqa: E402, E261 from logger import mylog, Logger # noqa: E402, E261 from helper import get_setting_value # noqa: E402, E261 from models.device_instance import DeviceInstance # noqa: E402, E261 import conf # noqa: E402, E261 conf.tz = timezone(get_setting_value('TIMEZONE')) Logger(get_setting_value('LOG_LEVEL')) pluginName = 'WIFICANARY' LOG_PATH = logPath + '/plugins' LOG_FILE = os.path.join(LOG_PATH, f'script.{pluginName}.log') RESULT_FILE = os.path.join(LOG_PATH, f'last_result.{pluginName}.log') plugin_objects = Plugin_Objects(RESULT_FILE) # Global signatures, independent of any trusted-AP baseline. PWNAGOTCHI_BSSID = 'de:ad:be:ef:de:ad' PINEAPPLE_OUI_MID = ('13', '37') # BSSID octets [1:3] == 13:37 # Weakest-to-strongest, used to detect a downgrade. SECURITY_RANK = {'open': 0, 'wep': 1, 'wpa': 2, 'wpa2': 3, 'wpa3': 4} def main(): """Scan once, compare against the configured trusted-AP baseline, and emit one CurrentScan row per anomaly found.""" mylog('verbose', [f'[{pluginName}] In script']) iface = get_setting_value('WIFICANARY_IFACE') if not iface: mylog('none', [f'[{pluginName}] WIFICANARY_IFACE is not set - nothing to scan']) plugin_objects.write_result_file() return 0 trusted_aps = get_trusted_aps() timeout = get_setting_value('WIFICANARY_RUN_TIMEOUT') or 60 aps = scan(iface, timeout) mylog('verbose', [f'[{pluginName}] Parsed {len(aps)} APs from scan on {iface}']) detections = [] detections += check_global_signatures(aps) detections += check_trusted_aps(aps, trusted_aps) detections += check_duplicate_ssid(aps, trusted_aps) escalate_known_devices(detections) for det in detections: plugin_objects.add_object( primaryId=det['bssid'], secondaryId=det['motor'], watched1=det['reason'], watched2=det['security'], watched3=det['signal'], watched4=det['oui'], extra=det['ssid'], foreignKey=det['bssid'], helpVal1=normalize_mac(det['bssid']), helpVal2='1', # scanCreatesDevice - every row here is an anomaly helpVal3='normal', # scanNotificationMode - these are meant to alert helpVal4='1', # scanPresence - detected in this scan cycle ) mylog('verbose', [f'[{pluginName}] {len(detections)} anomalies']) plugin_objects.write_result_file() return 0 def get_trusted_aps(): """Decode the WIFICANARY_trusted_aps nested setting into a list of dicts with ssid/bssid/security_set keys. security_set is the set of every encryption this network is allowed to legitimately use (e.g. a WPA2/WPA3 transition-mode AP would list both).""" raw_entries = get_setting_value('WIFICANARY_trusted_aps') or [] trusted = [] for raw in raw_entries: cfg = decode_settings_base64(raw) ssid = cfg.get('WIFICANARY_TRUSTED_SSID', '').strip() if not ssid: continue bssid = cfg.get('WIFICANARY_TRUSTED_BSSID', '').strip().lower() trusted.append({ 'ssid': ssid, 'bssid': normalize_mac(bssid) if bssid else '', 'security_set': parse_security_set(cfg.get('WIFICANARY_TRUSTED_SECURITY')), }) return trusted def parse_security_set(raw_value): """WIFICANARY_TRUSTED_SECURITY is a multi-select `array` setting - the frontend sends its value as a JSON-encoded list string (e.g. '["wpa2","wpa3"]'), not a real list, since it travels through the popupForm's generic decode_settings_base64() path rather than the top-level array-setting one. Falls back to {'wpa2'} for a blank/missing/ malformed value, matching config.json's own default_value.""" if not raw_value: return {'wpa2'} try: values = json.loads(raw_value) if isinstance(raw_value, str) else raw_value except (TypeError, ValueError): return {'wpa2'} security_set = {str(v).strip().lower() for v in values if str(v).strip()} return security_set or {'wpa2'} def scan(iface, timeout): """Run `iw dev scan` and parse the output into a list of AP dicts (bssid/ssid/security/signal/oui).""" cmd = ['sudo', 'iw', 'dev', iface, 'scan'] try: result = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout) except subprocess.TimeoutExpired: mylog('none', [f'[{pluginName}] scan on {iface} timed out after {timeout}s']) return [] except FileNotFoundError: mylog('none', [f'[{pluginName}] `iw` not found - is it installed on this host?']) return [] if result.returncode != 0: mylog('none', [f'[{pluginName}] scan on {iface} failed: {result.stderr.strip()}']) return [] return parse_iw_scan(result.stdout) def parse_iw_scan(output): """Parse `iw scan` text output into a list of AP dicts.""" aps = [] current = None for line in output.splitlines(): bss_match = re.match(r'^BSS ([0-9a-fA-F:]{17})', line) if bss_match: if current and current.get('ssid'): aps.append(current) current = { 'bssid': bss_match.group(1).lower(), 'ssid': '', 'security': 'open', 'signal': '', } continue if current is None: continue stripped = line.strip() m = re.match(r'^SSID:\s?(.*)$', stripped) if m: current['ssid'] = m.group(1) continue if stripped.startswith('capability:') and 'Privacy' in stripped: if current['security'] == 'open': # Privacy bit set but no RSN/WPA IE found below -> most likely WEP # (or a TKIP-only WPA1 network with no separate IE, rare in practice). current['security'] = 'wep' if stripped.startswith('RSN:'): current['security'] = 'wpa2' continue if stripped.startswith('WPA:'): if current['security'] not in ('wpa2', 'wpa3'): current['security'] = 'wpa' continue if 'Authentication suites' in stripped and 'SAE' in stripped: current['security'] = 'wpa3' continue m = re.match(r'^signal:\s*(-?\d+(?:\.\d+)?)\s*dBm', stripped) if m: current['signal'] = m.group(1) if current and current.get('ssid'): aps.append(current) for ap in aps: parts = ap['bssid'].split(':') ap['oui'] = ':'.join(parts[0:3]) if len(parts) >= 3 else '' return aps def check_global_signatures(aps): """Motors 1-2: absolute signatures that don't depend on any baseline - a known pwnagotchi BSSID, or a WiFi Pineapple's default OUI pattern.""" found = [] for ap in aps: parts = ap['bssid'].split(':') if ap['bssid'] == PWNAGOTCHI_BSSID: found.append(make_detection(ap, 'pwnagotchi_nearby', f"Pwnagotchi signature BSSID seen ({ap['bssid']})")) if len(parts) >= 3 and (parts[1], parts[2]) == PINEAPPLE_OUI_MID: found.append(make_detection(ap, 'pineapple_oui', f"WiFi Pineapple default OUI pattern on BSSID {ap['bssid']}")) return found def is_downgrade(observed_security, accepted): """True if `observed_security` is neither explicitly accepted nor at least as strong as the strongest accepted value - the shared threshold check_trusted_aps() uses for both a known radio weakening over time and a different radio cloning the SSID with lesser security.""" if observed_security in accepted: return False observed_rank = SECURITY_RANK.get(observed_security, 0) strongest_accepted_rank = max(SECURITY_RANK.get(s, 0) for s in accepted) return observed_rank < strongest_accepted_rank def check_trusted_aps(aps, trusted_aps): """Motors 3-5: evil twin / weaker-security clone, baseline AP absent while a clone is present, and security downgrade - all evaluated against the user's trusted-AP baseline (WIFICANARY_trusted_aps).""" found = [] for trust in trusted_aps: matches = [ap for ap in aps if ap['ssid'] == trust['ssid']] baseline_bssid_seen = any(ap['bssid'] == trust['bssid'] for ap in matches) if trust['bssid'] else True accepted = trust['security_set'] expected_desc = ' or '.join(sorted(accepted)) for ap in matches: if not is_downgrade(ap['security'], accepted): continue same_radio = trust['bssid'] and ap['bssid'] == trust['bssid'] if same_radio or not trust['bssid']: # The radio we already trust for this SSID (or, with no # BSSID configured, the only radio we have to go on). found.append(make_detection(ap, 'security_downgrade', f"'{trust['ssid']}' now broadcasting {ap['security']}, " f"expected {expected_desc}")) continue # A different BSSID broadcasting the same protected SSID with # weaker-than-accepted security. A same- or stronger-encrypted # different radio is check_duplicate_ssid's job instead, via # OUI mismatch, not this one's. if trust['bssid'] and not baseline_bssid_seen: found.append(make_detection(ap, 'absent_baseline_clone', f"'{trust['ssid']}' baseline AP ({trust['bssid']}) missing, " f"weaker clone ({ap['security']}) seen on {ap['bssid']}")) else: found.append(make_detection(ap, 'evil_twin', f"'{trust['ssid']}' cloned with weaker security ({ap['security']}) " f"by {ap['bssid']} (expected {expected_desc})")) return found def check_duplicate_ssid(aps, trusted_aps): """Motor 6: a trusted SSID broadcast by more than one OUI at once - a plausible impostor sharing a protected network's name. Restricted to SSIDs the user has explicitly claimed via the trusted-AP list, so an untracked network's own AP diversity (e.g. a cafe chain) never triggers this. A real multi-radio setup for the *same* trusted SSID (a range extender, a mesh kit - often a different OUI than the main AP) is expected to be listed as its own WIFICANARY_trusted_aps entry (same SSID, its own BSSID) - every trusted BSSID's OUI for a given SSID is whitelisted, not just one.""" trusted_ssids = {t['ssid'] for t in trusted_aps} trusted_ouis_by_ssid = {} for t in trusted_aps: if t['bssid']: trusted_ouis_by_ssid.setdefault(t['ssid'], set()).add(':'.join(t['bssid'].split(':')[:3])) found = [] by_ssid = {} for ap in aps: if ap['ssid'] in trusted_ssids: by_ssid.setdefault(ap['ssid'], []).append(ap) for ssid, group in by_ssid.items(): ouis = {ap['oui'] for ap in group} if len(ouis) < 2: continue trusted_ouis = trusted_ouis_by_ssid.get(ssid) if trusted_ouis: # One or more explicit trusted BSSIDs exist for this SSID - # their OUIs are the whitelist. Anything else sharing the SSID # is suspect regardless of how common it is in this scan. for ap in group: if ap['oui'] not in trusted_ouis: found.append(make_detection(ap, 'duplicate_ssid_diff_vendor', f"'{ssid}' also seen from OUI {ap['oui']} on {ap['bssid']} " f"(trusted OUIs for this SSID are {', '.join(sorted(trusted_ouis))})")) continue # No trusted BSSID configured for this SSID (wildcard-only entry) - # fall back to majority OUI as the presumed "expected" one. primary_oui = max(ouis, key=lambda o: sum(1 for ap in group if ap['oui'] == o)) for ap in group: if ap['oui'] != primary_oui: found.append(make_detection(ap, 'duplicate_ssid_diff_vendor', f"'{ssid}' also seen from OUI {ap['oui']} on {ap['bssid']} " f"(other APs for this SSID are {primary_oui})")) return found def escalate_known_devices(detections): """Motor 10 (see the addendum on issue #1789): a BSSID this plugin just flagged might not be a stranger's radio at all - it might be a device NetAlertX already knows and trusts, now behaving like an attacker (compromised firmware, a misconfigured AP mode, etc). That's a much more urgent signal than "unknown pineapple nearby", so it's called out separately - mutates each matching detection's motor/reason in place rather than returning a new list. One DeviceInstance().getAllByMacs() call for every distinct BSSID in this run, not one getByMac() per detection - a run can easily produce several detections (multiple motors firing on the same BSSID, or several rogue APs at once), and each would otherwise be its own DB round-trip. Only escalates when the existing Devices row was NOT itself created by a previous WIFICANARY run - otherwise every anomaly would trivially "escalate" against its own prior detection from run 2 onward.""" bssids = [det['bssid'] for det in detections] known_by_mac = DeviceInstance().getAllByMacs(bssids) for det in detections: existing = known_by_mac.get(det['bssid'].lower()) if not existing or (existing.get('devSourcePlugin') or '') == 'WIFICANARY': continue device_label = existing.get('devName') or det['bssid'] det['motor'] = f"{det['motor']}_known_device" det['reason'] = f"Known device '{device_label}' now behaving like a rogue AP: {det['reason']}" def make_detection(ap, motor, reason): """Build the dict consumed by main()'s add_object() call for one AP anomaly.""" return { 'bssid': ap['bssid'], 'ssid': ap['ssid'] or 'null', 'motor': motor, 'reason': reason, 'security': ap['security'], 'signal': ap['signal'] or 'null', 'oui': ap['oui'] or 'null', } if __name__ == '__main__': main()