feat: agent keeps SSH file sessions and runs file operations on the host by commands
This commit is contained in:
1 parent
f5d6bd30e8
commit
2859e69abe
4 files changed
+607
No files matched your search
@@ -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}
|
||||||
|
}
|
||||||
@@ -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")
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -26,6 +26,12 @@ func (a *Agent) runTask(ctx context.Context, taskID string) {
|
|||||||
res = a.doFSPut(ctx, env)
|
res = a.doFSPut(ctx, env)
|
||||||
case proto.TaskFSSend:
|
case proto.TaskFSSend:
|
||||||
res = a.doFSSend(ctx, env)
|
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:
|
case proto.TaskFSShipOut:
|
||||||
res = a.doShipOut(ctx, env)
|
res = a.doShipOut(ctx, env)
|
||||||
case proto.TaskFSShipIn:
|
case proto.TaskFSShipIn:
|
||||||
@@ -107,3 +113,15 @@ func itoa64(n int64) string {
|
|||||||
}
|
}
|
||||||
return string(b[i:])
|
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}
|
||||||
|
}
|
||||||
@@ -167,6 +167,11 @@ const (
|
|||||||
TaskDBAdd = "db.add" // add a row
|
TaskDBAdd = "db.add" // add a row
|
||||||
TaskDBDel = "db.del" // delete rows by their keys
|
TaskDBDel = "db.del" // delete rows by their keys
|
||||||
TaskDBSQL = "db.sql" // run a statement typed in the panel
|
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
|
TaskFSShipOut = "fs.ship.out" // pack a file or folder and send it to another machine
|
||||||
TaskFSShipIn = "fs.ship.in" // unpack what another machine sends
|
TaskFSShipIn = "fs.ship.in" // unpack what another machine sends
|
||||||
)
|
)
|
||||||
@@ -186,6 +191,18 @@ type FSXferArgs struct {
|
|||||||
Xfer string `json:"xfer"`
|
Xfer string `json:"xfer"`
|
||||||
Size int64 `json:"size"`
|
Size int64 `json:"size"`
|
||||||
Chunk int `json:"chunk"`
|
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.
|
// TaskEnvelope is what the agent fetches for a task id.
|
||||||
|
|||||||
Reference in new issue
Block a user