diff --git a/src/python/jw/pkg/lib/ec/ssh/AsyncSSH.py b/src/python/jw/pkg/lib/ec/ssh/AsyncSSH.py
index 67b110a3..dd8ec2ae 100644
--- a/src/python/jw/pkg/lib/ec/ssh/AsyncSSH.py
+++ b/src/python/jw/pkg/lib/ec/ssh/AsyncSSH.py
@@ -6,6 +6,8 @@ from ...log import *
 from ...ExecContext import Result
 from ..SSHClient import SSHClient as Base
 
+from .util import join_cmd
+
 _USE_DEFAULT_KNOWN_HOSTS = object()
 
 class AsyncSSH(Base):
@@ -52,7 +54,7 @@ class AsyncSSH(Base):
         if not cmd:
             raise ValueError("cmd must not be empty")
 
-        inner = f"exec {shlex.join(cmd)}"
+        inner = f"exec {join_cmd(cmd)}"
         if wd is not None:
             inner = f"cd {shlex.quote(wd)} && {inner}"
 
diff --git a/src/python/jw/pkg/lib/ec/ssh/Exec.py b/src/python/jw/pkg/lib/ec/ssh/Exec.py
index 8202c5c4..b1267800 100644
--- a/src/python/jw/pkg/lib/ec/ssh/Exec.py
+++ b/src/python/jw/pkg/lib/ec/ssh/Exec.py
@@ -2,10 +2,9 @@ from __future__ import annotations
 
 from typing import TYPE_CHECKING
 
-import shlex
-
 from ...util import run_cmd
 from ..SSHClient import SSHClient as Base
+from .util import join_cmd
 
 if TYPE_CHECKING:
     from ...ExecContext import Result
@@ -41,5 +40,5 @@ class Exec(Base):
 
     async def _run_ssh(self, cmd: list[str], cmd_input: str|None, *args, **kwargs) -> Result:
         self.__init_askpass()
-        return await run_cmd(['ssh', self.hostname, shlex.join(cmd)], cmd_input=cmd_input)
+        return await run_cmd(['ssh', self.hostname, join_cmd(cmd)], cmd_input=cmd_input)
 
diff --git a/src/python/jw/pkg/lib/ec/ssh/Paramiko.py b/src/python/jw/pkg/lib/ec/ssh/Paramiko.py
index 7ed4ae33..57c3bc44 100644
--- a/src/python/jw/pkg/lib/ec/ssh/Paramiko.py
+++ b/src/python/jw/pkg/lib/ec/ssh/Paramiko.py
@@ -3,12 +3,13 @@ from __future__ import annotations
 from typing import TYPE_CHECKING
 
 import paramiko # type: ignore # error: Library stubs not installed for "paramiko"
-import shlex
 
 from ...log import *
 from ...ExecContext import Result
 from ..SSHClient import SSHClient as Base
 
+from .util import join_cmd
+
 class Paramiko(Base):
 
     def __init__(self, *args, **kwargs) -> None:
@@ -45,9 +46,9 @@ class Paramiko(Base):
 
     async def _run_ssh(self, cmd: list[str], cmd_input: str|None, *args, **kwargs) -> Result:
         try:
-            stdin, stdout, stderr = self.__ssh.exec_command(shlex.join(cmd), timeout=self.__timeout)
+            stdin, stdout, stderr = self.__ssh.exec_command(join_cmd(cmd), timeout=self.__timeout)
         except Exception as e:
-            log(ERR, f'Command failed for {self.uri}: "{shlex.join(cmd)}"')
+            log(ERR, f'Command failed for {self.uri}: "{join_cmd(cmd)}"')
             raise
         if cmd_input is not None:
             stdin.write(cmd_input)
diff --git a/src/python/jw/pkg/lib/ec/ssh/util.py b/src/python/jw/pkg/lib/ec/ssh/util.py
new file mode 100644
index 00000000..b7cd6a9b
--- /dev/null
+++ b/src/python/jw/pkg/lib/ec/ssh/util.py
@@ -0,0 +1,41 @@
+# -*- coding: utf-8 -*-
+
+from typing import Iterable
+
+import shlex
+
+DEFAULT_SHELL_OPERATORS = {
+    # redirections
+    ">", ">>", "<", "<<", "<<-", "<&", ">&", "<>", ">|",
+    "1>", "1>>", "2>", "2>>",
+
+    # pipelines / control
+    "|", "||", "&", "&&", ";",
+
+    # grouping
+    "(", ")",
+}
+
+def join_cmd(
+    cmd: Iterable[str],
+    operators: set[str] = DEFAULT_SHELL_OPERATORS,
+) -> str:
+    """
+    Join a token list into a POSIX-shell command string.
+
+    Tokens in `operators` are emitted verbatim.
+    Everything else is shell-quoted with shlex.quote().
+
+    Example:
+        ["echo", "hello world", ">", "/tmp/out file"]
+        -> "echo 'hello world' > '/tmp/out file'"
+    """
+    ret: list[str] = []
+    for token in cmd:
+        if not isinstance(token, str):
+            token = str(token)
+        if token in operators:
+            ret.append(token)
+        else:
+            ret.append(shlex.quote(token))
+    return ' '.join(ret)