fix(tmux): explicit transport-only dispatch and safe remote quoting (#1496)
This commit is contained in:
@@ -0,0 +1,39 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Execute the SSH command through a shell; no network or real tmux access."""
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
import subprocess
|
||||
import tempfile
|
||||
|
||||
here = Path(__file__).resolve().parent
|
||||
with tempfile.TemporaryDirectory(prefix="tmux-remote-contract-") as temp:
|
||||
root = Path(temp)
|
||||
bindir = root / "bin"
|
||||
bindir.mkdir()
|
||||
(bindir / "ssh").write_text("#!/usr/bin/env python3\nimport os,subprocess,sys\nopen(os.environ['SSH_CALLS'],'a').write('ssh\\n')\nraise SystemExit(subprocess.call(['/bin/sh','-c',sys.argv[-1]]))\n")
|
||||
(bindir / "tmux").write_text("#!/usr/bin/env python3\nimport json,os,sys\na=sys.argv[1:]\nwith open(os.environ['CALLS'],'a') as f: f.write(json.dumps(a)+'\\n')\nif a and a[0]=='-L': a=a[2:]\nif a[0]=='display-message': print('%7')\nif a[0]=='load-buffer': open(os.environ['PAYLOAD'],'wb').write(sys.stdin.buffer.read())\n")
|
||||
for path in bindir.iterdir():
|
||||
path.chmod(0o755)
|
||||
env = dict(os.environ, PATH=str(bindir) + ':' + os.environ['PATH'], CALLS=str(root/'calls'), SSH_CALLS=str(root/'ssh-calls'), PAYLOAD=str(root/'payload'))
|
||||
env.pop('AGENT_SEND_SENDER', None)
|
||||
attack = "x'; printf 'UNAUTHORIZED_EXTRA_EFFECT\\n'; #"
|
||||
cases = [('target', ['-s', attack], 0), ('socket', ['-s', 'fixture', '-L', attack], 0), ('retry', ['-s', 'fixture', '-r', attack], 3)]
|
||||
for label, args, expected in cases:
|
||||
(root/'calls').write_text('')
|
||||
(root/'ssh-calls').write_text('')
|
||||
result = subprocess.run(['bash', str(here/'agent-send.sh'), '-S', 'review:src', '-n', 'fake', '-H', 'fake', '-m', 'safe\n你好'] + args, env=env, text=True, capture_output=True)
|
||||
assert result.returncode == expected, (label, result)
|
||||
assert 'UNAUTHORIZED_EXTRA_EFFECT' not in result.stdout, (label, result.stdout)
|
||||
calls = [json.loads(line) for line in (root/'calls').read_text().splitlines()]
|
||||
if expected == 3:
|
||||
assert not calls and not (root/'ssh-calls').read_text()
|
||||
else:
|
||||
assert result.stdout.strip() == 'transport dispatched; application acceptance unknown'
|
||||
command_calls = [a[2:] if a[0] == '-L' else a for a in calls]
|
||||
assert sum(a[0] == 'paste-buffer' for a in command_calls) == 1
|
||||
assert sum(a[0] == 'send-keys' for a in command_calls) == 1
|
||||
assert b'safe\n' in (root/'payload').read_bytes()
|
||||
if label == 'socket': assert all(a[:2] == ['-L', attack] for a in calls)
|
||||
if label == 'target': assert command_calls[0][3] == '=' + attack + ':0.0'
|
||||
print('PASS remote', label)
|
||||
Reference in New Issue
Block a user