diff --git a/changelogs/fragments/t9511_vyos_terminal_ansi_fixes.yaml b/changelogs/fragments/t9511_vyos_terminal_ansi_fixes.yaml
new file mode 100644
index 00000000..8234d939
--- /dev/null
+++ b/changelogs/fragments/t9511_vyos_terminal_ansi_fixes.yaml
@@ -0,0 +1,4 @@
+---
+bugfixes:
+ - vyos terminal plugin - strip the bracketed-paste escape sequences ``ESC[?2004h`` and ``ESC[?2004l`` emitted by vbash on rolling images with readline ``enable-bracketed-paste on``. Previously they leaked into command output, so ``vyos_command`` output started with ``[?2004l``, JSON output failed to parse and empty output was treated as non-empty (https://github.com/vyos/vyos.vyos/issues/XXX).
+ - vyos terminal plugin - fix loss of one character at every terminal-width (512 column) wrap in long output lines. The wrap artefact (space followed by backspace) was stripped by the generic backspace rule together with the following real character, corrupting output such as ``show configuration json`` on all VyOS versions (https://github.com/vyos/vyos.vyos/issues/XXX).
diff --git a/plugins/terminal/vyos.py b/plugins/terminal/vyos.py
index 29acb803..d2c98988 100644
--- a/plugins/terminal/vyos.py
+++ b/plugins/terminal/vyos.py
@@ -1,73 +1,83 @@
#
# (c) 2016 Red Hat Inc.
#
# This file is part of Ansible
#
# Ansible is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Ansible is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with Ansible. If not, see .
#
from __future__ import absolute_import, division, print_function
+
__metaclass__ = type
import os
import re
from ansible.errors import AnsibleConnectionFailure
from ansible_collections.ansible.netcommon.plugins.plugin_utils.terminal_base import TerminalBase
class TerminalModule(TerminalBase):
terminal_stdout_re = [
re.compile(rb"[\r\n]?[\w+\-\.:\/\[\]]+(?:\([^\)]+\)){,3}(?:>|#) ?$"),
re.compile(rb"\@[\w\-\.]+:\S+?[>#\$] ?$"),
]
terminal_stderr_re = [
re.compile(rb"\n\s*Invalid command:"),
re.compile(rb"\nCommit failed"),
re.compile(rb"\n\s+Set failed"),
re.compile(rb"\n\s+Delete failed"),
]
- ansi_re = TerminalBase.ansi_re + [
- # Color codes
- re.compile(rb"\x1b\[(\d+(;\d+)*)?m"),
- # Clear line (CSI K)
- re.compile(rb"\x1b\[K"),
- # Xterm change cursor mode (CSI ? 1 [h|l])
- re.compile(rb"\x1b\[\?1(h|l)"),
- # Xterm change keypad (ESC [=|>])
- re.compile(rb"\x1b(=|>)"),
- # Xterm window title string (OSC BEL)
- re.compile(rb"\x1b]0;[^\x07]*\x07"),
- ]
+ ansi_re = (
+ [
+ re.compile(
+ rb"\x1b\[\?2004[hl]",
+ ), # bracketed paste (rolling), before \x08. can eat its ESC
+ re.compile(rb" \x08"), # 512-col wrap artefact, before \x08. eats the real char
+ ]
+ + TerminalBase.ansi_re
+ + [
+ # Color codes
+ re.compile(rb"\x1b\[(\d+(;\d+)*)?m"),
+ # Clear line (CSI K)
+ re.compile(rb"\x1b\[K"),
+ # Xterm change cursor mode (CSI ? 1 [h|l])
+ re.compile(rb"\x1b\[\?1(h|l)"),
+ # Xterm change keypad (ESC [=|>])
+ re.compile(rb"\x1b(=|>)"),
+ # Xterm window title string (OSC <title string> BEL)
+ re.compile(rb"\x1b]0;[^\x07]*\x07"),
+ ]
+ )
terminal_config_prompt = re.compile(r"^.+#$")
try:
terminal_length = os.getenv("ANSIBLE_VYOS_TERMINAL_LENGTH", 10000)
terminal_length = int(terminal_length)
except ValueError:
raise AnsibleConnectionFailure(
"Invalid value set for vyos terminal length '%s', value should be a valid integer string"
% terminal_length,
)
def on_open_shell(self):
try:
for cmd in (b"set terminal length 0", b"set terminal width 512"):
self._exec_cli_command(cmd)
self._exec_cli_command(b"set terminal length %d" % self.terminal_length)
except AnsibleConnectionFailure:
raise AnsibleConnectionFailure("unable to set terminal parameters")
diff --git a/tests/unit/terminal/__init__.py b/tests/unit/terminal/__init__.py
new file mode 100644
index 00000000..e69de29b
diff --git a/tests/unit/terminal/test_vyos_terminal.py b/tests/unit/terminal/test_vyos_terminal.py
new file mode 100644
index 00000000..a1d455d9
--- /dev/null
+++ b/tests/unit/terminal/test_vyos_terminal.py
@@ -0,0 +1,66 @@
+#
+# This file is part of Ansible
+#
+# Ansible is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# Ansible is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with Ansible. If not, see <http://www.gnu.org/licenses/>.
+#
+from __future__ import absolute_import, division, print_function
+
+
+__metaclass__ = type
+
+import unittest
+
+from ansible_collections.vyos.vyos.plugins.terminal.vyos import TerminalModule
+
+
+def strip(data):
+ """Apply ansi_re in order, the same way network_cli._strip() does."""
+ for regex in TerminalModule.ansi_re:
+ data = regex.sub(b"", data)
+ return data
+
+
+class TestBracketedPaste(unittest.TestCase):
+ """Rolling vbash runs readline with enable-bracketed-paste on."""
+
+ def test_command_echo(self):
+ raw = b"\x1b[?2004l\r\x1b[?1h\x1b=\r" b"Version: VyOS 999.202609232109\x1b[m\r\n"
+ out = strip(raw)
+ self.assertNotIn(b"2004", out)
+ self.assertIn(b"Version: VyOS 999.202609232109", out)
+
+ def test_prompt(self):
+ raw = b"\x1b[?2004h\x1b]0;vyos@vyos: ~\x07vyos@vyos:~$ "
+ out = strip(raw)
+ self.assertEqual(out, b"vyos@vyos:~$ ")
+ self.assertTrue(any(r.search(out) for r in TerminalModule.terminal_stdout_re))
+
+ def test_empty_output(self):
+ self.assertEqual(strip(b"\x1b[?2004l\r\n").strip(), b"")
+
+
+class TestWrapArtefact(unittest.TestCase):
+ """At each terminal-width wrap the stream carries ESC[m, space, backspace, char."""
+
+ def test_wrap_lts(self):
+ raw = b'"ti\x1b[m \x08me3.vyos.net": {}'
+ self.assertEqual(strip(raw), b'"time3.vyos.net": {}')
+
+ def test_wrap_rolling_quote(self):
+ raw = b'"ssh": {}}, \x1b[m \x08"system": {'
+ self.assertEqual(strip(raw), b'"ssh": {}}, "system": {')
+
+ def test_wrap_rolling_letter(self):
+ raw = b'"al\x1b[m \x08l": {"level": "info"}'
+ self.assertEqual(strip(raw), b'"all": {"level": "info"}')