diff --git a/internal/agent/ship.go b/internal/agent/ship.go new file mode 100644 index 0000000..178fb5b --- /dev/null +++ b/internal/agent/ship.go @@ -0,0 +1,580 @@ +package agent + +import ( + "archive/tar" + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "io" + "io/fs" + "net/http" + "os" + "path" + "path/filepath" + "strconv" + "strings" + "sync" + "sync/atomic" + "syscall" + "time" + + "golang.org/x/crypto/ssh" + + "github.com/x-grk/longpolling-remote-access/internal/proto" +) + +// A file or folder goes to another machine as a tar stream: packed here on +// the fly, then either posted in pieces to a relay on the server for the +// other agent to unpack, or written by SSH straight into tar on that host. +// Nothing is staged on disk on either side. + +// shipMeter counts what has gone out, for the panel's progress bar. +type shipMeter struct { + total, done atomic.Int64 + files atomic.Int64 + note atomic.Value // string +} + +func (m *shipMeter) snapshot() proto.ShipProgress { + n, _ := m.note.Load().(string) + return proto.ShipProgress{Total: m.total.Load(), Done: m.done.Load(), Files: int(m.files.Load()), Note: n} +} + +// reportShip posts the progress to the job until ctx ends; the server +// answering 410 means the job was cancelled, and cancel is called. +func (a *Agent) reportShip(ctx context.Context, job string, m *shipMeter, cancel func()) { + if job == "" { + return + } + t := time.NewTicker(700 * time.Millisecond) + defer t.Stop() + for { + body, _ := json.Marshal(m.snapshot()) + req, _ := http.NewRequestWithContext(ctx, http.MethodPost, a.cfg.Host+"/api/agent/fs/job/"+job+"/progress", bytes.NewReader(body)) + a.authHeaders(req) + req.Header.Set("Content-Type", "application/json") + if resp, err := a.http.Do(req); err == nil { + io.Copy(io.Discard, resp.Body) + resp.Body.Close() + if resp.StatusCode == http.StatusGone { + cancel() + return + } + } + select { + case <-ctx.Done(): + return + case <-t.C: + } + } +} + +// shipName checks the name the thing gets at the other end. +func shipName(name string) error { + if name == "" || name == "." || name == ".." || strings.ContainsAny(name, "/\x00") { + return errors.New("недопустимое имя") + } + return nil +} + +// treeSize is how many bytes of files are under p. +func treeSize(ctx context.Context, p string) int64 { + var n int64 + _ = filepath.WalkDir(p, func(_ string, d fs.DirEntry, err error) error { + if err != nil || ctx.Err() != nil { + return nil + } + if d.Type().IsRegular() { + if info, err := d.Info(); err == nil { + n += info.Size() + } + } + return nil + }) + return n +} + +// packTar writes src as a tar stream whose entries start at name. Owners are +// left out: the numbers mean other people on another machine. +func packTar(ctx context.Context, w io.Writer, src, name string, m *shipMeter) error { + tw := tar.NewWriter(w) + err := filepath.WalkDir(src, func(p string, d fs.DirEntry, err error) error { + if err != nil { + return err + } + if err := ctx.Err(); err != nil { + return err + } + info, err := os.Lstat(p) + if err != nil { + return err + } + link := "" + switch { + case info.Mode()&fs.ModeSymlink != 0: + if link, err = os.Readlink(p); err != nil { + return err + } + case info.IsDir(), info.Mode().IsRegular(): + default: + return nil // sockets, devices, pipes: not sent + } + hdr, err := tar.FileInfoHeader(info, link) + if err != nil { + return err + } + rel, _ := filepath.Rel(src, p) + hdr.Name = name + if rel != "." { + hdr.Name = path.Join(name, filepath.ToSlash(rel)) + } + if info.IsDir() { + hdr.Name += "/" + } + hdr.Uid, hdr.Gid, hdr.Uname, hdr.Gname = 0, 0, "", "" + hdr.Format = tar.FormatPAX + if err := tw.WriteHeader(hdr); err != nil { + return err + } + if info.Mode().IsRegular() { + f, err := os.Open(p) + if err != nil { + return err + } + _, err = io.CopyN(tw, &countReader{r: f, n: &m.done}, hdr.Size) + f.Close() + if err != nil { + return fmt.Errorf("%s: %v", p, err) + } + } + m.files.Add(1) + return nil + }) + if err != nil { + return err + } + return tw.Close() +} + +type countReader struct { + r io.Reader + n *atomic.Int64 +} + +func (c *countReader) Read(p []byte) (int, error) { + n, err := c.r.Read(p) + c.n.Add(int64(n)) + return n, err +} + +// doShipOut sends Src to the other machine: by the relay, or by SSH. +func (a *Agent) doShipOut(parent context.Context, env proto.TaskEnvelope) proto.TaskResult { + var args proto.FSShipArgs + if err := json.Unmarshal(env.Args, &args); err != nil { + return proto.TaskResult{Error: err.Error()} + } + if err := shipName(args.Name); err != nil { + return proto.TaskResult{Error: err.Error()} + } + src := filepath.Clean(args.Src) + if _, err := os.Lstat(src); err != nil { + return proto.TaskResult{Error: err.Error()} + } + ctx, cancel := context.WithCancel(parent) + defer cancel() + m := &shipMeter{} + m.note.Store("подсчёт") + go a.reportShip(ctx, args.Job, m, cancel) + m.total.Store(treeSize(ctx, src)) + + var err error + if args.SSH != nil { + err = a.shipSSH(ctx, args, src, m) + } else { + m.note.Store("") + err = a.shipRelay(ctx, args, src, m) + } + if err != nil { + if ctx.Err() != nil && parent.Err() == nil { + err = errors.New("отменено") + } + return proto.TaskResult{Error: err.Error()} + } + res := map[string]any{"files": m.files.Load(), "bytes": m.done.Load()} + if args.Move && args.SSH != nil { + if err := os.RemoveAll(src); err != nil { + return proto.TaskResult{Error: "отправлено, но здесь не удалилось: " + err.Error()} + } + } + data, _ := json.Marshal(res) + return proto.TaskResult{OK: true, Data: data} +} + +func (a *Agent) relayURL(id string, i int) string { + return a.cfg.Host + "/api/agent/fs/relay/" + id + "/" + strconv.Itoa(i) +} + +// shipRelay posts the tar stream in pieces, a few at once; the last carries +// X-Last. +func (a *Agent) shipRelay(ctx context.Context, args proto.FSShipArgs, src string, m *shipMeter) error { + if args.Relay == "" || args.Chunk <= 0 { + return errors.New("не задана передача") + } + ctx, cancel := context.WithCancel(ctx) + defer cancel() + pr, pw := io.Pipe() + go func() { pw.CloseWithError(packTar(ctx, pw, src, args.Name, m)) }() + defer pr.Close() + + type piece struct { + i int + data []byte + last bool + } + ch := make(chan piece, 1) + var ( + mu sync.Mutex + first error + wg sync.WaitGroup + ) + setErr := func(err error) { + mu.Lock() + if first == nil { + first = err + cancel() + } + mu.Unlock() + } + for w := 0; w < xferWorkers; w++ { + wg.Add(1) + go func() { + defer wg.Done() + for p := range ch { + err := a.xferRequest(ctx, func(ctx context.Context) (*http.Request, error) { + req, err := http.NewRequestWithContext(ctx, http.MethodPost, a.relayURL(args.Relay, p.i), bytes.NewReader(p.data)) + if err == nil { + req.Header.Set("Content-Type", "application/octet-stream") + if p.last { + req.Header.Set("X-Last", "1") + } + } + return req, err + }, func(*http.Response) error { return nil }) + if err != nil { + setErr(err) + } + } + }() + } + for i := 0; ; i++ { + buf := make([]byte, args.Chunk) + n, err := io.ReadFull(pr, buf) + last := err == io.EOF || err == io.ErrUnexpectedEOF + if err != nil && !last { + setErr(err) + break + } + select { + case ch <- piece{i, buf[:n], last}: + case <-ctx.Done(): + } + if last || ctx.Err() != nil { + break + } + } + close(ch) + wg.Wait() + if first != nil { + var gone errXferGone + if errors.As(first, &gone) { + return errors.New(string(gone)) + } + return first + } + return ctx.Err() +} + +// doShipIn takes the tar stream from the relay and unpacks it into Dir. +func (a *Agent) doShipIn(ctx context.Context, env proto.TaskEnvelope) proto.TaskResult { + var args proto.FSShipArgs + if err := json.Unmarshal(env.Args, &args); err != nil { + return proto.TaskResult{Error: err.Error()} + } + if err := shipName(args.Name); err != nil { + return proto.TaskResult{Error: err.Error()} + } + dir := filepath.Clean(args.Dir) + if _, err := os.Lstat(filepath.Join(dir, args.Name)); err == nil && args.Conflict != "replace" && args.Conflict != "skip" { + return proto.TaskResult{Error: filepath.Join(dir, args.Name) + " уже существует"} + } + if err := os.MkdirAll(dir, 0o755); err != nil { + return proto.TaskResult{Error: err.Error()} + } + ctx, cancel := context.WithCancel(ctx) + defer cancel() + pr, pw := io.Pipe() + go func() { + for i := 0; ; i++ { + var data []byte + var last bool + err := a.xferRequest(ctx, func(ctx context.Context) (*http.Request, error) { + return http.NewRequestWithContext(ctx, http.MethodGet, a.relayURL(args.Relay, i), nil) + }, func(resp *http.Response) error { + b, err := io.ReadAll(io.LimitReader(resp.Body, int64(args.Chunk)+1)) + data, last = b, resp.Header.Get("X-Last") != "" + return err + }) + if err == nil { + _, err = pw.Write(data) + } + if err != nil { + pw.CloseWithError(err) + return + } + if last { + pw.Close() + return + } + } + }() + u := &unpacker{dir: dir, name: args.Name, conflict: args.Conflict} + if os.Geteuid() == 0 { + // New things belong to whoever owns the folder they land in. + if info, err := os.Stat(dir); err == nil { + if st, ok := info.Sys().(*syscall.Stat_t); ok { + u.uid, u.gid, u.chown = int(st.Uid), int(st.Gid), true + } + } + } + err := u.unpack(tar.NewReader(pr)) + if err == nil { + // Whatever follows the archive's end is read off, so the sender + // finishes. + _, err = io.Copy(io.Discard, pr) + } + pr.CloseWithError(errors.New("распаковка закончена")) + if err != nil { + var gone errXferGone + if errors.As(err, &gone) { + err = errors.New(string(gone)) + } + return proto.TaskResult{Error: err.Error()} + } + data, _ := json.Marshal(map[string]int{"done": u.done, "skipped": u.skipped}) + return proto.TaskResult{OK: true, Data: data} +} + +// unpacker writes a tar stream into dir, every entry under name, following +// the conflict rule for what is there already: folders merge, and files are +// replaced or kept. +type unpacker struct { + dir, name, conflict string + uid, gid int + chown bool + done, skipped int +} + +func (u *unpacker) unpack(tr *tar.Reader) error { + var skipUnder string + for { + hdr, err := tr.Next() + if err == io.EOF { + return nil + } + if err != nil { + return err + } + clean := path.Clean(hdr.Name) + if clean != u.name && !strings.HasPrefix(clean, u.name+"/") { + return fmt.Errorf("в архиве чужой путь %q", hdr.Name) + } + target := filepath.Join(u.dir, filepath.FromSlash(clean)) + if skipUnder != "" && strings.HasPrefix(target, skipUnder) { + continue + } + have, err := os.Lstat(target) + exists := err == nil + if exists && !(hdr.Typeflag == tar.TypeDir && have.IsDir()) { + switch u.conflict { + case "skip": + u.skipped++ + if hdr.Typeflag == tar.TypeDir { + skipUnder = target + string(filepath.Separator) + } + continue + case "replace": + if have.IsDir() || hdr.Typeflag != tar.TypeReg { + if err := os.RemoveAll(target); err != nil { + return err + } + } + default: + return fmt.Errorf("%s уже существует", target) + } + } + mode := os.FileMode(hdr.Mode).Perm() + switch hdr.Typeflag { + case tar.TypeDir: + if !exists || !have.IsDir() { + if err := os.MkdirAll(target, mode|0o700); err != nil { + return err + } + _ = os.Chmod(target, mode) + u.own(target) + } + case tar.TypeReg: + if err := u.writeFile(target, tr, mode, hdr.ModTime); err != nil { + return err + } + case tar.TypeSymlink: + if err := os.Symlink(hdr.Linkname, target); err != nil { + return err + } + u.own(target) + default: + continue + } + u.done++ + } +} + +// writeFile writes next to the target first and puts it in place when all of +// it is there. +func (u *unpacker) writeFile(target string, r io.Reader, mode os.FileMode, mtime time.Time) error { + tmp, err := os.CreateTemp(filepath.Dir(target), "."+filepath.Base(target)+".lpra-*") + if err != nil { + return err + } + if _, err := io.Copy(tmp, r); err != nil { + tmp.Close() + os.Remove(tmp.Name()) + return err + } + if err := tmp.Close(); err != nil { + os.Remove(tmp.Name()) + return err + } + _ = os.Chmod(tmp.Name(), mode) + _ = os.Chtimes(tmp.Name(), mtime, mtime) + u.own(tmp.Name()) + if err := os.Rename(tmp.Name(), target); err != nil { + os.Remove(tmp.Name()) + return err + } + return nil +} + +func (u *unpacker) own(p string) { + if u.chown { + _ = os.Lchown(p, u.uid, u.gid) + } +} + +// shq quotes s for a POSIX shell. +func shq(s string) string { return "'" + strings.ReplaceAll(s, "'", `'\''`) + "'" } + +// shipSSH writes the tar stream into tar on the SSH host, in Dir. +func (a *Agent) shipSSH(ctx context.Context, args proto.FSShipArgs, src string, m *shipMeter) error { + client, _, err := a.sshConnect(ctx, *args.SSH, func(_, note string) { m.note.Store(note) }) + if err != nil { + return err + } + defer client.Close() + go func() { <-ctx.Done(); client.Close() }() + dir := args.Dir + if dir == "" { + dir = "." + } + target := strings.TrimSuffix(dir, "/") + "/" + args.Name + + // What tar is there, and whether the target is. + m.note.Store("проверка на той стороне") + out, err := sshRun(client, "command -v tar >/dev/null 2>&1 || echo LPRA_NOTAR; tar --version 2>/dev/null | head -n 1; test -e "+shq(target)+" -o -L "+shq(target)+" && echo LPRA_EXISTS") + if err != nil { + return err + } + if strings.Contains(out, "LPRA_NOTAR") { + return errors.New("на той стороне нет tar") + } + if strings.Contains(out, "LPRA_EXISTS") && args.Conflict != "replace" && args.Conflict != "skip" { + return errors.New(target + " там уже существует") + } + gnu := strings.Contains(out, "GNU tar") + flags := "" + switch args.Conflict { + case "skip": + if gnu { + flags = " --skip-old-files" + } else { + flags = " -k" + } + case "replace": + if gnu { + flags = " --overwrite" + } + } + + sess, err := client.NewSession() + if err != nil { + return err + } + defer sess.Close() + var stderr bytes.Buffer + sess.Stderr = &limitBuf{b: &stderr, max: 4096} + stdin, err := sess.StdinPipe() + if err != nil { + return err + } + cmd := "mkdir -p -- " + shq(dir) + " && cd -- " + shq(dir) + " && tar -xf -" + flags + if err := sess.Start(cmd); err != nil { + return err + } + m.note.Store("") + packErr := packTar(ctx, stdin, src, args.Name, m) + stdin.Close() + waitErr := sess.Wait() + msg := strings.TrimSpace(stderr.String()) + if packErr != nil { + if msg != "" { + return fmt.Errorf("%v (%s)", packErr, msg) + } + return packErr + } + if waitErr != nil { + var ee *ssh.ExitError + if errors.As(waitErr, &ee) && msg != "" { + return errors.New("tar на той стороне: " + msg) + } + return waitErr + } + return nil +} + +// sshRun runs a command and returns its output. +func sshRun(c *ssh.Client, cmd string) (string, error) { + s, err := c.NewSession() + if err != nil { + return "", err + } + defer s.Close() + out, err := s.Output(cmd) + var ee *ssh.ExitError + if errors.As(err, &ee) { + err = nil // the checks may exit non-zero; what they print is the answer + } + return string(out), err +} + +type limitBuf struct { + b *bytes.Buffer + max int +} + +func (l *limitBuf) Write(p []byte) (int, error) { + if room := l.max - l.b.Len(); room > 0 { + l.b.Write(p[:min(len(p), room)]) + } + return len(p), nil +} diff --git a/internal/agent/ssh.go b/internal/agent/ssh.go index af9503a..06150cf 100644 --- a/internal/agent/ssh.go +++ b/internal/agent/ssh.go @@ -148,32 +148,22 @@ func (s *sshShell) Close() error { return nil } -// runSSH connects to the target and bridges its shell to the session. -func (a *Agent) runSSH(parent context.Context, sid string, t proto.SSHTarget) { - status := func(st, note string) { - a.postOut(context.Background(), sid, proto.OutMessage{Status: st, Note: note}) - } - fail := func(err error) { - log.Printf("ssh %s: %v", sid, err) - status(proto.SSHErr, err.Error()) - a.postOut(context.Background(), sid, proto.OutMessage{Data: b64.EncodeToString([]byte("\x1b[31m" + err.Error() + "\x1b[0m\r\n"))}) - a.postOut(context.Background(), sid, proto.OutMessage{Exit: true}) - } +// sshConnect connects and signs in to t, saying how it goes by status (the +// SSHStatus steps). It returns the client and the host key's fingerprint. +func (a *Agent) sshConnect(ctx context.Context, t proto.SSHTarget, status func(st, note string)) (*ssh.Client, string, error) { if t.Port == 0 { t.Port = 22 } addr := net.JoinHostPort(t.Host, strconv.Itoa(t.Port)) auth, err := sshAuth(t) if err != nil { - fail(err) - return + return nil, "", err } status(proto.SSHDial, "подключение к "+addr) d := net.Dialer{Timeout: 15 * time.Second, KeepAlive: 30 * time.Second} - conn, err := d.DialContext(parent, "tcp", addr) + conn, err := d.DialContext(ctx, "tcp", addr) if err != nil { - fail(fmt.Errorf("не подключиться к %s: %v", addr, err)) - return + return nil, "", fmt.Errorf("не подключиться к %s: %v", addr, err) } status(proto.SSHAuth, "вход как "+t.User) var fp string @@ -190,11 +180,33 @@ func (a *Agent) runSSH(parent context.Context, sid string, t proto.SSHTarget) { if strings.Contains(msg, "unable to authenticate") { msg = "сервер не принял вход как " + t.User + " - проверьте пароль или ключ" } - fail(errors.New(msg)) - return + return nil, "", errors.New(msg) } _ = conn.SetDeadline(time.Time{}) - client := ssh.NewClient(c, chans, reqs) + return ssh.NewClient(c, chans, reqs), fp, nil +} + +// runSSH connects to the target and bridges its shell to the session. +func (a *Agent) runSSH(parent context.Context, sid string, t proto.SSHTarget) { + status := func(st, note string) { + a.postOut(context.Background(), sid, proto.OutMessage{Status: st, Note: note}) + } + fail := func(err error) { + log.Printf("ssh %s: %v", sid, err) + status(proto.SSHErr, err.Error()) + a.postOut(context.Background(), sid, proto.OutMessage{Data: b64.EncodeToString([]byte("\x1b[31m" + err.Error() + "\x1b[0m\r\n"))}) + a.postOut(context.Background(), sid, proto.OutMessage{Exit: true}) + } + client, fp, err := a.sshConnect(parent, t, status) + if err != nil { + fail(err) + return + } + port := t.Port + if port == 0 { + port = 22 + } + addr := net.JoinHostPort(t.Host, strconv.Itoa(port)) sess, err := client.NewSession() if err != nil { client.Close() diff --git a/internal/agent/task.go b/internal/agent/task.go index 2dcd31e..bbf58c2 100644 --- a/internal/agent/task.go +++ b/internal/agent/task.go @@ -26,6 +26,10 @@ func (a *Agent) runTask(ctx context.Context, taskID string) { res = a.doFSPut(ctx, env) case proto.TaskFSSend: res = a.doFSSend(ctx, env) + case proto.TaskFSShipOut: + res = a.doShipOut(ctx, env) + case proto.TaskFSShipIn: + res = a.doShipIn(ctx, env) default: res = a.execTask(env) } diff --git a/internal/proto/proto.go b/internal/proto/proto.go index aff0acc..525b3a4 100644 --- a/internal/proto/proto.go +++ b/internal/proto/proto.go @@ -150,6 +150,8 @@ const ( TaskSysUserAdd = "sys.useradd" // create a new OS user TaskSysStat = "sys.stat" // CPU load, RAM and disk usage right now TaskSysSpeed = "sys.speed" // measure the link to the server (ping, down, up) + TaskFSShipOut = "fs.ship.out" // pack a file or folder and send it to another machine + TaskFSShipIn = "fs.ship.in" // unpack what another machine sends ) // FSReadCap bounds how many bytes TaskFSRead returns (5 MiB). The panel lets a @@ -313,3 +315,30 @@ type SysSpeedResult struct { // counted on its own side by then. Partial bool `json:"partial,omitempty"` } + +// FSShipArgs are the arguments of TaskFSShipOut and TaskFSShipIn: Src (on the +// sending machine) lands in Dir as Name (on the receiving one), following +// Conflict as FSCopyArgs does. The sender packs it as a tar stream and either +// posts it in pieces of Chunk bytes to Relay on the server, for the receiver +// to take, or with SSH set writes it to tar on that host itself, and with +// Move then deletes Src. Both report to Job how far they are. +type FSShipArgs struct { + Src string `json:"src,omitempty"` + Name string `json:"name"` + Dir string `json:"dir"` + Conflict string `json:"conflict,omitempty"` + Move bool `json:"move,omitempty"` + Relay string `json:"relay,omitempty"` + Chunk int `json:"chunk,omitempty"` + Job string `json:"job,omitempty"` + SSH *SSHTarget `json:"ssh,omitempty"` +} + +// ShipProgress is what a sending machine reports: bytes of files sent out of +// Total, and files done. +type ShipProgress struct { + Total int64 `json:"total"` + Done int64 `json:"done"` + Files int `json:"files"` + Note string `json:"note,omitempty"` // what is going on: connecting, signing in +} diff --git a/internal/server/server.go b/internal/server/server.go index 287d110..5c6fd00 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -59,6 +59,7 @@ type Server struct { relays speedRelays // speed tests that run through the browser meters speedMeters // byte counts of the speed tests under way xfers xfers // files on their way between a browser and an agent + ship shipping // files on their way from one machine to another challenges challenges // passkey challenges handed out wsAgents wsAgents // agents on their WebSocket now pinMisses pinMisses // wrong PINs per session @@ -263,6 +264,9 @@ func (s *Server) Handler() http.Handler { mux.Handle("PUT /api/panel/fs/xfer/{xid}/{i}", s.requireAdmin(s.handleXferPut)) mux.Handle("POST /api/panel/fs/xfer/{xid}/cancel", s.requireAdmin(s.handleXferCancel)) mux.Handle("GET /api/panel/fs/xfer/{xid}/result", s.requireAdmin(s.handleXferResult)) + mux.Handle("POST /api/panel/agents/{id}/fs/ship", s.requireAdmin(s.handleFSShip)) + mux.Handle("GET /api/panel/fs/job/{jid}", s.requireAdmin(s.handleShipJob)) + mux.Handle("POST /api/panel/fs/job/{jid}/cancel", s.requireAdmin(s.handleShipCancel)) mux.Handle("POST /api/panel/agents/{id}/fs/{op}", s.requireAdmin(s.handleFSOp)) // Machine status (admin). @@ -294,6 +298,9 @@ func (s *Server) Handler() http.Handler { mux.HandleFunc("POST /api/agent/speed/relay/up", s.handleSpeedRelayUp) mux.HandleFunc("GET /api/agent/fs/xfer/{xid}/{i}", s.handleAgentXferGet) mux.HandleFunc("POST /api/agent/fs/xfer/{xid}/{i}", s.handleAgentXferPost) + mux.HandleFunc("POST /api/agent/fs/job/{jid}/progress", s.handleShipProgress) + mux.HandleFunc("POST /api/agent/fs/relay/{rid}/{i}", s.handleRelayPost) + mux.HandleFunc("GET /api/agent/fs/relay/{rid}/{i}", s.handleRelayGet) return s.logRequests(mux) } diff --git a/internal/server/ship.go b/internal/server/ship.go new file mode 100644 index 0000000..a3c0083 --- /dev/null +++ b/internal/server/ship.go @@ -0,0 +1,414 @@ +package server + +import ( + "context" + "encoding/json" + "net/http" + "strconv" + "strings" + "sync" + "time" + + "github.com/x-grk/longpolling-remote-access/internal/proto" +) + +// Sending a file or folder to another machine. The sending agent packs it as +// a tar stream; to another agent the stream goes in pieces through a relay +// here, a few pieces held at most, and the receiving agent unpacks it as it +// comes. To an SSH host the sending agent writes it there itself. The panel +// follows a job: how far the sender is, and the result. + +// relay carries one stream between two agents, in numbered pieces. +type relay struct { + id, from, to string + + mu sync.Mutex + changed chan struct{} + chunks map[int][]byte + low int // pieces below it were taken + end int // the last piece's number, -1 until it came + err string +} + +func (x *relay) notify() { + close(x.changed) + x.changed = make(chan struct{}) +} + +// await waits until ready() holds and returns true with x.mu held. +func (x *relay) await(ctx context.Context, d time.Duration, ready func() bool) bool { + t := time.NewTimer(d) + defer t.Stop() + for { + x.mu.Lock() + if ready() { + return true + } + ch := x.changed + x.mu.Unlock() + select { + case <-ch: + case <-t.C: + return false + case <-ctx.Done(): + return false + } + } +} + +func (x *relay) fail(msg string) { + x.mu.Lock() + if x.err == "" { + x.err = msg + x.chunks = nil + x.notify() + } + x.mu.Unlock() +} + +type shipJob struct { + id string + mu sync.Mutex + changed chan struct{} + ver int + prog proto.ShipProgress + done bool + res proto.TaskResult + cancel context.CancelFunc + stopped bool +} + +func (j *shipJob) update(fn func()) { + j.mu.Lock() + fn() + j.ver++ + close(j.changed) + j.changed = make(chan struct{}) + j.mu.Unlock() +} + +type shipping struct { + mu sync.Mutex + relays map[string]*relay + jobs map[string]*shipJob +} + +func (c *shipping) relay(id string) *relay { + c.mu.Lock() + defer c.mu.Unlock() + return c.relays[id] +} + +func (c *shipping) job(id string) *shipJob { + c.mu.Lock() + defer c.mu.Unlock() + return c.jobs[id] +} + +// shipReq is what the panel asks for: Src on this agent to Dir/Name on agent +// To, or on the SSH host. +type shipReq struct { + Src string `json:"src"` + Name string `json:"name"` + Dir string `json:"dir"` + Conflict string `json:"conflict"` + Move bool `json:"move"` + To string `json:"to"` + SSH *proto.SSHTarget `json:"ssh"` +} + +// POST /api/panel/agents/{id}/fs/ship - start sending; the answer names the +// job to follow. +func (s *Server) handleFSShip(w http.ResponseWriter, r *http.Request) { + from := r.PathValue("id") + var q shipReq + if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 64<<10)).Decode(&q); err != nil { + writeJSON(w, proto.TaskResult{Error: "bad request"}) + return + } + q.Name, q.Dir = strings.TrimSpace(q.Name), strings.TrimSpace(q.Dir) + if q.Src == "" || q.Name == "" || strings.Contains(q.Name, "/") || q.Dir == "" { + writeJSON(w, proto.TaskResult{Error: "не указано, что и куда"}) + return + } + if !s.fsAgent(w, from, true) { + return + } + where := "" + if q.SSH != nil { + t := q.SSH + t.Host, t.User = strings.TrimSpace(t.Host), strings.TrimSpace(t.User) + if t.Port == 0 { + t.Port = 22 + } + if t.Host == "" || t.User == "" || t.Port < 1 || t.Port > 65535 { + writeJSON(w, proto.TaskResult{Error: "нужны хост, порт и пользователь"}) + return + } + where = "ssh:" + t.User + "@" + t.Host + ":" + strconv.Itoa(t.Port) + } else { + if q.To == "" { + writeJSON(w, proto.TaskResult{Error: "не выбрана машина"}) + return + } + if !s.fsAgent(w, q.To, true) { + return + } + where = "agent:" + q.To + } + ctx, cancel := context.WithTimeout(context.Background(), xferTimeout) + j := &shipJob{id: randToken()[:24], changed: make(chan struct{}), cancel: cancel} + s.ship.mu.Lock() + if s.ship.jobs == nil { + s.ship.jobs, s.ship.relays = map[string]*shipJob{}, map[string]*relay{} + } + s.ship.jobs[j.id] = j + s.ship.mu.Unlock() + if admin := s.currentAdmin(r); admin != nil { + op := "copy" + if q.Move { + op = "move" + } + s.db.Audit(admin.ID, from, "fs.ship."+op+":"+q.Src+" -> "+where+":"+strings.TrimSuffix(q.Dir, "/")+"/"+q.Name) + } + go s.runShip(ctx, j, from, q) + data, _ := json.Marshal(map[string]string{"job": j.id}) + writeJSON(w, proto.TaskResult{OK: true, Data: data}) +} + +func (s *Server) runShip(ctx context.Context, j *shipJob, from string, q shipReq) { + defer j.cancel() + res := s.ship1(ctx, j, from, q) + if !res.OK && j.stopped { + res.Error = "отменено" + } + j.update(func() { j.done, j.res = true, res }) + // The result waits a while for the panel to come for it. + time.AfterFunc(xferIdle, func() { + s.ship.mu.Lock() + delete(s.ship.jobs, j.id) + s.ship.mu.Unlock() + }) +} + +func oldAgent(res proto.TaskResult) proto.TaskResult { + if unknownOp(res) { + res.Error = "агент не умеет отправлять на другие машины - обновите его" + } + return res +} + +func (s *Server) ship1(ctx context.Context, j *shipJob, from string, q shipReq) proto.TaskResult { + args := proto.FSShipArgs{Src: q.Src, Name: q.Name, Dir: q.Dir, Conflict: q.Conflict, Job: j.id} + if q.SSH != nil { + args.SSH, args.Move = q.SSH, q.Move + res, err := s.waitTask(ctx, from, proto.TaskFSShipOut, args, xferTimeout) + if err != nil { + return proto.TaskResult{Error: err.Error()} + } + return oldAgent(res) + } + + x := &relay{id: randToken()[:24], from: from, to: q.To, changed: make(chan struct{}), chunks: map[int][]byte{}, end: -1} + s.ship.mu.Lock() + s.ship.relays[x.id] = x + s.ship.mu.Unlock() + defer func() { + s.ship.mu.Lock() + delete(s.ship.relays, x.id) + s.ship.mu.Unlock() + }() + go func() { <-ctx.Done(); x.fail("отменено") }() + args.Relay, args.Chunk = x.id, xferChunk + + var in proto.TaskResult + inDone := make(chan struct{}) + go func() { + defer close(inDone) + res, err := s.waitTask(ctx, q.To, proto.TaskFSShipIn, args, xferTimeout) + if err != nil { + res = proto.TaskResult{Error: err.Error()} + } + in = oldAgent(res) + if !in.OK { + x.fail(in.Error) + } + }() + out, err := s.waitTask(ctx, from, proto.TaskFSShipOut, args, xferTimeout) + if err != nil { + out = proto.TaskResult{Error: err.Error()} + } + out = oldAgent(out) + if !out.OK { + x.fail(out.Error) + } + <-inDone + if !in.OK { + return proto.TaskResult{Error: "на той машине: " + in.Error} + } + if !out.OK { + return out + } + if q.Move { + res, err := s.waitTask(ctx, from, proto.TaskFSDelTree, proto.FSPathArgs{Path: q.Src}, fsTaskTimeout) + if err == nil && !res.OK { + err = errText(res.Error) + } + if err != nil { + return proto.TaskResult{Error: "отправлено, но здесь не удалилось: " + err.Error()} + } + } + // The receiver's count: what it put down and what it kept as it was. + return in +} + +type errText string + +func (e errText) Error() string { return string(e) } + +// GET /api/panel/fs/job/{jid}?v=N - the job's state once it differs from +// version N, or after a while anyway. +func (s *Server) handleShipJob(w http.ResponseWriter, r *http.Request) { + j := s.ship.job(r.PathValue("jid")) + if j == nil { + writeJSON(w, map[string]any{"done": true, "result": proto.TaskResult{Error: "задание не найдено"}}) + return + } + seen, _ := strconv.Atoi(r.URL.Query().Get("v")) + t := time.NewTimer(25 * time.Second) + defer t.Stop() + for { + j.mu.Lock() + if j.ver != seen || j.done { + out := map[string]any{"v": j.ver, "progress": j.prog, "done": j.done} + if j.done { + out["result"] = j.res + } + j.mu.Unlock() + writeJSON(w, out) + return + } + ch := j.changed + j.mu.Unlock() + select { + case <-ch: + case <-t.C: + seen = -1 // answer with what there is + case <-r.Context().Done(): + return + } + } +} + +// POST /api/panel/fs/job/{jid}/cancel +func (s *Server) handleShipCancel(w http.ResponseWriter, r *http.Request) { + if j := s.ship.job(r.PathValue("jid")); j != nil { + j.update(func() { j.stopped = true }) + j.cancel() + } + w.WriteHeader(http.StatusNoContent) +} + +// POST /api/agent/fs/job/{jid}/progress - the sender's progress. 410: the +// job is over or cancelled, stop. +func (s *Server) handleShipProgress(w http.ResponseWriter, r *http.Request) { + if s.authenticateAgent(r) == nil { + http.Error(w, "unauthorized", http.StatusUnauthorized) + return + } + j := s.ship.job(r.PathValue("jid")) + var p proto.ShipProgress + if j == nil || json.NewDecoder(http.MaxBytesReader(w, r.Body, 4096)).Decode(&p) != nil { + http.Error(w, "gone", http.StatusGone) + return + } + stop := false + j.update(func() { j.prog, stop = p, j.stopped || j.done }) + if stop { + http.Error(w, "gone", http.StatusGone) + return + } + w.WriteHeader(http.StatusNoContent) +} + +// POST /api/agent/fs/relay/{rid}/{i} - piece i of the stream from the +// sender, X-Last on the last. 503: no room yet, again; 410: it is over. +func (s *Server) handleRelayPost(w http.ResponseWriter, r *http.Request) { + a := s.authenticateAgent(r) + if a == nil { + http.Error(w, "unauthorized", http.StatusUnauthorized) + return + } + x := s.ship.relay(r.PathValue("rid")) + i, err := strconv.Atoi(r.PathValue("i")) + if x == nil || x.from != a.ID || err != nil || i < 0 { + http.Error(w, "передача не найдена", http.StatusGone) + return + } + data, err := readPiece(w, r) + if err != nil { + http.Error(w, "кусок не дошёл", http.StatusBadRequest) + return + } + if !x.await(r.Context(), xferWait, func() bool { return x.err != "" || i < x.low+xferBuffer }) { + http.Error(w, "нет места, повторите", http.StatusServiceUnavailable) + return + } + defer x.mu.Unlock() + if x.err != "" { + http.Error(w, x.err, http.StatusGone) + return + } + if i >= x.low && x.chunks[i] == nil { + x.chunks[i] = data + if r.Header.Get("X-Last") != "" { + x.end = i + } + x.notify() + } + w.WriteHeader(http.StatusNoContent) +} + +// GET /api/agent/fs/relay/{rid}/{i} - piece i for the receiver, X-Last on +// the last. Asking for i says every piece below it is written, so those are +// let go. 503: not here yet, ask again; 410: it is over. +func (s *Server) handleRelayGet(w http.ResponseWriter, r *http.Request) { + a := s.authenticateAgent(r) + if a == nil { + http.Error(w, "unauthorized", http.StatusUnauthorized) + return + } + x := s.ship.relay(r.PathValue("rid")) + i, err := strconv.Atoi(r.PathValue("i")) + if x == nil || x.to != a.ID || err != nil || i < 0 { + http.Error(w, "передача не найдена", http.StatusGone) + return + } + x.mu.Lock() + if i > x.low { + for k := x.low; k < i; k++ { + delete(x.chunks, k) + } + x.low = i + x.notify() + } + x.mu.Unlock() + if !x.await(r.Context(), xferWait, func() bool { return x.err != "" || x.chunks[i] != nil }) { + http.Error(w, "кусок ещё не пришёл", http.StatusServiceUnavailable) + return + } + if x.err != "" { + msg := x.err + x.mu.Unlock() + http.Error(w, msg, http.StatusGone) + return + } + data, last := x.chunks[i], i == x.end + x.mu.Unlock() + w.Header().Set("Content-Type", "application/octet-stream") + w.Header().Set("Content-Length", strconv.Itoa(len(data))) + if last { + w.Header().Set("X-Last", "1") + } + _, _ = w.Write(data) +} diff --git a/internal/server/web/templates/files.html b/internal/server/web/templates/files.html index 57d1a23..1394ccc 100644 --- a/internal/server/web/templates/files.html +++ b/internal/server/web/templates/files.html @@ -107,6 +107,41 @@ .seg button + button { border-left:1px solid var(--border); } .seg button.on { background:#12202f; color:var(--text); box-shadow:inset 0 -2px 0 var(--accent); } .seg button:hover { background:#151a21; color:var(--text); } + .mv-top { display:flex; align-items:flex-end; gap:8px; } + .mv-top .mv-net { height:36px; width:36px; flex:none; color:var(--muted); transition:background .18s, color .18s, border-color .18s; } + .mv-top .mv-net:hover { color:var(--text); } + .mv-top .mv-net.on { color:var(--accent); border-color:var(--accent); background:#12202f; } + /* The other machine: slides open under the action. */ + .mv-remote { display:grid; grid-template-rows:0fr; opacity:0; transition:grid-template-rows .25s ease, opacity .2s ease; } + .mv-remote > div { overflow:hidden; min-height:0; } + .mv-remote.on { grid-template-rows:1fr; opacity:1; } + .mv-remote.on > div { overflow:visible; } + .mv-ssh { display:grid; gap:8px; margin-top:10px; padding:10px 12px; border:1px solid var(--border); border-radius:9px; background:#0f1318; + animation:mv-in .2s ease; } + .mv-ssh[hidden], .mv-a[hidden] { display:none; } + @keyframes mv-in { from { opacity:0; transform:translateY(-4px); } } + .mv-ssh label { margin:0; display:grid; gap:4px; font-size:12px; color:var(--muted); } + .mv-ssh input[type=number] { width:100%; height:36px; padding:6px 10px; } + .mv-row { display:flex; gap:10px; align-items:end; } + .mv-row .grow { flex:1; min-width:0; } + .mv-row .port { width:92px; flex:none; } + .mv-ssh .seg-auth { margin-top:2px; display:flex; } + .mv-ssh .seg-auth button { flex:1; padding:6px 8px; font-size:13px; } + @media (max-width:520px) { .mv-ssh .seg-auth button { font-size:12px; padding:6px 4px; white-space:nowrap; } } + .mv-a { display:grid; gap:8px; animation:mv-in .18s ease; } + .mv-a textarea { width:100%; box-sizing:border-box; font:12px/1.4 ui-monospace,Menlo,monospace; resize:vertical; min-height:80px; + background:var(--bg); color:var(--text); border:1px solid var(--border); border-radius:8px; padding:8px 10px; } + .mv-a textarea:focus { border-color:var(--accent); outline:none; } + p.mv-a { margin:0; font-size:13px; } + #mv-keyname { font-size:12px; overflow:hidden; text-overflow:ellipsis; white-space:nowrap; align-self:center; } + .mv-prog { margin-top:14px; animation:mv-in .2s ease; } + .mv-prog[hidden] { display:none; } + .mv-bar { height:6px; border-radius:3px; background:#1a2029; overflow:hidden; } + .mv-bar i { display:block; height:100%; width:0; background:var(--accent); border-radius:3px; transition:width .5s ease; } + .mv-bar.wait i { width:30% !important; animation:mv-slide 1.1s ease-in-out infinite; } + @keyframes mv-slide { from { transform:translateX(-100%); } to { transform:translateX(340%); } } + .mv-ptext { display:flex; align-items:center; justify-content:space-between; gap:10px; margin-top:8px; font-size:13px; color:var(--muted); } + .mv.busy input, .mv.busy select, .mv.busy textarea, .mv.busy .seg, .mv.busy .mv-net { pointer-events:none; opacity:.6; } .mv-dirbox { position:relative; } .mv-dirbox input { font-family:ui-monospace,Menlo,monospace; font-size:13px; } .mv-sug { position:absolute; left:0; right:0; top:calc(100% + 4px); z-index:5; max-height:220px; overflow:auto; @@ -342,10 +377,32 @@
-
- - +
+
+ + +
+
+
+ + + +
@@ -359,6 +416,10 @@ +
@@ -1139,18 +1200,24 @@ } // ---------- Move or copy ---------- - const M = { src: '', op: 'move', sug: [], sel: -1, seq: 0 }; - const dirCache = new Map(); // folder -> {at, names} of its subfolders + // remote: to another machine (to: its id) or an SSH host (to: 'ssh'). + const M = { src: '', op: 'move', sug: [], sel: -1, seq: 0, remote: false, to: '', auth: 'password', job: null }; + const dirCache = new Map(); // machine + folder -> {at, names} of its subfolders + // The machine the folder is on: this one, or the one it goes to. + const target = () => !M.remote ? agentID : M.to === 'ssh' ? '' : M.to; // Lists a folder for the suggestions, quietly: no spinner, kept a little. async function subdirs(path) { - const c = dirCache.get(path); + const on = target(); + if (!on) return []; + const key = on + '\n' + path; + const c = dirCache.get(key); if (c && Date.now() - c.at < 30000) return c.names; try { - const r = await fetch('/api/panel/agents/' + encodeURIComponent(agentID) + '/fs/list', { + const r = await fetch('/api/panel/agents/' + encodeURIComponent(on) + '/fs/list', { method: 'POST', headers: {'Content-Type': 'application/json'}, body: JSON.stringify({path}) }).then(r => r.json()); const names = r.ok ? (r.data.entries || []).filter(e => e.is_dir).map(e => e.name).sort(byNameStr) : []; - dirCache.set(path, {at: Date.now(), names}); + dirCache.set(key, {at: Date.now(), names}); return names; } catch (e) { return []; } } @@ -1158,8 +1225,139 @@ function setOp(op) { M.op = op; document.querySelectorAll('#mv-op button').forEach(b => { b.classList.toggle('on', b.dataset.op === op); b.setAttribute('aria-checked', b.dataset.op === op); }); - $('mv-go').textContent = op === 'move' ? 'Переместить' : 'Копировать'; + goLabel(); } + function goLabel() { + $('mv-go').textContent = (M.op === 'move' ? 'Переместить' : 'Копировать') + (M.remote ? ' туда' : ''); + } + // ---- to another machine, or by SSH ---- + function lastShip() { try { return JSON.parse(localStorage.getItem('lpra.ship.last') || '{}'); } catch (e) { return {}; } } + function setRemote(on) { + M.remote = on; + $('mv-net').classList.toggle('on', on); + $('mv-net').setAttribute('aria-pressed', on); + $('mv-remote').classList.toggle('on', on); + goLabel(); + if (on) loadTargets(); + $('mv-sug').hidden = true; + } + function setTo(v) { + M.to = v; + $('mv-ssh').hidden = v !== 'ssh'; + dirCache.clear(); + } + function setAuth(a) { + M.auth = a; + document.querySelectorAll('#mv-auth button').forEach(b => b.classList.toggle('on', b.dataset.auth === a)); + document.querySelectorAll('.mv-a').forEach(el => { el.hidden = el.dataset.for !== a; }); + } + async function loadTargets() { + const sel = $('mv-to'), last = lastShip(); + sel.innerHTML = ''; + let all = []; + try { all = await fetch('/api/panel/agents/presence', {cache: 'no-store'}).then(r => r.json()); } catch (e) { /* SSH only */ } + sel.innerHTML = ''; + all.filter(a => a.id !== agentID).forEach(a => { + const o = document.createElement('option'); + o.value = a.id; o.disabled = !a.online; + o.textContent = a.name + ' · ' + a.user + '@' + a.hostname + (a.online ? '' : ' (не в сети)'); + sel.appendChild(o); + }); + const o = document.createElement('option'); + o.value = 'ssh'; o.textContent = 'SSH-хост…'; + sel.appendChild(o); + const ok = v => [...sel.options].some(x => x.value === v && !x.disabled); + const first = [...sel.options].find(x => !x.disabled); + sel.value = ok(last.to) ? last.to : first ? first.value : 'ssh'; + setTo(sel.value); + if (last.ssh) { $('mv-host').value = last.ssh.host || ''; $('mv-port').value = last.ssh.port || 22; $('mv-user').value = last.ssh.user || ''; setAuth(last.ssh.auth || 'password'); } + } + $('mv-net').onclick = () => { if (!M.job) setRemote(!M.remote); }; + $('mv-to').onchange = ev => { setTo(ev.target.value); suggest(); }; + document.querySelectorAll('#mv-auth button').forEach(b => b.onclick = () => setAuth(b.dataset.auth)); + $('mv-keyload').onclick = () => $('mv-keyfile').click(); + $('mv-keyfile').onchange = () => { + const f = $('mv-keyfile').files[0]; + if (!f) return; + if (f.size > 32768) { toast('файл великоват для ключа', true); return; } + f.text().then(t => { $('mv-key').value = t.trim() + '\n'; $('mv-keyname').textContent = f.name; }); + $('mv-keyfile').value = ''; + }; + function sshTarget() { + const t = {host: $('mv-host').value.trim(), port: Number($('mv-port').value) || 22, user: $('mv-user').value.trim()}; + if (!t.host || !t.user) return {error: 'укажите хост и пользователя'}; + if (M.auth === 'password') { + if (!$('mv-pw').value) return {error: 'введите пароль'}; + t.password = $('mv-pw').value; + } else if (M.auth === 'key') { + if (!/PRIVATE KEY/.test($('mv-key').value)) return {error: 'вставьте закрытый ключ или загрузите файл'}; + t.key = $('mv-key').value; + if ($('mv-pass').value) t.passphrase = $('mv-pass').value; + } + return {t}; + } + function busy(on) { + document.querySelector('#mvdlg .mv').classList.toggle('busy', on); + $('mv-go').disabled = on; + $('mv-prog').hidden = !on; + } + // ship sends to the other machine and follows the job to its end. + async function ship(dir, name, conflict, move) { + const body = {src: M.src, name, dir, conflict, move}; + const keep = {to: M.to}; + if (M.to === 'ssh') { + const r = sshTarget(); + if (r.error) { toast(r.error, true); return; } + body.ssh = r.t; + keep.ssh = {host: r.t.host, port: r.t.port, user: r.t.user, auth: M.auth}; + } else { + if (!M.to) { toast('выберите машину', true); return; } + body.to = M.to; + keep.ssh = lastShip().ssh; + } + try { localStorage.setItem('lpra.ship.last', JSON.stringify(keep)); } catch (e) { /* not kept */ } + busy(true); + $('mv-bar').parentNode.classList.add('wait'); + $('mv-bar').style.width = '0'; + $('mv-ptext').textContent = 'подготовка…'; + const r = await api('ship', body); + $('mv-pw').value = ''; $('mv-pass').value = ''; + if (!r.ok) { busy(false); toast(r.error || 'ошибка', true); return; } + const job = M.job = r.data.job; + let v = 0, res = null; + const t0 = Date.now(), seen = []; + while (!res) { + let j; + try { j = await fetch('/api/panel/fs/job/' + job + '?v=' + v, {cache: 'no-store'}).then(x => x.json()); } + catch (e) { await new Promise(ok => setTimeout(ok, 1500)); continue; } + v = j.v || 0; + const p = j.progress || {}; + if (p.total > 0 || p.done > 0) { + $('mv-bar').parentNode.classList.remove('wait'); + $('mv-bar').style.width = (p.total ? Math.min(100, p.done / p.total * 100) : 0).toFixed(1) + '%'; + // Speed over the last few seconds. + seen.push([Date.now(), p.done]); + while (seen.length > 2 && Date.now() - seen[0][0] > 4000) seen.shift(); + const dt = (seen[seen.length - 1][0] - seen[0][0]) / 1000, rate = dt > 0.5 ? (seen[seen.length - 1][1] - seen[0][1]) / dt : 0; + $('mv-ptext').textContent = fmtSize(p.done) + ' из ' + fmtSize(p.total) + ' · файлов: ' + p.files + (rate ? ' · ' + fmtSize(Math.round(rate)) + '/с' : ''); + } else if (p.note) { + $('mv-ptext').textContent = p.note === 'подсчёт' ? 'подсчёт…' : p.note + '…'; + } + if (j.done) res = j.result; + } + M.job = null; + if (!res.ok) { busy(false); toast(res.error || 'ошибка', true); return; } + $('mv-bar').style.width = '100%'; + const d = res.data || {}, secs = Math.max(1, Math.round((Date.now() - t0) / 1000)); + const what = move ? 'перенесено' : 'отправлено'; + toast(what + (d.done != null ? ': ' + d.done : d.files != null ? ': ' + d.files : '') + (d.skipped ? ', оставлено как было: ' + d.skipped : '') + ' · ' + secs + ' с'); + setTimeout(() => { busy(false); if ($('mvdlg').open) $('mvdlg').close(); }, 400); + if (move) load(cwd); + } + $('mv-stop').onclick = () => { + if (M.job) fetch('/api/panel/fs/job/' + M.job + '/cancel', {method: 'POST'}); + $('mv-ptext').textContent = 'отмена…'; + }; document.querySelectorAll('#mv-op button').forEach(b => b.onclick = () => setOp(b.dataset.op)); // Suggestions: the subfolders of what is typed, matching its last part. async function suggest() { @@ -1212,6 +1410,7 @@ M.src = join(cwd, e.name); $('mv-src').textContent = M.src + (e.is_dir ? '/' : ''); setOp('move'); + if (!M.job) { busy(false); setRemote(false); } $('mv-dir').value = cwd === '/' ? '/' : cwd + '/'; $('mv-name').value = e.name; $('mv-conflict').value = 'error'; @@ -1227,6 +1426,7 @@ if (!name || name.includes('/')) { toast('имя - без «/»', true); return; } const dst = (dir === '/' ? '' : dir) + '/' + name; const conflict = $('mv-conflict').value, move = M.op === 'move'; + if (M.remote) { if (!M.job) ship(dir, name, conflict, move); return; } const btn = $('mv-go'); btn.disabled = true; let r = await api('copy', {src: M.src, dst, move, conflict});