Merge pull request #1873 from mauricio-camayo/fix/piholemon-bare-ip-as-mac

PIHOLEMON: never write a bare IP as a device MAC

server/plugins/pihole_monitor/README.md

@@ -96,6 +96,7 @@ Waiting for real, organic beaconing traffic to confirm notifications actually ar
 
 * 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.
+* Blocked traffic from an IP Pi-hole never tied to any MAC is written out under a fake MAC derived from that IP when **Generate fake MAC** is enabled; otherwise it's only logged, with no result row, so it can't create a device whose "MAC" is the IP.
 
 ---
 

server/plugins/pihole_monitor/pihole_monitor.py

@@ -543,14 +543,25 @@ def main():
 
     # 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
+    # associated with any MAC (not even an offline one) is keyed by a fake
-    # tracked under its own IP, so the signal isn't silently dropped.
+    # MAC derived from that IP when FAKE_MAC is on - same as device import -
+    # or else tracked under the bare IP for anomaly logging only, never
+    # written out as a result row (see the results loop). key_to_ip keeps the
+    # source IP behind each key (the busiest one, for a multi-IP MAC) so a
+    # row with no device-import entry still reports an IP, not its MAC.
     blocked_by_mac_by_source = {}
+    key_to_ip = {}
+    key_ip_counts = {}
     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)
+            key = ip_to_mac.get(ip)
+            if key is None:
+                key = string_to_fake_mac(ip) if fake_mac else ip
             mac_counts[key] = mac_counts.get(key, 0) + count
+            if count > key_ip_counts.get(key, -1):
+                key_ip_counts[key] = count
+                key_to_ip[key] = ip
         blocked_by_mac_by_source[label] = mac_counts
 
     if not stats_complete:
@@ -635,26 +646,32 @@ def main():
                 extra=detail,
                 foreignKey=str(device['mac']),
             )
-        else:
+        elif mac:
-            # No device-import row this run for `key` - either it's a real,
+            # No device-import row this run for `key`, but it's still a MAC:
-            # known MAC that's just offline-filtered above (still link the
+            # either a real, known device that's just offline-filtered above
-            # anomaly to that device's existing page via foreignKey), or a
+            # (link the anomaly to its existing page via foreignKey), or the
-            # bare IP Pi-hole has never associated with any MAC at all
+            # FAKE_MAC stand-in for an IP Pi-hole never associated with a MAC.
-            # (nothing to link to, foreignKey stays 'null').
-            known_mac = key if is_mac(key) else None
             plugin_objects.add_object(
                 primaryId=key,
-                secondaryId=key,
+                secondaryId=key_to_ip.get(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',
+                foreignKey=key,
             )
+        else:
+            # A bare IP with FAKE_MAC off. Results map primaryId to
+            # CurrentScan.scanMac, so writing it out would create a bogus
+            # device whose "MAC" is the IP. Keep tracking its history so an
+            # anomaly still gets logged below, but don't emit a row.
+            mylog('verbose', [f"[{pluginName}] No MAC for {key} (see Generate fake MAC setting) - not writing a result row"])
 
         if is_anomaly:
-            mylog('none', [f'[{pluginName}] Anomaly: {key} - {detail}'])
+            source_ip = key_to_ip.get(key)
+            who = f"{key} ({source_ip})" if source_ip and source_ip != key else key
+            mylog('none', [f'[{pluginName}] Anomaly: {who} - {detail}'])
 
         if stats_complete:
             # Always reset each source's diff reference point, even on a

test/plugins/test_pihole_monitor.py

@@ -815,8 +815,9 @@ def test_main_links_offline_device_resolves_owner_skips_invalid_mac_and_tracks_u
     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
+    an IP no device was ever seen on (FAKE_MAC off) is not written out,
-    that bare IP."""
+    since primaryId maps to CurrentScan.scanMac and would create a bogus
+    device whose "MAC" is the IP."""
     settings["PIHOLEMON_GET_OWNER"] = True
     settings["API_TOKEN"] = "tok"
 
@@ -848,8 +849,52 @@ def test_main_links_offline_device_resolves_owner_skips_invalid_mac_and_tracks_u
 
     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 calls_by_primary_id["aa:bb:cc:dd:ee:02"].kwargs["secondaryId"] == "10.0.0.2"  # maps to scanLastIP - an IP, not the MAC
     assert "not-a-real-mac" not in calls_by_primary_id
-    assert calls_by_primary_id["10.0.0.99"].kwargs["foreignKey"] == "null"
+    assert "10.0.0.99" not in calls_by_primary_id
+
+
+def test_main_unknown_ip_uses_fake_mac_when_enabled(isolated_state, settings):
+    """With FAKE_MAC on, blocked traffic on an IP Pi-hole never tied to a
+    MAC is written out under the fake MAC derived from that IP (same as
+    device import), never under the bare IP."""
+    settings["PIHOLEMON_FAKE_MAC"] = True
+
+    now = MagicMock()
+    now.timestamp.return_value = 2_000_000
+
+    with patch.object(pihole_monitor, "timeNowUTC", return_value=now), \
+         patch.object(pihole_monitor.PiholeSource, "fetch_devices", return_value=[]), \
+         patch.object(pihole_monitor.PiholeSource, "fetch_top_blocked_clients", return_value={"10.0.0.99": 5}), \
+         patch.object(pihole_monitor, "Plugin_Objects") as mock_plugin_objects:
+        assert pihole_monitor.main() == 0
+
+    calls_by_primary_id = {c.kwargs["primaryId"]: c for c in mock_plugin_objects.return_value.add_object.call_args_list}
+
+    assert "10.0.0.99" not in calls_by_primary_id
+    assert calls_by_primary_id["fa:ce:00:00:00:01"].kwargs["foreignKey"] == "fa:ce:00:00:00:01"  # stubbed string_to_fake_mac
+    assert calls_by_primary_id["fa:ce:00:00:00:01"].kwargs["secondaryId"] == "10.0.0.99"  # maps to scanLastIP - the source IP, not the fake MAC
+
+
+def test_main_anomaly_log_names_the_source_ip_for_a_fake_mac(isolated_state, settings):
+    """A fake MAC alone doesn't tell anyone which machine to look at - the
+    anomaly log line has to carry the source IP it was derived from."""
+    settings["PIHOLEMON_FAKE_MAC"] = True
+    settings["PIHOLEMON_MULTIPLIER"] = 4
+    settings["PIHOLEMON_MIN_BLOCKED"] = 1
+
+    with patch.object(pihole_monitor.PiholeSource, "fetch_devices", return_value=[]), \
+         patch.object(pihole_monitor.PiholeSource, "fetch_top_blocked_clients", return_value={"10.0.0.99": 5}), \
+         patch.object(pihole_monitor, "Plugin_Objects"), \
+         patch.object(pihole_monitor, "mylog") as mock_mylog:
+        # last_raw=0 so this run's raw count (5) is also its delta.
+        pihole_monitor.save_state({"fa:ce:00:00:00:01": {"last_raw": {"primary": 0}, "history": [[1, 0], [1, 0], [1, 0]]}})
+        assert pihole_monitor.main() == 0
+
+    anomaly_lines = [str(c.args[1]) for c in mock_mylog.call_args_list if "Anomaly:" in str(c.args[1])]
+    assert len(anomaly_lines) == 1
+    assert "fa:ce:00:00:00:01" in anomaly_lines[0]
+    assert "10.0.0.99" in anomaly_lines[0]
 
 
 def test_main_flags_anomaly_against_an_all_zero_baseline(isolated_state, settings):