Get askpass basically working

kittens/ask/main.py

@@ -74,7 +74,7 @@ class HistoryCompleter:
 def option_text() -> str:
     return '''\
 --type -t
-choices=line,yesno,choices
+choices=line,yesno,choices,password
 default=line
 Type of input. Defaults to asking for a line of text.
 
@@ -361,6 +361,13 @@ def main(args: List[str]) -> Response:
         loop.loop(handler)
         return {'items': items, 'response': handler.response}
 
+    if cli_opts.type == 'password':
+        import getpass
+        if cli_opts.message:
+            print(styled(cli_opts.message, bold=True))
+        q = getpass.getpass()
+        return {'items': items, 'response': q or ''}
+
     import readline as rl
     readline = rl
     from kitty.shell import init_readline

kitty/parser.c

@@ -1084,7 +1084,7 @@ dispatch_dcs(Screen *screen, PyObject DUMP_UNUSED *dump_callback) {
             } else IF_SIMPLE_PREFIX("kitty-print|", screen_handle_print)
             } else IF_SIMPLE_PREFIX("kitty-echo|", screen_handle_echo)
             } else IF_SIMPLE_PREFIX("kitty-ssh|", screen_handle_ssh)
-            } else IF_SIMPLE_PREFIX("kitty-askpass|", screen_handle_askpass)
+            } else IF_SIMPLE_PREFIX("kitty-ask|", screen_handle_askpass)
 #undef IF_SIMPLE_PREFIX
             } else {
                 REPORT_ERROR("Unrecognized DCS @ code: 0x%x", screen->parser_buf[1]);

shell-integration/ssh/askpass.py

@@ -3,22 +3,36 @@
 
 import json
 import os
-import struct
 import sys
+import time
 
 from kitty.shm import SharedMemory
 
 msg = sys.argv[-1]
 prompt = os.environ.get('SSH_ASKPASS_PROMPT', '')
 is_confirm = prompt == 'confirm'
-ask_cmdline = ['-m', msg, '--type', 'yesno' if is_confirm else 'password']
-if is_confirm:
-    ask_cmdline += ['--default', 'y']
-data = json.dumps(ask_cmdline).encode('utf-8')
-sz = struct.pack('>I', len(data))
-with SharedMemory(size=len(data) + len(sz) + 1, unlink_on_exit=True, prefix=f'askpass-{os.getpid()}-') as shm, open(os.ctermid(), 'wb') as tty:
+q = {
+    'prompt': msg,
+    'type': 'confirm' if is_confirm else 'get_line',
+    'is_password': True,
+}
+
+data = json.dumps(q)
+with SharedMemory(size=len(data) + 1 + SharedMemory.num_bytes_for_size, unlink_on_exit=True, prefix=f'askpass-{os.getpid()}-'
+                  ) as shm, open(os.ctermid(), 'wb') as tty:
     shm.write(b'\0')
-    shm.write(sz)
-    shm.write(data)
+    shm.write_data_with_size(data)
     shm.flush()
-    print(f'\x1bP@kitty-ask|{shm.name}\x1b\\', flush=True)
+    with open(os.ctermid(), 'wb') as f:
+        f.write(f'\x1bP@kitty-ask|{shm.name}\x1b\\'.encode('ascii'))
+    while True:
+        # TODO: Replace sleep() with a mutex and condition variable created in the shared memory
+        time.sleep(0.05)
+        shm.seek(0)
+        if shm.read(1) == b'\x01':
+            break
+    response = json.loads(shm.read_data_with_size())
+    if is_confirm:
+        response = 'yes' if response else 'no'
+    if response:
+        print(response, flush=True)