Harden remote kubectl argument quoting

Assistant: codex
Assistant-Model: gpt-5.6-sol
Assistant-Session: 01a02991-be07-7bb3-8b6d-e9701b5621de

scripts/audit-core-whitehat-e2-credentials-02.py

@@ -7,6 +7,7 @@ import argparse
 import json
 import os
 import secrets
+import shlex
 import stat
 import subprocess
 import sys
@@ -109,7 +110,7 @@ class Operator:
 
     def kubectl(self, args: list[str], *, label: str, input_text: str | None = None, allow_missing: bool = False) -> subprocess.CompletedProcess[str]:
         return safe_run(
-            ["ssh", "-o", "BatchMode=yes", self.remote, "kubectl", *args],
+            ["ssh", "-o", "BatchMode=yes", self.remote, shlex.join(["kubectl", *args])],
             label=label,
             input_text=input_text,
             allow_missing=allow_missing,

tests/test_audit_core_whitehat_e2_credentials_02.py

@@ -1,7 +1,9 @@
 from __future__ import annotations
 
 import importlib.util
+import subprocess
 import unittest
+from unittest import mock
 from datetime import UTC, datetime
 from pathlib import Path
 
@@ -17,6 +19,25 @@ SPEC.loader.exec_module(module)
 
 
 class CredentialProcedureTests(unittest.TestCase):
+    def test_remote_kubectl_shell_quotes_template_arguments(self) -> None:
+        operator = object.__new__(module.Operator)
+        operator.remote = "railiance01"
+        template = 'go-template={{range $k, $_ := .data}}{{$k}}{{"\\n"}}{{end}}'
+        completed = subprocess.CompletedProcess([], 0, stdout="token-a\ntoken-b\n", stderr="")
+        with mock.patch.object(module, "safe_run", return_value=completed) as safe_run:
+            operator.kubectl(
+                ["-n", "whitehat", "get", "secret", "example", "-o", template],
+                label="verify keys",
+            )
+
+        command = safe_run.call_args.args[0]
+        self.assertEqual(["ssh", "-o", "BatchMode=yes", "railiance01"], command[:4])
+        self.assertEqual(
+            "kubectl -n whitehat get secret example -o "
+            "'go-template={{range $k, $_ := .data}}{{$k}}{{\"\\n\"}}{{end}}'",
+            command[4],
+        )
+
     def test_add_and_remove_only_named_temporary_identities(self) -> None:
         original = [{"name": "user-engine", "tokens": ["existing"]}]
         updated = module.add_temporary_identities(original, {"token-a": "a", "token-b": "b"})