Separate Best Practice rules definition into dedicated best_practice_rules module

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)

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
 
-
+try:
-class BestPracticeRule:
+    from src.hub.best_practice_rules import BestPracticeRule, get_baseline_rules
-    """Represents a single modular Best Practice configuration check."""
+except ImportError:
-
+    try:
-    def __init__(
+        from best_practice_rules import BestPracticeRule, get_baseline_rules
-        self,
+    except ImportError:
-        rule_id: str,
+        from .best_practice_rules import BestPracticeRule, get_baseline_rules
-        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
-        }
 
 
 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
+        """Loads baseline checks defined in the separate best_practice_rules.py module."""
-        def eval_disk_space(node: Dict[str, Any]) -> Dict[str, Any]:
+        for rule in get_baseline_rules():
-            pct = float(node.get("disk_percent_free") or 0.0)
+            self.register_rule(rule)
-            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
-        ))
 
     def evaluate_node(self, node: Dict[str, Any]) -> List[Dict[str, Any]]:
         """Evaluates all applicable rules against a single server/node record."""

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