lpra/internal/agent/sshfs.go

package agent

// Files on an SSH host, for the panel's file panes: the agent keeps one SSH
// connection per pane and runs plain commands there - stat, cat, mv, rm - as
// the tar transfer does. No SFTP is needed on the host. A connection that
// dropped is opened again once with what it was opened with; one left alone
// for sshfsIdle is closed.

import (
	"archive/tar"
	"bytes"
	"context"
	"crypto/rand"
	"encoding/base64"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net/http"
	"path"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"time"
	"unicode/utf8"

	"golang.org/x/crypto/ssh"

	"github.com/x-grk/longpolling-remote-access/internal/proto"
)

const sshfsIdle = 10 * time.Minute

type sshfsConn struct {
	mu      sync.Mutex
	client  *ssh.Client
	target  proto.SSHTarget // kept in the agent's memory to reconnect once after a drop
	statFmt string          // "gnu", "bsd" or "ls"
	home    string
	last    time.Time
	busy    int // transfers under way: the sweep leaves the connection be
}

// hold marks the connection busy for a long transfer; the returned func ends
// it and starts the idle clock again.
func (c *sshfsConn) hold() func() {
	c.mu.Lock()
	c.busy++
	c.mu.Unlock()
	return func() {
		c.mu.Lock()
		c.busy--
		c.last = time.Now()
		c.mu.Unlock()
	}
}

// idle: nothing under way and unused for sshfsIdle by now.
func (c *sshfsConn) idle(now time.Time) bool {
	c.mu.Lock()
	defer c.mu.Unlock()
	return c.busy == 0 && now.Sub(c.last) > sshfsIdle
}

var (
	sshfsConns   sync.Map // id -> *sshfsConn
	sshfsJanitor sync.Once
)

// statCommand lists dir's entries as "type|size|mtime|mode|name" lines.
func statCommand(flavour, dir string) string {
	d := shq(strings.TrimSuffix(dir, "/") + "/")
	// Each line starts with LD for a link to a folder, so it opens like one.
	loop := "cd " + d + " && for f in .[!.]* ..?* *; do [ -e \"$f\" ] || [ -L \"$f\" ] || continue; " +
		"t=; [ -L \"$f\" ] && [ -d \"$f\" ] && t=LD; printf '%s|' \"$t\"; "
	switch flavour {
	case "bsd":
		return loop + "stat -f '%HT|%z|%m|%Sp|%N' -- \"$f\"; done"
	case "ls":
		return "cd " + d + " && LC_ALL=C ls -la"
	}
	return loop + "stat -c '%F|%s|%Y|%A|%n' -- \"$f\"; done"
}

// parseStatLine reads one line of statCommand's output.
func parseStatLine(line string) (proto.FSEntry, bool) {
	// The LD mark of statCommand: empty, or LD for a link to a folder.
	mark := ""
	if i := strings.IndexByte(line, '|'); i >= 0 && (line[:i] == "" || line[:i] == "LD") {
		mark, line = line[:i], line[i+1:]
	}
	p := strings.SplitN(line, "|", 5)
	if len(p) != 5 {
		return proto.FSEntry{}, false
	}
	size, err1 := strconv.ParseInt(p[1], 10, 64)
	mt, err2 := strconv.ParseInt(p[2], 10, 64)
	if err1 != nil || err2 != nil {
		return proto.FSEntry{}, false
	}
	t := strings.ToLower(p[0])
	return proto.FSEntry{Name: p[4], IsDir: strings.Contains(t, "directory") || mark == "LD", Size: size, MTime: mt, Mode: p[3],
		Link: strings.Contains(t, "link")}, true
}

// parseLsLine reads a line of "ls -la" for a host with no stat: no times.
func parseLsLine(line string) (proto.FSEntry, bool) {
	f := strings.Fields(line)
	if len(f) < 9 || len(f[0]) < 10 {
		return proto.FSEntry{}, false
	}
	size, err := strconv.ParseInt(f[4], 10, 64)
	if err != nil {
		return proto.FSEntry{}, false
	}
	// The name is what follows the eighth field, spaces and all.
	rest := line
	for i := 0; i < 8; i++ {
		rest = strings.TrimLeft(rest, " ")
		rest = rest[strings.IndexByte(rest, ' ')+1:]
	}
	name := strings.TrimLeft(rest, " ")
	if f[0][0] == 'l' {
		if i := strings.Index(name, " -> "); i >= 0 {
			name = name[:i]
		}
	}
	if name == "." || name == ".." {
		return proto.FSEntry{}, false
	}
	return proto.FSEntry{Name: name, IsDir: f[0][0] == 'd', Size: size, Mode: f[0]}, true
}

