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.<PREFIX>.decoded.<NodeName>.<N>.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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01PzCQWQVz9b2UzqLiGp2mPM
This commit is contained in:
Mauricio CamayoandClaude Opus 5.5 committed 2026-10-22 12:09:53 -05:00
1 parent 32e2ddd322
commit ca4e92c32d
7 files changed
+113 -8

No files matched your search

+3 -4
View File
@@ -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.<prefix>.encoded.Node_1.1.log
tmp_SyncHubNodeName = ""
if len(filename.split(".")) > 3:
tmp_SyncHubNodeName = filename.split(".")[2]
# e.g. Node_1 from last_result.<prefix>.decoded.Node_1.1.log
tmp_SyncHubNodeName = get_sync_hub_node_name(filename)
for line in newLines:
columns = line.split("|")
+1
View File
@@ -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
+1 -1
View File
@@ -611,7 +611,7 @@
"description": [
{
"language_code": "en_us",
"string": "The WiFi interface to scan with, e.g. <code>wlan0</code>. Station mode is enough - no monitor mode needed. Required."
"string": "The WiFi interface to scan with, e.g. <code>wlan0</code>. Station mode is enough - no monitor mode needed. Leave blank on a sync hub without WiFi to only process detections synced from nodes."
}
]
},
+11 -3
View File
@@ -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))
+11
View File
@@ -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.<PREFIX>.<encoded|decoded>.<NodeName>.<N>.log), or "" for a
local result file such as last_result.<PREFIX>.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):
+30
View File
@@ -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():
+56
View File
@@ -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.<PREFIX>.decoded.<NodeName>.<N>.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) == ""