From 64ce8f2b72c995486e84bdc9e7326eb94ac42ec4 Mon Sep 17 00:00:00 2001 From: iRedTea Date: Thu, 15 Oct 2026 22:05:04 +0300 Subject: [PATCH] fix: ended page scopes start no timers; SSH transfers keep their connection from the idle sweep; linked files download whole; only the sent item unpacks on an SSH host, replace and skip as on a machine --- internal/agent/ship.go | 8 +- internal/agent/sshfs.go | 100 +++++++++++++++------- internal/agent/sshfs_test.go | 65 ++++++++++++++ internal/server/web/static/app.css | 1 + internal/server/web/static/app.js | 19 +--- internal/server/web/static/wsutil.js | 24 +++++- internal/server/web/static/wsutil.test.js | 10 +++ 7 files changed, 175 insertions(+), 52 deletions(-) diff --git a/internal/agent/ship.go b/internal/agent/ship.go index b4ff13a..e07d7be 100644 --- a/internal/agent/ship.go +++ b/internal/agent/ship.go @@ -366,7 +366,7 @@ func (a *Agent) doShipIn(ctx context.Context, env proto.TaskEnvelope) proto.Task } }() if conn != nil { - err := a.unpackRemote(conn, pr, dir, args.Name, args.Conflict) + skipped, err := a.unpackRemote(conn, pr, dir, args.Name, args.Conflict) if err == nil { _, _ = io.Copy(io.Discard, pr) } @@ -374,7 +374,11 @@ func (a *Agent) doShipIn(ctx context.Context, env proto.TaskEnvelope) proto.Task if err != nil { return proto.TaskResult{Error: err.Error()} } - data, _ := json.Marshal(map[string]int{}) + res := map[string]int{} + if skipped { + res["skipped"] = 1 + } + data, _ := json.Marshal(res) return proto.TaskResult{OK: true, Data: data} } u := &unpacker{dir: dir, name: args.Name, conflict: args.Conflict} diff --git a/internal/agent/sshfs.go b/internal/agent/sshfs.go index 4fa5e9e..be496ea 100644 --- a/internal/agent/sshfs.go +++ b/internal/agent/sshfs.go @@ -39,6 +39,28 @@ type sshfsConn struct { statFmt string // "gnu", "bsd" or "ls" home string last time.Time + busy int // transfers under way: the sweep leaves the connection be +} + +// hold marks the connection busy for a long transfer; the returned func ends +// it and starts the idle clock again. +func (c *sshfsConn) hold() func() { + c.mu.Lock() + c.busy++ + c.mu.Unlock() + return func() { + c.mu.Lock() + c.busy-- + c.last = time.Now() + c.mu.Unlock() + } +} + +// idle: nothing under way and unused for sshfsIdle by now. +func (c *sshfsConn) idle(now time.Time) bool { + c.mu.Lock() + defer c.mu.Unlock() + return c.busy == 0 && now.Sub(c.last) > sshfsIdle } var ( @@ -151,11 +173,7 @@ func hasLine(lines []string, w string) bool { 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 { + if v.(*sshfsConn).idle(time.Now()) { sshfsClose(k.(string)) } return true @@ -410,14 +428,20 @@ func writeCommand(p string) string { 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) +// sizeCommand tells the size of p - of its target, if a link: what cat +// reads. +func sizeCommand(statFmt, p string) string { + switch statFmt { + case "bsd": + return "stat -L -f %z -- " + shq(p) + case "ls": + return "wc -c < " + shq(p) } - out, err := a.sshfsRun(c, cmd, nil) + return "stat -L -c %s -- " + shq(p) +} + +func (a *Agent) sshfsSize(c *sshfsConn, p string) (int64, error) { + out, err := a.sshfsRun(c, sizeCommand(c.statFmt, p), nil) if err != nil { return 0, err } @@ -483,6 +507,7 @@ func (a *Agent) doSSHFSSend(ctx context.Context, env proto.TaskEnvelope) proto.T if err != nil { return proto.TaskResult{Error: err.Error()} } + defer c.hold()() size, err := a.sshfsSize(c, args.Path) if err != nil { return proto.TaskResult{Error: err.Error()} @@ -518,7 +543,8 @@ func (a *Agent) doSSHFSSend(ctx context.Context, env proto.TaskEnvelope) proto.T return proto.TaskResult{Error: "передача прервана: " + err.Error()} } } - _ = s.Wait() + // All of it is read; the session closes (deferred) without waiting on a + // cat that may have more to say. data, _ := json.Marshal(map[string]any{"path": args.Path, "size": size}) return proto.TaskResult{OK: true, Data: data} } @@ -534,6 +560,7 @@ func (a *Agent) doSSHFSPut(ctx context.Context, env proto.TaskEnvelope) proto.Ta if err != nil { return proto.TaskResult{Error: err.Error()} } + defer c.hold()() s, err := a.sshfsSession(c) if err != nil { return proto.TaskResult{Error: err.Error()} @@ -598,7 +625,8 @@ func renameTar(w io.Writer, r io.Reader, base, name string) error { if i := strings.IndexByte(n, '/'); i >= 0 { top, rest = n[:i], n[i:] } - if top != base { + // Only base and what is under it - nothing that climbs out by "..". + if top != base || strings.Contains("/"+rest+"/", "/../") { continue } h.Name = name + rest @@ -615,6 +643,7 @@ func renameTar(w io.Writer, r io.Reader, base, name string) error { // packRemote writes src on the connection's host as a tar stream named name, // counting what goes for the progress. func (a *Agent) packRemote(ctx context.Context, w io.Writer, c *sshfsConn, src, name string, m *shipMeter) error { + defer c.hold()() s, err := a.sshfsSession(c) if err != nil { return err @@ -657,25 +686,34 @@ func (a *Agent) remoteSize(c *sshfsConn, src string) int64 { // unpackRemote puts a tar stream into dir on the connection's host, by the // conflict rule: tar keeps or replaces what is there, an "error" rule // refuses when name is there already. -func (a *Agent) unpackRemote(c *sshfsConn, r io.Reader, dir, name, conflict string) error { +func (a *Agent) unpackRemote(c *sshfsConn, r io.Reader, dir, name, conflict string) (skipped bool, err error) { + defer c.hold()() target := strings.TrimSuffix(dir, "/") + "/" + name out, _ := a.sshfsRun(c, "test -e "+shq(target)+" -o -L "+shq(target)+" && echo LPRA_EXISTS; tar --version 2>/dev/null | head -n 1", nil) - if strings.Contains(string(out), "LPRA_EXISTS") && conflict != "replace" && conflict != "skip" { - return errors.New(target + " уже существует") - } - gnu := strings.Contains(string(out), "GNU tar") - flags := "" - switch conflict { - case "skip": - flags = " -k" - if gnu { - flags = " --skip-old-files" - } - case "replace": - if gnu { - flags = " --overwrite" + if strings.Contains(string(out), "LPRA_EXISTS") { + switch conflict { + case "skip": + // Left as it was, as on a machine. + return true, nil + case "replace": + default: + return false, errors.New(target + " уже существует") } } - _, err := a.sshfsRun(c, "mkdir -p -- "+shq(dir)+" && cd -- "+shq(dir)+" && tar -xf -"+flags, r) - return err + // Only name and what is under it reaches the host's tar. + pr, pw := io.Pipe() + go func() { pw.CloseWithError(renameTar(pw, r, name, name)) }() + defer pr.Close() + _, err = a.sshfsRun(c, unpackCommand(dir, name, conflict, strings.Contains(string(out), "GNU tar")), pr) + return false, err +} + +// unpackCommand unpacks a tar stream of name into dir; "replace" removes +// what is there first, so the thing is replaced whole as on a machine. +func unpackCommand(dir, name, conflict string, gnu bool) string { + cmd := "mkdir -p -- " + shq(dir) + " && cd -- " + shq(dir) + " && tar -xf -" + if conflict == "replace" { + cmd = "rm -rf -- " + shq(strings.TrimSuffix(dir, "/")+"/"+name) + " && " + cmd + } + return cmd } diff --git a/internal/agent/sshfs_test.go b/internal/agent/sshfs_test.go index f269012..537bc8b 100644 --- a/internal/agent/sshfs_test.go +++ b/internal/agent/sshfs_test.go @@ -6,6 +6,7 @@ import ( "io" "strings" "testing" + "time" ) func TestParseStatLine(t *testing.T) { @@ -127,3 +128,67 @@ func TestZipCommand(t *testing.T) { t.Fatalf("unzip: %s", got) } } + +// A connection busy with a long transfer is not idle, however long ago it +// was opened; once the transfer ends the idle clock starts again. +func TestSweepSparesBusyConn(t *testing.T) { + c := &sshfsConn{last: time.Now().Add(-2 * sshfsIdle)} + release := c.hold() + if c.idle(time.Now()) { + t.Fatal("a busy connection was taken for idle") + } + release() + if c.idle(time.Now()) { + t.Fatal("just released: not idle yet") + } + if !c.idle(time.Now().Add(2 * sshfsIdle)) { + t.Fatal("unused for long: idle") + } +} + +// A file's size is its target's when it is a link: what cat will read. +func TestSizeCommandFollowsLinks(t *testing.T) { + for _, f := range []string{"gnu", "bsd"} { + if cmd := sizeCommand(f, "/x"); !strings.Contains(cmd, "stat -L ") { + t.Fatalf("%s: %s", f, cmd) + } + } +} + +// Nothing of a tar stream lands outside name: other entries and ".." paths +// are dropped. +func TestRenameTarKeepsInsideName(t *testing.T) { + var src bytes.Buffer + tw := tar.NewWriter(&src) + for _, n := range []string{"d/a", "d/../../etc/x", ".ssh/authorized_keys", "d/b"} { + _ = tw.WriteHeader(&tar.Header{Name: n, Mode: 0o644, Typeflag: tar.TypeReg}) + } + _ = tw.Close() + var out bytes.Buffer + if err := renameTar(&out, &src, "d", "d"); err != nil { + t.Fatal(err) + } + tr := tar.NewReader(&out) + var got []string + for { + h, err := tr.Next() + if err != nil { + break + } + got = append(got, h.Name) + } + if strings.Join(got, ",") != "d/a,d/b" { + t.Fatalf("entries: %v", got) + } +} + +// Into an SSH host, "replace" replaces the thing whole, as on a machine. +func TestUnpackCommandReplaces(t *testing.T) { + cmd := unpackCommand("/srv", "d", "replace", true) + if !strings.HasPrefix(cmd, "rm -rf -- '/srv/d' && ") || strings.Contains(cmd, "--overwrite") { + t.Fatalf("replace: %s", cmd) + } + if cmd := unpackCommand("/srv", "d", "error", false); cmd != "mkdir -p -- '/srv' && cd -- '/srv' && tar -xf -" { + t.Fatalf("error: %s", cmd) + } +} diff --git a/internal/server/web/static/app.css b/internal/server/web/static/app.css index d6017fb..799e1a7 100644 --- a/internal/server/web/static/app.css +++ b/internal/server/web/static/app.css @@ -780,6 +780,7 @@ p.mv-a { margin:0; font-size:13px; } .fpane.narrow .fm-bar { gap:5px; } .fpane.narrow .fm-bar .btn.ico { padding-left:8px; padding-right:8px; gap:4px; font-size:12.5px; } .fpane.narrow .fp-sel { gap:4px; } +.fpane.narrow .fp-selinfo { margin-right:6px; } .fpane.narrow .fp-sel .btn.sq { width:32px; height:32px; } /* The pane: its bar and list scroll inside it. */ .fpane-body { flex:1; min-height:0; overflow:auto; padding:10px 12px 12px; } diff --git a/internal/server/web/static/app.js b/internal/server/web/static/app.js index 6902afe..11a3565 100644 --- a/internal/server/web/static/app.js +++ b/internal/server/web/static/app.js @@ -15,24 +15,7 @@ function register(name, mod) { modules[name] = mod; } - // scope gives a module timers, requests and listeners that end with it. - function scope() { - const timers = new Set(), offs = [], ac = new AbortController(); - return { - timeout(fn, ms) { const id = setTimeout(() => { timers.delete(id); fn(); }, ms); timers.add(id); return id; }, - interval(fn, ms) { const id = setInterval(fn, ms); timers.add(id); return id; }, - clear(id) { clearTimeout(id); clearInterval(id); timers.delete(id); }, - fetch(url, opt) { - const o = Object.assign({credentials: 'same-origin'}, opt || {}); - // A request with its own signal ends on either. - o.signal = o.signal && AbortSignal.any ? AbortSignal.any([o.signal, ac.signal]) : (o.signal || ac.signal); - return fetch(url, o); - }, - on(target, ev, fn, opt) { target.addEventListener(ev, fn, opt); offs.push(() => target.removeEventListener(ev, fn, opt)); }, - signal: ac.signal, - end() { timers.forEach(id => { clearTimeout(id); clearInterval(id); }); timers.clear(); offs.forEach(f => f()); ac.abort(); } - }; - } + const scope = WSUtil.scope; function unmount(area) { const m = mounted.get(area); diff --git a/internal/server/web/static/wsutil.js b/internal/server/web/static/wsutil.js index 9a6accc..95a18e0 100644 --- a/internal/server/web/static/wsutil.js +++ b/internal/server/web/static/wsutil.js @@ -22,6 +22,28 @@ if (n < 1073741824) return (n / 1048576).toFixed(1) + ' M'; return (n / 1073741824).toFixed(1) + ' G'; } - const api = {shellQuote, shellQuoteAll, dropAction, fmtSize}; + // scope gives a module timers, requests and listeners that end with it; + // once ended it starts nothing new (a request's late answer cannot set a + // timer going again). + function scope() { + let ended = false; + const timers = new Set(), offs = [], ac = new AbortController(); + return { + timeout(fn, ms) { if (ended) return 0; const id = setTimeout(() => { timers.delete(id); fn(); }, ms); timers.add(id); return id; }, + interval(fn, ms) { if (ended) return 0; const id = setInterval(fn, ms); timers.add(id); return id; }, + clear(id) { clearTimeout(id); clearInterval(id); timers.delete(id); }, + fetch(url, opt) { + const o = Object.assign({credentials: 'same-origin'}, opt || {}); + // A request with its own signal ends on either. + o.signal = o.signal && AbortSignal.any ? AbortSignal.any([o.signal, ac.signal]) : (o.signal || ac.signal); + return fetch(url, o); + }, + on(target, ev, fn, opt) { target.addEventListener(ev, fn, opt); offs.push(() => target.removeEventListener(ev, fn, opt)); }, + signal: ac.signal, + end() { ended = true; timers.forEach(id => { clearTimeout(id); clearInterval(id); }); timers.clear(); offs.forEach(f => f()); ac.abort(); } + }; + } + + const api = {shellQuote, shellQuoteAll, dropAction, fmtSize, scope}; if (typeof module !== 'undefined') module.exports = api; else root.WSUtil = api; })(this); diff --git a/internal/server/web/static/wsutil.test.js b/internal/server/web/static/wsutil.test.js index 5017830..de57c76 100644 --- a/internal/server/web/static/wsutil.test.js +++ b/internal/server/web/static/wsutil.test.js @@ -39,3 +39,13 @@ test('dropAction: other SSH user on the same host is another place', () => { const b = {kind: 'ssh', id: 'm1|bob@h:22', label: 'bob@h'}; assert.strictEqual(U.dropAction({source: a, paths: ['/x'], dir: '/'}, {kind: 'term', source: b}).kind, 'deny'); }); + +test('scope: nothing new runs after it ended', async () => { + const s = U.scope(); + let n = 0; + s.end(); + s.timeout(() => n++, 1); + s.interval(() => n++, 1); + await new Promise(r => setTimeout(r, 20)); + assert.strictEqual(n, 0); +});