Fix parsing of redirect tokens (#5322)

* Fix parsing of redirect tokens

In the parser, wrap redirect tokens such as ">", "a>", and "2>1" in a
tuple (along with the name of the redirect file, if applicable) to
differentiate them from regular command line arguments.

This ensures that a command like ![echo ">"] is parsed correctly.

We now need 2 types of IO-redirect tokens (IOREDIRECT1 and IOREDIRECT2)
so that the parser knows whether or not an IO-redirect has a file
argument. For example, "a>" has type IOREDIRECT1, whereas "2>1" has
type IOREDIRECT2.

* update ioredir tests

* add tests

* add news

* fix multiple redirections error message

news/fix-redirect-structure.rst

@@ -0,0 +1,23 @@
+**Added:**
+
+* <news item>
+
+**Changed:**
+
+* <news item>
+
+**Deprecated:**
+
+* <news item>
+
+**Removed:**
+
+* <news item>
+
+**Fixed:**
+
+* Redirect tokens in quotes (e.g. ">", "2>", "2>1") are now correctly passed to commands as regular arguments.
+
+**Security:**
+
+* <news item>

tests/test_integrations.py

@@ -886,6 +886,27 @@ def _echo(args):
     assert out == exp
 
 
+@skip_if_no_xonsh
+@pytest.mark.parametrize(
+    "cmd, exp",
+    [
+        ("echo '>'", ">\n"),
+        ("echo '2>'", "2>\n"),
+        ("echo '2>1'", "2>1\n"),
+    ],
+)
+def test_redirect_argument(cmd, exp):
+    script = f"""
+#!/usr/bin/env xonsh
+def _echo(args):
+    print(' '.join(args))
+aliases['echo'] = _echo
+{cmd}
+"""
+    out, _, _ = run_xonsh(script)
+    assert out == exp
+
+
 # issue 3402
 @skip_if_no_xonsh
 @skip_if_on_windows

tests/test_lexer.py

@@ -415,11 +415,14 @@ def test_float_literals(case):
     assert check_token(case, ["NUMBER", case, 0])
 
 
-@pytest.mark.parametrize(
-    "case", ["2>1", "err>out", "o>", "all>", "e>o", "e>", "out>", "2>&1"]
-)
-def test_ioredir(case):
-    assert check_tokens_subproc(case, [("IOREDIRECT", case, 2)], stop=-2)
+@pytest.mark.parametrize("case", ["o>", "all>", "e>", "out>"])
+def test_ioredir1(case):
+    assert check_tokens_subproc(case, [("IOREDIRECT1", case, 2)], stop=-2)
+
+
+@pytest.mark.parametrize("case", ["2>1", "err>out", "e>o", "2>&1"])
+def test_ioredir2(case):
+    assert check_tokens_subproc(case, [("IOREDIRECT2", case, 2)], stop=-2)
 
 
 @pytest.mark.parametrize("case", [">", ">>", "<", "e>", "> ", ">>   ", "<  ", "e> "])

xonsh/lexer.py

@@ -23,7 +23,8 @@
     ERRORTOKEN,
     GREATER,
     INDENT,
-    IOREDIRECT,
+    IOREDIRECT1,
+    IOREDIRECT2,
     LESS,
     MATCH,
     NAME,
@@ -101,7 +102,8 @@ def token_map():
     }
     for op, typ in _op_map.items():
         tm[(OP, op)] = typ
-    tm[IOREDIRECT] = "IOREDIRECT"
+    tm[IOREDIRECT1] = "IOREDIRECT1"
+    tm[IOREDIRECT2] = "IOREDIRECT2"
     tm[STRING] = "STRING"
     tm[DOLLARNAME] = "DOLLAR_NAME"
     tm[NUMBER] = "NUMBER"
@@ -255,7 +257,7 @@ def handle_redirect(state, token):
     key = (typ, st) if (typ, st) in token_map else typ
     new_tok = _new_token(token_map[key], st, token.start)
     if state["pymode"][-1][0]:
-        if typ == IOREDIRECT:
+        if typ in (IOREDIRECT1, IOREDIRECT2):
             # Fix Python mode code that was incorrectly recognized as an
             # IOREDIRECT by the tokenizer (see issue #4994).
             # The tokenizer does not know when the code should be tokenized in
