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

Cleanup kitty dcs parsing

Browse source
This commit is contained in:
Kovid Goyal 2023-11-07 16:54:17 +05:30
parent 0a6d83901d
commit 50935b6c93
No known key found for this signature in database
GPG key ID: 06BC317B515ACE7C
6 changed files with 47 additions and 44 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
kitty/screen.c
View file

@ -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;

1
kitty/screen.h
View file

@ -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 *);

66
kitty/vt-parser.c
View file

@ -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);

9
kitty/window.py
View file

@ -193,9 +193,8 @@ def modify_argv_for_launch_with_cwd(self, argv: List[str], env: Optional[Dict[st
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 @@ def handle_remote_cmd(self, cmd: memoryview) -> None:
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)

6
kitty_tests/__init__.py
View file

@ -83,6 +83,7 @@ def clear(self) -> None:
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 @@ def on_mouse_event(self, event):
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')

4
kitty_tests/parser.py
View file

@ -423,6 +423,10 @@ def test_dcs_codes(self):
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()