clipboard kitten: Allow using a password to avoid repeated confirmation prompts when accessing the clipboard

Fixes #8789

Author: kovidgoyal

docs/changelog.rst

@@ -111,6 +111,10 @@ Detailed list of changes
 
 - A new :ref:`protocol extension <mouse_leave_window>` to notify terminal programs that have turned on SGR Pixel mouse reporting when the mouse leaves the window (:disc:`8808`)
 
+- clipboard kitten: Can now optionally take a password to avoid repeated
+  permission prompts when accessing the clipboard. Based on a
+  :ref:`protocol extension <clipboard_repeated_permission>`. (:iss:`8789`)
+
 - A new :option:`launch --hold-after-ssh` to not close a launched window
   that connects directly to a remote host because of
   :option:`launch --cwd`:code:`=current` when the connection ends (:pull:`8807`)

docs/clipboard.rst

@@ -138,6 +138,29 @@ the data, but create multiple references to it in the system clipboard. Alias
 packets can be sent anytime after the initial write packet and before the end
 of data packet.
 
+.. _clipboard_repeated_permission:
+
+Avoiding repeated permission prompts
+--------------------------------------
+
+.. versionadded:: using a password to avoid repeated confirmations was added in version 0.43.0
+
+If a program like an editor wants to make use of the system clipboard, by
+default, the user is prompted on every read request. This can become quite
+fatiguing. To avoid this situation, this protocol allows sending a password
+and human friendly name with ``type=write`` and ``type=read`` requests. The
+terminal can then ask the user to allow all future requests using that
+password. If the user agrees, future requests on the same tty will be
+automatically allowed by the terminal. The editor or other program using
+this facility should ideally use a password randomnly generated at startup,
+such as a UUID4. However, terminals may implement permanent/stored passwords.
+Users can then configure terminal programs they trust to use these password.
+
+The password and the human name are encoded using the ``pw`` and ``name`` keys
+in the metadata. The values are UTF-8 strings that are base64 encoded.
+Specifying a password without a human friendly name is equivalent to not
+specifying a password and the terminal must treat the request as though
+it had no password.
 
 Support for terminal multiplexers
 ------------------------------------

kittens/clipboard/main.go

@@ -3,7 +3,12 @@
 package clipboard
 
 import (
+	"fmt"
+	"io"
 	"os"
+	"strconv"
+	"strings"
+	"unicode"
 
 	"github.com/kovidgoyal/kitty/tools/cli"
 )
@@ -20,10 +25,47 @@ func run_mime_loop(opts *Options, args []string) (err error) {
 }
 
 func clipboard_main(cmd *cli.Command, opts *Options, args []string) (rc int, err error) {
+	if opts.Password != "" {
+		if opts.HumanName == "" {
+			return 1, fmt.Errorf("must specify --human-name when using a password")
+		}
+		ptype, val, found := strings.Cut(opts.Password, ":")
+		if !found {
+			return 1, fmt.Errorf("invalid password: %#v no password type specified", opts.Password)
+		}
+		switch ptype {
+		case "text":
+			opts.Password = val
+		case "fd":
+			if fd, err := strconv.Atoi(val); err == nil {
+				if f := os.NewFile(uintptr(fd), "password-fd"); f == nil {
+					return 1, fmt.Errorf("invalid file descriptor: %d", fd)
+				} else {
+					data, err := io.ReadAll(f)
+					f.Close()
+					if err != nil {
+						return 1, fmt.Errorf("failed to read from file descriptor: %d with error: %w", fd, err)
+					}
+					opts.Password = strings.TrimRightFunc(string(data), unicode.IsSpace)
+				}
+
+			} else {
+				return 1, fmt.Errorf("not a valid file descriptor number: %#v", val)
+			}
+		case "file":
+			if data, err := os.ReadFile(val); err == nil {
+				opts.Password = strings.TrimRightFunc(string(data), unicode.IsSpace)
+			} else {
+				return 1, fmt.Errorf("failed to read from file: %#v with error: %w", val, err)
+			}
+		}
+	}
 	if len(args) > 0 {
 		return 0, run_mime_loop(opts, args)
 	}
-
+	if opts.Password != "" || opts.HumanName != "" {
+		return 1, fmt.Errorf("cannot use --human-name or --password in filter mode")
+	}
 	return 0, run_plain_text_loop(opts)
 }
 

kittens/clipboard/main.py

