diff --git a/internal/agent/sshfs.go b/internal/agent/sshfs.go new file mode 100644 index 0000000..127123d --- /dev/null +++ b/internal/agent/sshfs.go @@ -0,0 +1,525 @@ +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 ( + "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; 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() }) + return map[string]string{"id": id, "home": c.home}, nil +} + +// 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 "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("неизвестная операция") +} + +// 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} +} diff --git a/internal/agent/sshfs_test.go b/internal/agent/sshfs_test.go new file mode 100644 index 0000000..327b378 --- /dev/null +++ b/internal/agent/sshfs_test.go @@ -0,0 +1,47 @@ +package agent + +import ( + "strings" + "testing" +) + +func TestParseStatLine(t *testing.T) { + // statCommand prints: type|size|mtime|mode|name + e, ok := parseStatLine("directory|4096|1760000000|drwxr-xr-x|my dir") + if !ok || !e.IsDir || e.Name != "my dir" || e.MTime != 1760000000 || e.Mode != "drwxr-xr-x" { + t.Fatalf("dir: %+v %v", e, ok) + } + e, ok = parseStatLine("regular file|12|1760000001|-rw-r--r--|a|b.txt") + if !ok || e.IsDir || e.Size != 12 || e.Name != "a|b.txt" { + t.Fatalf("file with | in name: %+v %v", e, ok) + } + if _, ok := parseStatLine("garbage"); ok { + t.Fatal("garbage parsed") + } +} + +func TestStatCommandQuotes(t *testing.T) { + for _, fl := range []string{"gnu", "bsd"} { + c := statCommand(fl, "/srv/it's here") + if !strings.Contains(c, `'/srv/it'\''s here/'`) { + t.Errorf("%s: dir not quoted: %s", fl, c) + } + } +} + +func TestParseLsLine(t *testing.T) { + e, ok := parseLsLine("-rw-r--r-- 1 root root 1234 Oct 15 10:00 my file.txt") + if !ok || e.Name != "my file.txt" || e.Size != 1234 || e.IsDir { + t.Fatalf("file: %+v %v", e, ok) + } + e, ok = parseLsLine("lrwxrwxrwx 1 root root 7 Oct 15 10:00 link -> target") + if !ok || e.Name != "link" { + t.Fatalf("link: %+v %v", e, ok) + } + if _, ok := parseLsLine("drwxr-xr-x 2 root root 4096 Oct 15 10:00 .."); ok { + t.Fatal(".. listed") + } + if _, ok := parseLsLine("total 8"); ok { + t.Fatal("total parsed") + } +} diff --git a/internal/agent/task.go b/internal/agent/task.go index 337ef55..ef1bf92 100644 --- a/internal/agent/task.go +++ b/internal/agent/task.go @@ -26,6 +26,12 @@ func (a *Agent) runTask(ctx context.Context, taskID string) { res = a.doFSPut(ctx, env) case proto.TaskFSSend: res = a.doFSSend(ctx, env) + case proto.TaskSSHFSSend: + res = a.doSSHFSSend(ctx, env) + case proto.TaskSSHFSPut: + res = a.doSSHFSPut(ctx, env) + case proto.TaskSSHFSOpen, proto.TaskSSHFSClose, proto.TaskSSHFSOp: + res = resultOf(a.sshfsTask(ctx, env)) case proto.TaskFSShipOut: res = a.doShipOut(ctx, env) case proto.TaskFSShipIn: @@ -107,3 +113,15 @@ func itoa64(n int64) string { } return string(b[i:]) } + +// resultOf makes a task's result of what it returned. +func resultOf(data any, err error) proto.TaskResult { + if err != nil { + return proto.TaskResult{Error: err.Error()} + } + raw, err := json.Marshal(data) + if err != nil { + return proto.TaskResult{Error: err.Error()} + } + return proto.TaskResult{OK: true, Data: raw} +} diff --git a/internal/proto/proto.go b/internal/proto/proto.go index 901e5dc..e8045fe 100644 --- a/internal/proto/proto.go +++ b/internal/proto/proto.go @@ -167,6 +167,11 @@ const ( TaskDBAdd = "db.add" // add a row TaskDBDel = "db.del" // delete rows by their keys TaskDBSQL = "db.sql" // run a statement typed in the panel + TaskSSHFSOpen = "sshfs.open" // connect to an SSH host for a file pane; result {id, home} + TaskSSHFSClose = "sshfs.close" // drop that connection + TaskSSHFSOp = "sshfs.op" // a file operation on the host: list, read, write, mkdir, delete, deltree, move, copy, stat, chmod + TaskSSHFSSend = "sshfs.send" // a file of the host to the browser in pieces (FSXferArgs with Conn) + TaskSSHFSPut = "sshfs.put" // a file from the browser to the host in pieces TaskFSShipOut = "fs.ship.out" // pack a file or folder and send it to another machine TaskFSShipIn = "fs.ship.in" // unpack what another machine sends ) @@ -186,6 +191,18 @@ type FSXferArgs struct { Xfer string `json:"xfer"` Size int64 `json:"size"` Chunk int `json:"chunk"` + // Conn is the SSH file connection the file is on (sshfs.send, sshfs.put). + Conn string `json:"conn,omitempty"` +} + +// SSHFSArgs are the arguments of the sshfs.* tasks: the connection (Conn), or +// where to connect (SSH, for open); for an operation its name (Op) and its +// arguments (Args) in the shapes of the fs.* tasks. +type SSHFSArgs struct { + Conn string `json:"conn,omitempty"` + SSH *SSHTarget `json:"ssh,omitempty"` + Op string `json:"op,omitempty"` + Args json.RawMessage `json:"args,omitempty"` } // TaskEnvelope is what the agent fetches for a task id.