Implement DnD protocol: t=s URI file transfer and t=d directory listing with tests

Agent-Logs-Url: https://github.com/kovidgoyal/kitty/sessions/a5aee38b-1746-4699-9dcc-2138f43cabe8

Co-authored-by: kovidgoyal <1308621+kovidgoyal@users.noreply.github.com>

kitty_tests/dnd.py

@@ -58,6 +58,28 @@ def client_request_data(mime: str = '', client_id: int = 0) -> bytes:
     return _osc(f'{meta};{mime}')
 
 
+def client_request_uri_data(idx: int, client_id: int = 0) -> bytes:
+    """Escape code a client sends to request a file from the URI list (t=s ; text/uri-list:idx)."""
+    meta = f'{DND_CODE};t=s'
+    if client_id:
+        meta += f':i={client_id}'
+    return _osc(f'{meta};text/uri-list:{idx}')
+
+
+def client_dir_read(handle_id: int, entry_num: int | None = None, client_id: int = 0) -> bytes:
+    """Escape code for a directory request (t=d:x=handle_id[:y=entry_num]).
+
+    * entry_num=None → close the directory handle.
+    * entry_num>=1   → read that entry (1-based).
+    """
+    meta = f'{DND_CODE};t=d:x={handle_id}'
+    if entry_num is not None:
+        meta += f':y={entry_num}'
+    if client_id:
+        meta += f':i={client_id}'
+    return _osc(meta)
+
+
 # ---- escape-code decoder used by assertions ---------------------------------
 
 _OSC_RE = re.compile(
@@ -474,3 +496,341 @@ class TestDnDProtocol(BaseTest):
             # Only the end signal should be present.
             self.assertEqual(len(r_events), 1, raw)
             self.ae(r_events[0]['payload'], b'')
+
+    # ---- t=s / t=d (remote file/directory transfer) tests ----------------
+
+    def _setup_uri_drop(self, screen, wid, cap, uri_list_data: bytes, mimes=None):
+        """Register, drop, deliver text/uri-list data, discard move/drop events."""
+        if mimes is None:
+            mimes = ['text/plain', 'text/uri-list']
+        parse_bytes(screen, client_register('text/plain text/uri-list'))
+        dnd_test_set_mouse_pos(wid, 0, 0, 0, 0)
+        dnd_test_fake_drop_event(wid, True, mimes)
+        cap.consume(wid)
+        # Client requests and receives the URI list
+        parse_bytes(screen, client_request_data('text/uri-list'))
+        dnd_test_fake_drop_data(wid, 'text/uri-list', uri_list_data)
+        cap.consume(wid)  # discard t=r data for text/uri-list
+
+    def test_uri_file_transfer_basic(self) -> None:
+        """t=s request sends the content of a regular file as t=r chunks."""
+        import os, tempfile
+        content = b'Hello, remote DnD world!\n' * 100
+        with tempfile.NamedTemporaryFile(delete=False) as f:
+            f.write(content)
+            fpath = f.name
+        try:
+            uri_list = f'file://{fpath}\r\n'.encode()
+            with dnd_test_window() as (osw, wid, screen, cap):
+                self._setup_uri_drop(screen, wid, cap, uri_list)
+                parse_bytes(screen, client_request_uri_data(0))
+                raw = cap.consume(wid)
+                events = parse_escape_codes_b64(raw)
+                r_events = [e for e in events if e['type'] == 'r']
+                self.assertTrue(r_events, 'no t=r events')
+                combined = b''.join(e['payload'] for e in r_events)
+                self.ae(combined, content)
+                # Last chunk must be the empty end-of-data signal
+                self.ae(r_events[-1]['payload'], b'')
+        finally:
+            os.unlink(fpath)
+
+    def test_uri_file_transfer_integrity(self) -> None:
+        """File content is transferred byte-for-byte (binary integrity)."""
+        import os, tempfile
+        # Use binary content with all byte values to check integrity
+        content = bytes(range(256)) * 512  # 128 KiB
+        with tempfile.NamedTemporaryFile(delete=False) as f:
+            f.write(content)
+            fpath = f.name
+        try:
+            uri_list = f'file://{fpath}\r\n'.encode()
+            with dnd_test_window() as (osw, wid, screen, cap):
+                self._setup_uri_drop(screen, wid, cap, uri_list)
+                parse_bytes(screen, client_request_uri_data(0))
+                raw = cap.consume(wid)
+                events = parse_escape_codes_b64(raw)
+                r_events = [e for e in events if e['type'] == 'r']
+                # Exclude the empty end-of-data entry when concatenating
+                combined = b''.join(e['payload'] for e in r_events if e['payload'])
+                self.ae(combined, content)
+        finally:
+            os.unlink(fpath)
+
+    def test_uri_file_transfer_enoent(self) -> None:
+        """t=s with an out-of-range index returns ENOENT."""
+        uri_list = b'file:///tmp/no_such_file_exists_dnd_test_xyz\r\n'
+        with dnd_test_window() as (osw, wid, screen, cap):
+            self._setup_uri_drop(screen, wid, cap, uri_list)
+            # Index 0 refers to a non-existent file
+            parse_bytes(screen, client_request_uri_data(0))
+            events = self._get_events(cap, wid)
+            self.assertEqual(len(events), 1, events)
+            self.ae(events[0]['type'], 'R')
+            self.assertIn(events[0]['payload'].strip(), [b'ENOENT', b'EPERM'])
+
+    def test_uri_file_transfer_out_of_bounds(self) -> None:
+        """t=s with an index beyond the URI list returns ENOENT."""
+        import os, tempfile
+        with tempfile.NamedTemporaryFile(delete=False) as f:
+            fpath = f.name
+        try:
+            uri_list = f'file://{fpath}\r\n'.encode()
+            with dnd_test_window() as (osw, wid, screen, cap):
+                self._setup_uri_drop(screen, wid, cap, uri_list)
+                parse_bytes(screen, client_request_uri_data(99))  # out of range
+                events = self._get_events(cap, wid)
+                self.assertEqual(len(events), 1, events)
+                self.ae(events[0]['type'], 'R')
+                self.ae(events[0]['payload'].strip(), b'ENOENT')
+        finally:
+            os.unlink(fpath)
+
+    def test_uri_request_without_uri_list_returns_einval(self) -> None:
+        """t=s without prior text/uri-list request returns EINVAL."""
+        import os, tempfile
+        with tempfile.NamedTemporaryFile(delete=False) as f:
+            fpath = f.name
+        try:
+            with dnd_test_window() as (osw, wid, screen, cap):
+                parse_bytes(screen, client_register('text/plain'))
+                dnd_test_set_mouse_pos(wid, 0, 0, 0, 0)
+                dnd_test_fake_drop_event(wid, True, ['text/plain', 'text/uri-list'])
+                cap.consume(wid)
+                # Do NOT request text/uri-list first
+                parse_bytes(screen, client_request_uri_data(0))
+                events = self._get_events(cap, wid)
+                self.assertEqual(len(events), 1, events)
+                self.ae(events[0]['type'], 'R')
+                self.ae(events[0]['payload'].strip(), b'EINVAL')
+        finally:
+            os.unlink(fpath)
+
+    def test_uri_non_regular_file_returns_einval(self) -> None:
+        """t=s for a non-regular file (e.g. /dev/null) returns EINVAL."""
+        uri_list = b'file:///dev/null\r\n'
+        with dnd_test_window() as (osw, wid, screen, cap):
+            self._setup_uri_drop(screen, wid, cap, uri_list)
+            parse_bytes(screen, client_request_uri_data(0))
+            events = self._get_events(cap, wid)
+            self.assertEqual(len(events), 1, events)
+            self.ae(events[0]['type'], 'R')
+            self.ae(events[0]['payload'].strip(), b'EINVAL')
+
+    def test_uri_directory_transfer_tree(self) -> None:
+        """Full directory tree transfer: listing, sub-dirs, file integrity."""
+        import os, tempfile, hashlib
+
+        # Build a tree: root/ a.txt  b/  b/c.txt  b/d/  b/d/e.txt
+        with tempfile.TemporaryDirectory() as root:
+            a_content = b'file a content\n' * 50
+            bc_content = bytes(range(256)) * 20  # binary data
+            bde_content = b'deep nested file\n'
+
+            (open(os.path.join(root, 'a.txt'), 'wb')).write(a_content)
+            os.makedirs(os.path.join(root, 'b', 'd'))
+            (open(os.path.join(root, 'b', 'c.txt'), 'wb')).write(bc_content)
+            (open(os.path.join(root, 'b', 'd', 'e.txt'), 'wb')).write(bde_content)
+
+            uri_list = f'file://{root}\r\n'.encode()
+            with dnd_test_window() as (osw, wid, screen, cap):
+                self._setup_uri_drop(screen, wid, cap, uri_list)
+
+                # Request the root directory (idx=0)
+                parse_bytes(screen, client_request_uri_data(0))
+                raw = cap.consume(wid)
+                events = parse_escape_codes_b64(raw)
+                d_events = [e for e in events if e['type'] == 'd']
+                self.assertTrue(d_events, 'expected t=d listing for root')
+
+                root_listing_payload = b''.join(
+                    chunk for e in d_events for chunk in e['chunks'] if chunk
+                )
+                root_handle_id = int(d_events[0]['meta']['x'])
+                self.assertGreater(root_handle_id, 0)
+
+                # Decode null-separated entries
+                root_entries = [e for e in root_listing_payload.split(b'\x00') if e]
+                # First entry is the unique identifier; remainder are file/dir names
+                self.assertGreater(len(root_entries), 1,
+                                   f'expected entries, got {root_entries}')
+                entry_names = {e.decode() for e in root_entries[1:]}
+                self.assertIn('a.txt', entry_names)
+                self.assertIn('b', entry_names)
+
+                # Find index of 'a.txt' in the entries list (1-based for t=d:y=)
+                entries_list = [e.decode() for e in root_entries[1:]]
+                a_idx = entries_list.index('a.txt') + 1
+                b_idx = entries_list.index('b') + 1
+
+                # Read a.txt
+                parse_bytes(screen, client_dir_read(root_handle_id, a_idx))
+                raw = cap.consume(wid)
+                events = parse_escape_codes_b64(raw)
+                r_events = [e for e in events if e['type'] == 'r']
+                a_data = b''.join(e['payload'] for e in r_events if e['payload'])
+                self.ae(a_data, a_content)
+
+                # Read sub-directory b → should get a new t=d listing
+                parse_bytes(screen, client_dir_read(root_handle_id, b_idx))
+                raw = cap.consume(wid)
+                events = parse_escape_codes_b64(raw)
+                b_d_events = [e for e in events if e['type'] == 'd']
+                self.assertTrue(b_d_events, 'expected t=d listing for b/')
+
+                b_listing_payload = b''.join(
+                    chunk for e in b_d_events for chunk in e['chunks'] if chunk
+                )
+                b_handle_id = int(b_d_events[0]['meta']['x'])
+                self.assertNotEqual(b_handle_id, root_handle_id)
+
+                b_entries = [e for e in b_listing_payload.split(b'\x00') if e]
+                b_names = {e.decode() for e in b_entries[1:]}
+                self.assertIn('c.txt', b_names)
+                self.assertIn('d', b_names)
+
+                b_entries_list = [e.decode() for e in b_entries[1:]]
+                bc_idx = b_entries_list.index('c.txt') + 1
+                bd_idx = b_entries_list.index('d') + 1
+
+                # Read b/c.txt (binary integrity)
+                parse_bytes(screen, client_dir_read(b_handle_id, bc_idx))
+                raw = cap.consume(wid)
+                events = parse_escape_codes_b64(raw)
+                r_events = [e for e in events if e['type'] == 'r']
+                bc_data = b''.join(e['payload'] for e in r_events if e['payload'])
+                self.ae(bc_data, bc_content)
+                # Check SHA-256 integrity
+                self.ae(hashlib.sha256(bc_data).digest(),
+                        hashlib.sha256(bc_content).digest())
+
+                # Read sub-directory b/d → yet another t=d listing
+                parse_bytes(screen, client_dir_read(b_handle_id, bd_idx))
+                raw = cap.consume(wid)
+                events = parse_escape_codes_b64(raw)
+                bd_d_events = [e for e in events if e['type'] == 'd']
+                self.assertTrue(bd_d_events, 'expected t=d listing for b/d/')
+
+                bd_listing_payload = b''.join(
+                    chunk for e in bd_d_events for chunk in e['chunks'] if chunk
+                )
+                bd_handle_id = int(bd_d_events[0]['meta']['x'])
+                bd_entries = [e for e in bd_listing_payload.split(b'\x00') if e]
+                bd_names = {e.decode() for e in bd_entries[1:]}
+                self.assertIn('e.txt', bd_names)
+
+                bd_entries_list = [e.decode() for e in bd_entries[1:]]
+                bde_idx = bd_entries_list.index('e.txt') + 1
+
+                # Read b/d/e.txt
+                parse_bytes(screen, client_dir_read(bd_handle_id, bde_idx))
+                raw = cap.consume(wid)
+                events = parse_escape_codes_b64(raw)
+                r_events = [e for e in events if e['type'] == 'r']
+                bde_data = b''.join(e['payload'] for e in r_events if e['payload'])
+                self.ae(bde_data, bde_content)
+
+                # Close all directory handles
+                parse_bytes(screen, client_dir_read(bd_handle_id))
+                parse_bytes(screen, client_dir_read(b_handle_id))
+                parse_bytes(screen, client_dir_read(root_handle_id))
+                # No error output expected from close operations
+                self._assert_no_output(cap, wid)
+
+    def test_dir_handle_close_and_reuse(self) -> None:
+        """Closing a directory handle invalidates it; subsequent requests return EINVAL."""
+        import os, tempfile
+        with tempfile.TemporaryDirectory() as root:
+            open(os.path.join(root, 'f.txt'), 'w').close()
+            uri_list = f'file://{root}\r\n'.encode()
+            with dnd_test_window() as (osw, wid, screen, cap):
+                self._setup_uri_drop(screen, wid, cap, uri_list)
+                parse_bytes(screen, client_request_uri_data(0))
+                raw = cap.consume(wid)
+                events = parse_escape_codes_b64(raw)
+                d_ev = [e for e in events if e['type'] == 'd']
+                self.assertTrue(d_ev)
+                hid = int(d_ev[0]['meta']['x'])
+
+                # Close the handle
+                parse_bytes(screen, client_dir_read(hid))
+                self._assert_no_output(cap, wid)
+
+                # Now try to read from the closed handle → EINVAL
+                parse_bytes(screen, client_dir_read(hid, 1))
+                events = self._get_events(cap, wid)
+                self.assertEqual(len(events), 1)
+                self.ae(events[0]['type'], 'R')
+                self.ae(events[0]['payload'].strip(), b'EINVAL')
+
+    def test_dir_entry_out_of_bounds_returns_enoent(self) -> None:
+        """Reading a directory entry with an out-of-range index returns ENOENT."""
+        import os, tempfile
+        with tempfile.TemporaryDirectory() as root:
+            open(os.path.join(root, 'only.txt'), 'w').close()
+            uri_list = f'file://{root}\r\n'.encode()
+            with dnd_test_window() as (osw, wid, screen, cap):
+                self._setup_uri_drop(screen, wid, cap, uri_list)
+                parse_bytes(screen, client_request_uri_data(0))
+                raw = cap.consume(wid)
+                events = parse_escape_codes_b64(raw)
+                d_ev = [e for e in events if e['type'] == 'd']
+                hid = int(d_ev[0]['meta']['x'])
+
+                # Entry 999 does not exist
+                parse_bytes(screen, client_dir_read(hid, 999))
+                events = self._get_events(cap, wid)
+                self.assertEqual(len(events), 1)
+                self.ae(events[0]['type'], 'R')
+                self.ae(events[0]['payload'].strip(), b'ENOENT')
+
+    def test_dir_unique_identifier_prevents_loops(self) -> None:
+        """Each directory listing starts with a unique id (dev:inode format)."""
+        import os, tempfile
+        with tempfile.TemporaryDirectory() as root:
+            sub = os.path.join(root, 'sub')
+            os.mkdir(sub)
+            uri_list = f'file://{root}\r\n'.encode()
+            with dnd_test_window() as (osw, wid, screen, cap):
+                self._setup_uri_drop(screen, wid, cap, uri_list)
+                parse_bytes(screen, client_request_uri_data(0))
+                raw = cap.consume(wid)
+                events = parse_escape_codes_b64(raw)
+                d_ev = [e for e in events if e['type'] == 'd']
+                root_payload = b''.join(
+                    chunk for e in d_ev for chunk in e['chunks'] if chunk
+                )
+                root_handle_id = int(d_ev[0]['meta']['x'])
+                root_uid = root_payload.split(b'\x00')[0].decode()
+                # uid must be non-empty and contain a colon (dev:inode)
+                self.assertIn(':', root_uid, f'uid={root_uid!r}')
+
+                # Get the sub directory listing to compare identifiers
+                entries = [e.decode() for e in root_payload.split(b'\x00')[1:] if e]
+                sub_idx = entries.index('sub') + 1
+                parse_bytes(screen, client_dir_read(root_handle_id, sub_idx))
+                raw = cap.consume(wid)
+                events = parse_escape_codes_b64(raw)
+                d_ev2 = [e for e in events if e['type'] == 'd']
+                sub_payload = b''.join(
+                    chunk for e in d_ev2 for chunk in e['chunks'] if chunk
+                )
+                sub_uid = sub_payload.split(b'\x00')[0].decode() if sub_payload else ''
+                self.assertIn(':', sub_uid, f'sub uid={sub_uid!r}')
+                # Root and sub must have different identifiers
+                self.assertNotEqual(root_uid, sub_uid)
+
+    def test_window_close_during_transfer_no_leak(self) -> None:
+        """Closing the window while dir handles are open frees all resources (no crash)."""
+        import os, tempfile
+        with tempfile.TemporaryDirectory() as root:
+            open(os.path.join(root, 'f.txt'), 'w').close()
+            uri_list = f'file://{root}\r\n'.encode()
+            # The context manager calls dnd_test_cleanup_fake_window on exit,
+            # which calls drop_free_data → drop_free_dir_handles.
+            with dnd_test_window() as (osw, wid, screen, cap):
+                self._setup_uri_drop(screen, wid, cap, uri_list)
+                parse_bytes(screen, client_request_uri_data(0))
+                cap.consume(wid)
+                # Intentionally leave the handle open – cleanup happens in __exit__
+