diff --git a/docs/changelog.rst b/docs/changelog.rst index dac6061e5..b511d9f02 100644 --- a/docs/changelog.rst +++ b/docs/changelog.rst @@ -319,6 +319,8 @@ Detailed list of changes - Graphics protocol: Make fileopen failure responses generic to avoid leaking any information about failed files +- edit-in-kitty: harden the code used to parse messages from clients + - Clipboard protocol: Report an ``EFBIG`` error to programs that try to write more data to the clipboard than allowed by :opt:`clipboard_max_size`, instead of silently truncating their data. Also fix :opt:`clipboard_max_size` being diff --git a/kitty/launch.py b/kitty/launch.py index a8f7a0f51..699a3dd1e 100644 --- a/kitty/launch.py +++ b/kitty/launch.py @@ -985,36 +985,102 @@ def parse_message(msg: str, simple: Container[str]) -> Iterator[tuple[str, str]] yield k, v +AbortSignal = Literal['closed', 'replaced', 'disconnected', 'interrupt', ''] + + +class EditRequest(NamedTuple): + args: list[str] + cwd: str + file_inode: tuple[int, int] + file_size: int + file_data: bytes + version: int + abort_signaled: AbortSignal + + +def parse_edit_message(msg: str) -> EditRequest: + # Only keys that are part of the edit protocol are recognized, all others + # are ignored. In particular, values from the message must never be used to + # set attributes on the EditCmd object directly, as that would allow any + # program that can write to the terminal to overwrite internal state, such + # as the temporary directory that is deleted once editing is finished. + args: list[str] = [] + cwd = '' + file_inode = -1, -1 + file_size = -1 + file_data = b'' + version = 0 + abort_signaled: AbortSignal = '' + simple = 'file_inode', 'file_data', 'abort_signaled', 'version' + for k, v in parse_message(msg, simple): + if k == 'file_inode': + parts = v.split(':') + if len(parts) > 2: + with suppress(ValueError): + dev, inode, sz = int(parts[0]), int(parts[1]), int(parts[2]) + file_inode, file_size = (dev, inode), sz + elif k == 'a': + args.append(v) + elif k == 'file_data': + import base64 + + file_data = base64.standard_b64decode(v) + elif k == 'version': + version = int(v) + elif k == 'cwd': + cwd = v + elif k == 'abort_signaled': + # interrupt is the only abort signal clients can send, the other + # values of abort_signaled are used for internal state only. + abort_signaled = 'interrupt' if v else '' + return EditRequest(args, cwd, file_inode, file_size, file_data, version, abort_signaled) + + class EditCmd: - abort_signaled: Literal['closed', 'replaced', 'disconnected', ''] = '' + # Using __slots__ means an attribute that is not listed here cannot be + # created on this object, not even accidentally, which keeps internal state + # such as tdir safe from data received over the edit protocol. + __slots__ = ( + 'abort_signaled', + 'args', + 'child_is_remote', + 'cwd', + 'editor_exit_code', + 'editor_window_id', + 'file_data', + 'file_inode', + 'file_localpath', + 'file_name', + 'file_size', + 'file_spec', + 'is_local_file', + 'last_mod_time', + 'line_number', + 'opts', + 'source_window_id', + 'tdir', + 'version', + ) + abort_signaled: AbortSignal + opts: LaunchCLIOptions def __init__(self, msg: str, child_is_remote: bool) -> None: + # set up the internal state that __del__ needs first, so that it works + # even if parsing the message below fails self.child_is_remote = child_is_remote self.tdir = '' - self.args: list[str] = [] - self.cwd = self.file_name = self.file_localpath = '' - self.file_data = b'' - self.file_inode = -1, -1 - self.file_size = -1 - self.version = 0 + self.abort_signaled = '' + self.file_name = self.file_localpath = '' self.source_window_id = self.editor_window_id = -1 self.editor_exit_code: int | None = None - simple = 'file_inode', 'file_data', 'abort_signaled', 'version' - for k, v in parse_message(msg, simple): - if k == 'file_inode': - q = map(int, v.split(':')) - self.file_inode = next(q), next(q) - self.file_size = next(q) - elif k == 'a': - self.args.append(v) - elif k == 'file_data': - import base64 - - self.file_data = base64.standard_b64decode(v) - elif k == 'version': - self.version = int(v) - else: - setattr(self, k, v) + req = parse_edit_message(msg) + self.args = req.args + self.cwd = req.cwd + self.file_inode = req.file_inode + self.file_size = req.file_size + self.file_data = req.file_data + self.version = req.version + self.abort_signaled = req.abort_signaled if self.abort_signaled: return if self.version > 0: @@ -1048,6 +1114,8 @@ class EditCmd: self.opts.cwd = os.path.dirname(self.file_localpath) def __del__(self) -> None: + # tdir is only ever set to a directory created by us in __init__, never + # to a value coming from the edit protocol message. if self.tdir: if not self.abort_signaled == 'disconnected': with suppress(OSError): diff --git a/kitty_tests/edit_in_kitty.py b/kitty_tests/edit_in_kitty.py new file mode 100644 index 000000000..1450bf93e --- /dev/null +++ b/kitty_tests/edit_in_kitty.py @@ -0,0 +1,110 @@ +#!/usr/bin/env python +# License: GPLv3 Copyright: 2026, Kovid Goyal + +import gc +import os +import tempfile +from base64 import standard_b64encode + +from kitty.launch import EditCmd, parse_edit_message + +from .base import BaseTest + + +def encode_msg(*fields: tuple[str, str], simple: frozenset[str] = frozenset({'file_inode', 'file_data', 'abort_signaled', 'version'})) -> str: + parts = [] + for k, v in fields: + if k not in simple: + v = standard_b64encode(v.encode('utf-8')).decode('ascii') + parts.append(f'{k}={v}') + return ','.join(parts) + + +class TestEditInKitty(BaseTest): + def setUp(self): + super().setUp() + self.set_options() + self.tdir = tempfile.mkdtemp() + + def tearDown(self): + self.rmtree_ignoring_errors(self.tdir) + super().tearDown() + + def test_edit_message_parsing(self): + path = os.path.join(self.tdir, 'file.txt') + with open(path, 'w') as f: + f.write('local') + st = os.stat(path) + msg = encode_msg( + ('cwd', self.tdir), + ('a', '--type=tab'), + ('a', '+17'), + ('a', 'file.txt'), + ('file_inode', f'{st.st_dev}:{st.st_ino}:{st.st_mtime_ns}'), + ('file_data', standard_b64encode(b'local').decode('ascii')), + ) + req = parse_edit_message(msg) + self.ae(req.args, ['--type=tab', '+17', 'file.txt']) + self.ae(req.cwd, self.tdir) + self.ae(req.file_inode, (st.st_dev, st.st_ino)) + self.ae(req.file_data, b'local') + self.ae(req.abort_signaled, '') + + c = EditCmd(msg, False) + self.ae(c.line_number, 17) + self.ae(c.file_name, 'file.txt') + self.ae(c.file_localpath, path) + self.assertTrue(c.is_local_file) + # no temporary directory is needed for a local file + self.ae(c.tdir, '') + + def test_edit_of_non_local_file_uses_own_tdir(self): + msg = encode_msg( + ('cwd', self.tdir), + ('a', 'no-such-file.txt'), + ('file_inode', '1:2:3'), + ('file_data', standard_b64encode(b'remote data').decode('ascii')), + ) + c = EditCmd(msg, True) + self.assertFalse(c.is_local_file) + self.assertTrue(c.tdir) + self.assertNotEqual(os.path.realpath(c.tdir), os.path.realpath(self.tdir)) + self.ae(c.file_localpath, os.path.join(c.tdir, 'no-such-file.txt')) + with open(c.file_localpath) as f: + self.ae(f.read(), 'remote data') + tdir = c.tdir + del c + gc.collect() + self.assertFalse(os.path.exists(tdir), 'The temporary directory created for the edit was not deleted') + + def test_edit_message_cannot_overwrite_internal_state(self): + # A program writing to the terminal must not be able to point tdir at a + # directory of its choosing, which would then be deleted recursively + # when the EditCmd object is destroyed. + victim = os.path.join(self.tdir, 'victim') + os.mkdir(victim) + with open(os.path.join(victim, 'data'), 'w') as f: + f.write('precious') + msg = encode_msg(('tdir', victim), ('abort_signaled', 'closed'), ('child_is_remote', 'yes'), ('is_local_file', 'yes')) + c = EditCmd(msg, False) + self.ae(c.tdir, '') + self.ae(c.abort_signaled, 'interrupt') + self.ae(c.child_is_remote, False) + del c + gc.collect() + self.assertTrue(os.path.exists(os.path.join(victim, 'data')), 'The edit protocol message deleted a directory of its choosing') + + def test_unknown_edit_message_keys_are_ignored(self): + msg = encode_msg(('cwd', self.tdir), ('a', 'x.txt'), ('some_unknown_key', 'value'), ('file_inode', '1:2:3')) + c = EditCmd(msg, False) + self.assertFalse(hasattr(c, 'some_unknown_key')) + # __slots__ prevents even a deliberate attempt at adding state + with self.assertRaises(AttributeError): + c.some_unknown_key = 'value' + + def test_malformed_edit_messages(self): + for inode in ('', 'a:b:c', '1:2', '1:2:x'): + req = parse_edit_message(encode_msg(('file_inode', inode))) + self.ae(req.file_inode, (-1, -1)) + self.ae(req.file_size, -1) + self.assertRaises(ValueError, EditCmd, encode_msg(('version', '1'), ('a', 'x.txt')), False)