Remote control: Allow matching for self window even over sockets when run inside a kitty window

Have kitty-tool send the value of KITTY_WINDOW_ID if present.

docs/rc_protocol.rst

@@ -15,6 +15,7 @@ Where ``<ESC>`` is the byte ``0x1b``. The JSON object has the form:
         "cmd": "command name",
         "version": "<kitty version>",
         "no_response": "<Optional Boolean>",
+        "kitty_window_id": "<Optional value of the KITTY_WINDOW_ID env var>",
         "payload": "<Optional JSON object>"
     }
 

kitty/boss.py

@@ -505,6 +505,18 @@ class Boss:
             return response
         if not pcmd:
             return response
+        self_window: Optional[Window] = None
+        if window is not None:
+            self_window = window
+        else:
+            try:
+                swid = int(pcmd.get('kitty_window_id', 0))
+            except Exception:
+                pass
+            else:
+                if swid > 0:
+                    self_window = self.window_id_map.get(swid)
+
         extra_data: Dict[str, Any] = {}
         try:
             allowed_unconditionally = (
@@ -513,12 +525,12 @@ class Boss:
         except PermissionError:
             return {'ok': False, 'error': 'Remote control disallowed by window specific password'}
         if allowed_unconditionally:
-            return self._execute_remote_command(pcmd, window, peer_id)
+            return self._execute_remote_command(pcmd, window, peer_id, self_window)
         q = is_cmd_allowed(pcmd, window, peer_id > 0, extra_data)
         if q is True:
-            return self._execute_remote_command(pcmd, window, peer_id)
+            return self._execute_remote_command(pcmd, window, peer_id, self_window)
         if q is None:
-            if self.ask_if_remote_cmd_is_allowed(pcmd, window, peer_id):
+            if self.ask_if_remote_cmd_is_allowed(pcmd, window, peer_id, self_window):
                 return AsyncResponse()
         response = {'ok': False, 'error': 'Remote control is disabled. Add allow_remote_control to your kitty.conf'}
         if q is False and pcmd.get('password'):
@@ -528,7 +540,9 @@ class Boss:
             return None
         return response
 
-    def ask_if_remote_cmd_is_allowed(self, pcmd: Dict[str, Any], window: Optional[Window] = None, peer_id: int = 0) -> bool:
+    def ask_if_remote_cmd_is_allowed(
+        self, pcmd: Dict[str, Any], window: Optional[Window] = None, peer_id: int = 0, self_window: Optional[Window] = None
+    ) -> bool:
         from kittens.tui.operations import styled
         in_flight = 0
         for w in self.window_id_map.values():
@@ -547,7 +561,7 @@ class Boss:
                   '\x1b[m' + styled(_(
                       'Note that allowing the password will allow all future actions using the same password, in this kitty instance.'
                   ), dim=True, italic=True)),
-            partial(self.remote_cmd_permission_received, pcmd, wid, peer_id),
+            partial(self.remote_cmd_permission_received, pcmd, wid, peer_id, self_window),
             'a;green:Allow request', 'p;yellow:Allow password', 'r;magenta:Deny request', 'd;red:Deny password',
             window=window, default='a', hidden_text=hidden_text
         )
@@ -556,7 +570,7 @@ class Boss:
         overlay_window.window_custom_type = 'remote_command_permission_dialog'
         return True
 
-    def remote_cmd_permission_received(self, pcmd: Dict[str, Any], window_id: int, peer_id: int, choice: str) -> None:
+    def remote_cmd_permission_received(self, pcmd: Dict[str, Any], window_id: int, peer_id: int, self_window: Optional[Window], choice: str) -> None:
         from .remote_control import encode_response_for_peer, set_user_password_allowed
         response: RCResponse = None
         window = self.window_id_map.get(window_id)
@@ -570,7 +584,7 @@ class Boss:
         elif choice in ('a', 'p'):
             if choice == 'p':
                 set_user_password_allowed(pcmd['password'], True)
-            response = self._execute_remote_command(pcmd, window, peer_id)
+            response = self._execute_remote_command(pcmd, window, peer_id, self_window)
         if window is not None and response is not None and not isinstance(response, AsyncResponse):
             window.send_cmd_response(response)
         if peer_id > 0:
@@ -579,10 +593,12 @@ class Boss:
             elif not isinstance(response, AsyncResponse):
                 send_data_to_peer(peer_id, encode_response_for_peer(response))
 
-    def _execute_remote_command(self, pcmd: Dict[str, Any], window: Optional[Window] = None, peer_id: int = 0) -> RCResponse:
+    def _execute_remote_command(
+        self, pcmd: Dict[str, Any], window: Optional[Window] = None, peer_id: int = 0, self_window: Optional[Window] = None
+    ) -> RCResponse:
         from .remote_control import handle_cmd
         try:
-            response = handle_cmd(self, window, pcmd, peer_id)
+            response = handle_cmd(self, window, pcmd, peer_id, self_window)
         except Exception as err:
             import traceback
             response = {'ok': False, 'error': str(err)}

kitty/remote_control.py

@@ -169,7 +169,9 @@ def close_active_stream(stream_id: str) -> None:
     active_streams.pop(stream_id, None)
 
 
-def handle_cmd(boss: BossType, window: Optional[WindowType], cmd: Dict[str, Any], peer_id: int) -> Union[Dict[str, Any], None, AsyncResponse]:
+def handle_cmd(
+    boss: BossType, window: Optional[WindowType], cmd: Dict[str, Any], peer_id: int, self_window: Optional[WindowType]
+) -> Union[Dict[str, Any], None, AsyncResponse]:
     v = cmd['version']
     no_response = cmd.get('no_response', False)
     if tuple(v)[:2] > version[:2]:
@@ -194,14 +196,14 @@ def handle_cmd(boss: BossType, window: Optional[WindowType], cmd: Dict[str, Any]
         payload['async_id'] = async_id
         if 'cancel_async' in cmd:
             active_async_requests.pop(async_id, None)
-            c.cancel_async_request(boss, window, PayloadGetter(c, payload))
+            c.cancel_async_request(boss, self_window or window, PayloadGetter(c, payload))
             return None
         active_async_requests[async_id] = monotonic()
         if len(active_async_requests) > 32:
             oldest = next(iter(active_async_requests))
             del active_async_requests[oldest]
     try:
-        ans = c.response_from_kitty(boss, window, PayloadGetter(c, payload))
+        ans = c.response_from_kitty(boss, self_window or window, PayloadGetter(c, payload))
     except Exception:
         if no_response:  # don't report errors if --no-response was used
             return None

tools/cmd/at/main.go

@@ -10,6 +10,7 @@ import (
 	"io"
 	"os"
 	"reflect"
+	"strconv"
 	"strings"
 	"time"
 	"unicode/utf16"
@@ -268,6 +269,10 @@ func send_rc_command(io_data *rc_io_data) (err error) {
 	if err != nil {
 		return err
 	}
+	wid, err := strconv.Atoi(os.Getenv("KITTY_WINDOW_ID"))
+	if err == nil && wid > 0 {
+		io_data.rc.KittyWindowId = uint(wid)
+	}
 	err = create_serializer(global_options.password, "", io_data)
 	if err != nil {
 		return

tools/utils/types.go

@@ -12,6 +12,7 @@ type RemoteControlCmd struct {
 	CancelAsync   bool   `json:"cancel_async,omitempty"`
 	Stream        bool   `json:"stream,omitempty"`
 	StreamId      string `json:"stream_id,omitempty"`
+	KittyWindowId uint   `json:"kitty_window_id,omitempty"`
 	Payload       any    `json:"payload,omitempty"`
 }