func randHex(n int) string {
	b := make([]byte, n)
	_, _ = rand.Read(b)
	return hex.EncodeToString(b)
}

// sshfsOpen connects for a file pane and tells the host's home.
func (a *Agent) sshfsOpen(ctx context.Context, t proto.SSHTarget) (any, error) {
	client, _, err := a.sshConnect(ctx, t, func(string, string) {})
	if err != nil {
		return nil, err
	}
	c := &sshfsConn{client: client, target: t, last: time.Now()}
	out, _ := sshRun(client, "stat -c %F / >/dev/null 2>&1 && echo GNU; stat -f %HT / >/dev/null 2>&1 && echo BSD; "+
		"command -v zip >/dev/null 2>&1 && echo ZIP; command -v unzip >/dev/null 2>&1 && echo UNZIP; "+
		"command -v python3 >/dev/null 2>&1 && echo PY; echo \"$HOME\"")
	lines := strings.Split(strings.TrimSpace(out), "\n")
	switch {
	case strings.Contains(out, "GNU"):
		c.statFmt = "gnu"
	case strings.Contains(out, "BSD"):
		c.statFmt = "bsd"
	default:
		c.statFmt = "ls"
	}
	c.home = strings.TrimSpace(lines[len(lines)-1])
	if !strings.HasPrefix(c.home, "/") {
		c.home = "/"
	}
	id := randHex(12)
	sshfsConns.Store(id, c)
	sshfsJanitor.Do(func() { go sshfsSweep() })
	// zip and unzip, where the host has them, for the pane's buttons.
	return map[string]any{"id": id, "home": c.home,
		"zip": hasLine(lines, "ZIP") || hasLine(lines, "PY"), "unzip": hasLine(lines, "UNZIP") || hasLine(lines, "PY")}, nil
}

func hasLine(lines []string, w string) bool {
	for _, l := range lines {
		if strings.TrimSpace(l) == w {
			return true
		}
	}
	return false
}

// sshfsSweep closes connections nobody used for sshfsIdle.
func sshfsSweep() {
	for range time.Tick(time.Minute) {
		sshfsConns.Range(func(k, v any) bool {
			if v.(*sshfsConn).idle(time.Now()) {
				sshfsClose(k.(string))
			}
			return true
		})
	}
}

func sshfsClose(id string) {
	if v, ok := sshfsConns.LoadAndDelete(id); ok {
		_ = v.(*sshfsConn).client.Close()
	}
}

func sshfsGet(id string) (*sshfsConn, error) {
	v, ok := sshfsConns.Load(id)
	if !ok {
		return nil, errors.New("подключение закрыто - откройте файлы по SSH снова")
	}
	c := v.(*sshfsConn)
	c.mu.Lock()
	c.last = time.Now()
	c.mu.Unlock()
	return c, nil
}

// session opens an SSH session on the connection; a connection that dropped
// is opened again once.
func (a *Agent) sshfsSession(c *sshfsConn) (*ssh.Session, error) {
	c.mu.Lock()
	client := c.client
	c.mu.Unlock()
	s, err := client.NewSession()
	if err == nil {
		return s, nil
	}
	ctx, cancel := context.WithTimeout(context.Background(), 40*time.Second)
	defer cancel()
	nc, _, cerr := a.sshConnect(ctx, c.target, func(string, string) {})
	if cerr != nil {
		return nil, fmt.Errorf("подключение к хосту потеряно: %v", cerr)
	}
	c.mu.Lock()
	_ = c.client.Close()
	c.client = nc
	c.mu.Unlock()
	return nc.NewSession()
}

