ayo/kitty
1
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

DRYer

Browse source
This commit is contained in:
Kovid Goyal 2022-07-29 12:51:36 +05:30
parent e2117ee8a4
commit 8202247b27
No known key found for this signature in database
GPG key ID: 06BC317B515ACE7C
2 changed files with 19 additions and 9 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

13
kitty/boss.py
View file

@ -439,23 +439,22 @@ def add_child(self, window: Window) -> None:
self.window_id_map[window.id] = window
def _handle_remote_command(self, cmd: str, window: Optional[Window] = None, peer_id: int = 0) -> Union[Dict[str, Any], None, AsyncResponse]:
from .remote_control import handle_cmd
from .remote_control import handle_cmd, parse_cmd
response = None
window = window or None
pcmd = parse_cmd(cmd)
if not pcmd:
return response
if self.allow_remote_control == 'y' or peer_id > 0 or getattr(window, 'allow_remote_control', False):
try:
response = handle_cmd(self, window, cmd, peer_id)
response = handle_cmd(self, window, pcmd, peer_id)
except Exception as err:
import traceback
response = {'ok': False, 'error': str(err)}
if not getattr(err, 'hide_traceback', False):
response['tb'] = traceback.format_exc()
else:
no_response = False
try:
no_response = json.loads(cmd).get('no_response')
except Exception:
pass
no_response = pcmd.get('no_response') or False
if not no_response:
response = {'ok': False, 'error': 'Remote control is disabled. Add allow_remote_control to your kitty.conf'}
return response

15
kitty/remote_control.py
View file

@ -32,8 +32,19 @@ def encode_response_for_peer(response: Any) -> bytes:
return b'\x1bP@kitty-cmd' + json.dumps(response).encode('utf-8') + b'\x1b\\'
def handle_cmd(boss: BossType, window: Optional[WindowType], serialized_cmd: str, peer_id: int) -> Union[Dict[str, Any], None, AsyncResponse]:
cmd = json.loads(serialized_cmd)
def parse_cmd(serialized_cmd: str) -> Dict[str, Any]:
try:
pcmd = json.loads(serialized_cmd)
if not isinstance(pcmd, dict):
return {}
except Exception:
return {}
if 'version' not in pcmd:
return {}
return pcmd
def handle_cmd(boss: BossType, window: Optional[WindowType], cmd: Dict[str, Any], peer_id: int) -> Union[Dict[str, Any], None, AsyncResponse]:
v = cmd['version']
no_response = cmd.get('no_response', False)
if tuple(v)[:2] > version[:2]: