Files
lpra/internal/agent/sshfs.go
T

682 lines
20 KiB
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"
"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
}
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, "/") + "/")
loop := "cd " + d + " && for f in .[!.]* ..?* *; do [ -e \"$f\" ] || [ -L \"$f\" ] || continue; "
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) {
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"), Size: size, MTime: mt, Mode: p[3]}, 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; 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"), "unzip": hasLine(lines, "UNZIP")}, 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 {
c := v.(*sshfsConn)
c.mu.Lock()
idle := time.Since(c.last) > sshfsIdle
c.mu.Unlock()
if idle {
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))
}
return "cd -- " + shq(dir) + " && zip -r -q -y " + shq(path.Clean(dest)) + " " + strings.Join(names, " "), 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 "unzip -o -q " + shq(path.Clean(p)) + " -d " + shq(path.Clean(dest))
}
// 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; }"
}
func (a *Agent) sshfsSize(c *sshfsConn, p string) (int64, error) {
cmd := "stat -c %s -- " + shq(p)
if c.statFmt == "bsd" {
cmd = "stat -f %z -- " + shq(p)
} else if c.statFmt == "ls" {
cmd = "wc -c < " + shq(p)
}
out, err := a.sshfsRun(c, cmd, 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()}
}
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()}
}
sizeHdr := strconv.FormatInt(size, 10)
n := pieceCount(size, args.Chunk)
for i := 0; i < n; i++ {
buf := make([]byte, min(int64(args.Chunk), size-int64(i)*int64(args.Chunk)))
if _, err := io.ReadFull(out, buf); err != nil {
return proto.TaskResult{Error: "файл изменился во время чтения: " + err.Error()}
}
err := a.xferRequest(ctx, 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 {
return proto.TaskResult{Error: "передача прервана: " + err.Error()}
}
}
_ = s.Wait()
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()}
}
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()}
}
n := pieceCount(args.Size, args.Chunk)
for i := 0; i < n; i++ {
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.Itoa(i), 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
})
if err == nil {
_, err = in.Write(data)
}
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 its top entry base - and everything
// under it - the name name: what the host's tar packs under the file's own
// name arrives under the name chosen for it. Entries other than base (the
// "._" companions a macOS tar adds) are left out.
func renameTar(w io.Writer, r io.Reader, base, name 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:]
}
if top != base {
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, src, name string, m *shipMeter) error {
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}
src = path.Clean(src)
if err := s.Start("COPYFILE_DISABLE=1 tar -cf - -C " + shq(path.Dir(src)) + " -- " + shq(path.Base(src))); err != nil {
return err
}
if err := renameTar(w, &countReader{r: out, n: &m.done}, path.Base(src), name); 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, name, conflict string) error {
target := strings.TrimSuffix(dir, "/") + "/" + name
out, _ := a.sshfsRun(c, "test -e "+shq(target)+" -o -L "+shq(target)+" && echo LPRA_EXISTS; tar --version 2>/dev/null | head -n 1", nil)
if strings.Contains(string(out), "LPRA_EXISTS") && conflict != "replace" && conflict != "skip" {
return errors.New(target + " уже существует")
}
gnu := strings.Contains(string(out), "GNU tar")
flags := ""
switch conflict {
case "skip":
flags = " -k"
if gnu {
flags = " --skip-old-files"
}
case "replace":
if gnu {
flags = " --overwrite"
}
}
_, err := a.sshfsRun(c, "mkdir -p -- "+shq(dir)+" && cd -- "+shq(dir)+" && tar -xf -"+flags, r)
return err
}