// run runs cmd on the host with stdin, returning its output; a failure
// carries what the command said on stderr.
func (a *Agent) sshfsRun(c *sshfsConn, cmd string, stdin io.Reader) ([]byte, error) {
	s, err := a.sshfsSession(c)
	if err != nil {
		return nil, err
	}
	defer s.Close()
	var out, errb bytes.Buffer
	s.Stdout = &out
	s.Stderr = &limitBuf{b: &errb, max: 4096}
	if stdin != nil {
		s.Stdin = stdin
	}
	if err := s.Run(cmd); err != nil {
		msg := strings.TrimSpace(errb.String())
		if msg == "" {
			msg = err.Error()
		}
		if len(msg) > 300 {
			msg = msg[:300]
		}
		return out.Bytes(), errors.New(msg)
	}
	return out.Bytes(), nil
}

// sshfsTask runs one sshfs.* task.
func (a *Agent) sshfsTask(ctx context.Context, env proto.TaskEnvelope) (any, error) {
	var args proto.SSHFSArgs
	if err := json.Unmarshal(env.Args, &args); err != nil {
		return nil, err
	}
	switch env.Type {
	case proto.TaskSSHFSOpen:
		if args.SSH == nil {
			return nil, errors.New("не указан хост")
		}
		return a.sshfsOpen(ctx, *args.SSH)
	case proto.TaskSSHFSClose:
		sshfsClose(args.Conn)
		return nil, nil
	}
	c, err := sshfsGet(args.Conn)
	if err != nil {
		return nil, err
	}
	return a.sshfsOp(c, args.Op, args.Args)
}

func (a *Agent) sshfsOp(c *sshfsConn, op string, raw json.RawMessage) (any, error) {
	switch op {
	case "list":
		var p proto.FSPathArgs
		_ = json.Unmarshal(raw, &p)
		dir := p.Path
		if dir == "" || strings.HasPrefix(dir, "~") {
			dir = c.home
		}
		out, err := a.sshfsRun(c, statCommand(c.statFmt, dir), nil)
		if err != nil && len(out) == 0 {
			return nil, err
		}
		entries := []proto.FSEntry{}
		for _, line := range strings.Split(string(out), "\n") {
			var e proto.FSEntry
			var ok bool
			if c.statFmt == "ls" {
				e, ok = parseLsLine(line)
			} else {
				e, ok = parseStatLine(line)
			}
			if ok {
				entries = append(entries, e)
			}
		}
		return proto.FSListResult{Path: path.Clean(dir), Entries: entries}, nil
	case "read":
		var p proto.FSPathArgs
		_ = json.Unmarshal(raw, &p)
		size, _ := a.sshfsSize(c, p.Path)
		out, err := a.sshfsRun(c, "head -c "+strconv.Itoa(proto.FSReadCap+1)+" -- "+shq(p.Path), nil)
		if err != nil {
			return nil, err
		}
		truncated := len(out) > proto.FSReadCap
		if truncated {
			out = out[:proto.FSReadCap]
		}
		return proto.FSReadResult{Path: p.Path, Size: size, Content: base64.StdEncoding.EncodeToString(out),
			Truncated: truncated, Binary: !utf8.Valid(out) || bytes.IndexByte(out, 0) >= 0}, nil
	case "write":
		var p proto.FSWriteArgs
		_ = json.Unmarshal(raw, &p)
		data, err := base64.StdEncoding.DecodeString(p.Data)
		if err != nil {
			return nil, err
		}
		_, err = a.sshfsRun(c, writeCommand(p.Path), bytes.NewReader(data))
		return map[string]string{"path": p.Path}, err
	case "mkdir":
		var p proto.FSPathArgs
		_ = json.Unmarshal(raw, &p)
		_, err := a.sshfsRun(c, "mkdir -p -- "+shq(p.Path), nil)
		return map[string]string{"path": p.Path}, err
	case "delete", "deltree":
		var p proto.FSPathArgs
		_ = json.Unmarshal(raw, &p)
		if p.Path == "" || path.Clean(p.Path) == "/" {
			return nil, errors.New("отказ: удаление корня запрещено")
		}
		flag := "-f"
		if op == "deltree" {
			flag = "-rf"
		}
		_, err := a.sshfsRun(c, "rm "+flag+" -- "+shq(p.Path), nil)
		return map[string]string{"path": p.Path}, err
	case "move":
		var p proto.FSMoveArgs
		_ = json.Unmarshal(raw, &p)
		_, err := a.sshfsRun(c, "mv -- "+shq(p.Src)+" "+shq(p.Dst), nil)
		return map[string]string{"src": p.Src, "dst": p.Dst}, err
	case "copy":
		var p proto.FSCopyArgs
		_ = json.Unmarshal(raw, &p)
		return nil, a.sshfsCopy(c, p)
	case "zip":
		var p proto.FSZipArgs
		_ = json.Unmarshal(raw, &p)
		cmd, err := zipCommand(p.Paths, p.Dest)
		if err != nil {
			return nil, err
		}
		_, err = a.sshfsRun(c, cmd, nil)
		return map[string]string{"dest": p.Dest}, err
	case "unzip":
		var p proto.FSUnzipArgs
		_ = json.Unmarshal(raw, &p)
		_, err := a.sshfsRun(c, unzipCommand(p.Path, p.Dest), nil)
		return map[string]string{"dest": p.Dest}, err
	case "stat":
		var p proto.FSPathArgs
		_ = json.Unmarshal(raw, &p)
		return a.sshfsStat(c, p.Path)
	case "chmod":
		var p proto.FSChmodArgs
		_ = json.Unmarshal(raw, &p)
		rec := ""
		if p.Recursive {
			rec = "-R "
		}
		var cmds []string
		if p.Mode != nil {
			cmds = append(cmds, "chmod "+rec+fmt.Sprintf("%o", *p.Mode)+" -- "+shq(p.Path))
		}
		if p.Owner != "" || p.Group != "" {
			who := p.Owner
			if p.Group != "" {
				who += ":" + p.Group
			}
			cmds = append(cmds, "chown "+rec+shq(who)+" -- "+shq(p.Path))
		}
		if len(cmds) == 0 {
			return nil, nil
		}
		_, err := a.sshfsRun(c, strings.Join(cmds, " && "), nil)
		return map[string]string{"path": p.Path}, err
	}
	return nil, errors.New("неизвестная операция")
}

