mirror of
https://github.com/jokob-sk/NetAlertX.git
synced 2026-11-08 00:13:25 -05:00
Merge branch 'next_release' of github.com:netalertx/NetAlertX into next_release
This commit is contained in:
commit
129ac8133b
9 files changed
+319
-33
No files matched your search
@@ -61,4 +61,4 @@ A worked example: a WiFi-scanning plugin PR added `iw` plus its `setcap` grant t
|
||||
|
||||
`scanPresence` defaults to `1` (`server/db/schema/app.sql`) when a plugin's `config.json` never maps it - every row that plugin reports asserts "currently online." That's correct for a plugin that only reports a device when it actually observed it *this cycle* (arp_scan, icmp_scan). It's wrong for a plugin whose data source is a **roster of known/historical entries** the upstream service remembers regardless of current connectivity (a client list built from DNS query history, a full device inventory export, anything that persists once discovered rather than expiring when the device goes quiet). Such a plugin re-reports the same entries every run whether or not the device is actually reachable, so the default `1` silently asserts permanent presence - read the actual data source (the API/file the script pulls from), not just the plugin's name, to tell which shape it is. `docs/PLUGINS_IMPORT_BEHAVIOR.md` already documents the fix (`scanPresence = 0` for "a reservation, a lease record, or a static IPAM entry"); this check exists because that guidance wasn't being verified against actual plugin behavior at review time.
|
||||
|
||||
If the plugin's own data does contain *some* real recency signal (an active DHCP lease, a last-seen timestamp within a threshold), prefer computing `scanPresence` per-row from that over a blanket static value - see `pihole_api_scan`'s `lastSeen`-vs-threshold check and `adguard_import`'s active-dynamic-lease check (`server/plugins/adguard_import/adguard_import.py`) for two different real implementations of the same idea. A `scanPresence = 0` row is an *abstention*, not a vote for offline - `update_presence_from_CurrentScan()`/`current_scan_presence_condition()` (`server/scan/presence.py`) only ever require *some* row asserting `1` to mark a device present, so a plugin correctly reporting "no evidence of presence" never suppresses a real scanner's positive signal for the same device. When writing the comment explaining this on a plugin's own `scanPresence` logic, state the plugin's own contract ("this value means no active lease was found, not that the device is offline") rather than asserting how the core presence-aggregation code behaves internally - that couples a leaf plugin file to framework internals it doesn't own and won't be updated if that logic ever changes.
|
||||
If the plugin's own data does contain *some* real recency signal (an active DHCP lease, a last-activity timestamp within a threshold), prefer computing `scanPresence` per-row from that over a blanket static value - see `pihole_api_scan`'s `lastQuery`-vs-threshold check (`server/plugins/pihole_api_scan/pihole_api_scan.py`) and `adguard_import`'s active-dynamic-lease check (`server/plugins/adguard_import/adguard_import.py`) for two different real implementations of the same idea. A timestamp the upstream keeps refreshing for every client it merely remembers is not a recency signal: Pi-hole FTL refreshes `ips[].lastSeen` for every client it holds in memory, so `pihole_api_scan` reads the device's `lastQuery` (time of its last DNS query). A `scanPresence = 0` row is an *abstention*, not a vote for offline - `update_presence_from_CurrentScan()`/`current_scan_presence_condition()` (`server/scan/presence.py`) only ever require *some* row asserting `1` to mark a device present, so a plugin correctly reporting "no evidence of presence" never suppresses a real scanner's positive signal for the same device. When writing the comment explaining this on a plugin's own `scanPresence` logic, state the plugin's own contract ("this value means no active lease was found, not that the device is offline") rather than asserting how the core presence-aggregation code behaves internally - that couples a leaf plugin file to framework internals it doesn't own and won't be updated if that logic ever changes.
|
||||
@@ -61,4 +61,4 @@ A worked example: a WiFi-scanning plugin PR added `iw` plus its `setcap` grant t
|
||||
|
||||
`scanPresence` defaults to `1` (`server/db/schema/app.sql`) when a plugin's `config.json` never maps it - every row that plugin reports asserts "currently online." That's correct for a plugin that only reports a device when it actually observed it *this cycle* (arp_scan, icmp_scan). It's wrong for a plugin whose data source is a **roster of known/historical entries** the upstream service remembers regardless of current connectivity (a client list built from DNS query history, a full device inventory export, anything that persists once discovered rather than expiring when the device goes quiet). Such a plugin re-reports the same entries every run whether or not the device is actually reachable, so the default `1` silently asserts permanent presence - read the actual data source (the API/file the script pulls from), not just the plugin's name, to tell which shape it is. `docs/PLUGINS_IMPORT_BEHAVIOR.md` already documents the fix (`scanPresence = 0` for "a reservation, a lease record, or a static IPAM entry"); this check exists because that guidance wasn't being verified against actual plugin behavior at review time.
|
||||
|
||||
If the plugin's own data does contain *some* real recency signal (an active DHCP lease, a last-seen timestamp within a threshold), prefer computing `scanPresence` per-row from that over a blanket static value - see `pihole_api_scan`'s `lastSeen`-vs-threshold check and `adguard_import`'s active-dynamic-lease check (`server/plugins/adguard_import/adguard_import.py`) for two different real implementations of the same idea. A `scanPresence = 0` row is an *abstention*, not a vote for offline - `update_presence_from_CurrentScan()`/`current_scan_presence_condition()` (`server/scan/presence.py`) only ever require *some* row asserting `1` to mark a device present, so a plugin correctly reporting "no evidence of presence" never suppresses a real scanner's positive signal for the same device. When writing the comment explaining this on a plugin's own `scanPresence` logic, state the plugin's own contract ("this value means no active lease was found, not that the device is offline") rather than asserting how the core presence-aggregation code behaves internally - that couples a leaf plugin file to framework internals it doesn't own and won't be updated if that logic ever changes.
|
||||
If the plugin's own data does contain *some* real recency signal (an active DHCP lease, a last-activity timestamp within a threshold), prefer computing `scanPresence` per-row from that over a blanket static value - see `pihole_api_scan`'s `lastQuery`-vs-threshold check (`server/plugins/pihole_api_scan/pihole_api_scan.py`) and `adguard_import`'s active-dynamic-lease check (`server/plugins/adguard_import/adguard_import.py`) for two different real implementations of the same idea. A timestamp the upstream keeps refreshing for every client it merely remembers is not a recency signal: Pi-hole FTL refreshes `ips[].lastSeen` for every client it holds in memory, so `pihole_api_scan` reads the device's `lastQuery` (time of its last DNS query). A `scanPresence = 0` row is an *abstention*, not a vote for offline - `update_presence_from_CurrentScan()`/`current_scan_presence_condition()` (`server/scan/presence.py`) only ever require *some* row asserting `1` to mark a device present, so a plugin correctly reporting "no evidence of presence" never suppresses a real scanner's positive signal for the same device. When writing the comment explaining this on a plugin's own `scanPresence` logic, state the plugin's own contract ("this value means no active lease was found, not that the device is offline") rather than asserting how the core presence-aggregation code behaves internally - that couples a leaf plugin file to framework internals it doesn't own and won't be updated if that logic ever changes.
|
||||
@@ -61,4 +61,4 @@ A worked example: a WiFi-scanning plugin PR added `iw` plus its `setcap` grant t
|
||||
|
||||
`scanPresence` defaults to `1` (`server/db/schema/app.sql`) when a plugin's `config.json` never maps it - every row that plugin reports asserts "currently online." That's correct for a plugin that only reports a device when it actually observed it *this cycle* (arp_scan, icmp_scan). It's wrong for a plugin whose data source is a **roster of known/historical entries** the upstream service remembers regardless of current connectivity (a client list built from DNS query history, a full device inventory export, anything that persists once discovered rather than expiring when the device goes quiet). Such a plugin re-reports the same entries every run whether or not the device is actually reachable, so the default `1` silently asserts permanent presence - read the actual data source (the API/file the script pulls from), not just the plugin's name, to tell which shape it is. `docs/PLUGINS_IMPORT_BEHAVIOR.md` already documents the fix (`scanPresence = 0` for "a reservation, a lease record, or a static IPAM entry"); this check exists because that guidance wasn't being verified against actual plugin behavior at review time.
|
||||
|
||||
If the plugin's own data does contain *some* real recency signal (an active DHCP lease, a last-seen timestamp within a threshold), prefer computing `scanPresence` per-row from that over a blanket static value - see `pihole_api_scan`'s `lastSeen`-vs-threshold check and `adguard_import`'s active-dynamic-lease check (`server/plugins/adguard_import/adguard_import.py`) for two different real implementations of the same idea. A `scanPresence = 0` row is an *abstention*, not a vote for offline - `update_presence_from_CurrentScan()`/`current_scan_presence_condition()` (`server/scan/presence.py`) only ever require *some* row asserting `1` to mark a device present, so a plugin correctly reporting "no evidence of presence" never suppresses a real scanner's positive signal for the same device. When writing the comment explaining this on a plugin's own `scanPresence` logic, state the plugin's own contract ("this value means no active lease was found, not that the device is offline") rather than asserting how the core presence-aggregation code behaves internally - that couples a leaf plugin file to framework internals it doesn't own and won't be updated if that logic ever changes.
|
||||
If the plugin's own data does contain *some* real recency signal (an active DHCP lease, a last-activity timestamp within a threshold), prefer computing `scanPresence` per-row from that over a blanket static value - see `pihole_api_scan`'s `lastQuery`-vs-threshold check (`server/plugins/pihole_api_scan/pihole_api_scan.py`) and `adguard_import`'s active-dynamic-lease check (`server/plugins/adguard_import/adguard_import.py`) for two different real implementations of the same idea. A timestamp the upstream keeps refreshing for every client it merely remembers is not a recency signal: Pi-hole FTL refreshes `ips[].lastSeen` for every client it holds in memory, so `pihole_api_scan` reads the device's `lastQuery` (time of its last DNS query). A `scanPresence = 0` row is an *abstention*, not a vote for offline - `update_presence_from_CurrentScan()`/`current_scan_presence_condition()` (`server/scan/presence.py`) only ever require *some* row asserting `1` to mark a device present, so a plugin correctly reporting "no evidence of presence" never suppresses a real scanner's positive signal for the same device. When writing the comment explaining this on a plugin's own `scanPresence` logic, state the plugin's own contract ("this value means no active lease was found, not that the device is offline") rather than asserting how the core presence-aggregation code behaves internally - that couples a leaf plugin file to framework internals it doesn't own and won't be updated if that logic ever changes.
|
||||
@@ -9,7 +9,7 @@ The plugin connects to your Pi-hole’s API and retrieves:
|
||||
* IP addresses
|
||||
* Hostnames (if available)
|
||||
* Vendor info
|
||||
* Last-seen timestamps
|
||||
* Time of the last DNS query
|
||||
|
||||
NetAlertX then uses this information to match or create devices in your system.
|
||||
|
||||
@@ -139,7 +139,9 @@ If names are missing, confirm they appear in Pi-hole’s own UI first.
|
||||
|
||||
### Notes
|
||||
|
||||
- Additional notes, limitations, Author info.
|
||||
- **Online status comes from DNS activity.** A device counts as online when its last DNS query to Pi-hole is at most `PIHOLEAPI_CONSIDER_ONLINE` seconds old. Pi-hole's per-IP "last seen" value is not used, because Pi-hole keeps refreshing it for as long as it holds the client in memory (about 24 hours in practice), also after the device has left.
|
||||
- Devices that send a DNS query only every few hours, such as smart plugs, powerline adapters or speakers, can show as offline although they are on. Run a presence scanner such as [ARPSCAN](https://docs.netalertx.com/plugins/arp_scan) on the same schedule. If PIHOLEAPI is your only presence source, raise `PIHOLEAPI_CONSIDER_ONLINE` instead (for example to `3600`). A device that has left the network then also counts as online for that long.
|
||||
- With `PIHOLEAPI_GET_OFFLINE` enabled, offline devices are still imported (name, vendor, IP), but they are not marked as online.
|
||||
|
||||
- Version: 1.0.0
|
||||
- Author: `jokob-sk`, `leiweibau`
|
||||
|
||||
@@ -259,7 +259,7 @@
|
||||
"description": [
|
||||
{
|
||||
"language_code": "en_us",
|
||||
"string": "Import offline devices. By default only online devices are retrieved."
|
||||
"string": "Also import offline devices, without marking them online. By default only online devices are imported."
|
||||
}
|
||||
]
|
||||
},
|
||||
@@ -281,13 +281,13 @@
|
||||
"name": [
|
||||
{
|
||||
"language_code": "en_us",
|
||||
"string": "Online if last seen"
|
||||
"string": "Online if last query"
|
||||
}
|
||||
],
|
||||
"description": [
|
||||
{
|
||||
"language_code": "en_us",
|
||||
"string": "Time in seconds to consider a device online."
|
||||
"string": "A device is online if its last DNS query is at most this many seconds old."
|
||||
}
|
||||
]
|
||||
},
|
||||
@@ -549,6 +549,22 @@
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
"column": "helpVal4",
|
||||
"mapped_to_column": "scanPresence",
|
||||
"css_classes": "col-sm-2",
|
||||
"show": false,
|
||||
"type": "none",
|
||||
"default_value": "",
|
||||
"options": [],
|
||||
"localized": ["name"],
|
||||
"name": [
|
||||
{
|
||||
"language_code": "en_us",
|
||||
"string": "N/A"
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
"column": "Dummy",
|
||||
"mapped_to_column": "scanSourcePlugin",
|
||||
|
||||
@@ -43,6 +43,7 @@ PIHOLEAPI_SES_CSRF = None
|
||||
PIHOLEAPI_API_MAXCLIENTS = None
|
||||
PIHOLEAPI_VERIFY_SSL = True
|
||||
PIHOLEAPI_GET_OFFLINE = False
|
||||
PIHOLEAPI_CONSIDER_ONLINE = 300
|
||||
PIHOLEAPI_RUN_TIMEOUT = 10
|
||||
PIHOLEAPI_FAKE_MAC = get_setting_value('PIHOLEAPI_FAKE_MAC')
|
||||
VERSION_DATE = "NAX-PIHOLEAPI-1.0"
|
||||
@@ -198,11 +199,17 @@ def get_pihole_network_devices():
|
||||
return data.get('devices', [])
|
||||
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
def is_device_online(last_query, now_ts, consider_online):
|
||||
"""Return True if last_query (Unix seconds of the device's last DNS query, 0 = never) is at most consider_online seconds old."""
|
||||
return last_query > 0 and (now_ts - last_query) <= consider_online
|
||||
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
def gather_device_entries():
|
||||
"""
|
||||
Build a list of device entries.
|
||||
Online status is determined by comparing lastSeen (in seconds) vs the current time.
|
||||
Build a list of device entries, one per device IP.
|
||||
A device is online if its last DNS query is at most PIHOLEAPI_CONSIDER_ONLINE seconds old.
|
||||
"""
|
||||
entries = []
|
||||
devices = get_pihole_network_devices()
|
||||
@@ -219,29 +226,22 @@ def gather_device_entries():
|
||||
if not device_ips:
|
||||
continue
|
||||
|
||||
# 1. Find the freshest timestamp across all IPs for this MAC
|
||||
# This ensures if the device is active on ANY IP, the MAC is considered online.
|
||||
max_last_seen = 0
|
||||
for ip_info in device_ips:
|
||||
ls = ip_info.get('lastSeen', 0)
|
||||
if ls > max_last_seen:
|
||||
max_last_seen = ls
|
||||
# 1. Determine online status from the device's last DNS query. ips[].lastSeen is not
|
||||
# used: FTL refreshes it for every client it still holds in memory, even when the
|
||||
# client is not in the ARP cache.
|
||||
last_query = device.get('lastQuery', 0)
|
||||
is_online = is_device_online(last_query, now_ts, PIHOLEAPI_CONSIDER_ONLINE)
|
||||
|
||||
# 2. Determine online status: (Current Time - Last Seen) <= PIHOLEAPI_CONSIDER_ONLINE
|
||||
# Math is in seconds.
|
||||
if (now_ts - max_last_seen) <= PIHOLEAPI_CONSIDER_ONLINE:
|
||||
is_online = True
|
||||
else:
|
||||
is_online = False
|
||||
|
||||
# 3. Skip if offline (and user doesn't want offline devices)
|
||||
# 2. Skip if offline (and user doesn't want offline devices)
|
||||
if not is_online and not PIHOLEAPI_GET_OFFLINE:
|
||||
mylog('verbose', [f'[{pluginName}] Not online in the last {PIHOLEAPI_CONSIDER_ONLINE}s, import of offline disabled (PIHOLEAPI_GET_OFFLINE) skipping device: {device}.'])
|
||||
query_age = f'{now_ts - last_query}s ago' if last_query > 0 else 'never'
|
||||
mylog('verbose', [f'[{pluginName}] Skipping offline device {hwaddr}: last DNS query {query_age}, '
|
||||
f'PIHOLEAPI_CONSIDER_ONLINE={PIHOLEAPI_CONSIDER_ONLINE}s, PIHOLEAPI_GET_OFFLINE disabled.'])
|
||||
continue
|
||||
|
||||
mac_vendor = device.get('macVendor', '')
|
||||
|
||||
# 4. Process each valid IP for the device
|
||||
# 3. Process each valid IP for the device
|
||||
for ip_info in device_ips:
|
||||
ip = ip_info.get('ip')
|
||||
|
||||
@@ -262,8 +262,9 @@ def gather_device_entries():
|
||||
'ip': ip,
|
||||
'name': name,
|
||||
'macVendor': mac_vendor,
|
||||
# Pass the Unix timestamp as a string for NAX tracking
|
||||
'lastQuery': str(max_last_seen) if max_last_seen > 0 else ""
|
||||
# Unix timestamp of the device's last DNS query, as a string for NAX tracking
|
||||
'lastQuery': str(last_query) if last_query > 0 else "",
|
||||
'isOnline': is_online
|
||||
})
|
||||
|
||||
return entries
|
||||
@@ -320,7 +321,10 @@ def main():
|
||||
watched3=str(entry['lastQuery']),
|
||||
watched4="",
|
||||
extra=pluginName,
|
||||
foreignKey=str(entry['mac'])
|
||||
foreignKey=str(entry['mac']),
|
||||
# Mapped to CurrentScan.scanPresence: an offline device imported via
|
||||
# PIHOLEAPI_GET_OFFLINE must not be marked as present
|
||||
helpVal4='1' if entry['isOnline'] else '0'
|
||||
)
|
||||
else:
|
||||
mylog('verbose', [f"[{pluginName}] Skipping invalid MAC (see PIHOLEAPI_FAKE_MAC setting): {entry['name']}|{entry['mac']}|{entry['ip']}"])
|
||||
|
||||
@@ -110,7 +110,7 @@ CREATE_CURRENT_SCAN = """
|
||||
scanType TEXT,
|
||||
scanCreatesDevice INTEGER NOT NULL DEFAULT 1,
|
||||
scanNotificationMode TEXT NOT NULL DEFAULT 'normal',
|
||||
scanPresence INTEGER NOT NULL DEFAULT 1
|
||||
scanPresence INTEGER NOT NULL DEFAULT 1 CHECK (scanPresence IN (0, 1))
|
||||
)
|
||||
"""
|
||||
|
||||
@@ -664,7 +664,7 @@ def make_plugin_event_row(prefix: str, primary_id: str, secondary_id="sec",
|
||||
watched1="val1", watched2="", watched3="",
|
||||
watched4="", changed="2026-01-01 00:00:00",
|
||||
extra="", user_data="", foreign_key="",
|
||||
status="not-processed", help_val1=None):
|
||||
status="not-processed", help_val1=None, help_val4=None):
|
||||
"""Build a tuple mimicking a raw plugin output row (19 columns + index)."""
|
||||
return (
|
||||
0, # index (placeholder, not used for events)
|
||||
@@ -685,7 +685,7 @@ def make_plugin_event_row(prefix: str, primary_id: str, secondary_id="sec",
|
||||
help_val1,
|
||||
None, # helpVal2
|
||||
None, # helpVal3
|
||||
None, # helpVal4
|
||||
help_val4,
|
||||
)
|
||||
|
||||
|
||||
|
||||
@@ -0,0 +1,216 @@
|
||||
"""Tests for the pihole_api_scan (PIHOLEAPI) plugin's online detection.
|
||||
|
||||
pihole_api_scan.py is loaded with its NetAlertX-internal dependencies stubbed
|
||||
out (same approach as test_adguard_import.py), so these tests run without a
|
||||
Pi-hole instance. get_pihole_network_devices() is patched per test.
|
||||
|
||||
Pi-hole FTL refreshes ips[].lastSeen for every client it still holds in
|
||||
memory, even when the client is not in the ARP cache, so lastSeen stays fresh
|
||||
for about 24 hours after a device left. The device's lastQuery (time of its
|
||||
last DNS query) is what tells the plugin whether a device is still active.
|
||||
"""
|
||||
|
||||
import datetime
|
||||
import importlib.util
|
||||
import os
|
||||
import re
|
||||
import sys
|
||||
import types
|
||||
from pathlib import Path
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
INSTALL_PATH = os.getenv("NETALERTX_APP", "/app")
|
||||
sys.path.extend([f"{INSTALL_PATH}/server/plugins", f"{INSTALL_PATH}/server"])
|
||||
|
||||
# Unix time of the debug run the bug report is based on (2026-10-31 20:05:10 UTC).
|
||||
NOW_TS = 1793477110
|
||||
|
||||
_MAC_PATTERN = re.compile(r"^([0-9a-f]{2}:){5}[0-9a-f]{2}$")
|
||||
|
||||
|
||||
def _fixed_now(as_string=True):
|
||||
"""Stand-in for timeNowUTC() that always returns NOW_TS."""
|
||||
now = datetime.datetime.fromtimestamp(NOW_TS, tz=datetime.timezone.utc)
|
||||
return now.strftime("%Y-%m-%d %H:%M:%S") if as_string else now
|
||||
|
||||
|
||||
def _load_pihole_api_scan_module():
|
||||
"""Import pihole_api_scan.py with NetAlertX modules stubbed, then restore sys.modules."""
|
||||
missing_module = object()
|
||||
previous_modules = {}
|
||||
|
||||
def stub(name, **attributes):
|
||||
"""Put a fake module with the given attributes into sys.modules, remembering the previous entry."""
|
||||
previous_modules[name] = sys.modules.get(name, missing_module)
|
||||
module = types.ModuleType(name)
|
||||
for attribute, value in attributes.items():
|
||||
setattr(module, attribute, value)
|
||||
sys.modules[name] = module
|
||||
|
||||
stub("plugin_helper", Plugin_Objects=MagicMock, is_mac=lambda value: bool(_MAC_PATTERN.match(value)))
|
||||
stub("utils.datetime_utils", timeNowUTC=_fixed_now)
|
||||
stub("logger", mylog=MagicMock(), Logger=MagicMock())
|
||||
stub("helper", get_setting_value=MagicMock(return_value="UTC"))
|
||||
stub("const", logPath="/tmp")
|
||||
stub("conf", tz=None)
|
||||
stub("pytz", timezone=MagicMock(return_value="UTC"))
|
||||
stub("utils.crypto_utils", string_to_fake_mac=MagicMock(return_value="fa:ce:00:00:00:01"))
|
||||
|
||||
module_path = Path(__file__).resolve().parents[2] / "server" / "plugins" / "pihole_api_scan" / "pihole_api_scan.py"
|
||||
spec = importlib.util.spec_from_file_location("pihole_api_scan_script", module_path)
|
||||
module = importlib.util.module_from_spec(spec)
|
||||
try:
|
||||
spec.loader.exec_module(module)
|
||||
finally:
|
||||
for name, previous_module in previous_modules.items():
|
||||
if previous_module is missing_module:
|
||||
sys.modules.pop(name, None)
|
||||
else:
|
||||
sys.modules[name] = previous_module
|
||||
|
||||
return module
|
||||
|
||||
|
||||
pihole_api_scan = _load_pihole_api_scan_module()
|
||||
|
||||
|
||||
def _api_device(hwaddr, ips, last_query, last_seen):
|
||||
"""Build one /api/network/devices entry with the fields Pi-hole FTL v6 returns."""
|
||||
return {
|
||||
"id": 1,
|
||||
"hwaddr": hwaddr,
|
||||
"interface": "eth0",
|
||||
"firstSeen": NOW_TS - 30 * 86400,
|
||||
"lastQuery": last_query,
|
||||
"numQueries": 42,
|
||||
"macVendor": "Test Vendor",
|
||||
"ips": [
|
||||
{"ip": ip, "name": f"host{index}.lan", "lastSeen": last_seen, "nameUpdated": last_seen}
|
||||
for index, ip in enumerate(ips)
|
||||
],
|
||||
}
|
||||
|
||||
|
||||
def _gather(devices, get_offline=False, consider_online=300):
|
||||
"""Run gather_device_entries() against the given API devices and plugin settings."""
|
||||
with patch.object(pihole_api_scan, "get_pihole_network_devices", return_value=devices), \
|
||||
patch.object(pihole_api_scan, "PIHOLEAPI_GET_OFFLINE", get_offline), \
|
||||
patch.object(pihole_api_scan, "PIHOLEAPI_CONSIDER_ONLINE", consider_online), \
|
||||
patch.object(pihole_api_scan, "PIHOLEAPI_FAKE_MAC", False):
|
||||
return pihole_api_scan.gather_device_entries()
|
||||
|
||||
|
||||
class TestOnlineFromLastQuery:
|
||||
def test_stale_last_query_with_fresh_last_seen_is_skipped(self):
|
||||
"""The reported bug: the phone's last DNS query is 5 h 47 min old, but
|
||||
Pi-hole still refreshes its ips[].lastSeen (10 s old)."""
|
||||
phone = _api_device("0a:60:b8:ad:b6:9d", ["192.168.178.64"], last_query=1793456269, last_seen=1793477100)
|
||||
|
||||
assert _gather([phone]) == []
|
||||
|
||||
def test_entry_reports_device_last_query(self):
|
||||
"""The 'Last Query' column (watched3) shows the last DNS query, not ips[].lastSeen."""
|
||||
laptop = _api_device("f8:01:b4:b4:67:94", ["192.168.178.52"], last_query=1793477050, last_seen=1793477105)
|
||||
|
||||
entries = _gather([laptop])
|
||||
|
||||
assert [entry["lastQuery"] for entry in entries] == ["1793477050"]
|
||||
|
||||
def test_recent_last_query_is_online(self):
|
||||
"""A device whose last DNS query is 60 s old is imported as online."""
|
||||
laptop = _api_device("f8:01:b4:b4:67:94", ["192.168.178.52"], last_query=1793477050, last_seen=1793477105)
|
||||
|
||||
entries = _gather([laptop])
|
||||
|
||||
assert [entry["isOnline"] for entry in entries] == [True]
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"last_query, expected_online",
|
||||
[
|
||||
(1793476810, [True]), # exactly 300 s old: still online
|
||||
(1793476809, []), # 301 s old: offline, so skipped
|
||||
(1793477140, [True]), # 30 s in the future (Pi-hole clock ahead): online
|
||||
],
|
||||
)
|
||||
def test_consider_online_window(self, last_query, expected_online):
|
||||
"""The PIHOLEAPI_CONSIDER_ONLINE window is inclusive, and a lastQuery slightly in the future counts as online."""
|
||||
laptop = _api_device("f8:01:b4:b4:67:94", ["192.168.178.52"], last_query=last_query, last_seen=1793477105)
|
||||
|
||||
entries = _gather([laptop], consider_online=300)
|
||||
|
||||
assert [entry["isOnline"] for entry in entries] == expected_online
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"last_query, consider_online",
|
||||
[
|
||||
(None, 300), # field missing in the API response
|
||||
(0, 300), # Pi-hole never saw a DNS query from this device
|
||||
(0, 10_000_000_000), # a huge window must not turn "never" into online
|
||||
],
|
||||
)
|
||||
def test_never_queried_device_is_offline(self, last_query, consider_online):
|
||||
"""A device without a lastQuery (missing or 0) is offline, whatever the window."""
|
||||
scale = _api_device("00:24:e4:d0:82:7c", ["192.168.178.50"], last_query=last_query, last_seen=1793477100)
|
||||
if last_query is None:
|
||||
del scale["lastQuery"]
|
||||
|
||||
entries = _gather([scale], get_offline=True, consider_online=consider_online)
|
||||
|
||||
assert [entry["isOnline"] for entry in entries] == [False]
|
||||
|
||||
def test_multi_ip_device_rows_share_online_state(self):
|
||||
"""lastQuery is per device, so every IP row of a device gets the same state,
|
||||
however fresh each IP's own lastSeen is."""
|
||||
tv = _api_device("f8:01:b4:b4:67:94", ["192.168.178.52", "192.168.178.53"], last_query=1793453291, last_seen=1793477100)
|
||||
|
||||
entries = _gather([tv], get_offline=True)
|
||||
|
||||
assert [(entry["ip"], entry["isOnline"]) for entry in entries] == [
|
||||
("192.168.178.52", False),
|
||||
("192.168.178.53", False),
|
||||
]
|
||||
|
||||
|
||||
def _use_settings(monkeypatch, settings):
|
||||
"""Serve settings to main() via get_setting_value, and restore the module globals main() rebinds after the test."""
|
||||
for name in ("PIHOLEAPI_URL", "PIHOLEAPI_PASSWORD", "PIHOLEAPI_API_MAXCLIENTS", "PIHOLEAPI_VERIFY_SSL",
|
||||
"PIHOLEAPI_RUN_TIMEOUT", "PIHOLEAPI_GET_OFFLINE", "PIHOLEAPI_CONSIDER_ONLINE"):
|
||||
monkeypatch.setattr(pihole_api_scan, name, getattr(pihole_api_scan, name))
|
||||
monkeypatch.setattr(pihole_api_scan, "get_setting_value", settings.get)
|
||||
|
||||
|
||||
def _run_main(monkeypatch, devices):
|
||||
"""Run main() with PIHOLEAPI_GET_OFFLINE on against the given API devices; return add_object() kwargs keyed by (MAC, IP)."""
|
||||
# Auth and the devices request are patched below, so main() needs no other settings.
|
||||
_use_settings(monkeypatch, {
|
||||
"PIHOLEAPI_URL": "http://pi.hole/",
|
||||
"PIHOLEAPI_GET_OFFLINE": True,
|
||||
"PIHOLEAPI_CONSIDER_ONLINE": 300,
|
||||
})
|
||||
plugin_objects = MagicMock()
|
||||
monkeypatch.setattr(pihole_api_scan, "pihole_api_auth", lambda: True)
|
||||
monkeypatch.setattr(pihole_api_scan, "pihole_api_deauth", lambda: None)
|
||||
monkeypatch.setattr(pihole_api_scan, "get_pihole_network_devices", lambda: devices)
|
||||
monkeypatch.setattr(pihole_api_scan, "PIHOLEAPI_FAKE_MAC", False)
|
||||
monkeypatch.setattr(pihole_api_scan, "plugin_objects", plugin_objects)
|
||||
|
||||
pihole_api_scan.main()
|
||||
return {(call.kwargs["primaryId"], call.kwargs["secondaryId"]): call.kwargs for call in plugin_objects.add_object.call_args_list}
|
||||
|
||||
|
||||
class TestPresenceFromOnlineState:
|
||||
def test_offline_import_is_not_present(self, monkeypatch):
|
||||
"""With PIHOLEAPI_GET_OFFLINE, an offline device is still imported, but its
|
||||
rows must not assert presence (helpVal4 maps to CurrentScan.scanPresence)."""
|
||||
phone = _api_device("0a:60:b8:ad:b6:9d", ["192.168.178.64"], last_query=1793456269, last_seen=1793477100)
|
||||
tv = _api_device("f8:01:b4:b4:67:94", ["192.168.178.52", "192.168.178.53"], last_query=1793477050, last_seen=1793477100)
|
||||
|
||||
calls = _run_main(monkeypatch, [phone, tv])
|
||||
|
||||
assert {key: kwargs.get("helpVal4") for key, kwargs in calls.items()} == {
|
||||
("0a:60:b8:ad:b6:9d", "192.168.178.64"): "0",
|
||||
("f8:01:b4:b4:67:94", "192.168.178.52"): "1",
|
||||
("f8:01:b4:b4:67:94", "192.168.178.53"): "1",
|
||||
}
|
||||
@@ -0,0 +1,48 @@
|
||||
"""
|
||||
PIHOLEAPI's config.json must route the plugin's helpVal4 to
|
||||
CurrentScan.scanPresence (server/plugin.py:process_plugin_events()).
|
||||
|
||||
pihole_api_scan.py writes helpVal4 '0' for a device it imports only because
|
||||
PIHOLEAPI_GET_OFFLINE is on. Without the mapping, scanPresence falls back to
|
||||
its default 1 and that offline device is marked as present.
|
||||
"""
|
||||
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
|
||||
from db_test_helpers import ( # noqa: E402
|
||||
make_plugin_db,
|
||||
make_plugin_dict,
|
||||
make_plugin_event_row,
|
||||
CREATE_CURRENT_SCAN,
|
||||
)
|
||||
|
||||
from plugin import process_plugin_events # noqa: E402
|
||||
|
||||
CONFIG_PATH = Path(__file__).resolve().parents[2] / "server" / "plugins" / "pihole_api_scan" / "config.json"
|
||||
|
||||
|
||||
def _piholeapi_plugin() -> dict:
|
||||
"""Return a plugin dict with PIHOLEAPI's real CurrentScan mapping from its config.json."""
|
||||
config = json.loads(CONFIG_PATH.read_text())
|
||||
plugin = make_plugin_dict(config["unique_prefix"])
|
||||
plugin["mapped_to_table"] = config["mapped_to_table"]
|
||||
plugin["database_column_definitions"] = config["database_column_definitions"]
|
||||
return plugin
|
||||
|
||||
|
||||
def test_offline_row_does_not_assert_presence():
|
||||
"""A PIHOLEAPI row with helpVal4 '0' lands in CurrentScan with scanPresence 0."""
|
||||
db, conn = make_plugin_db()
|
||||
conn.execute(CREATE_CURRENT_SCAN)
|
||||
conn.commit()
|
||||
try:
|
||||
row = make_plugin_event_row("PIHOLEAPI", "0a:60:b8:ad:b6:9d", secondary_id="192.168.178.64", help_val4="0")
|
||||
process_plugin_events(db, _piholeapi_plugin(), [row])
|
||||
|
||||
assert conn.execute("SELECT scanMac, scanPresence FROM CurrentScan").fetchall() == [("0a:60:b8:ad:b6:9d", 0)]
|
||||
finally:
|
||||
conn.close()
|
||||
Reference in new issue
Block a user