@@ -44,6 +44,18 @@
 type=bool-set
 Wait till the copy to clipboard is complete before exiting. Useful if running
 the kitten in a dedicated, ephemeral window. Only needed in filter mode.
+
+
+--password
+A password to use when accessing the clipboard. If the user chooses to accept the password
+future invocations of the kitten will not have a permission prompt in this tty session. Does not
+work in filter mode. Must be of the form: text:actual-password or fd:integer (a file descriptor
+number to read the password from) or file:path-to-file (a file from which to read the password).
+Note that you must also specify a human friendly name using the :option:`--human-name` flag.
+
+
+--human-name
+A human friendly name to show the user when asking for permission to access the clipboard.
 '''.format
 help_text = '''\
 Read or write to the system clipboard.

kittens/clipboard/read.go

@@ -329,9 +329,14 @@ func run_get_loop(opts *Options, args []string) (err error) {
 	if opts.UsePrimary {
 		basic_metadata["loc"] = "primary"
 	}
-
 	lp.OnInitialize = func() (string, error) {
 		lp.QueueWriteString(encode(basic_metadata, "."))
+		if opts.Password != "" {
+			basic_metadata["pw"] = base64.StdEncoding.EncodeToString(utils.UnsafeStringToBytes(opts.Password))
+		}
+		if opts.HumanName != "" {
+			basic_metadata["name"] = base64.StdEncoding.EncodeToString(utils.UnsafeStringToBytes(opts.HumanName))
+		}
 		return "", nil
 	}
 

kittens/clipboard/write.go

@@ -3,6 +3,7 @@
 package clipboard
 
 import (
+	"encoding/base64"
 	"errors"
 	"fmt"
 	"io"
@@ -80,6 +81,14 @@ func write_loop(inputs []*Input, opts *Options) (err error) {
 		if mime != "" {
 			ans["mime"] = mime
 		}
+		if ptype == "write" {
+			if opts.Password != "" {
+				ans["pw"] = base64.StdEncoding.EncodeToString(utils.UnsafeStringToBytes(opts.Password))
+			}
+			if opts.HumanName != "" {
+				ans["name"] = base64.StdEncoding.EncodeToString(utils.UnsafeStringToBytes(opts.HumanName))
+			}
+		}
 		return ans
 	}
 

kitty/clipboard.py

@@ -22,6 +22,7 @@
     get_options,
     set_clipboard_data_types,
 )
+from .typing_compat import WindowType
 from .utils import log_error
 
 READ_RESPONSE_CHUNK_SIZE = 4096
@@ -200,7 +201,7 @@ def encode_mime(x: str) -> str:
 
 
 def decode_metadata_value(k: str, x: str) -> str:
-    if k == 'mime':
+    if k in ('mime', 'name', 'pw'):
         import base64
         x = base64.standard_b64decode(x).decode('utf-8')
     return x
@@ -211,6 +212,8 @@ class ReadRequest(NamedTuple):
     mime_types: tuple[str, ...] = ('text/plain',)
     id: str = ''
     protocol_type: ProtocolType = ProtocolType.osc_52
+    human_name: str = ''
+    password: str = ''
 
     def encode_response(self, status: str = 'DATA', mime: str = '', payload: bytes | memoryview = b'') -> bytes:
         ans = f'{self.protocol_type.value};type=read:status={status}'
@@ -242,9 +245,11 @@ class WriteRequest:
 
     def __init__(
         self, is_primary_selection: bool = False, protocol_type: ProtocolType = ProtocolType.osc_52, id: str = '',
-        rollover_size: int = 16 * 1024 * 1024, max_size: int = -1,
+        rollover_size: int = 16 * 1024 * 1024, max_size: int = -1, human_name: str = '', password: str = '',
     ) -> None:
         self.decoder = StreamingBase64Decoder()
+        self.human_name = human_name
+        self.password = password
         self.id = id
         self.is_primary_selection = is_primary_selection
         self.protocol_type = protocol_type
@@ -255,6 +260,8 @@ def __init__(
         self.max_size = (get_options().clipboard_max_size * 1024 * 1024) if max_size < 0 else max_size
         self.aliases: dict[str, str] = {}
         self.committed = False
+        self.permission_pending = True
+        self.commit_pending = False
 
     def encode_response(self, status: str = 'OK') -> bytes:
         ans = f'{self.protocol_type.value};type=write:status={status}'
@@ -266,7 +273,11 @@ def encode_response(self, status: str = 'OK') -> bytes:
     def commit(self) -> None:
         if self.committed:
             return
+        if self.permission_pending:
+            self.commit_pending = True
+            return
         self.committed = True
+        self.commit_pending = False
         cp = get_boss().primary_selection if self.is_primary_selection else get_boss().clipboard
         if cp.enabled:
             for alias, src in self.aliases.items():
@@ -316,13 +327,21 @@ def data_for(self, mime: str = 'text/plain', offset: int = 0, size: int = -1) ->
         return self.tempfile.read(start+offset, size)
 
 
+class GrantedPermission:
+
+    def __init__(self, read: bool = False, write: bool = False):
+        self.read, self.write = read, write
+        self.write_ban = self.read_ban = False
+
+
 class ClipboardRequestManager:
 
     def __init__(self, window_id: int) -> None:
         self.window_id = window_id
         self.currently_asking_permission_for: ReadRequest | None = None
         self.in_flight_write_request: WriteRequest | None = None
         self.osc52_in_flight_write_requests: dict[ClipboardType, WriteRequest] = {}
+        self.granted_passwords: dict[str, GrantedPermission] = {}
 
     def parse_osc_5522(self, data: memoryview) -> None:
         import base64
@@ -349,13 +368,15 @@ def parse_osc_5522(self, data: memoryview) -> None:
             rr = ReadRequest(
                 is_primary_selection=m.get('loc', '') == 'primary',
                 mime_types=tuple(payload.decode('utf-8').split()),
-                protocol_type=ProtocolType.osc_5522, id=sanitize_id(m.get('id', ''))
+                protocol_type=ProtocolType.osc_5522, id=sanitize_id(m.get('id', '')),
+                human_name=m.get('name', ''), password=m.get('pw', ''),
             )
             self.handle_read_request(rr)
         elif typ == 'write':
             self.in_flight_write_request = WriteRequest(
                 is_primary_selection=m.get('loc', '') == 'primary',
-                protocol_type=ProtocolType.osc_5522, id=sanitize_id(m.get('id', ''))
+                protocol_type=ProtocolType.osc_5522, id=sanitize_id(m.get('id', '')),
+                human_name=m.get('name', ''), password=m.get('pw', ''),
             )
             self.handle_write_request(self.in_flight_write_request)
         elif typ == 'walias':
@@ -385,11 +406,17 @@ def parse_osc_5522(self, data: memoryview) -> None:
                     self.in_flight_write_request = None
                     raise
             else:
-                wr.flush_base64_data()
-                wr.commit()
-                self.in_flight_write_request = None
-                if w is not None:
-                    w.screen.send_escape_code_to_child(ESC_OSC, wr.encode_response(status='DONE'))
+                self.commit_write_request(wr)
+
+    def commit_write_request(self, wr: WriteRequest, needs_flush: bool = True) -> None:
+        if needs_flush:
+            wr.flush_base64_data()
+        wr.commit()
+        if wr.committed:
+            self.in_flight_write_request = None
+            w = get_boss().window_id_map.get(self.window_id)
+            if w is not None:
+                w.screen.send_escape_code_to_child(ESC_OSC, wr.encode_response(status='DONE'))
 
     def parse_osc_52(self, data: memoryview, is_partial: bool = False) -> None:
         idx = find_in_memoryview(data, ord(b';'))
@@ -423,15 +450,78 @@ def handle_write_request(self, wr: WriteRequest) -> None:
         self.fulfill_write_request(wr, allowed)
 
     def fulfill_write_request(self, wr: WriteRequest, allowed: bool = True) -> None:
+        wr.permission_pending = not allowed
         if wr.protocol_type is ProtocolType.osc_52:
             self.fulfill_legacy_write_request(wr, allowed)
             return
-        w = get_boss().window_id_map.get(self.window_id)
         cp = get_boss().primary_selection if wr.is_primary_selection else get_boss().clipboard
-        if not allowed or not cp.enabled:
+        w = get_boss().window_id_map.get(self.window_id)
+        if w is None:
             self.in_flight_write_request = None
-            if w is not None:
-                w.screen.send_escape_code_to_child(ESC_OSC, wr.encode_response(status='EPERM' if not allowed else 'ENOSYS'))
+            return
+        if not cp.enabled:
+            self.in_flight_write_request = None
+            w.screen.send_escape_code_to_child(ESC_OSC, wr.encode_response(status='ENOSYS'))
+            return
+        if not allowed:
+            if wr.password and wr.human_name:
+                if self.password_is_allowed_already(wr.password, for_write=True):
+                    wr.permission_pending = False
+                else:
+                    wid = w.id
+                    def callback(granted: bool) -> None:
+                        if wr is not self.in_flight_write_request:
+                            return
+                        if granted:
+                            wr.permission_pending = False
+                            if wr.commit_pending:
+                                self.commit_write_request(wr, needs_flush=False)
+                        else:
+                            w = get_boss().window_id_map.get(wid)
+                            if w is not None:
+                                w.screen.send_escape_code_to_child(ESC_OSC, wr.encode_response(status='EPERM'))
+                        self.in_flight_write_request = None
+
+                    self.request_permission(w, wr.human_name, wr.password, callback, for_write=True)
+            else:
+                self.in_flight_write_request = None
+                w.screen.send_escape_code_to_child(ESC_OSC, wr.encode_response(status='EPERM'))
+
+    def request_permission(self, window: WindowType, human_name: str, password: str, callback: Callable[[bool], None], for_write: bool = False) -> None:
+        if (gp := self.granted_passwords.get(password)) and (gp.write_ban if for_write else gp.read_ban):
+            callback(False)
+            return
+
+        def cb(q: str) -> None:
+            p = self.granted_passwords.get(password)
+            if p is None:
+                p = self.granted_passwords[password] = GrantedPermission()
+            callback(q in ('a', 'w'))
+            match q:
+                case 'w':
+                    if for_write:
+                        p.write = True
+                    else:
+                        p.read = True
+                case 'b':
+                    if for_write:
+                        p.write = False
+                        p.write_ban = True
+                    else:
+                        p.read = False
+                        p.read_ban = True
+        if for_write:
+            msg = _('The program {0} running in this window wants to write to the system clipboard.')
+        else:
+            msg = _('The program {0} running in this window wants to read from the system clipboard.')
+        msg += '\n\n' + ('If you choose "Always" similar requests from this program will be automatically allowed for the rest of this session.')
+        msg += '\n\n' + ('If you choose "Ban" similar requests from this program will be automatically dis-allowed for the rest of this session.')
+        from kittens.tui.operations import styled
+        get_boss().choose(msg.format(styled(human_name, fg='yellow')), cb, 'a;green:Allow', 'w;yellow:Always', 'd;red:Deny', 'b;red:Ban',
+                          default='d', window=window, title=_('A program wants to access the clipboard'))
+
+    def password_is_allowed_already(self, password: str, for_write: bool = False) -> bool:
+        return (q := self.granted_passwords.get(password)) is not None and (q.write if for_write else q.read)
 
     def fulfill_legacy_write_request(self, wr: WriteRequest, allowed: bool = True) -> None:
         cp = get_boss().primary_selection if wr.is_primary_selection else get_boss().clipboard
@@ -517,11 +607,24 @@ def ask_to_read_clipboard(self, rr: ReadRequest) -> None:
         w = get_boss().window_id_map.get(self.window_id)
         if w is not None:
             self.currently_asking_permission_for = rr
-            get_boss().confirm(_(
-                'A program running in this window wants to read from the system clipboard.'
-                ' Allow it to do so, once?'),
-                self.handle_clipboard_confirmation, window=w,
-            )
+            if rr.password and rr.human_name:
+                if self.password_is_allowed_already(rr.password):
+                    self.handle_clipboard_confirmation(True)
+                    return
+                if (p := self.granted_passwords.get(rr.password)) and p.read_ban:
+                    self.handle_clipboard_confirmation(False)
+                    return
+                self.request_permission(w, rr.human_name, rr.password, self.handle_clipboard_confirmation)
+            else:
+                if rr.human_name:
+                    msg = _(
+                    'The program {} running in this window wants to read from the system clipboard.'
+                    ' Allow it to do so, once?').format(rr.human_name)
+                else:
+                    msg = _(
+                    'A program running in this window wants to read from the system clipboard.'
+                    ' Allow it to do so, once?')
+                get_boss().confirm(msg, self.handle_clipboard_confirmation, window=w)
 
     def handle_clipboard_confirmation(self, confirmed: bool) -> None:
         rr = self.currently_asking_permission_for