mirror of
https://github.com/jokob-sk/NetAlertX.git
synced 2026-09-12 22:25:49 -04:00
Merge pull request #1765 from mauricio-camayo/add-pihole-monitor-plugin
Add pihole_monitor plugin: combined Pi-hole device import + query anomaly detection
This commit is contained in:
6 files changed
+2302
-1
No files matched your search
+19
-1
@@ -1,6 +1,6 @@
|
||||
# Integration with Pi-hole
|
||||
|
||||
NetAlertX includes three plugins for integrating with an existing Pi-hole installation. The first plugin imports devices through the Pi-hole v6 API, the second parses the `dhcp.leases` file generated by Pi-hole, and the third reads the Pi-hole SQLite database directly. You can use any of these approaches individually or combine them with each other and other [plugins](/docs/PLUGINS.md).
|
||||
NetAlertX includes four plugins for integrating with an existing Pi-hole installation. The first plugin imports devices through the Pi-hole v6 API, the second parses the `dhcp.leases` file generated by Pi-hole, the third reads the Pi-hole SQLite database directly, and the fourth flags devices with a blocked-query spike (a common malware/compromised-device signature) and, alongside that, imports devices from one or two Pi-hole v6 instances at once. You can use any of these approaches individually or combine them with each other and other [plugins](/docs/PLUGINS.md).
|
||||
|
||||
## Approach 1: `PIHOLEAPI` Plugin - Import devices directly from the Pi-hole v6 API
|
||||
|
||||
@@ -75,4 +75,22 @@ Check the [PIHOLE plugin README](https://github.com/netalertx/NetAlertX/tree/mai
|
||||
|
||||
---
|
||||
|
||||
## Approach 4: `PIHOLEMON` Plugin - Blocked-query anomaly detection (plus device import from one or two Pi-hole v6 instances)
|
||||
|
||||
This plugin's main job is different from the other three above: it watches each device's own blocked-query count and flags it when that count spikes well above its recent average - a common signature of malware or a compromised device beaconing out to blocklisted domains, not just device discovery. Bundled into the same connection is a second job, the same idea as `PIHOLEAPI`'s device import, extended to a primary **and** an optional secondary/failover Pi-hole instance under one set of settings.
|
||||
|
||||
**How it differs from `PIHOLEAPI`:**
|
||||
|
||||
* **Anomaly detection is the point, not a bonus.** `PIHOLEAPI` only imports devices. This plugin's reason to exist is watching a device you *already* know for a sudden change in its own behavior - device import is the second, supporting job, not the main one.
|
||||
* **Two Pi-hole instances, not one.** A primary and an optional secondary/failover are both checked and their results combined under one set of settings - useful if you run more than one Pi-hole (e.g. two resolvers for redundancy) and want a single device list and a single anomaly baseline across both, instead of a blind spot on whichever instance isn't being watched.
|
||||
* **Choose `PIHOLEAPI`** if you only run one Pi-hole and just want device import - it's the simpler, more focused option. **Choose `PIHOLEMON`** if you want the anomaly detection, or run two Pi-holes needing one configuration, or both.
|
||||
|
||||
Check the [PIHOLEMON plugin README](https://github.com/netalertx/NetAlertX/tree/main/server/plugins/pihole_monitor/) for the full settings reference, the `http://` vs `https://` trade-off, and troubleshooting.
|
||||
|
||||
### docker-compose changes
|
||||
|
||||
No changes are required - this plugin only talks to Pi-hole's API, the same as `PIHOLEAPI`.
|
||||
|
||||
---
|
||||
|
||||
Explore other [plugins](/docs/PLUGINS.md) to discover additional information about your network, or learn how to scan [remote networks](./REMOTE_NETWORKS.md).
|
||||
@@ -78,6 +78,7 @@ Device-detecting plugins insert values into the `CurrentScan` database table. T
|
||||
| `OMDSDNOPENAPI` | [omada_sdn_openapi](https://github.com/netalertx/NetAlertX/tree/main/server/plugins/omada_sdn_openapi/) | 📥/🆎 | OMADA TP-Link import via OpenAPI | 🖧 | |
|
||||
| `PIHOLE` | [pihole_scan](https://github.com/netalertx/NetAlertX/tree/main/server/plugins/pihole_scan/) | 🆎/📥 | Pi-hole device import & sync | | |
|
||||
| `PIHOLEAPI` | [pihole_api_scan](https://github.com/netalertx/NetAlertX/tree/main/server/plugins/pihole_api_scan/) | 🆎/📥 | Pi-hole device import & sync via API v6+ | | |
|
||||
| `PIHOLEMON` | [pihole_monitor](https://github.com/netalertx/NetAlertX/tree/main/server/plugins/pihole_monitor/) | 🆎/📥 | Blocked-query anomaly detection (includes primary and secondary DNS import from Pi-hole) | | |
|
||||
| `PUSHSAFER` | [_publisher_pushsafer](https://github.com/netalertx/NetAlertX/tree/main/server/plugins/_publisher_pushsafer/) | ▶️ | Pushsafer notifications | | |
|
||||
| `PUSHOVER` | [_publisher_pushover](https://github.com/netalertx/NetAlertX/tree/main/server/plugins/_publisher_pushover/) | ▶️ | Pushover notifications | | |
|
||||
| `RSTIMPRT` | [rest_import](https://github.com/netalertx/NetAlertX/tree/main/server/plugins/rest_import/) | 📥/🆎 | Import via a REST API endpoint | 🖧 | |
|
||||
|
||||
@@ -0,0 +1,128 @@
|
||||
## Overview - PIHOLEMON Plugin — Pi-hole Monitor
|
||||
|
||||
The **PIHOLEMON** plugin does two jobs against the same Pi-hole connection(s):
|
||||
|
||||
1. **Query anomaly detection** - flags a device whose *blocked*-query count spikes well above its own recent average, the classic signature of malware or a compromised device beaconing out to blocklisted domains. This is deliberately not the same thing as device discovery: it's about a device you already know suddenly behaving differently, not a new device showing up.
|
||||
2. **Device import** - same job as the official **PIHOLEAPI** (`pihole_api_scan`) plugin: pulls the device list from Pi-hole and feeds it into NetAlertX's normal device-scanner pipeline, so a device Pi-hole knows about but NetAlertX doesn't gets created automatically.
|
||||
|
||||
Both share one login per Pi-hole instance and one settings page, instead of being two separately configured pieces that happen to need the same credentials.
|
||||
|
||||
Two design choices worth knowing about:
|
||||
|
||||
* **Both a primary and an optional secondary/failover Pi-hole are checked, and their results combined.** Watching only one leaves an obvious blind spot for the anomaly detection - a device can simply point at the other resolver and never show up. Leave the secondary URL blank if you only run one Pi-hole; most setups do.
|
||||
* **The anomaly baseline is keyed by MAC address, not IP.** DHCP-assigned IPs change; since this plugin already has the device list from Pi-hole itself, it resolves IP to MAC from that same data - no separate lookup needed for that part. A device owner, if you use NetAlertX's `devOwner` field, is looked up via NetAlertX's own GraphQL API purely to make the anomaly label friendlier; it's optional and never blocks import or detection if unavailable.
|
||||
|
||||
### Why not just run two copies of PIHOLEAPI for two Pi-holes?
|
||||
|
||||
PIHOLEAPI doesn't support running two independent instances against different Pi-holes - both copies would read and write the same settings keys. PIHOLEMON was built to support a primary and secondary instance natively from the start.
|
||||
|
||||
### Quick setup guide
|
||||
|
||||
* You are running **Pi-hole v6** or newer on every instance you configure (this plugin uses `/api/auth`, `/api/network/devices`, and `/api/stats/top_clients`, none of which exist in v5).
|
||||
* An **App Password** is generated on each Pi-hole (`Settings → Web Interface / API → App Password`) - recommended over using the admin login password directly.
|
||||
* Like every non-core plugin, **When to run** (`PIHOLEMON_RUN`) defaults to `disabled` - set it to `schedule` (or another option) once your URL/password are filled in, or nothing runs.
|
||||
|
||||
#### 🔒 A note on `http://` vs `https://`
|
||||
|
||||
Most home Pi-hole setups (including the one this plugin was developed and tested against) run over plain `http://` on a trusted LAN, and that's what the examples below use - this plugin doesn't require `https://` or refuse an `http://` URL. Know the trade-off either way, though: over `http://`, the App Password/admin password is sent in cleartext on every run, readable by anything else that can see that network segment (a compromised device, a hostile guest network, etc.). If your Pi-hole's admin interface is reachable from anywhere less trusted than your own LAN, either put it behind `https://` (Pi-hole's own self-signed cert, or a reverse proxy with a real one) or keep it LAN-only. **Verify SSL** only matters once an instance is on `https://` - it's on by default, per instance (primary and secondary can each be on `http://` or `https://` independently), and turning it off to tolerate a self-signed cert accepts *any* certificate, including an attacker's; installing that self-signed CA as trusted on the machine running NetAlertX is the safer way to use a self-signed cert if you need one.
|
||||
|
||||
### Usage
|
||||
|
||||
- Head to **Settings** > **Pi-hole Monitor** to fill in the values below.
|
||||
|
||||
| Setting Key | Description |
|
||||
| ---------------------------------- | --------------------------------------------------------------------------------------------------- |
|
||||
| **PIHOLEMON_PRIMARY_URL** | Required. URL to your primary Pi-hole, e.g. `http://192.168.1.10/`. |
|
||||
| **PIHOLEMON_PRIMARY_PASSWORD** | App Password (or admin password) for the primary Pi-hole. |
|
||||
| **PIHOLEMON_PRIMARY_VERIFY_SSL** | Verify TLS certificates on the *primary* instance's `https://` URL. Default **on**. Only disable if that Pi-hole uses a self-signed certificate you can't install as trusted - see the security note below. |
|
||||
| **PIHOLEMON_SECONDARY_URL** | Optional. URL to a secondary/failover Pi-hole, e.g. `http://192.168.1.11:8080/`. Leave blank if you only run one. |
|
||||
| **PIHOLEMON_SECONDARY_PASSWORD** | Only used if a secondary URL is set. |
|
||||
| **PIHOLEMON_SECONDARY_VERIFY_SSL** | Same, for the *secondary* instance. Only used if a secondary URL is set. Default **on**. |
|
||||
| **PIHOLEMON_GET_OFFLINE** | Import devices even if not recently seen. Default off. |
|
||||
| **PIHOLEMON_CONSIDER_ONLINE** | Seconds since last seen to still count a device as online. Default `300`. |
|
||||
| **PIHOLEMON_API_MAXCLIENTS** | Maximum devices requested **per instance**'s device list. Default `500`. |
|
||||
| **PIHOLEMON_FAKE_MAC** | Generate a fake MAC from the IP for devices with a non-standard hardware address. Default off. |
|
||||
| **PIHOLEMON_GET_OWNER** | Look up an already-known device's owner for a friendlier anomaly label. Default **on**. Uses this app's own **GRAPHQL_PORT** and **API_TOKEN** settings (Settings → General) automatically - nothing else to configure. Disable if you don't use device owners. |
|
||||
| **PIHOLEMON_MULTIPLIER** | Flag a device when the blocked queries it generated *since the last run* exceed this many times its own recent per-run average. Default `4`. |
|
||||
| **PIHOLEMON_MIN_BLOCKED** | Ignore devices below this many blocked queries since the last run, even if the multiplier is exceeded. Default `20` - depends heavily on your Schedule (see the note below the settings table). |
|
||||
| **PIHOLEMON_HISTORY_DAYS** | How many days of recent runs to keep per device for the rolling baseline. A real time window, not a sample count - means the same thing regardless of your Schedule setting, and a faster schedule just adds more data points inside it. Default `7` (one week). |
|
||||
| **Watched** *(standard NetAlertX setting)* | Which columns count as "changed" for notification purposes. Defaults to `watchedValue4` (the anomaly/normal flag) only - the per-run blocked-query delta changes every run by design. |
|
||||
|
||||
#### A note on the blocked-query count: per-run delta, not a running total
|
||||
|
||||
Pi-hole's `/api/stats/top_clients` returns a count that's cumulative since Pi-hole's FTL service last started - it does not reset daily, and it's not a "since I last checked" delta. This plugin diffs each run's raw count against the one from its last run to get a real per-run increment, which is what `PIHOLEMON_MULTIPLIER`/`PIHOLEMON_MIN_BLOCKED` actually compare against - comparing the raw cumulative totals directly would make any device's traffic look like a growing "anomaly" purely from the counter never resetting. The very first run for a device, and any run right after Pi-hole/FTL restarts (the counter resets, so the raw count can drop below what was last seen), can't produce a valid delta - those runs establish a new reference point instead of evaluating an anomaly.
|
||||
| **Report on** *(standard NetAlertX setting)* | Which statuses actually notify. Defaults to `watched-changed` only, so you hear about it exactly when a device flips into (or out of) an anomaly. |
|
||||
|
||||
This plugin does **not** send notifications on its own - it relies on NetAlertX's own core, which diffs the columns picked in **Watched** between runs and, on a match against **Report on**, dispatches through whichever publisher(s) you already have enabled under **Settings → Notifications** (ntfy, Apprise, email, Telegram, ...).
|
||||
|
||||
One extra setting is required for that to actually reach you: NetAlertX's **Notification Processing** plugin (`NTFPRCS`) has its own **"Notify on"** setting (`NTFPRCS_INCLUDED_SECTIONS`), and its default value - `new_devices`, `down_devices`, `events` - does **not** include `plugins`. Without `plugins` in that list, this plugin's Watched/Report on matches are recorded correctly but never make it into a notification. Add `plugins` to `NTFPRCS_INCLUDED_SECTIONS` once, and it also covers any other plugin using the same mechanism, not just this one.
|
||||
|
||||
The default text notification for the `plugins` section is a generic vertical `Header: Value` dump. For something more readable, set `NTFPRCS`'s **"Text Template: Plugins"** (`NTFPRCS_TEXT_TEMPLATE_plugins`) to something like:
|
||||
|
||||
```
|
||||
{objectPrimaryId} [{watchedValue2}] @ {objectSecondaryId} → {watchedValue4} ({watchedValue3} blocked)
|
||||
```
|
||||
|
||||
Only the columns the `plugins` section actually selects are available as placeholders (`{plugin}`, `{objectPrimaryId}`, `{objectSecondaryId}`, `{dateTimeChanged}`, `{watchedValue1-4}`, `{status}`) - `extra` (where this plugin's `owner`/`ratio` detail lives) isn't one of them, so that richer text is only visible on the device's page in NetAlertX, not in the notification itself. This template setting is global to `NTFPRCS`, so it affects any plugin using Watched/Report on, not just this one.
|
||||
|
||||
If the same device (same MAC) is seen on both instances, the entry with the more recent "last seen" timestamp wins - it isn't imported twice, and its blocked-query counts from both instances are summed, not compared separately.
|
||||
|
||||
### Testing the notification pipeline end-to-end
|
||||
|
||||
Waiting for real, organic beaconing traffic to confirm notifications actually arrive isn't practical. The clean way to force a `normal` → `anomaly` transition on demand, without touching any internal state file by hand:
|
||||
|
||||
1. Temporarily set `PIHOLEMON_MULTIPLIER` to something like `1.01` and `PIHOLEMON_MIN_BLOCKED` to `1`.
|
||||
2. Run the plugin (wait for its schedule, or trigger it manually from the UI).
|
||||
3. Any device with a baseline and *any* blocked traffic should now flip to `anomaly` - confirming the whole chain: Watched/Report on match → `Plugins_Events` row → NetAlertX's Notification Processing (`NTFPRCS`, needs `plugins` in `NTFPRCS_INCLUDED_SECTIONS`, see above) → your configured publisher.
|
||||
4. Set `PIHOLEMON_MULTIPLIER`/`PIHOLEMON_MIN_BLOCKED` back to their real values afterward - left at the test values, everything with any traffic at all reads as an anomaly.
|
||||
|
||||
### ⚠️ Troubleshooting
|
||||
|
||||
---
|
||||
|
||||
#### ❌ Authentication failed / no data from a Pi-hole instance
|
||||
|
||||
* Confirm the URL includes the scheme (`http://`/`https://`) and, if not on the default port, the port too - e.g. `http://192.168.1.10/` ✔, `http://192.168.1.10/admin` ❌.
|
||||
* Confirm that instance is running **Pi-hole v6**, not v5.
|
||||
* SSL verification matches your setup (disable for self-signed certificates).
|
||||
* Try the App Password by hand first: `curl -X POST <url>/api/auth -d '{"password":"<app password>"}'` should return a `session.sid`.
|
||||
* Check the plugin log for `[PIHOLEMON] <label>: ...` lines - `label` is `primary` or `secondary`, so you can tell which instance is the problem.
|
||||
|
||||
---
|
||||
|
||||
#### ❌ Some devices are missing
|
||||
|
||||
* Devices with an invalid MAC are skipped unless **Generate fake MAC** is enabled - turning it on trades data consistency for coverage (the "MAC" becomes a stand-in derived from the IP, not a real hardware address).
|
||||
* Offline devices don't get a device-import row (name/vendor) unless **Import offline devices** is enabled - but their MAC is still used to attribute blocked-query counts correctly, so an offline-but-active device never falls back to being tracked by bare IP.
|
||||
|
||||
---
|
||||
|
||||
#### ❌ Notifications don't arrive even though an anomaly was detected
|
||||
|
||||
* Check NetAlertX's own **Notification Processing** settings - the `Notify on` setting (`NTFPRCS_INCLUDED_SECTIONS`) must include `plugins`, or Watched/Report on matches never reach a notification (see the note above `Watched`/`Report on` in the settings table).
|
||||
* Confirm you actually have a publisher enabled and working under **Settings → Notifications** (ntfy, Apprise, email, ...) - this plugin never sends anything itself, so a working publisher there is a prerequisite, not optional.
|
||||
|
||||
---
|
||||
|
||||
#### ❌ Anomaly label missing an owner
|
||||
|
||||
* NetAlertX doesn't have `devOwner` set for that device yet, or the GraphQL lookup isn't reachable - neither blocks anomaly detection, the label is just less friendly.
|
||||
* Confirm `PIHOLEMON_GET_OWNER` is enabled, and that this app's own **GRAPHQL_PORT**/**API_TOKEN** settings (Settings → General) are correct, if you expect owners to resolve.
|
||||
|
||||
---
|
||||
|
||||
#### ❌ Getting flagged for normal usage / not getting flagged for a real spike
|
||||
|
||||
* Both `PIHOLEMON_MULTIPLIER` and `PIHOLEMON_MIN_BLOCKED` are starting values, not tuned defaults - let it run for a week and adjust based on what's actually normal for your network. They apply *per run*, so if you change `PIHOLEMON_RUN_SCHD` to something much less frequent than the default, `PIHOLEMON_MIN_BLOCKED` in particular may need raising (fewer, larger runs mean more blocked queries naturally accumulate in each one).
|
||||
* `PIHOLEMON_HISTORY_DAYS` is a retention window, not a detection delay. A new device becomes evaluable after its 3rd successful run: the 1st anchors the raw counter, the 2nd records the first delta into the baseline history, and the 3rd is the first one with a baseline to compare against.
|
||||
* A device is also never flagged on the one run right after it's first seen, or right after Pi-hole/FTL restarts - see the delta note above the settings table.
|
||||
|
||||
### Notes
|
||||
|
||||
- Device-import parsing (online/offline handling, fake-MAC fallback, endpoints used) mirrors the official `pihole_api_scan` plugin - this isn't meant to reinvent that half, only to support a second Pi-hole instance without forking it.
|
||||
- This plugin never blocks, unblocks, or otherwise changes anything on either Pi-hole - it only reads.
|
||||
- The rolling per-device history is kept in its own `state.PIHOLEMON.json` file in NetAlertX's DB folder (survives upgrades, unlike the log folder) - safe to delete if you want to reset the baseline from scratch. The per-run `last_result` file stays in this plugin's log directory as usual.
|
||||
|
||||
- Version: 1.0.2
|
||||
- Author: [mauricio-camayo](https://github.com/mauricio-camayo/)
|
||||
- Release Date: `2026-08-31`
|
||||
@@ -0,0 +1,574 @@
|
||||
{
|
||||
"code_name": "pihole_monitor",
|
||||
"unique_prefix": "PIHOLEMON",
|
||||
"plugin_type": "device_scanner",
|
||||
"execution_order": "Layer_0",
|
||||
"enabled": true,
|
||||
"data_source": "script",
|
||||
"mapped_to_table": "CurrentScan",
|
||||
"data_filters": [
|
||||
{
|
||||
"compare_column": "objectPrimaryId",
|
||||
"compare_operator": "==",
|
||||
"compare_field_id": "txtMacFilter",
|
||||
"compare_js_template": "'{value}'.toString()",
|
||||
"compare_use_quotes": true
|
||||
}
|
||||
],
|
||||
"show_ui": true,
|
||||
"localized": ["display_name", "description", "icon"],
|
||||
"display_name": [
|
||||
{
|
||||
"language_code": "en_us",
|
||||
"string": "Pi-hole Monitor"
|
||||
}
|
||||
],
|
||||
"description": [
|
||||
{
|
||||
"language_code": "en_us",
|
||||
"string": "Flags devices with a spike in blocked DNS queries - a common malware/compromised-device signature - and imports devices from one or two Pi-hole v6 instances."
|
||||
}
|
||||
],
|
||||
"icon": [
|
||||
{
|
||||
"language_code": "en_us",
|
||||
"string": "<i class=\"fa-solid fa-shield-halved\"></i>"
|
||||
}
|
||||
],
|
||||
"params": [],
|
||||
"settings": [
|
||||
{
|
||||
"function": "RUN",
|
||||
"events": ["run"],
|
||||
"type": {
|
||||
"dataType": "string",
|
||||
"elements": [
|
||||
{ "elementType": "select", "elementOptions": [], "transformers": [] }
|
||||
]
|
||||
},
|
||||
"default_value": "disabled",
|
||||
"options": [
|
||||
"disabled",
|
||||
"once",
|
||||
"schedule",
|
||||
"always_after_scan",
|
||||
"on_new_device",
|
||||
"on_notification"
|
||||
],
|
||||
"localized": ["name", "description"],
|
||||
"name": [{ "language_code": "en_us", "string": "When to run" }],
|
||||
"description": [
|
||||
{
|
||||
"language_code": "en_us",
|
||||
"string": "<code>schedule</code> is recommended over <code>always_after_scan</code>."
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
"function": "RUN_SCHD",
|
||||
"type": {
|
||||
"dataType": "string",
|
||||
"elements": [
|
||||
{
|
||||
"elementType": "span",
|
||||
"elementOptions": [
|
||||
{ "cssClasses": "input-group-addon validityCheck" },
|
||||
{ "getStringKey": "Gen_ValidIcon" }
|
||||
],
|
||||
"transformers": []
|
||||
},
|
||||
{
|
||||
"elementType": "input",
|
||||
"elementOptions": [
|
||||
{ "focusout": "validateRegex(this)" },
|
||||
{ "base64Regex": "Xig/OlwqfCg/OlswLTldfFsxLTVdWzAtOV18WzAtOV0rLVswLTldKyg/Oi9bMC05XSspP3xcKi9bMC05XSspKSg/OiwoPzpbMC05XXxbMS01XVswLTldfFswLTldKy1bMC05XSsoPzovWzAtOV0rKT98XCovWzAtOV0rKSkqXHMrKD86XCp8KD86WzAtOV18MVswLTldfDJbMC0zXXxbMC05XSstWzAtOV0rKD86L1swLTldKyk/fFwqL1swLTldKykpKD86LCg/OlswLTldfDFbMC05XXwyWzAtM118WzAtOV0rLVswLTldKyg/Oi9bMC05XSspP3xcKi9bMC05XSspKSpccysoPzpcKnwoPzpbMS05XXxbMTJdWzAtOV18M1swMV18WzAtOV0rLVswLTldKyg/Oi9bMC05XSspP3xcKi9bMC05XSspKSg/OiwoPzpbMS05XXxbMTJdWzAtOV18M1swMV18WzAtOV0rLVswLTldKyg/Oi9bMC05XSspP3xcKi9bMC05XSspKSpccysoPzpcKnwoPzpbMS05XXwxWzAtMl18WzAtOV0rLVswLTldKyg/Oi9bMC05XSspP3xcKi9bMC05XSspKSg/OiwoPzpbMS05XXwxWzAtMl18WzAtOV0rLVswLTldKyg/Oi9bMC05XSspP3xcKi9bMC05XSspKSpccysoPzpcKnwoPzpbMC02XXxbMC02XS1bMC02XSg/Oi9bMC05XSspP3xcKi9bMC05XSspKSg/OiwoPzpbMC02XXxbMC02XS1bMC02XSg/Oi9bMC05XSspP3xcKi9bMC05XSspKSok" }
|
||||
],
|
||||
"transformers": []
|
||||
}
|
||||
]
|
||||
},
|
||||
"default_value": "*/5 * * * *",
|
||||
"options": [],
|
||||
"localized": ["name", "description"],
|
||||
"name": [{ "language_code": "en_us", "string": "Schedule" }],
|
||||
"description": [
|
||||
{ "language_code": "en_us", "string": "Only used when <code>RUN</code> is <code>schedule</code>. Default: every 5 minutes." }
|
||||
]
|
||||
},
|
||||
{
|
||||
"function": "SET_ALWAYS",
|
||||
"type": {
|
||||
"dataType": "array",
|
||||
"elements": [
|
||||
{ "elementType": "select", "elementOptions": [{ "multiple": "true", "orderable": "true" }], "transformers": [] }
|
||||
]
|
||||
},
|
||||
"default_value": ["devMac", "devLastIP", "devName", "devVendor"],
|
||||
"options": ["devMac", "devLastIP", "devName", "devVendor"],
|
||||
"localized": ["name", "description"],
|
||||
"name": [{ "language_code": "en_us", "string": "Set always columns" }],
|
||||
"description": [
|
||||
{
|
||||
"language_code": "en_us",
|
||||
"string": "These columns are treated as authoritative and overwrite existing values (including those set by other plugins), unless the current value was set by the user (<code>Source = USER</code> or <code>LOCKED</code>)."
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
"function": "SET_EMPTY",
|
||||
"type": {
|
||||
"dataType": "array",
|
||||
"elements": [
|
||||
{ "elementType": "select", "elementOptions": [{ "multiple": "true", "orderable": "true" }], "transformers": [] }
|
||||
]
|
||||
},
|
||||
"default_value": [],
|
||||
"options": ["devMac", "devLastIP", "devName", "devVendor", "devType", "devSourcePlugin"],
|
||||
"localized": ["name", "description"],
|
||||
"name": [{ "language_code": "en_us", "string": "Set empty columns" }],
|
||||
"description": [
|
||||
{ "language_code": "en_us", "string": "These columns are only overwritten if they are empty or their Source is <code>NEWDEV</code>." }
|
||||
]
|
||||
},
|
||||
{
|
||||
"function": "PRIMARY_URL",
|
||||
"type": {
|
||||
"dataType": "string",
|
||||
"elements": [
|
||||
{ "elementType": "input", "elementOptions": [], "transformers": [] }
|
||||
]
|
||||
},
|
||||
"default_value": "",
|
||||
"options": [],
|
||||
"localized": ["name", "description"],
|
||||
"name": [{ "language_code": "en_us", "string": "Primary Pi-hole URL" }],
|
||||
"description": [
|
||||
{ "language_code": "en_us", "string": "Required. URL to your primary Pi-hole instance, for example <code>http://pi.hole:8080/</code>." }
|
||||
]
|
||||
},
|
||||
{
|
||||
"function": "PRIMARY_PASSWORD",
|
||||
"type": {
|
||||
"dataType": "string",
|
||||
"elements": [
|
||||
{ "elementType": "input", "elementOptions": [{ "type": "password" }], "transformers": [] }
|
||||
]
|
||||
},
|
||||
"default_value": "",
|
||||
"options": [],
|
||||
"localized": ["name", "description"],
|
||||
"name": [{ "language_code": "en_us", "string": "Primary Pi-hole password" }],
|
||||
"description": [
|
||||
{ "language_code": "en_us", "string": "Web UI password or App Password for the primary Pi-hole. A dedicated App Password is recommended over the admin login password." }
|
||||
]
|
||||
},
|
||||
{
|
||||
"function": "PRIMARY_VERIFY_SSL",
|
||||
"type": {
|
||||
"dataType": "boolean",
|
||||
"elements": [
|
||||
{ "elementType": "input", "elementOptions": [{ "type": "checkbox" }], "transformers": [] }
|
||||
]
|
||||
},
|
||||
"default_value": true,
|
||||
"options": [],
|
||||
"localized": ["name", "description"],
|
||||
"name": [{ "language_code": "en_us", "string": "Primary: Verify SSL" }],
|
||||
"description": [
|
||||
{
|
||||
"language_code": "en_us",
|
||||
"string": "Only relevant for an <code>https://</code> primary URL. Disabling this accepts any certificate, including one from an attacker on the network - install the self-signed CA as trusted instead if possible. Leave on unless you understand the trade-off."
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
"function": "SECONDARY_URL",
|
||||
"type": {
|
||||
"dataType": "string",
|
||||
"elements": [
|
||||
{ "elementType": "input", "elementOptions": [], "transformers": [] }
|
||||
]
|
||||
},
|
||||
"default_value": "",
|
||||
"options": [],
|
||||
"localized": ["name", "description"],
|
||||
"name": [{ "language_code": "en_us", "string": "Secondary Pi-hole URL (optional)" }],
|
||||
"description": [
|
||||
{
|
||||
"language_code": "en_us",
|
||||
"string": "Optional. Leave empty if you only run one Pi-hole - most setups do. If set, this second instance (typically a failover/secondary resolver) is imported and its blocked-query counts combined with the primary's, so a device can't dodge detection by using the less-watched resolver, e.g. <code>http://192.168.1.11:8080/</code>."
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
"function": "SECONDARY_PASSWORD",
|
||||
"type": {
|
||||
"dataType": "string",
|
||||
"elements": [
|
||||
{ "elementType": "input", "elementOptions": [{ "type": "password" }], "transformers": [] }
|
||||
]
|
||||
},
|
||||
"default_value": "",
|
||||
"options": [],
|
||||
"localized": ["name", "description"],
|
||||
"name": [{ "language_code": "en_us", "string": "Secondary Pi-hole password (optional)" }],
|
||||
"description": [
|
||||
{ "language_code": "en_us", "string": "Only used if Secondary Pi-hole URL is set." }
|
||||
]
|
||||
},
|
||||
{
|
||||
"function": "SECONDARY_VERIFY_SSL",
|
||||
"type": {
|
||||
"dataType": "boolean",
|
||||
"elements": [
|
||||
{ "elementType": "input", "elementOptions": [{ "type": "checkbox" }], "transformers": [] }
|
||||
]
|
||||
},
|
||||
"default_value": true,
|
||||
"options": [],
|
||||
"localized": ["name", "description"],
|
||||
"name": [{ "language_code": "en_us", "string": "Secondary: Verify SSL" }],
|
||||
"description": [
|
||||
{
|
||||
"language_code": "en_us",
|
||||
"string": "Only relevant for an <code>https://</code> secondary URL. Same trade-off as the primary instance's Verify SSL - independent because primary and secondary can each be on http:// or https:// separately."
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
"function": "GET_OFFLINE",
|
||||
"type": {
|
||||
"dataType": "boolean",
|
||||
"elements": [
|
||||
{ "elementType": "input", "elementOptions": [{ "type": "checkbox" }], "transformers": [] }
|
||||
]
|
||||
},
|
||||
"default_value": false,
|
||||
"options": [],
|
||||
"localized": ["name", "description"],
|
||||
"name": [{ "language_code": "en_us", "string": "Import offline devices" }],
|
||||
"description": [
|
||||
{ "language_code": "en_us", "string": "If enabled, devices not seen recently (see below) are still imported/updated." }
|
||||
]
|
||||
},
|
||||
{
|
||||
"function": "CONSIDER_ONLINE",
|
||||
"type": {
|
||||
"dataType": "integer",
|
||||
"elements": [
|
||||
{ "elementType": "input", "elementOptions": [{ "type": "number" }], "transformers": [] }
|
||||
]
|
||||
},
|
||||
"default_value": 300,
|
||||
"options": [],
|
||||
"localized": ["name", "description"],
|
||||
"name": [{ "language_code": "en_us", "string": "Online if last seen (seconds)" }],
|
||||
"description": [
|
||||
{ "language_code": "en_us", "string": "Time in seconds to consider a device online." }
|
||||
]
|
||||
},
|
||||
{
|
||||
"function": "API_MAXCLIENTS",
|
||||
"type": {
|
||||
"dataType": "integer",
|
||||
"elements": [
|
||||
{ "elementType": "input", "elementOptions": [{ "type": "number" }], "transformers": [] }
|
||||
]
|
||||
},
|
||||
"default_value": 500,
|
||||
"options": [],
|
||||
"localized": ["name", "description"],
|
||||
"name": [{ "language_code": "en_us", "string": "Max clients per instance" }],
|
||||
"description": [
|
||||
{ "language_code": "en_us", "string": "Maximum number of devices requested from each Pi-hole instance's device list." }
|
||||
]
|
||||
},
|
||||
{
|
||||
"function": "FAKE_MAC",
|
||||
"type": {
|
||||
"dataType": "boolean",
|
||||
"elements": [
|
||||
{ "elementType": "input", "elementOptions": [{ "type": "checkbox" }], "transformers": [] }
|
||||
]
|
||||
},
|
||||
"default_value": false,
|
||||
"options": [],
|
||||
"localized": ["name", "description"],
|
||||
"name": [{ "language_code": "en_us", "string": "Generate fake MAC from IP" }],
|
||||
"description": [
|
||||
{
|
||||
"language_code": "en_us",
|
||||
"string": "Devices with a non-standard hardware address are skipped unless enabled, in which case a deterministic fake MAC is derived from the IP - accept the data-consistency trade-off that implies."
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
"function": "GET_OWNER",
|
||||
"type": {
|
||||
"dataType": "boolean",
|
||||
"elements": [
|
||||
{ "elementType": "input", "elementOptions": [{ "type": "checkbox" }], "transformers": [] }
|
||||
]
|
||||
},
|
||||
"default_value": true,
|
||||
"options": [],
|
||||
"localized": ["name", "description"],
|
||||
"name": [{ "language_code": "en_us", "string": "Look up device owner" }],
|
||||
"description": [
|
||||
{
|
||||
"language_code": "en_us",
|
||||
"string": "If enabled, looks up an already-known device's owner (<code>devOwner</code>) - device import itself never depends on this. Disable if you don't use device owners or don't want the extra lookup."
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
"function": "MULTIPLIER",
|
||||
"type": {
|
||||
"dataType": "string",
|
||||
"elements": [
|
||||
{ "elementType": "input", "elementOptions": [{ "type": "number" }], "transformers": [] }
|
||||
]
|
||||
},
|
||||
"default_value": "4",
|
||||
"options": [],
|
||||
"localized": ["name", "description"],
|
||||
"name": [{ "language_code": "en_us", "string": "Anomaly multiplier" }],
|
||||
"description": [
|
||||
{ "language_code": "en_us", "string": "Flag a device when the blocked queries it generated since the last run exceed this many times its own recent per-run average. Starting value, tune after a week of real data." }
|
||||
]
|
||||
},
|
||||
{
|
||||
"function": "MIN_BLOCKED",
|
||||
"type": {
|
||||
"dataType": "string",
|
||||
"elements": [
|
||||
{ "elementType": "input", "elementOptions": [{ "type": "number" }], "transformers": [] }
|
||||
]
|
||||
},
|
||||
"default_value": "20",
|
||||
"options": [],
|
||||
"localized": ["name", "description"],
|
||||
"name": [{ "language_code": "en_us", "string": "Minimum blocked queries" }],
|
||||
"description": [
|
||||
{ "language_code": "en_us", "string": "Ignore devices below this many blocked queries since the last run, even if the multiplier is exceeded (e.g. 1 -> 5 is technically 5x but is statistical noise). Depends on your Schedule - a more frequent schedule means fewer blocked queries accumulate per run, so a lower value may be needed." }
|
||||
]
|
||||
},
|
||||
{
|
||||
"function": "HISTORY_DAYS",
|
||||
"type": {
|
||||
"dataType": "string",
|
||||
"elements": [
|
||||
{ "elementType": "input", "elementOptions": [{ "type": "number" }], "transformers": [] }
|
||||
]
|
||||
},
|
||||
"default_value": "7",
|
||||
"options": [],
|
||||
"localized": ["name", "description"],
|
||||
"name": [{ "language_code": "en_us", "string": "History window (days)" }],
|
||||
"description": [
|
||||
{ "language_code": "en_us", "string": "How many days of recent runs to keep per device for the rolling baseline. A real time window, not a sample count - it means the same thing regardless of your <code>Schedule</code> above, and a faster schedule only adds more data points inside it. Default 7 (one week)." }
|
||||
]
|
||||
},
|
||||
{
|
||||
"function": "WATCH",
|
||||
"type": {
|
||||
"dataType": "array",
|
||||
"elements": [
|
||||
{ "elementType": "select", "elementOptions": [{ "multiple": "true", "orderable": "true" }], "transformers": [] }
|
||||
]
|
||||
},
|
||||
"default_value": ["watchedValue4"],
|
||||
"options": ["watchedValue1", "watchedValue2", "watchedValue3", "watchedValue4"],
|
||||
"localized": ["name", "description"],
|
||||
"name": [{ "language_code": "en_us", "string": "Watched" }],
|
||||
"description": [
|
||||
{
|
||||
"language_code": "en_us",
|
||||
"string": "Which columns count as \"changed\" for notification purposes. Default is <code>watchedValue4</code> (the anomaly/normal flag) only - <code>watchedValue3</code> (raw blocked-query count) fluctuates every run by design and would otherwise notify constantly."
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
"function": "REPORT_ON",
|
||||
"type": {
|
||||
"dataType": "array",
|
||||
"elements": [
|
||||
{ "elementType": "select", "elementOptions": [{ "multiple": "true", "orderable": "true" }], "transformers": [] }
|
||||
]
|
||||
},
|
||||
"default_value": ["watched-changed"],
|
||||
"options": ["new", "watched-changed", "watched-not-changed", "missing-in-last-scan"],
|
||||
"localized": ["name", "description"],
|
||||
"name": [{ "language_code": "en_us", "string": "Report on" }],
|
||||
"description": [
|
||||
{
|
||||
"language_code": "en_us",
|
||||
"string": "Which statuses generate a notification, dispatched through whichever publisher(s) you already have enabled under Settings (ntfy, Apprise, email, ...) - this plugin never sends notifications on its own. Default: only <code>watched-changed</code>, i.e. when a device flips between normal and anomaly."
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
"function": "CMD",
|
||||
"type": {
|
||||
"dataType": "string",
|
||||
"elements": [
|
||||
{ "elementType": "input", "elementOptions": [{ "readonly": "true" }], "transformers": [] }
|
||||
]
|
||||
},
|
||||
"default_value": "python3 /app/server/plugins/pihole_monitor/pihole_monitor.py",
|
||||
"options": [],
|
||||
"localized": ["name", "description"],
|
||||
"name": [{ "language_code": "en_us", "string": "Command" }],
|
||||
"description": [{ "language_code": "en_us", "string": "Command to run. This can not be changed." }]
|
||||
},
|
||||
{
|
||||
"function": "RUN_TIMEOUT",
|
||||
"type": {
|
||||
"dataType": "integer",
|
||||
"elements": [
|
||||
{ "elementType": "input", "elementOptions": [{ "type": "number" }], "transformers": [] }
|
||||
]
|
||||
},
|
||||
"default_value": 60,
|
||||
"options": [],
|
||||
"localized": ["name", "description"],
|
||||
"name": [{ "language_code": "en_us", "string": "Run timeout" }],
|
||||
"description": [
|
||||
{ "language_code": "en_us", "string": "Maximum time in seconds to wait for the script to finish (covers device list + stats for up to two Pi-hole instances, plus optional owner lookups)." }
|
||||
]
|
||||
}
|
||||
],
|
||||
"database_column_definitions": [
|
||||
{
|
||||
"column": "index",
|
||||
"css_classes": "col-sm-1",
|
||||
"show": true,
|
||||
"type": "none",
|
||||
"default_value": "",
|
||||
"options": [],
|
||||
"localized": ["name"],
|
||||
"name": [{ "language_code": "en_us", "string": "Index" }]
|
||||
},
|
||||
{
|
||||
"column": "objectPrimaryId",
|
||||
"mapped_to_column": "scanMac",
|
||||
"css_classes": "col-sm-2",
|
||||
"show": true,
|
||||
"type": "device_name_mac",
|
||||
"default_value": "",
|
||||
"options": [],
|
||||
"localized": ["name"],
|
||||
"name": [{ "language_code": "en_us", "string": "Device" }]
|
||||
},
|
||||
{
|
||||
"column": "objectSecondaryId",
|
||||
"mapped_to_column": "scanLastIP",
|
||||
"css_classes": "col-sm-2",
|
||||
"show": true,
|
||||
"type": "device_ip",
|
||||
"default_value": "",
|
||||
"options": [],
|
||||
"localized": ["name"],
|
||||
"name": [{ "language_code": "en_us", "string": "IP" }]
|
||||
},
|
||||
{
|
||||
"column": "watchedValue1",
|
||||
"mapped_to_column": "scanName",
|
||||
"css_classes": "col-sm-2",
|
||||
"show": true,
|
||||
"type": "label",
|
||||
"default_value": "",
|
||||
"options": [],
|
||||
"localized": ["name"],
|
||||
"name": [{ "language_code": "en_us", "string": "Name" }]
|
||||
},
|
||||
{
|
||||
"column": "watchedValue2",
|
||||
"mapped_to_column": "scanVendor",
|
||||
"css_classes": "col-sm-2",
|
||||
"show": true,
|
||||
"type": "label",
|
||||
"default_value": "",
|
||||
"options": [],
|
||||
"localized": ["name"],
|
||||
"name": [{ "language_code": "en_us", "string": "Vendor" }]
|
||||
},
|
||||
{
|
||||
"column": "watchedValue3",
|
||||
"css_classes": "col-sm-1",
|
||||
"show": true,
|
||||
"type": "label",
|
||||
"default_value": "",
|
||||
"options": [],
|
||||
"localized": ["name"],
|
||||
"name": [{ "language_code": "en_us", "string": "Blocked queries" }]
|
||||
},
|
||||
{
|
||||
"column": "watchedValue4",
|
||||
"css_classes": "col-sm-1",
|
||||
"show": true,
|
||||
"type": "replace",
|
||||
"default_value": "",
|
||||
"options": [
|
||||
{
|
||||
"equals": "anomaly",
|
||||
"replacement": "<div style='text-align:center'><i class='fa-solid fa-triangle-exclamation'></i></div>"
|
||||
},
|
||||
{
|
||||
"equals": "normal",
|
||||
"replacement": "<div style='text-align:center'><i class='fa-solid fa-circle-check'></i></div>"
|
||||
}
|
||||
],
|
||||
"localized": ["name"],
|
||||
"name": [{ "language_code": "en_us", "string": "Anomaly?" }]
|
||||
},
|
||||
{
|
||||
"column": "extra",
|
||||
"mapped_to_column": "scanSourcePlugin",
|
||||
"css_classes": "col-sm-3",
|
||||
"show": true,
|
||||
"type": "label",
|
||||
"default_value": "",
|
||||
"options": [],
|
||||
"localized": ["name"],
|
||||
"name": [{ "language_code": "en_us", "string": "Detail" }]
|
||||
},
|
||||
{
|
||||
"column": "dateTimeChanged",
|
||||
"css_classes": "col-sm-2",
|
||||
"show": true,
|
||||
"type": "label",
|
||||
"default_value": "",
|
||||
"options": [],
|
||||
"localized": ["name"],
|
||||
"name": [{ "language_code": "en_us", "string": "Last checked" }]
|
||||
},
|
||||
{
|
||||
"column": "status",
|
||||
"css_classes": "col-sm-1",
|
||||
"show": true,
|
||||
"type": "replace",
|
||||
"default_value": "",
|
||||
"options": [
|
||||
{
|
||||
"equals": "watched-not-changed",
|
||||
"replacement": "<div style='text-align:center'><i class='fa-solid fa-square-check'></i><div></div>"
|
||||
},
|
||||
{
|
||||
"equals": "watched-changed",
|
||||
"replacement": "<div style='text-align:center'><i class='fa-solid fa-triangle-exclamation'></i></div>"
|
||||
},
|
||||
{
|
||||
"equals": "new",
|
||||
"replacement": "<div style='text-align:center'><i class='fa-solid fa-circle-plus'></i></div>"
|
||||
},
|
||||
{
|
||||
"equals": "missing-in-last-scan",
|
||||
"replacement": "<div style='text-align:center'><i class='fa-solid fa-question'></i></div>"
|
||||
}
|
||||
],
|
||||
"localized": ["name"],
|
||||
"name": [{ "language_code": "en_us", "string": "Status" }]
|
||||
}
|
||||
]
|
||||
}
|
||||
@@ -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-<address>" 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())
|
||||
@@ -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}
|
||||
Reference in new issue
Block a user