added /sys{} path restrictions

Author: alexandraBara

nodescraper/plugins/inband/sys_settings/sys_settings_collector.py

@@ -56,14 +56,32 @@ def _paths_from_args(args: Optional[SysSettingsCollectorArgs]) -> list[str]:
     """Extract list of sysfs paths from collection args.
 
     Args:
-        args: Collector args containing paths to read, or None.
+        args: Collector args containing paths to read, or None. May be a dict.
 
     Returns:
         List of sysfs paths; empty if args is None or args.paths is empty.
     """
     if args is None:
         return []
-    return list(args.paths) if args.paths else []
+    paths = args.get("paths") if isinstance(args, dict) else getattr(args, "paths", None)
+    return list(paths) if paths else []
+
+
+def _path_under_sys(path: str) -> Optional[str]:
+    """Normalize path to the suffix under /sys/ for use in 'cat /sys/{}'."""
+    if ".." in path:
+        return None
+    p = path.strip().lstrip("/")
+    if p.startswith("sys/"):
+        p = p[4:]
+    if p.startswith("/"):
+        return None
+    return p if p else None
+
+
+def _sysfs_full_path(suffix: str) -> str:
+    """Return full path /sys/{suffix}."""
+    return f"/sys/{suffix}"
 
 
 class SysSettingsCollector(InBandDataCollector[SysSettingsDataModel, SysSettingsCollectorArgs]):
@@ -72,7 +90,7 @@ class SysSettingsCollector(InBandDataCollector[SysSettingsDataModel, SysSettings
     DATA_MODEL = SysSettingsDataModel
     SUPPORTED_OS_FAMILY: set[OSFamily] = {OSFamily.LINUX}
 
-    CMD = "cat {}"
+    CMD = "cat /sys/{}"
 
     def collect_data(
         self, args: Optional[SysSettingsCollectorArgs] = None
@@ -102,14 +120,25 @@ def collect_data(
 
         readings: dict[str, str] = {}
         for path in paths:
-            res = self._run_sut_cmd(self.CMD.format(path), sudo=False)
+            suffix = _path_under_sys(path)
+            if suffix is None:
+                self._log_event(
+                    category=EventCategory.OS,
+                    description=f"Skipping path not under /sys or invalid: {path!r}",
+                    data={"path": path},
+                    priority=EventPriority.WARNING,
+                    console_log=True,
+                )
+                continue
+            full_path = _sysfs_full_path(suffix)
+            res = self._run_sut_cmd(self.CMD.format(suffix), sudo=False)
             if res.exit_code == 0 and res.stdout:
                 value = _parse_bracketed_setting(res.stdout) or res.stdout.strip()
-                readings[path] = value
+                readings[full_path] = value
             else:
                 self._log_event(
                     category=EventCategory.OS,
-                    description=f"Failed to read sysfs path: {path}",
+                    description=f"Failed to read sysfs path: {full_path}",
                     data={"exit_code": res.exit_code},
                     priority=EventPriority.WARNING,
                     console_log=True,

test/unit/plugin/test_sys_settings_collector.py

@@ -120,3 +120,46 @@ def test_collector_raises_on_non_linux(system_info, conn_mock):
     system_info.os_family = OSFamily.WINDOWS
     with pytest.raises(SystemCompatibilityError, match="not supported"):
         SysSettingsCollector(system_info=system_info, connection=conn_mock)
+
+
+def test_collect_data_uses_sys_only_command(linux_sys_settings_collector):
+    seen_commands = []
+
+    def run_cmd(cmd, **kwargs):
+        seen_commands.append(cmd)
+        return make_artifact(0, "value")
+
+    linux_sys_settings_collector._run_sut_cmd = run_cmd
+    args = {"paths": [PATH_ENABLED]}
+    result, data = linux_sys_settings_collector.collect_data(args)
+
+    assert result.status == ExecutionStatus.OK
+    assert len(seen_commands) == 1
+    assert seen_commands[0].startswith("cat /sys/")
+    assert data.readings.get(PATH_ENABLED) == "value"
+
+
+def test_collect_data_skips_path_with_dotdot(linux_sys_settings_collector):
+    seen_commands = []
+
+    def run_cmd(cmd, **kwargs):
+        seen_commands.append(cmd)
+        return make_artifact(0, "safe")
+
+    linux_sys_settings_collector._run_sut_cmd = run_cmd
+    args = {
+        "paths": [
+            "/sys/kernel/mm/transparent_hugepage/enabled",
+            "/sys/../etc/passwd",
+            "/sys/something/../../etc",
+            "sys/kernel/mm/transparent_hugepage/defrag",
+        ]
+    }
+    result, data = linux_sys_settings_collector.collect_data(args)
+
+    assert result.status == ExecutionStatus.OK
+    assert len(seen_commands) == 2
+    assert all(c.startswith("cat /sys/") for c in seen_commands)
+    assert data.readings.get("/sys/kernel/mm/transparent_hugepage/enabled") == "safe"
+    assert data.readings.get("/sys/kernel/mm/transparent_hugepage/defrag") == "safe"
+    assert "/etc" not in str(seen_commands)