mirror of
https://github.com/kovidgoyal/kitty.git
synced 2026-09-01 14:53:30 +00:00
remote_file kitten: port ask UI to Go, wrap and shrink Python to result-handler shim
Adds kittens/remote_file/main.go (ask menu, hostname-mismatch prompt, save-path prompt with tab completion, edit/open/save actions) on top of the ControlMaster/ssh helpers from the previous two commits, registers it in tools/cmd/tool/main.go, and adds remote_file to shell-integration/ssh/kitty's wrapped_kittens list so both `kitty +kitten remote_file` and ssh-session invocations exec the Go binary directly. main.py is shrunk to option_text (kept importable by kitty/cli_stub.py and used for --doc/--help generation), is_ssh_kitten_sentinel (kept importable by kitty/window.py), and the handle_result shim boss.py runs in-process to open the returned file. Also drops the dead `-h` alias on --hostname: Python's CLI layer always special-cases -h/--help ahead of user options, so it never worked as a hostname shorthand, but the Go CLI layer hard-errors on the collision; removing it is a no-op for existing behavior (boss.py only ever passes --hostname) and unblocks `kitten remote-file --help`. Also adds the same reset_terminal() calls Python makes between UI phases (ask menu -> action, hostname-mismatch prompt, overwrite prompt, $EDITOR handoff) so the Go and Python screens match. Manually verified parity against Python (ask menu, cancel, open/edit/save download+upload round-trips, hostname-mismatch Y/N, save-path prompt and overwrite O/A/R/N) using a fake-ssh test double, and confirmed post-wrap that `kitty +kitten remote_file` execs the Go kitten binary.
This commit is contained in:
OBrutus
parent
36f93e72a7
commit
a0978a74a2
5 changed files with 418 additions and 323 deletions
|
|
@ -167,6 +167,10 @@ func edit_loop(master *ControlMaster, editor []string) error {
|
|||
return err
|
||||
}
|
||||
mtime := st.ModTime()
|
||||
// Mirrors main.py handle_action's edit branch: reset the terminal (clear
|
||||
// whatever the ask-menu/hostname-prompt phases drew) right before handing
|
||||
// the screen to the user's $EDITOR, and again once it exits.
|
||||
reset_terminal()
|
||||
argv := append(append([]string{}, editor...), master.Dest)
|
||||
cmd := exec.Command(argv[0], argv[1:]...)
|
||||
cmd.Stdin, cmd.Stdout, cmd.Stderr = os.Stdin, os.Stdout, os.Stderr
|
||||
|
|
@ -178,6 +182,7 @@ func edit_loop(master *ControlMaster, editor []string) error {
|
|||
for {
|
||||
select {
|
||||
case <-done:
|
||||
reset_terminal()
|
||||
if master.IsAlive() {
|
||||
if err := master.Upload(false); err != nil {
|
||||
return fmt.Errorf("failed to upload %s: %w", master.remote_path, err)
|
||||
|
|
|
|||
396
kittens/remote_file/main.go
Normal file
396
kittens/remote_file/main.go
Normal file
|
|
@ -0,0 +1,396 @@
|
|||
// License: GPLv3 Copyright: 2026, Kovid Goyal, <kovid at kovidgoyal.net>
|
||||
|
||||
package remote_file
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"io"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
|
||||
"github.com/kovidgoyal/kitty/kittens/choose_files"
|
||||
"github.com/kovidgoyal/kitty/tools/cli"
|
||||
"github.com/kovidgoyal/kitty/tools/cli/markup"
|
||||
"github.com/kovidgoyal/kitty/tools/tui"
|
||||
"github.com/kovidgoyal/kitty/tools/tui/loop"
|
||||
"github.com/kovidgoyal/kitty/tools/tui/readline"
|
||||
"github.com/kovidgoyal/kitty/tools/utils"
|
||||
)
|
||||
|
||||
var _ = fmt.Print
|
||||
|
||||
// reset_terminal performs a full terminal reset (RIS), exactly mirroring
|
||||
// kittens/tui/operations.py:reset_terminal()'s escape sequence
|
||||
// ('\033]\033\\\033c'). Python's remote_file calls this between UI phases
|
||||
// (after the ask menu, around hostname-mismatch prompts, around the
|
||||
// overwrite-prompt, and around the $EDITOR invocation) to clear whatever was
|
||||
// drawn by the previous phase/loop before drawing the next; we mirror those
|
||||
// call sites exactly so the two implementations look the same on screen.
|
||||
func reset_terminal() {
|
||||
fmt.Print("\x1b]\x1b\\\x1bc")
|
||||
}
|
||||
|
||||
// get_key_press shows `draw` and waits for one of the runes in allowed.
|
||||
// Returns deflt on Esc/Ctrl+C. Mirrors kittens/tui/utils.py:get_key_press.
|
||||
func get_key_press(draw func(lp *loop.Loop, ctx *markup.Context), allowed, deflt string) (ans string, err error) {
|
||||
lp, err := loop.NewForSimpleInteraction()
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
ctx := markup.New(true)
|
||||
ans = deflt
|
||||
lp.OnInitialize = func() (string, error) {
|
||||
lp.SetCursorVisible(false)
|
||||
draw(lp, ctx)
|
||||
return "", nil
|
||||
}
|
||||
lp.OnText = func(text string, from_key_event, in_bracketed_paste bool) error {
|
||||
text = strings.ToLower(text)
|
||||
if allowed != "" && strings.Contains(allowed, text) {
|
||||
ans = text
|
||||
lp.Quit(0)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
lp.OnKeyEvent = func(e *loop.KeyEvent) error {
|
||||
if e.MatchesPressOrRepeat("esc") || e.MatchesPressOrRepeat("ctrl+c") {
|
||||
e.Handled = true
|
||||
lp.Quit(1)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
err = lp.Run()
|
||||
lp.KillIfSignalled()
|
||||
return ans, err
|
||||
}
|
||||
|
||||
// wait_for_any_key blocks until the user presses any single key. Unlike
|
||||
// get_key_press it does not filter by an allowed set, matching Python's
|
||||
// show_error which breaks on the first byte read regardless of its value.
|
||||
func wait_for_any_key() error {
|
||||
lp, err := loop.NewForSimpleInteraction()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
lp.OnInitialize = func() (string, error) {
|
||||
lp.SetCursorVisible(false)
|
||||
return "", nil
|
||||
}
|
||||
lp.OnText = func(text string, from_key_event, in_bracketed_paste bool) error {
|
||||
lp.Quit(0)
|
||||
return nil
|
||||
}
|
||||
lp.OnKeyEvent = func(e *loop.KeyEvent) error {
|
||||
e.Handled = true
|
||||
lp.Quit(0)
|
||||
return nil
|
||||
}
|
||||
err = lp.Run()
|
||||
lp.KillIfSignalled()
|
||||
return err
|
||||
}
|
||||
|
||||
func show_error(msg string) {
|
||||
ctx := markup.New(true)
|
||||
fmt.Fprintln(os.Stderr, ctx.Err(msg))
|
||||
fmt.Println()
|
||||
fmt.Println("Press any key to quit")
|
||||
_ = wait_for_any_key()
|
||||
}
|
||||
|
||||
func ask_action(opts *Options) (string, error) {
|
||||
draw := func(lp *loop.Loop, ctx *markup.Context) {
|
||||
hostname := opts.Hostname
|
||||
if hostname == "" {
|
||||
hostname = "unknown"
|
||||
}
|
||||
lp.Println("What would you like to do with the remote file on " + ctx.Magenta(hostname) + ":")
|
||||
lp.Println(ctx.Yellow(opts.Path))
|
||||
lp.Println()
|
||||
lp.Println(ctx.Green("E") + "dit the file")
|
||||
lp.Println(lp.SprintStyled("dim", "The file will be downloaded and opened in an editor. Any changes you save will be automatically sent back to the remote machine"))
|
||||
lp.Println()
|
||||
lp.Println(ctx.Green("O") + "pen the file")
|
||||
lp.Println(lp.SprintStyled("dim", "The file will be downloaded and opened by the default open program"))
|
||||
lp.Println()
|
||||
lp.Println(ctx.Green("S") + "ave the file")
|
||||
lp.Println(lp.SprintStyled("dim", "The file will be downloaded to a destination you select"))
|
||||
lp.Println()
|
||||
lp.Println(ctx.Green("C") + "ancel")
|
||||
}
|
||||
response, err := get_key_press(draw, "ceos", "c")
|
||||
if err != nil {
|
||||
return "cancel", err
|
||||
}
|
||||
return map[string]string{"e": "edit", "o": "open", "s": "save"}[response], nil
|
||||
}
|
||||
|
||||
// check_hostname_matches prompts the user when the remote hostname does not
|
||||
// match the hyperlink hostname; mirrors main.py ControlMaster.check_hostname_matches.
|
||||
func check_hostname_matches(master *ControlMaster, cli_hostname string) (bool, error) {
|
||||
if master.conn.IsSSHKitten {
|
||||
return true, nil
|
||||
}
|
||||
q := master.remote_hostname()
|
||||
if q == "" || hostname_matches(cli_hostname, q) {
|
||||
return true, nil
|
||||
}
|
||||
reset_terminal()
|
||||
draw := func(lp *loop.Loop, ctx *markup.Context) {
|
||||
lp.Println("The remote hostname " + ctx.Green(q) + " does not match the")
|
||||
lp.Println("hostname in the hyperlink " + ctx.Err(cli_hostname))
|
||||
lp.Println("This indicates that kitty has not connected to the correct remote machine.")
|
||||
lp.Println("This can happen, for example, when using nested SSH sessions.")
|
||||
lp.Printf("The hostname kitty used to connect was: %s", ctx.Yellow(master.conn.Hostname))
|
||||
if master.conn.Port > 0 {
|
||||
lp.Printf(" with port: %d", master.conn.Port)
|
||||
}
|
||||
lp.Println()
|
||||
lp.Println()
|
||||
lp.Println("Do you want to continue anyway?")
|
||||
lp.Println(ctx.Green("Y") + "es\t" + ctx.Err("N") + "o")
|
||||
}
|
||||
response, err := get_key_press(draw, "yn", "n")
|
||||
reset_terminal()
|
||||
return response == "y", err
|
||||
}
|
||||
|
||||
// get_save_path reads a single line of input for the destination path,
|
||||
// with filename tab-completion, mirroring kittens/tui/path_completer.py's
|
||||
// PathCompleter/get_path. aborted is true on Ctrl+C/EOF, matching Python's
|
||||
// catch of KeyboardInterrupt/EOFError in save_as.
|
||||
func get_save_path(prompt string) (result string, aborted bool, err error) {
|
||||
lp, err := loop.New(loop.NoAlternateScreen, loop.NoRestoreColors)
|
||||
if err != nil {
|
||||
return "", false, err
|
||||
}
|
||||
rl := readline.New(lp, readline.RlInit{Prompt: prompt, Completer: choose_files.FilePromptCompleter(nil)})
|
||||
lp.OnInitialize = func() (string, error) { rl.Start(); return "", nil }
|
||||
lp.OnFinalize = func() string { rl.End(); return "" }
|
||||
lp.OnResumeFromStop = func() error { rl.Start(); return nil }
|
||||
lp.OnResize = rl.OnResize
|
||||
lp.OnKeyEvent = func(event *loop.KeyEvent) error {
|
||||
if event.MatchesPressOrRepeat("ctrl+c") {
|
||||
aborted = true
|
||||
lp.Quit(0)
|
||||
return nil
|
||||
}
|
||||
kerr := rl.OnKeyEvent(event)
|
||||
if kerr != nil {
|
||||
if kerr == io.EOF {
|
||||
aborted = true
|
||||
lp.Quit(0)
|
||||
return nil
|
||||
}
|
||||
if kerr == readline.ErrAcceptInput {
|
||||
result = rl.AllText()
|
||||
lp.Quit(0)
|
||||
return nil
|
||||
}
|
||||
return kerr
|
||||
}
|
||||
if event.Handled {
|
||||
rl.Redraw()
|
||||
}
|
||||
return nil
|
||||
}
|
||||
lp.OnText = func(text string, from_key_event, in_bracketed_paste bool) error {
|
||||
terr := rl.OnText(text, from_key_event, in_bracketed_paste)
|
||||
if terr == nil {
|
||||
rl.Redraw()
|
||||
}
|
||||
return terr
|
||||
}
|
||||
err = lp.Run()
|
||||
rl.Shutdown()
|
||||
if err != nil {
|
||||
return "", false, err
|
||||
}
|
||||
if ds := lp.DeathSignalName(); ds != "" {
|
||||
return "", false, fmt.Errorf("killed by signal: %s", ds)
|
||||
}
|
||||
return result, aborted, nil
|
||||
}
|
||||
|
||||
func master_show_error(m *ControlMaster, msg string) {
|
||||
if m.LastErrorLog != "" {
|
||||
fmt.Fprintln(os.Stderr, m.LastErrorLog)
|
||||
m.LastErrorLog = ""
|
||||
}
|
||||
show_error(msg)
|
||||
}
|
||||
|
||||
// save_as mirrors main.py:save_as. hostname is the CLI --hostname value used
|
||||
// by check_hostname_matches.
|
||||
func save_as(conn *SSHConnectionData, remote_path, hostname string) error {
|
||||
ddir := utils.CacheDir()
|
||||
if err := os.MkdirAll(ddir, 0o755); err != nil {
|
||||
return err
|
||||
}
|
||||
last_used_store := filepath.Join(ddir, "remote-file-last-used.txt")
|
||||
last_used_path := os.TempDir()
|
||||
if b, err := os.ReadFile(last_used_store); err == nil {
|
||||
last_used_path = string(b)
|
||||
}
|
||||
last_used_file := filepath.Join(last_used_path, filepath.Base(remote_path))
|
||||
ctx := markup.New(true)
|
||||
fmt.Println("Where do you want to save the file? Leaving it blank will save it as:", ctx.Yellow(last_used_file))
|
||||
cwd, _ := os.Getwd()
|
||||
fmt.Println("Relative paths will be resolved from:", ctx.Bold(cwd))
|
||||
fmt.Println()
|
||||
|
||||
dest, aborted, err := get_save_path("> ")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if aborted {
|
||||
return nil
|
||||
}
|
||||
if dest != "" {
|
||||
dest = utils.Expanduser(os.ExpandEnv(dest))
|
||||
if st, err := os.Stat(dest); err == nil && st.IsDir() {
|
||||
dest = filepath.Join(dest, filepath.Base(remote_path))
|
||||
}
|
||||
if abs, err := filepath.Abs(dest); err == nil {
|
||||
_ = os.WriteFile(last_used_store, []byte(filepath.Dir(abs)), 0o644)
|
||||
dest = abs
|
||||
}
|
||||
} else {
|
||||
dest = last_used_file
|
||||
}
|
||||
if _, err := os.Stat(dest); err == nil {
|
||||
reset_terminal()
|
||||
draw := func(lp *loop.Loop, mctx *markup.Context) {
|
||||
lp.Println("The file " + mctx.Yellow(dest) + " already exists. What would you like to do?")
|
||||
lp.Println(mctx.Green("O") + "verwrite " + mctx.Green("A") + "bort Auto " + mctx.Green("R") + "ename " + mctx.Green("N") + "ew name")
|
||||
}
|
||||
response, err := get_key_press(draw, "anor", "a")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
switch response {
|
||||
case "a":
|
||||
return nil
|
||||
case "n":
|
||||
reset_terminal()
|
||||
return save_as(conn, remote_path, hostname)
|
||||
case "r":
|
||||
dest = auto_rename_dest(dest, func(p string) bool { _, err := os.Stat(p); return err == nil })
|
||||
}
|
||||
}
|
||||
if d := filepath.Dir(dest); d != "" {
|
||||
if err := os.MkdirAll(d, 0o755); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
master := new_control_master(conn, remote_path, dest)
|
||||
if err := master.Start(); err != nil {
|
||||
return err
|
||||
}
|
||||
defer master.Close()
|
||||
ok, err := check_hostname_matches(master, hostname)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if !ok {
|
||||
return nil
|
||||
}
|
||||
if err := master.Download(); err != nil {
|
||||
master_show_error(master, "Failed to copy file from remote machine")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// handle_action returns the path to open locally (for the "open" action) or "".
|
||||
func handle_action(action string, opts *Options) (result string, err error) {
|
||||
conn, err := parse_conn_data(opts.SshConnectionData)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
remote_path := opts.Path
|
||||
switch action {
|
||||
case "open":
|
||||
fmt.Println("Opening", opts.Path, "from", opts.Hostname)
|
||||
tdir, err := os.MkdirTemp("", "kitty-remote-file")
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
dest := filepath.Join(tdir, filepath.Base(remote_path))
|
||||
master := new_control_master(conn, remote_path, dest)
|
||||
if err := master.Start(); err != nil {
|
||||
return "", err
|
||||
}
|
||||
defer master.Close()
|
||||
ok, err := check_hostname_matches(master, opts.Hostname)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
if !ok {
|
||||
return "", nil
|
||||
}
|
||||
if err := master.Download(); err != nil {
|
||||
master_show_error(master, "Failed to copy file from remote machine")
|
||||
return "", nil
|
||||
}
|
||||
return dest, nil
|
||||
case "edit":
|
||||
fmt.Println("Editing", opts.Path, "from", opts.Hostname)
|
||||
master := new_control_master(conn, remote_path, "")
|
||||
if err := master.Start(); err != nil {
|
||||
return "", err
|
||||
}
|
||||
defer master.Close()
|
||||
ok, err := check_hostname_matches(master, opts.Hostname)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
if !ok {
|
||||
return "", nil
|
||||
}
|
||||
if err := master.Download(); err != nil {
|
||||
master_show_error(master, "Failed to download "+remote_path)
|
||||
return "", nil
|
||||
}
|
||||
if err := edit_loop(master, get_editor()); err != nil {
|
||||
master_show_error(master, err.Error())
|
||||
}
|
||||
case "save":
|
||||
fmt.Println("Saving", opts.Path, "from", opts.Hostname)
|
||||
if err := save_as(conn, remote_path, opts.Hostname); err != nil {
|
||||
return "", err
|
||||
}
|
||||
}
|
||||
return "", nil
|
||||
}
|
||||
|
||||
func main(cmd *cli.Command, opts *Options, args []string) (rc int, err error) {
|
||||
action := opts.Mode
|
||||
if action == "ask" {
|
||||
action, err = ask_action(opts)
|
||||
reset_terminal()
|
||||
if err != nil {
|
||||
return 1, err
|
||||
}
|
||||
}
|
||||
if action == "" || action == "cancel" {
|
||||
return 0, nil
|
||||
}
|
||||
result, err := handle_action(action, opts)
|
||||
if err != nil {
|
||||
show_error(err.Error())
|
||||
return 1, nil
|
||||
}
|
||||
if result != "" {
|
||||
serialized, err := tui.KittenOutputSerializer()(result)
|
||||
if err != nil {
|
||||
return 1, err
|
||||
}
|
||||
os.Stdout.WriteString(serialized)
|
||||
}
|
||||
return 0, nil
|
||||
}
|
||||
|
||||
func EntryPoint(parent *cli.Command) {
|
||||
create_cmd(parent, main)
|
||||
}
|
||||
|
|
@ -2,33 +2,18 @@
|
|||
# License: GPLv3 Copyright: 2020, Kovid Goyal <kovid at kovidgoyal.net>
|
||||
|
||||
|
||||
import json
|
||||
import os
|
||||
import shlex
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
import time
|
||||
from typing import Any, Optional
|
||||
|
||||
from kitty.cli import parse_args
|
||||
from kitty.cli_stub import RemoteFileCLIOptions
|
||||
from kitty.constants import cache_dir
|
||||
from kitty.typing_compat import BossType
|
||||
from kitty.utils import SSHConnectionData, command_for_open, get_editor, open_cmd
|
||||
from kitty.utils import command_for_open, open_cmd
|
||||
|
||||
from ..tui.handler import result_handler
|
||||
from ..tui.operations import faint, raw_mode, reset_terminal, styled
|
||||
from ..tui.utils import get_key_press
|
||||
|
||||
# Must match kittens/remote_file/ssh.go is_ssh_kitten_sentinel. Kept importable
|
||||
# here because kitty/window.py:handle_remote_file imports it directly.
|
||||
is_ssh_kitten_sentinel = '!#*&$#($ssh-kitten)(##$'
|
||||
|
||||
|
||||
def key(x: str) -> str:
|
||||
return styled(x, bold=True, fg='green')
|
||||
|
||||
|
||||
def option_text() -> str:
|
||||
return '''\
|
||||
--mode -m
|
||||
|
|
@ -41,7 +26,7 @@ Which mode to operate in.
|
|||
Path to the remote file.
|
||||
|
||||
|
||||
--hostname -h
|
||||
--hostname
|
||||
Hostname of the remote host.
|
||||
|
||||
|
||||
|
|
@ -50,312 +35,12 @@ The data used to connect over ssh.
|
|||
'''
|
||||
|
||||
|
||||
def show_error(msg: str) -> None:
|
||||
print(styled(msg, fg='red'), file=sys.stderr)
|
||||
print()
|
||||
print('Press any key to quit', flush=True)
|
||||
with raw_mode():
|
||||
while True:
|
||||
try:
|
||||
q = sys.stdin.buffer.read(1)
|
||||
if q:
|
||||
break
|
||||
except (KeyboardInterrupt, EOFError):
|
||||
break
|
||||
|
||||
|
||||
def ask_action(opts: RemoteFileCLIOptions) -> str:
|
||||
print('What would you like to do with the remote file on {}:'.format(styled(opts.hostname or 'unknown', bold=True, fg='magenta')))
|
||||
print(styled(opts.path or '', fg='yellow', fg_intense=True))
|
||||
print()
|
||||
|
||||
def help_text(x: str) -> str:
|
||||
return faint(x)
|
||||
|
||||
print('{}dit the file'.format(key('E')))
|
||||
print(help_text('The file will be downloaded and opened in an editor. Any changes you save will'
|
||||
' be automatically sent back to the remote machine'))
|
||||
print()
|
||||
|
||||
print('{}pen the file'.format(key('O')))
|
||||
print(help_text('The file will be downloaded and opened by the default open program'))
|
||||
print()
|
||||
|
||||
print('{}ave the file'.format(key('S')))
|
||||
print(help_text('The file will be downloaded to a destination you select'))
|
||||
print()
|
||||
|
||||
print('{}ancel'.format(key('C')))
|
||||
print()
|
||||
|
||||
sys.stdout.flush()
|
||||
response = get_key_press('ceos', 'c')
|
||||
return {'e': 'edit', 'o': 'open', 's': 'save'}.get(response, 'cancel')
|
||||
|
||||
|
||||
def hostname_matches(from_hyperlink: str, actual: str) -> bool:
|
||||
if from_hyperlink == actual:
|
||||
return True
|
||||
if from_hyperlink.partition('.')[0] == actual.partition('.')[0]:
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
class ControlMaster:
|
||||
|
||||
def __init__(self, conn_data: SSHConnectionData, remote_path: str, cli_opts: RemoteFileCLIOptions, dest: str = ''):
|
||||
self.conn_data = conn_data
|
||||
self.cli_opts = cli_opts
|
||||
self.remote_path = remote_path
|
||||
self.dest = dest
|
||||
self.tdir = ''
|
||||
self.last_error_log = ''
|
||||
self.cmd_prefix = cmd = [
|
||||
conn_data.binary, '-o', f'ControlPath=~/.ssh/kitty-rf-{os.getpid()}-%C',
|
||||
'-o', 'TCPKeepAlive=yes', '-o', 'ControlPersist=yes'
|
||||
]
|
||||
self.is_ssh_kitten = conn_data.binary is is_ssh_kitten_sentinel
|
||||
if self.is_ssh_kitten:
|
||||
del cmd[:]
|
||||
self.batch_cmd_prefix = cmd
|
||||
sk_cmdline = json.loads(conn_data.identity_file)
|
||||
while '-t' in sk_cmdline:
|
||||
sk_cmdline.remove('-t')
|
||||
cmd.extend(sk_cmdline[:-2])
|
||||
else:
|
||||
if conn_data.port:
|
||||
cmd.extend(['-p', str(conn_data.port)])
|
||||
if conn_data.identity_file:
|
||||
cmd.extend(['-i', conn_data.identity_file])
|
||||
self.batch_cmd_prefix = cmd + ['-o', 'BatchMode=yes']
|
||||
|
||||
def check_call(self, cmd: list[str]) -> None:
|
||||
p = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, stdin=subprocess.DEVNULL)
|
||||
stdout = p.communicate()[0]
|
||||
if p.wait() != 0:
|
||||
out = stdout.decode('utf-8', 'replace')
|
||||
raise Exception(f'The ssh command: {shlex.join(cmd)} failed with exit code {p.returncode} and output: {out}')
|
||||
|
||||
def __enter__(self) -> 'ControlMaster':
|
||||
if not self.is_ssh_kitten:
|
||||
self.check_call(
|
||||
self.cmd_prefix + ['-o', 'ControlMaster=auto', '-fN', self.conn_data.hostname])
|
||||
self.check_call(
|
||||
self.batch_cmd_prefix + ['-O', 'check', self.conn_data.hostname])
|
||||
if not self.dest:
|
||||
self.tdir = tempfile.mkdtemp()
|
||||
self.dest = os.path.join(self.tdir, os.path.basename(self.remote_path))
|
||||
return self
|
||||
|
||||
def __exit__(self, *a: Any) -> None:
|
||||
if not self.is_ssh_kitten:
|
||||
subprocess.Popen(
|
||||
self.batch_cmd_prefix + ['-O', 'exit', self.conn_data.hostname],
|
||||
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, stdin=subprocess.DEVNULL
|
||||
).wait()
|
||||
if self.tdir:
|
||||
shutil.rmtree(self.tdir)
|
||||
|
||||
@property
|
||||
def is_alive(self) -> bool:
|
||||
if self.is_ssh_kitten:
|
||||
return True
|
||||
return subprocess.Popen(
|
||||
self.batch_cmd_prefix + ['-O', 'check', self.conn_data.hostname],
|
||||
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, stdin=subprocess.DEVNULL
|
||||
).wait() == 0
|
||||
|
||||
def check_hostname_matches(self) -> bool:
|
||||
if self.is_ssh_kitten:
|
||||
return True
|
||||
cp = subprocess.run(self.batch_cmd_prefix + [self.conn_data.hostname, 'hostname', '-f'], stdout=subprocess.PIPE,
|
||||
stderr=subprocess.DEVNULL, stdin=subprocess.DEVNULL)
|
||||
if cp.returncode == 0:
|
||||
q = tuple(filter(None, cp.stdout.decode('utf-8').strip().splitlines()))[-1]
|
||||
if not hostname_matches(self.cli_opts.hostname or '', q):
|
||||
print(reset_terminal(), end='')
|
||||
print(f'The remote hostname {styled(q, fg="green")} does not match the')
|
||||
print(f'hostname in the hyperlink {styled(self.cli_opts.hostname or "", fg="red")}')
|
||||
print('This indicates that kitty has not connected to the correct remote machine.')
|
||||
print('This can happen, for example, when using nested SSH sessions.')
|
||||
print(f'The hostname kitty used to connect was: {styled(self.conn_data.hostname, fg="yellow")}', end='')
|
||||
if self.conn_data.port is not None:
|
||||
print(f' with port: {self.conn_data.port}')
|
||||
print()
|
||||
print()
|
||||
print('Do you want to continue anyway?')
|
||||
print(
|
||||
f'{styled("Y", fg="green")}es',
|
||||
f'{styled("N", fg="red")}o', sep='\t'
|
||||
)
|
||||
sys.stdout.flush()
|
||||
response = get_key_press('yn', 'n')
|
||||
print(reset_terminal(), end='')
|
||||
return response == 'y'
|
||||
return True
|
||||
|
||||
def show_error(self, msg: str) -> None:
|
||||
if self.last_error_log:
|
||||
print(self.last_error_log, file=sys.stderr)
|
||||
self.last_error_log = ''
|
||||
show_error(msg)
|
||||
|
||||
def download(self) -> bool:
|
||||
cmdline = self.batch_cmd_prefix + [self.conn_data.hostname, 'cat', shlex.quote(self.remote_path)]
|
||||
with open(self.dest, 'wb') as f:
|
||||
cp = subprocess.run(cmdline, stdout=f, stderr=subprocess.PIPE, stdin=subprocess.DEVNULL)
|
||||
if cp.returncode != 0:
|
||||
self.last_error_log = f'The command: {shlex.join(cmdline)} failed\n' + cp.stderr.decode()
|
||||
return False
|
||||
return True
|
||||
|
||||
def upload(self, suppress_output: bool = True) -> bool:
|
||||
cmd_prefix = self.cmd_prefix if suppress_output else self.batch_cmd_prefix
|
||||
cmd = cmd_prefix + [self.conn_data.hostname, 'cat', '>', shlex.quote(self.remote_path)]
|
||||
if not suppress_output:
|
||||
print(shlex.join(cmd))
|
||||
with open(self.dest, 'rb') as f:
|
||||
if suppress_output:
|
||||
cp = subprocess.run(cmd, stdin=f, capture_output=True)
|
||||
if cp.returncode == 0:
|
||||
return True
|
||||
self.last_error_log = f'The command: {shlex.join(cmd)} failed\n' + cp.stdout.decode()
|
||||
else:
|
||||
return subprocess.run(cmd, stdin=f).returncode == 0
|
||||
return False
|
||||
|
||||
|
||||
Result = Optional[str]
|
||||
|
||||
|
||||
def main(args: list[str]) -> Result:
|
||||
msg = 'Ask the user what to do with the remote file. For internal use by kitty, do not run it directly.'
|
||||
try:
|
||||
cli_opts, items = parse_args(args[1:], option_text, '', msg, 'kitty +kitten remote_file', result_class=RemoteFileCLIOptions)
|
||||
except SystemExit as e:
|
||||
if e.code != 0:
|
||||
print(e.args[0])
|
||||
input('Press Enter to quit')
|
||||
raise SystemExit(e.code)
|
||||
|
||||
try:
|
||||
action = ask_action(cli_opts)
|
||||
finally:
|
||||
print(reset_terminal(), end='', flush=True)
|
||||
try:
|
||||
return handle_action(action, cli_opts)
|
||||
except Exception:
|
||||
print(reset_terminal(), end='', flush=True)
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
show_error('Failed with unhandled exception')
|
||||
return None
|
||||
|
||||
|
||||
def save_as(conn_data: SSHConnectionData, remote_path: str, cli_opts: RemoteFileCLIOptions) -> None:
|
||||
ddir = cache_dir()
|
||||
os.makedirs(ddir, exist_ok=True)
|
||||
last_used_store_path = os.path.join(ddir, 'remote-file-last-used.txt')
|
||||
try:
|
||||
with open(last_used_store_path) as f:
|
||||
last_used_path = f.read()
|
||||
except FileNotFoundError:
|
||||
last_used_path = tempfile.gettempdir()
|
||||
last_used_file = os.path.join(last_used_path, os.path.basename(remote_path))
|
||||
print(
|
||||
'Where do you want to save the file? Leaving it blank will save it as:',
|
||||
styled(last_used_file, fg='yellow')
|
||||
)
|
||||
print('Relative paths will be resolved from:', styled(os.getcwd(), fg_intense=True, bold=True))
|
||||
print()
|
||||
from ..tui.path_completer import get_path
|
||||
try:
|
||||
dest = get_path()
|
||||
except (KeyboardInterrupt, EOFError):
|
||||
return
|
||||
if dest:
|
||||
dest = os.path.expandvars(os.path.expanduser(dest))
|
||||
if os.path.isdir(dest):
|
||||
dest = os.path.join(dest, os.path.basename(remote_path))
|
||||
with open(last_used_store_path, 'w') as f:
|
||||
f.write(os.path.dirname(os.path.abspath(dest)))
|
||||
else:
|
||||
dest = last_used_file
|
||||
if os.path.exists(dest):
|
||||
print(reset_terminal(), end='')
|
||||
print(f'The file {styled(dest, fg="yellow")} already exists. What would you like to do?')
|
||||
print(f'{key("O")}verwrite {key("A")}bort Auto {key("R")}ename {key("N")}ew name')
|
||||
response = get_key_press('anor', 'a')
|
||||
if response == 'a':
|
||||
return
|
||||
if response == 'n':
|
||||
print(reset_terminal(), end='')
|
||||
return save_as(conn_data, remote_path, cli_opts)
|
||||
|
||||
if response == 'r':
|
||||
q = dest
|
||||
c = 0
|
||||
while os.path.exists(q):
|
||||
c += 1
|
||||
b, ext = os.path.splitext(dest)
|
||||
q = f'{b}-{c}{ext}'
|
||||
dest = q
|
||||
if os.path.dirname(dest):
|
||||
os.makedirs(os.path.dirname(dest), exist_ok=True)
|
||||
with ControlMaster(conn_data, remote_path, cli_opts, dest=dest) as master:
|
||||
if master.check_hostname_matches():
|
||||
if not master.download():
|
||||
master.show_error('Failed to copy file from remote machine')
|
||||
|
||||
|
||||
def handle_action(action: str, cli_opts: RemoteFileCLIOptions) -> Result:
|
||||
cli_data = json.loads(cli_opts.ssh_connection_data or '')
|
||||
if cli_data and cli_data[0] == is_ssh_kitten_sentinel:
|
||||
conn_data = SSHConnectionData(is_ssh_kitten_sentinel, cli_data[-1], -1, identity_file=json.dumps(cli_data[1:]))
|
||||
else:
|
||||
conn_data = SSHConnectionData(*cli_data)
|
||||
remote_path = cli_opts.path or ''
|
||||
if action == 'open':
|
||||
print('Opening', cli_opts.path, 'from', cli_opts.hostname)
|
||||
dest = os.path.join(tempfile.mkdtemp(), os.path.basename(remote_path))
|
||||
with ControlMaster(conn_data, remote_path, cli_opts, dest=dest) as master:
|
||||
if master.check_hostname_matches():
|
||||
if master.download():
|
||||
return dest
|
||||
master.show_error('Failed to copy file from remote machine')
|
||||
elif action == 'edit':
|
||||
print('Editing', cli_opts.path, 'from', cli_opts.hostname)
|
||||
editor = get_editor()
|
||||
with ControlMaster(conn_data, remote_path, cli_opts) as master:
|
||||
if not master.check_hostname_matches():
|
||||
return None
|
||||
if not master.download():
|
||||
master.show_error(f'Failed to download {remote_path}')
|
||||
return None
|
||||
mtime = os.path.getmtime(master.dest)
|
||||
print(reset_terminal(), end='', flush=True)
|
||||
editor_process = subprocess.Popen(editor + [master.dest])
|
||||
while editor_process.poll() is None:
|
||||
time.sleep(0.1)
|
||||
newmtime = os.path.getmtime(master.dest)
|
||||
if newmtime > mtime:
|
||||
mtime = newmtime
|
||||
if master.is_alive:
|
||||
master.upload()
|
||||
print(reset_terminal(), end='', flush=True)
|
||||
if master.is_alive:
|
||||
if not master.upload(suppress_output=False):
|
||||
master.show_error(f'Failed to upload {remote_path}')
|
||||
else:
|
||||
master.show_error(f'Failed to upload {remote_path}, SSH master process died')
|
||||
elif action == 'save':
|
||||
print('Saving', cli_opts.path, 'from', cli_opts.hostname)
|
||||
save_as(conn_data, remote_path, cli_opts)
|
||||
return None
|
||||
def main(args: list[str]) -> None:
|
||||
raise SystemExit('This should be run as kitten remote_file')
|
||||
|
||||
|
||||
@result_handler()
|
||||
def handle_result(args: list[str], data: Result, target_window_id: int, boss: BossType) -> None:
|
||||
def handle_result(args: list[str], data: str | None, target_window_id: int, boss: BossType) -> None:
|
||||
if data:
|
||||
from kitty.fast_data_types import get_options
|
||||
cmd = command_for_open(get_options().open_url_with)
|
||||
|
|
@ -364,3 +49,9 @@ def handle_result(args: list[str], data: Result, target_window_id: int, boss: Bo
|
|||
|
||||
if __name__ == '__main__':
|
||||
main(sys.argv)
|
||||
elif __name__ == '__doc__':
|
||||
cd = sys.cli_docs # type: ignore
|
||||
cd['usage'] = ''
|
||||
cd['options'] = option_text
|
||||
cd['help_text'] = 'Ask the user what to do with the remote file. For internal use by kitty, do not run it directly.'
|
||||
cd['short_desc'] = 'Handle remote files'
|
||||
|
|
|
|||
|
|
@ -24,7 +24,7 @@ exec_kitty() {
|
|||
|
||||
|
||||
is_wrapped_kitten() {
|
||||
wrapped_kittens="clipboard icat hyperlinked_grep ask hints unicode_input ssh themes diff show_key transfer query_terminal choose-files command-palette resize_window broadcast"
|
||||
wrapped_kittens="clipboard icat hyperlinked_grep ask hints unicode_input ssh themes diff show_key transfer query_terminal choose-files command-palette resize_window broadcast remote_file"
|
||||
[ -n "$1" ] && {
|
||||
case " $wrapped_kittens " in
|
||||
*" $1 "*) printf "%s" "$1" ;;
|
||||
|
|
|
|||
|
|
@ -21,6 +21,7 @@ import (
|
|||
"github.com/kovidgoyal/kitty/kittens/panel"
|
||||
"github.com/kovidgoyal/kitty/kittens/query_terminal"
|
||||
"github.com/kovidgoyal/kitty/kittens/quick_access_terminal"
|
||||
"github.com/kovidgoyal/kitty/kittens/remote_file"
|
||||
"github.com/kovidgoyal/kitty/kittens/resize_window"
|
||||
"github.com/kovidgoyal/kitty/kittens/show_key"
|
||||
"github.com/kovidgoyal/kitty/kittens/ssh"
|
||||
|
|
@ -72,6 +73,8 @@ func KittyToolEntryPoints(root *cli.Command) {
|
|||
resize_window.EntryPoint(root)
|
||||
// broadcast
|
||||
broadcast.EntryPoint(root)
|
||||
// remote_file
|
||||
remote_file.EntryPoint(root)
|
||||
// unicode_input
|
||||
unicode_input.EntryPoint(root)
|
||||
// show_key
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue