fix(template): escape environment variable literals

template/server/messaging.py

@@ -1,6 +1,7 @@
 import datetime
 import json
 import logging
+import shlex
 import uuid
 import asyncio
 
@@ -155,29 +156,32 @@ def _get_execute_request(
     def _set_env_var_snippet(self, key: str, value: str) -> str:
         """Get environment variable set command for the current language."""
         if self.language == "python":
-            return f"import os; os.environ['{key}'] = '{value}'"
+            return f"import os; os.environ[{key!r}] = {value!r}"
         elif self.language in ["javascript", "typescript"]:
-            return f"process.env['{key}'] = '{value}'"
+            return f"process.env[{json.dumps(key)}] = {json.dumps(value)}"
         elif self.language == "r":
-            return f'Sys.setenv({key} = "{value}")'
+            return (
+                f"do.call(Sys.setenv, setNames(list({json.dumps(value, ensure_ascii=False)}), "
+                f"{json.dumps(key, ensure_ascii=False)}))"
+            )
         elif self.language == "java":
-            return f'System.setProperty("{key}", "{value}");'
+            return f"System.setProperty({json.dumps(key)}, {json.dumps(value)});"
         elif self.language == "bash":
-            return f"export {key}='{value}'"
+            return f"export {shlex.quote(key + '=' + value)}"
         return ""
 
     def _delete_env_var_snippet(self, key: str) -> str:
         """Get environment variable delete command for the current language."""
         if self.language == "python":
-            return f"import os; del os.environ['{key}']"
+            return f"import os; del os.environ[{key!r}]"
         elif self.language in ["javascript", "typescript"]:
-            return f"delete process.env['{key}']"
+            return f"delete process.env[{json.dumps(key)}]"
         elif self.language == "r":
-            return f"Sys.unsetenv('{key}')"
+            return f"Sys.unsetenv({json.dumps(key, ensure_ascii=False)})"
         elif self.language == "java":
-            return f'System.clearProperty("{key}");'
+            return f"System.clearProperty({json.dumps(key)});"
         elif self.language == "bash":
-            return f"unset {key}"
+            return f"unset {shlex.quote(key)}"
         return ""
 
     def _set_env_vars_code(self, env_vars: Dict[StrictStr, str]) -> str:

tests/env_vars/test_escaping.py

@@ -0,0 +1,48 @@
+import json
+import shlex
+
+import pytest
+
+from harness import CodeInterpreter
+
+
+@pytest.mark.skip_debug
+@pytest.mark.parametrize("language", ["python", "javascript", "typescript", "r", "java", "bash"])
+@pytest.mark.parametrize("value", ["it's", 'a"b', "line1\nline2", r"C:\new\token"])
+def test_env_literal_round_trip_and_cleanup(client: CodeInterpreter, language, value):
+    # Bash requires identifier-shaped variable names; other kernels accept strings.
+    key = "ESCAPED_TOKEN" if language == "bash" else "ESCAPED_'\"\\\n_TOKEN"
+    key_literal, value_literal = json.dumps(key), json.dumps(value)
+    checks = {
+        "python": (
+            f"import os; assert os.environ[{key!r}] == {value!r}",
+            f"import os; assert {key!r} not in os.environ",
+        ),
+        "javascript": (
+            f"if (process.env[{key_literal}] !== {value_literal}) throw Error('value');",
+            f"if ({key_literal} in process.env) throw Error('cleanup');",
+        ),
+        "r": (
+            f"stopifnot(Sys.getenv({key_literal}) == {value_literal})",
+            f"stopifnot(is.na(Sys.getenv({key_literal}, unset=NA)))",
+        ),
+        "java": (
+            f"if (!System.getProperty({key_literal}).equals({value_literal})) "
+            'throw new RuntimeException("value");',
+            f"if (System.getProperty({key_literal}) != null) "
+            'throw new RuntimeException("cleanup");',
+        ),
+        "bash": (
+            f'test "$ESCAPED_TOKEN" = {shlex.quote(value)} || echo VALUE_MISMATCH',
+            'test "${ESCAPED_TOKEN+x}" != x || echo CLEANUP_FAILED',
+        ),
+    }
+    checks["typescript"] = checks["javascript"]
+    client.wait_for_kernel(language)
+    check_value, check_cleanup = checks[language]
+    execution = client.run_code(check_value, language=language, envs={key: value})
+    assert execution.error is None
+    assert not execution.stdout
+    cleanup = client.run_code(check_cleanup, language=language)
+    assert cleanup.error is None
+    assert not cleanup.stdout