From 50935b6c932fd66eeebdc5992e639ca7175541b2 Mon Sep 17 00:00:00 2001 From: Kovid Goyal Date: Tue, 7 Nov 2023 16:54:17 +0530 Subject: [PATCH] Cleanup kitty dcs parsing --- kitty/screen.c | 5 ---- kitty/screen.h | 1 - kitty/vt-parser.c | 66 ++++++++++++++++++++++------------------- kitty/window.py | 9 ++---- kitty_tests/__init__.py | 6 +++- kitty_tests/parser.py | 4 +++ 6 files changed, 47 insertions(+), 44 deletions(-) diff --git a/kitty/screen.c b/kitty/screen.c index 3bcfc951f..e3d502490 100644 --- a/kitty/screen.c +++ b/kitty/screen.c @@ -2188,11 +2188,6 @@ process_cwd_notification(Screen *self, unsigned int code, const char *data, size } // we ignore OSC 6 document reporting as we dont have a use for it } -void -screen_handle_cmd(Screen *self, PyObject *cmd) { - CALLBACK("handle_remote_cmd", "O", cmd); -} - bool screen_send_signal_for_key(Screen *self, char key) { int ret = 0; diff --git a/kitty/screen.h b/kitty/screen.h index df10d2d9d..cd511bb04 100644 --- a/kitty/screen.h +++ b/kitty/screen.h @@ -203,7 +203,6 @@ void screen_repeat_character(Screen *self, unsigned int count); void screen_delete_characters(Screen *self, unsigned int count); void screen_erase_characters(Screen *self, unsigned int count); void screen_set_margins(Screen *self, unsigned int top, unsigned int bottom); -void screen_handle_cmd(Screen *, PyObject *cmd); void screen_push_colors(Screen *, unsigned int); void screen_pop_colors(Screen *, unsigned int); void screen_report_color_stack(Screen *); diff --git a/kitty/vt-parser.c b/kitty/vt-parser.c index 0ef7516a6..ea755dadd 100644 --- a/kitty/vt-parser.c +++ b/kitty/vt-parser.c @@ -627,6 +627,40 @@ activate_pending_mode(PS *self) { self->pending_mode.state = PENDING_NORMAL; } +static bool +parse_kitty_dcs(PS *self, uint8_t *buf, size_t bufsz) { +#define starts_with(x) startswith(buf, bufsz, x, literal_strlen(x)) +#define inc(x) buf += literal_strlen(x); bufsz -= literal_strlen(x) +#define dispatch(prefix, func, delta) {\ + if (starts_with(prefix)) {\ + inc(prefix); buf -= delta; bufsz += delta; \ + PyObject *cmd = PyMemoryView_FromMemory((char*)buf, bufsz, PyBUF_READ); \ + if (cmd) { \ + REPORT_OSC(func, cmd); \ + screen_handle_kitty_dcs(self->screen, #func, cmd); \ + Py_DECREF(cmd); \ + } else PyErr_Clear(); \ + return true; \ + }} + if (!starts_with("kitty-")) return false; + inc("kitty-"); + + dispatch("cmd{", handle_remote_cmd, 1); + dispatch("overlay-ready|", handle_overlay_ready, 0) + dispatch("kitten-result|", handle_kitten_result, 0) + dispatch("print|", handle_remote_print, 0) + dispatch("echo|", handle_remote_echo, 0) + dispatch("ssh|", handle_remote_ssh, 0) + dispatch("ask|", handle_remote_askpass, 0) + dispatch("clone|", handle_remote_clone, 0) + dispatch("edit|", handle_remote_edit, 0) + + return false; +#undef dispatch +#undef starts_with +#undef inc +} + static void dispatch_dcs(PS *self, uint8_t *buf, size_t bufsz, bool is_extended UNUSED) { if (bufsz < 2) return; @@ -665,37 +699,7 @@ dispatch_dcs(PS *self, uint8_t *buf, size_t bufsz, bool is_extended UNUSED) { REPORT_UKNOWN_ESCAPE_CODE("DCS", buf); } break; case '@': - if (startswith(buf + 1, bufsz - 2, "kitty-", sizeof("kitty-") - 1)) { - if (startswith(buf + 7, bufsz - 2, "cmd{", sizeof("cmd{") - 1)) { - PyObject *cmd = PyMemoryView_FromMemory((char*)buf + 10, bufsz - 10, PyBUF_READ); - if (cmd != NULL) { - REPORT_OSC2(screen_handle_cmd, (char)buf[0], cmd); - screen_handle_cmd(self->screen, cmd); - Py_DECREF(cmd); - } else PyErr_Clear(); -#define IF_SIMPLE_PREFIX(prefix, func) \ - if (startswith(buf + 7, bufsz - 1, prefix, sizeof(prefix) - 1)) { \ - const size_t pp_size = sizeof("kitty") + sizeof(prefix); \ - PyObject *msg = PyMemoryView_FromMemory((char*)buf + pp_size, bufsz - pp_size, PyBUF_READ); \ - if (msg != NULL) { \ - REPORT_OSC2(func, (char)buf[0], msg); \ - screen_handle_kitty_dcs(self->screen, #func, msg); \ - Py_DECREF(msg); \ - } else PyErr_Clear(); - - } else IF_SIMPLE_PREFIX("overlay-ready|", handle_overlay_ready) - } else IF_SIMPLE_PREFIX("kitten-result|", handle_kitten_result) - } else IF_SIMPLE_PREFIX("print|", handle_remote_print) - } else IF_SIMPLE_PREFIX("echo|", handle_remote_echo) - } else IF_SIMPLE_PREFIX("ssh|", handle_remote_ssh) - } else IF_SIMPLE_PREFIX("ask|", handle_remote_askpass) - } else IF_SIMPLE_PREFIX("clone|", handle_remote_clone) - } else IF_SIMPLE_PREFIX("edit|", handle_remote_edit) -#undef IF_SIMPLE_PREFIX - } else { - REPORT_UKNOWN_ESCAPE_CODE("DCS", buf); - } - } + if (!parse_kitty_dcs(self, buf + 1, bufsz-1)) REPORT_UKNOWN_ESCAPE_CODE("DCS", buf); break; default: REPORT_UKNOWN_ESCAPE_CODE("DCS", buf); diff --git a/kitty/window.py b/kitty/window.py index e859d9585..484ec393e 100644 --- a/kitty/window.py +++ b/kitty/window.py @@ -193,9 +193,8 @@ class CwdRequest: def process_title_from_child(title: memoryview, is_base64: bool, default_title: str) -> str: if is_base64: - from base64 import standard_b64decode try: - stitle = standard_b64decode(title).decode('utf-8', 'replace') + stitle = base64_decode(title).decode('utf-8', 'replace') except Exception: stitle = 'undecodeable title' else: @@ -450,8 +449,7 @@ def cmd_output(screen: Screen, which: CommandOutput = CommandOutput.last_run, as def process_remote_print(msg: memoryview) -> str: - from base64 import standard_b64decode - return replace_c0_codes_except_nl_space_tab(standard_b64decode(msg)).decode('utf-8', 'replace') + return replace_c0_codes_except_nl_space_tab(base64_decode(msg)).decode('utf-8', 'replace') class EdgeWidths: @@ -1240,8 +1238,7 @@ class Window: get_boss().handle_remote_cmd(cmd, self) def handle_remote_echo(self, msg: memoryview) -> None: - from base64 import standard_b64decode - data = standard_b64decode(msg) + data = base64_decode(msg) # ensure we are not writing any control char back as this can lead to command injection on shell prompts # Any bytes outside the printable ASCII range are removed. data = re.sub(rb'[^ -~]', b'', data) diff --git a/kitty_tests/__init__.py b/kitty_tests/__init__.py index cdfccb037..e45994884 100644 --- a/kitty_tests/__init__.py +++ b/kitty_tests/__init__.py @@ -83,6 +83,7 @@ class Callbacks: self.wtcbuf = b'' self.iconbuf = self.colorbuf = self.ctbuf = '' self.titlebuf = [] + self.printbuf = [] self.notifications = [] self.open_urls = [] self.cc_buf = [] @@ -110,7 +111,10 @@ class Callbacks: def handle_remote_print(self, msg): text = process_remote_print(msg) - print(text, file=sys.__stdout__, end='', flush=True) + self.printbuf.append(text) + + def handle_remote_cmd(self, msg): + pass def handle_remote_clone(self, msg): msg = str(msg, 'utf-8') diff --git a/kitty_tests/parser.py b/kitty_tests/parser.py index 3fd527134..31e9e5763 100644 --- a/kitty_tests/parser.py +++ b/kitty_tests/parser.py @@ -423,6 +423,10 @@ class TestParser(BaseTest): c.clear() pb('\033P$qr\033\\', ('screen_request_capabilities', ord('$'), 'r')) self.ae(c.wtcbuf, f'\033P1$r{s.margin_top + 1};{s.margin_bottom + 1}r\033\\'.encode('ascii')) + pb('\033P@kitty-cmd{abc\033\\', ('handle_remote_cmd', '{abc')) + p = base64_encode('abcd').decode() + pb(f'\033P@kitty-print|{p}\033\\', ('handle_remote_print', p)) + self.ae(['abcd'], s.callbacks.printbuf) def test_pending(self): s = self.create_screen()