From 3281919056c442e4ba6f5abed7ee2c5130de31ff Mon Sep 17 00:00:00 2001 From: Kovid Goyal Date: Tue, 1 Sep 2026 20:14:03 +0530 Subject: [PATCH] ssh kitten: Verify owner and permissions of SHM memory used for askpass --- docs/changelog.rst | 2 ++ kittens/ssh/utils.py | 7 +------ kitty/shm.py | 10 ++++++++++ kitty/window.py | 47 +++++++++++++++++++++++++++++++++----------- kitty_tests/shm.py | 11 +++++++++++ 5 files changed, 59 insertions(+), 18 deletions(-) diff --git a/docs/changelog.rst b/docs/changelog.rst index 6e70b3654..8d059689f 100644 --- a/docs/changelog.rst +++ b/docs/changelog.rst @@ -281,6 +281,8 @@ Detailed list of changes - Text sizing protocol: Fix a buffer overflow when a natural width text sizing escape code contains a grapheme cluster longer than four codepoints (:cve:`2026-80431`) +- ssh kitten askpass: Verify owner and permissions of SHM memory used for askpass + - Graphics protocol: Fix a crash when transmitting image data via a file or shared memory object (``t=f``, ``t=t`` or ``t=s``) and the client truncates it while kitty is reading from it. diff --git a/kittens/ssh/utils.py b/kittens/ssh/utils.py index b3f6a9901..88595b3bc 100644 --- a/kittens/ssh/utils.py +++ b/kittens/ssh/utils.py @@ -159,17 +159,12 @@ def create_shared_memory(data: Any, prefix: str) -> str: def read_data_from_shared_memory(shm_name: str) -> Any: import json - import stat from kitty.shm import SharedMemory with SharedMemory(shm_name, readonly=True) as shm: shm.unlink() - if shm.stats.st_uid != os.geteuid() or shm.stats.st_gid != os.getegid(): - raise ValueError(f'Incorrect owner on pwfile: uid={shm.stats.st_uid} gid={shm.stats.st_gid}') - mode = stat.S_IMODE(shm.stats.st_mode) - if mode != stat.S_IREAD | stat.S_IWRITE: - raise ValueError(f'Incorrect permissions on pwfile: 0o{mode:03o}') + shm.verify_owner_and_mode() return json.loads(shm.read_data_with_size()) diff --git a/kitty/shm.py b/kitty/shm.py index 6cafb67ae..a4fb2ad1c 100644 --- a/kitty/shm.py +++ b/kitty/shm.py @@ -160,6 +160,16 @@ class SharedMemory: def fileno(self) -> int: return self._fd + def verify_owner_and_mode(self) -> None: + """Ensure the shared memory object is owned by us and not accessible to + any other user. Must be called when opening objects whose name comes + from an untrusted source, as anyone can create an object of that name.""" + if self.stats.st_uid != os.geteuid() or self.stats.st_gid != os.getegid(): + raise ValueError(f'Incorrect owner on shared memory object: uid={self.stats.st_uid} gid={self.stats.st_gid}') + mode = stat.S_IMODE(self.stats.st_mode) + if mode != stat.S_IREAD | stat.S_IWRITE: + raise ValueError(f'Incorrect permissions on shared memory object: 0o{mode:03o}') + def __repr__(self) -> str: return f'{self.__class__.__name__}({self.name!r}, size={self.size})' diff --git a/kitty/window.py b/kitty/window.py index 4a0ccd29b..884b1ae5c 100644 --- a/kitty/window.py +++ b/kitty/window.py @@ -1808,23 +1808,40 @@ class Window: def handle_remote_askpass(self, msgb: memoryview) -> None: from .shm import SharedMemory - msg = str(msgb, 'utf-8') - with SharedMemory(name=msg, readonly=True) as shm: + name = str(msgb, 'utf-8') + try: + shm = SharedMemory(name=name) + except OSError as err: + log_error(f'Ignoring ask request as opening its shared memory object failed with error: {err}') + return + try: + # The askpass client polls its own mapping for the answer, so + # unlinking the name here does not affect it and prevents anyone + # else from opening this object while we are using it. + shm.unlink() + shm.verify_owner_and_mode() shm.seek(1) data = json.loads(shm.read_data_with_size()) + q = str(data['type']) + message: str = sanitize_control_codes(data['message']) + choices = tuple(map(sanitize_control_codes, data['choices'])) if q == 'choose' else () + except Exception as err: + shm.close() + log_error(f'Ignoring invalid ask request with error: {err}') + return def callback(ans: Any) -> None: - data = json.dumps(ans) - with SharedMemory(name=msg) as shm: + try: shm.seek(1) - shm.write_data_with_size(data) + shm.write_data_with_size(json.dumps(ans)) shm.flush() shm.seek(0) shm.write(b'\x01') + finally: + shm.close() - message: str = data['message'] window_title = 'A program wants your input' - if data['type'] == 'confirm': + if q == 'confirm': get_boss().confirm( message, callback, @@ -1833,16 +1850,22 @@ class Window: confirm_on_accept=bool(data.get('confirm_on_accept', True)), title=window_title, ) - elif data['type'] == 'choose': - get_boss().choose(message, callback, *data['choices'], window=self, default=data.get('default', ''), title=window_title) - elif data['type'] == 'get_line': + elif q == 'choose': + get_boss().choose(message, callback, *choices, window=self, default=data.get('default', ''), title=window_title) + elif q == 'get_line': which = 'password' if data.get('is_password') else 'input' message = f'\x1b[33mA program running in this window is asking for your {which}\x1b[m\n\n{message}' get_boss().get_line( - message, callback, window=self, is_password=bool(data.get('is_password')), prompt=data.get('prompt', '> '), window_title=window_title + message, + callback, + window=self, + is_password=bool(data.get('is_password')), + prompt=sanitize_control_codes(data.get('prompt', '> ')), + window_title=window_title, ) else: - log_error(f'Ignoring ask request with unknown type: {data["type"]}') + shm.close() + log_error(f'Ignoring ask request with unknown type: {q}') def handle_remote_print(self, msg: memoryview) -> None: text = process_remote_print(msg) diff --git a/kitty_tests/shm.py b/kitty_tests/shm.py index 88cec0e55..046720b80 100644 --- a/kitty_tests/shm.py +++ b/kitty_tests/shm.py @@ -27,3 +27,14 @@ class SHMTest(BaseTest): with SharedMemory(name=name, unlink_on_exit=True) as shm: q = shm.read_data_with_size() self.assertEqual(data, q) + + def test_shm_ownership_verification(self): + with SharedMemory(size=64, unlink_on_exit=True) as shm: + shm.verify_owner_and_mode() + for mode in (0o644, 0o400, 0o660): + os.fchmod(shm.fileno(), mode) + with SharedMemory(name=shm.name, readonly=True) as shm2: + self.assertRaises(ValueError, shm2.verify_owner_and_mode) + os.fchmod(shm.fileno(), 0o600) + with SharedMemory(name=shm.name, readonly=True) as shm2: + shm2.verify_owner_and_mode()