mirror/NetAlertX
1
0
Fork 0
mirror of https://github.com/jokob-sk/NetAlertX.git synced 2026-07-25 15:08:10 -04:00
Code Issues Packages Projects Releases Wiki Activity

BE+FE: report findinfg fixes + css

Browse Source
This commit is contained in:
jokob-sk
2026-07-12 10:26:53 +10:00
parent 3aa7bc5538
commit 1636d155c7
11 changed files with 238 additions and 87 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
front/css/app.css
View File

@@ -2222,6 +2222,7 @@ textarea[readonly],
.plugins-description
{
padding-top: 20px;
padding-left: 5px;
}
.pluginsCore .helpIcon
@@ -2230,6 +2231,20 @@ textarea[readonly],
font-size: x-small;
}
.pluginsCore .tab-content
{
padding-left: 5px;
}
.pluginsCore .tab-content .row
{
margin: 0px;
}
.plugin-obj-purge{
padding: 5px;
}
/*Hidden special button*/
@media (max-width: 365px) {

71
front/php/server/query_replace_config.php
View File

@@ -4,62 +4,71 @@
//------------------------------------------------------------------------------
// Check if authenticated
require_once $_SERVER['DOCUMENT_ROOT'] . '/php/templates/security.php';
// Get init.php
require dirname(__FILE__).'/../server/init.php';
// Get init.php
require dirname(__FILE__) . '/../server/init.php';
// ---- IMPORTS ----
header('Content-Type: text/plain; charset=utf-8');
global $configFolderPath;
//------------------------------------------------------------------------------
// Handle incoming requests
if ($_SERVER['REQUEST_METHOD'] === 'POST') {
// Access the 'config' parameter from the POST request
$base64Data = $_POST['base64data'] ?? null;
if (!$base64Data) {
$msg = "Missing 'base64data' parameter.";
echo $msg;
http_response_code(400); // Bad request
die($msg);
http_response_code(400);
die("Missing 'base64data' parameter.");
}
$fileName = $_POST['fileName'] ?? null;
if (!$fileName) {
$msg = "Missing 'fileName' parameter.";
echo $msg;
http_response_code(400); // Bad request
die($msg);
http_response_code(400);
die("Missing 'fileName' parameter.");
}
//--------------------------------------------------------------------------
// Only allow known configuration files
$safeName = basename($fileName);
$allowedFiles = [
'app.conf',
'workflows.json',
'devices.csv',
];
if (!in_array($safeName, $allowedFiles, true)) {
http_response_code(400);
die("Invalid fileName.");
}
$fullPath = rtrim($configFolderPath, DIRECTORY_SEPARATOR)
. DIRECTORY_SEPARATOR
. $safeName;
//--------------------------------------------------------------------------
// Decode incoming base64 data
$input = base64_decode($base64Data, true);
if ($input === false) {
$msg = "Invalid base64 data.";
echo $msg;
http_response_code(400); // Bad request
die($msg);
http_response_code(400);
die("Invalid base64 data.");
}
$fullPath = $configFolderPath.$fileName;
//--------------------------------------------------------------------------
// Backup the original file
if (file_exists($fullPath)) {
copy($fullPath, $fullPath . ".bak");
}
//--------------------------------------------------------------------------
// Write the new configuration
$file = fopen($fullPath, "w");
if (!$file) {
$msg = "Unable to open file: ". $fullPath;
echo $msg;
http_response_code(500); // Server error
die($msg);
if (file_put_contents($fullPath, $input, LOCK_EX) === false) {
http_response_code(500);
die("Unable to save configuration file.");
}
fwrite($file, $input);
fclose($file);
echo "Configuration file saved successfully: " .$fileName ;
}
?>
echo "Configuration file saved successfully: " . $safeName;
}

2
front/plugins/sync/sync.py
View File

@@ -80,10 +80,10 @@ def main():
mylog('verbose', [f'[{pluginName}] plugins_to_sync {plugins_to_sync}'])
index = 0
for plugin in all_plugins:
pref = plugin["unique_prefix"]
index = 0
if pref in plugins_to_sync:
index += 1
mylog('verbose', [f'[{pluginName}] synching "{pref}" ({index}/{len(plugins_to_sync)})'])

1
front/pluginsCore.php
View File

@@ -667,6 +667,7 @@ function initializeDataTables(prefix, colDefinitions, pluginObj) {
}
const skelId = `#skel-${tableId.replace('Table_', 'Target_')}`;
$(`#${tableId}`).DataTable({
autoWidth: false,
processing: true,
serverSide: true,
paging: true,

22
server/api_server/graphql_endpoint.py
View File

@@ -537,7 +537,7 @@ class Query(ObjectType):
langStrings = []
# --- CORE JSON FILES ---
language_folder = '/app/front/php/templates/language/'
language_folder = INSTALL_PATH + '/front/php/templates/language/'
if os.path.exists(language_folder):
for filename in os.listdir(language_folder):
if filename.endswith('.json') and filename != 'languages.json':
@@ -607,15 +607,25 @@ class Query(ObjectType):
langStrings = [ls for ls in langStrings if ls.langStringKey == langStringKey]
# --- Fallback to en_us if enabled and requested lang is missing ---
# Build a lookup map once without modifying the cached language lists.
if fallback_to_en and langCode and langCode != "en_us":
en_map = {}
# Index core English strings
for e in _langstrings_cache.get("core_en_us", []):
en_map.setdefault(e.langStringKey, e)
# Index English plugin strings (do not overwrite core strings)
for p in _langstrings_cache.get("plugin", []):
if p.langCode == "en_us":
en_map.setdefault(p.langStringKey, p)
# Replace empty translations with their English equivalent
for i, ls in enumerate(langStrings):
if not ls.langStringText: # empty string triggers fallback
# try to get en_us version
en_list = _langstrings_cache.get("core_en_us", [])
en_list += [p for p in _langstrings_cache.get("plugin", []) if p.langCode == "en_us"]
en_fallback = [e for e in en_list if e.langStringKey == ls.langStringKey]
en_fallback = en_map.get(ls.langStringKey)
if en_fallback:
langStrings[i] = en_fallback[0]
langStrings[i] = en_fallback
mylog('trace', f'[graphql_schema] Collected {len(langStrings)} language strings (langCode={langCode}, key={langStringKey}, fallback_to_en={fallback_to_en})')

26
server/api_server/mcp_endpoint.py
View File

@@ -36,6 +36,7 @@ from urllib.parse import quote
from flask import Blueprint, request, jsonify, Response, stream_with_context
import requests
from pydantic import ValidationError
from collections import deque
from helper import get_setting_value
from logger import mylog
@@ -403,12 +404,12 @@ def map_openapi_to_mcp_tools(spec: Dict[str, Any]) -> List[Dict[str, Any]]:
mcp_upgrade = tool["_is_mcp"] and not existing["_is_mcp"]
# Upgrade if same route type but current is POST and existing is GET
method_upgrade = (tool["_is_mcp"] == existing["_is_mcp"]) and tool["_is_post"] and not existing["_is_post"]
if mcp_upgrade or method_upgrade:
tools_map[original_op_id] = tool
# Final cleanup: remove internal preference flags and ensure tools have the original names
# unless we explicitly want the suffixed ones.
# unless we explicitly want the suffixed ones.
# The user said "Eliminate Duplicate Tool Names", so we should use original_op_id as the tool name.
final_tools = []
_tool_name_to_operation_id: Dict[str, str] = {}
@@ -450,7 +451,7 @@ def find_route_for_tool(tool_name: str) -> Optional[Dict[str, Any]]:
# Apply same preference logic as map_openapi_to_mcp_tools to ensure we pick the
# same route definition that generated the tool schema.
# Priority 1: MCP routes (they have specialized paths/behavior)
mcp_candidates = [c for c in candidates if c["path"].startswith("/mcp/")]
pool = mcp_candidates if mcp_candidates else candidates
@@ -779,7 +780,6 @@ def _execute_tool(route: Dict[str, Any], args: Dict[str, Any]) -> Dict[str, Any]
# =============================================================================
# MCP RESOURCES
# =============================================================================
def get_log_dir() -> str:
"""Get the log directory from environment or settings."""
log_dir = os.getenv("NETALERTX_LOG")
@@ -849,6 +849,7 @@ def _list_resources() -> List[Dict[str, Any]]:
def _read_resource(uri: str) -> List[Dict[str, Any]]:
"""Read a resource by URI."""
# Handle API Specification
if uri == "netalertx://api/openapi.json":
from flask import current_app
@@ -870,17 +871,22 @@ def _read_resource(uri: str) -> List[Dict[str, Any]]:
# Security: ensure path is within log directory
real_log_dir = os.path.realpath(log_dir)
real_path = os.path.realpath(file_path)
# Use os.path.commonpath or append separator to prevent prefix attacks
if not (real_path.startswith(real_log_dir + os.sep) or real_path == real_log_dir):
return [{"uri": uri, "text": "Access denied: path outside log directory"}]
if os.path.exists(file_path):
if os.path.exists(real_path):
try:
# Read last 500 lines to avoid overwhelming context
# Stream the file and retain only the last 500 lines
with open(real_path, "r", encoding="utf-8", errors="replace") as f:
lines = f.readlines()
content = "".join(lines[-500:])
return [{"uri": uri, "mimeType": "text/plain", "text": content}]
last_lines = deque(f, maxlen=500)
return [{
"uri": uri,
"mimeType": "text/plain",
"text": "".join(last_lines)
}]
except Exception as e:
return [{"uri": uri, "text": f"Error reading file: {e}"}]

8
server/api_server/openapi/validation.py
View File

@@ -55,7 +55,9 @@ def validate_request(
Features:
- Auto-registers the endpoint with the OpenAPI spec generator.
- Validates JSON body against `request_model` (for POST/PUT).
- Validates JSON body against `request_model` (for POST/PUT/PATCH/DELETE).
- Validates query parameters against `request_model` for GET requests.
- Allows HEAD requests to pass through without request validation.
- Injects the validated Pydantic model as the first argument to the view function.
- Supports auth_callable to check permissions before validation.
- Returns 422 (default) if validation fails.
@@ -171,6 +173,10 @@ def validate_request(
"error": "Invalid query parameters",
"message": "Unable to process query parameters"
}), 400
elif request.method == "HEAD":
# HEAD requests do not contain a meaningful request body and should
# behave like GET without running query/body validation.
pass
else:
# Unsupported HTTP method with a request_model - fail explicitly
mylog("verbose", [f"[Validation] Unsupported HTTP method {request.method} for {operation_id} with request_model"])

33
server/db/db_upgrade.py
View File

@@ -424,6 +424,34 @@ def ensure_Indexes(sql) -> bool:
"idx_dev_parentmac",
"CREATE INDEX idx_dev_parentmac ON Devices(devParentMAC)",
),
(
"idx_ses_lower_mac_date",
"""
CREATE INDEX idx_ses_lower_mac_date
ON Sessions(LOWER(sesMac), sesDateTimeConnection, sesDateTimeDisconnection, sesStillConnected)
""",
),
(
"idx_eve_lower_mac_date_type",
"""
CREATE INDEX idx_eve_lower_mac_date_type
ON Events(LOWER(eveMac), eveDateTime, eveEventType)
""",
),
(
"idx_dev_lower_mac",
"""
CREATE INDEX idx_dev_lower_mac
ON Devices(LOWER(devMac))
""",
),
(
"idx_dev_lower_parentmac",
"""
CREATE INDEX idx_dev_lower_parentmac
ON Devices(LOWER(devParentMAC))
""",
),
# Optional filter indexes
("idx_dev_site", "CREATE INDEX idx_dev_site ON Devices(devSite)"),
("idx_dev_group", "CREATE INDEX idx_dev_group ON Devices(devGroup)"),
@@ -437,11 +465,8 @@ def ensure_Indexes(sql) -> bool:
(
"idx_plugins_plugin_mac_ip",
"CREATE INDEX idx_plugins_plugin_mac_ip ON Plugins_Objects(plugin, objectPrimaryId, objectSecondaryId)",
), # Issue #1251: Optimize name resolution lookup
),
# Plugins_History: covers both the db_cleanup window function
# (PARTITION BY plugin ORDER BY dateTimeChanged DESC) and the
# API query (SELECT * … ORDER BY dateTimeChanged DESC).
# Without this, both ops do a full 48k-row table sort on every cycle.
(
"idx_plugins_history_plugin_dt",
"CREATE INDEX idx_plugins_history_plugin_dt ON Plugins_History(plugin, dateTimeChanged DESC)",

2
server/helper.py
View File

@@ -302,7 +302,7 @@ def setting_value_to_python_type(set_type, set_value):
elif dataType == "integer" and (elementType == "input" or elementType == "select"):
# handle storing/retrieving boolean values as 1/0
if set_value.lower() not in ["true", "false"] and isinstance(set_value, str):
if isinstance(set_value, str) and set_value.lower() not in ["true", "false"]:
value = int(set_value)
elif isinstance(set_value, bool):

12
server/models/device_instance.py
View File

@@ -213,11 +213,17 @@ class DeviceInstance:
)
ports = []
for o in objs:
try:
port = int(str(o.get('objectSecondaryId') or '').strip())
except (TypeError, ValueError):
continue
port = int(o.get('objectSecondaryId') or 0)
ports.append({"port": port, "service": o.get('watchedValue2', '')})
ports.append({
"port": port,
"service": o.get('watchedValue2', '')
})
return ports

133
server/scan/device_handling.py
View File

@@ -581,6 +581,20 @@ def print_scan_stats(db):
sql.execute(query)
stats = sql.fetchall()
# guard against empty resuolts crashing on stats[0] access below
if not stats:
stats = [{
"devices_detected": 0,
"new_devices": 0,
"down_alerts": 0,
"new_down_alerts": 0,
"new_connections": 0,
"disconnections": 0,
"ip_changes": 0,
"scanSourcePlugin": None,
"scan_method_count": 0,
}]
mylog("verbose", f"[Scan Stats] Devices Detected.......: {stats[0]['devices_detected']}",)
mylog("verbose", f"[Scan Stats] New Devices............: {stats[0]['new_devices']}")
mylog("verbose", f"[Scan Stats] Down Alerts............: {stats[0]['down_alerts']}")
@@ -1354,41 +1368,100 @@ def check_mac_or_internet(input_str):
return False
# -------------------------------------------------------------------------------
# Lookup unknown vendors on devices
def query_MAC_vendor(pMAC):
pMACstr = str(pMAC)
# ------------------------------------------------------------------------------
# Vendor database cache
#
# The IEEE OUI database is loaded into memory on first use to avoid repeatedly
# opening and linearly scanning the vendors file for every MAC address lookup.
# The cache is automatically rebuilt if the selected vendors file changes or is
# updated on disk.
# ------------------------------------------------------------------------------
_vendor_cache = None
_vendor_cache_path = None
_vendor_cache_mtime = None
filePath = vendorsPath
if os.path.isfile(vendorsPathNewest):
filePath = vendorsPathNewest
# ------------------------------------------------------------------------------
def _load_vendor_cache():
"""
Load the MAC vendor database into an in-memory dictionary.
# Check MAC parameter
mac = pMACstr.replace(":", "").lower()
if len(pMACstr) != 17 or len(mac) != 12:
return -2 # return -2 if ignored MAC
The cache maps the first six hexadecimal characters of a MAC address
(the OUI prefix) to the corresponding vendor name.
# Search vendor in HW Vendors DB
mac_start_string6 = mac[0:6]
The cache is only rebuilt when:
- It has not been loaded yet.
- The selected vendors file changes.
- The vendors file modification time changes.
Returns:
None
"""
global _vendor_cache, _vendor_cache_path, _vendor_cache_mtime
file_path = vendorsPathNewest if os.path.isfile(vendorsPathNewest) else vendorsPath
try:
with open(filePath, "r") as f:
for line in f:
line_lower = (
line.lower()
) # Convert line to lowercase for case-insensitive matching
if line_lower.startswith(mac_start_string6):
parts = line.split("\t", 1)
if len(parts) > 1:
vendor = parts[1].strip()
mylog("debug", [f"[Vendor Check] Found '{vendor}' for '{pMAC}' in {vendorsPath}"], )
return vendor
else:
mylog("debug", [f'[Vendor Check] ⚠ ERROR: Match found, but line could not be processed: "{line_lower}"'],)
return -1
mtime = os.path.getmtime(file_path)
except OSError:
mtime = None
# Cache is already up-to-date.
if (
_vendor_cache is not None and _vendor_cache_path == file_path and _vendor_cache_mtime == mtime
):
return
cache = {}
try:
with open(file_path, "r") as f:
for line in f:
parts = line.rstrip("\n").split("\t", 1)
# Expected format:
# <OUI>\t<Vendor Name>
if len(parts) == 2:
cache[parts[0].lower()] = parts[1]
return -1 # MAC address not found in the database
except FileNotFoundError:
mylog("none", [f"[Vendor Check] ⚠ ERROR: Vendors file {vendorsPath} not found."])
return -1
mylog("none", [f"[Vendor Check] ⚠ ERROR: Vendors file '{file_path}' not found."])
_vendor_cache = cache
_vendor_cache_path = file_path
_vendor_cache_mtime = mtime
# ------------------------------------------------------------------------------
def query_MAC_vendor(pMAC):
"""
Look up the hardware vendor for a MAC address.
The lookup uses an in-memory cache of the IEEE OUI database for O(1)
performance. The vendor database is loaded on first use and automatically
reloaded if the underlying file changes.
Args:
pMAC (str): MAC address in the format 'AA:BB:CC:DD:EE:FF'.
Returns:
str: Vendor name if found.
-1 : Vendor not found or vendor database unavailable.
-2 : Invalid MAC address format.
"""
pMACstr = str(pMAC)
mac = pMACstr.replace(":", "").lower()
# Validate expected MAC address format.
if len(pMACstr) != 17 or len(mac) != 12:
return -2
_load_vendor_cache()
vendor = _vendor_cache.get(mac[:6])
if vendor:
mylog("debug", [f"[Vendor Check] Found '{vendor}' for '{pMAC}'"])
return vendor
return -1