ssh kitten: Verify owner and permissions of SHM memory used for askpass

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.

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())
 
 

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})'
 

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)

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()