// zipCommand packs paths - items of one folder - into the archive dest, from
// that folder, under their own names. Paths are absolute and names start
// with "./", so none reads as an option.
func zipCommand(paths []string, dest string) (string, error) {
	if len(paths) == 0 {
		return "", errors.New("не выбраны файлы")
	}
	dir := path.Dir(path.Clean(paths[0]))
	names := make([]string, len(paths))
	for i, p := range paths {
		p = path.Clean(p)
		if path.Dir(p) != dir {
			return "", errors.New("в архив - только из одной папки")
		}
		names[i] = shq("./" + path.Base(p))
	}
	// zip where the host has it, else Python's zipfile, which most have.
	items := strings.Join(names, " ")
	return "cd -- " + shq(dir) + " && if command -v zip >/dev/null 2>&1; then zip -r -q -y " + shq(path.Clean(dest)) + " " + items + "; " +
		"else python3 -m zipfile -c " + shq(path.Clean(dest)) + " " + items + "; fi", nil
}

// unzipCommand unpacks the archive p into dest (its own folder if empty),
// replacing what is there, as on a machine.
func unzipCommand(p, dest string) string {
	if dest == "" {
		dest = path.Dir(path.Clean(p))
	}
	return "if command -v unzip >/dev/null 2>&1; then unzip -o -q " + shq(path.Clean(p)) + " -d " + shq(path.Clean(dest)) + "; " +
		"else python3 -m zipfile -e " + shq(path.Clean(p)) + " " + shq(path.Clean(dest)) + "; fi"
}

// writeCommand writes stdin to p through a file beside it, so a broken
// transfer leaves the old file as it was.
func writeCommand(p string) string {
	tmp := path.Join(path.Dir(p), "."+path.Base(p)+".lpra-"+randHex(4))
	return "cat > " + shq(tmp) + " && mv -f -- " + shq(tmp) + " " + shq(p) + " || { rm -f -- " + shq(tmp) + "; exit 1; }"
}

// sizeCommand tells the size of p - of its target, if a link: what cat
// reads.
func sizeCommand(statFmt, p string) string {
	switch statFmt {
	case "bsd":
		return "stat -L -f %z -- " + shq(p)
	case "ls":
		return "wc -c < " + shq(p)
	}
	return "stat -L -c %s -- " + shq(p)
}

func (a *Agent) sshfsSize(c *sshfsConn, p string) (int64, error) {
	out, err := a.sshfsRun(c, sizeCommand(c.statFmt, p), nil)
	if err != nil {
		return 0, err
	}
	return strconv.ParseInt(strings.TrimSpace(string(out)), 10, 64)
}

