From e9a845a6eaffc24649ad1a0b568dcd3eecbb07b7 Mon Sep 17 00:00:00 2001 From: iRedTea Date: Wed, 14 Oct 2026 19:06:59 +0300 Subject: [PATCH] feat: files go in 4 MB pieces, several at once and each retried on its own, so they pass a proxy like Cloudflare and survive a bad link (0.0.8) --- VERSION.txt | 2 +- deploy/nginx.conf.example | 6 +- internal/agent/fsstream.go | 258 +++++++--- internal/agent/task.go | 8 +- internal/proto/proto.go | 22 +- internal/server/fsstream.go | 572 +++++++++++++++++------ internal/server/server.go | 17 +- internal/server/web/templates/files.html | 82 +++- 8 files changed, 715 insertions(+), 252 deletions(-) diff --git a/VERSION.txt b/VERSION.txt index 5a5831a..d169b2f 100644 --- a/VERSION.txt +++ b/VERSION.txt @@ -1 +1 @@ -0.0.7 +0.0.8 diff --git a/deploy/nginx.conf.example b/deploy/nginx.conf.example index debb1aa..233be8e 100644 --- a/deploy/nginx.conf.example +++ b/deploy/nginx.conf.example @@ -21,9 +21,9 @@ server { include /etc/letsencrypt/options-ssl-nginx.conf; # managed by Certbot ssl_dhparam /etc/letsencrypt/ssl-dhparams.pem; # managed by Certbot - # Files go through the panel as plain request bodies, up to 512 MB each - # way (an upload from the browser, a download sent up by the agent). - client_max_body_size 520m; + # Files go through the panel in 4 MB pieces; 64m leaves room for a client + # older than 0.0.8, which takes a file of up to 32 MB in one request. + client_max_body_size 64m; location / { proxy_pass http://127.0.0.1:8091; diff --git a/internal/agent/fsstream.go b/internal/agent/fsstream.go index 19675fb..0e3e413 100644 --- a/internal/agent/fsstream.go +++ b/internal/agent/fsstream.go @@ -1,42 +1,166 @@ package agent import ( + "bytes" "context" "encoding/json" + "errors" "fmt" "io" - "log" "net/http" "os" "path/filepath" "strconv" + "strings" + "sync" "syscall" "time" "github.com/x-grk/longpolling-remote-access/internal/proto" ) -// File transfers that do not fit a task's JSON: the bytes go as a plain HTTP -// body through a stream the server holds open for the browser on the other -// end, so a file is never held whole in memory. +// Files go to and from the browser in pieces, each a request of its own +// through the server: small enough for any proxy in front, and sent again +// alone when the link drops it. A few go at once, so a good link is kept as +// busy as by one long stream. -func (a *Agent) streamURL(id string) string { return a.cfg.Host + "/api/agent/fs/stream/" + id } +const ( + // xferWorkers is how many pieces are in flight at once. + xferWorkers = 4 + // xferPieceTimeout bounds one piece's request: a link that slow, or + // stuck, gets the piece sent again. + xferPieceTimeout = 5 * time.Minute + // xferTries bounds how often a piece is tried over a failing link. + xferTries = 30 +) -// doFSUpload writes the file the browser is uploading. It lands in a temporary +// errXferGone is the server saying the transfer is over: the browser left +// or gave up. +type errXferGone string + +func (e errXferGone) Error() string { return string(e) } + +func (a *Agent) xferURL(id string, i int) string { + return a.cfg.Host + "/api/agent/fs/xfer/" + id + "/" + strconv.Itoa(i) +} + +// xferRequest makes one request for a piece, trying again over a failing link. +// A 503 is the server waiting on the other side: asked again at once. +func (a *Agent) xferRequest(ctx context.Context, newReq func(context.Context) (*http.Request, error), keep func(*http.Response) error) error { + fails := 0 + for { + if err := ctx.Err(); err != nil { + return err + } + rctx, cancel := context.WithTimeout(ctx, xferPieceTimeout) + req, err := newReq(rctx) + if err != nil { + cancel() + return err + } + a.authHeaders(req) + resp, err := a.stream.Do(req) + if err == nil { + switch { + case resp.StatusCode == http.StatusGone: + msg, _ := io.ReadAll(io.LimitReader(resp.Body, 512)) + resp.Body.Close() + cancel() + return errXferGone(strings.TrimSpace(string(msg))) + case resp.StatusCode == http.StatusServiceUnavailable: + io.Copy(io.Discard, resp.Body) + resp.Body.Close() + cancel() + continue + case resp.StatusCode < 300: + err = keep(resp) + resp.Body.Close() + cancel() + if err == nil { + return nil + } + default: + io.Copy(io.Discard, resp.Body) + resp.Body.Close() + err = errStatus(resp.StatusCode) + } + } + cancel() + if fails++; fails >= xferTries { + return fmt.Errorf("сеть не держит передачу: %w", err) + } + select { + case <-time.After(min(time.Duration(fails)*time.Second, 10*time.Second)): + case <-ctx.Done(): + return ctx.Err() + } + } +} + +// pieces runs fn for every piece, xferWorkers at a time, until one fails. +func pieces(ctx context.Context, n int, fn func(ctx context.Context, i int) error) error { + ctx, cancel := context.WithCancel(ctx) + defer cancel() + var ( + mu sync.Mutex + next int + first error + wg sync.WaitGroup + ) + for w := 0; w < min(xferWorkers, n); w++ { + wg.Add(1) + go func() { + defer wg.Done() + for { + mu.Lock() + if first != nil || next >= n { + mu.Unlock() + return + } + i := next + next++ + mu.Unlock() + if err := fn(ctx, i); err != nil { + mu.Lock() + if first == nil { + first = err + cancel() + } + mu.Unlock() + return + } + } + }() + } + wg.Wait() + return first +} + +func pieceCount(size int64, chunk int) int { + if size <= 0 { + return 1 + } + return int((size + int64(chunk) - 1) / int64(chunk)) +} + +// doFSPut writes the file the browser is uploading. It lands in a temporary // file next to the target first, so a broken upload leaves the old file as it // was. -func (a *Agent) doFSUpload(ctx context.Context, env proto.TaskEnvelope) proto.TaskResult { - var args proto.FSStreamArgs +func (a *Agent) doFSPut(ctx context.Context, env proto.TaskEnvelope) proto.TaskResult { + var args proto.FSXferArgs if err := json.Unmarshal(env.Args, &args); err != nil { return proto.TaskResult{Error: err.Error()} } - if args.Size > proto.FSFileCap { + if args.Size < 0 || args.Size > proto.FSFileCap || args.Chunk <= 0 { return proto.TaskResult{Error: "файл больше 512 МБ"} } dir, base := filepath.Split(args.Path) if dir == "" { dir = "." } + if info, err := os.Stat(args.Path); err == nil && info.IsDir() { + return proto.TaskResult{Error: "это каталог"} + } tmp, err := os.CreateTemp(dir, "."+base+".lpra-*") if err != nil { return proto.TaskResult{Error: err.Error()} @@ -49,29 +173,52 @@ func (a *Agent) doFSUpload(ctx context.Context, env proto.TaskEnvelope) proto.Ta } }() - req, err := http.NewRequestWithContext(ctx, http.MethodGet, a.streamURL(args.Stream), nil) + // Pieces written so far, to tell the server which it can let go. + n := pieceCount(args.Size, args.Chunk) + var ( + mu sync.Mutex + got = make([]bool, n) + contig int + ) + err = pieces(ctx, n, func(ctx context.Context, i int) error { + want := min(int64(args.Chunk), args.Size-int64(i)*int64(args.Chunk)) + mu.Lock() + ack := contig + mu.Unlock() + 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(ack), 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 { + return err + } + if _, err := tmp.WriteAt(data, int64(i)*int64(args.Chunk)); err != nil { + return err + } + mu.Lock() + got[i] = true + for contig < n && got[contig] { + contig++ + } + mu.Unlock() + return nil + }) if err != nil { - return proto.TaskResult{Error: err.Error()} - } - a.authHeaders(req) - resp, err := a.stream.Do(req) - if err != nil { - return proto.TaskResult{Error: err.Error()} - } - defer resp.Body.Close() - if resp.StatusCode != http.StatusOK { - return proto.TaskResult{Error: errStatus(resp.StatusCode).Error()} - } - n, err := io.Copy(tmp, io.LimitReader(resp.Body, proto.FSFileCap+1)) - if err != nil { - log.Printf("fs receive %s: broke after %d of %d bytes: %v", args.Path, n, args.Size, err) + var gone errXferGone + if errors.As(err, &gone) { + return proto.TaskResult{Error: "загрузка прервана: " + string(gone)} + } return proto.TaskResult{Error: "загрузка прервана: " + err.Error()} } - if n > proto.FSFileCap { - return proto.TaskResult{Error: "файл больше 512 МБ"} - } - if args.Size >= 0 && n != args.Size { - return proto.TaskResult{Error: fmt.Sprintf("загрузка прервана: пришло %d из %d байт", n, args.Size)} + if err := tmp.Truncate(args.Size); err != nil { + return proto.TaskResult{Error: err.Error()} } if err := tmp.Close(); err != nil { return proto.TaskResult{Error: err.Error()} @@ -79,9 +226,6 @@ func (a *Agent) doFSUpload(ctx context.Context, env proto.TaskEnvelope) proto.Ta // A file that is being replaced keeps its mode and owner. mode := os.FileMode(0o644) if info, err := os.Stat(args.Path); err == nil { - if info.IsDir() { - return proto.TaskResult{Error: "это каталог"} - } mode = info.Mode().Perm() if st, ok := info.Sys().(*syscall.Stat_t); ok { _ = os.Chown(tmp.Name(), int(st.Uid), int(st.Gid)) @@ -94,16 +238,19 @@ func (a *Agent) doFSUpload(ctx context.Context, env proto.TaskEnvelope) proto.Ta return proto.TaskResult{Error: err.Error()} } done = true - data, _ := json.Marshal(map[string]any{"path": args.Path, "size": n}) + data, _ := json.Marshal(map[string]any{"path": args.Path, "size": args.Size}) return proto.TaskResult{OK: true, Data: data} } -// doFSGet sends a file to the browser that asked for it. -func (a *Agent) doFSGet(ctx context.Context, env proto.TaskEnvelope) proto.TaskResult { - var args proto.FSStreamArgs +// doFSSend sends a file to the browser that asked for it. +func (a *Agent) doFSSend(ctx context.Context, env proto.TaskEnvelope) proto.TaskResult { + var args proto.FSXferArgs if err := json.Unmarshal(env.Args, &args); err != nil { return proto.TaskResult{Error: err.Error()} } + if args.Chunk <= 0 { + return proto.TaskResult{Error: "не задан размер куска"} + } f, err := os.Open(args.Path) if err != nil { return proto.TaskResult{Error: err.Error()} @@ -116,28 +263,29 @@ func (a *Agent) doFSGet(ctx context.Context, env proto.TaskEnvelope) proto.TaskR if info.IsDir() { return proto.TaskResult{Error: "это каталог"} } - if info.Size() > proto.FSFileCap { + size := info.Size() + if size > proto.FSFileCap { return proto.TaskResult{Error: "файл больше 512 МБ"} } - req, err := http.NewRequestWithContext(ctx, http.MethodPost, a.streamURL(args.Stream), io.LimitReader(f, info.Size())) + sizeHdr := strconv.FormatInt(size, 10) + err = pieces(ctx, pieceCount(size, args.Chunk), func(ctx context.Context, i int) error { + off := int64(i) * int64(args.Chunk) + buf := make([]byte, min(int64(args.Chunk), size-off)) + if _, err := f.ReadAt(buf, off); err != nil && err != io.EOF { + return err + } + return 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()} + return proto.TaskResult{Error: "передача прервана: " + err.Error()} } - req.ContentLength = info.Size() - req.Header.Set("X-Size", strconv.FormatInt(info.Size(), 10)) - a.authHeaders(req) - start := time.Now() - resp, err := a.stream.Do(req) - if err != nil { - log.Printf("fs send %s (%d bytes): failed after %s: %v", args.Path, info.Size(), time.Since(start).Round(time.Millisecond), err) - return proto.TaskResult{Error: err.Error()} - } - io.Copy(io.Discard, resp.Body) - resp.Body.Close() - if resp.StatusCode != http.StatusNoContent { - log.Printf("fs send %s (%d bytes): server answered %d after %s", args.Path, info.Size(), resp.StatusCode, time.Since(start).Round(time.Millisecond)) - return proto.TaskResult{Error: "браузер не дочитал файл"} - } - data, _ := json.Marshal(map[string]any{"path": args.Path, "size": info.Size()}) + data, _ := json.Marshal(map[string]any{"path": args.Path, "size": size}) return proto.TaskResult{OK: true, Data: data} } diff --git a/internal/agent/task.go b/internal/agent/task.go index 2dc5602..2dcd31e 100644 --- a/internal/agent/task.go +++ b/internal/agent/task.go @@ -22,10 +22,10 @@ func (a *Agent) runTask(ctx context.Context, taskID string) { res = a.doUpdate() case proto.TaskSysSpeed: res = a.doSpeed(ctx, env) - case proto.TaskFSUpload: - res = a.doFSUpload(ctx, env) - case proto.TaskFSGet: - res = a.doFSGet(ctx, env) + case proto.TaskFSPut: + res = a.doFSPut(ctx, env) + case proto.TaskFSSend: + res = a.doFSSend(ctx, env) default: res = a.execTask(env) } diff --git a/internal/proto/proto.go b/internal/proto/proto.go index 35e2076..364b3dd 100644 --- a/internal/proto/proto.go +++ b/internal/proto/proto.go @@ -85,8 +85,8 @@ const ( TaskFSZip = "fs.zip" // zip selected paths (native archive/zip) TaskFSMove = "fs.move" // move/rename TaskFSDownload = "fs.download" // read a whole file (larger cap) for browser download - TaskFSUpload = "fs.upload" // write a file from a stream the agent fetches (0.0.7+) - TaskFSGet = "fs.get" // send a file as a stream the agent posts (0.0.7+) + TaskFSPut = "fs.put" // write a file the browser sends in pieces (0.0.8+) + TaskFSSend = "fs.send" // send a file to the browser in pieces (0.0.8+) TaskSysUpdate = "sys.update" // download the latest agent binary and restart TaskSysUsers = "sys.users" // report the agent's user and the login-capable users @@ -99,17 +99,17 @@ const ( // file be edited only when it fits, so that a save never cuts it short. const FSReadCap = 5 * 1024 * 1024 -// FSFileCap bounds a file sent either way as a stream (TaskFSUpload, -// TaskFSGet). +// FSFileCap bounds a file sent either way in pieces (TaskFSPut, TaskFSSend). const FSFileCap = 512 << 20 -// FSStreamArgs are the arguments of TaskFSUpload and TaskFSGet: the file, and -// the stream on the server that carries its bytes (GET it for an upload, POST -// to it for a download). Size is the upload's length, -1 when not known. -type FSStreamArgs struct { - Path string `json:"path"` - Stream string `json:"stream"` - Size int64 `json:"size"` +// FSXferArgs are the arguments of TaskFSPut and TaskFSSend: the file, and the +// transfer on the server whose pieces carry it - GET them for a put, POST them +// for a send, Chunk bytes each. Size is a put's length; -1 for a send. +type FSXferArgs struct { + Path string `json:"path"` + Xfer string `json:"xfer"` + Size int64 `json:"size"` + Chunk int `json:"chunk"` } // TaskEnvelope is what the agent fetches for a task id. diff --git a/internal/server/fsstream.go b/internal/server/fsstream.go index fd752ea..5ef7b52 100644 --- a/internal/server/fsstream.go +++ b/internal/server/fsstream.go @@ -6,94 +6,258 @@ import ( "encoding/json" "errors" "io" - "log" "net/http" "path" "strconv" "strings" "sync" - "sync/atomic" "time" "github.com/x-grk/longpolling-remote-access/internal/proto" ) -// Files go between the browser and the agent as plain HTTP bodies, joined on -// the server by a pipe: the browser's upload is the body of the agent's GET, -// the agent's POST is the body of the browser's download. Nothing is held -// whole in memory and nothing is base64. +// Files go between the browser and the agent in pieces. Each piece is a +// request of its own, small enough for any proxy in front (Cloudflare takes +// 100 MB a request at most), and one that fails is sent again instead of the +// whole file. Several pieces are in flight at once, so a good link stays as +// busy as it would with one long stream. The server holds only a few pieces +// of a transfer: whoever sends waits while they wait to be taken. +// +// A download keeps one plain response towards the browser, which a proxy +// passes on whatever its size: links, previews and the hex editor read it. -// fsStreamTimeout bounds one file transfer, however slow the links. -const fsStreamTimeout = 2 * time.Hour +const ( + // xferChunk is the size of a piece. + xferChunk = 4 << 20 + // xferBuffer is how many pieces of a transfer the server holds. + xferBuffer = 8 + // xferWait is how long a request waits for its piece, or for room for it, + // before it is told to ask again: well under a proxy's timeout. + xferWait = 50 * time.Second + // xferIdle ends a transfer nobody has touched for this long. + xferIdle = 10 * time.Minute + // xferTimeout bounds one transfer, however slow the links. + xferTimeout = 2 * time.Hour +) -// fsLegacyCap is the most an agent that predates the streams can take in or +// fsLegacyCap is the most an agent that predates the pieces can take in or // hand over: the file goes base64 inside a task then. const fsLegacyCap = 32 << 20 -// fsStream is one file on its way. -type fsStream struct { - agentID string - r *io.PipeReader - w *io.PipeWriter - size int64 // bytes to come, -1 when not known - ready chan struct{} // download: closed when the agent starts sending - once sync.Once - taken atomic.Bool // the agent has come for it +// xfer is one file on its way. +type xfer struct { + id, agentID string + + mu sync.Mutex + changed chan struct{} // closed and replaced on every change + size int64 // -1 until known + n int // pieces in all, once the size is known + chunks map[int][]byte + // low is the first piece still to go: the ones below it were written by + // the agent (upload) or to the browser (download). + low int + sized chan struct{} // closed once the size is known + err string // why the transfer failed + done bool + result proto.TaskResult + last time.Time } -// start marks the download as under way with its size; false when it already -// is. -func (st *fsStream) start(size int64) bool { - ok := false - st.once.Do(func() { - st.size = size - close(st.ready) - ok = true - }) - return ok +func pieces(size int64) int { + if size <= 0 { + return 1 // an empty file is one empty piece + } + return int((size + xferChunk - 1) / xferChunk) } -type fsStreams struct { +// pieceLen is how long piece i of the file is. +func (x *xfer) pieceLen(i int) int64 { + return min(int64(xferChunk), x.size-int64(i)*xferChunk) +} + +// notify wakes whoever waits on the transfer; x.mu is held. +func (x *xfer) notify() { + x.last = time.Now() + close(x.changed) + x.changed = make(chan struct{}) +} + +// await waits until ready() holds and returns true with x.mu held; false, +// with it free, when the wait ran out or ctx ended. +func (x *xfer) 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 *xfer) fail(msg string) { + x.mu.Lock() + if x.err == "" && !x.done { + x.err = msg + x.chunks = nil + x.notify() + } + x.mu.Unlock() +} + +func (x *xfer) finish(res proto.TaskResult) { + x.mu.Lock() + x.done, x.result, x.chunks = true, res, nil + x.notify() + x.mu.Unlock() +} + +// setSize records the file's size; false when it differs from the one known. +func (x *xfer) setSize(size int64) bool { + x.mu.Lock() + defer x.mu.Unlock() + if x.size >= 0 { + return x.size == size + } + x.size, x.n = size, pieces(size) + close(x.sized) + x.notify() + return true +} + +// put stores piece i once there is room for it, and answers with the HTTP +// status for its sender. A piece already taken or held is accepted again: it +// is a retry. +func (x *xfer) put(ctx context.Context, i int, data []byte) (int, string) { + if !x.await(ctx, xferWait, func() bool { return x.err != "" || x.done || i < x.low+xferBuffer }) { + return http.StatusServiceUnavailable, "нет места, повторите" + } + defer x.mu.Unlock() + if x.err != "" || x.done { + return http.StatusGone, x.gone() + } + if i < 0 || i >= x.n || int64(len(data)) != x.pieceLen(i) { + return http.StatusBadRequest, "не тот кусок" + } + if i >= x.low && x.chunks[i] == nil { + x.chunks[i] = data + x.notify() + } + return http.StatusNoContent, "" +} + +// take hands over piece i, the next one in order, and lets it go. +func (x *xfer) take(ctx context.Context, i int, d time.Duration) ([]byte, error) { + if !x.await(ctx, d, func() bool { return x.err != "" || x.chunks[i] != nil }) { + return nil, errors.New("кусок не пришёл вовремя") + } + defer x.mu.Unlock() + if x.err != "" { + return nil, errors.New(x.err) + } + data := x.chunks[i] + delete(x.chunks, i) + x.low = i + 1 + x.notify() + return data, nil +} + +func (x *xfer) gone() string { + if x.err != "" { + return x.err + } + return "передача закончена" +} + +// Read makes an upload a plain reader, for an agent that takes a file whole. +type xferReader struct { + ctx context.Context + x *xfer + i int + buf []byte +} + +func (r *xferReader) Read(p []byte) (int, error) { + for len(r.buf) == 0 { + if r.i >= r.x.n { + return 0, io.EOF + } + data, err := r.x.take(r.ctx, r.i, xferIdle) + if err != nil { + return 0, err + } + r.buf, r.i = data, r.i+1 + } + n := copy(p, r.buf) + r.buf = r.buf[n:] + return n, nil +} + +type xfers struct { mu sync.Mutex - m map[string]*fsStream + m map[string]*xfer } -func (c *fsStreams) open(agentID string, size int64) (string, *fsStream) { - st := &fsStream{agentID: agentID, size: size, ready: make(chan struct{})} - st.r, st.w = io.Pipe() - id := randToken()[:24] +func (c *xfers) open(agentID string, size int64) *xfer { + x := &xfer{ + id: randToken()[:24], agentID: agentID, size: -1, + changed: make(chan struct{}), chunks: map[int][]byte{}, + sized: make(chan struct{}), last: time.Now(), + } + if size >= 0 { + x.size, x.n = size, pieces(size) + close(x.sized) + } c.mu.Lock() if c.m == nil { - c.m = map[string]*fsStream{} + c.m = map[string]*xfer{} } - c.m[id] = st + // Forget the ones left behind: a closed tab, a client gone for good. + for id, old := range c.m { + old.mu.Lock() + stale := time.Since(old.last) > xferIdle + old.mu.Unlock() + if stale { + old.fail("передача брошена") + delete(c.m, id) + } + } + c.m[x.id] = x c.mu.Unlock() - return id, st + return x } -// get returns the stream if it belongs to the agent asking. -func (c *fsStreams) get(id, agentID string) *fsStream { +func (c *xfers) get(id string) *xfer { c.mu.Lock() defer c.mu.Unlock() - if st := c.m[id]; st != nil && st.agentID == agentID { - return st + return c.m[id] +} + +// forAgent returns the transfer if it is the asking agent's. +func (c *xfers) forAgent(id, agentID string) *xfer { + if x := c.get(id); x != nil && x.agentID == agentID { + return x } return nil } -func (c *fsStreams) drop(id string) { +func (c *xfers) drop(id string) { c.mu.Lock() - st := c.m[id] delete(c.m, id) c.mu.Unlock() - if st != nil { - st.r.CloseWithError(io.ErrClosedPipe) - st.w.CloseWithError(io.ErrClosedPipe) - } } -// waitTask runs a task on the agent and waits for its result, or for the -// browser to leave. +// waitTask runs a task on the agent and waits for its result, or for ctx to +// end. func (s *Server) waitTask(ctx context.Context, agentID, typ string, args any, timeout time.Duration) (proto.TaskResult, error) { raw, _ := json.Marshal(args) taskID, ch := s.hub.CreateTask(agentID, typ, raw) @@ -134,67 +298,172 @@ func (s *Server) fsAgent(w http.ResponseWriter, agentID string, asJSON bool) boo return true } -// POST /api/panel/agents/{id}/fs/upload?path=... - write a file; the request -// body is the file, passed on to the agent as it arrives. -func (s *Server) handleFSUpload(w http.ResponseWriter, r *http.Request) { +// readPiece reads a piece's body into a buffer of its own size, so pieces +// leave no trail of grown buffers behind. +func readPiece(w http.ResponseWriter, r *http.Request) ([]byte, error) { + if r.ContentLength >= 0 && r.ContentLength <= xferChunk { + data := make([]byte, r.ContentLength) + _, err := io.ReadFull(r.Body, data) + return data, err + } + return io.ReadAll(http.MaxBytesReader(w, r.Body, xferChunk)) +} + +// ---------- upload: browser -> agent ---------- + +// POST /api/panel/agents/{id}/fs/upload?path=...&size=N - start writing a +// file. The answer names the transfer and the piece size; the browser then +// PUTs the pieces and asks for the result. +func (s *Server) handleFSUploadStart(w http.ResponseWriter, r *http.Request) { agentID, p := r.PathValue("id"), r.URL.Query().Get("path") if !s.fsAgent(w, agentID, true) { return } - if p == "" { - writeJSON(w, proto.TaskResult{Error: "не указан путь"}) + size, err := strconv.ParseInt(r.URL.Query().Get("size"), 10, 64) + if p == "" || err != nil || size < 0 { + writeJSON(w, proto.TaskResult{Error: "не указан путь или размер"}) return } - if r.ContentLength > proto.FSFileCap { + if size > proto.FSFileCap { writeJSON(w, proto.TaskResult{Error: "файл больше 512 МБ"}) return } - body := http.MaxBytesReader(w, r.Body, proto.FSFileCap) - id, st := s.fsStreams.open(agentID, r.ContentLength) - defer s.fsStreams.drop(id) - copied := make(chan struct{}) - go func() { - defer close(copied) - n, err := io.Copy(st.w, body) - if err != nil && !errors.Is(err, io.ErrClosedPipe) { - log.Printf("fs upload %s to %s: browser stream broke after %d of %d bytes: %v", p, agentID, n, r.ContentLength, err) - } - st.w.CloseWithError(err) // nil: the reader sees the end of the file - }() - - args := proto.FSStreamArgs{Path: p, Stream: id, Size: r.ContentLength} - res, err := s.waitTask(r.Context(), agentID, proto.TaskFSUpload, args, fsStreamTimeout) - if err == nil && unknownOp(res) && !st.taken.Load() { - res = s.legacyUpload(r.Context(), agentID, p, st.r) - } - // Stop the copy and let it end before the handler does: the body is not - // to be read after that. - st.r.CloseWithError(io.ErrClosedPipe) - <-copied - if err != nil { - writeJSON(w, proto.TaskResult{Error: err.Error()}) - return - } - writeJSON(w, res) + x := s.xfers.open(agentID, size) + go s.runUpload(x, p) + data, _ := json.Marshal(map[string]any{"id": x.id, "chunk": xferChunk}) + writeJSON(w, proto.TaskResult{OK: true, Data: data}) } -// legacyUpload writes a file through an agent that predates the streams. -func (s *Server) legacyUpload(ctx context.Context, agentID, p string, body io.Reader) proto.TaskResult { - data, err := io.ReadAll(io.LimitReader(body, fsLegacyCap+1)) +// runUpload has the agent write the file and keeps the result for the +// browser to collect. +func (s *Server) runUpload(x *xfer, p string) { + ctx, cancel := context.WithTimeout(context.Background(), xferTimeout) + defer cancel() + args := proto.FSXferArgs{Path: p, Xfer: x.id, Size: x.size, Chunk: xferChunk} + res, err := s.waitTask(ctx, x.agentID, proto.TaskFSPut, args, xferTimeout) + if err == nil && unknownOp(res) { + res = s.legacyUpload(ctx, x, p) + } + if err != nil { + res = proto.TaskResult{Error: err.Error()} + } + if !res.OK { + x.fail(res.Error) + } + x.finish(res) + // The result waits a while for the browser to come for it. + time.AfterFunc(xferIdle, func() { s.xfers.drop(x.id) }) +} + +// legacyUpload writes a file through an agent that predates the pieces. +func (s *Server) legacyUpload(ctx context.Context, x *xfer, p string) proto.TaskResult { + if x.size > fsLegacyCap { + return proto.TaskResult{Error: "клиент принимает файлы до 32 МБ - обновите его"} + } + data, err := io.ReadAll(&xferReader{ctx: ctx, x: x}) if err != nil { return proto.TaskResult{Error: err.Error()} } - if len(data) > fsLegacyCap { - return proto.TaskResult{Error: "клиент принимает файлы до 32 МБ - обновите его"} - } args := proto.FSWriteArgs{Path: p, Data: base64.StdEncoding.EncodeToString(data)} - res, err := s.waitTask(ctx, agentID, proto.TaskFSWrite, args, fsTaskTimeout) + res, err := s.waitTask(ctx, x.agentID, proto.TaskFSWrite, args, fsTaskTimeout) if err != nil { return proto.TaskResult{Error: err.Error()} } return res } +// PUT /api/panel/fs/xfer/{xid}/{i} - piece i of a file being uploaded. 503: +// no room yet, send it again; 410: the transfer is over. +func (s *Server) handleXferPut(w http.ResponseWriter, r *http.Request) { + x := s.xfers.get(r.PathValue("xid")) + i, err := strconv.Atoi(r.PathValue("i")) + if x == nil || err != nil { + http.Error(w, "передача не найдена", http.StatusGone) + return + } + data, err := readPiece(w, r) + if err != nil { + http.Error(w, "кусок не дошёл", http.StatusBadRequest) + return + } + code, msg := x.put(r.Context(), i, data) + if code == http.StatusNoContent { + w.WriteHeader(code) + return + } + http.Error(w, msg, code) +} + +// POST /api/panel/fs/xfer/{xid}/cancel - the browser gives the upload up. +func (s *Server) handleXferCancel(w http.ResponseWriter, r *http.Request) { + if x := s.xfers.get(r.PathValue("xid")); x != nil { + x.fail("отменено") + } + w.WriteHeader(http.StatusNoContent) +} + +// GET /api/panel/fs/xfer/{xid}/result - the upload's result, once the agent +// has written the file; {"done":false} means ask again. +func (s *Server) handleXferResult(w http.ResponseWriter, r *http.Request) { + x := s.xfers.get(r.PathValue("xid")) + if x == nil { + writeJSON(w, map[string]any{"done": true, "result": proto.TaskResult{Error: "передача не найдена"}}) + return + } + if !x.await(r.Context(), xferWait, func() bool { return x.done }) { + writeJSON(w, map[string]any{"done": false}) + return + } + res := x.result + x.mu.Unlock() + writeJSON(w, map[string]any{"done": true, "result": res}) +} + +// GET /api/agent/fs/xfer/{xid}/{i}?ack=K - piece i of the upload, for the +// agent to write. K says the agent has written every piece below it, so +// those are let go. 503: not here yet, ask again; 410: the transfer is over. +func (s *Server) handleAgentXferGet(w http.ResponseWriter, r *http.Request) { + a := s.authenticateAgent(r) + if a == nil { + http.Error(w, "unauthorized", http.StatusUnauthorized) + return + } + x := s.xfers.forAgent(r.PathValue("xid"), a.ID) + i, err := strconv.Atoi(r.PathValue("i")) + if x == nil || err != nil { + http.Error(w, "передача не найдена", http.StatusGone) + return + } + if ack, err := strconv.Atoi(r.URL.Query().Get("ack")); err == nil { + x.mu.Lock() + if ack > x.low { + for k := x.low; k < ack; k++ { + delete(x.chunks, k) + } + x.low = ack + 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 := x.chunks[i] + x.mu.Unlock() + w.Header().Set("Content-Type", "application/octet-stream") + w.Header().Set("Content-Length", strconv.Itoa(len(data))) + _, _ = w.Write(data) +} + +// ---------- download: agent -> browser ---------- + // inlineTypes are the files the panel shows in place rather than downloads. var inlineTypes = map[string]string{ ".png": "image/png", ".jpg": "image/jpeg", ".jpeg": "image/jpeg", ".gif": "image/gif", @@ -229,35 +498,38 @@ func fileHeaders(w http.ResponseWriter, p string, inline bool) { w.Header().Set("Cache-Control", "no-store") } -// GET /api/panel/agents/{id}/fs/download?path=...[&inline=1] - the file, -// streamed from the agent as it reads it. inline=1 serves pictures, video and -// sound for the page to show. +// GET /api/panel/agents/{id}/fs/download?path=...[&inline=1] - the file, as +// one response fed by the pieces the agent sends. inline=1 serves pictures, +// video and sound for the page to show. func (s *Server) handleFSDownload(w http.ResponseWriter, r *http.Request) { agentID, p := r.PathValue("id"), r.URL.Query().Get("path") inline := r.URL.Query().Get("inline") != "" if !s.fsAgent(w, agentID, false) { return } - id, st := s.fsStreams.open(agentID, -1) - defer s.fsStreams.drop(id) - raw, _ := json.Marshal(proto.FSStreamArgs{Path: p, Stream: id, Size: -1}) - taskID, ch := s.hub.CreateTask(agentID, proto.TaskFSGet, raw) + x := s.xfers.open(agentID, -1) + defer s.xfers.drop(x.id) + defer x.fail("браузер ушёл") // stops the agent if the response ends early + + raw, _ := json.Marshal(proto.FSXferArgs{Path: p, Xfer: x.id, Size: -1, Chunk: xferChunk}) + taskID, ch := s.hub.CreateTask(agentID, proto.TaskFSSend, raw) defer s.hub.DiscardTask(taskID) + // The agent's result: an error ends the transfer wherever it stands. + results := make(chan proto.TaskResult, 1) + go func() { + select { + case res := <-ch: + if !res.OK && !unknownOp(res) { + x.fail(res.Error) + } + results <- res + case <-r.Context().Done(): + } + }() select { - case <-st.ready: - fileHeaders(w, p, inline) - if st.size >= 0 { - w.Header().Set("Content-Length", strconv.FormatInt(st.size, 10)) - } - if n, err := io.Copy(w, st.r); err != nil { - // Cut the response off, so the browser does not take a part of - // the file for all of it. - log.Printf("fs download %s from %s: cut off after %d of %d bytes: %v", p, agentID, n, st.size, err) - st.r.CloseWithError(err) - panic(http.ErrAbortHandler) - } - case res := <-ch: + case <-x.sized: + case res := <-results: if unknownOp(res) { s.legacyDownload(w, r, agentID, p, inline) return @@ -267,13 +539,33 @@ func (s *Server) handleFSDownload(w http.ResponseWriter, r *http.Request) { msg = "агент не отдал файл" } http.Error(w, msg, http.StatusBadGateway) + return case <-r.Context().Done(): + return case <-time.After(fsTaskTimeout): http.Error(w, "таймаут ожидания агента", http.StatusGatewayTimeout) + return + } + + fileHeaders(w, p, inline) + w.Header().Set("Content-Length", strconv.FormatInt(x.size, 10)) + rc := http.NewResponseController(w) + for i := 0; i < x.n; i++ { + data, err := x.take(r.Context(), i, xferIdle) + if err == nil { + _, err = w.Write(data) + } + if err != nil { + // Cut the response off, so the browser does not take a part of + // the file for all of it. + x.fail(err.Error()) + panic(http.ErrAbortHandler) + } + _ = rc.Flush() } } -// legacyDownload gets a file from an agent that predates the streams. +// legacyDownload gets a file from an agent that predates the pieces. func (s *Server) legacyDownload(w http.ResponseWriter, r *http.Request, agentID, p string, inline bool) { res, err := s.waitTask(r.Context(), agentID, proto.TaskFSDownload, proto.FSPathArgs{Path: p}, fsTaskTimeout) if err != nil { @@ -297,59 +589,35 @@ func (s *Server) legacyDownload(w http.ResponseWriter, r *http.Request, agentID, _, _ = w.Write(data) } -// GET /api/agent/fs/stream/{sid} - the file the browser is uploading, for the -// agent to write. -func (s *Server) handleAgentFSStreamGet(w http.ResponseWriter, r *http.Request) { +// POST /api/agent/fs/xfer/{xid}/{i} - piece i of the file the browser asked +// for, X-Size its whole size. 503: no room yet, send it again; 410: the +// browser is gone. +func (s *Server) handleAgentXferPost(w http.ResponseWriter, r *http.Request) { a := s.authenticateAgent(r) if a == nil { http.Error(w, "unauthorized", http.StatusUnauthorized) return } - st := s.fsStreams.get(r.PathValue("sid"), a.ID) - if st == nil || !st.taken.CompareAndSwap(false, true) { - http.Error(w, "gone", http.StatusGone) - return - } - w.Header().Set("Content-Type", "application/octet-stream") - w.Header().Set("X-Accel-Buffering", "no") - if st.size >= 0 { - w.Header().Set("Content-Length", strconv.FormatInt(st.size, 10)) - } - if n, err := io.Copy(w, st.r); err != nil { - // The browser broke off: cut the response, so the agent does not - // take a part of the file for all of it. - log.Printf("fs upload to %s: cut off after %d of %d bytes: %v", a.ID, n, st.size, err) - panic(http.ErrAbortHandler) - } -} - -// POST /api/agent/fs/stream/{sid} - the file the browser asked for, sent by -// the agent. 204 once the browser has it all. -func (s *Server) handleAgentFSStreamPost(w http.ResponseWriter, r *http.Request) { - a := s.authenticateAgent(r) - if a == nil { - http.Error(w, "unauthorized", http.StatusUnauthorized) - return - } - st := s.fsStreams.get(r.PathValue("sid"), a.ID) - if st == nil { - http.Error(w, "gone", http.StatusGone) + x := s.xfers.forAgent(r.PathValue("xid"), a.ID) + i, err := strconv.Atoi(r.PathValue("i")) + if x == nil || err != nil { + http.Error(w, "передача не найдена", http.StatusGone) return } size, err := strconv.ParseInt(r.Header.Get("X-Size"), 10, 64) - if err != nil { - size = -1 - } - if !st.taken.CompareAndSwap(false, true) || !st.start(size) { - http.Error(w, "gone", http.StatusGone) + if err != nil || size < 0 || size > proto.FSFileCap || !x.setSize(size) { + http.Error(w, "не тот размер", http.StatusBadRequest) return } - n, err := io.Copy(st.w, http.MaxBytesReader(w, r.Body, proto.FSFileCap)) - st.w.CloseWithError(err) + data, err := readPiece(w, r) if err != nil { - log.Printf("fs download from %s: agent stream broke after %d of %d bytes: %v", a.ID, n, size, err) - http.Error(w, "gone", http.StatusGone) + http.Error(w, "кусок не дошёл", http.StatusBadRequest) return } - w.WriteHeader(http.StatusNoContent) + code, msg := x.put(r.Context(), i, data) + if code == http.StatusNoContent { + w.WriteHeader(code) + return + } + http.Error(w, msg, code) } diff --git a/internal/server/server.go b/internal/server/server.go index 929cf2f..c89bd5b 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -53,10 +53,10 @@ type Server struct { tunnelMu sync.Mutex // serializes port changes tunnelPort atomic.Int64 - stats statCache // last load sample per agent, for the status page - relays speedRelays // speed tests that run through the browser - meters speedMeters // byte counts of the speed tests under way - fsStreams fsStreams // files on their way between a browser and an agent + stats statCache // last load sample per agent, for the status page + 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 } // New builds the server, parsing embedded templates. @@ -230,7 +230,10 @@ func (s *Server) Handler() http.Handler { // Panel file manager (admin-only). mux.Handle("GET /api/panel/agents/{id}/fs/download", s.requireAdmin(s.handleFSDownload)) - mux.Handle("POST /api/panel/agents/{id}/fs/upload", s.requireAdmin(s.handleFSUpload)) + mux.Handle("POST /api/panel/agents/{id}/fs/upload", s.requireAdmin(s.handleFSUploadStart)) + 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/{op}", s.requireAdmin(s.handleFSOp)) // Machine status (admin). @@ -259,8 +262,8 @@ func (s *Server) Handler() http.Handler { mux.HandleFunc("POST /api/agent/speed/up", s.handleSpeedUp) mux.HandleFunc("GET /api/agent/speed/relay/down", s.handleSpeedRelayDown) mux.HandleFunc("POST /api/agent/speed/relay/up", s.handleSpeedRelayUp) - mux.HandleFunc("GET /api/agent/fs/stream/{sid}", s.handleAgentFSStreamGet) - mux.HandleFunc("POST /api/agent/fs/stream/{sid}", s.handleAgentFSStreamPost) + mux.HandleFunc("GET /api/agent/fs/xfer/{xid}/{i}", s.handleAgentXferGet) + mux.HandleFunc("POST /api/agent/fs/xfer/{xid}/{i}", s.handleAgentXferPost) return s.logRequests(mux) } diff --git a/internal/server/web/templates/files.html b/internal/server/web/templates/files.html index f85b840..e7a6d6d 100644 --- a/internal/server/web/templates/files.html +++ b/internal/server/web/templates/files.html @@ -398,6 +398,64 @@ // ---------- File viewer ---------- const fsURL = (what, path) => '/api/panel/agents/' + encodeURIComponent(agentID) + '/fs/' + what + '?path=' + encodeURIComponent(path); + + // Sends a file to the client in pieces: each a request of its own (a proxy + // in front takes only so much at once), a few in flight together so a good + // link stays busy, and one the link drops is sent again alone. onFrac gets + // how much has gone; the answer is the client's, once it has written it. + const SEND_WORKERS = 4; + async function sendFile(path, blob, onFrac) { + let st; + try { + st = await fetch(fsURL('upload', path) + '&size=' + blob.size, {method: 'POST'}).then(r => r.json()); + } catch (e) { return {ok: false, error: 'ошибка сети'}; } + if (!st.ok) return st; + const {id, chunk} = st.data; + const xurl = '/api/panel/fs/xfer/' + encodeURIComponent(id); + const n = Math.max(1, Math.ceil(blob.size / chunk)); + const sent = new Array(n).fill(0); + const report = () => onFrac && onFrac(blob.size ? Math.min(1, sent.reduce((a, b) => a + b, 0) / blob.size) : 1); + let next = 0, failed = null; + const putPiece = (i, part) => new Promise(resolve => { + const x = new XMLHttpRequest(); + x.open('PUT', xurl + '/' + i); + x.setRequestHeader('Content-Type', 'application/octet-stream'); + x.upload.onprogress = ev => { sent[i] = ev.loaded; report(); }; + x.onload = () => resolve({status: x.status, text: x.responseText}); + x.onerror = x.ontimeout = () => resolve({status: 0}); + x.timeout = 5 * 60 * 1000; + x.send(part); + }); + async function worker() { + while (!failed && next < n) { + const i = next++, part = blob.slice(i * chunk, (i + 1) * chunk); + for (let tries = 0; !failed; ) { + const r = await putPiece(i, part); + if (r.status === 204) { sent[i] = part.size; report(); break; } + if (r.status === 410 || r.status === 400) { failed = (r.text || '').trim() || 'передача прервана'; break; } + sent[i] = 0; report(); + if (r.status === 503) continue; // the client is busy: just again + if (++tries >= 30) { failed = 'сеть не держит передачу'; break; } + await new Promise(res => setTimeout(res, Math.min(tries, 10) * 1000)); + } + } + } + await Promise.all(Array.from({length: Math.min(SEND_WORKERS, n)}, worker)); + if (failed) { + fetch(xurl + '/cancel', {method: 'POST'}).catch(() => {}); + return {ok: false, error: failed}; + } + for (let tries = 0; ; ) { + try { + const r = await fetch(xurl + '/result', {cache: 'no-store'}).then(r => r.json()); + if (r.done) return r.result; + tries = 0; + } catch (e) { + if (++tries >= 30) return {ok: false, error: 'нет ответа о записи файла'}; + await new Promise(res => setTimeout(res, 2000)); + } + } + } // Writes a file as it is, with no base64: the request body is the file. async function putFile(path, body) { // The save button turns into a small spinner meanwhile, and takes no @@ -408,10 +466,8 @@ btn.disabled = true; setBusy(true); try { - const r = await fetch(fsURL('upload', path), {method: 'POST', headers: {'Content-Type': 'application/octet-stream'}, body}); - return (r.headers.get('Content-Type') || '').startsWith('application/json') ? await r.json() : {ok: false, error: 'HTTP ' + r.status}; - } catch (e) { return {ok: false, error: 'ошибка сети'}; } - finally { setBusy(false); btn.classList.remove('saving'); btn.disabled = false; } + return await sendFile(path, new Blob([body])); + } finally { setBusy(false); btn.classList.remove('saving'); btn.disabled = false; } } const IMG = /\.(png|jpe?g|gif|webp|bmp|ico|avif|svg)$/i, VID = /\.(mp4|m4v|webm|ogv|mov)$/i, AUD = /\.(mp3|wav|ogg|oga|m4a|aac|flac|opus)$/i; @@ -960,22 +1016,10 @@ $('busy').onclick = () => { if (!uploads.length) return; $('updlg').showModal(); renderUploads(); }; $('updlg-close').onclick = () => $('updlg').close(); - // XHR rather than fetch: it reports how much of the request has gone out. - // The body is the file itself, which the panel passes on to the client as - // it comes, so the bar follows the whole way. + // The pieces go on to the client as they come, so the bar follows the + // whole way. const UP_CAP = 512 * 1024 * 1024; - function writeWithProgress(path, file, onFrac) { - return new Promise(resolve => { - const x = new XMLHttpRequest(); - x.open('POST', fsURL('upload', path)); - x.setRequestHeader('Content-Type', 'application/octet-stream'); - x.upload.onprogress = ev => { if (ev.lengthComputable) onFrac(ev.loaded / ev.total); }; - x.upload.onload = () => onFrac(1); - x.onload = () => { try { resolve(JSON.parse(x.responseText)); } catch (e) { resolve({ok: false, error: 'HTTP ' + x.status}); } }; - x.onerror = () => resolve({ok: false, error: 'ошибка сети'}); - x.send(file); - }); - } + const writeWithProgress = (path, file, onFrac) => sendFile(path, file, onFrac); async function uploadFiles(files) { clearTimeout(upDoneTimer);