KernelModulePlugin update

nodescraper/plugins/inband/kernel_module/kernel_module_collector.py

@@ -26,11 +26,12 @@
 from typing import Optional
 
 from nodescraper.base import InBandDataCollector
+from nodescraper.connection.inband import TextFileArtifact
 from nodescraper.connection.inband.inband import CommandArtifact
 from nodescraper.enums import EventCategory, EventPriority, ExecutionStatus, OSFamily
 from nodescraper.models import TaskResult
 
-from .kernel_module_data import KernelModuleDataModel
+from .kernel_module_data import KernelModuleDataModel, ModuleInfo, ModuleParameter
 
 
 class KernelModuleCollector(InBandDataCollector[KernelModuleDataModel, None]):
@@ -39,6 +40,7 @@ class KernelModuleCollector(InBandDataCollector[KernelModuleDataModel, None]):
     DATA_MODEL = KernelModuleDataModel
     CMD_WINDOWS = "wmic os get Version /Value"
     CMD = "cat /proc/modules"
+    CMD_MODINFO_AMDGPU = "modinfo amdgpu"
 
     def parse_proc_modules(self, output: dict) -> dict:
         """Parse command output and return dict of modules
@@ -60,6 +62,77 @@ def parse_proc_modules(self, output: dict) -> dict:
             }
         return modules
 
+    def _parse_modinfo(self, output: str) -> Optional[ModuleInfo]:
+        """Parse modinfo command output into structured ModuleInfo
+
+        Args:
+            output (str): modinfo command output
+
+        Returns:
+            Optional[ModuleInfo]: parsed module information or None if parsing fails
+        """
+        if not output or not output.strip():
+            return None
+
+        module_info = ModuleInfo()
+
+        for line in output.splitlines():
+            line = line.strip()
+            if not line or ":" not in line:
+                continue
+
+            field, _, value = line.partition(":")
+            field = field.strip()
+            value = value.strip()
+
+            if field == "filename":
+                module_info.filename = value
+            elif field == "version":
+                module_info.version = value
+            elif field == "license":
+                module_info.license = value
+            elif field == "description":
+                module_info.description = value
+            elif field == "author":
+                module_info.author.append(value)
+            elif field == "firmware":
+                module_info.firmware.append(value)
+            elif field == "srcversion":
+                module_info.srcversion = value
+            elif field == "depends":
+                if value:
+                    module_info.depends = [dep.strip() for dep in value.split(",") if dep.strip()]
+            elif field == "name":
+                module_info.name = value
+            elif field == "vermagic":
+                module_info.vermagic = value
+            elif field == "sig_id":
+                module_info.sig_id = value
+            elif field == "signer":
+                module_info.signer = value
+            elif field == "sig_key":
+                module_info.sig_key = value
+            elif field == "sig_hashalgo":
+                module_info.sig_hashalgo = value
+            elif field == "parm":
+                param_name, param_desc = value.split(":", 1) if ":" in value else (value, "")
+                param_name = param_name.strip()
+                param_desc = param_desc.strip()
+
+                param_type = None
+                if param_desc and "(" in param_desc and ")" in param_desc:
+                    type_start = param_desc.rfind("(")
+                    type_end = param_desc.rfind(")")
+                    if type_start < type_end:
+                        param_type = param_desc[type_start + 1 : type_end].strip()
+                        param_desc = param_desc[:type_start].strip()
+
+                module_info.parm.append(
+                    ModuleParameter(name=param_name, type=param_type, description=param_desc)
+                )
+
+        return module_info
+
     def get_module_parameters(self, module_name: str) -> dict:
         """Fetches parameter names and values for a given kernel module using _run_sut_cmd
 
@@ -143,8 +216,39 @@ def collect_data(self, args=None) -> tuple[TaskResult, Optional[KernelModuleData
         else:
             kernel_modules = self.collect_all_module_info()
 
+        amdgpu_modinfo = None
+        if self.system_info.os_family != OSFamily.WINDOWS:
+            # Collect and parse modinfo amdgpu output
+            modinfo_res = self._run_sut_cmd(self.CMD_MODINFO_AMDGPU)
+            if modinfo_res.exit_code == 0 and modinfo_res.stdout:
+                amdgpu_modinfo = self._parse_modinfo(modinfo_res.stdout)
+                if amdgpu_modinfo:
+                    self.result.artifacts.append(
+                        TextFileArtifact(filename="modinfo_amdgpu.txt", contents=modinfo_res.stdout)
+                    )
+                else:
+                    self._log_event(
+                        category=EventCategory.OS,
+                        description="Could not parse modinfo amdgpu output",
+                        data={"command": modinfo_res.command},
+                        priority=EventPriority.WARNING,
+                    )
+            else:
+                self._log_event(
+                    category=EventCategory.OS,
+                    description="Could not collect modinfo amdgpu output",
+                    data={
+                        "command": modinfo_res.command,
+                        "exit_code": modinfo_res.exit_code,
+                        "stderr": modinfo_res.stderr,
+                    },
+                    priority=EventPriority.WARNING,
+                )
+
         if kernel_modules:
-            km_data = KernelModuleDataModel(kernel_modules=kernel_modules)
+            km_data = KernelModuleDataModel(
+                kernel_modules=kernel_modules, amdgpu_modinfo=amdgpu_modinfo
+            )
             self._log_event(
                 category="KERNEL_READ",
                 description="Kernel modules read",

nodescraper/plugins/inband/kernel_module/kernel_module_data.py

@@ -24,8 +24,37 @@
 #
 ###############################################################################
 
+from typing import Optional
+
+from pydantic import BaseModel, Field
+
 from nodescraper.models import DataModel
 
 
+class ModuleParameter(BaseModel):
+    name: str
+    type: Optional[str] = None
+    description: Optional[str] = None
+
+
+class ModuleInfo(BaseModel):
+    filename: Optional[str] = None
+    version: Optional[str] = None
+    license: Optional[str] = None
+    description: Optional[str] = None
+    author: list[str] = Field(default_factory=list)
+    firmware: list[str] = Field(default_factory=list)
+    srcversion: Optional[str] = None
+    depends: list[str] = Field(default_factory=list)
+    name: Optional[str] = None
+    vermagic: Optional[str] = None
+    sig_id: Optional[str] = None
+    signer: Optional[str] = None
+    sig_key: Optional[str] = None
+    sig_hashalgo: Optional[str] = None
+    parm: list[ModuleParameter] = Field(default_factory=list)
+
+
 class KernelModuleDataModel(DataModel):
     kernel_modules: dict
+    amdgpu_modinfo: Optional[ModuleInfo] = None

test/unit/plugin/test_amdsmi_collector.py

@@ -1,3 +1,28 @@
+###############################################################################
+#
+# MIT License
+#
+# Copyright (c) 2025 Advanced Micro Devices, Inc.
+#
+# Permission is hereby granted, free of charge, to any person obtaining a copy
+# of this software and associated documentation files (the "Software"), to deal
+# in the Software without restriction, including without limitation the rights
+# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+# copies of the Software, and to permit persons to whom the Software is
+# furnished to do so, subject to the following conditions:
+#
+# The above copyright notice and this permission notice shall be included in all
+# copies or substantial portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+# SOFTWARE.
+#
+###############################################################################
 import json
 from typing import Any
 from unittest.mock import MagicMock

test/unit/plugin/test_kernel_module_analyzer.py

@@ -1,3 +1,28 @@
+###############################################################################
+#
+# MIT License
+#
+# Copyright (c) 2025 Advanced Micro Devices, Inc.
+#
+# Permission is hereby granted, free of charge, to any person obtaining a copy
+# of this software and associated documentation files (the "Software"), to deal
+# in the Software without restriction, including without limitation the rights
+# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+# copies of the Software, and to permit persons to whom the Software is
+# furnished to do so, subject to the following conditions:
+#
+# The above copyright notice and this permission notice shall be included in all
+# copies or substantial portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+# SOFTWARE.
+#
+###############################################################################
 import pytest
 
 from nodescraper.enums.eventcategory import EventCategory