Use literal quoting for env vars sent over ssh by clone
This commit is contained in:
parent
25f022cc14
commit
775584b5a5
@ -106,28 +106,39 @@ def set_env_in_cmdline(env: Dict[str, str], argv: List[str]) -> None:
|
||||
quote_pat = re.compile('([\\`"])')
|
||||
|
||||
|
||||
def quote_env_val(x: str) -> str:
|
||||
x = quote_pat.sub(r'\\\1', x)
|
||||
x = x.replace('$(', r'\$(') # prevent execution with $()
|
||||
return f'"{x}"'
|
||||
def quote_env_val(x: str, literal_quote: bool = False) -> str:
|
||||
if not literal_quote:
|
||||
x = quote_pat.sub(r'\\\1', x)
|
||||
x = x.replace('$(', r'\$(') # prevent execution with $()
|
||||
return f'"{x}"'
|
||||
if "'" in x:
|
||||
x = quote_pat.sub(r'\\\1', x)
|
||||
x = x.replace('$', r'\$')
|
||||
return f'"{x}"'
|
||||
return f'{x}'
|
||||
|
||||
|
||||
def serialize_env(env: Dict[str, str], base_env: Dict[str, str], for_python: bool = False) -> bytes:
|
||||
def serialize_env(literal_env: Dict[str, str], env: Dict[str, str], base_env: Dict[str, str], for_python: bool = False) -> bytes:
|
||||
lines = []
|
||||
literal_quote = True
|
||||
|
||||
if for_python:
|
||||
def a(k: str, val: str = '', prefix: str = 'export') -> None:
|
||||
if val:
|
||||
lines.append(f'{prefix} {json.dumps((k, val))}')
|
||||
lines.append(f'{prefix} {json.dumps((k, val, literal_quote))}')
|
||||
else:
|
||||
lines.append(f'{prefix} {json.dumps((k,))}')
|
||||
else:
|
||||
def a(k: str, val: str = '', prefix: str = 'export') -> None:
|
||||
if val:
|
||||
lines.append(f'{prefix} {shlex.quote(k)}={quote_env_val(val)}')
|
||||
lines.append(f'{prefix} {shlex.quote(k)}={quote_env_val(val, literal_quote)}')
|
||||
else:
|
||||
lines.append(f'{prefix} {shlex.quote(k)}')
|
||||
|
||||
for k, v in literal_env.items():
|
||||
a(k, v)
|
||||
|
||||
literal_quote = False
|
||||
for k in sorted(env):
|
||||
v = env[k]
|
||||
if v == DELETE_ENV_VAR:
|
||||
@ -188,7 +199,6 @@ def make_tarfile(ssh_opts: SSHOptions, base_env: Dict[str, str], compression: st
|
||||
'TERM': os.environ.get('TERM') or kitty_opts().term,
|
||||
'COLORTERM': 'truecolor',
|
||||
}
|
||||
env.update(literal_env)
|
||||
env.update(ssh_opts.env)
|
||||
for q in ('KITTY_WINDOW_ID', 'WINDOWID'):
|
||||
val = os.environ.get(q)
|
||||
@ -202,7 +212,7 @@ def make_tarfile(ssh_opts: SSHOptions, base_env: Dict[str, str], compression: st
|
||||
env['KITTY_LOGIN_CWD'] = ssh_opts.cwd
|
||||
if ssh_opts.remote_kitty != 'no':
|
||||
env['KITTY_REMOTE'] = ssh_opts.remote_kitty
|
||||
env_script = serialize_env(env, base_env, for_python=compression != 'gz')
|
||||
env_script = serialize_env(literal_env, env, base_env, for_python=compression != 'gz')
|
||||
buf = io.BytesIO()
|
||||
with tarfile.open(mode=f'w:{compression}', fileobj=buf, encoding='utf-8') as tf:
|
||||
rd = ssh_opts.remote_dir.rstrip('/')
|
||||
|
||||
@ -143,7 +143,7 @@ copy --exclude */w.* d1
|
||||
self.ae(len(glob.glob(f'{remote_home}/{tname}/*/xterm-kitty')), 2)
|
||||
|
||||
def test_ssh_env_vars(self):
|
||||
tset = '$A-$(echo no)-`echo no2` "something"'
|
||||
tset = '$A-$(echo no)-`echo no2` "something\nelse"'
|
||||
for sh in self.all_possible_sh:
|
||||
with self.subTest(sh=sh), tempfile.TemporaryDirectory() as tdir:
|
||||
os.mkdir(os.path.join(tdir, 'cwd'))
|
||||
|
||||
@ -93,8 +93,9 @@ def apply_env_vars(raw):
|
||||
if len(parts) == 1:
|
||||
key, val = parts[0], ''
|
||||
else:
|
||||
key, val = parts
|
||||
val = os.path.expandvars(val)
|
||||
key, val, literal_quote = parts
|
||||
if not literal_quote:
|
||||
val = os.path.expandvars(val)
|
||||
os.environ[key] = val
|
||||
|
||||
for line in raw.splitlines():
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user