me0nline
39e0e6f219
CI Test Suite / Run Component Tests & Pipeline Verification (push) Successful in 2m21s
445 lines
21 KiB
Python
445 lines
21 KiB
Python
"""
|
|
Unit and integration tests for LOGAR Server Roles detection,
|
|
Windows Server vs Windows Client differentiation, /client-info subpage,
|
|
and Best Practice baseline engine.
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import json
|
|
import sqlite3
|
|
import unittest
|
|
from unittest.mock import patch, MagicMock
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
from src.common.system_info import (
|
|
get_os_details,
|
|
get_ip_addresses,
|
|
get_full_system_inventory,
|
|
get_installed_windows_server_roles,
|
|
get_linux_distro_info,
|
|
parse_os_release_text
|
|
)
|
|
from src.hub.best_practices import BestPracticeEngine, BestPracticeRule, BEST_PRACTICES_ENGINE
|
|
from src.hub.best_practice_rules import BASELINE_RULES, get_baseline_rules
|
|
from src.hub.web_ui import WebUIRenderer
|
|
from src.hub.Server import app, SERVER_STATE, DEFAULT_DB_FILE
|
|
from src.hub.auth_manager import SessionManager
|
|
|
|
|
|
class TestServerRolesAndBestPractices(unittest.TestCase):
|
|
|
|
def setUp(self):
|
|
self.client = TestClient(app)
|
|
self.db_path = SERVER_STATE.get("config", {}).get("db_path") or DEFAULT_DB_FILE
|
|
if not os.path.exists(self.db_path):
|
|
self.db_path = DEFAULT_DB_FILE
|
|
SERVER_STATE.setdefault("config", {})["db_path"] = self.db_path
|
|
from src.hub.Server import init_db
|
|
init_db(self.db_path)
|
|
self.session_token = SessionManager.generate_session_token("admin", db_path=self.db_path)
|
|
self.auth_headers = {"Authorization": f"Bearer {self.session_token}"}
|
|
|
|
def test_os_details_classification(self):
|
|
"""Verifies OS classification returns valid structure with is_server flag."""
|
|
details = get_os_details()
|
|
self.assertIn("os_type", details)
|
|
self.assertIn("os_name", details)
|
|
self.assertIn("is_server", details)
|
|
self.assertIn("hostname", details)
|
|
self.assertIn("ip_addresses", details)
|
|
self.assertIsInstance(details["is_server"], bool)
|
|
self.assertIn(details["os_type"], ["windows", "windows_server", "linux", "darwin", "unknown"])
|
|
|
|
def test_full_system_inventory(self):
|
|
"""Verifies full system inventory includes CPU, RAM, disk, and server roles."""
|
|
inv = get_full_system_inventory()
|
|
self.assertIn("hostname", inv)
|
|
self.assertIn("os_type", inv)
|
|
self.assertIn("cpu_count", inv)
|
|
self.assertIn("memory_total_mb", inv)
|
|
self.assertIn("disk_percent_free", inv)
|
|
self.assertIn("server_roles", inv)
|
|
self.assertIsInstance(inv["server_roles"], list)
|
|
|
|
def test_best_practices_engine_evaluation(self):
|
|
"""Tests that baseline best practice checks evaluate properly."""
|
|
engine = BestPracticeEngine()
|
|
rules = engine.get_rules()
|
|
self.assertGreaterEqual(len(rules), 4)
|
|
|
|
# Test node with healthy specs
|
|
healthy_node = {
|
|
"client_id": "test-dc-01",
|
|
"hostname": "DC01.corp.local",
|
|
"os_type": "windows_server",
|
|
"os_name": "Windows Server 2022 Datacenter",
|
|
"is_server": True,
|
|
"server_roles": ["Active Directory Domain Services", "DNS Server"],
|
|
"disk_percent_free": 45.0,
|
|
"memory_percent": 55.0,
|
|
"client_version": "1.2.0",
|
|
"update_status": "up_to_date"
|
|
}
|
|
res = engine.evaluate_node(healthy_node)
|
|
self.assertTrue(all(r["status"] == "PASS" for r in res))
|
|
|
|
# Test DC with bad practice (co-located IIS Web Server)
|
|
unhealthy_dc = {
|
|
"client_id": "test-dc-bad",
|
|
"hostname": "DC-BAD.corp.local",
|
|
"os_type": "windows_server",
|
|
"os_name": "Windows Server 2022 Datacenter",
|
|
"is_server": True,
|
|
"server_roles": ["Active Directory Domain Services", "Web Server (IIS)"],
|
|
"disk_percent_free": 5.0, # Critical disk space (< 10%)
|
|
"memory_percent": 96.0, # Exhausted memory (>= 95%)
|
|
"client_version": "1.0.0",
|
|
"update_status": "outdated"
|
|
}
|
|
res_bad = engine.evaluate_node(unhealthy_dc)
|
|
statuses = {r["rule_id"]: r["status"] for r in res_bad}
|
|
self.assertEqual(statuses.get("BP-AD-001"), "FAIL") # DC co-location conflict
|
|
self.assertEqual(statuses.get("BP-SYS-001"), "FAIL") # Critical disk
|
|
self.assertEqual(statuses.get("BP-SYS-002"), "FAIL") # Exhausted RAM
|
|
self.assertEqual(statuses.get("BP-SEC-003"), "WARN") # Outdated client
|
|
|
|
def test_best_practices_custom_rule_extensibility(self):
|
|
"""Tests user extensibility: custom rules can be defined and registered dynamically."""
|
|
engine = BestPracticeEngine()
|
|
custom_rule = BestPracticeRule(
|
|
rule_id="BP-CUSTOM-001",
|
|
category="Custom Compliance",
|
|
title="Custom Domain Suffix Enforcement",
|
|
description="All fleet nodes must belong to the internal domain.",
|
|
recommendation="Join host to corp.internal.",
|
|
severity="WARNING",
|
|
eval_fn=lambda node: {"status": "PASS" if "corp.internal" in node.get("hostname", "") else "FAIL", "message": "Domain check."}
|
|
)
|
|
engine.register_rule(custom_rule)
|
|
rules = engine.get_rules()
|
|
self.assertTrue(any(r["rule_id"] == "BP-CUSTOM-001" for r in rules))
|
|
|
|
eval_pass = engine.evaluate_node({"hostname": "server1.corp.internal"})
|
|
self.assertTrue(any(r["rule_id"] == "BP-CUSTOM-001" and r["status"] == "PASS" for r in eval_pass))
|
|
|
|
eval_fail = engine.evaluate_node({"hostname": "server1.public.com"})
|
|
self.assertTrue(any(r["rule_id"] == "BP-CUSTOM-001" and r["status"] == "FAIL" for r in eval_fail))
|
|
|
|
def test_rules_defined_in_separate_file(self):
|
|
"""Verifies Best Practice rules are defined in separate best_practice_rules module."""
|
|
rules = get_baseline_rules()
|
|
self.assertGreaterEqual(len(rules), 5)
|
|
for r in rules:
|
|
self.assertIsInstance(r, BestPracticeRule)
|
|
self.assertTrue(r.rule_id.startswith("BP-"))
|
|
self.assertTrue(r.title)
|
|
self.assertTrue(r.description)
|
|
self.assertTrue(r.recommendation)
|
|
self.assertIn(r.severity, ["CRITICAL", "WARNING", "RECOMMENDED"])
|
|
self.assertIsNotNone(r.eval_fn)
|
|
|
|
# Engine automatically registers all rules from best_practice_rules module
|
|
engine = BestPracticeEngine()
|
|
loaded_ids = [r["rule_id"] for r in engine.get_rules()]
|
|
self.assertIn("BP-SYS-001", loaded_ids)
|
|
self.assertIn("BP-SYS-002", loaded_ids)
|
|
self.assertIn("BP-AD-001", loaded_ids)
|
|
self.assertIn("BP-WIN-002", loaded_ids)
|
|
self.assertIn("BP-SEC-003", loaded_ids)
|
|
|
|
def test_client_info_html_rendering(self):
|
|
"""Verifies GET /client-info redirects to /login when unauthenticated, and redirects to /webui#server-roles when authenticated."""
|
|
# Unauthenticated redirects to /login
|
|
res_unauth = self.client.get("/client-info", follow_redirects=False)
|
|
self.assertEqual(res_unauth.status_code, 303)
|
|
self.assertEqual(res_unauth.headers.get("location"), "/login")
|
|
|
|
# Authenticated redirects to /webui#server-roles
|
|
auth_client = TestClient(app)
|
|
auth_client.cookies.set(SessionManager.SESSION_COOKIE_NAME, self.session_token)
|
|
|
|
res1 = auth_client.get("/client-info", follow_redirects=False)
|
|
self.assertEqual(res1.status_code, 303)
|
|
self.assertEqual(res1.headers.get("location"), "/webui#server-roles")
|
|
|
|
res2 = auth_client.get("/webui/client-info", follow_redirects=False)
|
|
self.assertEqual(res2.status_code, 303)
|
|
self.assertEqual(res2.headers.get("location"), "/webui#server-roles")
|
|
|
|
# Following redirect leads to main dashboard with Server Roles tab pane
|
|
res_follow = auth_client.get("/client-info", follow_redirects=True)
|
|
self.assertEqual(res_follow.status_code, 200)
|
|
self.assertIn("id=\"tabPane-server-roles\"", res_follow.text)
|
|
self.assertIn("Node Telemetry & Server Roles Inventory", res_follow.text)
|
|
|
|
def test_dashboard_tabs_structure(self):
|
|
"""Verifies primary tabs (Fleet Forwarders, Server Roles & Client Directory, Events, Administration) and secondary admin subtabs (Hermes, Sub-CA, DB, Updates)."""
|
|
res = self.client.get("/webui", cookies={SessionManager.SESSION_COOKIE_NAME: self.session_token})
|
|
self.assertEqual(res.status_code, 200)
|
|
html = res.text
|
|
|
|
# Primary navigation tabs (only 4 tabs on top row)
|
|
self.assertIn("id=\"mainTabsNav\"", html)
|
|
self.assertIn("id=\"tabBtn-fleet\"", html)
|
|
self.assertIn(">💻 Fleet Forwarders</button>", html)
|
|
self.assertIn("id=\"tabBtn-server-roles\"", html)
|
|
self.assertIn(">🖥️ Server Roles & Client Directory</button>", html)
|
|
self.assertIn("id=\"tabBtn-events\"", html)
|
|
self.assertIn(">📊 Events</button>", html)
|
|
self.assertIn("id=\"tabBtn-admin\"", html)
|
|
self.assertIn(">⚙️ Administration", html)
|
|
|
|
# Administration sub-navigation tabs (row below)
|
|
self.assertIn("id=\"adminSubtabs\"", html)
|
|
self.assertNotIn("⚙️ Administration:</span>", html)
|
|
self.assertIn("id=\"tabBtn-baseline\"", html)
|
|
self.assertIn(">🛡️ Baseline Config</button>", html)
|
|
self.assertIn("id=\"tabBtn-hermes\"", html)
|
|
self.assertIn(">🤖 Hermes AI & Tokens</button>", html)
|
|
self.assertIn("id=\"tabBtn-pki\"", html)
|
|
self.assertIn(">📜 Sub-CA & Licensing</button>", html)
|
|
self.assertIn("id=\"tabBtn-db\"", html)
|
|
self.assertIn(">💾 DB Engine</button>", html)
|
|
self.assertIn("id=\"tabBtn-updates\"", html)
|
|
self.assertIn(">📦 Updates & Distribution", html)
|
|
|
|
# Tab panes
|
|
self.assertIn("id=\"tabPane-fleet\"", html)
|
|
self.assertIn("id=\"tabPane-server-roles\"", html)
|
|
self.assertIn("id=\"tabPane-events\"", html)
|
|
self.assertIn("id=\"tabPane-baseline\"", html)
|
|
self.assertIn("id=\"tabPane-pki\"", html)
|
|
self.assertIn("id=\"tabPane-db\"", html)
|
|
self.assertIn("id=\"tabPane-updates\"", html)
|
|
self.assertIn("id=\"tabPane-hermes\"", html)
|
|
|
|
# Telemetry JSON inspector modal
|
|
self.assertIn("id=\"telemetryModal\"", html)
|
|
self.assertIn("id=\"modalJsonContent\"", html)
|
|
|
|
# Server Roles & Client Directory tab should not contain locked/inactive Best Practice Baseline card or banner
|
|
server_roles_pane = html.split('id="tabPane-server-roles"')[1].split('id="tabPane-')[0]
|
|
self.assertNotIn("Best Practice Baseline", server_roles_pane)
|
|
self.assertNotIn("Requires 'best_practices' license capability", server_roles_pane)
|
|
|
|
def test_api_client_info_endpoint(self):
|
|
"""Verifies GET /api/client-info provides structured JSON inventory."""
|
|
res = self.client.get("/api/client-info", headers=self.auth_headers)
|
|
self.assertEqual(res.status_code, 200)
|
|
data = res.json()
|
|
self.assertEqual(data.get("status"), "success")
|
|
self.assertIn("total_nodes", data)
|
|
self.assertIn("windows_servers", data)
|
|
self.assertIn("windows_workstations", data)
|
|
self.assertIn("linux_nodes", data)
|
|
self.assertIn("clients", data)
|
|
|
|
def test_api_best_practices_status_unlicensed(self):
|
|
"""Verifies /api/best-practices/status returns locked when unlicensed."""
|
|
with patch("src.hub.Server.is_feature_licensed", return_value=False):
|
|
res = self.client.get("/api/best-practices/status", headers=self.auth_headers)
|
|
self.assertEqual(res.status_code, 200)
|
|
data = res.json()
|
|
self.assertEqual(data.get("status"), "locked")
|
|
self.assertFalse(data.get("licensed"))
|
|
self.assertIn("enterprise capability", data.get("message", "").lower())
|
|
|
|
def test_api_best_practices_status_licensed(self):
|
|
"""Verifies /api/best-practices/status returns active evaluation when feature is licensed."""
|
|
with patch("src.hub.Server.is_feature_licensed", return_value=True):
|
|
res = self.client.get("/api/best-practices/status", headers=self.auth_headers)
|
|
self.assertEqual(res.status_code, 200)
|
|
data = res.json()
|
|
self.assertEqual(data.get("status"), "active")
|
|
self.assertTrue(data.get("licensed"))
|
|
self.assertIn("evaluation", data)
|
|
eval_data = data["evaluation"]
|
|
self.assertIn("score", eval_data)
|
|
self.assertIn("pass_count", eval_data)
|
|
self.assertIn("warn_count", eval_data)
|
|
self.assertIn("fail_count", eval_data)
|
|
|
|
def test_admin_version_info_modal(self):
|
|
"""Verifies admin dropdown menu contains Version & System Info option and renders telemetry modal in /webui."""
|
|
from src.version import LOGAR_VERSION
|
|
|
|
auth_client = TestClient(app)
|
|
auth_client.cookies.set(SessionManager.SESSION_COOKIE_NAME, self.session_token)
|
|
|
|
# Check /webui
|
|
res_webui = auth_client.get("/webui")
|
|
self.assertEqual(res_webui.status_code, 200)
|
|
self.assertIn("id=\"userDropdown\"", res_webui.text)
|
|
self.assertIn("Version & System Info", res_webui.text)
|
|
self.assertIn("openVersionInfoModal()", res_webui.text)
|
|
self.assertIn("id=\"versionInfoModal\"", res_webui.text)
|
|
self.assertIn(f"v{LOGAR_VERSION}", res_webui.text)
|
|
self.assertIn("Server Hub Release", res_webui.text)
|
|
self.assertIn("Hosted Client Binaries", res_webui.text)
|
|
self.assertIn("Python Environment:", res_webui.text)
|
|
self.assertIn("Host Platform & OS:", res_webui.text)
|
|
self.assertIn("TLS 1.3 mTLS Encrypted", res_webui.text)
|
|
self.assertNotIn("Repository Update Mode:", res_webui.text)
|
|
self.assertNotIn("Distribution Source:", res_webui.text)
|
|
self.assertNotIn("Storage Backend Engine:", res_webui.text)
|
|
self.assertNotIn("License Status & Seats:", res_webui.text)
|
|
|
|
# Check /client-info redirects to /webui#server-roles and shows the version info modal
|
|
res_client_info = auth_client.get("/client-info", follow_redirects=True)
|
|
self.assertEqual(res_client_info.status_code, 200)
|
|
self.assertIn("id=\"userDropdown\"", res_client_info.text)
|
|
self.assertIn("Version & System Info", res_client_info.text)
|
|
self.assertIn("id=\"versionInfoModal\"", res_client_info.text)
|
|
self.assertIn(f"v{LOGAR_VERSION}", res_client_info.text)
|
|
|
|
def test_linux_distro_collection_ubuntu(self):
|
|
"""Verifies parsing of standard Ubuntu os-release data."""
|
|
ubuntu_os_release = (
|
|
'NAME="Ubuntu"\n'
|
|
'VERSION="22.04.4 LTS (Jammy Jellyfish)"\n'
|
|
'ID=ubuntu\n'
|
|
'ID_LIKE=debian\n'
|
|
'PRETTY_NAME="Ubuntu 22.04.4 LTS"\n'
|
|
'VERSION_ID="22.04"\n'
|
|
'VERSION_CODENAME=jammy\n'
|
|
)
|
|
import tempfile
|
|
with tempfile.NamedTemporaryFile("w", delete=False, encoding="utf-8") as f:
|
|
f.write(ubuntu_os_release)
|
|
tmp_path = f.name
|
|
try:
|
|
info = get_linux_distro_info(os_release_path=tmp_path)
|
|
self.assertEqual(info["distro"], "Ubuntu")
|
|
self.assertEqual(info["distro_version"], "22.04")
|
|
self.assertEqual(info["distro_id"], "ubuntu")
|
|
self.assertEqual(info["distro_codename"], "jammy")
|
|
self.assertEqual(info["pretty_name"], "Ubuntu 22.04.4 LTS")
|
|
finally:
|
|
if os.path.exists(tmp_path):
|
|
os.unlink(tmp_path)
|
|
|
|
def test_linux_distro_collection_debian(self):
|
|
"""Verifies parsing of standard Debian os-release data."""
|
|
debian_os_release = (
|
|
'PRETTY_NAME="Debian GNU/Linux 12 (bookworm)"\n'
|
|
'NAME="Debian GNU/Linux"\n'
|
|
'VERSION_ID="12"\n'
|
|
'VERSION="12 (bookworm)"\n'
|
|
'VERSION_CODENAME=bookworm\n'
|
|
'ID=debian\n'
|
|
)
|
|
import tempfile
|
|
with tempfile.NamedTemporaryFile("w", delete=False, encoding="utf-8") as f:
|
|
f.write(debian_os_release)
|
|
tmp_path = f.name
|
|
try:
|
|
info = get_linux_distro_info(os_release_path=tmp_path)
|
|
self.assertEqual(info["distro"], "Debian GNU/Linux")
|
|
self.assertEqual(info["distro_version"], "12")
|
|
self.assertEqual(info["distro_id"], "debian")
|
|
self.assertEqual(info["distro_codename"], "bookworm")
|
|
self.assertEqual(info["pretty_name"], "Debian GNU/Linux 12 (bookworm)")
|
|
finally:
|
|
if os.path.exists(tmp_path):
|
|
os.unlink(tmp_path)
|
|
|
|
def test_linux_distro_collection_rhel_and_alpine(self):
|
|
"""Verifies parsing of RHEL and Alpine os-release data."""
|
|
rhel_release = (
|
|
'NAME="Red Hat Enterprise Linux"\n'
|
|
'VERSION="9.3 (Plow)"\n'
|
|
'ID="rhel"\n'
|
|
'ID_LIKE="fedora"\n'
|
|
'VERSION_ID="9.3"\n'
|
|
'PRETTY_NAME="Red Hat Enterprise Linux 9.3 (Plow)"\n'
|
|
)
|
|
alpine_release = (
|
|
'NAME="Alpine Linux"\n'
|
|
'ID=alpine\n'
|
|
'VERSION_ID=3.19.1\n'
|
|
'PRETTY_NAME="Alpine Linux v3.19"\n'
|
|
)
|
|
import tempfile
|
|
with tempfile.NamedTemporaryFile("w", delete=False, encoding="utf-8") as f:
|
|
f.write(rhel_release)
|
|
rhel_path = f.name
|
|
with tempfile.NamedTemporaryFile("w", delete=False, encoding="utf-8") as f:
|
|
f.write(alpine_release)
|
|
alpine_path = f.name
|
|
try:
|
|
rhel_info = get_linux_distro_info(os_release_path=rhel_path)
|
|
self.assertEqual(rhel_info["distro"], "Red Hat Enterprise Linux")
|
|
self.assertEqual(rhel_info["distro_version"], "9.3")
|
|
self.assertEqual(rhel_info["distro_id"], "rhel")
|
|
|
|
alpine_info = get_linux_distro_info(os_release_path=alpine_path)
|
|
self.assertEqual(alpine_info["distro"], "Alpine Linux")
|
|
self.assertEqual(alpine_info["distro_version"], "3.19.1")
|
|
self.assertEqual(alpine_info["distro_id"], "alpine")
|
|
finally:
|
|
if os.path.exists(rhel_path):
|
|
os.unlink(rhel_path)
|
|
if os.path.exists(alpine_path):
|
|
os.unlink(alpine_path)
|
|
|
|
def test_get_os_details_linux_simulation(self):
|
|
"""Tests get_os_details when simulated on Linux."""
|
|
ubuntu_os_release = (
|
|
'NAME="Ubuntu"\n'
|
|
'VERSION="24.04 LTS (Noble Numbat)"\n'
|
|
'ID=ubuntu\n'
|
|
'VERSION_ID="24.04"\n'
|
|
'PRETTY_NAME="Ubuntu 24.04 LTS"\n'
|
|
'VERSION_CODENAME=noble\n'
|
|
)
|
|
import tempfile
|
|
with tempfile.NamedTemporaryFile("w", delete=False, encoding="utf-8") as f:
|
|
f.write(ubuntu_os_release)
|
|
tmp_path = f.name
|
|
try:
|
|
with patch("sys.platform", "linux"):
|
|
details = get_os_details(os_release_path=tmp_path)
|
|
self.assertEqual(details["os_type"], "linux")
|
|
self.assertTrue(details["is_server"])
|
|
self.assertEqual(details["distro"], "Ubuntu")
|
|
self.assertEqual(details["distro_version"], "24.04")
|
|
self.assertEqual(details["distro_id"], "ubuntu")
|
|
self.assertEqual(details["distro_codename"], "noble")
|
|
self.assertEqual(details["os_name"], "Ubuntu 24.04 LTS")
|
|
finally:
|
|
if os.path.exists(tmp_path):
|
|
os.unlink(tmp_path)
|
|
|
|
def test_webui_renders_linux_distro_badge_and_specs(self):
|
|
"""Verifies /webui properly formats Linux distribution badges and specs."""
|
|
db_path = self.db_path
|
|
conn = sqlite3.connect(db_path)
|
|
c = conn.cursor()
|
|
c.execute("""
|
|
INSERT OR REPLACE INTO clients (
|
|
client_id, hostname, os_type, cert_fingerprint, status, first_seen, last_seen,
|
|
client_version, os_name, is_server, server_roles, client_info_json
|
|
) VALUES (
|
|
'test-linux-node-01', 'ubuntu-srv-01.corp.internal', 'linux', 'FP-LINUX-TEST-01', 'active',
|
|
'2026-10-11 10:00:00', '2026-10-11 10:05:00', '1.2.0', 'Ubuntu 22.04.4 LTS', 1,
|
|
'["Web Server (Nginx)", "SSH Remote Access"]',
|
|
'{"distro": "Ubuntu", "distro_version": "22.04", "distro_id": "ubuntu", "distro_codename": "jammy", "architecture": "x86_64", "cpu_count": 4, "memory_total_mb": 8192}'
|
|
)
|
|
""")
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
auth_client = TestClient(app)
|
|
auth_client.cookies.set(SessionManager.SESSION_COOKIE_NAME, self.session_token)
|
|
res = auth_client.get("/webui")
|
|
self.assertEqual(res.status_code, 200)
|
|
self.assertIn("ubuntu-srv-01.corp.internal", res.text)
|
|
self.assertIn("Ubuntu 22.04.4 LTS", res.text)
|
|
self.assertIn("UBUNTU", res.text)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|