Bash integration: Fix clone-in-kitty not working on bash >= 5.2 if environment variable values contain newlines or other special characters

Bash >= 5.2 changed the export command to output values using $' escaping when they contain special characters.
Fixes #5629

docs/changelog.rst

@@ -51,6 +51,8 @@ Detailed list of changes
 
 - Add an option :opt:`background_tint_gaps` to control background image tinting for window gaps (:iss:`5596`)
 
+- Bash integration: Fix ``clone-in-kitty`` not working on bash >= 5.2 if environment variable values contain newlines or other special characters (:iss:`5629`)
+
 
 0.26.5 [2022-11-07]
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

kitty/bash.py

@@ -0,0 +1,138 @@
+#!/usr/bin/env python
+# License: GPLv3 Copyright: 2022, Kovid Goyal <kovid at kovidgoyal.net>
+
+import string
+from typing import Dict, List, Tuple
+
+ansi_c_escapes = {
+    'a': '\a',
+    'b': '\b',
+    'e': '\x1b',
+    'E': '\x1b',
+    'f': '\f',
+    'n': '\n',
+    'r': '\r',
+    't': '\t',
+    'v': '\v',
+    '\\': '\\',
+    "'": "'",
+    '"': '"',
+    '?': '?',
+}
+
+
+def ctrl_mask_char(ch: str) -> str:
+    try:
+        o = ord(ch)
+    except Exception:
+        return ch
+    if o > 127:
+        return ch
+    return chr(o & 0b0011111)
+
+
+def read_digit(text: str, pos: int, max_len: int, valid_digits: str, base: int) -> Tuple[str, int]:
+    epos = pos
+    while (epos - pos) <= max_len and epos < len(text) and text[epos] in valid_digits:
+        epos += 1
+    raw = text[pos:epos]
+    try:
+        return chr(int(raw, base)), epos
+    except Exception:
+        return raw, epos
+
+
+def read_hex_digit(text: str, pos: int, max_len: int) -> Tuple[str, int]:
+    return read_digit(text, pos, max_len, string.digits + 'abcdefABCDEF', 16)
+
+
+def read_octal_digit(text: str, pos: int) -> Tuple[str, int]:
+    return read_digit(text, pos, 3, '01234567', 8)
+
+
+def decode_ansi_c_quoted_string(text: str, pos: int) -> Tuple[str, int]:
+    buf: List[str] = []
+    a = buf.append
+    while pos < len(text):
+        ch = text[pos]
+        pos += 1
+        if ch == '\\':
+            ec = text[pos]
+            pos += 1
+            ev = ansi_c_escapes.get(ec)
+            if ev is None:
+                if ec == 'c' and pos + 1 < len(text):
+                    a(ctrl_mask_char(text[pos]))
+                    pos += 1
+                elif ec in 'xuU' and pos + 1 < len(text):
+                    hd, pos = read_hex_digit(text, pos, {'x': 2, 'u': 4, 'U': 8}[ec])
+                    a(hd)
+                elif ec.isdigit():
+                    hd, pos = read_octal_digit(text, pos-1)
+                    a(hd)
+                else:
+                    a(ec)
+            else:
+                a(ev)
+        elif ch == "'":
+            break
+        else:
+            a(ch)
+    return ''.join(buf), pos
+
+
+def decode_double_quoted_string(text: str, pos: int) -> Tuple[str, int]:
+    escapes = r'"\$`'
+    buf: List[str] = []
+    a = buf.append
+    while pos < len(text):
+        ch = text[pos]
+        pos += 1
+        if ch == '\\':
+            if text[pos] in escapes:
+                a(text[pos])
+                pos += 1
+                continue
+            a(ch)
+        elif ch == '"':
+            break
+        else:
+            a(ch)
+    return ''.join(buf), pos
+
+
+def parse_modern_bash_env(text: str) -> Dict[str, str]:
+    ans = {}
+    for line in text.splitlines():
+        idx = line.find('=')
+        if idx < 0:
+            break
+        key = line[:idx].rpartition(' ')[2]
+        val = line[idx+1:]
+        if val.startswith('"'):
+            val = decode_double_quoted_string(val, 1)[0]
+        else:
+            val = decode_ansi_c_quoted_string(val, 2)[0]
+        ans[key] = val
+    return ans
+
+
+def parse_bash_env(text: str, bash_version: str) -> Dict[str, str]:
+    # See https://www.gnu.org/software/bash/manual/html_node/Double-Quotes.html
+    parts = bash_version.split('.')
+    bv = tuple(map(int, parts[:2]))
+    if bv >= (5, 2):
+        return parse_modern_bash_env(text)
+    ans = {}
+    pos = 0
+    while pos < len(text):
+        idx = text.find('="', pos)
+        if idx < 0:
+            break
+        i = text.rfind(' ', 0, idx)
+        if i < 0:
+            break
+        key = text[i+1:idx]
+        pos = idx + 2
+        ans[key], pos = decode_double_quoted_string(text, pos)
+    return ans

kitty/launch.py

