Use encryption for bypass

docs/file-transfer-protocol.rst

@@ -458,6 +458,12 @@ The value of ``bypass`` is of the form ``hash_function_name : hash_value``
    they will introduce a lot of latency to starting a session and in any case
    there is no mathematical proof that **any** hash function is not brute-forceable.
 
+Terminal implementations are free to use their own more advanced hashing
+schemes, with prefixes other than those starting with ``sha``, which are
+reserved. For instance, kitty uses a scheme based on public key encryption
+via :envvar:`KITTY_PUBLIC_KEY`. For details of this scheme, see the
+``check_bypass()`` function in the kitty source code.
+
 Encoding of transfer commands as escape codes
 ------------------------------------------------
 

kittens/transfer/send.go

@@ -361,7 +361,13 @@ func (self *SendManager) start_transfer() string {
 
 func (self *SendManager) initialize() {
 	if self.bypass != "" {
-		self.bypass = encode_bypass(self.request_id, self.bypass)
+		q, err := encode_bypass(self.request_id, self.bypass)
+		if err == nil {
+			self.bypass = q
+		} else {
+			fmt.Fprintln(os.Stderr, "Ignoring password because of error:", err)
+		}
+
 	}
 	self.fid_map = make(map[string]*File, len(self.files))
 	for _, f := range self.files {

kittens/transfer/utils.go

@@ -4,13 +4,13 @@ package transfer
 
 import (
 	"crypto/rand"
-	"crypto/sha256"
 	"encoding/hex"
 	"fmt"
 	"os"
 	"path/filepath"
 	"strings"
 
+	"kitty/tools/crypto"
 	"kitty/tools/utils"
 )
 
@@ -33,10 +33,20 @@ func home_path() string {
 	return global_home
 }
 
-func encode_bypass(request_id string, bypass string) string {
+func encode_bypass(request_id string, bypass string) (string, error) {
 	q := request_id + ";" + bypass
-	sum := sha256.Sum256(utils.UnsafeStringToBytes(q))
-	return fmt.Sprintf("sha256:%x", sum)
+	if pkey_encoded := os.Getenv("KITTY_PUBLIC_KEY"); pkey_encoded != "" {
+		encryption_protocol, pubkey, err := crypto.DecodePublicKey(pkey_encoded)
+		if err != nil {
+			return "", err
+		}
+		encrypted, err := crypto.Encrypt_data(utils.UnsafeStringToBytes(q), pubkey, encryption_protocol)
+		if err != nil {
+			return "", err
+		}
+		return fmt.Sprintf("kitty-1:%s", utils.UnsafeBytesToString(encrypted)), nil
+	}
+	return "", fmt.Errorf("KITTY_PUBLIC_KEY env var not set, cannot transmit password securely")
 }
 
 func abspath(path string, use_home ...bool) string {

kitty/file_transmission.py

@@ -1,12 +1,13 @@
 #!/usr/bin/env python
 # License: GPLv3 Copyright: 2021, Kovid Goyal <kovid at kovidgoyal.net>
 
+import base64
 import errno
+import json
 import os
 import re
 import stat
 import tempfile
-from base64 import standard_b64decode
 from collections import defaultdict, deque
 from contextlib import suppress
 from dataclasses import Field, dataclass, field, fields
@@ -14,12 +15,12 @@
 from functools import partial
 from gettext import gettext as _
 from itertools import count
-from time import monotonic
+from time import monotonic, time_ns
 from typing import IO, Any, Callable, DefaultDict, Deque, Dict, Iterable, Iterator, List, Optional, Tuple, Union, cast
 
 from kittens.transfer.librsync import LoadSignature, PatchFile, delta_for_file, signature_of_file
 from kittens.transfer.utils import IdentityCompressor, ZlibCompressor, abspath, expand_home, home_path
-from kitty.fast_data_types import FILE_TRANSFER_CODE, OSC, add_timer, base64_encode, get_boss, get_options
+from kitty.fast_data_types import FILE_TRANSFER_CODE, OSC, AES256GCMDecrypt, add_timer, base64_encode, get_boss, get_options
 from kitty.types import run_once
 
 from .utils import log_error
@@ -252,7 +253,7 @@ def b64decode(val: memoryview) -> bytes:
     if extra != 0:
         padding = b'=' * (4 - extra)
         val = memoryview(bytes(val) + padding)
-    return standard_b64decode(val)
+    return base64.standard_b64decode(val)
 
 
 @dataclass
@@ -494,6 +495,29 @@ def write_data(self, all_files: Dict[str, 'DestFile'], data: bytes, is_last: boo
                 self.apply_metadata()
 
 
+def check_bypass(password: str, request_id: str, bypass_data: str) -> bool:
+    protocol, sep, bypass_data = bypass_data.partition(':')
+    if protocol == 'kitty-1':
+        try:
+            pcmd = json.loads(bypass_data)
+            pubkey = pcmd.get('pubkey', '')
+            if not pubkey:
+                return False
+            ekey = get_boss().encryption_key
+            d = AES256GCMDecrypt(ekey.derive_secret(base64.b85decode(pubkey)), base64.b85decode(pcmd['iv']), base64.b85decode(pcmd['tag']))
+            data = d.add_data_to_be_decrypted(base64.b85decode(pcmd['encrypted']), True)
+            timestamp, sep, payload = data.decode('utf-8').partition(':')
+            delta = time_ns() - int(timestamp)
+            if abs(delta) > 5 * 60 * 1e9:
+                return False
+            return payload == f'{request_id};{password}'
+        except Exception:
+            return False
+    elif protocol == 'sha256':
+        return (encode_bypass(request_id, password) == bypass_data) if password else False
+    return False
+
+
 class ActiveReceive:
     id: str
     files: Dict[str, DestFile]
@@ -504,7 +528,7 @@ def __init__(self, request_id: str, quiet: int, bypass: str) -> None:
         self.bypass_ok: Optional[bool] = None
         if bypass:
             byp = get_options().file_transfer_confirmation_bypass
-            self.bypass_ok = (encode_bypass(request_id, byp) == bypass) if byp else False
+            self.bypass_ok = check_bypass(byp, request_id, bypass)
         self.files = {}
         self.last_activity_at = monotonic()
         self.send_acknowledgements = quiet < 1

tools/crypto/crypto.go

@@ -11,9 +11,12 @@ import (
 	"crypto/sha256"
 	"encoding/json"
 	"fmt"
-	"github.com/jamesruan/go-rfc1924/base85"
 	"kitty/tools/utils"
+	"strconv"
+	"strings"
 	"time"
+
+	"github.com/jamesruan/go-rfc1924/base85"
 )
 
 func curve25519_key_pair() (private_key []byte, public_key []byte, err error) {
@@ -103,6 +106,15 @@ func EncodePublicKey(pubkey []byte, encryption_protocol string) (ans string, err
 	return
 }
 
+func DecodePublicKey(raw string) (encryption_protocol string, pubkey []byte, err error) {
+	encryption_protocol, encoded_pubkey, found := strings.Cut(raw, ":")
+	if !found {
+		return "", nil, fmt.Errorf("Invalid encoded pubkey, no : in string")
+	}
+	pubkey, err = b85_decode(encoded_pubkey)
+	return
+}
+
 func Encrypt_cmd(cmd *utils.RemoteControlCmd, password string, other_pubkey []byte, encryption_protocol string) (encrypted_cmd utils.EncryptedRemoteControlCmd, err error) {
 	cmd.Password = password
 	cmd.Timestamp = time.Now().UnixNano()
@@ -119,4 +131,21 @@ func Encrypt_cmd(cmd *utils.RemoteControlCmd, password string, other_pubkey []by
 	return
 }
 
+func Encrypt_data(data []byte, other_pubkey []byte, encryption_protocol string) (ans []byte, err error) {
+	d := make([]byte, 0, len(data)+32)
+	d = append(d, []byte(fmt.Sprintf("%s:", strconv.FormatInt(time.Now().UnixNano(), 10)))...)
+	d = append(d, data...)
+	iv, tag, ciphertext, pubkey, err := encrypt(d, other_pubkey, encryption_protocol)
+	if err != nil {
+		return
+	}
+	ec := utils.EncryptedRemoteControlCmd{
+		IV: b85_encode(iv), Tag: b85_encode(tag), Pubkey: b85_encode(pubkey), Encrypted: b85_encode(ciphertext)}
+	if encryption_protocol != "1" {
+		ec.EncProto = encryption_protocol
+	}
+	ans, err = json.Marshal(ec)
+	return
+}
+
 // }}}