func (a *Agent) sshfsStat(c *sshfsConn, p string) (any, error) {
	cmd := "stat -c '%F|%s|%Y|%a|%u|%g|%U|%G' -- " + shq(p)
	if c.statFmt == "bsd" {
		cmd = "stat -f '%HT|%z|%m|%Lp|%u|%g|%Su|%Sg' -- " + shq(p)
	}
	out, err := a.sshfsRun(c, cmd, nil)
	if err != nil {
		return nil, err
	}
	f := strings.Split(strings.TrimSpace(string(out)), "|")
	if len(f) != 8 {
		return nil, errors.New("не удалось прочитать свойства")
	}
	size, _ := strconv.ParseInt(f[1], 10, 64)
	mt, _ := strconv.ParseInt(f[2], 10, 64)
	mode, _ := strconv.ParseUint(f[3], 8, 32)
	uid, _ := strconv.Atoi(f[4])
	gid, _ := strconv.Atoi(f[5])
	return proto.FSStatResult{Path: p, IsDir: strings.Contains(strings.ToLower(f[0]), "directory"), Size: size, MTime: mt,
		Mode: uint32(mode), UID: uid, GID: gid, Owner: f[6], Group: f[7], Users: []string{}, Groups: []string{}}, nil
}

// sshfsCopy copies or moves on the host, by the conflict rule for what is
// there already.
func (a *Agent) sshfsCopy(c *sshfsConn, p proto.FSCopyArgs) error {
	src, dst := shq(p.Src), shq(p.Dst)
	exists, _ := a.sshfsRun(c, "test -e "+dst+" -o -L "+dst+" && echo EXISTS", nil)
	there := strings.Contains(string(exists), "EXISTS")
	if there {
		switch p.Conflict {
		case "replace":
			if _, err := a.sshfsRun(c, "rm -rf -- "+dst, nil); err != nil {
				return err
			}
		case "skip":
			return nil
		default:
			return fmt.Errorf("%s уже существует", p.Dst)
		}
	}
	cmd := "cp -a -- " + src + " " + dst
	if p.Move {
		cmd = "mv -- " + src + " " + dst
	}
	_, err := a.sshfsRun(c, cmd, nil)
	return err
}

// doSSHFSSend sends a file of the host to the browser in pieces, as
// doFSSend does with a local one: cat's output, piece after piece.
func (a *Agent) doSSHFSSend(ctx context.Context, env proto.TaskEnvelope) proto.TaskResult {
	var args proto.FSXferArgs
	if err := json.Unmarshal(env.Args, &args); err != nil || args.Chunk <= 0 {
		return proto.TaskResult{Error: "плохой запрос"}
	}
	c, err := sshfsGet(args.Conn)
	if err != nil {
		return proto.TaskResult{Error: err.Error()}
	}
	defer c.hold()()
	size, err := a.sshfsSize(c, args.Path)
	if err != nil {
		return proto.TaskResult{Error: err.Error()}
	}
	if size > proto.FSFileCap {
		return proto.TaskResult{Error: "файл больше 512 МБ"}
	}
	s, err := a.sshfsSession(c)
	if err != nil {
		return proto.TaskResult{Error: err.Error()}
	}
	defer s.Close()
	out, _ := s.StdoutPipe()
	if err := s.Start("cat -- " + shq(args.Path)); err != nil {
		return proto.TaskResult{Error: err.Error()}
	}
	// Read in order from cat, posted to the server a few at a time.
	sizeHdr := strconv.FormatInt(size, 10)
	n := pieceCount(size, args.Chunk)
	pctx, cancel := context.WithCancel(ctx)
	defer cancel()
	var (
		wg    sync.WaitGroup
		once  sync.Once
		first error
	)
	fail := func(err error) { once.Do(func() { first = err; cancel() }) }
	sem := make(chan struct{}, xferWorkers)
	for i := 0; i < n && pctx.Err() == nil; i++ {
		buf := make([]byte, min(int64(args.Chunk), size-int64(i)*int64(args.Chunk)))
		if _, err := io.ReadFull(out, buf); err != nil {
			fail(errors.New("файл изменился во время чтения: " + err.Error()))
			break
		}
		select {
		case sem <- struct{}{}:
		case <-pctx.Done():
		}
		if pctx.Err() != nil {
			break
		}
		wg.Add(1)
		go func(i int, buf []byte) {
			defer func() { <-sem; wg.Done() }()
			err := a.xferRequest(pctx, func(ctx context.Context) (*http.Request, error) {
				req, err := http.NewRequestWithContext(ctx, http.MethodPost, a.xferURL(args.Xfer, i), bytes.NewReader(buf))
				if err == nil {
					req.Header.Set("X-Size", sizeHdr)
					req.Header.Set("Content-Type", "application/octet-stream")
				}
				return req, err
			}, func(*http.Response) error { return nil })
			if err != nil {
				fail(errors.New("передача прервана: " + err.Error()))
			}
		}(i, buf)
	}
	wg.Wait()
	if first == nil && ctx.Err() != nil {
		first = ctx.Err()
	}
	if first != nil {
		return proto.TaskResult{Error: first.Error()}
	}
	// All of it is read; the session closes (deferred) without waiting on a
	// cat that may have more to say.
	data, _ := json.Marshal(map[string]any{"path": args.Path, "size": size})
	return proto.TaskResult{OK: true, Data: data}
}

// doSSHFSPut writes a file the browser uploads onto the host, through a file
// beside it.
func (a *Agent) doSSHFSPut(ctx context.Context, env proto.TaskEnvelope) proto.TaskResult {
	var args proto.FSXferArgs
	if err := json.Unmarshal(env.Args, &args); err != nil || args.Chunk <= 0 || args.Size < 0 || args.Size > proto.FSFileCap {
		return proto.TaskResult{Error: "плохой запрос"}
	}
	c, err := sshfsGet(args.Conn)
	if err != nil {
		return proto.TaskResult{Error: err.Error()}
	}
	defer c.hold()()
	s, err := a.sshfsSession(c)
	if err != nil {
		return proto.TaskResult{Error: err.Error()}
	}
	defer s.Close()
	in, _ := s.StdinPipe()
	var errb bytes.Buffer
	s.Stderr = &limitBuf{b: &errb, max: 4096}
	if err := s.Start(writeCommand(args.Path)); err != nil {
		return proto.TaskResult{Error: err.Error()}
	}
	// Pieces come in a few ahead; stdin takes them in order. ack: how many
	// are written, for the server to let go.
	n := pieceCount(args.Size, args.Chunk)
	var written atomic.Int64
	err = ordered(ctx, n, xferWorkers, func(ctx context.Context, i int) ([]byte, error) {
		want := min(int64(args.Chunk), args.Size-int64(i)*int64(args.Chunk))
		var data []byte
		err := a.xferRequest(ctx, func(ctx context.Context) (*http.Request, error) {
			return http.NewRequestWithContext(ctx, http.MethodGet, a.xferURL(args.Xfer, i)+"?ack="+strconv.FormatInt(written.Load(), 10), nil)
		}, func(resp *http.Response) error {
			b, err := io.ReadAll(io.LimitReader(resp.Body, int64(args.Chunk)+1))
			if err == nil && int64(len(b)) != want {
				err = errors.New("кусок не той длины")
			}
			data = b
			return err
		})
		return data, err
	}, func(i int, b []byte) error {
		_, err := in.Write(b)
		written.Store(int64(i + 1))
		return err
	})
	if err != nil {
		_ = s.Close()
		return proto.TaskResult{Error: "загрузка прервана: " + err.Error()}
	}
	_ = in.Close()
	if err := s.Wait(); err != nil {
		msg := strings.TrimSpace(errb.String())
		if msg == "" {
			msg = err.Error()
		}
		return proto.TaskResult{Error: msg}
	}
	data, _ := json.Marshal(map[string]any{"path": args.Path, "size": args.Size})
	return proto.TaskResult{OK: true, Data: data}
}

// renameTar copies a tar stream, giving each top entry named in names - and
// everything under it - its new name: what the host's tar packs under the
// file's own name arrives under the name chosen for it. Other entries (the
// "._" companions a macOS tar adds) are left out.
func renameTar(w io.Writer, r io.Reader, names map[string]string) error {
	tr, tw := tar.NewReader(r), tar.NewWriter(w)
	for {
		h, err := tr.Next()
		if err == io.EOF {
			return tw.Close()
		}
		if err != nil {
			return err
		}
		n := strings.TrimPrefix(h.Name, "./")
		top, rest := n, ""
		if i := strings.IndexByte(n, '/'); i >= 0 {
			top, rest = n[:i], n[i:]
		}
		// Only base and what is under it - nothing that climbs out by "..".
		name, ok := names[top]
		if !ok || strings.Contains("/"+rest+"/", "/../") {
			continue
		}
		h.Name = name + rest
		h.Uid, h.Gid, h.Uname, h.Gname = 0, 0, "", ""
		if err := tw.WriteHeader(h); err != nil {
			return err
		}
		if _, err := io.Copy(tw, tr); err != nil {
			return err
		}
	}
}

