dev -> main

.github/labeler.yml

@@ -0,0 +1,47 @@
+# Path-based PR labels for actions/labeler v5+.
+# See: https://github.com/actions/labeler#pull-request-labeler
+
+documentation:
+  - changed-files:
+      - any-glob-to-any-file:
+          - docs/**
+          - README.md
+          - .github/**/*.md
+
+ci:
+  - changed-files:
+      - any-glob-to-any-file:
+          - .github/**
+
+tests:
+  - changed-files:
+      - any-glob-to-any-file:
+          - test/**
+
+plugins-inband:
+  - changed-files:
+      - any-glob-to-any-file:
+          - nodescraper/plugins/inband/**
+
+plugins-ooband:
+  - changed-files:
+      - any-glob-to-any-file:
+          - nodescraper/plugins/ooband/**
+
+plugins-serviceability:
+  - changed-files:
+      - any-glob-to-any-file:
+          - nodescraper/plugins/serviceability/**
+
+plugins-generic:
+  - changed-files:
+      - any-glob-to-any-file:
+          - nodescraper/plugins/generic_collection/**
+
+framework:
+  - changed-files:
+      - any-glob-to-any-file:
+          - nodescraper/base/**
+          - nodescraper/interfaces/**
+          - nodescraper/cli/**
+          - nodescraper/models/**

.github/scripts/plugin_convention_warnings.py

@@ -14,6 +14,14 @@
    ``pydantic.Field(...)`` with a non-empty ``description=`` (for help/CLI
    text). ``ClassVar`` fields, ``_``-prefixed names, and ``model_config`` are
    skipped.
+
+3. **Shell quoting** — In ``Collector`` / ``Analyzer`` methods, values from
+   ``args.*``, ``cmd_spec.*``, or sensitive function parameters (names ending
+   in ``_path`` / ``_file``, or ``url``, ``boot``, ``folder``, ``host``) that
+   are interpolated into shell commands (``_run_sut_cmd``, ``_run_amd_smi``,
+   ``_run_dell_command``) must be wrapped in ``shell_quote(...)`` or assigned
+   from a prior ``shell_quote(...)`` call in the same function.
+   ``GenericCollectionCollector`` is excluded (user commands are intentional).
 """
 
 from __future__ import annotations
@@ -40,6 +48,9 @@
     }
 )
 
+_SHELL_RUN_METHODS = frozenset({"_run_sut_cmd", "_run_amd_smi", "_run_dell_command"})
+_SKIP_SHELL_QUOTE_CLASSES = frozenset({"GenericCollectionCollector"})
+
 
 def _is_stringish(expr: ast.expr) -> bool:
     if isinstance(expr, ast.Constant) and isinstance(expr.value, str):
@@ -224,6 +235,179 @@ def _check_args_fields(path: Path, tree: ast.Module) -> list[str]:
     return msgs
 
 
+def _is_shell_quote_call(expr: ast.expr) -> bool:
+    if not isinstance(expr, ast.Call):
+        return False
+    func = expr.func
+    if isinstance(func, ast.Name) and func.id == "shell_quote":
+        return True
+    return isinstance(func, ast.Attribute) and func.attr == "shell_quote"
+
+
+def _func_param_names(func: ast.FunctionDef | ast.AsyncFunctionDef) -> set[str]:
+    names: set[str] = set()
+    for arg in func.args.args + func.args.kwonlyargs:
+        if arg.arg != "self":
+            names.add(arg.arg)
+    if func.args.vararg:
+        names.add(func.args.vararg.arg)
+    return names
+
+
+def _collect_shell_quote_safe_names(func: ast.FunctionDef | ast.AsyncFunctionDef) -> set[str]:
+    safe: set[str] = set()
+    for node in ast.walk(func):
+        if not isinstance(node, ast.Assign):
+            continue
+        for target in node.targets:
+            if isinstance(target, ast.Name) and _is_shell_quote_call(node.value):
+                safe.add(target.id)
+    return safe
+
+
+def _expr_preview(expr: ast.expr) -> str:
+    try:
+        return ast.unparse(expr)
+    except Exception:
+        return "<expr>"
+
+
+def _is_sensitive_param_name(name: str) -> bool:
+    if name in {"url", "boot", "folder", "hostname", "host"}:
+        return True
+    return name.endswith("_path") or name.endswith("_file")
+
+
+def _is_user_controlled_leaf(expr: ast.expr, param_names: set[str]) -> bool:
+    if isinstance(expr, ast.Attribute) and isinstance(expr.value, ast.Name):
+        base = expr.value.id
+        if base in ("args", "cmd_spec"):
+            return True
+    if isinstance(expr, ast.Name) and expr.id in param_names:
+        return _is_sensitive_param_name(expr.id)
+    return False
+
+
+def _contains_user_controlled(expr: ast.expr, param_names: set[str]) -> bool:
+    if _is_user_controlled_leaf(expr, param_names):
+        return True
+    for child in ast.iter_child_nodes(expr):
+        if isinstance(child, ast.expr) and _contains_user_controlled(child, param_names):
+            return True
+    return False
+
+
+def _is_trivially_safe_expr(expr: ast.expr, safe_names: set[str]) -> bool:
+    if _is_shell_quote_call(expr):
+        return True
+    if isinstance(expr, ast.Constant):
+        return True
+    if isinstance(expr, ast.Name) and expr.id in safe_names:
+        return True
+    if isinstance(expr, ast.Attribute) and isinstance(expr.value, ast.Name):
+        base = expr.value.id
+        if base == "self" and (
+            expr.attr == "CMD" or expr.attr.startswith("CMD_") or expr.attr in ("AMD_SMI_EXE",)
+        ):
+            return True
+    return False
+
+
+def _command_insert_issues(
+    expr: ast.expr,
+    param_names: set[str],
+    safe_names: set[str],
+) -> list[tuple[int, str]]:
+    issues: list[tuple[int, str]] = []
+
+    def note(value: ast.expr) -> None:
+        if _is_trivially_safe_expr(value, safe_names):
+            return
+        if _contains_user_controlled(value, param_names):
+            issues.append((getattr(value, "lineno", 0), _expr_preview(value)))
+
+    if isinstance(expr, ast.JoinedStr):
+        for part in expr.values:
+            if isinstance(part, ast.FormattedValue):
+                note(part.value)
+    elif isinstance(expr, ast.Call) and isinstance(expr.func, ast.Attribute):
+        if expr.func.attr == "format":
+            for kw in expr.keywords:
+                if kw.arg is not None and kw.value is not None:
+                    note(kw.value)
+    elif isinstance(expr, ast.BinOp):
+        note(expr.left)
+        note(expr.right)
+
+    return issues
+
+
+def _resolve_command_exprs(
+    func: ast.FunctionDef | ast.AsyncFunctionDef, expr: ast.expr
+) -> list[ast.expr]:
+    if not isinstance(expr, ast.Name):
+        return [expr]
+
+    matches: list[ast.expr] = []
+    for node in ast.walk(func):
+        if isinstance(node, ast.Assign):
+            for target in node.targets:
+                if isinstance(target, ast.Name) and target.id == expr.id:
+                    matches.append(node.value)
+        elif (
+            isinstance(node, ast.AnnAssign)
+            and isinstance(node.target, ast.Name)
+            and node.target.id == expr.id
+            and node.value is not None
+        ):
+            matches.append(node.value)
+    return matches or [expr]
+
+
+def _check_function_shell_quoting(
+    path: Path,
+    class_name: str,
+    func: ast.FunctionDef | ast.AsyncFunctionDef,
+) -> list[str]:
+    msgs: list[str] = []
+    param_names = _func_param_names(func)
+    safe_names = _collect_shell_quote_safe_names(func)
+
+    for node in ast.walk(func):
+        if not isinstance(node, ast.Call):
+            continue
+        if not isinstance(node.func, ast.Attribute):
+            continue
+        if node.func.attr not in _SHELL_RUN_METHODS:
+            continue
+        if not node.args:
+            continue
+
+        command_expr = node.args[0]
+        for resolved in _resolve_command_exprs(func, command_expr):
+            for lineno, preview in _command_insert_issues(resolved, param_names, safe_names):
+                line = lineno or node.lineno
+                msgs.append(
+                    f"{path}:{line}: [{class_name}.{func.name}] user-controlled value "
+                    f"{preview} in shell command must use shell_quote(...)."
+                )
+    return msgs
+
+
+def _check_shell_quoting(path: Path, tree: ast.Module) -> list[str]:
+    """Rule #3: warn when config/param values reach shell commands without shell_quote."""
+    msgs: list[str] = []
+    for node in tree.body:
+        if not isinstance(node, ast.ClassDef) or not _is_collector_or_analyzer_class(node):
+            continue
+        if node.name in _SKIP_SHELL_QUOTE_CLASSES:
+            continue
+        for stmt in node.body:
+            if isinstance(stmt, (ast.FunctionDef, ast.AsyncFunctionDef)):
+                msgs.extend(_check_function_shell_quoting(path, node.name, stmt))
+    return msgs
+
+
 def main() -> None:
     if not PLUGIN_ROOT.is_dir():
         sys.stderr.write(f"warning: plugins directory not found: {PLUGIN_ROOT}\n")
@@ -242,8 +426,10 @@ def main() -> None:
 
         if "collector" in name and name.endswith(".py"):
             all_msgs.extend(_check_cmd_prefixes(rel, tree))
+            all_msgs.extend(_check_shell_quoting(rel, tree))
         if "analyzer" in name and name.endswith(".py"):
             all_msgs.extend(_check_cmd_prefixes(rel, tree))
+            all_msgs.extend(_check_shell_quoting(rel, tree))
 
         if name == "collector_args.py" or name == "analyzer_args.py":
             all_msgs.extend(_check_args_fields(rel, tree))

.github/workflows/label.yml

@@ -0,0 +1,23 @@
+# This workflow will triage pull requests and apply a label based on the
+# paths that are modified in the pull request.
+#
+# To use this workflow, you will need to set up a .github/labeler.yml
+# file with configuration.  For more information, see:
+# https://github.com/actions/labeler
+
+name: Labeler
+on: [pull_request_target]
+
+jobs:
+  label:
+
+    runs-on: ubuntu-latest
+    permissions:
+      contents: read
+      pull-requests: write
+
+    steps:
+    - uses: actions/labeler@v5
+      with:
+        repo-token: "${{ secrets.GITHUB_TOKEN }}"
+        sync-labels: true