vh/NetAlertX
1
0
Fork 0
mirror of https://github.com/jokob-sk/NetAlertX.git synced 2026-04-15 22:51:37 -07:00
Code Issues Actions Packages Projects Releases Wiki Activity

FIX: lowercase MAC normalization across project v0.2

Browse Source 
Signed-off-by: jokob-sk <jokob.sk@gmail.com>
This commit is contained in:
jokob-sk
2026-02-07 14:02:54 +11:00
parent 946ad00253
commit 827b5d2ad3
11 changed files with 42 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files
front/network.php
+12 -10
View File
@@ -96,12 +96,14 @@
COUNT(child.devMac) AS node_ports_count COUNT(child.devMac) AS node_ports_count
FROM Devices AS parent FROM Devices AS parent
LEFT JOIN Devices AS child LEFT JOIN Devices AS child
ON child.devParentMAC = parent.devMac /* CRITICAL FIX: COLLATE NOCASE ensures the join works
WHERE parent.devType IN ( even if devParentMAC is uppercase and devMac is lowercase
${networkDeviceTypes}) */
ON child.devParentMAC = parent.devMac COLLATE NOCASE
WHERE parent.devType IN (${networkDeviceTypes})
AND parent.devIsArchived = 0 AND parent.devIsArchived = 0
GROUP BY parent.devMac, parent.devName, parent.devPresentLastScan, GROUP BY parent.devMac, parent.devName, parent.devPresentLastScan,
parent.devType, parent.devParentMAC, parent.devIcon, parent.devAlertDown parent.devType, parent.devParentMAC, parent.devIcon, parent.devAlertDown
ORDER BY parent.devName; ORDER BY parent.devName;
`; `;
@@ -381,9 +383,8 @@
} }
// ---------------------------------------------------- // ----------------------------------------------------
function loadConnectedDevices(node_mac) { function loadConnectedDevices(node_mac) {
// Standardize the input just in case
// 1. Force to lowercase to match the new DB standard
const normalized_mac = node_mac.toLowerCase(); const normalized_mac = node_mac.toLowerCase();
const sql = ` const sql = `
@@ -397,13 +398,14 @@
ELSE 'Unknown status' ELSE 'Unknown status'
END AS devStatus END AS devStatus
FROM Devices FROM Devices
WHERE devParentMac = '${normalized_mac}'`; /* Using COLLATE NOCASE here solves the 'TEXT' vs 'NOCASE' mismatch */
WHERE devParentMac = '${normalized_mac}' COLLATE NOCASE`;
const id = node_mac.replace(/:/g, '_'); // Keep the ID generation consistent
const id = normalized_mac.replace(/:/g, '_');
const wrapperHtml = ` const wrapperHtml = `
<table class="table table-bordered table-striped node-leafs-table " id="table_leafs_${id}" data-node-mac="${normalized_mac}"> <table class="table table-bordered table-striped node-leafs-table " id="table_leafs_${id}" data-node-mac="${normalized_mac}">
</table>`; </table>`;
loadDeviceTable({ loadDeviceTable({
front/plugins/plugin_helper.py
+2 -2
View File
@@ -91,11 +91,11 @@ def is_typical_router_ip(ip_address):
def is_mac(input): def is_mac(input):
input_str = str(input).lower().strip() # Convert to string and lowercase so non-string values won't raise errors input_str = str(input).lower().strip() # Convert to string and lowercase so non-string values won't raise errors
# Full MAC (6 octets) e.g. AA:BB:CC:DD:EE:FF # Full MAC (6 octets) e.g. aa:bb:cc:dd:ee:ff
full_mac_re = re.compile(r"^[0-9a-f]{2}([-:]?)[0-9a-f]{2}(\1[0-9a-f]{2}){4}$") full_mac_re = re.compile(r"^[0-9a-f]{2}([-:]?)[0-9a-f]{2}(\1[0-9a-f]{2}){4}$")
# Wildcard prefix format: exactly 3 octets followed by a trailing '*' component # Wildcard prefix format: exactly 3 octets followed by a trailing '*' component
# Examples: AA:BB:CC:* # Examples: aa:bb:cc:*
wildcard_re = re.compile(r"^[0-9a-f]{2}[-:]?[0-9a-f]{2}[-:]?[0-9a-f]{2}[-:]?\*$") wildcard_re = re.compile(r"^[0-9a-f]{2}[-:]?[0-9a-f]{2}[-:]?[0-9a-f]{2}[-:]?\*$")
if full_mac_re.match(input_str) or wildcard_re.match(input_str): if full_mac_re.match(input_str) or wildcard_re.match(input_str):
server/api_server/openapi/schemas.py
+2 -2
View File
@@ -458,7 +458,7 @@ class DeleteDevicesRequest(BaseModel):
{ {
"summary": "Delete specific devices", "summary": "Delete specific devices",
"value": { "value": {
"macs": ["AA:BB:CC:DD:EE:FF", "AA:BB:CC:DD:*"], "macs": ["aa:bb:cc:dd:ee:ff", "aa:bb:cc:dd:*"],
"confirm_delete_all": False "confirm_delete_all": False
} }
} }
@@ -570,7 +570,7 @@ class WakeOnLanResponse(BaseResponse):
output: Optional[str] = Field( output: Optional[str] = Field(
None, None,
description="Command output", description="Command output",
json_schema_extra={"examples": ["Sent magic packet to AA:BB:CC:DD:EE:FF"]} json_schema_extra={"examples": ["Sent magic packet to aa:bb:cc:dd:ee:ff"]}
) )
test/api_endpoints/test_device_endpoints.py
+2 -2
View File
@@ -28,7 +28,7 @@ def client():
@pytest.fixture @pytest.fixture
def test_mac(): def test_mac():
# Generate a unique MAC for each test run # Generate a unique MAC for each test run
return "AA:BB:CC:" + ":".join(f"{random.randint(0, 255):02X}" for _ in range(3)) return "aa:bb:cc:" + ":".join(f"{random.randint(0, 255):02X}" for _ in range(3))
def auth_headers(token): def auth_headers(token):
@@ -100,7 +100,7 @@ def test_copy_device(client, api_token, test_mac):
assert resp.status_code == 200 assert resp.status_code == 200
# Step 2: Generate a target MAC # Step 2: Generate a target MAC
target_mac = "AA:BB:CC:" + ":".join( target_mac = "aa:bb:cc:" + ":".join(
f"{random.randint(0, 255):02X}" for _ in range(3) f"{random.randint(0, 255):02X}" for _ in range(3)
) )
test/api_endpoints/test_devices_endpoints.py
+1 -1
View File
@@ -196,6 +196,6 @@ def test_devices_by_status(client, api_token, test_mac):
def test_delete_test_devices(client, api_token): def test_delete_test_devices(client, api_token):
# Delete by MAC # Delete by MAC
resp = client.delete("/devices", json={"macs": ["AA:BB:CC:*"]}, headers=auth_headers(api_token)) resp = client.delete("/devices", json={"macs": ["aa:bb:cc:*"]}, headers=auth_headers(api_token))
assert resp.status_code == 200 assert resp.status_code == 200
assert resp.json.get("success") is True assert resp.json.get("success") is True
test/api_endpoints/test_mcp_extended_endpoints.py
+4 -4
View File
@@ -78,7 +78,7 @@ def test_reset_device_props(mock_reset, client, api_token):
def test_copy_device(mock_copy, client, api_token): def test_copy_device(mock_copy, client, api_token):
"""Test POST /device/copy.""" """Test POST /device/copy."""
mock_copy.return_value = {"success": True} mock_copy.return_value = {"success": True}
payload = {"macFrom": "00:11:22:33:44:55", "macTo": "AA:BB:CC:DD:EE:FF"} payload = {"macFrom": "00:11:22:33:44:55", "macTo": "aa:bb:cc:dd:ee:ff"}
response = client.post('/device/copy', response = client.post('/device/copy',
json=payload, json=payload,
@@ -86,21 +86,21 @@ def test_copy_device(mock_copy, client, api_token):
assert response.status_code == 200 assert response.status_code == 200
assert response.get_json() == {"success": True} assert response.get_json() == {"success": True}
mock_copy.assert_called_with("00:11:22:33:44:55", "AA:BB:CC:DD:EE:FF") mock_copy.assert_called_with("00:11:22:33:44:55", "aa:bb:cc:dd:ee:ff")
@patch('models.device_instance.DeviceInstance.deleteDevices') @patch('models.device_instance.DeviceInstance.deleteDevices')
def test_delete_devices_bulk(mock_delete, client, api_token): def test_delete_devices_bulk(mock_delete, client, api_token):
"""Test DELETE /devices.""" """Test DELETE /devices."""
mock_delete.return_value = {"success": True} mock_delete.return_value = {"success": True}
payload = {"macs": ["00:11:22:33:44:55", "AA:BB:CC:DD:EE:FF"]} payload = {"macs": ["00:11:22:33:44:55", "aa:bb:cc:dd:ee:ff"]}
response = client.delete('/devices', response = client.delete('/devices',
json=payload, json=payload,
headers=auth_headers(api_token)) headers=auth_headers(api_token))
assert response.status_code == 200 assert response.status_code == 200
mock_delete.assert_called_with(["00:11:22:33:44:55", "AA:BB:CC:DD:EE:FF"]) mock_delete.assert_called_with(["00:11:22:33:44:55", "aa:bb:cc:dd:ee:ff"])
@patch('models.device_instance.DeviceInstance.deleteAllWithEmptyMacs') @patch('models.device_instance.DeviceInstance.deleteAllWithEmptyMacs')
test/api_endpoints/test_mcp_tools_endpoints.py
+13 -13
View File
@@ -30,7 +30,7 @@ def test_get_device_info_ip_partial(mock_db_conn, client, api_token):
# Mock database connection - DeviceInstance._fetchall calls conn.execute().fetchall() # Mock database connection - DeviceInstance._fetchall calls conn.execute().fetchall()
mock_conn = MagicMock() mock_conn = MagicMock()
mock_execute_result = MagicMock() mock_execute_result = MagicMock()
mock_execute_result.fetchall.return_value = [{"devName": "Test Device", "devMac": "AA:BB:CC:DD:EE:FF", "devLastIP": "192.168.1.50"}] mock_execute_result.fetchall.return_value = [{"devName": "Test Device", "devMac": "aa:bb:cc:dd:ee:ff", "devLastIP": "192.168.1.50"}]
mock_conn.execute.return_value = mock_execute_result mock_conn.execute.return_value = mock_execute_result
mock_db_conn.return_value = mock_conn mock_db_conn.return_value = mock_conn
@@ -92,7 +92,7 @@ def test_get_open_ports_ip(mock_device_db_conn, mock_plugin_db_conn, client, api
# Mock for PluginObjectInstance.getByField (returns port data) # Mock for PluginObjectInstance.getByField (returns port data)
mock_execute_result.fetchall.return_value = [{"Object_SecondaryID": "22", "Watched_Value2": "ssh"}, {"Object_SecondaryID": "80", "Watched_Value2": "http"}] mock_execute_result.fetchall.return_value = [{"Object_SecondaryID": "22", "Watched_Value2": "ssh"}, {"Object_SecondaryID": "80", "Watched_Value2": "http"}]
# Mock for DeviceInstance.getByIP (returns device with MAC) # Mock for DeviceInstance.getByIP (returns device with MAC)
mock_execute_result.fetchone.return_value = {"devMac": "AA:BB:CC:DD:EE:FF"} mock_execute_result.fetchone.return_value = {"devMac": "aa:bb:cc:dd:ee:ff"}
mock_conn.execute.return_value = mock_execute_result mock_conn.execute.return_value = mock_execute_result
mock_plugin_db_conn.return_value = mock_conn mock_plugin_db_conn.return_value = mock_conn
@@ -119,7 +119,7 @@ def test_get_open_ports_mac_resolve(mock_plugin_db_conn, client, api_token):
mock_conn.execute.return_value = mock_execute_result mock_conn.execute.return_value = mock_execute_result
mock_plugin_db_conn.return_value = mock_conn mock_plugin_db_conn.return_value = mock_conn
payload = {"target": "AA:BB:CC:DD:EE:FF"} payload = {"target": "aa:bb:cc:dd:ee:ff"}
response = client.post("/device/open_ports", json=payload, headers=auth_headers(api_token)) response = client.post("/device/open_ports", json=payload, headers=auth_headers(api_token))
assert response.status_code == 200 assert response.status_code == 200
@@ -163,7 +163,7 @@ def test_get_recent_alerts(mock_db_conn, client, api_token):
mock_conn = MagicMock() mock_conn = MagicMock()
mock_execute_result = MagicMock() mock_execute_result = MagicMock()
now = datetime.now().strftime("%Y-%m-%d %H:%M:%S") now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
mock_execute_result.fetchall.return_value = [{"eve_DateTime": now, "eve_EventType": "New Device", "eve_MAC": "AA:BB:CC:DD:EE:FF"}] mock_execute_result.fetchall.return_value = [{"eve_DateTime": now, "eve_EventType": "New Device", "eve_MAC": "aa:bb:cc:dd:ee:ff"}]
mock_conn.execute.return_value = mock_execute_result mock_conn.execute.return_value = mock_execute_result
mock_db_conn.return_value = mock_conn mock_db_conn.return_value = mock_conn
@@ -186,12 +186,12 @@ def test_set_device_alias(mock_update_col, client, api_token):
mock_update_col.return_value = {"success": True, "message": "Device alias updated"} mock_update_col.return_value = {"success": True, "message": "Device alias updated"}
payload = {"alias": "New Device Name"} payload = {"alias": "New Device Name"}
response = client.post("/device/AA:BB:CC:DD:EE:FF/set-alias", json=payload, headers=auth_headers(api_token)) response = client.post("/device/aa:bb:cc:dd:ee:ff/set-alias", json=payload, headers=auth_headers(api_token))
assert response.status_code == 200 assert response.status_code == 200
data = response.get_json() data = response.get_json()
assert data["success"] is True assert data["success"] is True
mock_update_col.assert_called_once_with("AA:BB:CC:DD:EE:FF", "devName", "New Device Name") mock_update_col.assert_called_once_with("aa:bb:cc:dd:ee:ff", "devName", "New Device Name")
@patch("models.device_instance.DeviceInstance.updateDeviceColumn") @patch("models.device_instance.DeviceInstance.updateDeviceColumn")
@@ -214,15 +214,15 @@ def test_set_device_alias_not_found(mock_update_col, client, api_token):
@patch("api_server.api_server_start.wakeonlan") @patch("api_server.api_server_start.wakeonlan")
def test_wol_wake_device(mock_wakeonlan, client, api_token): def test_wol_wake_device(mock_wakeonlan, client, api_token):
"""Test wol_wake_device.""" """Test wol_wake_device."""
mock_wakeonlan.return_value = {"success": True, "message": "WOL packet sent to AA:BB:CC:DD:EE:FF"} mock_wakeonlan.return_value = {"success": True, "message": "WOL packet sent to aa:bb:cc:dd:ee:ff"}
payload = {"devMac": "AA:BB:CC:DD:EE:FF"} payload = {"devMac": "aa:bb:cc:dd:ee:ff"}
response = client.post("/nettools/wakeonlan", json=payload, headers=auth_headers(api_token)) response = client.post("/nettools/wakeonlan", json=payload, headers=auth_headers(api_token))
assert response.status_code == 200 assert response.status_code == 200
data = response.get_json() data = response.get_json()
assert data["success"] is True assert data["success"] is True
assert "AA:BB:CC:DD:EE:FF" in data["message"] assert "aa:bb:cc:dd:ee:ff" in data["message"]
def test_wol_wake_device_invalid_mac(client, api_token): def test_wol_wake_device_invalid_mac(client, api_token):
@@ -249,7 +249,7 @@ def test_get_latest_device(mock_db_conn, client, api_token):
mock_execute_result = MagicMock() mock_execute_result = MagicMock()
mock_execute_result.fetchone.return_value = { mock_execute_result.fetchone.return_value = {
"devName": "Latest Device", "devName": "Latest Device",
"devMac": "AA:BB:CC:DD:EE:FF", "devMac": "aa:bb:cc:dd:ee:ff",
"devLastIP": "192.168.1.100", "devLastIP": "192.168.1.100",
"devFirstConnection": "2025-12-07 10:30:00", "devFirstConnection": "2025-12-07 10:30:00",
} }
@@ -262,7 +262,7 @@ def test_get_latest_device(mock_db_conn, client, api_token):
data = response.get_json() data = response.get_json()
assert len(data) >= 1, "Expected at least one device in response" assert len(data) >= 1, "Expected at least one device in response"
assert data[0]["devName"] == "Latest Device" assert data[0]["devName"] == "Latest Device"
assert data[0]["devMac"] == "AA:BB:CC:DD:EE:FF" assert data[0]["devMac"] == "aa:bb:cc:dd:ee:ff"
def test_openapi_spec(client, api_token): def test_openapi_spec(client, api_token):
@@ -293,7 +293,7 @@ def test_mcp_devices_export_csv(mock_db_conn, client, api_token):
"""Test MCP devices export in CSV format.""" """Test MCP devices export in CSV format."""
mock_conn = MagicMock() mock_conn = MagicMock()
mock_execute_result = MagicMock() mock_execute_result = MagicMock()
mock_execute_result.fetchall.return_value = [{"devMac": "AA:BB:CC:DD:EE:FF", "devName": "Test Device", "devLastIP": "192.168.1.1"}] mock_execute_result.fetchall.return_value = [{"devMac": "aa:bb:cc:dd:ee:ff", "devName": "Test Device", "devLastIP": "192.168.1.1"}]
mock_conn.execute.return_value = mock_execute_result mock_conn.execute.return_value = mock_execute_result
mock_db_conn.return_value = mock_conn mock_db_conn.return_value = mock_conn
@@ -310,7 +310,7 @@ def test_mcp_devices_export_json(mock_export, client, api_token):
"""Test MCP devices export in JSON format.""" """Test MCP devices export in JSON format."""
mock_export.return_value = { mock_export.return_value = {
"format": "json", "format": "json",
"data": [{"devMac": "AA:BB:CC:DD:EE:FF", "devName": "Test Device", "devLastIP": "192.168.1.1"}], "data": [{"devMac": "aa:bb:cc:dd:ee:ff", "devName": "Test Device", "devLastIP": "192.168.1.1"}],
"columns": ["devMac", "devName", "devLastIP"], "columns": ["devMac", "devName", "devLastIP"],
} }
test/scan/test_device_field_lock.py
+1 -1
View File
@@ -31,7 +31,7 @@ def client():
@pytest.fixture @pytest.fixture
def test_mac(): def test_mac():
"""Generate a test MAC address.""" """Generate a test MAC address."""
return "AA:BB:CC:DD:EE:FF" return "aa:bb:cc:dd:ee:ff"
@pytest.fixture @pytest.fixture
test/scan/test_ip_update_logic.py
+3 -3
View File
@@ -38,7 +38,7 @@ def test_ipv6_update_preserves_ipv4(scan_db, mock_device_handling):
# 1️⃣ Create device with IPv4 # 1️⃣ Create device with IPv4
cur.execute( cur.execute(
"INSERT INTO Devices (devMac, devLastIP, devPrimaryIPv4, devName) VALUES (?, ?, ?, ?)", "INSERT INTO Devices (devMac, devLastIP, devPrimaryIPv4, devName) VALUES (?, ?, ?, ?)",
("AA:BB:CC:DD:EE:FF", "192.168.1.10", "192.168.1.10", "Device") ("aa:bb:cc:dd:ee:ff", "192.168.1.10", "192.168.1.10", "Device")
) )
# 2️⃣ Insert a scan reporting IPv6 # 2️⃣ Insert a scan reporting IPv6
@@ -47,7 +47,7 @@ def test_ipv6_update_preserves_ipv4(scan_db, mock_device_handling):
INSERT INTO CurrentScan (scanMac, scanLastIP, scanSourcePlugin, scanLastConnection) INSERT INTO CurrentScan (scanMac, scanLastIP, scanSourcePlugin, scanLastConnection)
VALUES (?, ?, ?, ?) VALUES (?, ?, ?, ?)
""", """,
("AA:BB:CC:DD:EE:FF", "2001:db8::1", "TEST_PLUGIN", "2025-01-01 01:00:00") ("aa:bb:cc:dd:ee:ff", "2001:db8::1", "TEST_PLUGIN", "2025-01-01 01:00:00")
) )
scan_db.commit() scan_db.commit()
@@ -61,7 +61,7 @@ def test_ipv6_update_preserves_ipv4(scan_db, mock_device_handling):
# 4️⃣ Verify the device fields # 4️⃣ Verify the device fields
row = cur.execute( row = cur.execute(
"SELECT devLastIP, devPrimaryIPv4, devPrimaryIPv6 FROM Devices WHERE devMac = ?", "SELECT devLastIP, devPrimaryIPv4, devPrimaryIPv6 FROM Devices WHERE devMac = ?",
("AA:BB:CC:DD:EE:FF",), ("aa:bb:cc:dd:ee:ff",),
).fetchone() ).fetchone()
assert row["devLastIP"] == "2001:db8::1" # Latest IP is now IPv6 assert row["devLastIP"] == "2001:db8::1" # Latest IP is now IPv6
test/server/test_graphql_endpoints.py
+1 -1
View File
@@ -24,7 +24,7 @@ def client():
@pytest.fixture @pytest.fixture
def test_mac(): def test_mac():
# Generate a unique MAC for each test run # Generate a unique MAC for each test run
return "AA:BB:CC:" + ":".join(f"{random.randint(0, 255):02X}" for _ in range(3)) return "aa:bb:cc:" + ":".join(f"{random.randint(0, 255):02X}" for _ in range(3))
def auth_headers(token): def auth_headers(token):
test/test_plugin_helper.py
+1 -1
View File
@@ -15,7 +15,7 @@ def test_normalize_mac_preserves_wildcard():
# Call once and assert deterministic result # Call once and assert deterministic result
result = normalize_mac("aabbcc*") result = normalize_mac("aabbcc*")
assert result == "AA:BB:CC:*", f"Expected 'AA:BB:CC:*' but got '{result}'" assert result == "AA:BB:CC:*", f"Expected 'AA:BB:CC:*' but got '{result}'"
assert normalize_mac("aa:bb:cc:dd:ee:ff") == "AA:BB:CC:DD:EE:FF" assert normalize_mac("aa:bb:cc:dd:ee:ff") == "aa:bb:cc:dd:ee:ff"
def test_normalize_mac_preserves_internet_root(): def test_normalize_mac_preserves_internet_root():