Initial commit: haul — desktop SFTP client
A two-pane local/remote file browser over SSH, split out of dial so file transfer lives in its own tool. The two share ~/.ssh/config as their base. - internal/sshconf: ~/.ssh/config parsing, for the saved-host list. - internal/sshx: dialling and auth — agent, keys, password, host-key verification, ProxyJump. - internal/vfs: the FS interface both sides are driven through, implemented over the local disk and over SFTP. - internal/xfer: recursive copy engine with progress reporting, cancellation, conflict policy, and resume of partial transfers. - internal/ui: the Fyne GUI — connect window (saved hosts plus quick connect) and the two-pane browser session. Build with `make build`; the Makefile locates the Go toolchain at ~/.local/go, which is deliberately kept off PATH. Cross-compiling needs a per-target C toolchain because Fyne is cgo, so `make dist` builds the native binary only — the alternatives are documented in the Makefile. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,268 @@
|
||||
// Package xfer implements haul's SFTP file-transfer engine: recursive push/pull
|
||||
// with aggregate progress, per-file overwrite policy, and same-session resume.
|
||||
//
|
||||
// Push (local→remote) and pull (remote→local) share one recursive copy routine
|
||||
// by abstracting each side behind vfs.FS, so the tree-walking, resume, and
|
||||
// conflict logic is written once and exercised in both directions.
|
||||
package xfer
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"os"
|
||||
"sync/atomic"
|
||||
|
||||
"github.com/pkg/sftp"
|
||||
|
||||
"gitea.apointless.space/bsncubed/haul/internal/vfs"
|
||||
)
|
||||
|
||||
// ErrCanceled is returned by Push/Pull when Cancel was called mid-transfer.
|
||||
// Files already copied are left in place; the partial file is left on disk so
|
||||
// that re-running the same transfer on this Engine resumes it.
|
||||
var ErrCanceled = errors.New("transfer canceled")
|
||||
|
||||
// Conflict is the decision for a destination file that already existed before
|
||||
// this transfer began.
|
||||
type Conflict int
|
||||
|
||||
const (
|
||||
Overwrite Conflict = iota // replace the existing file
|
||||
Skip // leave the existing file, skip the source
|
||||
Rename // write the source under a non-colliding name
|
||||
)
|
||||
|
||||
// Progress is a snapshot of an in-flight transfer, reported to OnProgress.
|
||||
type Progress struct {
|
||||
CurrentName string // file currently copying
|
||||
FilesDone int
|
||||
FilesTotal int
|
||||
BytesDone int64 // cumulative across all files
|
||||
BytesTotal int64
|
||||
}
|
||||
|
||||
// ConflictFunc decides what to do about a pre-existing destination file. name
|
||||
// is the destination path. Returning the same decision for every call is the
|
||||
// "apply to all" behaviour.
|
||||
type ConflictFunc func(name string) Conflict
|
||||
|
||||
// ProgressFunc receives periodic progress updates. It must not block.
|
||||
type ProgressFunc func(Progress)
|
||||
|
||||
// Engine performs transfers over one SFTP connection. A single Engine instance
|
||||
// remembers which destination files it has created this session, which is what
|
||||
// makes same-session resume work: re-running an interrupted transfer resumes
|
||||
// partial files it started rather than treating them as pre-existing conflicts.
|
||||
type Engine struct {
|
||||
remote vfs.FS
|
||||
OnProgress ProgressFunc
|
||||
OnConflict ConflictFunc
|
||||
|
||||
created map[string]bool // destination paths this session started/finished
|
||||
canceled atomic.Bool // set by Cancel, cleared at the start of each run
|
||||
}
|
||||
|
||||
// New returns an Engine bound to an SFTP client.
|
||||
func New(c *sftp.Client) *Engine {
|
||||
return &Engine{remote: vfs.Remote{C: c}, created: map[string]bool{}}
|
||||
}
|
||||
|
||||
// Cancel asks an in-flight transfer to stop. It is safe to call from another
|
||||
// goroutine (the GUI's stop button) and takes effect at the next file or write
|
||||
// boundary; Push/Pull then return ErrCanceled.
|
||||
func (e *Engine) Cancel() { e.canceled.Store(true) }
|
||||
|
||||
// Push copies local sources into the remote directory (recursively for dirs).
|
||||
func (e *Engine) Push(localSources []string, remoteDir string) error {
|
||||
return e.run(vfs.Local{}, localSources, e.remote, remoteDir)
|
||||
}
|
||||
|
||||
// Pull copies remote sources into the local directory (recursively for dirs).
|
||||
func (e *Engine) Pull(remoteSources []string, localDir string) error {
|
||||
return e.run(e.remote, remoteSources, vfs.Local{}, localDir)
|
||||
}
|
||||
|
||||
func (e *Engine) run(src vfs.FS, sources []string, dst vfs.FS, dstDir string) error {
|
||||
e.canceled.Store(false)
|
||||
prog := Progress{}
|
||||
// First pass: count files and total bytes for aggregate progress.
|
||||
for _, s := range sources {
|
||||
n, bytes, err := countTree(src, s)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
prog.FilesTotal += n
|
||||
prog.BytesTotal += bytes
|
||||
}
|
||||
if err := dst.MkdirAll(dstDir); err != nil {
|
||||
return err
|
||||
}
|
||||
for _, s := range sources {
|
||||
target := dst.Join(dstDir, src.Base(s))
|
||||
if err := e.copyTree(src, s, dst, target, &prog); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// countTree returns the file count and total byte size under p.
|
||||
func countTree(src vfs.FS, p string) (int, int64, error) {
|
||||
fi, err := src.Stat(p)
|
||||
if err != nil {
|
||||
return 0, 0, err
|
||||
}
|
||||
if !fi.IsDir() {
|
||||
return 1, fi.Size(), nil
|
||||
}
|
||||
entries, err := src.ReadDir(p)
|
||||
if err != nil {
|
||||
return 0, 0, err
|
||||
}
|
||||
var files int
|
||||
var bytes int64
|
||||
for _, ent := range entries {
|
||||
n, b, err := countTree(src, src.Join(p, ent.Name()))
|
||||
if err != nil {
|
||||
return 0, 0, err
|
||||
}
|
||||
files += n
|
||||
bytes += b
|
||||
}
|
||||
return files, bytes, nil
|
||||
}
|
||||
|
||||
func (e *Engine) copyTree(src vfs.FS, srcPath string, dst vfs.FS, dstPath string, prog *Progress) error {
|
||||
if e.canceled.Load() {
|
||||
return ErrCanceled
|
||||
}
|
||||
fi, err := src.Stat(srcPath)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if fi.IsDir() {
|
||||
if err := dst.MkdirAll(dstPath); err != nil {
|
||||
return err
|
||||
}
|
||||
entries, err := src.ReadDir(srcPath)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
for _, ent := range entries {
|
||||
child := dst.Join(dstPath, ent.Name())
|
||||
if err := e.copyTree(src, src.Join(srcPath, ent.Name()), dst, child, prog); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
return e.copyFile(src, srcPath, fi.Size(), dst, dstPath, prog)
|
||||
}
|
||||
|
||||
// copyFile copies a single regular file, honouring conflict policy for
|
||||
// pre-existing destinations and resuming partials this session started.
|
||||
func (e *Engine) copyFile(src vfs.FS, srcPath string, srcSize int64, dst vfs.FS, dstPath string, prog *Progress) error {
|
||||
var offset int64
|
||||
|
||||
if info, err := dst.Stat(dstPath); err == nil {
|
||||
if e.created[dstPath] {
|
||||
// Our own partial from an interrupted attempt: resume from its end.
|
||||
if info.Size() <= srcSize {
|
||||
offset = info.Size()
|
||||
}
|
||||
} else {
|
||||
// Pre-existing file: ask the caller what to do.
|
||||
switch e.conflict(dstPath) {
|
||||
case Skip:
|
||||
prog.FilesDone++
|
||||
prog.BytesDone += srcSize
|
||||
e.report(prog, dst.Base(dstPath))
|
||||
return nil
|
||||
case Rename:
|
||||
dstPath = uniqueName(dst, dstPath)
|
||||
case Overwrite:
|
||||
// fall through; Create truncates.
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
in, err := src.Open(srcPath)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer in.Close()
|
||||
|
||||
var out io.WriteCloser
|
||||
if offset > 0 {
|
||||
if _, err := in.Seek(offset, io.SeekStart); err != nil {
|
||||
return err
|
||||
}
|
||||
if out, err = dst.OpenAt(dstPath, offset); err != nil {
|
||||
return err
|
||||
}
|
||||
} else if out, err = dst.Create(dstPath); err != nil {
|
||||
return err
|
||||
}
|
||||
e.created[dstPath] = true
|
||||
|
||||
prog.BytesDone += offset // resumed bytes already count as done
|
||||
pw := &progressWriter{w: out, prog: prog, name: dst.Base(dstPath), report: e.report, canceled: &e.canceled}
|
||||
if _, err := io.Copy(pw, in); err != nil {
|
||||
out.Close()
|
||||
return fmt.Errorf("copying %s: %w", srcPath, err)
|
||||
}
|
||||
if err := out.Close(); err != nil {
|
||||
return err
|
||||
}
|
||||
prog.FilesDone++
|
||||
e.report(prog, dst.Base(dstPath))
|
||||
return nil
|
||||
}
|
||||
|
||||
func (e *Engine) conflict(name string) Conflict {
|
||||
if e.OnConflict == nil {
|
||||
return Overwrite
|
||||
}
|
||||
return e.OnConflict(name)
|
||||
}
|
||||
|
||||
func (e *Engine) report(prog *Progress, name string) {
|
||||
if e.OnProgress == nil {
|
||||
return
|
||||
}
|
||||
p := *prog
|
||||
p.CurrentName = name
|
||||
e.OnProgress(p)
|
||||
}
|
||||
|
||||
// uniqueName returns dstPath with a numeric suffix that doesn't yet exist.
|
||||
func uniqueName(dst vfs.FS, dstPath string) string {
|
||||
for i := 1; ; i++ {
|
||||
cand := fmt.Sprintf("%s.%d", dstPath, i)
|
||||
if _, err := dst.Stat(cand); errors.Is(err, os.ErrNotExist) || err != nil {
|
||||
return cand
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// progressWriter tallies bytes as they're written and emits progress updates.
|
||||
// It is also where a cancel lands mid-file, so a single large file can be
|
||||
// interrupted rather than only being stoppable between files.
|
||||
type progressWriter struct {
|
||||
w io.Writer
|
||||
prog *Progress
|
||||
name string
|
||||
report func(*Progress, string)
|
||||
canceled *atomic.Bool
|
||||
}
|
||||
|
||||
func (p *progressWriter) Write(b []byte) (int, error) {
|
||||
if p.canceled != nil && p.canceled.Load() {
|
||||
return 0, ErrCanceled
|
||||
}
|
||||
n, err := p.w.Write(b)
|
||||
p.prog.BytesDone += int64(n)
|
||||
p.report(p.prog, p.name)
|
||||
return n, err
|
||||
}
|
||||
@@ -0,0 +1,160 @@
|
||||
package xfer
|
||||
|
||||
import (
|
||||
"io"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
|
||||
"github.com/pkg/sftp"
|
||||
)
|
||||
|
||||
// fakeConn adapts two pipes into an io.ReadWriteCloser for the SFTP server.
|
||||
type fakeConn struct {
|
||||
io.Reader
|
||||
io.WriteCloser
|
||||
}
|
||||
|
||||
// newTestEngine spins up an in-memory SFTP server over pipes and returns an
|
||||
// Engine talking to it, so transfers can be exercised without a real sshd.
|
||||
func newTestEngine(t *testing.T) *Engine {
|
||||
t.Helper()
|
||||
// OS pipes (kernel-buffered) avoid the handshake deadlock that unbuffered
|
||||
// io.Pipe causes between the SFTP client and server.
|
||||
srvR, cliW, err := os.Pipe() // client writes -> server reads
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
cliR, srvW, err := os.Pipe() // server writes -> client reads
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
server, err := sftp.NewServer(fakeConn{srvR, srvW})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
go server.Serve()
|
||||
|
||||
client, err := sftp.NewClientPipe(cliR, cliW)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
// Close the server first so the client's recv goroutine sees EOF and
|
||||
// client.Close() can return (otherwise the two block each other).
|
||||
t.Cleanup(func() { server.Close(); client.Close() })
|
||||
return New(client)
|
||||
}
|
||||
|
||||
func write(t *testing.T, p, content string) {
|
||||
t.Helper()
|
||||
if err := os.MkdirAll(filepath.Dir(p), 0o755); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := os.WriteFile(p, []byte(content), 0o644); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
func read(t *testing.T, p string) string {
|
||||
t.Helper()
|
||||
b, err := os.ReadFile(p)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return string(b)
|
||||
}
|
||||
|
||||
func TestPushPullRecursive(t *testing.T) {
|
||||
e := newTestEngine(t)
|
||||
local := t.TempDir()
|
||||
remote := t.TempDir()
|
||||
|
||||
// A directory tree on the local side.
|
||||
write(t, filepath.Join(local, "proj", "a.txt"), "alpha")
|
||||
write(t, filepath.Join(local, "proj", "sub", "b.txt"), "bravo")
|
||||
|
||||
var last Progress
|
||||
e.OnProgress = func(p Progress) { last = p }
|
||||
|
||||
if err := e.Push([]string{filepath.Join(local, "proj")}, remote); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if got := read(t, filepath.Join(remote, "proj", "a.txt")); got != "alpha" {
|
||||
t.Errorf("a.txt = %q", got)
|
||||
}
|
||||
if got := read(t, filepath.Join(remote, "proj", "sub", "b.txt")); got != "bravo" {
|
||||
t.Errorf("b.txt = %q", got)
|
||||
}
|
||||
if last.FilesDone != 2 || last.FilesTotal != 2 {
|
||||
t.Errorf("files done/total = %d/%d, want 2/2", last.FilesDone, last.FilesTotal)
|
||||
}
|
||||
if last.BytesDone != last.BytesTotal || last.BytesTotal != 10 {
|
||||
t.Errorf("bytes done/total = %d/%d, want 10/10", last.BytesDone, last.BytesTotal)
|
||||
}
|
||||
|
||||
// Pull it back into a fresh local dir.
|
||||
local2 := t.TempDir()
|
||||
if err := e.Pull([]string{filepath.Join(remote, "proj")}, local2); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if got := read(t, filepath.Join(local2, "proj", "sub", "b.txt")); got != "bravo" {
|
||||
t.Errorf("pulled b.txt = %q", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestResumePartial(t *testing.T) {
|
||||
e := newTestEngine(t)
|
||||
local := t.TempDir()
|
||||
remote := t.TempDir()
|
||||
|
||||
write(t, filepath.Join(local, "big.txt"), "hello world") // 11 bytes
|
||||
// Simulate an interrupted prior attempt: a 5-byte partial that THIS engine
|
||||
// created, so it should resume (append) rather than restart or conflict.
|
||||
dst := filepath.Join(remote, "big.txt")
|
||||
write(t, dst, "hello")
|
||||
e.created[dst] = true
|
||||
|
||||
if err := e.Push([]string{filepath.Join(local, "big.txt")}, remote); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if got := read(t, dst); got != "hello world" {
|
||||
t.Errorf("resume result = %q, want 'hello world'", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestConflictPolicies(t *testing.T) {
|
||||
cases := []struct {
|
||||
name string
|
||||
policy Conflict
|
||||
wantDst string // content at original dst path
|
||||
wantCopy string // content at dst.1 (only for Rename), "" if none
|
||||
}{
|
||||
{"skip", Skip, "OLD", ""},
|
||||
{"overwrite", Overwrite, "NEW", ""},
|
||||
{"rename", Rename, "OLD", "NEW"},
|
||||
}
|
||||
for _, tc := range cases {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
e := newTestEngine(t)
|
||||
local := t.TempDir()
|
||||
remote := t.TempDir()
|
||||
write(t, filepath.Join(local, "f.txt"), "NEW")
|
||||
dst := filepath.Join(remote, "f.txt")
|
||||
write(t, dst, "OLD") // pre-existing, engine did NOT create it
|
||||
e.OnConflict = func(string) Conflict { return tc.policy }
|
||||
|
||||
if err := e.Push([]string{filepath.Join(local, "f.txt")}, remote); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if got := read(t, dst); got != tc.wantDst {
|
||||
t.Errorf("dst = %q, want %q", got, tc.wantDst)
|
||||
}
|
||||
if tc.wantCopy != "" {
|
||||
if got := read(t, dst+".1"); got != tc.wantCopy {
|
||||
t.Errorf("renamed copy = %q, want %q", got, tc.wantCopy)
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user