@@ -193,8 +316,19 @@ function logout_user(): void {
? '?next=' . htmlspecialchars($_GET['next'], ENT_QUOTES, 'UTF-8')
: '';
?>" method="post">
+
+
+
+
+
+
+
diff --git a/front/php/server/query_graphql.php b/front/php/server/query_graphql.php
index 8c7331343..d8353fe74 100755
--- a/front/php/server/query_graphql.php
+++ b/front/php/server/query_graphql.php
@@ -12,6 +12,9 @@
// Helper function to get GraphQL URL (you can replace this with environment variables)
function getGraphQLUrl() {
$port = getSettingValue("GRAPHQL_PORT"); // Port for the GraphQL server
+ if (empty($port) || !is_numeric($port)) {
+ $port = 20211;
+ }
return "0.0.0.0:$port/graphql"; // Full URL to the GraphQL endpoint
}
diff --git a/front/php/templates/auth.php b/front/php/templates/auth.php
index a31b6412f..b626fc073 100755
--- a/front/php/templates/auth.php
+++ b/front/php/templates/auth.php
@@ -29,11 +29,29 @@
}
$config_file_lines = file($config_file);
-$config_file_lines = array_values(preg_grep('/^SETPWD_password.*=/', $config_file_lines));
-$password_line = explode("'", $config_file_lines[0]);
+$config_file_lines_pw = array_values(preg_grep('/^SETPWD_password\s*=/', $config_file_lines ?: []));
+if (empty($config_file_lines_pw) || substr_count($config_file_lines_pw[0], "'") < 2) {
+ http_response_code(401);
+ echo 'Unauthorized 401';
+ exit;
+}
+$password_line = explode("'", $config_file_lines_pw[0], 3);
$nax_Password = $password_line[1];
-if (isset($_COOKIE[$CookieSaveLoginName]) && $nax_Password === $_COOKIE[$CookieSaveLoginName]) {
+$config_file_lines_token = array_values(preg_grep('/^API_TOKEN\s*=/', $config_file_lines ?: []));
+$token_line = !empty($config_file_lines_token) ? explode("'", $config_file_lines_token[0], 3) : [];
+$api_token = $token_line[1] ?? '';
+if (empty($api_token)) {
+ $api_token = getenv('API_TOKEN') ?: '';
+}
+if ($api_token === '') {
+ http_response_code(401);
+ echo 'Unauthorized 401';
+ exit;
+}
+
+$expected_cookie = hash_hmac('sha256', $nax_Password, $api_token);
+if (isset($_COOKIE[$CookieSaveLoginName]) && hash_equals($expected_cookie, $_COOKIE[$CookieSaveLoginName])) {
$isAuthenticated = true;
}
diff --git a/front/php/templates/language/en_us.json b/front/php/templates/language/en_us.json
index 6df32b0cc..89388f489 100755
--- a/front/php/templates/language/en_us.json
+++ b/front/php/templates/language/en_us.json
@@ -420,6 +420,7 @@
"Login_Remember": "Remember",
"Login_Remember_small": "(valid for 7 days)",
"Login_Submit": "Log in",
+ "Login_Username": "Username",
"Login_Toggle_Alert_headline": "Password Alert!",
"Login_Toggle_Info": "Password Information",
"Login_Toggle_Info_headline": "Password Information",
diff --git a/front/php/templates/security.php b/front/php/templates/security.php
index af30651dc..00cdfda74 100755
--- a/front/php/templates/security.php
+++ b/front/php/templates/security.php
@@ -90,7 +90,17 @@ function redirect($url) {
// Handle logout
if (!empty($_REQUEST['action']) && $_REQUEST['action'] == 'logout') {
session_destroy();
- setcookie(COOKIE_SAVE_LOGIN_NAME, "", time() - 3600);
+
+ // Determine protocol for secure cookie flag
+ $protocol = (!empty($_SERVER['HTTPS']) && $_SERVER['HTTPS'] !== 'off') ? 'https://' : 'http://';
+
+ setcookie(COOKIE_SAVE_LOGIN_NAME, "", [
+ 'expires' => time() - 3600,
+ 'path' => '/',
+ 'httponly' => true,
+ 'secure' => $protocol === 'https://',
+ 'samesite' => 'Strict',
+ ]);
redirect('index.php');
}
@@ -102,6 +112,10 @@ function redirect($url) {
// Handle web protection and password
$nax_WebProtection = strtolower(trim(getConfigLine('/^SETPWD_enable_password.*=/', $configLines)[1] ?? 'false'));
+
+// Auth plugins force web protection when enabled
+if (strtolower(trim(getConfigLine('/^LDAP_enabled\s*=/', $configLines)[1] ?? 'false')) === 'true') $nax_WebProtection = 'true';
+
$nax_Password = getConfigValue('/^SETPWD_password.*=/', $configLines);
$api_token = getConfigValue('/^API_TOKEN.*=/', $configLines, "'");
@@ -109,7 +123,7 @@ function redirect($url) {
// Authentication Handling
if ($nax_WebProtection == 'true') {
- if ($authHeader === $expectedToken) {
+ if (!empty($api_token) && $authHeader === $expectedToken) {
$_SESSION['login'] = 1; // User authenticated with bearer token
} elseif (!empty($authHeader)) {
echo "[Security] Incorrect Bearer Token";
diff --git a/requirements.txt b/requirements.txt
index 18fe9cbc3..51d0e03ee 100755
--- a/requirements.txt
+++ b/requirements.txt
@@ -34,4 +34,5 @@ freebox-api
mcp
psutil
pydantic>=2.0,<3.0
+ldap3>=2.9
fritzconnection>=1.15.1
diff --git a/scripts/run_tests_in_docker_environment.sh b/scripts/run_tests_in_docker_environment.sh
index 22711452d..fd78214a3 100755
--- a/scripts/run_tests_in_docker_environment.sh
+++ b/scripts/run_tests_in_docker_environment.sh
@@ -47,7 +47,7 @@ docker run -d --name netalertx-test-container \
# --- 5. Install Python test dependencies ---
echo "--- Installing Python test dependencies into venv ---"
-docker exec netalertx-test-container pip3 install --break-system-packages pytest docker debugpy selenium
+docker exec netalertx-test-container pip3 install --break-system-packages pytest docker debugpy selenium ldap3
# --- 6. Execute Setup Script ---
echo "--- Executing setup script inside the container ---"
@@ -96,7 +96,7 @@ docker exec netalertx-test-container /bin/bash -c " \
# --- 9. Execute Tests ---
echo "--- Executing tests inside the container ---"
docker exec netalertx-test-container /bin/bash -c " \
- cd /workspaces/NetAlertX && pytest -m 'not (docker or compose or feature_complete)' --cache-clear -o cache_dir=/tmp/.pytest_cache; \
+ cd /workspaces/NetAlertX && pytest test/ -m 'not (docker or compose or feature_complete)' --cache-clear -o cache_dir=/tmp/.pytest_cache; \
"
# --- 10. Final Teardown ---
diff --git a/server/api_server/api_server_start.py b/server/api_server/api_server_start.py
index 7ca8afbf7..4150f3d89 100755
--- a/server/api_server/api_server_start.py
+++ b/server/api_server/api_server_start.py
@@ -1,4 +1,5 @@
import threading
+import time
import sys
import os
from datetime import timedelta
@@ -20,6 +21,7 @@
from db.db_helper import get_date_from_period # noqa: E402 [flake8 lint suppression]
from app_state import updateState # noqa: E402 [flake8 lint suppression]
from utils.datetime_utils import timeNowUTC # noqa: E402 [flake8 lint suppression]
+from auth.manager import AuthManager # noqa: E402 [flake8 lint suppression]
from .graphql_endpoint import devicesSchema # noqa: E402 [flake8 lint suppression]
from .history_endpoint import delete_online_history # noqa: E402 [flake8 lint suppression]
@@ -105,6 +107,7 @@
EventListRequest,
LanguagesResponse,
PluginStatsResponse,
+ LoginRequest, LoginResponse,
)
from .sse_endpoint import ( # noqa: E402 [flake8 lint suppression]
@@ -115,6 +118,40 @@
# Flask application
app = Flask(__name__)
+# ---------------------------------------------------------------------------
+# Brute-force rate limiter for the login endpoint
+# ---------------------------------------------------------------------------
+_failed_logins_lock = threading.Lock()
+_failed_logins: dict[str, list[float]] = {} # IP -> list of failure timestamps
+_MAX_FAILED_ATTEMPTS = 5
+_LOCKOUT_WINDOW_SECONDS = 300 # 5 minutes
+
+
+def _record_failed_login(ip: str) -> None:
+ """Record a failed login attempt for rate limiting."""
+ now = time.monotonic()
+ with _failed_logins_lock:
+ attempts = _failed_logins.setdefault(ip, [])
+ attempts.append(now)
+ # Prune old entries outside the window
+ _failed_logins[ip] = [t for t in attempts if now - t < _LOCKOUT_WINDOW_SECONDS]
+
+
+def _is_rate_limited(ip: str) -> bool:
+ """Check if an IP has exceeded the max failed login attempts."""
+ now = time.monotonic()
+ with _failed_logins_lock:
+ attempts = _failed_logins.get(ip, [])
+ recent = [t for t in attempts if now - t < _LOCKOUT_WINDOW_SECONDS]
+ _failed_logins[ip] = recent
+ return len(recent) >= _MAX_FAILED_ATTEMPTS
+
+
+def _clear_failed_logins(ip: str) -> None:
+ """Clear failed login history on successful auth."""
+ with _failed_logins_lock:
+ _failed_logins.pop(ip, None)
+
@app.errorhandler(500)
@app.errorhandler(Exception)
@@ -2034,6 +2071,54 @@ def sync_endpoint_post(payload=None):
# --------------------------
# Auth endpoint
# --------------------------
+@app.route("/api/auth/login", methods=["POST"])
+@validate_request(
+ operation_id="login",
+ summary="Login",
+ description="Authenticate via the active provider and return the provider used.",
+ request_model=LoginRequest,
+ response_model=LoginResponse,
+ tags=["auth"],
+ auth_callable=is_authorized
+)
+def login(payload=None):
+ client_ip = request.headers.get("X-Forwarded-For", request.remote_addr or "unknown").split(",")[0].strip()
+
+ if _is_rate_limited(client_ip):
+ mylog("warning", [f"[auth] Rate-limited login attempt from {client_ip}"])
+ return jsonify(
+ {
+ "success": False,
+ "message": "Too many failed attempts. Try again later.",
+ "error": "Too many failed attempts. Try again later.",
+ }
+ ), 429
+
+ username = (payload.username or "admin").strip() if payload is not None else "admin"
+ password = payload.password if payload is not None else ""
+
+ result = AuthManager().authenticate(username, password)
+ if result.success:
+ _clear_failed_logins(client_ip)
+ return jsonify(
+ {
+ "success": True,
+ "message": "Authentication successful",
+ "username": result.username,
+ "provider": result.provider,
+ }
+ ), 200
+
+ _record_failed_login(client_ip)
+ return jsonify(
+ {
+ "success": False,
+ "message": "Invalid credentials",
+ "error": "Invalid credentials",
+ }
+ ), 401
+
+
@app.route("/auth", methods=["GET"])
@validate_request(
operation_id="check_auth",
diff --git a/server/api_server/openapi/schemas.py b/server/api_server/openapi/schemas.py
index edec8c58a..038a1b838 100644
--- a/server/api_server/openapi/schemas.py
+++ b/server/api_server/openapi/schemas.py
@@ -151,6 +151,49 @@ class PaginatedResponse(BaseResponse):
per_page: int = Field(50, ge=1, le=500, description="Items per page")
+# =============================================================================
+# AUTH SCHEMAS
+# =============================================================================
+
+
+class LoginRequest(BaseModel):
+ """Credentials supplied to the /api/auth/login endpoint."""
+ model_config = ConfigDict(
+ json_schema_extra={
+ "examples": [{"username": "admin", "password": "secret"}]
+ }
+ )
+
+ username: str = Field(
+ "",
+ max_length=256,
+ description="Login name. For LDAP this is the value matched by LDAP_user_filter.",
+ )
+ password: str = Field(
+ ...,
+ min_length=1,
+ max_length=1024,
+ description="Plain-text password. Transmitted over the internal loopback only.",
+ )
+
+
+class LoginResponse(BaseResponse):
+ """Response returned by /api/auth/login."""
+ model_config = ConfigDict(
+ json_schema_extra={
+ "examples": [
+ {"success": True, "message": "Authentication successful",
+ "username": "admin", "provider": "local"},
+ ]
+ }
+ )
+
+ username: Optional[str] = Field(None, description="Authenticated username")
+ provider: Optional[Literal["local", "ldap"]] = Field(
+ None, description="Provider that authenticated the user"
+ )
+
+
# =============================================================================
# DEVICE SCHEMAS
# =============================================================================
diff --git a/server/auth/__init__.py b/server/auth/__init__.py
new file mode 100644
index 000000000..707f581d3
--- /dev/null
+++ b/server/auth/__init__.py
@@ -0,0 +1 @@
+"""Authentication providers for NetAlertX."""
diff --git a/server/auth/base.py b/server/auth/base.py
new file mode 100644
index 000000000..487bea6ef
--- /dev/null
+++ b/server/auth/base.py
@@ -0,0 +1,48 @@
+"""
+Base classes for the NetAlertX pluggable authentication system.
+
+Each concrete provider must implement :class:`AuthProvider` and return an
+:class:`AuthResult` object so callers never need to import provider-specific
+modules directly.
+"""
+
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+from dataclasses import dataclass
+
+
+@dataclass
+class AuthResult:
+ """Outcome of an authentication attempt."""
+
+ success: bool
+ username: str = ""
+ provider: str = ""
+ error: str = ""
+
+ @classmethod
+ def ok(cls, username: str, provider: str) -> "AuthResult":
+ return cls(success=True, username=username, provider=provider)
+
+ @classmethod
+ def fail(cls, provider: str, error: str = "Invalid credentials") -> "AuthResult":
+ return cls(success=False, provider=provider, error=error)
+
+
+class AuthProvider(ABC):
+ """Abstract base class every auth provider must implement."""
+
+ #: Short lowercase identifier shown in the ``provider`` field of AuthResult
+ name: str = "unknown"
+
+ @abstractmethod
+ def authenticate(self, username: str, password: str) -> AuthResult:
+ """
+ Validate *username* / *password* and return an :class:`AuthResult`.
+
+ Implementations MUST NOT raise exceptions for invalid credentials โ
+ they should return ``AuthResult.fail(...)`` instead. Only
+ infrastructure errors (e.g. LDAP server unreachable) should propagate
+ as exceptions so the caller can decide how to handle them gracefully.
+ """
diff --git a/server/auth/ldap_provider.py b/server/auth/ldap_provider.py
new file mode 100644
index 000000000..ded7d6756
--- /dev/null
+++ b/server/auth/ldap_provider.py
@@ -0,0 +1,306 @@
+"""
+LDAP authentication provider for NetAlertX.
+
+Uses the *search-then-bind* pattern which is compatible with both OpenLDAP and
+Microsoft Active Directory:
+
+1. Connect to the LDAP server (optionally over TLS/StartTLS).
+2. Bind with a read-only service account (``LDAP_bind_dn`` / ``LDAP_bind_password``).
+3. Search for the user entry whose ``LDAP_user_filter`` matches *username*.
+4. Re-bind as that user with the supplied *password*.
+5. A successful re-bind means the credentials are valid.
+
+Configuration settings (via NetAlertX plugin ``auth_ldap``)
+-------------------------------------------------------------
+- ``LDAP_server`` โ hostname or IP of the LDAP/AD server
+- ``LDAP_port`` โ default 389 (636 for LDAPS)
+- ``LDAP_use_ssl`` โ True โ LDAPS (port 636), False โ plain / StartTLS
+- ``LDAP_use_start_tls`` โ True โ issue StartTLS on a plain-text connection
+- ``LDAP_bind_dn`` โ service-account DN for the initial search bind
+- ``LDAP_bind_password`` โ service-account password
+- ``LDAP_base_dn`` โ base DN for the user search
+- ``LDAP_user_filter`` โ search filter template; ``{username}`` is replaced at
+ runtime. Examples:
+ OpenLDAP : ``(uid={username})``
+ Active Directory: ``(sAMAccountName={username})``
+- ``LDAP_username_attribute`` โ attribute that holds the login name (default ``uid``)
+"""
+
+from __future__ import annotations
+
+import re
+import ssl
+from typing import Optional
+
+from helper import get_setting_value
+from logger import mylog, sanitize_for_log
+from auth.base import AuthProvider, AuthResult
+
+try:
+ import ldap3
+except ImportError:
+ ldap3 = None
+
+
+# ---------------------------------------------------------------------------
+# LDAP filter escaping (RFC 4515)
+# ---------------------------------------------------------------------------
+
+_LDAP_ESCAPE_RE = re.compile(r'[\\*()\x00/]')
+_LDAP_ESCAPE_MAP = {
+ '\\': r'\5c',
+ '*': r'\2a',
+ '(': r'\28',
+ ')': r'\29',
+ '\x00': r'\00',
+ '/': r'\2f',
+}
+
+
+def _escape_ldap_filter(value: str) -> str:
+ """Escape special characters in an LDAP filter value (RFC 4515 ยง4)."""
+ return _LDAP_ESCAPE_RE.sub(lambda m: _LDAP_ESCAPE_MAP[m.group(0)], value)
+
+
+# ---------------------------------------------------------------------------
+# DN value escaping (RFC 4514)
+# ---------------------------------------------------------------------------
+
+_DN_ESCAPE_RE = re.compile(r'[,+\"\\<>;#=]')
+
+
+def _escape_dn_value(value: str) -> str:
+ """Escape special characters in an LDAP DN attribute value (RFC 4514 Section 2.4)."""
+ escaped = _DN_ESCAPE_RE.sub(lambda m: '\\' + m.group(0), value)
+ if escaped.startswith(' '):
+ escaped = '\\ ' + escaped[1:]
+ if escaped.endswith(' ') and not escaped.endswith('\\ '):
+ escaped = escaped[:-1] + '\\ '
+ if escaped.startswith('#'):
+ escaped = '\\#' + escaped[1:]
+ return escaped
+
+
+class LdapProvider(AuthProvider):
+ """Authenticate against an LDAP / Active Directory server."""
+
+ name = "ldap"
+ USER_NOT_FOUND = "User not found"
+
+ # ------------------------------------------------------------------
+ # Public interface
+ # ------------------------------------------------------------------
+
+ def authenticate(self, username: str, password: str) -> AuthResult:
+ if not username or not password:
+ return AuthResult.fail(self.name, "Username and password are required")
+
+ if ldap3 is None:
+ mylog("none", ["[auth.ldap] ldap3 package is not installed"])
+ return AuthResult.fail(self.name, "LDAP library not available")
+
+ cfg = self._read_config()
+ if not cfg.get("server"):
+ return AuthResult.fail(self.name, "LDAP server not configured")
+
+ if not cfg["use_ssl"] and not cfg["use_start_tls"]:
+ mylog(
+ "warning",
+ [
+ "[auth.ldap] WARNING: Neither LDAPS nor StartTLS is enabled. "
+ "Credentials will be sent in cleartext."
+ ],
+ )
+
+ tls_obj = None
+ if cfg["use_ssl"] or cfg["use_start_tls"]:
+ validate = ssl.CERT_REQUIRED if cfg["tls_verify_cert"] else ssl.CERT_NONE
+ ca_certs_file = cfg["ca_cert_path"] if cfg["ca_cert_path"] else None
+ tls_obj = ldap3.Tls(validate=validate, ca_certs_file=ca_certs_file)
+
+ server_obj = ldap3.Server(
+ cfg["server"],
+ port=cfg["port"],
+ use_ssl=cfg["use_ssl"],
+ tls=tls_obj,
+ connect_timeout=cfg["timeout"],
+ get_info=ldap3.NONE,
+ )
+
+ try:
+ if cfg.get("direct_bind_format"):
+ safe_username = _escape_dn_value(username)
+ user_dn = cfg["direct_bind_format"].replace("{username}", safe_username)
+ else:
+ user_dn = self._resolve_user_dn(ldap3, server_obj, cfg, username)
+
+ if user_dn is None:
+ return AuthResult.fail(self.name, self.USER_NOT_FOUND)
+
+ return self._bind_as_user(ldap3, server_obj, cfg, user_dn, username, password)
+
+ except Exception as exc:
+ mylog("none", [f"[auth.ldap] Unexpected error for user '{sanitize_for_log(username)}': {exc}"])
+ raise exc
+
+ # ------------------------------------------------------------------
+ # Private helpers
+ # ------------------------------------------------------------------
+
+ def _get_server_obj(self, ldap3, cfg: dict):
+ tls_obj = None
+ if cfg["use_ssl"] or cfg["use_start_tls"]:
+ validate = ssl.CERT_REQUIRED if cfg.get("tls_verify_cert", True) else ssl.CERT_NONE
+ ca_certs_file = cfg.get("ca_cert_path") if cfg.get("ca_cert_path") else None
+ tls_obj = ldap3.Tls(validate=validate, ca_certs_file=ca_certs_file)
+
+ return ldap3.Server(
+ cfg["server"],
+ port=cfg["port"],
+ use_ssl=cfg["use_ssl"],
+ tls=tls_obj,
+ connect_timeout=cfg["timeout"],
+ get_info=ldap3.NONE,
+ )
+
+ def _read_config(self) -> dict:
+ def get_setting(key_name: str, default_value, type_cast):
+ db_val = get_setting_value(key_name)
+ if db_val is not None and db_val != "":
+ if type_cast is bool:
+ return str(db_val).lower() in ("true", "1", "yes") if isinstance(db_val, str) else bool(db_val)
+ try:
+ return type_cast(db_val)
+ except ValueError:
+ mylog("warning", [f"[auth.ldap] Invalid {type_cast.__name__} in database setting {key_name}: {db_val}"])
+ return default_value
+
+ return default_value
+
+ # IMPORTANT: never pass the returned dict to a log function โ it contains bind_password.
+ cfg = {
+ "enabled": get_setting("LDAP_enabled", False, bool),
+ "server": get_setting("LDAP_server", "", str).strip(),
+ "port": get_setting("LDAP_port", 389, int),
+ "use_ssl": get_setting("LDAP_use_ssl", False, bool),
+ "use_start_tls": get_setting("LDAP_use_start_tls", False, bool),
+ "tls_verify_cert": get_setting("LDAP_tls_verify_cert", True, bool),
+ "ca_cert_path": get_setting("LDAP_ca_cert_path", "", str).strip(),
+ "disable_local_admin": get_setting("LDAP_disable_local_admin", False, bool),
+ "direct_bind_format": get_setting("LDAP_direct_bind_format", "", str).strip(),
+ "bind_dn": get_setting("LDAP_bind_dn", "", str).strip(),
+ "bind_password": get_setting("LDAP_bind_password", "", str).strip(),
+ "base_dn": get_setting("LDAP_base_dn", "", str).strip(),
+ "user_filter": get_setting("LDAP_user_filter", "(uid={username})", str).strip(),
+ "username_attr": get_setting("LDAP_username_attribute", "uid", str).strip(),
+ "timeout": 5,
+ }
+
+ server = cfg["server"]
+ # Validate that the server value matches an RFC 1123 hostname/FQDN,
+ # an IPv4 address (e.g. 192.168.1.10), or bracketed IPv6 address (e.g. [::1])
+ if server and not re.match(
+ r'^(?:[a-zA-Z0-9](?:[a-zA-Z0-9\-]{0,61}[a-zA-Z0-9])?(?:\.[a-zA-Z0-9](?:[a-zA-Z0-9\-]{0,61}[a-zA-Z0-9])?)*'
+ r'|\d{1,3}(?:\.\d{1,3}){3}'
+ r'|\[[\da-fA-F:]+\])$',
+ server
+ ):
+ mylog("warning", [f"[auth.ldap] LDAP_server value '{sanitize_for_log(server)}' does not look like a valid hostname or IP"])
+
+ return cfg
+
+ def _create_secure_connection(self, ldap3, server_obj, cfg: dict, user: Optional[str], password: Optional[str], authentication):
+ """
+ Creates a secure LDAP connection, handling the StartTLS sequence
+ correctly before binding.
+ """
+ conn = ldap3.Connection(
+ server_obj,
+ user=user,
+ password=password,
+ auto_bind=ldap3.AUTO_BIND_NONE,
+ authentication=authentication,
+ )
+
+ if cfg["use_start_tls"] and not cfg["use_ssl"]:
+ if not conn.start_tls():
+ mylog("warning", [f"[auth.ldap] StartTLS negotiation failed: {conn.result}"])
+ conn.unbind()
+ return conn, False
+
+ if not conn.bind():
+ return conn, False
+
+ return conn, True
+
+ def _resolve_user_dn(self, ldap3, server_obj, cfg: dict, username: str) -> Optional[str]:
+ """
+ Bind with the service account and search for the user's DN.
+ Returns the DN string on success, None if user is not found.
+ Raises if the connection itself fails.
+ """
+ safe_username = _escape_ldap_filter(username)
+ search_filter = cfg["user_filter"].replace("{username}", safe_username)
+
+ authentication = ldap3.SIMPLE if cfg["bind_dn"] else ldap3.ANONYMOUS
+ conn, bind_success = self._create_secure_connection(
+ ldap3, server_obj, cfg,
+ user=cfg["bind_dn"] or None,
+ password=cfg["bind_password"] or None,
+ authentication=authentication
+ )
+
+ try:
+ if not bind_success:
+ mylog("none", [f"[auth.ldap] Service-account bind failed: {conn.result}"])
+ raise ConnectionError(f"LDAP service-account bind failed: {conn.result}")
+
+ search_success = conn.search(
+ search_base=cfg["base_dn"],
+ search_filter=search_filter,
+ search_scope=ldap3.SUBTREE,
+ attributes=[cfg["username_attr"]],
+ size_limit=2,
+ )
+
+ if not search_success:
+ mylog("warning", [f"[auth.ldap] Search operation failed for user '{sanitize_for_log(username)}': {conn.result}"])
+ return AuthResult(False, self.USER_NOT_FOUND, provider=self.name, username=username)
+
+ entries = conn.entries
+ if not entries or len(entries) != 1:
+ mylog("verbose", [
+ f"[auth.ldap] User '{sanitize_for_log(username)}' not found "
+ f"(got {len(entries)} entries for filter {search_filter})"
+ ])
+ return None
+
+ return entries[0].entry_dn
+
+ finally:
+ conn.unbind()
+
+ def _bind_as_user(
+ self, ldap3, server_obj, cfg: dict,
+ user_dn: str, username: str, password: str,
+ ) -> AuthResult:
+ """
+ Attempt to bind as *user_dn* using the supplied *password*.
+ A successful bind confirms valid credentials.
+ """
+ conn, bind_success = self._create_secure_connection(
+ ldap3, server_obj, cfg,
+ user=user_dn,
+ password=password,
+ authentication=ldap3.SIMPLE
+ )
+
+ try:
+ if bind_success:
+ return AuthResult.ok(username, self.name)
+
+ mylog("verbose", [f"[auth.ldap] User bind failed for DN '{sanitize_for_log(user_dn)}': {conn.result}"])
+ return AuthResult.fail(self.name)
+
+ finally:
+ conn.unbind()
diff --git a/server/auth/local_provider.py b/server/auth/local_provider.py
new file mode 100644
index 000000000..f29bedc41
--- /dev/null
+++ b/server/auth/local_provider.py
@@ -0,0 +1,67 @@
+"""
+Local password authentication provider.
+
+Validates the incoming password by comparing its SHA-256 digest against the
+hash stored in the ``SETPWD_password`` setting โ the same value that the PHP
+login page checks. This provider is always available as a fallback regardless
+of whether LDAP is enabled.
+
+Password handling
+-----------------
+- The stored value is the hex-encoded SHA-256 digest of the *raw* password.
+- Comparison uses :func:`hmac.compare_digest` to prevent timing attacks,
+ matching PHP's ``hash_equals`` behaviour.
+- No username check is performed for the local provider โ any non-empty
+ username is accepted (backward-compatible with the single-user model).
+"""
+
+from __future__ import annotations
+
+import hashlib
+import hmac
+
+from helper import get_setting_value
+from logger import mylog
+from auth.base import AuthProvider, AuthResult
+
+
+class LocalProvider(AuthProvider):
+ """Authenticate against the SETPWD_password SHA-256 hash."""
+
+ name = "local"
+
+ @classmethod
+ def is_fallback_allowed(cls, username: str) -> bool:
+ """
+ Determine if the given username is allowed to fall back to local authentication
+ when an external provider (like LDAP) fails. Currently, only the built-in
+ 'admin' account is permitted as a recovery identity.
+ """
+ return username == "admin"
+
+ def authenticate(self, username: str, password: str) -> AuthResult:
+ username = (username or "").strip()
+ if not username:
+ return AuthResult.fail(self.name, "Username must not be empty")
+
+ if not password:
+ return AuthResult.fail(self.name, "Password must not be empty")
+
+ stored_hash: str = get_setting_value("SETPWD_password") or ""
+
+ if not stored_hash:
+ mylog("verbose", ["[auth.local] SETPWD_password is not set"])
+ return AuthResult.fail(self.name, "Local password not configured")
+
+ # NOTE: SHA-256 is used here purely to *replicate* the legacy PHP hash
+ # comparison (``hash('sha256', $password)``). The hash is computed by
+ # the PHP login page and stored in SETPWD_password; this code only
+ # verifies it โ it does NOT use SHA-256 to *store* a new password.
+ # Replacing this with bcrypt/Argon2 would require a coordinated change
+ # to the PHP side and is tracked separately.
+ incoming_hash = hashlib.sha256(password.encode("utf-8")).hexdigest() # nosec B324
+
+ if hmac.compare_digest(stored_hash.lower(), incoming_hash.lower()):
+ return AuthResult.ok(username, self.name)
+
+ return AuthResult.fail(self.name)
diff --git a/server/auth/manager.py b/server/auth/manager.py
new file mode 100644
index 000000000..2bab566b8
--- /dev/null
+++ b/server/auth/manager.py
@@ -0,0 +1,101 @@
+"""
+AuthManager โ selects and dispatches to the correct authentication provider.
+
+Decision logic
+--------------
+1. If ``LDAP_enabled`` setting is ``True`` โ use :class:`LdapProvider`.
+2. Otherwise โ use :class:`LocalProvider`.
+
+The manager does NOT cache the provider across requests so that toggling
+``LDAP_enabled`` at runtime takes effect without a server restart.
+"""
+
+from __future__ import annotations
+
+from helper import get_setting_value
+from logger import mylog, sanitize_for_log
+from auth.base import AuthProvider, AuthResult
+from auth.local_provider import LocalProvider
+from auth.ldap_provider import LdapProvider
+
+
+class AuthManager:
+ """Thin dispatcher that picks the active :class:`AuthProvider`."""
+
+ def get_provider(self) -> AuthProvider:
+ """Return the :class:`AuthProvider` appropriate for the current config."""
+ ldap_enabled = LdapProvider()._read_config().get("enabled", False)
+ if ldap_enabled:
+ return LdapProvider()
+ return LocalProvider()
+
+ def authenticate(self, username: str, password: str) -> AuthResult:
+ """Authenticate *username* / *password* with the active provider."""
+ ldap_provider = LdapProvider()
+ ldap_cfg = ldap_provider._read_config()
+ ldap_enabled = ldap_cfg.get("enabled", False)
+
+ if ldap_enabled:
+ # Check requirements
+ setpwd_enabled = get_setting_value("SETPWD_enable_password")
+ disable_local = ldap_cfg.get("disable_local_admin", False)
+
+ if not setpwd_enabled and not disable_local:
+ mylog(
+ "warning",
+ [
+ "[auth.manager] LDAP is enabled but "
+ "SETPWD_enable_password is disabled. Local admin account is still active unless explicitly disabled in LDAP settings "
+ "(not recommended)."
+ ],
+ )
+
+ mylog("verbose", ["[auth.manager] Trying LDAP provider"])
+ try:
+ ldap_result = ldap_provider.authenticate(username, password)
+ if ldap_result.success:
+ return ldap_result
+
+ # If the user doesn't exist in LDAP, we can fallback to local auth
+ if ldap_result.error == LdapProvider.USER_NOT_FOUND:
+ if not disable_local:
+ # Only the built-in local admin is a recovery account;
+ # all other identities must exist in LDAP.
+ if username != "admin":
+ mylog("verbose", [f"[auth.manager] Local fallback denied for non-admin user '{sanitize_for_log(username)}'"])
+ return ldap_result
+ mylog("warning", ["[auth.manager] User not found in LDAP, falling back to local provider"])
+ local_result = LocalProvider().authenticate(username, password)
+ if not local_result.success:
+ mylog("verbose", [f"[auth.manager] Authentication failed for user '{sanitize_for_log(username)}' via both ldap and local"])
+ return local_result
+ else:
+ mylog("verbose", ["[auth.manager] User not found in LDAP, but local fallback is disabled."])
+ return ldap_result
+
+ # LDAP returned failure without exception, which means invalid credentials
+ # We should NOT fallback if the LDAP server explicitly rejected them
+ if not disable_local:
+ mylog("warning", ["[auth.manager] LDAP explicitly rejected credentials, no fallback to local provider"])
+ else:
+ mylog("verbose", [f"[auth.manager] Authentication failed for user '{sanitize_for_log(username)}' via ldap"])
+ return ldap_result
+ except Exception as e:
+ # Infrastructure error
+ mylog("warning", [f"[auth.manager] LDAP infrastructure error: {e}."])
+
+ if not disable_local and username == "admin":
+ mylog("warning", ["[auth.manager] LDAP failed, falling back to local provider for 'admin' user"])
+ local_result = LocalProvider().authenticate(username, password)
+ if not local_result.success:
+ mylog("verbose", [f"[auth.manager] Authentication failed for user '{sanitize_for_log(username)}' via both ldap and local"])
+ return local_result
+
+ mylog("warning", ["[auth.manager] Authentication failed securely (fail-closed)."])
+ return AuthResult.fail("ldap", "LDAP infrastructure error")
+
+ mylog("verbose", ["[auth.manager] Using local provider"])
+ local_result = LocalProvider().authenticate(username, password)
+ if not local_result.success:
+ mylog("verbose", [f"[auth.manager] Authentication failed for user '{sanitize_for_log(username)}' via local"])
+ return local_result
diff --git a/server/logger.py b/server/logger.py
index c3a8c3eca..ee0d76e5a 100755
--- a/server/logger.py
+++ b/server/logger.py
@@ -161,3 +161,9 @@ def append_line_to_file(pPath, pText):
# Create default logger instance and backward-compatible global mylog
logger = Logger(conf.LOG_LEVEL)
mylog = logger.mylog
+
+
+def sanitize_for_log(value):
+ if value is None:
+ return ""
+ return str(value).encode("unicode_escape").decode("ascii")
diff --git a/server/plugins/auth_ldap/README.md b/server/plugins/auth_ldap/README.md
new file mode 100644
index 000000000..5181b3b32
--- /dev/null
+++ b/server/plugins/auth_ldap/README.md
@@ -0,0 +1,173 @@
+## Overview
+
+The LDAP Authentication plugin adds directory-backed login to NetAlertX. It supports both OpenLDAP and Microsoft Active Directory using a search-then-bind flow, and it can also use a direct bind format when your directory supports user principal names or predictable distinguished names.
+
+### Features
+
+- Authenticate users against LDAP or Active Directory from the NetAlertX login page
+- Support OpenLDAP-style search filters and Active Directory `sAMAccountName` lookups
+- Support LDAPS and StartTLS with optional custom CA bundle validation
+- Allow anonymous search, service-account search, or direct bind authentication
+- Optionally disable the local admin fallback for strict LDAP-only access
+- Built-in brute-force rate limiting on login attempts
+
+> [!WARNING]
+> If local admin fallback remains enabled, the local `admin` account can still authenticate when LDAP is enabled. Change the local password and enable **Disable local admin account** if you want strict LDAP-only login.
+
+### Quick Setup Guide
+
+> [!NOTE]
+> NetAlertX does not define individual environment variables for LDAP configuration. All LDAP options are standard application settings configured through the Web UI (**Settings > LDAP Authentication**). To initialize or override settings headlessly in automated or containerized environments, see [Headless / Automated Configuration (`APP_CONF_OVERRIDE`)](#headless--automated-configuration-app_conf_override).
+
+1. Open **Settings** and navigate to **LDAP Authentication**.
+2. Enable **Enable LDAP login**.
+3. Enter the LDAP server hostname and port.
+4. Choose one transport mode:
+ - **Use LDAPS (SSL)** for TLS from connect time, usually port `636`
+ - **Use StartTLS** for upgrading a plain LDAP connection, usually port `389`
+5. Choose one bind strategy:
+ - **Search then bind**: configure **Bind DN**, **Bind password**, **Base DN**, and **User search filter**
+ - **Direct bind**: set **Direct Bind Format** and leave search-specific fields unused
+6. Set **Username attribute** to match your directory, usually `uid` for OpenLDAP or `sAMAccountName` for Active Directory.
+7. If you want LDAP to be mandatory, enable **Disable local admin account (Recommended)** after confirming LDAP login works.
+
+#### Required Settings
+
+- **Enable LDAP login** (`LDAP_enabled`): Turns on LDAP-backed authentication and shows the username field on the login form.
+- **LDAP server** (`LDAP_server`): Hostname or IP address of the LDAP or Active Directory server.
+- **LDAP port** (`LDAP_port`): Directory port. Typical values are `389` for LDAP or StartTLS and `636` for LDAPS.
+
+#### Authentication Mode Settings
+
+- **Direct Bind Format (Optional)** (`LDAP_direct_bind_format`): Optional user DN or UPN template such as `{username}@example.com` or `uid={username},ou=users,dc=example,dc=com`. When this is set, search settings are ignored.
+- **Bind DN (service account)** (`LDAP_bind_dn`): Read-only service account DN used for the initial search bind. Leave empty for anonymous bind if your directory allows it.
+- **Bind password** (`LDAP_bind_password`): Password for the service account.
+- **Base DN** (`LDAP_base_dn`): Search root used to find the authenticating user.
+- **User search filter** (`LDAP_user_filter`): Search filter template containing `{username}`. Example OpenLDAP filter: `(uid={username})`. Example Active Directory filter: `(sAMAccountName={username})`.
+- **Username attribute** (`LDAP_username_attribute`): Attribute that stores the login name.
+
+#### TLS and Certificate Settings
+
+- **Use LDAPS (SSL)** (`LDAP_use_ssl`): Connect using TLS immediately.
+- **Use StartTLS** (`LDAP_use_start_tls`): Upgrade a plain LDAP connection to TLS.
+- **Verify TLS Certificate** (`LDAP_tls_verify_cert`): Require a valid server certificate. Disable only for temporary testing.
+- **CA Certificate Bundle Path** (`LDAP_ca_cert_path`): Absolute path to a CA bundle when your directory uses a private CA.
+
+#### Access Control Settings
+
+- **Disable local admin account (Recommended)** (`LDAP_disable_local_admin`): Disables the local `admin` fallback and enforces LDAP-only login.
+
+### Headless / Automated Configuration (`APP_CONF_OVERRIDE`)
+
+For automated and containerized deployments, all NetAlertX settings (including LDAP) can be initialized or overridden using the standard `APP_CONF_OVERRIDE` environment variable containing a JSON dictionary of setting keys and values. Individual setting environment variables are not supported to avoid duplicating configuration mechanisms.
+
+| Setting Key | Description | Default |
+|-------------|-------------|---------|
+| `LDAP_enabled` | Enable or disable LDAP authentication (`true` or `false`) | `false` |
+| `LDAP_server` | Hostname or IP address of the LDAP/AD server | `""` |
+| `LDAP_port` | TCP port for LDAP connections | `389` |
+| `LDAP_use_ssl` | Use LDAPS (TLS from the start). Mutually exclusive with StartTLS. (`true`/`false`) | `false` |
+| `LDAP_use_start_tls` | Upgrade plain-text connection using STARTTLS. (`true`/`false`) | `false` |
+| `LDAP_tls_verify_cert` | Require a valid TLS certificate from the server. | `true` |
+| `LDAP_ca_cert_path` | Absolute path to a custom CA certificate bundle for validation (e.g. `/data/my-ca.pem`) | `""` |
+| `LDAP_disable_local_admin` | When `true`, completely disables the local fallback password. | `false` |
+| `LDAP_direct_bind_format` | Format for direct binding (e.g., `{username}@example.com`). If set, bind DN and search filters are ignored. | `""` |
+| `LDAP_bind_dn` | Distinguished Name of the read-only service account used to search for user entries. | `""` |
+| `LDAP_bind_password` | Password for the service account. | `""` |
+| `LDAP_base_dn` | Base DN to search under (e.g., `ou=users,dc=example,dc=com`) | `""` |
+| `LDAP_user_filter` | Search filter template. `{username}` is replaced at runtime. | `(uid={username})` |
+| `LDAP_username_attribute` | LDAP attribute that holds the login name. | `uid` |
+
+### Docker Compose Example
+
+```yaml
+services:
+ netalertx:
+ image: netalertx:latest
+ environment:
+ - APP_CONF_OVERRIDE={"LDAP_enabled":true,"LDAP_server":"ldap.example.com","LDAP_port":636,"LDAP_use_ssl":true,"LDAP_tls_verify_cert":true,"LDAP_ca_cert_path":"/data/ca-certificates.crt","LDAP_bind_dn":"cn=readonly,dc=example,dc=com","LDAP_bind_password":"super_secret_password","LDAP_base_dn":"ou=users,dc=example,dc=com","LDAP_user_filter":"(uid={username})","LDAP_disable_local_admin":true}
+ volumes:
+ - ./netalertx_data:/data
+ # Mount the CA certificate into the container so NetAlertX can verify the LDAP server
+ - ./my-company-ca.crt:/data/ca-certificates.crt:ro
+```
+
+### Common Configuration Examples
+
+#### OpenLDAP
+* **`LDAP_port`**: `389` (or `636` for LDAPS)
+* **`LDAP_username_attribute`**: `uid`
+* **`LDAP_user_filter`**: `(uid={username})`
+
+#### Active Directory
+* **`LDAP_port`**: `389` (or `636` for LDAPS)
+* **`LDAP_username_attribute`**: `sAMAccountName`
+* **`LDAP_user_filter`**: `(sAMAccountName={username})`
+
+**Advanced Active Directory Filter Examples:**
+* Require user to be in a specific group (e.g., `NetAlertX`):
+ ```text
+ (&(memberOf=CN=NetAlertX,OU=Groups,DC=example,DC=com)(sAMAccountName={username}))
+ ```
+* Restrict login to a specific user only (e.g., `my_username`):
+ ```text
+ (&(sAMAccountName=my_username)(sAMAccountName={username}))
+ ```
+
+### Usage
+
+1. Configure the settings described above in the UI or via `APP_CONF_OVERRIDE`.
+2. Save the settings.
+3. Test logging in through the NetAlertX login page with an LDAP user while the local admin fallback is still active.
+4. Once verified, enable **Disable local admin account** if desired.
+
+### Security Notes
+
+- Always enable TLS certificate verification in production to prevent Man-in-the-Middle attacks.
+- Prefer LDAPS or StartTLS for production deployments.
+- If you use a private CA, place the CA bundle in a persistent path such as `/config/my-ca.pem` and set **CA Certificate Bundle Path**.
+- The bind password is stored in `app.conf`. Treat the config file as sensitive data and restrict access accordingly.
+- Enable **Disable local admin account** only after you have verified at least one LDAP login path works.
+
+#### Brute-Force Rate Limiting
+To protect against automated password guessing and credential stuffing, the `/api/auth/login` endpoint enforces per-client-IP in-memory rate limiting defined in `server/api_server/api_server_start.py`:
+- **Maximum failed attempts:** 5 failed attempts (`_MAX_FAILED_ATTEMPTS = 5`)
+- **Lockout window:** 300 seconds / 5 minutes (`_LOCKOUT_WINDOW_SECONDS = 300`)
+- **Lockout response:** When rate-limited, the endpoint rejects the request with HTTP 429 (`Too Many Requests`) and payload `{"success": false, "message": "Too many failed attempts. Try again later.", "error": "Too many failed attempts. Try again later."}`.
+- **Reset on success:** Successful logins immediately reset the failure counter and clear the lockout record for that client IP.
+- **Client IP resolution:** Client IP is determined from the first entry of the `X-Forwarded-For` header or `request.remote_addr`.
+
+### Troubleshooting
+
+#### LDAP server not configured
+
+- Verify **Enable LDAP login** is on.
+- Confirm **LDAP server** and **LDAP port** are saved correctly.
+
+#### User not found
+
+- Check **Base DN** and **User search filter**.
+- Confirm the **Username attribute** matches your directory schema.
+
+#### TLS or certificate failures
+
+- Make sure **Use LDAPS (SSL)** and **Use StartTLS** are not both enabled at the same time.
+- Confirm the CA bundle path is valid when certificate verification is enabled.
+- Test with certificate verification enabled before considering temporary relaxation for lab environments.
+
+#### Login falls back to local admin
+
+- This is expected when **Disable local admin account** is off.
+- Turn it on only after LDAP login succeeds consistently.
+
+### Technical Notes
+
+- Plugin type: `system`
+- Data source: `template`
+- Login flow: search then bind, or direct bind if `LDAP_direct_bind_format` is set
+
+### Version
+
+- Version: 1.0.0
+- Author: NetAlertX contributors
+- Release Date: October 2026
diff --git a/server/plugins/auth_ldap/config.json b/server/plugins/auth_ldap/config.json
new file mode 100644
index 000000000..890bafde2
--- /dev/null
+++ b/server/plugins/auth_ldap/config.json
@@ -0,0 +1,505 @@
+{
+ "code_name": "auth_ldap",
+ "template_type": "database-entry",
+ "unique_prefix": "LDAP",
+ "plugin_type": "system",
+ "enabled": true,
+ "data_source": "template",
+ "show_ui": false,
+ "localized": [
+ "display_name",
+ "description",
+ "icon"
+ ],
+ "display_name": [
+ {
+ "language_code": "en_us",
+ "string": "LDAP Authentication"
+ }
+ ],
+ "description": [
+ {
+ "language_code": "en_us",
+ "string": "Authenticate users against an LDAP or Active Directory server."
+ }
+ ],
+ "icon": [
+ {
+ "language_code": "en_us",
+ "string": "
"
+ }
+ ],
+ "params": [],
+ "settings": [
+ {
+ "function": "enabled",
+ "type": {
+ "dataType": "boolean",
+ "elements": [
+ {
+ "elementType": "input",
+ "elementOptions": [
+ {
+ "type": "checkbox"
+ }
+ ],
+ "transformers": []
+ }
+ ]
+ },
+ "default_value": false,
+ "options": [],
+ "localized": [
+ "name",
+ "description"
+ ],
+ "name": [
+ {
+ "language_code": "en_us",
+ "string": "Enable LDAP login"
+ }
+ ],
+ "description": [
+ {
+ "language_code": "en_us",
+ "string": "Validate user credentials against the configured LDAP server."
+ }
+ ]
+ },
+ {
+ "function": "server",
+ "type": {
+ "dataType": "string",
+ "elements": [
+ {
+ "elementType": "input",
+ "elementOptions": [],
+ "transformers": []
+ }
+ ]
+ },
+ "maxLength": 253,
+ "default_value": "",
+ "options": [],
+ "localized": [
+ "name",
+ "description"
+ ],
+ "name": [
+ {
+ "language_code": "en_us",
+ "string": "LDAP server"
+ }
+ ],
+ "description": [
+ {
+ "language_code": "en_us",
+ "string": "Hostname or IP address of LDAP server. Example:
ldap.example.com"
+ }
+ ]
+ },
+ {
+ "function": "port",
+ "type": {
+ "dataType": "integer",
+ "elements": [
+ {
+ "elementType": "input",
+ "elementOptions": [
+ {
+ "type": "number"
+ }
+ ],
+ "transformers": []
+ }
+ ]
+ },
+ "default_value": 389,
+ "options": [],
+ "localized": [
+ "name",
+ "description"
+ ],
+ "name": [
+ {
+ "language_code": "en_us",
+ "string": "LDAP port"
+ }
+ ],
+ "description": [
+ {
+ "language_code": "en_us",
+ "string": "TCP port for LDAP connections (default
389, or
636 for LDAPS)."
+ }
+ ]
+ },
+ {
+ "function": "use_ssl",
+ "type": {
+ "dataType": "boolean",
+ "elements": [
+ {
+ "elementType": "input",
+ "elementOptions": [
+ {
+ "type": "checkbox"
+ }
+ ],
+ "transformers": []
+ }
+ ]
+ },
+ "default_value": false,
+ "options": [],
+ "localized": [
+ "name",
+ "description"
+ ],
+ "name": [
+ {
+ "language_code": "en_us",
+ "string": "Use LDAPS (SSL)"
+ }
+ ],
+ "description": [
+ {
+ "language_code": "en_us",
+ "string": "Enable LDAPS (implicit TLS on port 636). Mutually exclusive with StartTLS."
+ }
+ ]
+ },
+ {
+ "function": "use_start_tls",
+ "type": {
+ "dataType": "boolean",
+ "elements": [
+ {
+ "elementType": "input",
+ "elementOptions": [
+ {
+ "type": "checkbox"
+ }
+ ],
+ "transformers": []
+ }
+ ]
+ },
+ "default_value": false,
+ "options": [],
+ "localized": [
+ "name",
+ "description"
+ ],
+ "name": [
+ {
+ "language_code": "en_us",
+ "string": "Use StartTLS"
+ }
+ ],
+ "description": [
+ {
+ "language_code": "en_us",
+ "string": "Upgrade connection to TLS via STARTTLS (port 389). Mutually exclusive with LDAPS."
+ }
+ ]
+ },
+ {
+ "function": "tls_verify_cert",
+ "type": {
+ "dataType": "boolean",
+ "elements": [
+ {
+ "elementType": "input",
+ "elementOptions": [
+ {
+ "type": "checkbox"
+ }
+ ],
+ "transformers": []
+ }
+ ]
+ },
+ "default_value": true,
+ "options": [],
+ "localized": [
+ "name",
+ "description"
+ ],
+ "name": [
+ {
+ "language_code": "en_us",
+ "string": "Verify TLS Certificate"
+ }
+ ],
+ "description": [
+ {
+ "language_code": "en_us",
+ "string": "Verify the LDAP server TLS certificate against trusted CAs."
+ }
+ ]
+ },
+ {
+ "function": "ca_cert_path",
+ "type": {
+ "dataType": "string",
+ "elements": [
+ {
+ "elementType": "input",
+ "elementOptions": [],
+ "transformers": []
+ }
+ ]
+ },
+ "maxLength": 512,
+ "default_value": "",
+ "options": [],
+ "localized": [
+ "name",
+ "description"
+ ],
+ "name": [
+ {
+ "language_code": "en_us",
+ "string": "CA Certificate Bundle Path"
+ }
+ ],
+ "description": [
+ {
+ "language_code": "en_us",
+ "string": "Path to custom CA certificate bundle (e.g.
/config/my-ca.pem)."
+ }
+ ]
+ },
+ {
+ "function": "disable_local_admin",
+ "type": {
+ "dataType": "boolean",
+ "elements": [
+ {
+ "elementType": "input",
+ "elementOptions": [
+ {
+ "type": "checkbox"
+ }
+ ],
+ "transformers": []
+ }
+ ]
+ },
+ "default_value": false,
+ "options": [],
+ "localized": [
+ "name",
+ "description"
+ ],
+ "name": [
+ {
+ "language_code": "en_us",
+ "string": "Disable local admin account (Recommended)"
+ }
+ ],
+ "description": [
+ {
+ "language_code": "en_us",
+ "string": "Disable local admin login to enforce LDAP-only authentication."
+ }
+ ]
+ },
+ {
+ "function": "direct_bind_format",
+ "type": {
+ "dataType": "string",
+ "elements": [
+ {
+ "elementType": "input",
+ "elementOptions": [],
+ "transformers": []
+ }
+ ]
+ },
+ "maxLength": 512,
+ "default_value": "",
+ "options": [],
+ "localized": [
+ "name",
+ "description"
+ ],
+ "name": [
+ {
+ "language_code": "en_us",
+ "string": "Direct Bind Format (Optional)"
+ }
+ ],
+ "description": [
+ {
+ "language_code": "en_us",
+ "string": "Direct bind template (e.g.
{username}@example.com). Bypasses user search if set."
+ }
+ ]
+ },
+ {
+ "function": "bind_dn",
+ "type": {
+ "dataType": "string",
+ "elements": [
+ {
+ "elementType": "input",
+ "elementOptions": [],
+ "transformers": []
+ }
+ ]
+ },
+ "maxLength": 512,
+ "default_value": "",
+ "options": [],
+ "localized": [
+ "name",
+ "description"
+ ],
+ "name": [
+ {
+ "language_code": "en_us",
+ "string": "Bind DN (service account)"
+ }
+ ],
+ "description": [
+ {
+ "language_code": "en_us",
+ "string": "Service account DN for user search (e.g.
cn=readonly,dc=example,dc=com). Blank for anonymous."
+ }
+ ]
+ },
+ {
+ "function": "bind_password",
+ "type": {
+ "dataType": "string",
+ "elements": [
+ {
+ "elementType": "input",
+ "elementOptions": [
+ {
+ "type": "password"
+ }
+ ],
+ "transformers": []
+ }
+ ]
+ },
+ "maxLength": 256,
+ "default_value": "",
+ "options": [],
+ "localized": [
+ "name",
+ "description"
+ ],
+ "name": [
+ {
+ "language_code": "en_us",
+ "string": "Bind password"
+ }
+ ],
+ "description": [
+ {
+ "language_code": "en_us",
+ "string": "Password for the service account above."
+ }
+ ]
+ },
+ {
+ "function": "base_dn",
+ "type": {
+ "dataType": "string",
+ "elements": [
+ {
+ "elementType": "input",
+ "elementOptions": [],
+ "transformers": []
+ }
+ ]
+ },
+ "maxLength": 512,
+ "default_value": "",
+ "options": [],
+ "localized": [
+ "name",
+ "description"
+ ],
+ "name": [
+ {
+ "language_code": "en_us",
+ "string": "Base DN"
+ }
+ ],
+ "description": [
+ {
+ "language_code": "en_us",
+ "string": "Base search DN. Example:
ou=users,dc=example,dc=com"
+ }
+ ]
+ },
+ {
+ "function": "user_filter",
+ "type": {
+ "dataType": "string",
+ "elements": [
+ {
+ "elementType": "input",
+ "elementOptions": [],
+ "transformers": []
+ }
+ ]
+ },
+ "maxLength": 512,
+ "default_value": "(uid={username})",
+ "options": [],
+ "localized": [
+ "name",
+ "description"
+ ],
+ "name": [
+ {
+ "language_code": "en_us",
+ "string": "User search filter"
+ }
+ ],
+ "description": [
+ {
+ "language_code": "en_us",
+ "string": "Search filter for user lookup. Example:
(uid={username}) or
(sAMAccountName={username})."
+ }
+ ]
+ },
+ {
+ "function": "username_attribute",
+ "type": {
+ "dataType": "string",
+ "elements": [
+ {
+ "elementType": "input",
+ "elementOptions": [],
+ "transformers": []
+ }
+ ]
+ },
+ "maxLength": 64,
+ "default_value": "uid",
+ "options": [],
+ "localized": [
+ "name",
+ "description"
+ ],
+ "name": [
+ {
+ "language_code": "en_us",
+ "string": "Username attribute"
+ }
+ ],
+ "description": [
+ {
+ "language_code": "en_us",
+ "string": "LDAP attribute for login name (typically
uid or
sAMAccountName)."
+ }
+ ]
+ }
+ ]
+}
diff --git a/test/api_endpoints/test_auth_endpoints.py b/test/api_endpoints/test_auth_endpoints.py
index be87cca6c..d796a759c 100644
--- a/test/api_endpoints/test_auth_endpoints.py
+++ b/test/api_endpoints/test_auth_endpoints.py
@@ -2,7 +2,9 @@
import sys
import os
+import hashlib
import pytest
+from unittest.mock import patch
# Register NetAlertX directories
INSTALL_PATH = os.getenv("NETALERTX_APP", "/app")
@@ -64,3 +66,84 @@ def test_auth_invalid_token(client):
assert data is not None
assert data.get("success") is False
assert "not authorized" in data.get("message", "").lower()
+
+
+# -------------------------
+# LOGIN ENDPOINT TESTS
+# -------------------------
+
+_DEFAULT_PW = "123456"
+_DEFAULT_PW_HASH = hashlib.sha256(_DEFAULT_PW.encode()).hexdigest()
+
+
+def test_login_valid_local_credentials(client, api_token):
+ """POST /api/auth/login with correct local password returns 200."""
+ with patch("auth.local_provider.get_setting_value", return_value=_DEFAULT_PW_HASH), \
+ patch("auth.manager.get_setting_value", return_value=False):
+ resp = client.post(
+ "/api/auth/login",
+ headers=auth_headers(api_token),
+ json={"username": "admin", "password": _DEFAULT_PW},
+ )
+ assert resp.status_code == 200
+ data = resp.get_json()
+ assert data is not None
+ assert data.get("success") is True
+ assert data.get("provider") == "local"
+ assert data.get("username") == "admin"
+
+
+def test_login_wrong_local_password(client, api_token):
+ """POST /api/auth/login with incorrect password returns 401."""
+ with patch("auth.local_provider.get_setting_value", return_value=_DEFAULT_PW_HASH), \
+ patch("auth.manager.get_setting_value", return_value=False):
+ resp = client.post(
+ "/api/auth/login",
+ headers=auth_headers(api_token),
+ json={"username": "admin", "password": "totally_wrong"},
+ )
+ assert resp.status_code == 401
+ data = resp.get_json()
+ assert data is not None
+ assert data.get("success") is False
+
+
+def test_login_missing_password_field(client, api_token):
+ """POST /api/auth/login without password returns 422 validation error."""
+ resp = client.post("/api/auth/login", headers=auth_headers(api_token), json={"username": "admin"})
+ assert resp.status_code == 422
+
+
+def test_login_missing_username_field(client, api_token):
+ """POST /api/auth/login without username defaults to admin."""
+ with patch("auth.manager.LdapProvider._read_config", return_value={"enabled": False}), \
+ patch("auth.local_provider.get_setting_value", return_value=_DEFAULT_PW_HASH), \
+ patch("auth.manager.get_setting_value", return_value=False):
+ resp = client.post("/api/auth/login", headers=auth_headers(api_token), json={"password": _DEFAULT_PW})
+ assert resp.status_code == 200
+ data = resp.get_json()
+ assert data.get("username") == "admin"
+
+
+def test_login_empty_body(client, api_token):
+ """POST /api/auth/login with empty JSON object returns 422."""
+ resp = client.post("/api/auth/login", headers=auth_headers(api_token), json={})
+ assert resp.status_code == 422
+
+
+def test_login_no_json_content_type(client, api_token):
+ """POST /api/auth/login with wrong content-type returns 415."""
+ resp = client.post(
+ "/api/auth/login",
+ headers=auth_headers(api_token),
+ data="username=admin&password=123456",
+ content_type="application/x-www-form-urlencoded",
+ )
+ assert resp.status_code == 415
+
+
+def test_login_endpoint_requires_post(client):
+ """GET /api/auth/login should return 405 because the route is POST-only."""
+ resp = client.get("/api/auth/login")
+ assert resp.status_code == 405
+
diff --git a/test/backend/test_auth_providers.py b/test/backend/test_auth_providers.py
new file mode 100644
index 000000000..8d2af113e
--- /dev/null
+++ b/test/backend/test_auth_providers.py
@@ -0,0 +1,314 @@
+"""
+Unit tests for the NetAlertX auth provider stack.
+
+These tests use unittest.mock.patch so they can run without a live database,
+LDAP server, or running Flask application.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import sys
+import os
+from unittest.mock import patch, MagicMock
+
+import pytest
+
+@pytest.fixture(autouse=True)
+def mock_all_mylogs():
+ with patch("auth.local_provider.mylog"), \
+ patch("auth.ldap_provider.mylog"), \
+ patch("auth.manager.mylog"), \
+ patch("helper.mylog"), \
+ patch("logger.mylog"):
+ yield
+
+# Register NetAlertX server directory so bare module imports resolve
+INSTALL_PATH = os.getenv("NETALERTX_APP", "/app")
+sys.path.extend([f"{INSTALL_PATH}/front/plugins", f"{INSTALL_PATH}/server"])
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+def _sha256(value: str) -> str:
+ return hashlib.sha256(value.encode()).hexdigest()
+
+
+# ===========================================================================
+# AuthResult
+# ===========================================================================
+
+class TestAuthResult:
+ def test_ok_factory(self):
+ from auth.base import AuthResult
+ r = AuthResult.ok("alice", "local")
+ assert r.success is True
+ assert r.username == "alice"
+ assert r.provider == "local"
+ assert r.error == ""
+
+ def test_fail_factory(self):
+ from auth.base import AuthResult
+ r = AuthResult.fail("ldap", "bad creds")
+ assert r.success is False
+ assert r.provider == "ldap"
+ assert r.error == "bad creds"
+
+
+# ===========================================================================
+# LocalProvider
+# ===========================================================================
+
+class TestLocalProvider:
+ def test_correct_password(self):
+ from auth.local_provider import LocalProvider
+ pw = "correct_horse"
+ with patch("auth.local_provider.get_setting_value", return_value=_sha256(pw)):
+ result = LocalProvider().authenticate("admin", pw)
+ assert result.success is True
+ assert result.provider == "local"
+ assert result.username == "admin"
+
+ def test_wrong_password(self):
+ from auth.local_provider import LocalProvider
+ pw = "correct_horse"
+ with patch("auth.local_provider.get_setting_value", return_value=_sha256(pw)):
+ result = LocalProvider().authenticate("admin", "wrong_password")
+ assert result.success is False
+ assert result.provider == "local"
+
+ def test_empty_password_rejected(self):
+ from auth.local_provider import LocalProvider
+ with patch("auth.local_provider.get_setting_value", return_value=_sha256("abc")):
+ result = LocalProvider().authenticate("admin", "")
+ assert result.success is False
+
+ def test_missing_stored_hash(self):
+ from auth.local_provider import LocalProvider
+ with patch("auth.local_provider.get_setting_value", return_value=""), \
+ patch("auth.local_provider.mylog"):
+ result = LocalProvider().authenticate("admin", "anything")
+ assert result.success is False
+
+ def test_default_password_hash(self):
+ """The shipped default password '123456' must authenticate correctly."""
+ from auth.local_provider import LocalProvider
+ default_hash = "8d969eef6ecad3c29a3a629280e686cf0c3f5d5a86aff3ca12020c923adc6c92"
+ with patch("auth.local_provider.get_setting_value", return_value=default_hash):
+ result = LocalProvider().authenticate("admin", "123456")
+ assert result.success is True
+
+ def test_any_username_accepted(self):
+ """Local provider does not validate username โ any value should work."""
+ from auth.local_provider import LocalProvider
+ pw = "secret"
+ with patch("auth.local_provider.get_setting_value", return_value=_sha256(pw)):
+ result = LocalProvider().authenticate("anyone", pw)
+ assert result.success is True
+ assert result.username == "anyone"
+
+
+# ===========================================================================
+# LdapProvider
+# ===========================================================================
+
+class TestLdapProvider:
+ """LdapProvider is tested with a fully mocked ldap3 module."""
+
+ def _cfg(self, **overrides):
+ cfg = {
+ "server": "ldap.example.com",
+ "port": 389,
+ "use_ssl": False,
+ "use_start_tls": False,
+ "bind_dn": "cn=svc,dc=example,dc=com",
+ "bind_password": "svc_pass",
+ "base_dn": "dc=example,dc=com",
+ "user_filter": "(uid={username})",
+ "username_attr": "uid",
+ "timeout": 5,
+ }
+ cfg.update(overrides)
+ return cfg
+
+ def _make_ldap3_mock(self, *, bind_success=True, search_entries=1, user_bind_success=True):
+ """Build a mock ldap3 module with the minimal interface needed."""
+ ldap3 = MagicMock()
+ ldap3.NONE = "NONE"
+ ldap3.SIMPLE = "SIMPLE"
+ ldap3.ANONYMOUS = "ANONYMOUS"
+ ldap3.AUTO_BIND_NONE = "AUTO_BIND_NONE"
+ ldap3.SUBTREE = "SUBTREE"
+
+ # Service account connection
+ svc_conn = MagicMock()
+ svc_conn.bind.return_value = bind_success
+ svc_conn.result = {"description": "success"} if bind_success else {"description": "invalidCredentials"}
+ fake_entries = [MagicMock(entry_dn="uid=alice,dc=example,dc=com") for _ in range(search_entries)]
+ svc_conn.entries = fake_entries
+
+ # User connection
+ user_conn = MagicMock()
+ user_conn.bind.return_value = user_bind_success
+ user_conn.result = {"description": "success"} if user_bind_success else {"description": "invalidCredentials"}
+
+ # Connection() always returns svc_conn first, then user_conn
+ ldap3.Connection.side_effect = [svc_conn, user_conn]
+ ldap3.Server.return_value = MagicMock()
+
+ return ldap3, svc_conn, user_conn
+
+ def test_successful_authentication(self):
+ from auth.ldap_provider import LdapProvider
+ provider = LdapProvider()
+ ldap3_mock, _, _ = self._make_ldap3_mock()
+
+ with patch("auth.ldap_provider.ldap3", ldap3_mock), \
+ patch("auth.ldap_provider.get_setting_value", side_effect=lambda k: self._cfg().get(k.replace("LDAP_", ""), "")):
+ provider._read_config = lambda: self._cfg()
+ result = provider.authenticate("alice", "password")
+
+ assert result.success is True
+ assert result.provider == "ldap"
+ assert result.username == "alice"
+
+ def test_service_account_bind_fails(self):
+ from auth.ldap_provider import LdapProvider
+ provider = LdapProvider()
+ ldap3_mock, _, _ = self._make_ldap3_mock(bind_success=False)
+
+ with patch.dict("sys.modules", {"ldap3": ldap3_mock}):
+ provider._read_config = lambda: self._cfg()
+ with pytest.raises(ConnectionError):
+ provider._resolve_user_dn(ldap3_mock, MagicMock(), self._cfg(), "alice")
+
+ def test_user_not_found_in_ldap(self):
+ from auth.ldap_provider import LdapProvider
+ provider = LdapProvider()
+ ldap3_mock, _, _ = self._make_ldap3_mock(search_entries=0)
+
+ with patch.dict("sys.modules", {"ldap3": ldap3_mock}):
+ provider._read_config = lambda: self._cfg()
+ result = provider._resolve_user_dn(ldap3_mock, MagicMock(), self._cfg(), "unknown_user")
+ assert result is None
+
+ def test_user_password_wrong(self):
+ from auth.ldap_provider import LdapProvider
+ provider = LdapProvider()
+
+ ldap3 = MagicMock()
+ ldap3.SIMPLE = "SIMPLE"
+ ldap3.AUTO_BIND_NONE = "AUTO_BIND_NONE"
+
+ user_conn = MagicMock()
+ user_conn.bind.return_value = False
+ user_conn.result = {"description": "invalidCredentials"}
+ ldap3.Connection.return_value = user_conn
+
+ with patch.dict("sys.modules", {"ldap3": ldap3}):
+ provider._read_config = lambda: self._cfg()
+ result = provider._bind_as_user(
+ ldap3, MagicMock(), self._cfg(),
+ "uid=alice,dc=example,dc=com", "alice", "wrong"
+ )
+ assert result.success is False
+
+ def test_missing_server_config(self):
+ from auth.ldap_provider import LdapProvider
+ provider = LdapProvider()
+ with patch("auth.ldap_provider.get_setting_value", return_value=""):
+ result = provider.authenticate("alice", "pw")
+ assert result.success is False
+
+ def test_ldap_filter_escaping(self):
+ """Special characters in username must be escaped in the LDAP filter."""
+ from auth.ldap_provider import _escape_ldap_filter
+ assert _escape_ldap_filter("a*b(c)d\\e\x00f") == r"a\2ab\28c\29d\5ce\00f"
+
+ def test_missing_ldap3_package(self):
+ """Graceful failure when ldap3 is not installed."""
+ from auth.ldap_provider import LdapProvider
+ provider = LdapProvider()
+ provider._read_config = lambda: self._cfg()
+
+ with patch("auth.ldap_provider.ldap3", None):
+ result = provider.authenticate("alice", "pw")
+ assert result.success is False
+
+
+# ===========================================================================
+# AuthManager
+# ===========================================================================
+
+class TestAuthManager:
+ def test_uses_local_when_ldap_disabled(self):
+ from auth.manager import AuthManager
+ with patch("auth.manager.get_setting_value", return_value=False):
+ manager = AuthManager()
+ provider = manager.get_provider()
+ assert provider.name == "local"
+
+ def test_uses_ldap_when_ldap_enabled(self):
+ from auth.manager import AuthManager
+ with patch("auth.manager.LdapProvider._read_config", return_value={"enabled": True}):
+ manager = AuthManager()
+ provider = manager.get_provider()
+ assert provider.name == "ldap"
+
+ def test_authenticate_uses_local_when_ldap_disabled(self):
+ from auth.manager import AuthManager
+ from auth.base import AuthResult
+
+ manager = AuthManager()
+ with patch("auth.manager.get_setting_value", return_value=False), \
+ patch("auth.manager.LocalProvider.authenticate", return_value=AuthResult.ok("admin", "local")) as mock_auth:
+ result = manager.authenticate("admin", "pass")
+
+ mock_auth.assert_called_once_with("admin", "pass")
+ assert result.success is True
+
+ def test_authenticate_uses_ldap_when_ldap_enabled(self):
+ from auth.manager import AuthManager
+ from auth.base import AuthResult
+
+ def mock_settings(key):
+ if key == "SETPWD_enable_password":
+ return True
+ return False
+
+ manager = AuthManager()
+ with patch("auth.manager.LdapProvider._read_config", return_value={"enabled": True, "disable_local_admin": True}), \
+ patch("auth.manager.get_setting_value", side_effect=mock_settings), \
+ patch("auth.manager.LdapProvider.authenticate", return_value=AuthResult.ok("admin", "ldap")) as mock_auth:
+ result = manager.authenticate("admin", "pass")
+
+ mock_auth.assert_called_once_with("admin", "pass")
+ assert result.success is True
+
+ def test_authenticate_fallback_to_local(self):
+ from auth.manager import AuthManager
+ from auth.base import AuthResult
+ from auth.ldap_provider import LdapProvider
+
+ def mock_settings(key):
+ if key == "LDAP_enabled":
+ return True
+ if key == "SETPWD_enable_password":
+ return True
+ if key == "LDAP_disable_local_admin":
+ return True
+ return False
+
+ manager = AuthManager()
+ with patch("auth.manager.LdapProvider._read_config", return_value={"enabled": True, "disable_local_admin": False}), \
+ patch("auth.manager.get_setting_value", side_effect=mock_settings), \
+ patch("auth.manager.LdapProvider.authenticate", return_value=AuthResult.fail("ldap", LdapProvider.USER_NOT_FOUND)), \
+ patch("auth.manager.LocalProvider.authenticate", return_value=AuthResult.ok("admin", "local")) as mock_local_auth:
+ result = manager.authenticate("admin", "pass")
+
+ mock_local_auth.assert_called_once_with("admin", "pass")
+ assert result.success is True
+ assert result.provider == "local"
+
diff --git a/test/docker_tests/test_container_environment.py b/test/docker_tests/test_container_environment.py
index 1ca1eb019..d96b2e269 100644
--- a/test/docker_tests/test_container_environment.py
+++ b/test/docker_tests/test_container_environment.py
@@ -58,7 +58,7 @@
VOLUME_MAP = CONTAINER_TARGETS
-pytestmark = [pytest.mark.docker, pytest.mark.feature_complete]
+pytestmark = [pytest.mark.docker, pytest.mark.feature_complete, pytest.mark.allow_socket]
def _unique_label(prefix: str) -> str:
@@ -174,15 +174,20 @@ def _setup_mount_tree(
# Ensure every mount point is world-writable so arbitrary UID/GID can write
for p in paths.values():
- if p.is_dir():
- p.chmod(0o777)
- for child in p.iterdir():
- if child.is_dir():
- child.chmod(0o777)
- else:
- child.chmod(0o666)
- else:
- p.chmod(0o666)
+ try:
+ if p.is_dir():
+ p.chmod(0o777)
+ for child in p.iterdir():
+ if child.is_dir():
+ child.chmod(0o777)
+ else:
+ child.chmod(0o666)
+ else:
+ p.chmod(0o666)
+ except PermissionError:
+ # Best-effort; if we don't own it (e.g. root-owned from previous run),
+ # we rely on Docker-based chown/cleanup helpers.
+ pass
return paths
@@ -1212,7 +1217,6 @@ def test_mandatory_folders_creation(tmp_path: pathlib.Path) -> None:
"services_run",
"nginx_conf",
]:
- paths[key].chmod(0o777)
_chown_netalertx(paths[key]) # Ensure all directories are owned by netalertx
volumes = _build_volume_args_for_keys(paths, {"data"})
@@ -1256,7 +1260,6 @@ def test_writable_config_validation(tmp_path: pathlib.Path) -> None:
"services_run",
"nginx_conf",
]:
- paths[key].chmod(0o777)
_chown_netalertx(paths[key])
volumes = _build_volume_args_for_keys(paths, {"data"})
diff --git a/test/docker_tests/test_docker_compose_scenarios.py b/test/docker_tests/test_docker_compose_scenarios.py
index da04c285b..713b07883 100644
--- a/test/docker_tests/test_docker_compose_scenarios.py
+++ b/test/docker_tests/test_docker_compose_scenarios.py
@@ -127,7 +127,7 @@ def _discover_host_addresses() -> tuple[str, ...]:
# Test project name prefixes to clean up before runs
_TEST_PROJECT_PREFIXES = ("netalertx-missing", "netalertx-normal", "netalertx-custom", "netalertx-host", "netalertx-ram", "netalertx-dataloss")
-pytestmark = [pytest.mark.docker, pytest.mark.compose]
+pytestmark = [pytest.mark.docker, pytest.mark.compose, pytest.mark.allow_socket]
IMAGE = os.environ.get("NETALERTX_TEST_IMAGE", "netalertx-test")
diff --git a/test/docker_tests/test_docker_compose_unit.py b/test/docker_tests/test_docker_compose_unit.py
index 411654861..778f1d9be 100644
--- a/test/docker_tests/test_docker_compose_unit.py
+++ b/test/docker_tests/test_docker_compose_unit.py
@@ -1,6 +1,8 @@
import subprocess
+import pytest
+@pytest.mark.allow_socket
def test_run_docker_compose_returns_output(monkeypatch, tmp_path):
"""Unit test that verifies `_run_docker_compose` returns a CompletedProcess
instance with an `output` attribute (combined stdout+stderr). This uses
diff --git a/test/docker_tests/test_ldap_ui.py b/test/docker_tests/test_ldap_ui.py
new file mode 100644
index 000000000..2cf60b7e7
--- /dev/null
+++ b/test/docker_tests/test_ldap_ui.py
@@ -0,0 +1,338 @@
+import json
+import os
+import pathlib
+import socket
+import subprocess
+import sys
+import time
+import pytest
+import yaml
+
+if hasattr(socket, "_original_socket"):
+ socket.socket = socket._original_socket
+
+# Add docker_tests to sys.path to import helpers
+sys.path.insert(0, os.path.dirname(__file__))
+
+from test_docker_compose_scenarios import ( # noqa: E402
+ _create_test_data_dirs,
+ CONTAINER_PATHS,
+ IMAGE,
+ _select_custom_ports,
+)
+
+pytestmark = [pytest.mark.docker, pytest.mark.compose, pytest.mark.feature_complete, pytest.mark.allow_socket]
+
+LDIF_CONTENT = """dn: ou=users,dc=example,dc=org
+objectClass: organizationalUnit
+ou: users
+
+dn: uid=testuser,ou=users,dc=example,dc=org
+objectClass: top
+objectClass: account
+objectClass: posixAccount
+objectClass: shadowAccount
+cn: testuser
+uid: testuser
+uidNumber: 10000
+gidNumber: 10000
+homeDirectory: /home/testuser
+userPassword: testpassword
+loginShell: /bin/bash
+"""
+
+
+def test_ldap_ui_login(tmp_path: pathlib.Path):
+ """
+ Spins up an LDAP server and NetAlertX container using Docker Compose,
+ configures LDAP in NetAlertX, and runs a Selenium UI test to log in.
+ """
+ project_name = f"netalertx-ldap-{int(time.time())}"
+
+ repo_root = pathlib.Path(__file__).resolve().parents[2]
+ base_dir = tmp_path
+
+ # Create test data directories for NetAlertX
+ _create_test_data_dirs(base_dir)
+ subprocess.run(["chmod", "-R", "777", str(base_dir / "test_data")], check=True)
+
+ # Create LDAP bootstrap directory
+ ldap_bootstrap_dir = base_dir / "ldap_bootstrap"
+ ldap_bootstrap_dir.mkdir(parents=True, exist_ok=True)
+ ldif_file = ldap_bootstrap_dir / "users.ldif"
+ ldif_file.write_text(LDIF_CONTENT)
+ ldif_file.chmod(0o666)
+
+ # Choose a port
+ http_port = _select_custom_ports()
+ graphql_port = _select_custom_ports({http_port})
+
+ data_volume_name = f"{project_name}_data"
+
+ compose_config = {
+ "version": "3.8",
+ "volumes": {
+ data_volume_name: {}
+ },
+ "services": {
+ "ldap": {
+ "image": "osixia/openldap:1.5.0",
+ "environment": {
+ "LDAP_ORGANISATION": "Example Inc.",
+ "LDAP_DOMAIN": "example.org",
+ "LDAP_ADMIN_PASSWORD": "admin",
+ "LDAP_REMOVE_CONFIG_AFTER_SETUP": "false"
+ }
+ },
+ "netalertx": {
+ "image": IMAGE,
+ "cap_drop": ["ALL"],
+ "cap_add": ["NET_RAW", "NET_ADMIN", "NET_BIND_SERVICE"],
+ "tmpfs": ["/tmp:mode=777"],
+ "ports": [
+ f"{http_port}:20211",
+ f"{graphql_port}:20212"
+ ],
+ "depends_on": ["ldap"],
+ "volumes": [
+ {
+ "type": "volume",
+ "source": data_volume_name,
+ "target": CONTAINER_PATHS["data"],
+ "read_only": False,
+ }
+ ],
+ "environment": {
+ "TZ": "UTC",
+ "NETALERTX_DEBUG": "1",
+ "NETALERTX_CHECK_ONLY": "0", # We want the container to stay running
+ "SKIP_STARTUP_CHECKS": "host optimization",
+ "LOADED_PLUGINS": (
+ "['ARPSCAN', 'AVAHISCAN', 'INTRNT', 'NBTSCAN', 'NSLOOKUP', 'DIGSCAN', "
+ "'SMTP', 'LDAP', 'CSVBCKP', 'CUSTPROP', 'DBCLNP', 'MAINT', 'NEWDEV', "
+ "'NTFPRCS', 'SETPWD', 'SYNC', 'UI', 'VNDRPDT', 'WORKFLOWS']"
+ ),
+ "APP_CONF_OVERRIDE": json.dumps({
+ "LDAP_enabled": True,
+ "LDAP_server": "ldap",
+ "LDAP_port": 389,
+ "LDAP_base_dn": "dc=example,dc=org",
+ "LDAP_user_filter": "(uid={username})",
+ "LDAP_username_attribute": "uid",
+ "LDAP_bind_dn": "cn=admin,dc=example,dc=org",
+ "LDAP_bind_password": "admin",
+ "LDAP_use_ssl": False,
+ "LDAP_use_start_tls": False,
+ "SETPWD_enable_password": True
+ })
+ }
+ }
+ }
+ }
+
+ compose_file = base_dir / "docker-compose.yml"
+ with open(compose_file, "w") as f:
+ yaml.dump(compose_config, f)
+
+ print(f"Starting docker-compose for {project_name} on HTTP port {http_port}...")
+
+ def get_container_name(service):
+ try:
+ cmd = [
+ "docker", "ps",
+ "--filter", f"label=com.docker.compose.project={project_name}",
+ "--filter", f"label=com.docker.compose.service={service}",
+ "--format", "{{.Names}}",
+ ]
+ return subprocess.run(cmd, capture_output=True, text=True, check=True, timeout=10).stdout.strip().split('\n')[0]
+ except Exception:
+ return f"{project_name}-{service}-1"
+
+ try:
+ # Start detached using subprocess directly to avoid test_docker_compose_scenarios.py's auto-teardown
+ subprocess.run(
+ ["docker", "compose", "-f", str(compose_file), "-p", project_name, "up", "-d"],
+ check=True,
+ stdout=sys.stdout,
+ stderr=sys.stderr,
+ timeout=180
+ )
+
+ container_name = get_container_name("netalertx")
+ ldap_container = get_container_name("ldap")
+
+ # Connect the test container to the compose network to access the internal IPs directly
+ is_container = False
+ test_container_id = None
+
+ try:
+ with open("/proc/self/cpuset", "r") as f:
+ content = f.read().strip()
+ if "docker" in content:
+ is_container = True
+ test_container_id = content.split("/")[-1].strip()
+ except Exception:
+ pass
+
+ if not is_container:
+ try:
+ with open("/proc/1/cpuset", "r") as f:
+ content = f.read().strip()
+ if "docker" in content:
+ is_container = True
+ test_container_id = content.split("/")[-1].strip()
+ except Exception:
+ pass
+
+ if not test_container_id:
+ try:
+ with open("/etc/hostname", "r") as f:
+ test_container_id = f.read().strip()
+ except Exception:
+ test_container_id = socket.gethostname()
+
+ network_name = f"{project_name}_default"
+ if is_container:
+ print(f"Connecting test container {test_container_id} to network {network_name}...")
+ res = subprocess.run(
+ ["docker", "network", "connect", network_name, test_container_id],
+ check=False,
+ capture_output=True,
+ text=True,
+ )
+ if res.returncode != 0 and "already exists" not in res.stderr:
+ raise RuntimeError(f"Failed to connect to network {network_name}: {res.stderr}")
+
+ # Check host resolution
+ try:
+ socket.getaddrinfo(container_name, 20211)
+ except socket.gaierror:
+ # Some environments take a second to propagate DNS
+ time.sleep(2)
+ try:
+ socket.getaddrinfo(container_name, 20211)
+ except socket.gaierror as e:
+ print(f"Warning: Could not resolve hostname {container_name} after connecting to network: {e}")
+
+ # Wait for LDAP to become ready
+ print("Waiting for LDAP server to accept connections...")
+ for _ in range(30):
+ try:
+ subprocess.run(
+ ["docker", "exec", ldap_container, "ldapwhoami", "-x", "-D", "cn=admin,dc=example,dc=org", "-w", "admin"],
+ capture_output=True, check=True, timeout=5
+ )
+ print("LDAP server is ready!")
+ break
+ except Exception:
+ time.sleep(2)
+ else:
+ print("Warning: LDAP server did not become ready, proceeding anyway to attempt seeding.")
+
+ target_host = container_name if is_container else "127.0.0.1"
+ target_port = 20211 if is_container else http_port
+
+ print(f"Waiting for NetAlertX internal container on {target_host}:{target_port}...")
+ for _ in range(45):
+ try:
+ with socket.create_connection((target_host, target_port), timeout=2):
+ print("Port is ready!")
+ break
+ except OSError:
+ time.sleep(2)
+ else:
+ print(f"Warning: NetAlertX port {target_port} on {target_host} did not become ready for socket check, but we will proceed to selenium.")
+ print("\n--- BEGIN CONTAINER LOGS ---")
+ subprocess.run(["docker", "logs", container_name], check=False, timeout=10)
+ print("\n--- BEGIN PYTHON STDERR ---")
+ subprocess.run(["docker", "exec", container_name, "cat", "/tmp/log/stderr.log"], check=False, timeout=10)
+ print("--- END LOGS ---\n")
+
+ # Seed LDAP via docker exec
+ print(f"Seeding LDAP users into {ldap_container}...")
+ subprocess.run(
+ ["docker", "exec", "-i", ldap_container, "ldapadd", "-x", "-D", "cn=admin,dc=example,dc=org", "-w", "admin"],
+ input=LDIF_CONTENT,
+ text=True,
+ check=True,
+ timeout=10
+ )
+
+ # Now, the netalertx container should be up. Let's run the selenium test INSIDE the container.
+ # Give PHP/Nginx extra time to fully bootstrap UI
+ time.sleep(5)
+
+ # Force LDAP_enabled and password protection in app.conf so the PHP UI detects it
+ print("Forcing LDAP_enabled to true in app.conf for the UI...")
+ sed_script = (
+ "sed '/LDAP_enabled/d' /data/config/app.conf | sed '/SETPWD_enable_password/d' > /tmp/app.conf "
+ "&& echo 'LDAP_enabled=True' >> /tmp/app.conf "
+ "&& echo 'SETPWD_enable_password=True' >> /tmp/app.conf "
+ "&& cat /tmp/app.conf > /data/config/app.conf"
+ )
+ subprocess.run(
+ ["docker", "exec", "-u", "root", container_name, "/bin/sh", "-c", sed_script],
+ check=True,
+ timeout=10,
+ )
+
+ # Run the selenium test directly targeting the compose stack's accessible address
+ print(f"Running UI test against {target_host} on port {target_port}...")
+ ldap_ui_test = repo_root / "test" / "ui" / "test_ui_ldap_login.py"
+ cmd = [
+ sys.executable, "-m", "pytest", str(ldap_ui_test), "-v", "-s"
+ ]
+
+ env = os.environ.copy()
+ env["BASE_URL"] = f"http://{target_host}:{target_port}"
+ env["UI_BASE_URL"] = f"http://{target_host}:{target_port}"
+
+ result = subprocess.run(
+ cmd,
+ env=env,
+ stdout=subprocess.PIPE,
+ stderr=subprocess.PIPE,
+ text=True,
+ timeout=180
+ )
+
+ print("--- UI Test STDOUT ---")
+ print(result.stdout)
+ print("--- UI Test STDERR ---")
+ print(result.stderr)
+
+ if result.returncode != 0:
+ print("--- PHP ERROR LOG ---")
+ subprocess.run(["docker", "exec", container_name, "cat", "/tmp/log/app.php_errors.log"], check=False)
+ print("--- NGINX ERROR LOG ---")
+ subprocess.run(["docker", "exec", container_name, "cat", "/tmp/log/nginx-error.log"], check=False)
+ print("--- APP LOG (LAST 50) ---")
+ subprocess.run(["docker", "exec", container_name, "tail", "-n", "50", "/tmp/log/app.log"], check=False)
+ print("--- APP.CONF ---")
+ subprocess.run(["docker", "exec", container_name, "cat", "/data/config/app.conf"], check=False)
+
+ # If test fails, try to fetch python/nginx logs from the container
+ logs_cmd = ["docker", "logs", container_name]
+ logs_res = subprocess.run(logs_cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, timeout=30)
+ print("--- NETALERTX LOGS ---")
+ print(logs_res.stdout)
+ print(logs_res.stderr)
+
+ ldap_logs_cmd = ["docker", "logs", f"{project_name}-ldap-1"]
+ ldap_logs_res = subprocess.run(ldap_logs_cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, timeout=30)
+ print("--- LDAP LOGS ---")
+ print(ldap_logs_res.stdout)
+ print(ldap_logs_res.stderr)
+
+ assert result.returncode == 0, "LDAP UI test failed!"
+
+ finally:
+ print("Tearing down docker compose stack...")
+ subprocess.run(
+ ["docker", "compose", "-f", str(compose_file), "-p", project_name, "down", "-v"],
+ stdout=sys.stdout,
+ stderr=sys.stderr,
+ text=True,
+ check=False,
+ timeout=60
+ )
diff --git a/test/docker_tests/test_ports_available.py b/test/docker_tests/test_ports_available.py
index e50b61bd5..666f1bd4d 100644
--- a/test/docker_tests/test_ports_available.py
+++ b/test/docker_tests/test_ports_available.py
@@ -26,7 +26,7 @@
"services_run": "/tmp/run",
}
-pytestmark = [pytest.mark.docker, pytest.mark.feature_complete]
+pytestmark = [pytest.mark.docker, pytest.mark.feature_complete, pytest.mark.allow_socket]
@pytest.fixture(scope="function")
diff --git a/test/server/test_ldap_provider.py b/test/server/test_ldap_provider.py
new file mode 100644
index 000000000..56a4f868f
--- /dev/null
+++ b/test/server/test_ldap_provider.py
@@ -0,0 +1,100 @@
+import unittest
+from unittest.mock import MagicMock, patch
+
+from server.auth.ldap_provider import LdapProvider
+
+class TestLdapProvider(unittest.TestCase):
+ def test_start_tls_before_bind(self):
+ mock_ldap3 = MagicMock()
+ # Create a mock connection object
+ mock_conn = MagicMock()
+ mock_ldap3.Connection.return_value = mock_conn
+
+ # Configure a test scenario where start_tls is enabled
+ provider = LdapProvider()
+
+ cfg = {
+ "use_start_tls": True,
+ "use_ssl": False,
+ "bind_dn": "cn=admin,dc=example,dc=com",
+ "bind_password": "password123",
+ "base_dn": "dc=example,dc=com",
+ "user_filter": "(uid={username})",
+ "username_attr": "uid",
+ "server": "ldap.example.com",
+ "port": 389,
+ "timeout": 5,
+ }
+
+ mock_server = MagicMock()
+
+ # Call the secure connection method
+ provider._create_secure_connection(
+ mock_ldap3,
+ mock_server,
+ cfg,
+ user=cfg["bind_dn"],
+ password=cfg["bind_password"],
+ authentication=mock_ldap3.SIMPLE
+ )
+
+ # Assert start_tls was called
+ mock_conn.start_tls.assert_called_once()
+
+ # Assert bind was called
+ mock_conn.bind.assert_called_once()
+
+ # Determine the order of calls
+ mock_calls = mock_conn.mock_calls
+ start_tls_idx = next(i for i, call in enumerate(mock_calls) if call[0] == 'start_tls')
+ bind_idx = next(i for i, call in enumerate(mock_calls) if call[0] == 'bind')
+
+ # The critical assertion: start_tls must happen BEFORE bind
+ self.assertLess(start_tls_idx, bind_idx, "start_tls MUST be called before bind")
+
+ @patch('server.auth.ldap_provider.ldap3')
+ @patch('server.auth.ldap_provider.ssl')
+ def test_tls_verify_cert(self, mock_ssl, mock_ldap3):
+ provider = LdapProvider()
+
+ # Mock _read_config to return our test cfg
+ cfg = {
+ "server": "ldap.example.com",
+ "port": 636,
+ "use_ssl": True,
+ "use_start_tls": False,
+ "tls_verify_cert": True,
+ "ca_cert_path": "/path/to/ca.crt",
+ "bind_dn": "",
+ "bind_password": "",
+ "base_dn": "dc=example,dc=com",
+ "user_filter": "(uid={username})",
+ "username_attr": "uid",
+ "timeout": 5,
+ }
+ provider._read_config = MagicMock(return_value=cfg)
+
+ provider._resolve_user_dn = MagicMock(return_value="uid=test,dc=example,dc=com")
+ provider._bind_as_user = MagicMock(return_value=MagicMock(success=True))
+
+ mock_tls_obj = MagicMock()
+ mock_ldap3.Tls.return_value = mock_tls_obj
+ mock_ssl.CERT_REQUIRED = 2
+
+ provider.authenticate("testuser", "testpassword")
+
+ # Assert Tls object created with validation and CA path
+ mock_ldap3.Tls.assert_called_once_with(validate=2, ca_certs_file="/path/to/ca.crt")
+
+ # Assert Server object created with tls object
+ mock_ldap3.Server.assert_called_once_with(
+ "ldap.example.com",
+ port=636,
+ use_ssl=True,
+ tls=mock_tls_obj,
+ connect_timeout=5,
+ get_info=mock_ldap3.NONE
+ )
+
+if __name__ == '__main__':
+ unittest.main()
diff --git a/test/server/test_ldap_provider_env.py b/test/server/test_ldap_provider_env.py
new file mode 100644
index 000000000..469c327d5
--- /dev/null
+++ b/test/server/test_ldap_provider_env.py
@@ -0,0 +1,66 @@
+import os
+import unittest
+from unittest.mock import patch
+
+from server.auth.ldap_provider import LdapProvider
+
+
+class TestLdapProviderEnv(unittest.TestCase):
+ @patch("server.auth.ldap_provider.get_setting_value")
+ @patch.dict(os.environ, {}, clear=True)
+ def test_read_config_no_env(self, mock_get_setting):
+ """Test that settings are read from the database if no ENV is set."""
+ mock_get_setting.side_effect = lambda k: {
+ "LDAP_server": "db.example.com",
+ "LDAP_port": "3890",
+ "LDAP_use_ssl": "true",
+ "LDAP_tls_verify_cert": "1",
+ "LDAP_disable_local_admin": "false"
+ }.get(k, None)
+
+ provider = LdapProvider()
+ cfg = provider._read_config()
+
+ self.assertEqual(cfg["server"], "db.example.com")
+ self.assertEqual(cfg["port"], 3890)
+ self.assertEqual(cfg["use_ssl"], True)
+ self.assertEqual(cfg["tls_verify_cert"], True)
+ self.assertEqual(cfg["disable_local_admin"], False)
+
+ @patch("server.auth.ldap_provider.get_setting_value")
+ @patch.dict(os.environ, {
+ "LDAP_SERVER": "env.example.com",
+ "LDAP_PORT": "6360",
+ "LDAP_USE_SSL": "false",
+ "LDAP_TLS_VERIFY_CERT": "0",
+ "LDAP_DISABLE_LOCAL_ADMIN": "true",
+ "LDAP_BIND_DN": "cn=admin,dc=env,dc=com",
+ "LDAP_BIND_PASSWORD": "envpassword"
+ }, clear=True)
+ def test_read_config_with_env(self, mock_get_setting):
+ """Settings always come from get_setting_value (app.conf), env vars are ignored."""
+ mock_get_setting.side_effect = lambda k: {
+ "LDAP_server": "db.example.com",
+ "LDAP_port": "3890",
+ "LDAP_use_ssl": "true",
+ "LDAP_tls_verify_cert": "1",
+ "LDAP_disable_local_admin": "false",
+ "LDAP_bind_dn": "cn=dbadmin,dc=db,dc=com",
+ "LDAP_bind_password": "dbpassword"
+ }.get(k, None)
+
+ provider = LdapProvider()
+ cfg = provider._read_config()
+
+ self.assertEqual(cfg["server"], "db.example.com")
+ self.assertEqual(cfg["port"], 3890)
+ self.assertEqual(cfg["use_ssl"], True)
+ self.assertEqual(cfg["tls_verify_cert"], True)
+ self.assertEqual(cfg["disable_local_admin"], False)
+ self.assertEqual(cfg["bind_dn"], "cn=dbadmin,dc=db,dc=com")
+ # bind_password now reads from get_setting_value, not env vars
+ self.assertEqual(cfg["bind_password"], "dbpassword")
+
+
+if __name__ == '__main__':
+ unittest.main()
diff --git a/test/ui/test_ui_ldap_login.py b/test/ui/test_ui_ldap_login.py
new file mode 100644
index 000000000..60b179804
--- /dev/null
+++ b/test/ui/test_ui_ldap_login.py
@@ -0,0 +1,70 @@
+#!/usr/bin/env python3
+"""
+LDAP Login UI Test
+Tests LDAP authentication login functionality using the UI.
+This test is meant to be run inside the container when LDAP is configured.
+"""
+
+import sys
+import os
+import pytest
+from selenium.webdriver.common.by import By
+
+# Add test directory to path
+sys.path.insert(0, os.path.dirname(__file__))
+
+from test_helpers import BASE_URL, wait_for_page_load # noqa: E402
+
+
+def test_ldap_login_success(driver):
+ """Test: Successful login using LDAP credentials"""
+ print("Navigating to login page...")
+ driver.get(f"{BASE_URL}/index.php")
+ wait_for_page_load(driver)
+
+ # Verify LDAP username field is present
+ username_field = driver.find_elements(By.NAME, "loginusername")
+ if not username_field:
+ pytest.skip("LDAP login field not present; LDAP is likely disabled in this environment.")
+
+ username_input = username_field[0]
+ password_input = driver.find_element(By.NAME, "loginpassword")
+
+ print("Entering credentials...")
+ username_input.send_keys("testuser")
+ password_input.send_keys("testpassword")
+
+ submit_button = driver.find_element(By.CSS_SELECTOR, "button[type='submit']")
+ submit_button.click()
+
+ # Wait for page to respond to form submission
+ print("Waiting for login to complete...")
+
+ # Wait for URL to change away from login page
+ from selenium.webdriver.support.ui import WebDriverWait
+ try:
+ WebDriverWait(driver, 10).until(
+ lambda d: '/devices.php' in d.current_url or '/index.php' not in d.current_url
+ )
+ except Exception as e:
+ print(f"FAILED TO REDIRECT! Current URL: {driver.current_url}")
+ print("--- PAGE BODY TEXT ---")
+ try:
+ print(driver.find_element(By.TAG_NAME, "body").text)
+ except Exception:
+ pass
+ print("--- PAGE LOGS ---")
+ try:
+ for entry in driver.get_log('browser'):
+ print(entry)
+ except Exception:
+ pass
+ raise e
+
+ wait_for_page_load(driver, timeout=5)
+
+ # Should be redirected to devices page or dashboard
+ print(f"Current URL after login: {driver.current_url}")
+ assert '/devices.php' in driver.current_url or '/index.php' not in driver.current_url, \
+ f"Expected redirect after successful LDAP login, got {driver.current_url}"
+ print("LDAP login successful!")
diff --git a/test/ui/test_ui_login.py b/test/ui/test_ui_login.py
index faccdb587..7ca396724 100644
--- a/test/ui/test_ui_login.py
+++ b/test/ui/test_ui_login.py
@@ -17,6 +17,71 @@
from .test_helpers import BASE_URL, wait_for_page_load # noqa: E402
+def _is_setting_line(line, key):
+ """Return True only for 'KEY=value' lines, not 'KEY__metadata=...' lines."""
+ return line.startswith(key + "=") and not line.startswith(key + "__")
+
+
+@pytest.fixture(scope="module", autouse=True)
+def ensure_web_protection():
+ """Ensure web protection is enabled for login tests and restored after."""
+ config_path = "/data/config/app.conf"
+
+ # Read original state
+ was_enabled = False
+ original_password_line = None
+ original_enable_line = None
+
+ if os.path.exists(config_path):
+ with open(config_path, "r") as f:
+ for line in f:
+ if _is_setting_line(line, "SETPWD_enable_password"):
+ original_enable_line = line
+ if "true" in line.lower():
+ was_enabled = True
+ elif _is_setting_line(line, "SETPWD_password"):
+ original_password_line = line
+
+ if not was_enabled:
+ # Enable web protection โ only replace value lines, preserve __metadata
+ if os.path.exists(config_path):
+ with open(config_path, "r") as f:
+ lines = f.readlines()
+ with open(config_path, "w") as f:
+ for line in lines:
+ if _is_setting_line(line, "SETPWD_enable_password"):
+ continue # will rewrite below
+ if _is_setting_line(line, "SETPWD_password"):
+ continue # will rewrite below
+ f.write(line)
+ f.write("SETPWD_enable_password=True\n")
+ if original_password_line:
+ f.write(original_password_line)
+ else:
+ f.write("SETPWD_password='8d969eef6ecad3c29a3a629280e686cf0c3f5d5a86aff3ca12020c923adc6c92'\n")
+ time.sleep(2) # Let disk sync catch up
+
+ yield
+
+ # Restore original state โ only replace value lines, preserve __metadata
+ if not was_enabled:
+ if os.path.exists(config_path):
+ with open(config_path, "r") as f:
+ lines = f.readlines()
+ with open(config_path, "w") as f:
+ for line in lines:
+ if _is_setting_line(line, "SETPWD_enable_password"):
+ continue
+ if _is_setting_line(line, "SETPWD_password"):
+ continue
+ f.write(line)
+ if original_enable_line:
+ f.write(original_enable_line)
+ else:
+ f.write("SETPWD_enable_password=False\n")
+ if original_password_line:
+ f.write(original_password_line)
+
def get_login_password():
"""Get login password from config file or environment
@@ -45,7 +110,7 @@ def get_login_password():
with open(config_path, 'r') as f:
for line in f:
# Only look for SETPWD_password lines (not other config like API keys)
- if 'SETPWD_password' in line and '=' in line:
+ if line.startswith('SETPWD_password'):
# Extract the value between quotes
value = line.split('=', 1)[1].strip()
# Remove quotes
@@ -67,21 +132,19 @@ def get_login_password():
continue
# If we couldn't determine the password from config, try default password
- print("โน Password not determinable from config, trying default passwords...")
+ print("[i] Password not determinable from config, falling back to default password...")
- # For now, return first test password to try
- # Tests will skip if login fails
- return None
+ return "123456"
def require_login_page(driver):
- """Skip the test if the login form is not present (web protection disabled)."""
+ """Assert the login form is present now that web protection is forced ON for testing."""
fields = driver.find_elements(By.NAME, "loginpassword")
if not fields:
- pytest.skip(
- "Web protection is disabled (SETPWD_enable_password != true); "
- "login page is not shown on this instance"
- )
+ print("\n--- PAGE SOURCE ---")
+ print(driver.page_source)
+ print("--- END PAGE SOURCE ---\n")
+ assert fields, "Web protection was enabled but the login page was not shown on this instance"
def perform_login(driver, password=None):
@@ -137,9 +200,6 @@ def test_login_redirects_to_devices(driver):
time.sleep(1)
# Should be redirected to devices page
- if '/devices.php' not in driver.current_url:
- pytest.skip(f"Login failed or not configured. URL: {driver.current_url}")
-
assert '/devices.php' in driver.current_url, \
f"Expected redirect to devices.php, got {driver.current_url}"
@@ -172,9 +232,6 @@ def test_login_with_deep_link_preserves_hash(driver):
current_url = driver.current_url
print(f"URL after login with deep link: {current_url}")
- if '/devices.php' not in current_url:
- pytest.skip(f"Login failed or redirect not configured. URL: {current_url}")
-
# Verify the hash fragment is preserved
assert '#device-123' in current_url, f"Expected #device-123 hash in URL, got {current_url}"
@@ -207,9 +264,6 @@ def test_login_with_deep_link_to_network_page(driver):
current_url = driver.current_url
print(f"URL after login with network.php deep link: {current_url}")
- if '/network.php' not in current_url:
- pytest.skip(f"Login failed or redirect not configured. URL: {current_url}")
-
# Verify the hash fragment is preserved
assert '#settings-panel' in current_url, f"Expected #settings-panel hash in URL, got {current_url}"
@@ -229,8 +283,6 @@ def test_login_without_next_parameter(driver):
# Should redirect to default devices page
current_url = driver.current_url
- if '/devices.php' not in current_url:
- pytest.skip(f"Login failed or not configured. URL: {current_url}")
assert '/devices.php' in current_url, f"Expected default redirect to devices.php, got {current_url}"
diff --git a/test/ui/test_ui_maintenance.py b/test/ui/test_ui_maintenance.py
index 0ebab6968..32c9bbdb0 100644
--- a/test/ui/test_ui_maintenance.py
+++ b/test/ui/test_ui_maintenance.py
@@ -32,13 +32,13 @@ def test_export_csv_button_works(driver):
import os
import glob
- # Use 127.0.0.1 instead of localhost to avoid IPv6 resolution issues in the browser
- # which can lead to "Failed to fetch" if the server is only listening on IPv4.
- target_url = f"{BASE_URL}/maintenance.php".replace("localhost", "127.0.0.1")
- driver.get(target_url)
+ # Initialize session storage by visiting index first
+ driver.get(f"{BASE_URL}/index.php")
wait_for_page_load(driver, timeout=10)
- # Clear any existing downloads
+ target_url = f"{BASE_URL}/maintenance.php"
+ driver.get(target_url)
+ wait_for_page_load(driver, timeout=10) # Clear any existing downloads
download_dir = getattr(driver, 'download_dir', '/tmp/selenium_downloads')
for f in glob.glob(f"{download_dir}/*.csv"):
os.remove(f)