From ca4e92c32d5f9b8b6e1144d37dfe86fc033c7999 Mon Sep 17 00:00:00 2001 From: Mauricio Camayo Date: Thu, 22 Oct 2026 12:09:53 -0500 Subject: [PATCH] Sync + WIFICANARY: attribute synced detections to their node Fixes found running WIFICANARY on a WiFi satellite node that pushes its detections to a hub with no WiFi of its own. Synced plugin objects were stored with syncHubNodeName = "decoded" instead of the node's name. plugin.py took filename.split(".")[2], but by the time the hub processes the file decode_and_rename_files() has renamed it to last_result..decoded...log, so index 2 is the literal marker. The parsing now lives in utils.plugin_utils.get_sync_hub_node_name(), next to the function that produces these names; it keeps node names that contain dots and returns "" for local result files. WIFICANARY's absent_baseline_clone was evaluated per trusted_aps entry, so with a main AP + range extender pair, a weak clone seen while only the main AP was in range also raised absent_baseline_clone through the extender's entry, on top of the correct evil_twin. The baseline now counts as present if any trusted radio for that SSID is visible; single-entry behaviour is unchanged. Leaving WIFICANARY_IFACE blank is how such a hub ingests the node's detections (core processes the synced files after the plugin runs), but it logged an unconditional "not set - nothing to scan" every cycle. It is now a verbose "hub-only mode" message, and the node + hub setup is documented in the setting description and the plugin README. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01PzCQWQVz9b2UzqLiGp2mPM --- server/plugin.py | 7 ++-- server/plugins/wificanary/README.md | 1 + server/plugins/wificanary/config.json | 2 +- server/plugins/wificanary/script.py | 14 +++++-- server/utils/plugin_utils.py | 11 +++++ test/plugins/test_wificanary.py | 30 ++++++++++++++ test/server/test_sync_hub_node_name.py | 56 ++++++++++++++++++++++++++ 7 files changed, 113 insertions(+), 8 deletions(-) create mode 100644 test/server/test_sync_hub_node_name.py diff --git a/server/plugin.py b/server/plugin.py index 62927b505..7397b2c22 100755 --- a/server/plugin.py +++ b/server/plugin.py @@ -23,6 +23,7 @@ from utils.plugin_utils import ( resolve_wildcards_arr, handle_empty, decode_and_rename_files, + get_sync_hub_node_name, primary_id_is_mac ) from models.notification_instance import NotificationInstance @@ -484,10 +485,8 @@ def execute_plugin(db, all_plugins, plugin): # cleanup - select only lines containing a separator to filter out unnecessary data newLines = list(filter(lambda x: "|" in x, newLines)) - # Store e.g. Node_1 from last_result..encoded.Node_1.1.log - tmp_SyncHubNodeName = "" - if len(filename.split(".")) > 3: - tmp_SyncHubNodeName = filename.split(".")[2] + # e.g. Node_1 from last_result..decoded.Node_1.1.log + tmp_SyncHubNodeName = get_sync_hub_node_name(filename) for line in newLines: columns = line.split("|") diff --git a/server/plugins/wificanary/README.md b/server/plugins/wificanary/README.md index e0113143d..78eec702d 100644 --- a/server/plugins/wificanary/README.md +++ b/server/plugins/wificanary/README.md @@ -15,6 +15,7 @@ Runs a periodic passive WiFi scan (`iw scan`, no monitor mode) and flags rogue A - Enable the plugin (`WIFICANARY_RUN` → `schedule`) and set a schedule in `WIFICANARY_RUN_SCHD`. - A detection creates a new, dangerous-by-default `Devices` entry for the rogue BSSID (even though it never associated with your network) - turn off `WIFICANARY_IMPORT_ON` if you'd rather tune your trusted-AP list against the plugin's history first, without devices being created yet. - Pwnagotchi and WiFi Pineapple signature checks run unconditionally, regardless of `WIFICANARY_trusted_aps`. +- No WiFi on your main NetAlertX host? Run this plugin on a small WiFi-capable node (e.g. a Raspberry Pi) and push its findings with the [Sync Hub](../sync/README.md) plugin (`SYNC_plugins` = `WIFICANARY`). On the hub, load `WIFICANARY` too with `WIFICANARY_RUN` → `schedule` and `WIFICANARY_IFACE` left **blank**: it skips the local scan (hub-only mode) but its run is what makes core ingest the detections synced from the node, tagged with the node's name. - If a detected rogue BSSID turns out to already be a device NetAlertX knows from another source (ARP, DHCP, an importer...), the finding is escalated in place - the reason is rewritten to name the known device, and the motor gets a `_known_device` suffix (e.g. `evil_twin_known_device`) so a [Workflow](https://docs.netalertx.com/WORKFLOWS) rule can route it to a more urgent channel than a stranger's radio. ### Notes diff --git a/server/plugins/wificanary/config.json b/server/plugins/wificanary/config.json index 4e51758e3..3bcf024cd 100644 --- a/server/plugins/wificanary/config.json +++ b/server/plugins/wificanary/config.json @@ -611,7 +611,7 @@ "description": [ { "language_code": "en_us", - "string": "The WiFi interface to scan with, e.g. wlan0. Station mode is enough - no monitor mode needed. Required." + "string": "The WiFi interface to scan with, e.g. wlan0. Station mode is enough - no monitor mode needed. Leave blank on a sync hub without WiFi to only process detections synced from nodes." } ] }, diff --git a/server/plugins/wificanary/script.py b/server/plugins/wificanary/script.py index 59cb9f357..a24587472 100644 --- a/server/plugins/wificanary/script.py +++ b/server/plugins/wificanary/script.py @@ -58,7 +58,10 @@ def main(): iface = get_setting_value('WIFICANARY_IFACE') if not iface: - mylog('none', [f'[{pluginName}] WIFICANARY_IFACE is not set - nothing to scan']) + # Hub-only mode (e.g. a sync hub with no WiFi hardware): no local scan, + # but the run itself still matters - core processes the results synced + # from WIFICANARY nodes after this script exits. + mylog('verbose', [f'[{pluginName}] WIFICANARY_IFACE not set - hub-only mode, skipping local scan']) plugin_objects.write_result_file() return 0 @@ -263,8 +266,13 @@ def check_trusted_aps(aps, trusted_aps): for trust in trusted_aps: matches = [ap for ap in aps if ap['ssid'] == trust['ssid']] - other_trusted_bssids = trusted_bssids_by_ssid.get(trust['ssid'], set()) - {trust['bssid']} - baseline_bssid_seen = any(ap['bssid'] == trust['bssid'] for ap in matches) if trust['bssid'] else True + ssid_trusted_bssids = trusted_bssids_by_ssid.get(trust['ssid'], set()) + other_trusted_bssids = ssid_trusted_bssids - {trust['bssid']} + # The baseline counts as present if ANY trusted radio for this SSID + # is visible, not just this entry's own: with a main AP + extender + # pair, an out-of-range extender while the main AP is up isn't an + # "absent baseline" - the clone is a plain evil twin. + baseline_bssid_seen = any(ap['bssid'] in ssid_trusted_bssids for ap in matches) if trust['bssid'] else True accepted = trust['security_set'] expected_desc = ' or '.join(sorted(accepted)) diff --git a/server/utils/plugin_utils.py b/server/utils/plugin_utils.py index f7dc53c09..60df25b15 100755 --- a/server/utils/plugin_utils.py +++ b/server/utils/plugin_utils.py @@ -346,6 +346,17 @@ def decode_and_rename_files(file_dir, file_prefix): return files_to_process +# ------------------------------------------------------------------- +def get_sync_hub_node_name(filename): + """Return the sync node name embedded in a received plugin result file name + (last_result.....log), or "" for a + local result file such as last_result..log.""" + parts = filename.split(".") + if len(parts) >= 6 and parts[2] in ("encoded", "decoded"): + return ".".join(parts[3:-2]) # node names may themselves contain dots + return "" + + # ------------------------------------------------------------------ # Retrieve the value for a plugin's setting, prioritizing user-defined values over defaults. def get_set_value_for_init(plugin, c_d, setting_key): diff --git a/test/plugins/test_wificanary.py b/test/plugins/test_wificanary.py index e47a4c517..dce6ea3d3 100644 --- a/test/plugins/test_wificanary.py +++ b/test/plugins/test_wificanary.py @@ -407,6 +407,34 @@ def test_two_trusted_bssids_same_ssid_no_false_positive(): assert wificanary.check_trusted_aps(aps, trusted) == [] +def test_out_of_range_extender_does_not_make_clone_absent_baseline(): + # Main AP + extender, each its own entry; only the main AP is in range. + # An open clone is a plain evil twin - the protected network is up, so + # the out-of-range extender must not also raise absent_baseline_clone. + trusted = [ + {'ssid': 'HomeWiFi', 'bssid': 'aa:bb:cc:11:22:33', 'security_set': {'wpa2', 'wpa3'}}, + {'ssid': 'HomeWiFi', 'bssid': '44:55:66:aa:bb:cc', 'security_set': {'wpa2', 'wpa3'}}, + ] + aps = [ + _ap('aa:bb:cc:11:22:33', 'HomeWiFi', 'wpa3'), + _ap('ff:ee:dd:99:88:77', 'HomeWiFi', 'open'), + ] + found = wificanary.check_trusted_aps(aps, trusted) + assert {d['motor'] for d in found} == {'evil_twin'} + + +def test_clone_with_every_trusted_radio_absent_is_absent_baseline(): + # Same pair, neither radio in range - now the clone really is standing + # in for a missing baseline. + trusted = [ + {'ssid': 'HomeWiFi', 'bssid': 'aa:bb:cc:11:22:33', 'security_set': {'wpa2', 'wpa3'}}, + {'ssid': 'HomeWiFi', 'bssid': '44:55:66:aa:bb:cc', 'security_set': {'wpa2', 'wpa3'}}, + ] + aps = [_ap('ff:ee:dd:99:88:77', 'HomeWiFi', 'open')] + found = wificanary.check_trusted_aps(aps, trusted) + assert {d['motor'] for d in found} == {'absent_baseline_clone'} + + def test_trusted_extender_with_stricter_main_ap_not_flagged_as_clone(): # Regression (CodeRabbit): a main AP entry requiring a *stronger* # accepted set (wpa3 only) than a separately-trusted extender (wpa2) @@ -679,6 +707,8 @@ def test_main_returns_early_without_iface(): wificanary.main() mock_scan.assert_not_called() wificanary.plugin_objects.add_object.assert_not_called() + # hub-only mode still writes an (empty) result file so the run completes normally + wificanary.plugin_objects.write_result_file.assert_called_once() def test_main_end_to_end_one_detection(): diff --git a/test/server/test_sync_hub_node_name.py b/test/server/test_sync_hub_node_name.py new file mode 100644 index 000000000..780463715 --- /dev/null +++ b/test/server/test_sync_hub_node_name.py @@ -0,0 +1,56 @@ +""" +Tests for utils.plugin_utils.get_sync_hub_node_name(). + +Regression coverage: server/plugin.py used filename.split(".")[2] to tag +plugin objects received from a sync node, but by the time the hub processes +them decode_and_rename_files() has renamed the file to +last_result..decoded...log - so index 2 is the literal +"decoded", and every synced plugin object (e.g. a WIFICANARY detection from a +WiFi satellite) was stored with syncHubNodeName = "decoded" instead of the +node's name. + +Run from inside the NetAlertX container (utils.plugin_utils imports conf, +logger and helper): + + pytest "test/server/test_sync_hub_node_name.py" -v +""" + +import os +import sys + +import pytest + +INSTALL_PATH = os.getenv("NETALERTX_APP", "/app") +sys.path.extend([f"{INSTALL_PATH}/server/plugins", f"{INSTALL_PATH}/server"]) + +from utils.plugin_utils import get_sync_hub_node_name # noqa: E402 + + +@pytest.mark.parametrize( + "filename, expected", + [ + # What the hub actually processes: already decrypted and renamed + ("last_result.WIFICANARY.decoded.satelite.6.log", "satelite"), + # Same file before decode_and_rename_files() touches it + ("last_result.WIFICANARY.encoded.satelite.6.log", "satelite"), + ("last_result.ARPSCAN.decoded.Node_1.1.log", "Node_1"), + # Node names containing dots must survive intact + ("last_result.ARPSCAN.decoded.sat.office.12.log", "sat.office"), + ], +) +def test_node_name_from_received_file(filename, expected): + """A received node file yields the node's name, never the encoded/decoded marker.""" + assert get_sync_hub_node_name(filename) == expected + + +@pytest.mark.parametrize( + "filename", + [ + "last_result.ARPSCAN.log", # local plugin result + "last_result.log", # sync's own progress file + "last_result.satelite.log", # PULL-mode device dump (handled by sync.py, not here) + ], +) +def test_local_result_file_has_no_node_name(filename): + """Local result files are not tagged with any sync node.""" + assert get_sync_hub_node_name(filename) == ""