From 3a9ca7af14865f8a80ba215927ae468012f6af1c Mon Sep 17 00:00:00 2001 From: Manny <107129718+mlgtcode@users.noreply.github.com> Date: Sun, 26 Jul 2026 19:53:22 +0200 Subject: [PATCH] Hotfix for win and general fixes --- agent360/plugins/system.py | 140 +++++++++++++++++++++++++++++-------- 1 file changed, 112 insertions(+), 28 deletions(-) diff --git a/agent360/plugins/system.py b/agent360/plugins/system.py index eb0e947..5118130 100644 --- a/agent360/plugins/system.py +++ b/agent360/plugins/system.py @@ -5,7 +5,9 @@ except ImportError: netifaces = None import os +import shlex import platform +import socket from subprocess import Popen, PIPE import sys import time @@ -20,11 +22,18 @@ def systemCommand(Command, newlines=True): Output = "" Error = "" try: - proc = Popen(Command.split(), stdout=PIPE) - Output = proc.communicate()[0] + if isinstance(Command, str): + Command = shlex.split(Command) + proc = Popen(Command, stdout=PIPE, stderr=PIPE) + Output, Error = proc.communicate() except Exception: pass + if isinstance(Output, bytes): + Output = Output.decode("utf-8", "replace") + if isinstance(Error, bytes): + Error = Error.decode("utf-8", "replace") + if Output: if newlines is True: Stdout = Output.split("\n") @@ -40,6 +49,11 @@ def systemCommand(Command, newlines=True): return (Stdout, Stderr) +def parse_cpuinfo_line(line): + key, separator, value = line.partition(':') + return key.strip(), value.strip() if separator else "" + + def linux_hardware_memory(): block_size = 0 try: @@ -67,6 +81,31 @@ def ip_addresses(): ip_list['v4'] = {} ip_list['v6'] = {} if netifaces is None: + try: + addrs = psutil.net_if_addrs() + except Exception: + return ip_list + + for interface, addresses in addrs.items(): + for address in addresses: + if address.family == socket.AF_INET: + if interface not in ip_list['v4']: + ip_list['v4'][interface] = [] + ip_list['v4'][interface].append([{ + 'addr': address.address, + 'netmask': address.netmask, + 'broadcast': address.broadcast, + 'ptp': getattr(address, 'ptp', None), + }]) + elif address.family == socket.AF_INET6: + if interface not in ip_list['v6']: + ip_list['v6'][interface] = [] + ip_list['v6'][interface].append([{ + 'addr': address.address, + 'netmask': address.netmask, + 'broadcast': address.broadcast, + 'ptp': getattr(address, 'ptp', None), + }]) return ip_list for interface in netifaces.interfaces(): link = netifaces.ifaddresses(interface) @@ -81,6 +120,41 @@ def ip_addresses(): return ip_list +def windows_cpu_brand(): + brand = platform.processor() + if brand: + return brand + + brand = os.environ.get('PROCESSOR_IDENTIFIER', '') + if brand: + return brand + + try: + output = systemCommand('wmic cpu get name', False)[0] + if output: + for line in output.split('\n'): + line = line.strip() + if line and line.lower() != 'name': + return line + except Exception: + pass + + return "Unknown CPU" + + +def linux_os_name(): + if distro is not None: + try: + return str(' '.join(distro.linux_distribution(full_distribution_name=True))) + except Exception: + pass + + try: + return str(' '.join(platform.linux_distribution())) + except Exception: + return platform.platform() + + class Plugin(plugins.BasePlugin): __name__ = 'system' @@ -89,56 +163,66 @@ def run(self, *unused): cpu = {} cpu['brand'] = "Unknown CPU" cpu['count'] = 0 - if(os.path.isfile("/proc/cpuinfo")): - f = open('/proc/cpuinfo') - if f: - for line in f: + if os.path.isfile("/proc/cpuinfo"): + with open('/proc/cpuinfo', 'r') as cpuinfo_file: + for line in cpuinfo_file: # Ignore the blank line separating the information between # details about two processing units if line.strip(): - if "model name" == line.rstrip('\n').split(':')[0].strip(): - cpu['brand'] = line.rstrip('\n').split(':')[1].strip() - if "Processor" == line.rstrip('\n').split(':')[0].strip(): - cpu['brand'] = line.rstrip('\n').split(':')[1].strip() - if "processor" == line.rstrip('\n').split(':')[0].strip(): - cpu['count'] = line.rstrip('\n').split(':')[1].strip() + key, value = parse_cpuinfo_line(line) + if key == "model name": + cpu['brand'] = value + if key == "Processor": + cpu['brand'] = value + if key == "processor": + cpu['count'] = value if cpu['brand'] == "Unknown CPU": - f = os.popen('lscpu').read().split('\n') - if f: - for line in f: + cpuinfo_output = systemCommand('lscpu', False)[0] + if cpuinfo_output: + for line in cpuinfo_output.split('\n'): # Ignore the blank line separating the information between # details about two processing units if line.strip(): - if "Model name" == line.rstrip('\n').split(':')[0].strip(): - cpu['brand'] = line.rstrip('\n').split(':')[1].strip() - if "Processor" == line.rstrip('\n').split(':')[0].strip(): - cpu['brand'] = line.rstrip('\n').split(':')[1].strip() - if "CPU(s)" == line.rstrip('\n').split(':')[0].strip(): - cpu['count'] = line.rstrip('\n').split(':')[1].strip() + key, value = parse_cpuinfo_line(line) + if key == "Model name": + cpu['brand'] = value + if key == "Processor": + cpu['brand'] = value + if key == "CPU(s)": + cpu['count'] = value mem = psutil.virtual_memory().total if sys.platform == "linux" or sys.platform == "linux2": hw_mem = linux_hardware_memory() if hw_mem != 0: mem = hw_mem - if distro is None: - systeminfo['os'] = str(' '.join(platform.linux_distribution())) - else: - systeminfo['os'] = str(' '.join(distro.linux_distribution(full_distribution_name=True))) + systeminfo['os'] = linux_os_name() elif sys.platform == "darwin": systeminfo['os'] = "Mac OS %s" % platform.mac_ver()[0] - cpu['brand'] = str(systemCommand('sysctl machdep.cpu.brand_string', False)[0]).split(': ')[1] + sysctl_output = systemCommand('sysctl machdep.cpu.brand_string', False)[0] + if sysctl_output: + brand_line = sysctl_output.split(': ', 1) + if len(brand_line) > 1: + cpu['brand'] = brand_line[1] #cpu['count'] = systemCommand('sysctl hw.ncpu') elif sys.platform == "freebsd10" or sys.platform == "freebsd11": systeminfo['os'] = "FreeBSD %s" % platform.release() - cpu['brand'] = str(systemCommand('sysctl hw.model', False)[0]).split(': ')[1] - cpu['count'] = systemCommand('sysctl hw.ncpu') + sysctl_output = systemCommand('sysctl hw.model', False)[0] + if sysctl_output: + brand_line = sysctl_output.split(': ', 1) + if len(brand_line) > 1: + cpu['brand'] = brand_line[1] + cpu_count_output = systemCommand('sysctl hw.ncpu', False)[0] + if cpu_count_output: + cpu['count'] = cpu_count_output.splitlines()[-1].strip() elif sys.platform == "win32": # https://learn.microsoft.com/en-us/windows/release-health/windows11-release-information if sys.getwindowsversion().build >= 22000: systeminfo['os'] = "{} {}".format(platform.uname()[0], 11) else: systeminfo['os'] = "{} {}".format(platform.uname()[0], platform.uname()[2]) + cpu['brand'] = windows_cpu_brand() + cpu['count'] = psutil.cpu_count() systeminfo['cpu'] = cpu['brand'] systeminfo['cores'] = cpu['count'] systeminfo['memory'] = mem