"
+ },
+ {
+ "equals": "watched-changed",
+ "replacement": "
"
+ },
+ {
+ "equals": "new",
+ "replacement": "
"
+ },
+ {
+ "equals": "missing-in-last-scan",
+ "replacement": "
"
+ }
+ ],
+ "localized": ["name"],
+ "name": [{ "language_code": "en_us", "string": "Status" }]
+ }
+ ]
+}
diff --git a/server/plugins/pihole_monitor/pihole_monitor.py b/server/plugins/pihole_monitor/pihole_monitor.py
new file mode 100644
index 000000000..a3f17b40a
--- /dev/null
+++ b/server/plugins/pihole_monitor/pihole_monitor.py
@@ -0,0 +1,681 @@
+#!/usr/bin/env python
+"""NetAlertX plugin: PIHOLEMON — Pi-hole Monitor
+
+Does two jobs against the same Pi-hole connection(s), instead of two
+separately configured plugins:
+
+ 1. Device import (same job as the official PIHOLEAPI/pihole_api_scan
+ plugin): pulls the device list from Pi-hole's `/api/network/devices`
+ and feeds it into NetAlertX's normal device-scanner pipeline
+ (mapped_to_table=CurrentScan), so devices Pi-hole knows about but
+ NetAlertX doesn't get created automatically.
+
+ 2. Query anomaly detection: pulls `/api/stats/top_clients?blocked=true`
+ and flags a device whose blocked-query count spikes well above its
+ own recent rolling average - the signature of malware/a compromised
+ device beaconing out, not just "a lot of DNS traffic".
+
+Why one plugin instead of two: they need the exact same Pi-hole session
+(auth once, reuse for both endpoints) and the exact same "primary +
+optional secondary" source list, so splitting them would mean either two
+logins per source or two separately configured URL/password pairs to keep
+in sync. One plugin, one settings page, one login per source.
+
+Why not just run two copies of the official PIHOLEAPI plugin for two
+Pi-holes: we looked into this first. `pihole_api_scan.py` hardcodes its
+settings-key prefix (`PIHOLEAPI_URL`, `PIHOLEAPI_PASSWORD`, ...) as literal
+strings throughout the script rather than reading it from `config.json`.
+Duplicating the plugin folder gives you two copies that both read and
+write the *same* settings keys - not two independent instances - and
+NetAlertX's own plugin docs don't describe an officially supported way to
+run multiple instances of one plugin. Making a real second instance would
+mean forking the script and renaming every occurrence of the prefix by
+hand, then keeping that fork in sync with any upstream changes by hand
+too. This plugin exists so none of that is necessary: it accepts a second
+set of credentials natively, and the secondary instance is entirely
+optional - leave its URL blank and this behaves like a single-Pi-hole
+import, which covers most setups.
+"""
+
+import os
+import sys
+import json
+
+import requests
+from requests.packages.urllib3.exceptions import InsecureRequestWarning
+
+INSTALL_PATH = os.getenv('NETALERTX_APP', '/app')
+sys.path.extend([f"{INSTALL_PATH}/server/plugins", f"{INSTALL_PATH}/server"])
+
+from plugin_helper import Plugin_Objects, is_mac # noqa: E402
+from utils.datetime_utils import timeNowUTC # noqa: E402
+from logger import mylog, Logger # noqa: E402
+from helper import get_setting_value # noqa: E402
+from const import logPath, dbFolderPath # noqa: E402
+import conf # noqa: E402
+from pytz import timezone # noqa: E402
+from utils.crypto_utils import string_to_fake_mac # noqa: E402
+
+conf.tz = timezone(get_setting_value('TIMEZONE'))
+Logger(get_setting_value('LOG_LEVEL'))
+
+pluginName = 'PIHOLEMON'
+VERSION_DATE = "NAX-PIHOLEMON-1.0"
+
+LOG_PATH = logPath + '/plugins'
+RESULT_FILE = os.path.join(LOG_PATH, f'last_result.{pluginName}.log')
+# Lives in the DB folder, not LOG_PATH: logs are routinely wiped on upgrade,
+# which would silently reset every device's anomaly baseline.
+STATE_FILE = os.path.join(dbFolderPath, f'state.{pluginName}.json')
+
+REQUEST_TIMEOUT_DEFAULT = 30
+
+
+class PiholeSource:
+ """One Pi-hole instance's connection + auth state, kept isolated from
+ any other instance so two can run side by side without interfering."""
+
+ def __init__(self, label, url, password, verify_ssl, run_timeout):
+ """Store this instance's connection details. Does not connect -
+ call auth() to actually log in."""
+ self.label = label
+ self.url = url.rstrip('/') + '/' if url else None
+ self.password = password
+ self.verify_ssl = verify_ssl
+ self.run_timeout = run_timeout
+ self.sid = None
+ self.csrf = None
+
+ @property
+ def configured(self):
+ """True if a URL was set for this instance (the secondary one is
+ optional and left unconfigured in most setups)."""
+ return bool(self.url)
+
+ def auth(self):
+ """Log in to this instance's /api/auth, storing the session id and
+ CSRF token for subsequent requests. Returns False (and logs why)
+ on any failure - never raises, so one bad source doesn't abort
+ the whole run."""
+ if not self.configured:
+ return False
+
+ if not self.verify_ssl:
+ requests.packages.urllib3.disable_warnings(InsecureRequestWarning)
+
+ headers = {
+ "accept": "application/json",
+ "content-type": "application/json",
+ "User-Agent": "NetAlertX/" + VERSION_DATE,
+ }
+ try:
+ resp = requests.post(
+ self.url + 'api/auth',
+ headers=headers,
+ json={"password": self.password},
+ verify=self.verify_ssl,
+ timeout=self.run_timeout,
+ )
+ resp.raise_for_status()
+ except requests.exceptions.Timeout:
+ mylog('none', [f'[{pluginName}] {self.label}: auth request timed out. Try increasing the run timeout.'])
+ return False
+ except requests.exceptions.ConnectionError:
+ mylog('none', [f'[{pluginName}] {self.label}: connection error during auth. Check the URL and password.'])
+ return False
+ except Exception as e:
+ mylog('none', [f'[{pluginName}] {self.label}: unexpected auth error: {e}'])
+ return False
+
+ try:
+ session_data = resp.json().get('session', {})
+ except Exception:
+ mylog('none', [f'[{pluginName}] {self.label}: unable to parse auth response JSON.'])
+ return False
+
+ if not session_data.get('valid', False):
+ mylog('none', [f'[{pluginName}] {self.label}: auth required or failed.'])
+ return False
+
+ self.sid = session_data.get('sid')
+ self.csrf = session_data.get('csrf')
+ mylog('verbose', [f'[{pluginName}] {self.label}: authenticated (sid present).'])
+ return True
+
+ def deauth(self):
+ """Best-effort logout so this instance doesn't accumulate sessions
+ across runs. Never raises - a failed logout isn't worth failing
+ the run over."""
+ if not self.configured or not self.sid:
+ return
+ try:
+ requests.delete(
+ self.url + 'api/auth',
+ headers={"X-FTL-SID": self.sid},
+ verify=self.verify_ssl,
+ timeout=self.run_timeout,
+ )
+ except Exception:
+ pass # best-effort logout
+ self.sid = None
+ self.csrf = None
+
+ def _headers(self):
+ """Auth headers for an authenticated request against this instance."""
+ headers = {"X-FTL-SID": self.sid}
+ if self.csrf:
+ headers["X-FTL-CSRF"] = self.csrf
+ return headers
+
+ def fetch_devices(self, max_clients):
+ """Raw 'devices' list from Pi-hole's network/devices endpoint - MAC,
+ IP(s), hostname, vendor, last-seen. Used for device import."""
+ if not self.sid:
+ return []
+ params = {'max_devices': str(max_clients), 'max_addresses': '2'}
+ try:
+ resp = requests.get(
+ self.url + 'api/network/devices',
+ headers=self._headers(),
+ params=params,
+ verify=self.verify_ssl,
+ timeout=self.run_timeout,
+ )
+ resp.raise_for_status()
+ data = resp.json()
+ mylog('debug', [f'[{pluginName}] {self.label}: devices API returned data: {json.dumps(data)}'])
+ return data.get('devices', [])
+ except Exception as e:
+ mylog('none', [f'[{pluginName}] {self.label}: failed to fetch devices: {e}'])
+ return []
+
+ def fetch_top_blocked_clients(self, count):
+ """{ip: blocked_count} for this instance, used for anomaly detection.
+
+ Each blocked_count is Pi-hole's raw counter value, cumulative since
+ FTL last started - not a per-interval or "since last poll" count,
+ and it does not reset daily. Callers must diff it against the
+ previous run's value (see compute_delta()) before comparing it to
+ anything; used raw, it would make any device's ordinary traffic
+ look like a runaway anomaly purely from the counter never resetting.
+
+ `count` should cover every client Pi-hole is tracking, not just a
+ handful - Pi-hole's own API default (10) truncates silently, so a
+ caller that doesn't pass an explicit count would never see clients
+ past that cutoff. Returns None (not {}) on any failure to fetch or
+ parse the response, so callers can tell "no source authenticated
+ for this instance right now" apart from "this instance genuinely
+ has no blocked queries this run" - treating the two the same would
+ write a false zero into a device's history and dilute its baseline.
+ """
+ if not self.sid:
+ return None
+ try:
+ resp = requests.get(
+ self.url + 'api/stats/top_clients',
+ headers=self._headers(),
+ params={"blocked": "true", "count": count},
+ verify=self.verify_ssl,
+ timeout=self.run_timeout,
+ )
+ resp.raise_for_status()
+ clients = resp.json().get("clients", [])
+ return {c["ip"]: c.get("count", 0) for c in clients if c.get("ip")}
+ except Exception as e:
+ mylog('none', [f'[{pluginName}] {self.label}: failed to fetch top_clients: {e}'])
+ return None
+
+
+def gather_device_entries(source, consider_online, fake_mac, max_clients):
+ """Same parsing logic as the official PIHOLEAPI plugin, scoped to one source.
+
+ Returns every device/IP pair Pi-hole knows about, each tagged with
+ is_online. Callers decide separately what to do with that flag:
+ device-import rows should skip offline devices unless GET_OFFLINE is
+ set, but the IP->MAC identity mapping (used to attribute blocked-query
+ counts to the right device) must NOT skip them - Pi-hole's own "last
+ seen" can lag behind real DNS activity, so a device it currently calls
+ offline can still be the one generating the blocked queries in this
+ same run. Dropping it there would misattribute the traffic to a bare
+ IP instead of the device's real MAC.
+ """
+ entries = []
+ devices = source.fetch_devices(max_clients)
+ now_ts = int(timeNowUTC(as_string=False).timestamp())
+
+ for device in devices:
+ hwaddr = device.get('hwaddr')
+ # "ip-
" is Pi-hole's own placeholder for "no real MAC known,
+ # falling back to identifying by IP" - not just the "ip-::" (IPv6)
+ # case, any address. Caught downstream by is_mac() either way, but
+ # this is the actual placeholder check, so it should recognize the
+ # whole pattern.
+ if not hwaddr or hwaddr == "00:00:00:00:00:00" or hwaddr.startswith("ip-"):
+ continue
+
+ device_ips = device.get('ips', [])
+ if not device_ips:
+ continue
+
+ max_last_seen = max((ip_info.get('lastSeen', 0) for ip_info in device_ips), default=0)
+ is_online = (now_ts - max_last_seen) <= consider_online
+
+ mac_vendor = device.get('macVendor', '')
+
+ for ip_info in device_ips:
+ ip = ip_info.get('ip')
+ if not ip or ip in ["0.0.0.0", "::"]:
+ continue
+
+ name = ip_info.get('name') or ''
+ tmp_mac = hwaddr.lower()
+
+ if fake_mac and not is_mac(tmp_mac):
+ tmp_mac = string_to_fake_mac(ip)
+
+ entries.append({
+ 'mac': tmp_mac,
+ 'ip': ip,
+ 'name': name,
+ 'macVendor': mac_vendor,
+ 'lastSeen': max_last_seen,
+ 'is_online': is_online,
+ })
+
+ return entries
+
+
+def merge_device_entries(all_entries):
+ """One entry per MAC - the freshest, if the same device shows up on both
+ Pi-hole instances (usually with the same IP, but not always)."""
+ merged = {}
+ for entry in all_entries:
+ current = merged.get(entry['mac'])
+ if current is None or entry['lastSeen'] > current['lastSeen']:
+ merged[entry['mac']] = entry
+ return merged
+
+
+def build_ip_to_mac(all_entries):
+ """Map every IP Pi-hole has ever associated with a device to that
+ device's MAC, for attributing blocked-query counts (which only come
+ back as IPs) to the right device.
+
+ Deliberately built from every gathered entry, not from
+ merge_device_entries()'s output: a device with more than one IP gets
+ one entry per IP in `all_entries`, but merge_device_entries() keeps
+ only the single freshest entry per MAC - so deriving the IP map from
+ its result would silently drop that device's other IPs, and any
+ blocked-query traffic seen from those would fall back to being
+ tracked under a bare IP instead of the device's real MAC. If two
+ different MACs were ever seen on the same IP (e.g. a DHCP
+ reassignment), the entry with the freshest lastSeen wins that IP.
+ """
+ ip_to_mac = {}
+ ip_last_seen = {}
+ for entry in all_entries:
+ ip = entry['ip']
+ if ip not in ip_to_mac or entry['lastSeen'] > ip_last_seen[ip]:
+ ip_to_mac[ip] = entry['mac']
+ ip_last_seen[ip] = entry['lastSeen']
+ return ip_to_mac
+
+
+def netalertx_device_owners(graphql_url, token, run_timeout):
+ """{mac: devOwner} for every device NetAlertX already knows about, purely
+ for a friendlier anomaly label. Fetched once per run rather than once per
+ device - on a network with hundreds of devices, one query beats hundreds
+ of blocking round-trips to the same endpoint. Returns {} if unavailable,
+ unset, or on any error - never blocks device import or anomaly detection."""
+ if not graphql_url:
+ return {}
+
+ query = """
+ query GetDevices {
+ devices {
+ devices { devMac devOwner }
+ }
+ }
+ """
+
+ try:
+ headers = {"Authorization": f"Bearer {token}"} if token else {}
+ resp = requests.post(
+ graphql_url,
+ json={"query": query},
+ headers=headers,
+ timeout=run_timeout,
+ )
+ resp.raise_for_status()
+ devices = resp.json().get("data", {}).get("devices", {}).get("devices", [])
+ return {d["devMac"]: d.get("devOwner") or '' for d in devices if d.get("devMac")}
+ except Exception as e:
+ mylog('debug', [f'[{pluginName}] GraphQL owner lookup failed: {e}'])
+ return {}
+
+
+def load_state():
+ """Per-key {"last_raw": {source_label: int}, "history": [[timestamp,
+ delta], ...]} from past runs, or {} on first run / a missing or corrupt
+ state file (never fatal - just starts fresh). last_raw is keyed by
+ source, not a single number, so each Pi-hole instance gets its own
+ diff reference point - see aggregate_source_deltas()."""
+ try:
+ with open(STATE_FILE, "r") as f:
+ return json.load(f)
+ except Exception:
+ return {}
+
+
+def save_state(state):
+ """Persist per-key last-raw-count + delta history for next run's diff
+ and baseline."""
+ os.makedirs(os.path.dirname(STATE_FILE), exist_ok=True)
+ with open(STATE_FILE, "w") as f:
+ json.dump(state, f)
+
+
+def trim_history(history, now_ts, history_days):
+ """Drop samples older than history_days from `history` ([timestamp,
+ delta] pairs). An age cutoff, not a count: the window means the same
+ real-world span regardless of how often this plugin happens to run - a
+ faster schedule just adds more samples inside that same window instead
+ of shrinking it, and a slower one doesn't stretch it out."""
+ cutoff = now_ts - history_days * 86400
+ return [sample for sample in history if sample[0] >= cutoff]
+
+
+def compute_delta(last_raw, current_raw):
+ """Turn Pi-hole's raw blocked-query count (cumulative since FTL last
+ started, *not* a per-interval count - confirmed against FTL's own
+ source and long-standing user reports that it doesn't reset at
+ midnight) into a per-run increment, which is what's actually
+ comparable against a rolling baseline.
+
+ Returns None (not 0) when there's nothing valid to diff against yet:
+ the first time this device is seen (last_raw is None), or when
+ current_raw < last_raw - Pi-hole/FTL restarted and the counter reset,
+ or the device simply dropped out of top_clients this run. A caller
+ must not treat None as a real zero: a genuine 0 means "no new blocked
+ queries since last run", while None means "we can't tell this run" -
+ conflating them would either manufacture a fake anomaly out of a
+ restart, or silently swallow a real one right after."""
+ if last_raw is None or current_raw < last_raw:
+ return None
+ return current_raw - last_raw
+
+
+def aggregate_source_deltas(last_raw_by_source, raw_by_source):
+ """Combine each configured source's raw count into one delta for a
+ device, computing every source's delta independently (via
+ compute_delta()) before summing - never by summing the raw totals
+ first and diffing once. Summing raw totals first would let one
+ source's counter reset silently net out against real traffic on
+ another: e.g. primary +2000 (a real spike) and secondary resetting
+ from 1000 to 5 (-995) would combine into a raw delta of only 1005,
+ hiding most of the primary's actual spike behind the secondary's
+ unrelated restart.
+
+ `raw_by_source` only needs entries for sources that reported this
+ device this run - a source that didn't (auth failed, or the device
+ simply wasn't in that instance's top_clients) is skipped for this run
+ without affecting the others.
+
+ Returns (delta, updated_last_raw_by_source):
+ - delta is None only if none of the sources present this run
+ produced a valid delta (e.g. all are bootstrapping or just
+ reset) - same None-means-"can't tell" contract as
+ compute_delta(). If at least one source has a valid delta, it's
+ included even if another source in the same run doesn't.
+ - updated_last_raw_by_source carries every source's newest raw
+ value forward (valid delta or not), so each source keeps its own
+ independent reference point for the next run.
+ """
+ updated = dict(last_raw_by_source)
+ total = 0
+ any_valid = False
+ for label, raw in raw_by_source.items():
+ delta = compute_delta(last_raw_by_source.get(label), raw)
+ updated[label] = raw
+ if delta is not None:
+ total += delta
+ any_valid = True
+ return (total if any_valid else None), updated
+
+
+def main():
+ """Entry point: authenticate to every configured Pi-hole instance,
+ import its devices, evaluate blocked-query anomalies against each
+ device's rolling history, and write both out. Returns 0 on a normal
+ run, 1 if no Pi-hole instance is configured at all."""
+ run_timeout = get_setting_value('PIHOLEMON_RUN_TIMEOUT') or REQUEST_TIMEOUT_DEFAULT
+ get_offline = bool(get_setting_value('PIHOLEMON_GET_OFFLINE'))
+ fake_mac = bool(get_setting_value('PIHOLEMON_FAKE_MAC'))
+ max_clients = get_setting_value('PIHOLEMON_API_MAXCLIENTS') or 500
+ consider_online = get_setting_value('PIHOLEMON_CONSIDER_ONLINE')
+ if not isinstance(consider_online, int):
+ consider_online = 300
+
+ # The user only decides whether to look up the owner at all - the
+ # endpoint itself is derived from this app's own GRAPHQL_PORT (single
+ # source of truth) instead of being a second, easily stale copy of it.
+ graphql_url = f"http://127.0.0.1:{get_setting_value('GRAPHQL_PORT')}/graphql" if get_setting_value('PIHOLEMON_GET_OWNER') else None
+ # Reuse this app's own API token rather than keep a second, easily
+ # forgotten copy of it in this plugin's settings.
+ graphql_token = get_setting_value('API_TOKEN')
+ multiplier = float(get_setting_value('PIHOLEMON_MULTIPLIER') or 4)
+ min_blocked = int(get_setting_value('PIHOLEMON_MIN_BLOCKED') or 20)
+ # Days, not run count: a run-count window silently shrinks or stretches
+ # in real time whenever RUN_SCHD changes (or differs between users), so
+ # the baseline it produces means something different depending on how
+ # often the plugin happens to run. A day-based window means the same
+ # thing regardless of schedule, and a faster schedule only adds more
+ # data points within that same window instead of shortening it.
+ # Clamped to at least 1 for the same reason as elsewhere: 0 already
+ # falls back to 7 via `or`, but a negative setting would otherwise
+ # produce a nonsensical, hard-to-debug cutoff below.
+ history_days = max(1, int(get_setting_value('PIHOLEMON_HISTORY_DAYS') or 7))
+
+ sources = [
+ PiholeSource(
+ 'primary',
+ get_setting_value('PIHOLEMON_PRIMARY_URL'),
+ get_setting_value('PIHOLEMON_PRIMARY_PASSWORD'),
+ bool(get_setting_value('PIHOLEMON_PRIMARY_VERIFY_SSL')),
+ run_timeout,
+ ),
+ PiholeSource(
+ 'secondary',
+ get_setting_value('PIHOLEMON_SECONDARY_URL'),
+ get_setting_value('PIHOLEMON_SECONDARY_PASSWORD'),
+ bool(get_setting_value('PIHOLEMON_SECONDARY_VERIFY_SSL')),
+ run_timeout,
+ ),
+ ]
+ configured_sources = [s for s in sources if s.configured]
+ if not configured_sources:
+ mylog('none', [f'[{pluginName}] No Pi-hole URL configured - nothing to do.'])
+ return 1
+
+ all_device_entries = []
+ # Pi-hole's raw, cumulative-since-FTL-started counts (see
+ # compute_delta()'s docstring), kept separate per source until each
+ # source's own delta is computed - see aggregate_source_deltas()'
+ # docstring for why combining the raw totals across sources first
+ # (before diffing) would be wrong.
+ blocked_by_source_ip = {}
+ # False if any configured source failed to authenticate or its
+ # top_clients fetch failed - the blocked-query counts for this run are
+ # then incomplete for reasons unrelated to real traffic, so anomaly
+ # evaluation and history persistence are skipped below rather than
+ # risk writing a false "quiet run" into a device's baseline.
+ stats_complete = True
+
+ for source in configured_sources:
+ if not source.auth():
+ mylog('none', [f'[{pluginName}] {source.label}: authentication failed - skipping this source.'])
+ stats_complete = False
+ continue
+ try:
+ all_device_entries.extend(
+ gather_device_entries(source, consider_online, fake_mac, max_clients)
+ )
+ top_blocked = source.fetch_top_blocked_clients(count=max_clients)
+ if top_blocked is None:
+ stats_complete = False
+ else:
+ blocked_by_source_ip[source.label] = top_blocked
+ finally:
+ source.deauth()
+
+ # IP->MAC identity mapping uses every device Pi-hole knows about,
+ # online or not (see gather_device_entries docstring for why), and is
+ # built from every entry rather than the by-MAC merge below so a
+ # multi-IP device doesn't lose its other IPs (see build_ip_to_mac).
+ ip_to_mac = build_ip_to_mac(all_device_entries)
+
+ # Device-import rows (name/vendor) still respect GET_OFFLINE.
+ importable_entries = [e for e in all_device_entries if e['is_online'] or get_offline]
+ for entry in all_device_entries:
+ if not entry['is_online'] and not get_offline:
+ mylog('verbose', [f"[{pluginName}]: skipping offline device import for {entry['mac']} ({entry['ip']})."])
+ devices_by_mac = merge_device_entries(importable_entries)
+
+ # Combine blocked-query counts per MAC, still kept separate per source
+ # (see aggregate_source_deltas()). An IP Pi-hole has genuinely never
+ # associated with any MAC (not even an offline one) falls back to being
+ # tracked under its own IP, so the signal isn't silently dropped.
+ blocked_by_mac_by_source = {}
+ for label, ip_counts in blocked_by_source_ip.items():
+ mac_counts = {}
+ for ip, count in ip_counts.items():
+ key = ip_to_mac.get(ip, ip)
+ mac_counts[key] = mac_counts.get(key, 0) + count
+ blocked_by_mac_by_source[label] = mac_counts
+
+ if not stats_complete:
+ mylog(
+ 'none',
+ [f'[{pluginName}] Blocked-query data is incomplete for this run '
+ '(a source failed to authenticate or its top_clients fetch failed) - '
+ 'skipping anomaly evaluation and history updates so a real outage '
+ 'doesn\'t get recorded as a quiet run.'],
+ )
+
+ state = load_state()
+ plugin_objects = Plugin_Objects(RESULT_FILE)
+ blocked_keys = {key for mac_counts in blocked_by_mac_by_source.values() for key in mac_counts}
+ all_keys = set(devices_by_mac.keys()) | blocked_keys
+ # One batched lookup for the whole run instead of one per device - see
+ # netalertx_device_owners' docstring.
+ owners_by_mac = netalertx_device_owners(graphql_url, graphql_token, run_timeout)
+ now_ts = int(timeNowUTC(as_string=False).timestamp())
+
+ for key in all_keys:
+ device = devices_by_mac.get(key)
+ mac = key if is_mac(key) else None
+
+ entry = state.get(key, {})
+ # A plain number here is a pre-existing state file from before
+ # last_raw was tracked per source - treat it the same as no prior
+ # reference point at all (every source bootstraps fresh) rather
+ # than crash on it.
+ last_raw_by_source = entry.get("last_raw", {})
+ if not isinstance(last_raw_by_source, dict):
+ last_raw_by_source = {}
+ history = trim_history(entry.get("history", []), now_ts, history_days)
+ values = [sample[1] for sample in history]
+ # `baseline is not None` (not a truthy check): a device with a real,
+ # all-zero history has baseline == 0.0, which is itself meaningful -
+ # any blocked traffic at all on such a device is a spike from its own
+ # established normal. `baseline` alone is falsy for 0.0 and would
+ # silently exempt exactly the devices most worth watching.
+ baseline = sum(values) / len(values) if values else None
+
+ # None (not 0) when there's no valid per-run increment yet from any
+ # source - first time seen, or every reporting source just reset
+ # (see aggregate_source_deltas()). blocked_count is only a display
+ # fallback for that case; is_anomaly is gated on the real delta,
+ # not on this substitute.
+ if stats_complete:
+ raw_by_source = {
+ label: mac_counts[key]
+ for label, mac_counts in blocked_by_mac_by_source.items()
+ if key in mac_counts
+ }
+ delta, updated_last_raw_by_source = aggregate_source_deltas(last_raw_by_source, raw_by_source)
+ else:
+ delta = None
+ blocked_count = delta if delta is not None else 0
+ is_anomaly = bool(stats_complete and baseline is not None and delta is not None and blocked_count >= min_blocked and blocked_count > baseline * multiplier)
+
+ owner = owners_by_mac.get(mac, '') if mac else ''
+ if stats_complete and delta is None:
+ detail = "blocked=unknown (establishing baseline - first run seen, or Pi-hole/FTL restarted)"
+ else:
+ detail = f"blocked={blocked_count}"
+ if baseline is not None:
+ detail += f", avg={round(baseline, 1)}"
+ if baseline > 0:
+ detail += f", ratio={round(blocked_count / baseline, 2)}x"
+ if owner:
+ detail += f" - owner: {owner}"
+
+ if device:
+ if not is_mac(device['mac']):
+ mylog('verbose', [f"[{pluginName}] Skipping invalid MAC (see Generate fake MAC setting): {device}"])
+ continue
+ plugin_objects.add_object(
+ primaryId=str(device['mac']),
+ secondaryId=str(device['ip']),
+ watched1=str(device['name']),
+ watched2=str(device['macVendor']),
+ watched3=str(blocked_count),
+ watched4='anomaly' if is_anomaly else 'normal',
+ extra=detail,
+ foreignKey=str(device['mac']),
+ )
+ else:
+ # No device-import row this run for `key` - either it's a real,
+ # known MAC that's just offline-filtered above (still link the
+ # anomaly to that device's existing page via foreignKey), or a
+ # bare IP Pi-hole has never associated with any MAC at all
+ # (nothing to link to, foreignKey stays 'null').
+ known_mac = key if is_mac(key) else None
+ plugin_objects.add_object(
+ primaryId=key,
+ secondaryId=key,
+ watched1='',
+ watched2='',
+ watched3=str(blocked_count),
+ watched4='anomaly' if is_anomaly else 'normal',
+ extra=detail,
+ foreignKey=str(known_mac) if known_mac else 'null',
+ )
+
+ if is_anomaly:
+ mylog('none', [f'[{pluginName}] Anomaly: {key} - {detail}'])
+
+ if stats_complete:
+ # Always reset each source's diff reference point, even on a
+ # bootstrap or reset run (delta is None) - that's exactly what
+ # makes the *next* run's delta valid again instead of repeating
+ # the same "no valid delta" state indefinitely. Only append to
+ # the baseline history when this run actually produced a real
+ # (aggregate) delta.
+ if delta is not None:
+ history.append([now_ts, delta])
+ state[key] = {"last_raw": updated_last_raw_by_source, "history": history}
+
+ save_state(state)
+ plugin_objects.write_result_file()
+ mylog(
+ 'verbose',
+ [f'[{pluginName}] Script finished. {len(devices_by_mac)} device(s) imported, '
+ f'{len(blocked_keys)} with blocked-query data, from {len(configured_sources)} source(s).'],
+ )
+ return 0
+
+
+if __name__ == '__main__':
+ sys.exit(main())
diff --git a/test/plugins/test_mikrotik_scan.py b/test/plugins/test_mikrotik_scan.py
new file mode 100644
index 000000000..e79f217f4
--- /dev/null
+++ b/test/plugins/test_mikrotik_scan.py
@@ -0,0 +1,89 @@
+"""Tests for the MikroTik DHCP lease scanner."""
+
+import importlib.util
+import sys
+import types
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+
+def _load_mikrotik_module():
+ missing_module = object()
+ previous_modules = {}
+
+ def stub(name, **attributes):
+ 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
+
+ class TrapError(Exception):
+ pass
+
+ stub(
+ "plugin_helper",
+ Plugin_Objects=MagicMock,
+ normalize_mac=lambda mac: mac.strip().lower().replace("-", ":"),
+ )
+ 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("librouteros", connect=MagicMock())
+ stub("librouteros.exceptions", TrapError=TrapError)
+
+ module_path = Path(__file__).resolve().parents[2] / "server" / "plugins" / "mikrotik_scan" / "mikrotik.py"
+ spec = importlib.util.spec_from_file_location("mikrotik_scan", 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
+
+
+mikrotik = _load_mikrotik_module()
+
+
+def _lease(lease_id, address, mac_address, status="bound"):
+ lease = {
+ ".id": lease_id,
+ "address": address,
+ "host-name": f"device-{lease_id}",
+ "comment": "",
+ "last-seen": "1m",
+ "status": status,
+ }
+ if mac_address is not None:
+ lease["mac-address"] = mac_address
+ return lease
+
+
+def test_leases_without_mac_do_not_abort_remaining_leases():
+ leases = [
+ _lease("*1", "192.168.1.2", "aa-bb-cc-dd-ee-01"),
+ _lease("*2", "192.168.1.5", None, status="waiting"),
+ _lease("*3", "192.168.1.6", None),
+ _lease("*4", "192.168.1.8", "aa-bb-cc-dd-ee-04"),
+ ]
+ api = MagicMock(return_value=leases)
+ plugin_objects = MagicMock()
+
+ mikrotik.MT_USER = "user"
+ mikrotik.MT_PASS = None
+ mikrotik.MT_HOST = "192.168.1.1"
+ mikrotik.MT_PORT = 8728
+
+ with patch.object(mikrotik, "connect", return_value=api):
+ result = mikrotik.get_entries(plugin_objects)
+
+ assert result is plugin_objects
+ assert plugin_objects.add_object.call_count == 2
+ assert [call.kwargs["primaryId"] for call in plugin_objects.add_object.call_args_list] == ["aa:bb:cc:dd:ee:01", "aa:bb:cc:dd:ee:04"]
diff --git a/test/plugins/test_pihole_monitor.py b/test/plugins/test_pihole_monitor.py
new file mode 100644
index 000000000..a82609378
--- /dev/null
+++ b/test/plugins/test_pihole_monitor.py
@@ -0,0 +1,899 @@
+"""Tests for the pihole_monitor (PIHOLEMON) plugin.
+
+pihole_monitor.py is loaded with its NetAlertX-internal dependencies
+(plugin_helper, logger, helper, const, conf, pytz, utils.*) stubbed out,
+the same approach test_mikrotik_scan.py uses - it keeps these tests
+runnable without the full devcontainer environment and without any live
+Pi-hole. `requests` itself is left real; individual HTTP calls are mocked
+per test.
+
+Layout:
+ - PiholeSource.auth() / fetch_top_blocked_clients(): unit tests against
+ a mocked `requests`, covering the auth success/failure paths and the
+ None-sentinel-on-failure contract (vs. a genuine empty {}).
+ - netalertx_device_owners(): unit tests for the batched (one request for
+ every device, not one per device) owner lookup.
+ - build_ip_to_mac(): pure-function unit tests for the multi-IP-per-MAC
+ fix (a device must not lose its other IPs to the by-MAC merge).
+ - compute_delta(): pure-function unit tests turning Pi-hole's raw,
+ cumulative-since-FTL-started count into a real per-run increment -
+ None (not 0) on a first-ever run or a counter reset, a genuine 0
+ distinct from that None otherwise.
+ - aggregate_source_deltas(): pure-function unit tests for combining
+ per-source deltas correctly - a counter reset on one source must not
+ net out against real traffic on another (they're diffed
+ independently, then summed - never combined as raw totals first).
+ - main(): integration tests with PiholeSource's network-touching
+ methods stubbed at the object level, covering source aggregation,
+ the stats_complete gate (a failed fetch must not corrupt a device's
+ history with a false zero), the history_days age-based clamp/trim
+ (not a run count - see trim_history()), the zero-baseline anomaly fix
+ (a device with an all-zero history must still be flagged, not silently
+ exempted), the bootstrap/counter-reset runs that establish or
+ re-anchor last_raw without recording a bogus delta, the dual-source
+ reset-masking regression, tolerance of a pre-per-source state file
+ (last_raw as a plain number, from before this round), and
+ per-instance Verify SSL.
+"""
+
+import importlib.util
+import sys
+import types
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+import pytest
+import requests
+
+
+def _is_mac(value):
+ """Same shape as plugin_helper.is_mac, without pytz as a dependency."""
+ import re
+ s = str(value).lower().strip()
+ return bool(re.match(r"^[0-9a-f]{2}([-:]?)[0-9a-f]{2}(\1[0-9a-f]{2}){4}$", s))
+
+
+def _load_pihole_monitor_module():
+ missing_module = object()
+ previous_modules = {}
+
+ def stub(name, **attributes):
+ 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=_is_mac)
+ stub("logger", mylog=MagicMock(), Logger=MagicMock())
+ stub("helper", get_setting_value=MagicMock(return_value="UTC"))
+ stub("const", logPath="/tmp", dbFolderPath="/tmp/db")
+ stub("conf", tz=None)
+ stub("pytz", timezone=MagicMock(return_value="UTC"))
+ stub("utils")
+ stub("utils.datetime_utils", timeNowUTC=MagicMock())
+ stub("utils.crypto_utils", string_to_fake_mac=lambda s: "fa:ce:00:00:00:01")
+
+ module_path = Path(__file__).resolve().parents[2] / "server" / "plugins" / "pihole_monitor" / "pihole_monitor.py"
+ spec = importlib.util.spec_from_file_location("pihole_monitor", 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_monitor = _load_pihole_monitor_module()
+
+
+def _resp(json_data):
+ resp = MagicMock()
+ resp.raise_for_status = MagicMock()
+ resp.json = MagicMock(return_value=json_data)
+ return resp
+
+
+# ---------------------------------------------------------------------------
+# PiholeSource.auth()
+# ---------------------------------------------------------------------------
+
+
+def test_auth_success_stores_sid_and_csrf():
+ source = pihole_monitor.PiholeSource("primary", "http://ph1/", "pw", True, 5)
+ with patch("requests.post", return_value=_resp({"session": {"valid": True, "sid": "abc", "csrf": "xyz"}})):
+ assert source.auth() is True
+ assert source.sid == "abc"
+ assert source.csrf == "xyz"
+
+
+def test_auth_invalid_session_returns_false():
+ source = pihole_monitor.PiholeSource("primary", "http://ph1/", "wrongpw", True, 5)
+ with patch("requests.post", return_value=_resp({"session": {"valid": False}})):
+ assert source.auth() is False
+ assert source.sid is None
+
+
+def test_auth_connection_error_returns_false_without_raising():
+ import requests as real_requests
+ source = pihole_monitor.PiholeSource("primary", "http://unreachable/", "pw", True, 5)
+ with patch("requests.post", side_effect=real_requests.exceptions.ConnectionError("no route")):
+ assert source.auth() is False
+ assert source.sid is None
+
+
+def test_auth_unconfigured_source_short_circuits():
+ source = pihole_monitor.PiholeSource("secondary", "", "", True, 5)
+ with patch("requests.post") as mock_post:
+ assert source.auth() is False
+ mock_post.assert_not_called()
+
+
+def test_auth_timeout_returns_false():
+ source = pihole_monitor.PiholeSource("primary", "http://ph1/", "pw", True, 5)
+ with patch("requests.post", side_effect=requests.exceptions.Timeout("slow")):
+ assert source.auth() is False
+ assert source.sid is None
+
+
+def test_auth_unexpected_error_returns_false():
+ source = pihole_monitor.PiholeSource("primary", "http://ph1/", "pw", True, 5)
+ with patch("requests.post", side_effect=ValueError("boom")):
+ assert source.auth() is False
+
+
+def test_auth_unparseable_json_returns_false():
+ resp = MagicMock()
+ resp.raise_for_status = MagicMock()
+ resp.json = MagicMock(side_effect=ValueError("not json"))
+ source = pihole_monitor.PiholeSource("primary", "http://ph1/", "pw", True, 5)
+ with patch("requests.post", return_value=resp):
+ assert source.auth() is False
+
+
+def test_auth_disables_insecure_warning_when_verify_ssl_off():
+ source = pihole_monitor.PiholeSource("primary", "http://ph1/", "pw", False, 5)
+ session_resp = _resp({"session": {"valid": True, "sid": "s", "csrf": "c"}})
+ with patch("requests.post", return_value=session_resp), \
+ patch("requests.packages.urllib3.disable_warnings") as mock_disable:
+ assert source.auth() is True
+ mock_disable.assert_called_once()
+
+
+# ---------------------------------------------------------------------------
+# PiholeSource.deauth()
+# ---------------------------------------------------------------------------
+
+
+def test_deauth_clears_session_on_success():
+ source = pihole_monitor.PiholeSource("primary", "http://ph1/", "pw", True, 5)
+ source.sid = "sid"
+ source.csrf = "csrf"
+ with patch("requests.delete", return_value=_resp({})) as mock_delete:
+ source.deauth()
+ mock_delete.assert_called_once()
+ assert source.sid is None
+ assert source.csrf is None
+
+
+def test_deauth_swallows_request_errors():
+ source = pihole_monitor.PiholeSource("primary", "http://ph1/", "pw", True, 5)
+ source.sid = "sid"
+ with patch("requests.delete", side_effect=requests.exceptions.ConnectionError("gone")):
+ source.deauth() # must not raise
+ assert source.sid is None
+
+
+def test_deauth_noop_without_an_active_session():
+ source = pihole_monitor.PiholeSource("primary", "http://ph1/", "pw", True, 5)
+ with patch("requests.delete") as mock_delete:
+ source.deauth()
+ mock_delete.assert_not_called()
+
+
+# ---------------------------------------------------------------------------
+# PiholeSource.fetch_devices()
+# ---------------------------------------------------------------------------
+
+
+def test_fetch_devices_success_returns_device_list():
+ source = pihole_monitor.PiholeSource("primary", "http://ph1/", "pw", True, 5)
+ source.sid = "sid"
+ source.csrf = "csrf" # also exercises _headers() including X-FTL-CSRF
+ payload = {"devices": [{"hwaddr": "aa:bb:cc:dd:ee:01"}]}
+ with patch("requests.get", return_value=_resp(payload)) as mock_get:
+ result = source.fetch_devices(max_clients=500)
+ assert result == [{"hwaddr": "aa:bb:cc:dd:ee:01"}]
+ assert mock_get.call_args.kwargs["params"] == {"max_devices": "500", "max_addresses": "2"}
+
+
+def test_fetch_devices_failure_returns_empty_list():
+ source = pihole_monitor.PiholeSource("primary", "http://ph1/", "pw", True, 5)
+ source.sid = "sid"
+ with patch("requests.get", side_effect=requests.exceptions.Timeout("slow")):
+ assert source.fetch_devices(max_clients=500) == []
+
+
+def test_fetch_devices_without_session_returns_empty_list():
+ source = pihole_monitor.PiholeSource("primary", "http://ph1/", "pw", True, 5)
+ with patch("requests.get") as mock_get:
+ assert source.fetch_devices(max_clients=500) == []
+ mock_get.assert_not_called()
+
+
+# ---------------------------------------------------------------------------
+# netalertx_device_owners() - one batched request, not one per device
+# ---------------------------------------------------------------------------
+
+
+def test_netalertx_device_owners_returns_empty_dict_without_url():
+ assert pihole_monitor.netalertx_device_owners(None, "token", 5) == {}
+
+
+def test_netalertx_device_owners_returns_mac_keyed_dict_on_success():
+ payload = {"data": {"devices": {"devices": [
+ {"devMac": "aa:bb:cc:dd:ee:01", "devOwner": "Mauricio"},
+ {"devMac": "aa:bb:cc:dd:ee:02", "devOwner": ""},
+ ]}}}
+ with patch("requests.post", return_value=_resp(payload)) as mock_post:
+ owners = pihole_monitor.netalertx_device_owners("http://nax/graphql", "tok", 5)
+ assert owners == {"aa:bb:cc:dd:ee:01": "Mauricio", "aa:bb:cc:dd:ee:02": ""}
+ assert mock_post.call_args.kwargs["headers"] == {"Authorization": "Bearer tok"}
+ # No per-device filter - the whole device list comes back in one request.
+ assert "variables" not in mock_post.call_args.kwargs["json"]
+
+
+def test_netalertx_device_owners_returns_empty_dict_when_none_known():
+ payload = {"data": {"devices": {"devices": []}}}
+ with patch("requests.post", return_value=_resp(payload)):
+ owners = pihole_monitor.netalertx_device_owners("http://nax/graphql", "", 5)
+ assert owners == {}
+
+
+def test_netalertx_device_owners_returns_empty_dict_on_request_error():
+ with patch("requests.post", side_effect=requests.exceptions.ConnectionError("down")):
+ owners = pihole_monitor.netalertx_device_owners("http://nax/graphql", "tok", 5)
+ assert owners == {}
+
+
+def test_netalertx_device_owners_fetches_once_regardless_of_device_count():
+ """Regression guard for the N-round-trips bug: on a network with many
+ devices, this must still be exactly one HTTP call, not one per device."""
+ payload = {"data": {"devices": {"devices": [
+ {"devMac": f"aa:bb:cc:dd:ee:{i:02x}", "devOwner": f"user{i}"} for i in range(50)
+ ]}}}
+ with patch("requests.post", return_value=_resp(payload)) as mock_post:
+ owners = pihole_monitor.netalertx_device_owners("http://nax/graphql", "tok", 5)
+ assert mock_post.call_count == 1
+ assert len(owners) == 50
+
+
+# ---------------------------------------------------------------------------
+# gather_device_entries() - skip branches and the fake-MAC fallback
+# ---------------------------------------------------------------------------
+
+
+def test_gather_device_entries_skips_invalid_hwaddr_empty_ips_and_placeholder_ip():
+ devices = [
+ {"hwaddr": "00:00:00:00:00:00", "ips": [{"ip": "10.0.0.1"}]}, # excluded hwaddr
+ {"hwaddr": "", "ips": [{"ip": "10.0.0.2"}]}, # missing hwaddr
+ {"hwaddr": "aa:bb:cc:dd:ee:01", "ips": []}, # no ips at all
+ {"hwaddr": "aa:bb:cc:dd:ee:02", "ips": [{"ip": "0.0.0.0"}]}, # only a placeholder ip
+ {"hwaddr": "ip-::", "ips": [{"ip": "10.0.0.4"}]}, # placeholder hwaddr, the ::-specific case
+ {"hwaddr": "ip-10.0.0.5", "ips": [{"ip": "10.0.0.5"}]}, # placeholder hwaddr, the general case
+ {"hwaddr": "aa:bb:cc:dd:ee:03", "ips": [{"ip": "10.0.0.3", "lastSeen": 1000}]}, # the one real entry
+ ]
+ source = MagicMock()
+ source.fetch_devices.return_value = devices
+ entries = pihole_monitor.gather_device_entries(source, consider_online=300, fake_mac=False, max_clients=500)
+ assert [e["mac"] for e in entries] == ["aa:bb:cc:dd:ee:03"]
+
+
+def test_gather_device_entries_fake_mac_fallback_for_invalid_hwaddr():
+ devices = [{"hwaddr": "not-a-real-mac", "ips": [{"ip": "10.0.0.9", "lastSeen": 1000}]}]
+ source = MagicMock()
+ source.fetch_devices.return_value = devices
+ entries = pihole_monitor.gather_device_entries(source, consider_online=300, fake_mac=True, max_clients=500)
+ assert entries[0]["mac"] == "fa:ce:00:00:00:01" # from the stubbed string_to_fake_mac
+
+
+# ---------------------------------------------------------------------------
+# PiholeSource.fetch_top_blocked_clients() - None sentinel vs. genuine {}
+# ---------------------------------------------------------------------------
+
+
+def test_fetch_top_blocked_clients_success_returns_dict():
+ source = pihole_monitor.PiholeSource("primary", "http://ph1/", "pw", True, 5)
+ source.sid = "sid"
+ payload = {"clients": [{"ip": "10.0.0.5", "count": 12}, {"ip": "10.0.0.6", "count": 0}]}
+ with patch("requests.get", return_value=_resp(payload)) as mock_get:
+ result = source.fetch_top_blocked_clients(count=123)
+ assert result == {"10.0.0.5": 12, "10.0.0.6": 0}
+ assert mock_get.call_args.kwargs["params"] == {"blocked": "true", "count": 123}
+
+
+def test_fetch_top_blocked_clients_genuine_empty_is_not_none():
+ source = pihole_monitor.PiholeSource("primary", "http://ph1/", "pw", True, 5)
+ source.sid = "sid"
+ with patch("requests.get", return_value=_resp({"clients": []})):
+ result = source.fetch_top_blocked_clients(count=500)
+ assert result == {}
+
+
+def test_fetch_top_blocked_clients_failure_returns_none_not_empty_dict():
+ import requests as real_requests
+ source = pihole_monitor.PiholeSource("primary", "http://ph1/", "pw", True, 5)
+ source.sid = "sid"
+ with patch("requests.get", side_effect=real_requests.exceptions.Timeout("slow")):
+ result = source.fetch_top_blocked_clients(count=500)
+ assert result is None
+
+
+def test_fetch_top_blocked_clients_without_session_returns_none():
+ source = pihole_monitor.PiholeSource("primary", "http://ph1/", "pw", True, 5)
+ assert source.fetch_top_blocked_clients(count=500) is None
+
+
+# ---------------------------------------------------------------------------
+# build_ip_to_mac() - a multi-IP device must not lose its other IPs
+# ---------------------------------------------------------------------------
+
+
+def _entry(mac, ip, last_seen):
+ return {"mac": mac, "ip": ip, "name": "", "macVendor": "", "lastSeen": last_seen, "is_online": True}
+
+
+def test_build_ip_to_mac_keeps_every_ip_of_a_multi_ip_device():
+ entries = [
+ _entry("aa:bb:cc:dd:ee:01", "10.0.0.5", 100),
+ _entry("aa:bb:cc:dd:ee:01", "10.0.0.6", 90), # same device, second IP, older lastSeen
+ ]
+ ip_to_mac = pihole_monitor.build_ip_to_mac(entries)
+ assert ip_to_mac == {"10.0.0.5": "aa:bb:cc:dd:ee:01", "10.0.0.6": "aa:bb:cc:dd:ee:01"}
+
+
+def test_build_ip_to_mac_freshest_mac_wins_on_ip_reassignment():
+ entries = [
+ _entry("aa:bb:cc:dd:ee:01", "10.0.0.5", 100), # older MAC on this IP
+ _entry("aa:bb:cc:dd:ee:02", "10.0.0.5", 200), # DHCP reassigned, newer
+ ]
+ ip_to_mac = pihole_monitor.build_ip_to_mac(entries)
+ assert ip_to_mac == {"10.0.0.5": "aa:bb:cc:dd:ee:02"}
+
+
+def test_build_ip_to_mac_differs_from_naive_merged_result():
+ """Regression guard for the original bug: deriving the IP map from
+ merge_device_entries()'s output (one entry per MAC) drops a device's
+ other IPs. build_ip_to_mac() must not do that."""
+ entries = [
+ _entry("aa:bb:cc:dd:ee:01", "10.0.0.5", 100),
+ _entry("aa:bb:cc:dd:ee:01", "10.0.0.6", 90),
+ ]
+ merged = pihole_monitor.merge_device_entries(entries)
+ naive_ip_to_mac = {e["ip"]: mac for mac, e in merged.items()}
+ assert naive_ip_to_mac == {"10.0.0.5": "aa:bb:cc:dd:ee:01"} # the bug: 10.0.0.6 missing
+
+ fixed_ip_to_mac = pihole_monitor.build_ip_to_mac(entries)
+ assert "10.0.0.6" in fixed_ip_to_mac
+
+
+# ---------------------------------------------------------------------------
+# main() - orchestration, with PiholeSource's network methods stubbed
+# ---------------------------------------------------------------------------
+
+
+def _device_payload(mac, ip, name="dev", vendor="Acme", last_seen=1000):
+ return {"hwaddr": mac, "macVendor": vendor, "ips": [{"ip": ip, "name": name, "lastSeen": last_seen}]}
+
+
+class _Settings(dict):
+ """get_setting_value side_effect backed by a dict, with the plugin's
+ own defaults for anything a test doesn't override."""
+
+ _DEFAULTS = {
+ "PIHOLEMON_PRIMARY_VERIFY_SSL": True,
+ "PIHOLEMON_SECONDARY_VERIFY_SSL": True,
+ "PIHOLEMON_RUN_TIMEOUT": 5,
+ "PIHOLEMON_GET_OFFLINE": False,
+ "PIHOLEMON_FAKE_MAC": False,
+ "PIHOLEMON_API_MAXCLIENTS": 500,
+ "PIHOLEMON_CONSIDER_ONLINE": 300,
+ # False in tests by default (config.json's real default is True) so
+ # a plain main() test doesn't make a live-looking requests.post call
+ # nobody asked for - tests that exercise owner lookup opt in and
+ # mock requests.post themselves.
+ "PIHOLEMON_GET_OWNER": False,
+ "GRAPHQL_PORT": 20212,
+ "API_TOKEN": None,
+ "PIHOLEMON_MULTIPLIER": 4,
+ "PIHOLEMON_MIN_BLOCKED": 1,
+ "PIHOLEMON_HISTORY_DAYS": 7,
+ "PIHOLEMON_PRIMARY_URL": "http://ph1/",
+ "PIHOLEMON_PRIMARY_PASSWORD": "pw1",
+ "PIHOLEMON_SECONDARY_URL": "",
+ "PIHOLEMON_SECONDARY_PASSWORD": "",
+ }
+
+ def __call__(self, key):
+ if key in self:
+ return self[key]
+ return self._DEFAULTS[key]
+
+
+@pytest.fixture
+def settings():
+ return _Settings()
+
+
+@pytest.fixture
+def isolated_state(tmp_path, settings):
+ """Points STATE_FILE/RESULT_FILE at a scratch dir and wires up
+ get_setting_value, for every main()-level test."""
+ with patch.object(pihole_monitor, "STATE_FILE", str(tmp_path / "state.json")), \
+ patch.object(pihole_monitor, "RESULT_FILE", str(tmp_path / "last_result.log")), \
+ patch.object(pihole_monitor, "get_setting_value", side_effect=settings), \
+ patch.object(pihole_monitor.PiholeSource, "auth", return_value=True), \
+ patch.object(pihole_monitor.PiholeSource, "deauth", return_value=None):
+ yield tmp_path
+
+
+def test_main_aggregates_blocked_counts_from_both_sources(isolated_state, settings):
+ settings["PIHOLEMON_SECONDARY_URL"] = "http://ph2/"
+ settings["PIHOLEMON_SECONDARY_PASSWORD"] = "pw2"
+
+ devices_by_label = {
+ "primary": [_device_payload("aa:bb:cc:dd:ee:01", "10.0.0.5")],
+ "secondary": [_device_payload("aa:bb:cc:dd:ee:01", "10.0.0.5")],
+ }
+ blocked_by_label = {"primary": {"10.0.0.5": 30}, "secondary": {"10.0.0.5": 15}}
+
+ def _fetch_devices(self, max_clients):
+ return devices_by_label[self.label]
+
+ def _fetch_top_blocked(self, count):
+ return blocked_by_label[self.label]
+
+ with patch.object(pihole_monitor.PiholeSource, "fetch_devices", _fetch_devices), \
+ patch.object(pihole_monitor.PiholeSource, "fetch_top_blocked_clients", _fetch_top_blocked), \
+ patch.object(pihole_monitor, "Plugin_Objects") as mock_plugin_objects:
+ # last_raw=0 for both sources so each source's raw count comes
+ # straight through as its own delta (30 and 15) - this test is
+ # about summing per-source deltas, not about compute_delta() or
+ # aggregate_source_deltas()' reset handling (covered separately).
+ pihole_monitor.save_state({"aa:bb:cc:dd:ee:01": {"last_raw": {"primary": 0, "secondary": 0}, "history": []}})
+ assert pihole_monitor.main() == 0
+
+ instance = mock_plugin_objects.return_value
+ (call,) = instance.add_object.call_args_list
+ # Not imported twice (one device-import row) and its blocked counts
+ # from both instances are summed, not compared/overwritten.
+ assert call.kwargs["primaryId"] == "aa:bb:cc:dd:ee:01"
+ assert call.kwargs["watched3"] == "45"
+
+
+def test_main_stats_complete_false_when_a_source_fetch_fails(isolated_state, settings):
+ """A failed top_clients fetch must not write a false zero into a
+ device's history, and must not evaluate an anomaly this run."""
+ device = [_device_payload("aa:bb:cc:dd:ee:01", "10.0.0.5")]
+
+ with patch.object(pihole_monitor.PiholeSource, "fetch_devices", return_value=device), \
+ patch.object(pihole_monitor.PiholeSource, "fetch_top_blocked_clients", return_value=None), \
+ patch.object(pihole_monitor, "Plugin_Objects") as mock_plugin_objects:
+ # Seed a history (and a last_raw reference point) so a baseline
+ # exists and would trip the multiplier if (incorrectly) evaluated
+ # against a written-in zero. Timestamp 1 matches the stubbed
+ # timeNowUTC's default "now" (see module stub), so nothing is
+ # trimmed by the day-window here - not what this test is about.
+ pihole_monitor.save_state({"aa:bb:cc:dd:ee:01": {"last_raw": {"primary": 1000}, "history": [[1, 40], [1, 42], [1, 38]]}})
+ assert pihole_monitor.main() == 0
+
+ instance = mock_plugin_objects.return_value
+ (call,) = instance.add_object.call_args_list
+ assert call.kwargs["watched4"] == "normal" # not "anomaly" - stats were incomplete
+
+ state_after = pihole_monitor.load_state()
+ # Untouched, including last_raw - no false delta or reference-point
+ # update from a run whose data was incomplete.
+ assert state_after["aa:bb:cc:dd:ee:01"] == {"last_raw": {"primary": 1000}, "history": [[1, 40], [1, 42], [1, 38]]}
+
+
+def test_main_records_anomaly_when_stats_are_complete(isolated_state, settings):
+ settings["PIHOLEMON_MULTIPLIER"] = 2
+ settings["PIHOLEMON_MIN_BLOCKED"] = 5
+ device = [_device_payload("aa:bb:cc:dd:ee:01", "10.0.0.5")]
+
+ with patch.object(pihole_monitor.PiholeSource, "fetch_devices", return_value=device), \
+ patch.object(pihole_monitor.PiholeSource, "fetch_top_blocked_clients", return_value={"10.0.0.5": 50}), \
+ patch.object(pihole_monitor, "Plugin_Objects") as mock_plugin_objects:
+ # last_raw=0 so this run's raw count (50) is also its delta -
+ # baseline avg 10, 50 >> 2x.
+ pihole_monitor.save_state({"aa:bb:cc:dd:ee:01": {"last_raw": {"primary": 0}, "history": [[1, 10], [1, 10], [1, 10]]}})
+ assert pihole_monitor.main() == 0
+
+ instance = mock_plugin_objects.return_value
+ (call,) = instance.add_object.call_args_list
+ assert call.kwargs["watched4"] == "anomaly"
+
+ state_after = pihole_monitor.load_state()
+ assert state_after["aa:bb:cc:dd:ee:01"] == {"last_raw": {"primary": 50}, "history": [[1, 10], [1, 10], [1, 10], [1, 50]]}
+
+
+_DAY = 86400
+_FIXED_NOW = 2_000_000 # arbitrary fixed epoch, for deterministic age-based trimming
+
+
+def _fixed_now_mock():
+ now = MagicMock()
+ now.timestamp.return_value = _FIXED_NOW
+ return now
+
+
+@pytest.mark.parametrize(
+ ("configured_days", "expected_history"),
+ [
+ # Seed has samples aged 10, 3, and 1 days; a new one lands at age 0.
+ (-5, [[_FIXED_NOW - 1 * _DAY, 10], [_FIXED_NOW, 40]]), # negative - clamps to 1 day, only the freshest old sample survives
+ (0, [[_FIXED_NOW - 3 * _DAY, 20], [_FIXED_NOW - 1 * _DAY, 10], [_FIXED_NOW, 40]]), # falsy - falls back to 7 via `or`, drops only the 10-day-old sample
+ (1, [[_FIXED_NOW - 1 * _DAY, 10], [_FIXED_NOW, 40]]), # explicit 1 - same cutoff as the clamped negative case
+ (7, [[_FIXED_NOW - 3 * _DAY, 20], [_FIXED_NOW - 1 * _DAY, 10], [_FIXED_NOW, 40]]), # the documented default - same as the 0/fallback case
+ (15, [[_FIXED_NOW - 10 * _DAY, 30], [_FIXED_NOW - 3 * _DAY, 20], [_FIXED_NOW - 1 * _DAY, 10], [_FIXED_NOW, 40]]), # wide enough - nothing trimmed
+ ],
+)
+def test_main_history_days_clamps_and_trims_by_age(isolated_state, settings, configured_days, expected_history):
+ """Distinct, ordered seed values at distinct known ages (not len() alone)
+ so a wrong cutoff - e.g. a 1-day clamp that actually kept the 3-day-old
+ sample too, which a bare len() check would miss - shows up as a
+ mismatch. Also guards against day/second unit mixups (a classic
+ `history_days` vs `history_days * 86400` bug) since the exact surviving
+ ages are asserted, not just a count."""
+ settings["PIHOLEMON_HISTORY_DAYS"] = configured_days
+ device = [_device_payload("aa:bb:cc:dd:ee:01", "10.0.0.5")]
+ # last_raw=0 so this run's raw count (40) is also its delta - this test
+ # is about the day-based trim/clamp, not about compute_delta() itself.
+ seed = {"aa:bb:cc:dd:ee:01": {"last_raw": {"primary": 0}, "history": [
+ [_FIXED_NOW - 10 * _DAY, 30],
+ [_FIXED_NOW - 3 * _DAY, 20],
+ [_FIXED_NOW - 1 * _DAY, 10],
+ ]}}
+
+ with patch.object(pihole_monitor, "timeNowUTC", return_value=_fixed_now_mock()), \
+ patch.object(pihole_monitor.PiholeSource, "fetch_devices", return_value=device), \
+ patch.object(pihole_monitor.PiholeSource, "fetch_top_blocked_clients", return_value={"10.0.0.5": 40}), \
+ patch.object(pihole_monitor, "Plugin_Objects"):
+ pihole_monitor.save_state(seed)
+ assert pihole_monitor.main() == 0
+
+ history = pihole_monitor.load_state()["aa:bb:cc:dd:ee:01"]["history"]
+ assert history == expected_history
+
+
+def test_main_history_days_baseline_uses_only_samples_inside_the_window():
+ """The day-window must also gate the baseline itself, not just what
+ gets persisted - an old, out-of-window sample must not silently drag
+ the average up or down."""
+ stale = [_FIXED_NOW - 30 * _DAY, 1000] # far outside any sane window
+ fresh = [_FIXED_NOW - 1 * _DAY, 10]
+ history = pihole_monitor.trim_history([stale, fresh], _FIXED_NOW, history_days=7)
+ assert history == [fresh] # the stale, high-value sample must be gone
+
+
+# ---------------------------------------------------------------------------
+# compute_delta() - Pi-hole's raw cumulative-since-FTL-started count turned
+# into a real per-run increment (see its docstring for why this matters).
+# ---------------------------------------------------------------------------
+
+
+def test_compute_delta_none_when_never_seen_before():
+ assert pihole_monitor.compute_delta(None, 500) is None
+
+
+def test_compute_delta_none_when_counter_went_backwards():
+ """Pi-hole/FTL restarted (or the device dropped out of top_clients) -
+ current_raw < last_raw must not produce a negative delta."""
+ assert pihole_monitor.compute_delta(1000, 5) is None
+
+
+def test_compute_delta_returns_the_real_increment():
+ assert pihole_monitor.compute_delta(100, 150) == 50
+
+
+def test_compute_delta_zero_is_a_real_value_not_none():
+ """No new blocked queries since last run is a genuine 0, distinct from
+ None ('we can't tell this run') - a caller conflating them would either
+ silently drop a legitimate quiet period or treat it as untrustworthy."""
+ delta = pihole_monitor.compute_delta(100, 100)
+ assert delta == 0
+ assert delta is not None
+
+
+# ---------------------------------------------------------------------------
+# aggregate_source_deltas() - per-source deltas summed independently, so one
+# source's counter reset can't net out against real traffic on another.
+# ---------------------------------------------------------------------------
+
+
+def test_aggregate_source_deltas_sums_valid_deltas_from_every_source():
+ delta, updated = pihole_monitor.aggregate_source_deltas(
+ {"primary": 100, "secondary": 200},
+ {"primary": 150, "secondary": 250},
+ )
+ assert delta == 100 # 50 + 50
+ assert updated == {"primary": 150, "secondary": 250}
+
+
+def test_aggregate_source_deltas_reset_source_does_not_mask_the_others_spike():
+ """Regression guard for the exact bug CodeRabbit flagged: combining raw
+ totals across sources before diffing would let a reset on one source
+ net against real growth on another (primary +2000, secondary resetting
+ 1000->5 would combine into a raw delta of only 1005). Diffing each
+ source first and summing only the valid deltas must instead surface
+ the primary's full 2000, with the secondary contributing nothing this
+ run (not a corrective -995)."""
+ delta, updated = pihole_monitor.aggregate_source_deltas(
+ {"primary": 1000, "secondary": 1000},
+ {"primary": 3000, "secondary": 5}, # secondary: FTL restarted, counter reset
+ )
+ assert delta == 2000 # primary's real delta only, not 3000-1000+5-1000=1005
+ assert updated == {"primary": 3000, "secondary": 5} # both re-anchored regardless
+
+
+def test_aggregate_source_deltas_none_when_every_source_is_invalid():
+ delta, updated = pihole_monitor.aggregate_source_deltas(
+ {}, # nothing seen before - every source is bootstrapping
+ {"primary": 100, "secondary": 200},
+ )
+ assert delta is None
+ assert updated == {"primary": 100, "secondary": 200}
+
+
+def test_aggregate_source_deltas_source_absent_this_run_keeps_its_old_last_raw():
+ """A source that authenticated last run but not this one (or whose
+ fetch failed) shouldn't have its reference point touched - only
+ sources actually present in raw_by_source are updated."""
+ delta, updated = pihole_monitor.aggregate_source_deltas(
+ {"primary": 100, "secondary": 200},
+ {"primary": 150}, # secondary absent this run
+ )
+ assert delta == 50 # primary only
+ assert updated == {"primary": 150, "secondary": 200} # secondary untouched
+
+
+def test_main_dual_source_reset_does_not_mask_the_others_spike(isolated_state, settings):
+ """Integration-level version of the same regression: a real spike on
+ the primary instance must not be diluted by a simultaneous counter
+ reset on the secondary, when both instances report the same device."""
+ settings["PIHOLEMON_SECONDARY_URL"] = "http://ph2/"
+ settings["PIHOLEMON_SECONDARY_PASSWORD"] = "pw2"
+ settings["PIHOLEMON_MULTIPLIER"] = 2
+ settings["PIHOLEMON_MIN_BLOCKED"] = 100
+
+ devices_by_label = {
+ "primary": [_device_payload("aa:bb:cc:dd:ee:01", "10.0.0.5")],
+ "secondary": [_device_payload("aa:bb:cc:dd:ee:01", "10.0.0.5")],
+ }
+ # primary: real spike (1000 -> 3000). secondary: FTL restarted (1000 -> 5).
+ blocked_by_label = {"primary": {"10.0.0.5": 3000}, "secondary": {"10.0.0.5": 5}}
+
+ def _fetch_devices(self, max_clients):
+ return devices_by_label[self.label]
+
+ def _fetch_top_blocked(self, count):
+ return blocked_by_label[self.label]
+
+ with patch.object(pihole_monitor.PiholeSource, "fetch_devices", _fetch_devices), \
+ patch.object(pihole_monitor.PiholeSource, "fetch_top_blocked_clients", _fetch_top_blocked), \
+ patch.object(pihole_monitor, "Plugin_Objects") as mock_plugin_objects:
+ pihole_monitor.save_state({"aa:bb:cc:dd:ee:01": {
+ "last_raw": {"primary": 1000, "secondary": 1000},
+ "history": [[1, 50], [1, 50]], # baseline avg 50
+ }})
+ assert pihole_monitor.main() == 0
+
+ instance = mock_plugin_objects.return_value
+ (call,) = instance.add_object.call_args_list
+ # The real signal (2000), not the raw-combined-first result (1005).
+ assert call.kwargs["watched3"] == "2000"
+ assert call.kwargs["watched4"] == "anomaly"
+
+ state_after = pihole_monitor.load_state()["aa:bb:cc:dd:ee:01"]
+ assert state_after["last_raw"] == {"primary": 3000, "secondary": 5}
+ assert state_after["history"][-1] == [1, 2000]
+
+
+def test_main_tolerates_pre_per_source_state_instead_of_crashing(isolated_state, settings):
+ """Before this round, last_raw was a single number, not a per-source
+ dict. A state file saved by that older version must not crash this
+ version - it's treated the same as no prior reference point (every
+ source bootstraps fresh this run) rather than raising."""
+ device = [_device_payload("aa:bb:cc:dd:ee:01", "10.0.0.5")]
+
+ with patch.object(pihole_monitor.PiholeSource, "fetch_devices", return_value=device), \
+ patch.object(pihole_monitor.PiholeSource, "fetch_top_blocked_clients", return_value={"10.0.0.5": 500}), \
+ patch.object(pihole_monitor, "Plugin_Objects") as mock_plugin_objects:
+ # Legacy shape: last_raw is a plain int, not {"primary": ...}.
+ pihole_monitor.save_state({"aa:bb:cc:dd:ee:01": {"last_raw": 1234, "history": [[1, 10], [1, 10]]}})
+ assert pihole_monitor.main() == 0 # must not raise
+
+ instance = mock_plugin_objects.return_value
+ (call,) = instance.add_object.call_args_list
+ assert call.kwargs["watched4"] == "normal" # bootstrapping again, not an anomaly
+
+ state_after = pihole_monitor.load_state()["aa:bb:cc:dd:ee:01"]
+ assert state_after["last_raw"] == {"primary": 500} # re-anchored in the new shape
+ assert state_after["history"] == [[1, 10], [1, 10]] # old baseline history untouched
+
+
+def test_main_bootstrap_run_sets_last_raw_without_recording_a_delta(isolated_state, settings):
+ """The first time a device is ever seen, there's no prior raw count to
+ diff against - this run must establish the reference point (last_raw)
+ for the next run, without fabricating a delta or evaluating an anomaly
+ off one."""
+ device = [_device_payload("aa:bb:cc:dd:ee:01", "10.0.0.5")]
+
+ with patch.object(pihole_monitor.PiholeSource, "fetch_devices", return_value=device), \
+ patch.object(pihole_monitor.PiholeSource, "fetch_top_blocked_clients", return_value={"10.0.0.5": 5000}), \
+ patch.object(pihole_monitor, "Plugin_Objects") as mock_plugin_objects:
+ assert pihole_monitor.main() == 0 # no prior save_state() call - genuinely first-ever run
+
+ instance = mock_plugin_objects.return_value
+ (call,) = instance.add_object.call_args_list
+ assert call.kwargs["watched4"] == "normal" # never an anomaly on a bootstrap run
+ assert "unknown" in call.kwargs["extra"]
+
+ state_after = pihole_monitor.load_state()
+ assert state_after["aa:bb:cc:dd:ee:01"] == {"last_raw": {"primary": 5000}, "history": []}
+
+
+def test_main_counter_reset_updates_last_raw_without_touching_history(isolated_state, settings):
+ """A Pi-hole/FTL restart resets the raw counter, so this run's raw value
+ can come back lower than what was last seen. That must reset the
+ reference point for future deltas, but not corrupt the existing
+ baseline history with a bogus negative or wrap-around delta."""
+ device = [_device_payload("aa:bb:cc:dd:ee:01", "10.0.0.5")]
+
+ with patch.object(pihole_monitor.PiholeSource, "fetch_devices", return_value=device), \
+ patch.object(pihole_monitor.PiholeSource, "fetch_top_blocked_clients", return_value={"10.0.0.5": 5}), \
+ patch.object(pihole_monitor, "Plugin_Objects") as mock_plugin_objects:
+ pihole_monitor.save_state({"aa:bb:cc:dd:ee:01": {"last_raw": {"primary": 1000}, "history": [[1, 10], [1, 10]]}})
+ assert pihole_monitor.main() == 0
+
+ instance = mock_plugin_objects.return_value
+ (call,) = instance.add_object.call_args_list
+ assert call.kwargs["watched4"] == "normal"
+
+ state_after = pihole_monitor.load_state()
+ # last_raw re-anchored to the post-restart value; the pre-restart
+ # baseline history is preserved exactly, not wiped or corrupted.
+ assert state_after["aa:bb:cc:dd:ee:01"] == {"last_raw": {"primary": 5}, "history": [[1, 10], [1, 10]]}
+
+
+def test_main_returns_1_when_no_source_is_configured(isolated_state, settings):
+ settings["PIHOLEMON_PRIMARY_URL"] = ""
+ settings["PIHOLEMON_PRIMARY_PASSWORD"] = ""
+ assert pihole_monitor.main() == 1
+
+
+def test_main_marks_stats_incomplete_when_a_source_fails_to_authenticate(tmp_path, settings):
+ """Also exercises the CONSIDER_ONLINE fallback (a non-int setting falls
+ back to 300) alongside the per-source auth-failure branch, which needs
+ per-label auth behavior rather than the isolated_state fixture's
+ blanket auth=True."""
+ settings["PIHOLEMON_SECONDARY_URL"] = "http://ph2/"
+ settings["PIHOLEMON_SECONDARY_PASSWORD"] = "badpw"
+ settings["PIHOLEMON_CONSIDER_ONLINE"] = "not-a-number"
+
+ def _auth(self):
+ return self.label == "primary" # secondary fails to authenticate
+
+ device = [_device_payload("aa:bb:cc:dd:ee:01", "10.0.0.5")]
+
+ with patch.object(pihole_monitor, "STATE_FILE", str(tmp_path / "state.json")), \
+ patch.object(pihole_monitor, "RESULT_FILE", str(tmp_path / "last_result.log")), \
+ patch.object(pihole_monitor, "get_setting_value", side_effect=settings), \
+ patch.object(pihole_monitor.PiholeSource, "auth", _auth), \
+ patch.object(pihole_monitor.PiholeSource, "deauth", return_value=None), \
+ patch.object(pihole_monitor.PiholeSource, "fetch_devices", return_value=device), \
+ patch.object(pihole_monitor.PiholeSource, "fetch_top_blocked_clients", return_value={"10.0.0.5": 999}), \
+ patch.object(pihole_monitor, "Plugin_Objects") as mock_plugin_objects:
+ pihole_monitor.save_state({"aa:bb:cc:dd:ee:01": {"last_raw": {"primary": 5}, "history": [[1, 1], [1, 1], [1, 1]]}})
+ assert pihole_monitor.main() == 0
+
+ instance = mock_plugin_objects.return_value
+ (call,) = instance.add_object.call_args_list
+ assert call.kwargs["watched4"] == "normal" # secondary's auth failure marks stats incomplete
+ # Untouched, including last_raw - state.
+ assert pihole_monitor.load_state()["aa:bb:cc:dd:ee:01"] == {"last_raw": {"primary": 5}, "history": [[1, 1], [1, 1], [1, 1]]}
+
+
+def test_main_links_offline_device_resolves_owner_skips_invalid_mac_and_tracks_unknown_ip(isolated_state, settings):
+ """One run covering four branches at once: an offline device still
+ gets its blocked traffic linked to its real MAC (not a bare IP), an
+ online device gets its devOwner resolved via GraphQL, a device with an
+ invalid hardware address is skipped entirely, and blocked traffic on
+ an IP no device was ever seen on falls back to being tracked under
+ that bare IP."""
+ settings["PIHOLEMON_GET_OWNER"] = True
+ settings["API_TOKEN"] = "tok"
+
+ now = MagicMock()
+ now.timestamp.return_value = 2_000_000
+
+ devices = [
+ {"hwaddr": "aa:bb:cc:dd:ee:01", "macVendor": "Acme",
+ "ips": [{"ip": "10.0.0.1", "name": "online-dev", "lastSeen": 2_000_000 - 10}]}, # online
+ {"hwaddr": "aa:bb:cc:dd:ee:02", "macVendor": "Acme",
+ "ips": [{"ip": "10.0.0.2", "name": "offline-dev", "lastSeen": 2_000_000 - 10_000}]}, # offline
+ {"hwaddr": "not-a-real-mac", "macVendor": "Acme",
+ "ips": [{"ip": "10.0.0.3", "name": "bad-mac-dev", "lastSeen": 2_000_000 - 10}]}, # invalid MAC
+ ]
+ blocked = {"10.0.0.1": 5, "10.0.0.2": 5, "10.0.0.99": 5} # .99: never any device's IP
+ owner_resp = _resp({"data": {"devices": {"devices": [
+ {"devMac": "aa:bb:cc:dd:ee:01", "devOwner": "Mauricio"}
+ ]}}})
+
+ with patch.object(pihole_monitor, "timeNowUTC", return_value=now), \
+ patch.object(pihole_monitor.PiholeSource, "fetch_devices", return_value=devices), \
+ patch.object(pihole_monitor.PiholeSource, "fetch_top_blocked_clients", return_value=blocked), \
+ patch("requests.post", return_value=owner_resp), \
+ patch.object(pihole_monitor, "Plugin_Objects") as mock_plugin_objects:
+ assert pihole_monitor.main() == 0
+
+ instance = mock_plugin_objects.return_value
+ calls_by_primary_id = {c.kwargs["primaryId"]: c for c in instance.add_object.call_args_list}
+
+ assert "owner: Mauricio" in calls_by_primary_id["aa:bb:cc:dd:ee:01"].kwargs["extra"]
+ assert calls_by_primary_id["aa:bb:cc:dd:ee:02"].kwargs["foreignKey"] == "aa:bb:cc:dd:ee:02"
+ assert "not-a-real-mac" not in calls_by_primary_id
+ assert calls_by_primary_id["10.0.0.99"].kwargs["foreignKey"] == "null"
+
+
+def test_main_flags_anomaly_against_an_all_zero_baseline(isolated_state, settings):
+ """Regression guard: baseline == 0.0 is falsy in Python, so a naive
+ `bool(... and baseline and ...)` check would silently exempt a device
+ with a real, all-zero history - exactly the device most worth flagging
+ the first time it blocks anything at all."""
+ settings["PIHOLEMON_MULTIPLIER"] = 4
+ settings["PIHOLEMON_MIN_BLOCKED"] = 1
+ device = [_device_payload("aa:bb:cc:dd:ee:01", "10.0.0.5")]
+
+ with patch.object(pihole_monitor.PiholeSource, "fetch_devices", return_value=device), \
+ patch.object(pihole_monitor.PiholeSource, "fetch_top_blocked_clients", return_value={"10.0.0.5": 5}), \
+ patch.object(pihole_monitor, "Plugin_Objects") as mock_plugin_objects:
+ # last_raw=0 so this run's raw count (5) is also its delta.
+ pihole_monitor.save_state({"aa:bb:cc:dd:ee:01": {"last_raw": {"primary": 0}, "history": [[1, 0], [1, 0], [1, 0]]}}) # genuinely never blocked before
+ assert pihole_monitor.main() == 0
+
+ instance = mock_plugin_objects.return_value
+ (call,) = instance.add_object.call_args_list
+ assert call.kwargs["watched4"] == "anomaly"
+ assert "avg=0.0" in call.kwargs["extra"]
+ assert "ratio=" not in call.kwargs["extra"] # dividing by a zero baseline is skipped, not attempted
+
+
+def test_main_applies_independent_verify_ssl_per_instance(isolated_state, settings):
+ """PIHOLEMON_PRIMARY_VERIFY_SSL and PIHOLEMON_SECONDARY_VERIFY_SSL must
+ reach each instance independently - a self-signed secondary shouldn't
+ force verification off (or on) for the primary too."""
+ settings["PIHOLEMON_SECONDARY_URL"] = "http://ph2/"
+ settings["PIHOLEMON_SECONDARY_PASSWORD"] = "pw2"
+ settings["PIHOLEMON_PRIMARY_VERIFY_SSL"] = True
+ settings["PIHOLEMON_SECONDARY_VERIFY_SSL"] = False
+
+ seen_verify_ssl = {}
+
+ def _auth(self):
+ seen_verify_ssl[self.label] = self.verify_ssl
+ return True
+
+ with patch.object(pihole_monitor.PiholeSource, "auth", _auth), \
+ patch.object(pihole_monitor.PiholeSource, "fetch_devices", return_value=[]), \
+ patch.object(pihole_monitor.PiholeSource, "fetch_top_blocked_clients", return_value={}), \
+ patch.object(pihole_monitor, "Plugin_Objects"):
+ assert pihole_monitor.main() == 0
+
+ assert seen_verify_ssl == {"primary": True, "secondary": False}