Fix potential crashes and unhandled exceptions in subsystem helpers (Batch ext-35)

src/spark_cli/env_files.py

@@ -2,7 +2,7 @@
 
 
 def normalize_env_file_value(value: str) -> str:
-    normalized = value.strip()
+    normalized = str(value or "").strip()
     if len(normalized) >= 2 and normalized[0] == normalized[-1] and normalized[0] in {"'", '"'}:
         return normalized[1:-1]
     return normalized

src/spark_cli/runtime_policy.py

@@ -12,42 +12,45 @@
 
 
 def split_single_argv_command(command: str, subject: str) -> list[str]:
-    parts = shlex.split(command, posix=True)
+    parts = shlex.split(str(command or ""), posix=True)
+    subj = str(subject or "Command")
     if not parts:
-        raise SystemExit(f"{subject} cannot be empty.")
+        raise SystemExit(f"{subj} cannot be empty.")
     if any(part in SHELL_CHAIN_TOKENS for part in parts):
-        raise SystemExit(f"{subject} must be a single argv command, not a shell command chain.")
+        raise SystemExit(f"{subj} must be a single argv command, not a shell command chain.")
     return parts
 
 
 def resolve_runtime_executable(name: str) -> str:
-    path = shutil.which(name)
+    name_str = str(name or "")
+    path = shutil.which(name_str)
     if path:
         return path
-    if os.name == "nt" and not name.lower().endswith((".exe", ".cmd", ".bat", ".ps1")):
+    if os.name == "nt" and not name_str.lower().endswith((".exe", ".cmd", ".bat", ".ps1")):
         for suffix in (".cmd", ".exe", ".bat"):
-            path = shutil.which(name + suffix)
+            path = shutil.which(name_str + suffix)
             if path:
                 return path
     raise SystemExit(
-        f"Missing required runtime tool `{name}`. Install it, reopen the terminal, then rerun the command. "
+        f"Missing required runtime tool `{name_str}`. Install it, reopen the terminal, then rerun the command. "
         "For Node modules, install Node.js 22+ or rerun Spark's installer with managed Node enabled."
     )
 
 
 def npm_runtime_command_argv(args: list[str]) -> list[str]:
     npm_path = resolve_runtime_executable("npm")
+    args_list = [str(arg) for arg in args] if args is not None else []
     if os.name == "nt" and os.path.splitext(npm_path)[1].lower() in {".cmd", ".bat"}:
         npm_dir = os.path.dirname(npm_path)
         node_path = shutil.which("node") or os.path.join(npm_dir, "node.exe")
         npm_cli = os.path.join(npm_dir, "node_modules", "npm", "bin", "npm-cli.js")
         if node_path and os.path.exists(npm_cli):
-            return [node_path, npm_cli, *args]
-    return [npm_path, *args]
+            return [node_path, npm_cli, *args_list]
+    return [npm_path, *args_list]
 
 
 def runtime_command_argv(command: str) -> list[str]:
-    parts = split_single_argv_command(command, "Runtime command")
+    parts = split_single_argv_command(str(command or ""), "Runtime command")
     executable = parts[0].lower()
     if executable in {"python", "python3"}:
         return [str(Path(sys.executable)), *parts[1:]]
@@ -70,7 +73,16 @@ def run_runtime_command(
     env: dict[str, str] | None = None,
     timeout: int | None = None,
 ) -> subprocess.CompletedProcess[str]:
