From 72d9646cc4a16f99c427611e29d7bfeece4fad4f Mon Sep 17 00:00:00 2001 From: max Date: Sun, 11 Oct 2026 09:40:47 +0200 Subject: [PATCH] Separate Best Practice rules definition into dedicated best_practice_rules module --- src/hub/best_practice_rules.py | 196 ++++++++++++++++++ src/hub/best_practices.py | 183 ++-------------- tests/test_server_roles_and_best_practices.py | 23 ++ 3 files changed, 234 insertions(+), 168 deletions(-) create mode 100644 src/hub/best_practice_rules.py diff --git a/src/hub/best_practice_rules.py b/src/hub/best_practice_rules.py new file mode 100644 index 0000000..7cebee3 --- /dev/null +++ b/src/hub/best_practice_rules.py @@ -0,0 +1,196 @@ +""" +LOGAR Best Practice & Security Baseline Rules Definition +Contains modular configuration checks, role-specific baselines, +and security hardening rules evaluated across enrolled server nodes. +Separated from the core evaluation engine to allow simple addition, +customization, and maintenance of compliance checks. +""" + +import json +from typing import Dict, Any, List, Optional, Callable + + +class BestPracticeRule: + """Represents a single modular Best Practice configuration check.""" + + def __init__( + self, + rule_id: str, + category: str, + title: str, + description: str, + recommendation: str, + severity: str = "WARNING", # "CRITICAL" | "WARNING" | "RECOMMENDED" + target_role: Optional[str] = None, + target_os: Optional[str] = None, + eval_fn: Optional[Callable[[Dict[str, Any]], Dict[str, Any]]] = None + ): + self.rule_id = rule_id + self.category = category + self.title = title + self.description = description + self.recommendation = recommendation + self.severity = severity + self.target_role = target_role + self.target_os = target_os + self.eval_fn = eval_fn + + def to_dict(self) -> Dict[str, Any]: + return { + "rule_id": self.rule_id, + "category": self.category, + "title": self.title, + "description": self.description, + "recommendation": self.recommendation, + "severity": self.severity, + "target_role": self.target_role, + "target_os": self.target_os + } + + +# ============================================================================ +# Rule Evaluation Functions +# ============================================================================ + +def eval_disk_space(node: Dict[str, Any]) -> Dict[str, Any]: + """BP-SYS-001: Verifies primary volume retains healthy free capacity.""" + pct = float(node.get("disk_percent_free") or 0.0) + if pct <= 0.0: + return {"status": "NOT_APPLICABLE", "message": "Disk capacity metrics pending telemetry collection."} + if pct < 10.0: + return {"status": "FAIL", "message": f"Critical disk headroom: only {pct}% free capacity remaining on primary volume."} + if pct < 20.0: + return {"status": "WARN", "message": f"Low disk headroom: {pct}% free capacity remaining (recommended >= 20%)."} + return {"status": "PASS", "message": f"Healthy disk capacity: {pct}% free capacity available."} + + +def eval_memory_load(node: Dict[str, Any]) -> Dict[str, Any]: + """BP-SYS-002: Verifies RAM load remains safely below critical paging saturation.""" + pct = float(node.get("memory_percent") or 0.0) + if pct <= 0.0: + return {"status": "NOT_APPLICABLE", "message": "Memory metrics pending telemetry collection."} + if pct >= 95.0: + return {"status": "FAIL", "message": f"Exhausted memory headroom: {pct}% active memory consumption."} + if pct >= 88.0: + return {"status": "WARN", "message": f"Elevated memory load: {pct}% utilization (recommended < 85%)."} + return {"status": "PASS", "message": f"Memory utilization optimal: {pct}% in use."} + + +def eval_dc_co_location(node: Dict[str, Any]) -> Dict[str, Any]: + """BP-AD-001: Verifies Active Directory Domain Controllers do not co-locate high-risk services.""" + roles = node.get("server_roles") or [] + if isinstance(roles, str): + try: + roles = json.loads(roles) + except Exception: + roles = [r.strip() for r in roles.split(",") if r.strip()] + + roles_lower = [str(r).lower() for r in roles] + has_ad = any("active directory" in r for r in roles_lower) + if not has_ad: + return {"status": "NOT_APPLICABLE", "message": "Node is not designated as an Active Directory Domain Controller."} + + conflicts = [] + if any("web server" in r or "iis" in r for r in roles_lower): + conflicts.append("Web Server (IIS)") + if any("remote desktop" in r or "terminal" in r for r in roles_lower): + conflicts.append("Remote Desktop Services") + if any("print" in r for r in roles_lower): + conflicts.append("Print & Document Services") + + if conflicts: + return {"status": "FAIL", "message": f"Active Directory Domain Controller co-locates non-recommended roles: {', '.join(conflicts)}."} + return {"status": "PASS", "message": "Domain Controller adheres to dedicated role isolation standards."} + + +def eval_server_classification(node: Dict[str, Any]) -> Dict[str, Any]: + """BP-WIN-002: Verifies operating system classification and role inventory alignment.""" + is_server = bool(node.get("is_server")) + os_type = str(node.get("os_type", "")).lower() + os_name = str(node.get("os_name", "")) + roles = node.get("server_roles") or [] + if isinstance(roles, str): + try: + roles = json.loads(roles) + except Exception: + roles = [] + + if "server" in os_type or "server" in os_name.lower(): + if len(roles) > 0: + return {"status": "PASS", "message": f"Windows Server edition identified ({os_name}) with {len(roles)} detected enterprise roles."} + else: + return {"status": "WARN", "message": f"Windows Server identified ({os_name}), but no server roles currently indexed."} + elif os_type in ("windows", "workstation"): + return {"status": "PASS", "message": f"Workstation node ({os_name or 'Windows Client'}) correctly classified without server role footprint."} + return {"status": "PASS", "message": f"Host classification verified: {os_name or os_type}."} + + +def eval_agent_version(node: Dict[str, Any]) -> Dict[str, Any]: + """BP-SEC-003: Ensures edge forwarder binary version is actively patched.""" + ver = str(node.get("client_version") or "").strip() + up_status = str(node.get("update_status") or "up_to_date").lower() + if not ver: + return {"status": "WARN", "message": "Client version not yet reported to hub."} + if up_status == "outdated": + return {"status": "WARN", "message": f"Edge client version (v{ver}) is outdated relative to hosted hub binaries."} + return {"status": "PASS", "message": f"Edge forwarder binary (v{ver}) is current."} + + +# ============================================================================ +# Baseline Rule Definitions +# ============================================================================ + +BASELINE_RULES: List[BestPracticeRule] = [ + BestPracticeRule( + rule_id="BP-SYS-001", + category="System Health", + title="System Volume Storage Headroom (> 20% Free)", + description="Verifies that server and workstation system drives retain at least 20% free space to ensure unhindered OS paging, log buffering, and update installation.", + recommendation="Expand volume storage or clear transient files to maintain healthy write headroom.", + severity="CRITICAL", + eval_fn=eval_disk_space + ), + BestPracticeRule( + rule_id="BP-SYS-002", + category="System Health", + title="Memory Saturation & Buffer Safety (< 85% Load)", + description="Ensures system RAM utilization remains below critical saturation thresholds to prevent kernel thread swapping and out-of-memory service crashes.", + recommendation="Review high-consumption service processes or allocate additional physical RAM to this server.", + severity="WARNING", + eval_fn=eval_memory_load + ), + BestPracticeRule( + rule_id="BP-AD-001", + category="Active Directory", + title="Domain Controller Role Isolation Baseline", + description="Domain Controllers should be dedicated solely to identity services. Co-locating IIS, Remote Desktop, or Print Spooler introduces lateral privilege escalation attack vectors.", + recommendation="Migrate auxiliary web and print services to separate member servers.", + severity="CRITICAL", + target_role="Active Directory Domain Services", + target_os="windows_server", + eval_fn=eval_dc_co_location + ), + BestPracticeRule( + rule_id="BP-WIN-002", + category="Role Baseline", + title="Operating System Edition & Server Role Alignment", + description="Verifies that nodes are accurately differentiated between client workstations and enterprise servers, and that active server roles are fully inventoried.", + recommendation="Ensure the LOGAR edge forwarder runs with sufficient permissions to query Windows Server Manager.", + severity="RECOMMENDED", + eval_fn=eval_server_classification + ), + BestPracticeRule( + rule_id="BP-SEC-003", + category="Security Baseline", + title="Edge Forwarder Binary Version Baseline", + description="Ensures forwarder nodes run the latest patched agent binary with current TLS 1.3 cryptographic cipher suites and security fixes.", + recommendation="Enable Fleet Auto-Update or update the forwarder executable to the latest hosted release.", + severity="RECOMMENDED", + eval_fn=eval_agent_version + ) +] + + +def get_baseline_rules() -> List[BestPracticeRule]: + """Returns a list of all pre-configured Best Practice rules.""" + return list(BASELINE_RULES) diff --git a/src/hub/best_practices.py b/src/hub/best_practices.py index ba70f64..836a441 100644 --- a/src/hub/best_practices.py +++ b/src/hub/best_practices.py @@ -3,49 +3,20 @@ LOGAR Best Practice Analyzer & Security Baseline Engine Provides a modular, extensible framework for verifying server configuration baselines, role-specific best practices, and security hardening across enrolled fleet nodes. Gated by the 'best_practices' enterprise license capability. +Rule definitions are maintained separately in best_practice_rules.py. """ import json from datetime import datetime, timezone from typing import Dict, Any, List, Optional, Callable - -class BestPracticeRule: - """Represents a single modular Best Practice configuration check.""" - - def __init__( - self, - rule_id: str, - category: str, - title: str, - description: str, - recommendation: str, - severity: str = "WARNING", # "CRITICAL" | "WARNING" | "RECOMMENDED" - target_role: Optional[str] = None, - target_os: Optional[str] = None, - eval_fn: Optional[Callable[[Dict[str, Any]], Dict[str, Any]]] = None - ): - self.rule_id = rule_id - self.category = category - self.title = title - self.description = description - self.recommendation = recommendation - self.severity = severity - self.target_role = target_role - self.target_os = target_os - self.eval_fn = eval_fn - - def to_dict(self) -> Dict[str, Any]: - return { - "rule_id": self.rule_id, - "category": self.category, - "title": self.title, - "description": self.description, - "recommendation": self.recommendation, - "severity": self.severity, - "target_role": self.target_role, - "target_os": self.target_os - } +try: + from src.hub.best_practice_rules import BestPracticeRule, get_baseline_rules +except ImportError: + try: + from best_practice_rules import BestPracticeRule, get_baseline_rules + except ImportError: + from .best_practice_rules import BestPracticeRule, get_baseline_rules class BestPracticeEngine: @@ -59,141 +30,17 @@ class BestPracticeEngine: """Registers a custom Best Practice check. Enables users to define rules later on.""" self._rules[rule.rule_id] = rule + def unregister_rule(self, rule_id: str): + """Removes a rule by its identifier.""" + self._rules.pop(rule_id, None) + def get_rules(self) -> List[Dict[str, Any]]: return [r.to_dict() for r in self._rules.values()] def _load_baseline_rules(self): - # 1. Disk Capacity Health - def eval_disk_space(node: Dict[str, Any]) -> Dict[str, Any]: - pct = float(node.get("disk_percent_free") or 0.0) - if pct <= 0.0: - return {"status": "NOT_APPLICABLE", "message": "Disk capacity metrics pending telemetry collection."} - if pct < 10.0: - return {"status": "FAIL", "message": f"Critical disk headroom: only {pct}% free capacity remaining on primary volume."} - if pct < 20.0: - return {"status": "WARN", "message": f"Low disk headroom: {pct}% free capacity remaining (recommended >= 20%)."} - return {"status": "PASS", "message": f"Healthy disk capacity: {pct}% free capacity available."} - - self.register_rule(BestPracticeRule( - rule_id="BP-SYS-001", - category="System Health", - title="System Volume Storage Headroom (> 20% Free)", - description="Verifies that server and workstation system drives retain at least 20% free space to ensure unhindered OS paging, log buffering, and update installation.", - recommendation="Expand volume storage or clear transient files to maintain healthy write headroom.", - severity="CRITICAL", - eval_fn=eval_disk_space - )) - - # 2. Memory Headroom - def eval_memory_load(node: Dict[str, Any]) -> Dict[str, Any]: - pct = float(node.get("memory_percent") or 0.0) - if pct <= 0.0: - return {"status": "NOT_APPLICABLE", "message": "Memory metrics pending telemetry collection."} - if pct >= 95.0: - return {"status": "FAIL", "message": f"Exhausted memory headroom: {pct}% active memory consumption."} - if pct >= 88.0: - return {"status": "WARN", "message": f"Elevated memory load: {pct}% utilization (recommended < 85%)."} - return {"status": "PASS", "message": f"Memory utilization optimal: {pct}% in use."} - - self.register_rule(BestPracticeRule( - rule_id="BP-SYS-002", - category="System Health", - title="Memory Saturation & Buffer Safety (< 85% Load)", - description="Ensures system RAM utilization remains below critical saturation thresholds to prevent kernel thread swapping and out-of-memory service crashes.", - recommendation="Review high-consumption service processes or allocate additional physical RAM to this server.", - severity="WARNING", - eval_fn=eval_memory_load - )) - - # 3. Server Role Segregation - def eval_dc_co_location(node: Dict[str, Any]) -> Dict[str, Any]: - roles = node.get("server_roles") or [] - if isinstance(roles, str): - try: - roles = json.loads(roles) - except Exception: - roles = [r.strip() for r in roles.split(",") if r.strip()] - - roles_lower = [str(r).lower() for r in roles] - has_ad = any("active directory" in r for r in roles_lower) - if not has_ad: - return {"status": "NOT_APPLICABLE", "message": "Node is not designated as an Active Directory Domain Controller."} - - # Check for high-risk co-located roles on Domain Controllers - conflicts = [] - if any("web server" in r or "iis" in r for r in roles_lower): - conflicts.append("Web Server (IIS)") - if any("remote desktop" in r or "terminal" in r for r in roles_lower): - conflicts.append("Remote Desktop Services") - if any("print" in r for r in roles_lower): - conflicts.append("Print & Document Services") - - if conflicts: - return {"status": "FAIL", "message": f"Active Directory Domain Controller co-locates non-recommended roles: {', '.join(conflicts)}."} - return {"status": "PASS", "message": "Domain Controller adheres to dedicated role isolation standards."} - - self.register_rule(BestPracticeRule( - rule_id="BP-AD-001", - category="Active Directory", - title="Domain Controller Role Isolation Baseline", - description="Domain Controllers should be dedicated solely to identity services. Co-locating IIS, Remote Desktop, or Print Spooler introduces lateral privilege escalation attack vectors.", - recommendation="Migrate auxiliary web and print services to separate member servers.", - severity="CRITICAL", - target_role="Active Directory Domain Services", - target_os="windows_server", - eval_fn=eval_dc_co_location - )) - - # 4. Windows Server Detection and Role Discovery - def eval_server_classification(node: Dict[str, Any]) -> Dict[str, Any]: - is_server = bool(node.get("is_server")) - os_type = str(node.get("os_type", "")).lower() - os_name = str(node.get("os_name", "")) - roles = node.get("server_roles") or [] - if isinstance(roles, str): - try: - roles = json.loads(roles) - except Exception: - roles = [] - - if "server" in os_type or "server" in os_name.lower(): - if len(roles) > 0: - return {"status": "PASS", "message": f"Windows Server edition identified ({os_name}) with {len(roles)} detected enterprise roles."} - else: - return {"status": "WARN", "message": f"Windows Server identified ({os_name}), but no server roles currently indexed."} - elif os_type in ("windows", "workstation"): - return {"status": "PASS", "message": f"Workstation node ({os_name}) correctly classified without server role footprint."} - return {"status": "PASS", "message": f"Host classification verified: {os_name or os_type}."} - - self.register_rule(BestPracticeRule( - rule_id="BP-WIN-002", - category="Role Baseline", - title="Operating System Edition & Server Role Alignment", - description="Verifies that nodes are accurately differentiated between client workstations and enterprise servers, and that active server roles are fully inventoried.", - recommendation="Ensure the LOGAR edge forwarder runs with sufficient permissions to query Windows Server Manager.", - severity="RECOMMENDED", - eval_fn=eval_server_classification - )) - - # 5. Forwarder Agent Freshness - def eval_agent_version(node: Dict[str, Any]) -> Dict[str, Any]: - ver = str(node.get("client_version") or "").strip() - up_status = str(node.get("update_status") or "up_to_date").lower() - if not ver: - return {"status": "WARN", "message": "Client version not yet reported to hub."} - if up_status == "outdated": - return {"status": "WARN", "message": f"Edge client version (v{ver}) is outdated relative to hosted hub binaries."} - return {"status": "PASS", "message": f"Edge forwarder binary (v{ver}) is current."} - - self.register_rule(BestPracticeRule( - rule_id="BP-SEC-003", - category="Security Baseline", - title="Edge Forwarder Binary Version Baseline", - description="Ensures forwarder nodes run the latest patched agent binary with current TLS 1.3 cryptographic cipher suites and security fixes.", - recommendation="Enable Fleet Auto-Update or update the forwarder executable to the latest hosted release.", - severity="RECOMMENDED", - eval_fn=eval_agent_version - )) + """Loads baseline checks defined in the separate best_practice_rules.py module.""" + for rule in get_baseline_rules(): + self.register_rule(rule) def evaluate_node(self, node: Dict[str, Any]) -> List[Dict[str, Any]]: """Evaluates all applicable rules against a single server/node record.""" diff --git a/tests/test_server_roles_and_best_practices.py b/tests/test_server_roles_and_best_practices.py index a537791..e700ae2 100644 --- a/tests/test_server_roles_and_best_practices.py +++ b/tests/test_server_roles_and_best_practices.py @@ -20,6 +20,7 @@ from src.common.system_info import ( get_installed_windows_server_roles ) 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 @@ -119,6 +120,28 @@ class TestServerRolesAndBestPractices(unittest.TestCase): 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 renders HTML when authenticated.""" # Unauthenticated redirects to /login