// packRemote writes src on the connection's host as a tar stream named name,
// counting what goes for the progress.
func (a *Agent) packRemote(ctx context.Context, w io.Writer, c *sshfsConn, srcs, names []string, m *shipMeter) error {
	defer c.hold()()
	s, err := a.sshfsSession(c)
	if err != nil {
		return err
	}
	defer s.Close()
	go func() { <-ctx.Done(); s.Close() }()
	out, _ := s.StdoutPipe()
	var errb bytes.Buffer
	s.Stderr = &limitBuf{b: &errb, max: 4096}
	// All from one folder: packed from there by their own names.
	dir := path.Dir(path.Clean(srcs[0]))
	rename := map[string]string{}
	cmd := "COPYFILE_DISABLE=1 tar -cf - -C " + shq(dir) + " --"
	for i, p := range srcs {
		b := path.Base(path.Clean(p))
		rename[b] = names[i]
		cmd += " " + shq(b)
	}
	if err := s.Start(cmd); err != nil {
		return err
	}
	if err := renameTar(w, &countReader{r: out, n: &m.done}, rename); err != nil {
		return err
	}
	if err := s.Wait(); err != nil {
		if msg := strings.TrimSpace(errb.String()); msg != "" {
			return errors.New("tar на хосте: " + msg)
		}
		return err
	}
	return nil
}

// remoteSize is about how many bytes src takes on the host, for progress.
func (a *Agent) remoteSize(c *sshfsConn, src string) int64 {
	out, err := a.sshfsRun(c, "du -sk -- "+shq(src), nil)
	if err != nil {
		return 0
	}
	f := strings.Fields(string(out))
	if len(f) == 0 {
		return 0
	}
	kb, _ := strconv.ParseInt(f[0], 10, 64)
	return kb * 1024
}

// unpackRemote puts a tar stream into dir on the connection's host, by the
// conflict rule: tar keeps or replaces what is there, an "error" rule
// refuses when name is there already.
func (a *Agent) unpackRemote(c *sshfsConn, r io.Reader, dir string, names []string, conflict string) (skipped int, err error) {
	defer c.hold()()
	base := strings.TrimSuffix(dir, "/") + "/"
	check := ""
	for _, n := range names {
		check += "test -e " + shq(base+n) + " -o -L " + shq(base+n) + " && echo " + shq("LPRA_EXISTS "+n) + "; "
	}
	out, _ := a.sshfsRun(c, check+"true", nil)
	keep := map[string]string{}
	var put []string
	for _, n := range names {
		if strings.Contains(string(out), "LPRA_EXISTS "+n+"\n") {
			switch conflict {
			case "skip":
				// Left as it was, as on a machine.
				skipped++
				continue
			case "replace":
			default:
				return 0, errors.New(base + n + " уже существует")
			}
		}
		keep[n] = n
		put = append(put, n)
	}
	if len(put) == 0 {
		return skipped, nil
	}
	// Only the items sent (and not kept back) reach the host's tar.
	pr, pw := io.Pipe()
	go func() { pw.CloseWithError(renameTar(pw, r, keep)) }()
	defer pr.Close()
	_, err = a.sshfsRun(c, unpackCommand(dir, put, conflict), pr)
	return skipped, err
}

// unpackCommand unpacks a tar stream of names into dir; "replace" removes
// what is there first, so each thing is replaced whole as on a machine.
func unpackCommand(dir string, names []string, conflict string) string {
	cmd := "mkdir -p -- " + shq(dir) + " && cd -- " + shq(dir) + " && tar -xf -"
	if conflict == "replace" {
		rm := "rm -rf --"
		for _, n := range names {
			rm += " " + shq(strings.TrimSuffix(dir, "/")+"/"+n)
		}
		cmd = rm + " && " + cmd
	}
	return cmd
}