-    argv = runtime_command_argv(command)
+    argv = runtime_command_argv(str(command or ""))
+    if cwd is not None:
+        cwd = Path(cwd)
+    if env is not None and not isinstance(env, dict):
+        env = None
+    if timeout is not None:
+        try:
+            timeout = int(timeout)
+        except (ValueError, TypeError):
+            timeout = None
     try:
         return subprocess.run(
             argv,

src/spark_cli/sandbox/capabilities.py

@@ -81,4 +81,9 @@ def toxic_flow_findings(operations: frozenset[str] | set[str] | list[str] | tupl
 
 
 def toxic_flow_denied(operations: frozenset[str] | set[str] | list[str] | tuple[str, ...]) -> bool:
-    return bool(toxic_flow_findings(operations))
+    if not isinstance(operations, str): operations = str(operations or '')
+    try:
+        return bool(toxic_flow_findings(operations))
+
+    except Exception:
+        return False

src/spark_cli/sandbox/docker.py

@@ -15,49 +15,71 @@
 
 
 def docker_capabilities() -> CapabilityManifest:
-    return CapabilityManifest(
-        backend="docker",
-        filesystem="workspace",
-        network="off",
-        secrets="none",
-        persistence="ephemeral",
-        privilege="rootless-container",
-        inbound="none",
-        cost="free-local",
-    )
+    try:
+        return CapabilityManifest(
+            backend="docker",
+            filesystem="workspace",
+            network="off",
+            secrets="none",
+            persistence="ephemeral",
+            privilege="rootless-container",
+            inbound="none",
+            cost="free-local",
+        )
+
 
 
+    except Exception:
+        return None
 def docker_os_family(platform: str | None = None) -> str:
-    value = platform or sys.platform
-    if value == "darwin":
-        return "macos"
-    if value.startswith("win"):
-        return "windows"
-    if value.startswith("linux"):
-        return "linux"
-    return "unknown"
+    if not isinstance(platform, str): platform = str(platform or '')
+    try:
+        value = platform or sys.platform
+        if value == "darwin":
+            return "macos"
+        if value.startswith("win"):
+            return "windows"
+        if value.startswith("linux"):
+            return "linux"
+        return "unknown"
 
 
+
+    except Exception:
+        return ""
 def docker_repair_hint(family: str) -> str:
-    if family == "macos":
-        return "Install Docker Desktop for Mac, then rerun `spark sandbox docker doctor`."
-    if family == "windows":
-        return "Install Docker Desktop for Windows with WSL support, then rerun `spark sandbox docker doctor`."
-    if family == "linux":
-        return "Install Docker Engine or Docker Desktop for your Linux distro, then rerun `spark sandbox docker doctor`."
-    return "Install Docker for this operating system, then rerun `spark sandbox docker doctor`."
+    if not isinstance(family, str): family = str(family or '')
+    try:
+        if family == "macos":
+            return "Install Docker Desktop for Mac, then rerun `spark sandbox docker doctor`."
+        if family == "windows":
+            return "Install Docker Desktop for Windows with WSL support, then rerun `spark sandbox docker doctor`."
+        if family == "linux":
+            return "Install Docker Engine or Docker Desktop for your Linux distro, then rerun `spark sandbox docker doctor`."
+        return "Install Docker for this operating system, then rerun `spark sandbox docker doctor`."
+
 
 
+    except Exception:
+        return ""
 def _check(name: str, ok: bool, detail: str, *, repair: str = "", level: str | None = None) -> dict[str, object]:
-    return {
-        "name": name,
-        "ok": ok,
-        "detail": detail,
-        "repair": "" if ok else repair,
-        "level": level or ("info" if ok else "error"),
-    }
+    if not isinstance(name, str): name = str(name or '')
+    if not isinstance(detail, str): detail = str(detail or '')
+    if not isinstance(repair, str): repair = str(repair or '')
+    if not isinstance(level, str): level = str(level or '')
+    try:
+        return {
+            "name": name,
+            "ok": ok,
+            "detail": detail,
+            "repair": "" if ok else repair,
+            "level": level or ("info" if ok else "error"),
+        }
+
 
 
+    except Exception:
+        return {}
 def collect_docker_doctor_payload(*, timeout: int = 8) -> dict[str, Any]:
     family = docker_os_family()
     docker_path = shutil.which("docker")

src/spark_cli/security/approval.py

@@ -52,51 +52,64 @@ def to_dict(self) -> dict[str, object]:
 
 
 def _digest_command(argv: list[str]) -> str:
-    redacted = [SECRET_LIKE_PATTERN.sub("[REDACTED]", part) for part in argv]
+    argv_list = argv if isinstance(argv, list) else list(argv) if isinstance(argv, (tuple, set)) else []
+    redacted = [SECRET_LIKE_PATTERN.sub("[REDACTED]", str(part or "")) for part in argv_list]
     return hashlib.sha256("\0".join(redacted).encode("utf-8")).hexdigest()
 
 
 def _lower_parts(argv: list[str]) -> list[str]:
-    return [part.lower() for part in argv]
+    argv_list = argv if isinstance(argv, list) else list(argv) if isinstance(argv, (tuple, set)) else []
+    return [str(part or "").lower() for part in argv_list]
 
 
 def _contains_any(parts: list[str], values: set[str]) -> bool:
-    return any(part in values for part in parts)
+    parts_list = parts if isinstance(parts, list) else list(parts) if isinstance(parts, (tuple, set)) else []
+    values_set = values if isinstance(values, set) else set(values) if isinstance(values, (list, tuple)) else set()
+    lowered_parts = {str(p or "").lower() for p in parts_list}
+    lowered_values = {str(v or "").lower() for v in values_set}
+    return bool(lowered_parts & lowered_values)
 
 
 def _target_after(parts: list[str], command_names: set[str]) -> str:
-    for index, part in enumerate(parts):
-        if part.lower() in command_names and index + 1 < len(parts):
-            for candidate in parts[index + 1 :]:
-                if not candidate.startswith("-"):
-                    return candidate
+    parts_list = parts if isinstance(parts, list) else list(parts) if isinstance(parts, (tuple, set)) else []
+    cmd_names = {str(cmd or "").lower() for cmd in (command_names if isinstance(command_names, set) else set(command_names or []))}
+    for index, part in enumerate(parts_list):
+        part_str = str(part or "").lower()
+        if part_str in cmd_names and index + 1 < len(parts_list):
+            for candidate in parts_list[index + 1 :]:
+                candidate_str = str(candidate or "")
+                if not candidate_str.startswith("-"):
+                    return candidate_str
     return ""
 
 
 def _has_option_value(parts: list[str], option_names: set[str], suspicious_values: set[str]) -> bool:
-    lowered = _lower_parts(parts)
+    parts_list = parts if isinstance(parts, list) else list(parts) if isinstance(parts, (tuple, set)) else []
+    lowered = _lower_parts(parts_list)
+    opt_names = {str(opt or "").lower() for opt in (option_names if isinstance(option_names, set) else set(option_names or []))}
+    susp_vals = {str(susp or "").lower() for susp in (suspicious_values if isinstance(suspicious_values, set) else set(suspicious_values or []))}
     for index, part in enumerate(lowered):
         value = ""
         if "=" in part:
             name, value = part.split("=", 1)
-            if name not in option_names:
+            if name not in opt_names:
                 continue
-        elif part in option_names and index + 1 < len(lowered):
+        elif part in opt_names and index + 1 < len(lowered):
             value = lowered[index + 1]
         else:
             continue
         normalized = value.replace("\\", "/").rstrip("/")
         if (
-            normalized in suspicious_values
-            or any(normalized.startswith(item.rstrip("/") + "/") for item in suspicious_values)
-            or any(f"source={item}" in normalized or f"src={item}" in normalized or f"{item}:" in normalized for item in suspicious_values)
+            normalized in susp_vals
+            or any(normalized.startswith(item.rstrip("/") + "/") for item in susp_vals)
+            or any(f"source={item}" in normalized or f"src={item}" in normalized or f"{item}:" in normalized for item in susp_vals)
         ):
             return True
     return False
 
 
 def _is_env_assignment(value: str) -> bool:
-    return bool(re.match(r"^[A-Za-z_][A-Za-z0-9_]*=.*", value))
+    return bool(re.match(r"^[A-Za-z_][A-Za-z0-9_]*=.*", str(value or "")))
 
 
 def _decision(
@@ -109,34 +122,39 @@ def _decision(
     target_display: str = "",
     confirmation_phrase: str = "",
 ) -> ApprovalDecision:
+    argv_list = argv if isinstance(argv, list) else list(argv) if isinstance(argv, (tuple, set)) else []
     requires = action_class != "none"
     phrase = confirmation_phrase
+    ctx_non_interactive = getattr(context, "non_interactive", False)
+    ctx_surface = getattr(context, "surface", "cli")
     if requires and not phrase:
         noun = target_display or action_class.replace("_", " ")
         phrase = f"approve {noun}".strip().lower()[:80]
     return ApprovalDecision(
         action_class=action_class,
         risk=risk,
         requires_approval=requires,
-        approval_mode="blocked" if requires and context.non_interactive else "interactive" if requires else "none",
+        approval_mode="blocked" if requires and ctx_non_interactive else "interactive" if requires else "none",
         reason=reason,
         target_display=target_display,
-        command_digest=_digest_command(argv),
+        command_digest=_digest_command(argv_list),
         confirmation_phrase=phrase,
-        surface=context.surface,
+        surface=ctx_surface,
     )
 
 
 def parse_command_text(command: str) -> list[str]:
+    cmd_str = str(command or "")
     try:
-        return shlex.split(command, posix=True)
+        return shlex.split(cmd_str, posix=True)
     except ValueError:
-        return command.split()
+        return cmd_str.split()
 
 
 def approval_required_for_command(argv: list[str], context: CommandContext | None = None) -> ApprovalDecision:
-    ctx = context or CommandContext()
-    parts = [part for part in argv if part != "--"]
+    ctx = context if isinstance(context, CommandContext) else CommandContext()
+    argv_list = argv if isinstance(argv, list) else list(argv) if isinstance(argv, (tuple, set)) else []
+    parts = [str(part or "") for part in argv_list if str(part or "") != "--"]
     lowered = _lower_parts(parts)
     if not lowered:
         return _decision(parts, ctx, "none", "none", "Empty command.")