diff --git a/src/clients/client_transport.py b/src/clients/client_transport.py index a62a3be..ba28eed 100644 --- a/src/clients/client_transport.py +++ b/src/clients/client_transport.py @@ -333,12 +333,18 @@ def send_encrypted_logs_over_socket( if expected_fp and actual_fp != expected_fp: raise ValueError(f"Server fingerprint mismatch! Expected {expected_fp}, but key has {actual_fp}.") + inventory = get_full_system_inventory() payload = { "server": actual_machine_id, "timestamp": datetime.now(timezone.utc).isoformat(), "logs": logs, "metrics": collect_system_metrics(), - "version": LOGAR_VERSION + "version": LOGAR_VERSION, + "os_type": inventory.get("os_type", node_os), + "os_name": inventory.get("os_name", ""), + "is_server": inventory.get("is_server", False), + "server_roles": inventory.get("server_roles", []), + "client_info": inventory } payload_json = json.dumps(payload) pgp_msg = pgpy.PGPMessage.new(payload_json) diff --git a/src/common/system_info.py b/src/common/system_info.py index e96b49e..f4c7a67 100644 --- a/src/common/system_info.py +++ b/src/common/system_info.py @@ -10,6 +10,7 @@ import json import socket import shutil import platform +import re import subprocess from typing import Dict, Any, List, Optional @@ -41,7 +42,154 @@ def get_ip_addresses() -> List[str]: return sorted(list(ips)) -def get_os_details() -> Dict[str, Any]: +def parse_os_release_text(content: str) -> Dict[str, str]: + """Parses key=value lines from standard Linux /etc/os-release files.""" + data = {} + for line in content.splitlines(): + line = line.strip() + if not line or line.startswith("#") or "=" not in line: + continue + k, v = line.split("=", 1) + k = k.strip() + v = v.strip().strip('"\'') + data[k] = v + return data + + +def get_linux_distro_info(os_release_path: Optional[str] = None) -> Dict[str, str]: + """ + Accurately detects Linux distribution name, version, ID, and codename. + Uses platform.freedesktop_os_release() (Python 3.10+), followed by standard + /etc/os-release and /usr/lib/os-release parsing, with legacy release file fallbacks. + """ + res = { + "distro": "", + "distro_version": "", + "distro_id": "", + "distro_codename": "", + "pretty_name": "" + } + + raw: Dict[str, str] = {} + + if os_release_path and os.path.exists(os_release_path): + try: + with open(os_release_path, "r", encoding="utf-8", errors="replace") as f: + raw = parse_os_release_text(f.read()) + except Exception: + pass + + if not raw: + # 1. Try standard library platform.freedesktop_os_release() (Python 3.10+) + if hasattr(platform, "freedesktop_os_release"): + try: + raw = dict(platform.freedesktop_os_release()) + except Exception: + pass + + if not raw: + # 2. Check standard /etc/os-release and /usr/lib/os-release + for fpath in ("/etc/os-release", "/usr/lib/os-release"): + if os.path.exists(fpath): + try: + with open(fpath, "r", encoding="utf-8", errors="replace") as f: + raw = parse_os_release_text(f.read()) + if raw: + break + except Exception: + pass + + if raw: + distro = raw.get("NAME") or raw.get("ID", "").capitalize() or "Linux" + version = raw.get("VERSION_ID") or raw.get("VERSION") or "" + pretty = raw.get("PRETTY_NAME") or f"{distro} {version}".strip() + distro_id = raw.get("ID", "").lower() + codename = raw.get("VERSION_CODENAME") or "" + if not codename and "(" in raw.get("VERSION", ""): + m = re.search(r"\(([^)]+)\)", raw.get("VERSION", "")) + if m: + codename = m.group(1).lower() + + res["distro"] = distro + res["distro_version"] = version + res["distro_id"] = distro_id + res["distro_codename"] = codename + res["pretty_name"] = pretty + return res + + # 3. Legacy file fallbacks for older or non-systemd Linux distributions + for rh_file in ("/etc/redhat-release", "/etc/centos-release", "/etc/system-release"): + if os.path.exists(rh_file): + try: + with open(rh_file, "r", encoding="utf-8", errors="replace") as f: + content = f.read().strip() + if content: + res["pretty_name"] = content + m = re.search(r"release\s+([\d\.]+)", content, re.IGNORECASE) + if m: + res["distro_version"] = m.group(1) + parts = content.split("release")[0].strip() + res["distro"] = parts or "Red Hat / CentOS" + res["distro_id"] = "rhel" if "red hat" in content.lower() else ("centos" if "centos" in content.lower() else "linux") + return res + except Exception: + pass + + if os.path.exists("/etc/debian_version"): + try: + with open("/etc/debian_version", "r", encoding="utf-8", errors="replace") as f: + ver = f.read().strip() + res["distro"] = "Debian" + res["distro_version"] = ver + res["distro_id"] = "debian" + res["pretty_name"] = f"Debian {ver}".strip() + return res + except Exception: + pass + + if os.path.exists("/etc/alpine-release"): + try: + with open("/etc/alpine-release", "r", encoding="utf-8", errors="replace") as f: + ver = f.read().strip() + res["distro"] = "Alpine Linux" + res["distro_version"] = ver + res["distro_id"] = "alpine" + res["pretty_name"] = f"Alpine Linux {ver}".strip() + return res + except Exception: + pass + + if os.path.exists("/etc/arch-release"): + res["distro"] = "Arch Linux" + res["distro_version"] = "Rolling" + res["distro_id"] = "arch" + res["pretty_name"] = "Arch Linux" + return res + + if os.path.exists("/etc/SuSE-release"): + try: + with open("/etc/SuSE-release", "r", encoding="utf-8", errors="replace") as f: + content = f.read() + m = re.search(r"VERSION\s*=\s*([\d\.]+)", content) + ver = m.group(1) if m else "" + res["distro"] = "openSUSE" + res["distro_version"] = ver + res["distro_id"] = "opensuse" + res["pretty_name"] = f"openSUSE {ver}".strip() + return res + except Exception: + pass + + # Generic Linux fallback + kernel_ver = platform.release() + res["distro"] = "Linux" + res["distro_version"] = kernel_ver + res["distro_id"] = "linux" + res["pretty_name"] = f"Linux {kernel_ver}".strip() + return res + + +def get_os_details(os_release_path: Optional[str] = None) -> Dict[str, Any]: """ Accurately classifies the operating system, explicitly differentiating between Windows Client (Workstation) and Windows Server, as well as Linux distributions. @@ -51,8 +199,12 @@ def get_os_details() -> Dict[str, Any]: "os_type": "windows_server" | "windows" | "linux" | "darwin" | "unknown", "os_name": str, # e.g. "Windows Server 2022 Datacenter", "Windows 11 Pro", "Ubuntu 22.04 LTS" "is_server": bool, # True for Windows Server or Linux Server; False for desktop clients + "distro": str, # e.g. "Ubuntu", "Debian GNU/Linux", "Red Hat Enterprise Linux" + "distro_version": str,# e.g. "22.04", "12", "9.3" + "distro_id": str, # e.g. "ubuntu", "debian", "rhel", "alpine" + "distro_codename": str,# e.g. "jammy", "bookworm" "edition": str, # e.g. "ServerDatacenter", "Professional", "Enterprise" - "build": str, # OS build number + "build": str, # OS build number or kernel release "architecture": str, # "AMD64", "x86_64", "ARM64" "hostname": str, "ip_addresses": List[str] @@ -61,6 +213,10 @@ def get_os_details() -> Dict[str, Any]: is_server = False os_type = "unknown" os_name = platform.platform() + distro = "" + distro_version = "" + distro_id = "" + distro_codename = "" edition = "" build = platform.version() arch = platform.machine() or ("64-bit" if sys.maxsize > 2**32 else "32-bit") @@ -105,15 +261,14 @@ def get_os_details() -> Dict[str, Any]: elif sys.platform.startswith("linux"): os_type = "linux" is_server = True # Linux in enterprise log forwarding contexts is typically server/node - if os.path.exists("/etc/os-release"): - try: - with open("/etc/os-release", "r", encoding="utf-8") as f: - for line in f: - if line.startswith("PRETTY_NAME="): - os_name = line.strip().split("=", 1)[1].strip('"\'') - break - except Exception: - pass + lin_info = get_linux_distro_info(os_release_path=os_release_path) + distro = lin_info.get("distro", "Linux") + distro_version = lin_info.get("distro_version", "") + distro_id = lin_info.get("distro_id", "") + distro_codename = lin_info.get("distro_codename", "") + os_name = lin_info.get("pretty_name") or f"{distro} {distro_version}".strip() or "Linux" + edition = distro_codename or distro + build = platform.release() elif sys.platform == "darwin": os_type = "darwin" is_server = False @@ -126,6 +281,10 @@ def get_os_details() -> Dict[str, Any]: "os_type": os_type, "os_name": os_name, "is_server": is_server, + "distro": distro, + "distro_version": distro_version, + "distro_id": distro_id, + "distro_codename": distro_codename, "edition": edition, "build": build, "architecture": arch, @@ -310,6 +469,10 @@ def get_full_system_inventory() -> Dict[str, Any]: "os_type": os_info.get("os_type"), "os_name": os_info.get("os_name"), "is_server": os_info.get("is_server"), + "distro": os_info.get("distro", ""), + "distro_version": os_info.get("distro_version", ""), + "distro_id": os_info.get("distro_id", ""), + "distro_codename": os_info.get("distro_codename", ""), "edition": os_info.get("edition"), "build": os_info.get("build"), "architecture": os_info.get("architecture"), diff --git a/src/hub/web_ui.py b/src/hub/web_ui.py index 6e44c65..523ce97 100644 --- a/src/hub/web_ui.py +++ b/src/hub/web_ui.py @@ -469,6 +469,9 @@ class WebUIRenderer: # OS classification badge & display name ost_lower = ost.lower() + distro_val = str(info.get("distro") or "").strip() + distro_ver_val = str(info.get("distro_version") or "").strip() + if "server" in ost_lower or is_srv: if "windows" in ost_lower or "win" in ost_lower: filter_cat = "windows_server" @@ -478,9 +481,11 @@ class WebUIRenderer: os_display_name = html.escape(os_name or "Windows Server") else: filter_cat = "linux" - os_badge_html = f'🐧 LINUX' - os_badge_full = '🐧 LINUX SERVER' - os_display_name = html.escape(os_name or "Linux Node") + lbl = distro_val.upper() if distro_val else "LINUX" + disp_name = os_name if os_name else (f"{distro_val} {distro_ver_val}".strip() or "Linux Server") + os_badge_html = f'🐧 {html.escape(lbl)}' + os_badge_full = f'🐧 {html.escape(lbl)} SERVER' + os_display_name = html.escape(disp_name) elif "windows" in ost_lower or "win" in ost_lower: filter_cat = "windows_client" os_badge_html = f'🖥️ WINDOWS' @@ -488,19 +493,36 @@ class WebUIRenderer: os_display_name = html.escape(os_name or "Windows Workstation") elif "linux" in ost_lower: filter_cat = "linux" - os_badge_html = f'🐧 LINUX' - os_badge_full = '🐧 LINUX' - os_display_name = html.escape(os_name or "Linux Node") + lbl = distro_val.upper() if distro_val else "LINUX" + disp_name = os_name if os_name else (f"{distro_val} {distro_ver_val}".strip() or "Linux Node") + os_badge_html = f'🐧 {html.escape(lbl)}' + os_badge_full = f'🐧 {html.escape(lbl)}' + os_display_name = html.escape(disp_name) else: filter_cat = "other" os_badge_html = f'{esc_ost.upper()}' os_badge_full = f'{html.escape(ost.upper())}' os_display_name = html.escape(os_name or ost) - edition_str = html.escape(str(info.get("edition") or "")) - build_str = html.escape(str(info.get("build") or "")) - arch_str = html.escape(str(info.get("architecture") or "")) - specs_sub = f"{edition_str} • Build {build_str} • {arch_str}".strip(" •") if (edition_str or build_str or arch_str) else "" + edition_str = html.escape(str(info.get("edition") or "").strip()) + build_str = html.escape(str(info.get("build") or "").strip()) + arch_str = html.escape(str(info.get("architecture") or "").strip()) + distro_ver_str = html.escape(distro_ver_val) + codename_str = html.escape(str(info.get("distro_codename") or "").strip()) + + if "linux" in ost_lower: + sub_parts = [] + if distro_ver_str and distro_ver_str not in os_display_name: + sub_parts.append(f"v{distro_ver_str}") + if codename_str: + sub_parts.append(f"({codename_str})") + if build_str: + sub_parts.append(f"Kernel {build_str}") + if arch_str: + sub_parts.append(arch_str) + specs_sub = " • ".join(sub_parts) + else: + specs_sub = f"{edition_str} • Build {build_str} • {arch_str}".strip(" •") if (edition_str or build_str or arch_str) else "" # Roles badges role_badges = [] @@ -602,7 +624,7 @@ class WebUIRenderer: # 2. Server Roles Row (Tab 2) server_roles_rows.append(f""" - + {esc_st.upper()}
{esc_host}{host_tag}
diff --git a/tests/test_server_roles_and_best_practices.py b/tests/test_server_roles_and_best_practices.py index e321510..7c13b4c 100644 --- a/tests/test_server_roles_and_best_practices.py +++ b/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()