@@ -310,7 +312,8 @@ def special_handlers():
         LESS: handle_redirect,
         GREATER: handle_redirect,
         RIGHTSHIFT: handle_redirect,
-        IOREDIRECT: handle_redirect,
+        IOREDIRECT1: handle_redirect,
+        IOREDIRECT2: handle_redirect,
         (OP, "<"): handle_redirect,
         (OP, ">"): handle_redirect,
         (OP, ">>"): handle_redirect,

xonsh/parsers/base.py

@@ -3432,12 +3432,20 @@ def p_subproc_atom_subproc_inject_bang(self, p):
 
     def p_subproc_atom_redirect(self, p):
         """
-        subproc_atom : GT
-                     | LT
-                     | RSHIFT
-                     | IOREDIRECT
-        """
-        p0 = ast.const_str(s=p[1], lineno=self.lineno, col_offset=self.col)
+        subproc_atom : GT WS subproc_atom
+                     | LT WS subproc_atom
+                     | RSHIFT WS subproc_atom
+                     | IOREDIRECT1 WS subproc_atom
+                     | IOREDIRECT2
+        """
+        operator = ast.const_str(s=p[1], lineno=self.lineno, col_offset=self.col)
+        elts = [operator] if len(p) == 2 else [operator, p[3]]
+        p0 = ast.Tuple(
+            elts=elts,
+            ctx=ast.Load(),
+            lineno=self.lineno,
+            col_offset=self.col,
+        )
         p0._cliarg_action = "append"
         p[0] = p0
 
@@ -3523,7 +3531,8 @@ def _attach_subproc_arg_part_rules(self):
             "LT",
             "LSHIFT",
             "RSHIFT",
-            "IOREDIRECT",
+            "IOREDIRECT1",
+            "IOREDIRECT2",
             "SEARCHPATH",
             "INDENT",
             "DEDENT",

xonsh/parsers/completion_context.py

@@ -330,7 +330,8 @@ class CompletionContextParser:
         "LT",
         "GT",
         "RSHIFT",
-        "IOREDIRECT",
+        "IOREDIRECT1",
+        "IOREDIRECT2",
     }
     used_tokens |= io_redir_tokens
     artificial_tokens = {"ANY"}

xonsh/procs/specs.py

@@ -172,10 +172,6 @@ def _O2E_MAP():
     return frozenset({f"{o}>{e}" for e in _REDIR_ERR for o in _REDIR_OUT if o != ""})
 
 
-def _is_redirect(x):
-    return isinstance(x, str) and _REDIR_REGEX.match(x)
-
-
 def safe_open(fname, mode, buffering=-1):
     """Safely attempts to open a file in for xonsh subprocs."""
     # file descriptors
@@ -401,7 +397,7 @@ def stdin(self, value):
         else:
             safe_close(value)
             msg = "Multiple inputs for stdin for {0!r}"
-            msg = msg.format(" ".join(self.args))
+            msg = msg.format(self.get_command_str())
             raise xt.XonshError(msg)
 
     @property
@@ -417,7 +413,7 @@ def stdout(self, value):
         else:
             safe_close(value)
             msg = "Multiple redirections for stdout for {0!r}"
-            msg = msg.format(" ".join(self.args))
+            msg = msg.format(self.get_command_str())
             raise xt.XonshError(msg)
 
     @property
@@ -433,9 +429,14 @@ def stderr(self, value):
         else:
             safe_close(value)
             msg = "Multiple redirections for stderr for {0!r}"
-            msg = msg.format(" ".join(self.args))
+            msg = msg.format(self.get_command_str())
             raise xt.XonshError(msg)
 
+    def get_command_str(self):
+        return " ".join(
+            " ".join(arg) if isinstance(arg, tuple) else arg for arg in self.args
+        )
+
     #
     # Execution methods
     #
@@ -579,8 +580,7 @@ def build(kls, cmd, *, cls=subprocess.Popen, **kwargs):
         spec = kls(cmd, cls=cls, **kwargs)
         # modifications that alter cmds must come after creating instance
         # perform initial redirects
-        spec.redirect_leading()
-        spec.redirect_trailing()
+        spec.resolve_redirects()
         # apply aliases
         spec.resolve_alias()
         spec.resolve_binary_loc()
@@ -590,26 +590,16 @@ def build(kls, cmd, *, cls=subprocess.Popen, **kwargs):
         spec.resolve_stack()
         return spec
 
-    def redirect_leading(self):
-        """Manage leading redirects such as with '< input.txt COMMAND'."""
-        while len(self.cmd) >= 3 and self.cmd[0] == "<":
-            self.stdin = safe_open(self.cmd[1], "r")
-            self.cmd = self.cmd[2:]
-
-    def redirect_trailing(self):
-        """Manages trailing redirects."""
-        while True:
-            cmd = self.cmd
-            if len(cmd) >= 3 and _is_redirect(cmd[-2]):
-                streams = _redirect_streams(cmd[-2], cmd[-1])
-                self.stdin, self.stdout, self.stderr = streams
-                self.cmd = cmd[:-2]
-            elif len(cmd) >= 2 and _is_redirect(cmd[-1]):
-                streams = _redirect_streams(cmd[-1])
+    def resolve_redirects(self):
+        """Manages redirects."""
+        new_cmd = []
+        for c in self.cmd:
+            if isinstance(c, tuple):
+                streams = _redirect_streams(*c)
                 self.stdin, self.stdout, self.stderr = streams
-                self.cmd = cmd[:-1]
             else:
-                break
+                new_cmd.append(c)
+        self.cmd = new_cmd
 
     def resolve_alias(self):
         """Sets alias in command, if applicable."""
@@ -667,8 +657,7 @@ def resolve_executable_commands(self):
         else:
             self.cmd = alias + self.cmd[1:]
             # resolve any redirects the aliases may have applied
-            self.redirect_leading()
-            self.redirect_trailing()
+            self.resolve_redirects()
         if self.binary_loc is None:
             return
         try:

xonsh/tokenize.py

@@ -110,7 +110,8 @@
     "ATDOLLAR",
     "ATEQUAL",
     "DOLLARNAME",
-    "IOREDIRECT",
+    "IOREDIRECT1",
+    "IOREDIRECT2",
     "MATCH",
     "CASE",
 ]
@@ -135,8 +136,11 @@
 SEARCHPATH = N_TOKENS
 tok_name[N_TOKENS] = "SEARCHPATH"
 N_TOKENS += 1
-IOREDIRECT = N_TOKENS
-tok_name[N_TOKENS] = "IOREDIRECT"
+IOREDIRECT1 = N_TOKENS
+tok_name[N_TOKENS] = "IOREDIRECT1"
+N_TOKENS += 1
+IOREDIRECT2 = N_TOKENS
+tok_name[N_TOKENS] = "IOREDIRECT2"
 N_TOKENS += 1
 DOLLARNAME = N_TOKENS
 tok_name[N_TOKENS] = "DOLLARNAME"
@@ -335,10 +339,11 @@ def maybe(*choices):
 )
 IORedirect = group(group(*_redir_map), f"{group(*_redir_names)}>>?")
 
-_redir_check_0 = set(_redir_map)
-_redir_check_1 = {f"{i}>" for i in _redir_names}.union(_redir_check_0)
+_redir_check_map = frozenset(_redir_map)
+
+_redir_check_1 = {f"{i}>" for i in _redir_names}
 _redir_check_2 = {f"{i}>>" for i in _redir_names}.union(_redir_check_1)
-_redir_check = frozenset(_redir_check_2)
+_redir_check_single = frozenset(_redir_check_2)
 
 Operator = group(
     r"\*\*=?",
@@ -1004,8 +1009,10 @@ def _tokenize(readline, encoding, tolerant=False, tokenize_ioredirects=True):
                     continue
                 token, initial = line[start:end], line[start]
 
-                if token in _redir_check:
-                    yield TokenInfo(IOREDIRECT, token, spos, epos, line)
+                if token in _redir_check_single:
+                    yield TokenInfo(IOREDIRECT1, token, spos, epos, line)
+                elif token in _redir_check_map:
+                    yield TokenInfo(IOREDIRECT2, token, spos, epos, line)
                 elif initial in numchars or (  # ordinary number
                     initial == "." and token != "." and token != "..."
                 ):