Collect Linux distribution and version telemetry across forwarders and web UI

tests/test_server_roles_and_best_practices.py

@@ -17,7 +17,9 @@ from src.common.system_info import (
     get_os_details,
     get_ip_addresses,
     get_full_system_inventory,
-    get_installed_windows_server_roles
+    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
@@ -283,6 +285,151 @@ class TestServerRolesAndBestPractices(unittest.TestCase):
         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()