@@ -606,39 +606,6 @@ def parse_opts_for_clone(args: List[str]) -> Tuple[LaunchCLIOptions, List[str]]:
     return default_opts, unsafe_args
 
 
-def parse_bash_env(text: str) -> Dict[str, str]:
-    # See https://www.gnu.org/software/bash/manual/html_node/Double-Quotes.html
-    ans = {}
-    pos = 0
-    escapes = r'"\$`'
-    while pos < len(text):
-        idx = text.find('="', pos)
-        if idx < 0:
-            break
-        i = text.rfind(' ', 0, idx)
-        if i < 0:
-            break
-        key = text[i+1:idx]
-        pos = idx + 2
-        buf: List[str] = []
-        a = buf.append
-        while pos < len(text):
-            ch = text[pos]
-            pos += 1
-            if ch == '\\':
-                if text[pos] in escapes:
-                    a(text[pos])
-                    pos += 1
-                    continue
-                a(ch)
-            elif ch == '"':
-                break
-            else:
-                a(ch)
-        ans[key] = ''.join(buf)
-    return ans
-
-
 def parse_null_env(text: str) -> Dict[str, str]:
     ans = {}
     for line in text.split('\0'):
@@ -782,12 +749,13 @@ class CloneCmd:
         self.shell = ''
         self.envfmt = 'default'
         self.pid = -1
+        self.bash_version = ''
         self.history = ''
         self.parse_message(msg)
         self.opts = parse_opts_for_clone(self.args)[0]
 
     def parse_message(self, msg: str) -> None:
-        simple = 'pid', 'envfmt', 'shell'
+        simple = 'pid', 'envfmt', 'shell', 'bash_version'
         for k, v in parse_message(msg, simple):
             if k in simple:
                 if k == 'pid':
@@ -797,7 +765,11 @@ class CloneCmd:
             elif k == 'a':
                 self.args.append(v)
             elif k == 'env':
-                env = parse_bash_env(v) if self.envfmt == 'bash' else parse_null_env(v)
+                if self.envfmt == 'bash':
+                    from .bash import parse_bash_env
+                    env = parse_bash_env(v, self.bash_version)
+                else:
+                    env = parse_null_env(v)
                 self.env = {k: v for k, v in env.items() if k not in {
                     'HOME', 'LOGNAME', 'USER', 'PWD',
                     # some people export these. We want the shell rc files to recreate them

kitty_tests/shell_integration.py

@@ -16,6 +16,7 @@ from kitty.fast_data_types import CURSOR_BEAM, CURSOR_BLOCK, CURSOR_UNDERLINE
 from kitty.shell_integration import (
     setup_bash_env, setup_fish_env, setup_zsh_env
 )
+from kitty.bash import decode_ansi_c_quoted_string
 
 from . import BaseTest
 
@@ -364,9 +365,21 @@ PS1="{ps1}"
         run_test('bash -l .bashrc', 'profile', rc='echo ok;read', wait_string='ok', assert_not_in=True)
         run_test('bash -il -- .bashrc', 'profile', rc='echo ok;read', wait_string='ok')
 
-        with self.run_shell(shell='bash', setup_env=partial(setup_env, set()), cmd='bash', rc=f'''PS1="{ps1}"\nexport ES=$'a\n `b` c\n$d' ''') as pty:
+        with self.run_shell(shell='bash', setup_env=partial(setup_env, set()), cmd='bash',
+                            rc=f'''PS1="{ps1}"\nexport ES=$'a\n `b` c\n$d'\nexport ES2="XXX" ''') as pty:
             pty.callbacks.clear()
             pty.send_cmd_to_child('clone-in-kitty')
             pty.wait_till(lambda: len(pty.callbacks.clone_cmds) == 1)
             env = pty.callbacks.clone_cmds[0].env
             self.ae(env.get('ES'), 'a\n `b` c\n$d', f'Screen contents: {pty.screen_contents()!r}')
+            self.ae(env.get('ES2'), 'XXX', f'Screen contents: {pty.screen_contents()!r}')
+        for q, e in {
+            'a': 'a',
+            r'a\ab': 'a\ab',
+            r'a\x7z': 'a\x07z',
+            r'a\7b': 'a\007b',
+            r'a\U1f345x': 'a🍅x',
+            r'a\c b': 'a\0b',
+        }.items():
+            q = q + "'"
+            self.ae(decode_ansi_c_quoted_string(q, 0)[0], e, f'Failed to decode: {q!r}')

shell-integration/bash/kitty.bash

@@ -332,7 +332,8 @@ _ksi_transmit_data() {
 }
 
 clone-in-kitty() {
-    builtin local data="shell=bash,pid=$$,cwd=$(builtin printf "%s" "$PWD" | builtin command base64),envfmt=bash,env=$(builtin export | builtin command base64)"
+    builtin local bv="${BASH_VERSINFO[0]}.${BASH_VERSINFO[1]}.${BASH_VERSINFO[2]}"
+    builtin local data="shell=bash,pid=$$,bash_version=$bv,cwd=$(builtin printf "%s" "$PWD" | builtin command base64),envfmt=bash,env=$(builtin export | builtin command base64)"
     while :; do
         case "$1" in
             "") break;;