Commit 5de0fe9

Antonio Mika  ·  2026-10-10 19:41:37 -0400 EDT
parent 38ed55f 
feat(send): support rsync merge filters and sftp renames
22 files changed,  +2501, -152
M pkg/apps/feeds/scp_hooks.go
M pkg/apps/pastes/api.go
A pkg/apps/pastes/api_test.go
M pkg/apps/pgs/rsync_storage_test.go
A pkg/apps/pgs/scp_sftp_test.go
M pkg/apps/pgs/uploader.go
M pkg/filehandlers/post_handler.go
A pkg/filehandlers/post_handler_test.go
M pkg/filehandlers/router_handler.go
M pkg/rsync/filter.go
M pkg/rsync/fs.go
M pkg/rsync/integration_test.go
M pkg/rsync/receiver.go
M pkg/rsync/sender.go
M pkg/rsync/server.go
M pkg/rsync/wire_test.go
M pkg/send/protocols/rsync/rsync.go
M pkg/send/protocols/scp/copy_from_client.go
M pkg/send/protocols/sftp/handler.go
M pkg/send/protocols/sftp/sftp.go
M pkg/send/protocols/sftp/writer.go
M pkg/send/utils/utils.go
M pkg/apps/feeds/scp_hooks.go
+2, -3
......@@ -37,9 +37,8 @@ func (p *FeedHooks) FileValidate(s *pssh.SSHServerConnSession, data *filehandler
3737 return false, err
3838 }
3939
40- // Because we need to support sshfs, sftp runs our Write handler twice
41- // and on the first pass we do not have access to the file data.
42- // In that case we should skip the parsing validation
40+ // SFTP validates a file when it is opened, before its contents
41+ // arrive, so there is nothing to parse yet.
4342 if data.Text == "" {
4443 return true, nil
4544 }
M pkg/apps/pastes/api.go
+2, -1
......@@ -192,11 +192,12 @@ func postHandler(w http.ResponseWriter, r *http.Request) {
192192 // we dont want to syntax highlight huge files
193193 if post.FileSize > 1*shared.MB {
194194 logger.Warn("paste too large to parse and apply syntax highlighting")
195- parsedText = post.Text
195+ parsedText = "<pre>" + template.HTMLEscapeString(post.Text) + "</pre>"
196196 } else {
197197 parsedText, err = ParseText(post.Filename, post.Text)
198198 if err != nil {
199199 logger.Error("could not parse text", "err", err)
200+ parsedText = "<pre>" + template.HTMLEscapeString(post.Text) + "</pre>"
200201 }
201202 if post.ExpiresAt != nil {
202203 expiresAt = post.ExpiresAt.Format(time.DateOnly)
A pkg/apps/pastes/api_test.go
+68, -0
......@@ -0,0 +1,68 @@
1+package pastes
2+
3+import (
4+ "context"
5+ "html/template"
6+ "log/slog"
7+ "net/http"
8+ "net/http/httptest"
9+ "strings"
10+ "testing"
11+ "time"
12+
13+ "github.com/picosh/pico/pkg/db"
14+ "github.com/picosh/pico/pkg/db/stub"
15+ "github.com/picosh/pico/pkg/shared"
16+ "github.com/picosh/pico/pkg/shared/router"
17+)
18+
19+type pasteDB struct {
20+ *stub.StubDB
21+ post *db.Post
22+}
23+
24+func (m *pasteDB) FindUserByName(string) (*db.User, error) {
25+ return &db.User{ID: "user-1", Name: "tester"}, nil
26+}
27+
28+func (m *pasteDB) FindPostWithSlug(string, string, string) (*db.Post, error) {
29+ return m.post, nil
30+}
31+
32+func TestLargePasteEscapesHTML(t *testing.T) {
33+ t.Chdir("../../..")
34+ logger := slog.New(slog.DiscardHandler)
35+ cfg := &shared.ConfigSite{
36+ Domain: "pastes.test", Protocol: "https", Space: "pastes", Logger: logger,
37+ }
38+ now := time.Now()
39+ text := strings.Repeat("x", shared.MB) + "\n<script>alert(1)</script>\n<img src=x onerror=alert(1)>\n& \" '"
40+ dbpool := &pasteDB{
41+ StubDB: stub.NewStubDB(logger),
42+ post: &db.Post{
43+ Filename: "test.html", Slug: "test.html", Username: "tester",
44+ Text: text, FileSize: len(text), PublishAt: &now, UpdatedAt: &now,
45+ },
46+ }
47+ api := &router.ApiConfig{Cfg: cfg, Dbpool: dbpool}
48+ handler := router.CreateServeBasic(createSubdomainRoutes(nil), api.CreateCtx(context.Background(), "tester"))
49+
50+ page := httptest.NewRecorder()
51+ handler(page, httptest.NewRequest(http.MethodGet, "https://tester.pastes.test/test.html", nil))
52+ if page.Code != http.StatusOK {
53+ t.Fatalf("page status = %d", page.Code)
54+ }
55+ want := "<pre>" + template.HTMLEscapeString(text) + "</pre>"
56+ if !strings.Contains(page.Body.String(), want) {
57+ t.Fatal("page does not contain the escaped paste in a pre element")
58+ }
59+ if strings.Contains(page.Body.String(), "<script>alert(1)</script>") {
60+ t.Fatal("page contains the uploaded script as executable HTML")
61+ }
62+
63+ raw := httptest.NewRecorder()
64+ handler(raw, httptest.NewRequest(http.MethodGet, "https://tester.pastes.test/test.html/raw", nil))
65+ if raw.Code != http.StatusOK || raw.Body.String() != text {
66+ t.Fatal("raw response does not preserve the uploaded text")
67+ }
68+}
M pkg/apps/pgs/rsync_storage_test.go
+130, -3
......@@ -4,6 +4,7 @@ import (
44 "bytes"
55 "crypto/rand"
66 "encoding/pem"
7+ "errors"
78 "fmt"
89 "io"
910 "log/slog"
......@@ -38,6 +39,10 @@ type rsyncEnv struct {
3839 bucket storage.Bucket
3940 rsh string
4041 feature *db.FeatureFlag
42+ // sshArgs are the client options scp and sftp need to reach the server.
43+ sshArgs []string
44+ addr string
45+ user UserSSH
4146 }
4247
4348 func newRsyncEnv(t *testing.T, st storage.StorageServe) *rsyncEnv {
......@@ -95,7 +100,8 @@ func newRsyncEnv(t *testing.T, st storage.StorageServe) *rsyncEnv {
95100 "ssh -p %s -i %s -o IdentitiesOnly=yes -o StrictHostKeyChecking=no -o UserKnownHostsFile=/dev/null -o LogLevel=ERROR",
96101 port, keyFile,
97102 )
98- return &rsyncEnv{t: t, st: st, bucket: bucket, rsh: rsh, feature: dbpool.Feature}
103+ sshArgs := []string{"-P", port, "-i", keyFile, "-o", "IdentitiesOnly=yes", "-o", "StrictHostKeyChecking=no", "-o", "UserKnownHostsFile=/dev/null", "-o", "LogLevel=ERROR"}
104+ return &rsyncEnv{t: t, st: st, bucket: bucket, rsh: rsh, feature: dbpool.Feature, sshArgs: sshArgs, addr: addr, user: user}
99105 }
100106
101107 // discardPubsub drops cache purges, which pgs sends from goroutines that can
......@@ -122,11 +128,24 @@ var statPatterns = map[string]*regexp.Regexp{
122128 }
123129
124130 func (e *rsyncEnv) rsync(args ...string) rsyncStats {
131+ e.t.Helper()
132+ return e.rsyncExit(0, args...)
133+}
134+
135+func (e *rsyncEnv) rsyncExit(want int, args ...string) rsyncStats {
125136 e.t.Helper()
126137 args = append([]string{"-e", e.rsh, "--stats"}, args...)
127138 out, err := exec.Command("rsync", args...).CombinedOutput()
139+ code := 0
128140 if err != nil {
129- e.t.Fatalf("rsync %q: %v\n%s", args, err, out)
141+ var exit *exec.ExitError
142+ if !errors.As(err, &exit) {
143+ e.t.Fatalf("rsync %q: %v\n%s", args, err, out)
144+ }
145+ code = exit.ExitCode()
146+ }
147+ if code != want {
148+ e.t.Fatalf("rsync %q: exit %d, want %d\n%s", args, code, want, out)
130149 }
131150 stat := func(name string) int {
132151 m := statPatterns[name].FindSubmatch(out)
......@@ -308,7 +327,7 @@ func TestRsyncStorage(t *testing.T) {
308327
309328 t.Run("oversized file is rejected", func(t *testing.T) {
310329 src.write("huge.bin", randomBytes(rsyncFileMax+1), t0)
311- s := env.rsync("-rt", src.dir+"/", remote)
330+ s := env.rsyncExit(23, "-rt", src.dir+"/", remote)
312331 if !strings.Contains(s.output, "huge.bin") {
313332 t.Fatalf("no warning about huge.bin\n%s", s.output)
314333 }
......@@ -397,6 +416,60 @@ func TestRsyncStorage(t *testing.T) {
397416 }
398417 }
399418
419+func TestRsyncStorageRejections(t *testing.T) {
420+ for name, newStorage := range rsyncStorageBackends() {
421+ t.Run(name, func(t *testing.T) {
422+ for _, reason := range []string{"ignore", "file limit", "storage quota"} {
423+ t.Run(reason, func(t *testing.T) {
424+ env := newRsyncEnv(t, newStorage(t))
425+ src := tree{t, t.TempDir()}
426+ remote := "localhost:/site"
427+ mtime := time.Unix(1_600_000_000, 0)
428+ wantExit := 23
429+ accepted := map[string]string{"z.txt": "ok"}
430+ switch reason {
431+ case "ignore":
432+ src.write("_pgs_ignore", []byte("a.txt\n"), mtime)
433+ env.rsync("-rt", src.dir+"/", remote)
434+ wantExit = 0
435+ case "file limit":
436+ env.feature.Data.FileMax = 3
437+ case "storage quota":
438+ src.write("z.txt", []byte("full"), mtime)
439+ env.rsync("-rt", src.dir+"/", remote)
440+ env.feature.Data.StorageMax = 4
441+ accepted["zz.txt"] = "x"
442+ src.write("zz.txt", []byte("x"), mtime)
443+ }
444+ src.write("a.txt", []byte("rejected"), mtime)
445+ src.write("z.txt", []byte("ok"), mtime)
446+ s := env.rsyncExit(wantExit, "-rt", "--remove-source-files", src.dir+"/", remote)
447+ if !strings.Contains(s.output, "a.txt") {
448+ t.Fatalf("missing per-file rejection message\n%s", s.output)
449+ }
450+ if reason == "ignore" && !strings.Contains(s.output, "a.txt skipped: file ignored: (/site/a.txt), https://pico.sh/pgs#-pgs-ignore\n") {
451+ t.Fatalf("unexpected ignore message\n%s", s.output)
452+ }
453+ if _, _, ok := env.stored("/site/a.txt"); ok {
454+ t.Fatal("rejected file was stored")
455+ }
456+ if data, _ := src.read("a.txt"); string(data) != "rejected" {
457+ t.Fatal("rejected source file changed")
458+ }
459+ for file, contents := range accepted {
460+ if data, _, ok := env.stored("/site/" + file); !ok || string(data) != contents {
461+ t.Errorf("accepted file %s not stored correctly", file)
462+ }
463+ if _, err := os.Stat(filepath.Join(src.dir, file)); !errors.Is(err, os.ErrNotExist) {
464+ t.Errorf("accepted file %s not removed from source: %v", file, err)
465+ }
466+ }
467+ })
468+ }
469+ })
470+ }
471+}
472+
400473 // TestRsyncStorageQuota checks that shrinking and deleting files frees quota
401474 // for files later in the same run.
402475 func TestRsyncStorageQuota(t *testing.T) {
......@@ -432,3 +505,57 @@ func TestRsyncStorageQuota(t *testing.T) {
432505 })
433506 }
434507 }
508+
509+// TestRsyncStorageMergeFilters checks rsync's per-directory filter files
510+// against pgs storage, where dotfiles are only stored when _pgs_ignore
511+// allows them.
512+func TestRsyncStorageMergeFilters(t *testing.T) {
513+ for name, newStorage := range rsyncStorageBackends() {
514+ t.Run(name, func(t *testing.T) {
515+ for _, allowDotfiles := range []bool{true, false} {
516+ t.Run(fmt.Sprintf("dotfiles=%v", allowDotfiles), func(t *testing.T) {
517+ env := newRsyncEnv(t, newStorage(t))
518+ src := tree{t, t.TempDir()}
519+ remote := "localhost:/site"
520+ mtime := time.Unix(1_600_000_000, 0)
521+ if allowDotfiles {
522+ src.write("_pgs_ignore", []byte("_private\n"), mtime)
523+ env.rsync("-rt", src.dir+"/", remote)
524+ }
525+ src.write(".rsync-filter", []byte("- keep.txt\n"), mtime)
526+ src.write("sub/.rsync-filter", []byte("- *.tmp\n"), mtime)
527+ for _, f := range []string{"index.html", "keep.txt", "stale.txt", "sub/a.tmp", "sub/b.html"} {
528+ src.write(f, []byte(f), mtime)
529+ }
530+ env.rsync("-rt", src.dir+"/", remote)
531+ _, _, stored := env.stored("/site/.rsync-filter")
532+ if stored != allowDotfiles {
533+ t.Fatalf(".rsync-filter stored=%v with dotfiles allowed=%v", stored, allowDotfiles)
534+ }
535+
536+ dst := tree{t, t.TempDir()}
537+ env.rsync("-rt", "-F", remote+"/", dst.dir+"/")
538+ for f, filtered := range map[string]bool{"keep.txt": true, "sub/a.tmp": true, "index.html": false, "sub/b.html": false} {
539+ _, err := os.Stat(filepath.Join(dst.dir, f))
540+ if got := err == nil; got != (!filtered || !allowDotfiles) {
541+ t.Errorf("download: %s present=%v", f, got)
542+ }
543+ }
544+
545+ for _, f := range []string{"keep.txt", "stale.txt", "sub/a.tmp"} {
546+ if err := os.Remove(filepath.Join(src.dir, f)); err != nil {
547+ t.Fatal(err)
548+ }
549+ }
550+ s := env.rsync("-rt", "-F", "--delete", src.dir+"/", remote)
551+ for f, protected := range map[string]bool{"keep.txt": true, "sub/a.tmp": true, "stale.txt": false} {
552+ _, _, ok := env.stored("/site/" + f)
553+ if ok != (protected && allowDotfiles) {
554+ t.Errorf("upload: %s stored=%v\n%s", f, ok, s.output)
555+ }
556+ }
557+ })
558+ }
559+ })
560+ }
561+}
A pkg/apps/pgs/scp_sftp_test.go
+632, -0
......@@ -0,0 +1,632 @@
1+package pgs
2+
3+import (
4+ "errors"
5+ "io"
6+ "os"
7+ "os/exec"
8+ "path/filepath"
9+ "strings"
10+ "testing"
11+ "time"
12+
13+ "github.com/pkg/sftp"
14+)
15+
16+// uploadModes upload the contents of a local directory to /site. OpenSSH's
17+// scp uses SFTP unless -O asks for the legacy protocol.
18+var uploadModes = map[string]func(e *rsyncEnv, dir string) *exec.Cmd{
19+ "scp-legacy": func(e *rsyncEnv, dir string) *exec.Cmd {
20+ return exec.Command("scp", append(append([]string{"-O", "-r"}, e.sshArgs...), dir+"/.", "localhost:/site")...)
21+ },
22+ "scp-sftp": func(e *rsyncEnv, dir string) *exec.Cmd {
23+ return exec.Command("scp", append(append([]string{"-r"}, e.sshArgs...), dir+"/.", "localhost:/site")...)
24+ },
25+ "sftp": func(e *rsyncEnv, dir string) *exec.Cmd {
26+ return e.sftp("-mkdir /site", "put -r "+dir+"/. /site")
27+ },
28+}
29+
30+func (e *rsyncEnv) sftp(cmds ...string) *exec.Cmd {
31+ cmd := exec.Command("sftp", append(append([]string{"-b", "-"}, e.sshArgs...), "localhost")...)
32+ cmd.Stdin = strings.NewReader(strings.Join(cmds, "\n") + "\n")
33+ return cmd
34+}
35+
36+func runCmd(t *testing.T, cmd *exec.Cmd) (string, int) {
37+ t.Helper()
38+ out, err := cmd.CombinedOutput()
39+ var exit *exec.ExitError
40+ switch {
41+ case errors.As(err, &exit):
42+ return string(out), exit.ExitCode()
43+ case err != nil:
44+ t.Fatalf("%s: %v", cmd.Path, err)
45+ }
46+ return string(out), 0
47+}
48+
49+func (e *rsyncEnv) assertStored(files map[string]string) {
50+ e.t.Helper()
51+ for name, want := range files {
52+ if got, _, ok := e.stored("/site/" + name); !ok || string(got) != want {
53+ e.t.Errorf("%s stored=%v contents=%q, want %q", name, ok, got, want)
54+ }
55+ }
56+}
57+
58+// waitStored waits for files to be stored. sshfs closes a file after the
59+// program writing it has exited, and the file is stored on close.
60+func (e *rsyncEnv) waitStored(files map[string]string) {
61+ e.t.Helper()
62+ for range 50 {
63+ done := true
64+ for name, want := range files {
65+ if got, _, ok := e.stored("/site/" + name); !ok || string(got) != want {
66+ done = false
67+ }
68+ }
69+ if done {
70+ return
71+ }
72+ time.Sleep(20 * time.Millisecond)
73+ }
74+ e.assertStored(files)
75+}
76+
77+func TestScpSftpUpload(t *testing.T) {
78+ newStorage := rsyncStorageBackends()["memory"]
79+ mtime := time.Unix(1_600_000_000, 0)
80+ for mode, upload := range uploadModes {
81+ t.Run(mode, func(t *testing.T) {
82+ t.Run("tree and overwrite", func(t *testing.T) {
83+ env := newRsyncEnv(t, newStorage(t))
84+ src := tree{t, t.TempDir()}
85+ files := map[string]string{"index.html": "index", "css/site.css": "css", "img/a/b.png": "png"}
86+ for name, contents := range files {
87+ src.write(name, []byte(contents), mtime)
88+ }
89+ if out, code := runCmd(t, upload(env, src.dir)); code != 0 {
90+ t.Fatalf("exit %d\n%s", code, out)
91+ }
92+ env.assertStored(files)
93+
94+ files["index.html"] = "changed"
95+ src.write("index.html", []byte("changed"), mtime)
96+ if out, code := runCmd(t, upload(env, src.dir)); code != 0 {
97+ t.Fatalf("exit %d\n%s", code, out)
98+ }
99+ env.assertStored(files)
100+ })
101+
102+ t.Run("rejected file fails without stopping the rest", func(t *testing.T) {
103+ env := newRsyncEnv(t, newStorage(t))
104+ env.feature.Data.FileMax = 100
105+ src := tree{t, t.TempDir()}
106+ ok := map[string]string{"a1.txt": "ok", "c3.txt": "ok", "sub/d4.txt": "ok"}
107+ for name, contents := range ok {
108+ src.write(name, []byte(contents), mtime)
109+ }
110+ big := strings.Repeat("x", 200)
111+ src.write("b2.bin", []byte(big), mtime)
112+ out, code := runCmd(t, upload(env, src.dir))
113+ if code == 0 {
114+ t.Errorf("rejected upload exited 0\n%s", out)
115+ }
116+ if !strings.Contains(out, "stream size exceeded") {
117+ t.Errorf("no rejection message\n%s", out)
118+ }
119+ env.assertStored(ok)
120+ if got, _, _ := env.stored("/site/b2.bin"); string(got) == big {
121+ t.Error("oversized file was stored")
122+ }
123+ })
124+
125+ t.Run("ignored dotfile", func(t *testing.T) {
126+ env := newRsyncEnv(t, newStorage(t))
127+ src := tree{t, t.TempDir()}
128+ ok := map[string]string{"a1.txt": "ok", "z9.txt": "ok"}
129+ for name, contents := range ok {
130+ src.write(name, []byte(contents), mtime)
131+ }
132+ src.write(".secret", []byte("dot"), mtime)
133+ out, code := runCmd(t, upload(env, src.dir))
134+ if !strings.Contains(out, "file ignored: (/site/.secret)") {
135+ t.Errorf("no ignore message\n%s", out)
136+ }
137+ if code != 0 {
138+ t.Errorf("ignored file failed the upload: exit %d\n%s", code, out)
139+ }
140+ env.assertStored(ok)
141+ if _, _, stored := env.stored("/site/.secret"); stored {
142+ t.Error("ignored file was stored")
143+ }
144+ })
145+ })
146+ }
147+}
148+
149+func TestScpSftpDownloadAndCommands(t *testing.T) {
150+ env := newRsyncEnv(t, rsyncStorageBackends()["memory"](t))
151+ src := tree{t, t.TempDir()}
152+ files := map[string]string{"index.html": "index", "css/site.css": "css", "old.txt": "old"}
153+ for name, contents := range files {
154+ src.write(name, []byte(contents), time.Unix(1_600_000_000, 0))
155+ }
156+ if out, code := runCmd(t, uploadModes["sftp"](env, src.dir)); code != 0 {
157+ t.Fatalf("exit %d\n%s", code, out)
158+ }
159+
160+ t.Run("sftp get", func(t *testing.T) {
161+ dst := t.TempDir()
162+ if out, code := runCmd(t, env.sftp("get -r /site "+dst)); code != 0 {
163+ t.Fatalf("exit %d\n%s", code, out)
164+ }
165+ for name, want := range files {
166+ if got, err := os.ReadFile(filepath.Join(dst, "site", name)); err != nil || string(got) != want {
167+ t.Errorf("%s: %q %v", name, got, err)
168+ }
169+ }
170+ })
171+
172+ t.Run("scp download over sftp", func(t *testing.T) {
173+ dst := t.TempDir()
174+ cmd := exec.Command("scp", append(append([]string{"-r"}, env.sshArgs...), "localhost:/site/css", dst)...)
175+ if out, code := runCmd(t, cmd); code != 0 {
176+ t.Fatalf("exit %d\n%s", code, out)
177+ }
178+ if got, err := os.ReadFile(filepath.Join(dst, "css/site.css")); err != nil || string(got) != "css" {
179+ t.Errorf("site.css: %q %v", got, err)
180+ }
181+ })
182+
183+ t.Run("legacy scp download is refused", func(t *testing.T) {
184+ cmd := exec.Command("scp", append(append([]string{"-O"}, env.sshArgs...), "localhost:/site/index.html", t.TempDir())...)
185+ out, code := runCmd(t, cmd)
186+ if code == 0 || !strings.Contains(out, "unsupported, use rsync or sftp") {
187+ t.Fatalf("exit %d\n%s", code, out)
188+ }
189+ })
190+
191+ t.Run("sftp ls rm", func(t *testing.T) {
192+ out, code := runCmd(t, env.sftp("ls /site", "rm /site/old.txt"))
193+ if code != 0 {
194+ t.Fatalf("exit %d\n%s", code, out)
195+ }
196+ for _, name := range []string{"index.html", "old.txt", "css"} {
197+ if !strings.Contains(out, name) {
198+ t.Errorf("ls is missing %s\n%s", name, out)
199+ }
200+ }
201+ if _, _, ok := env.stored("/site/old.txt"); ok {
202+ t.Error("rm did not delete the file")
203+ }
204+ env.assertStored(map[string]string{"index.html": "index", "css/site.css": "css"})
205+ })
206+}
207+
208+func TestSftpIgnoredFileKeepsBatchGoing(t *testing.T) {
209+ env := newRsyncEnv(t, rsyncStorageBackends()["memory"](t))
210+ src := tree{t, t.TempDir()}
211+ src.write(".secret", []byte("dot"), time.Unix(1_600_000_000, 0))
212+ src.write("after.txt", []byte("after"), time.Unix(1_600_000_000, 0))
213+ out, code := runCmd(t, env.sftp(
214+ "-mkdir /site",
215+ "put "+filepath.Join(src.dir, ".secret")+" /site/.secret",
216+ "put "+filepath.Join(src.dir, "after.txt")+" /site/after.txt",
217+ ))
218+ if code != 0 {
219+ t.Fatalf("exit %d\n%s", code, out)
220+ }
221+ env.assertStored(map[string]string{"after.txt": "after"})
222+ if _, _, ok := env.stored("/site/.secret"); ok {
223+ t.Error("ignored file was stored")
224+ }
225+}
226+
227+func TestSftpRejectedUploadKeepsStoredFile(t *testing.T) {
228+ for _, mode := range []string{"scp-sftp", "sftp"} {
229+ t.Run(mode, func(t *testing.T) {
230+ env := newRsyncEnv(t, rsyncStorageBackends()["memory"](t))
231+ src := tree{t, t.TempDir()}
232+ src.write("page.html", []byte("original"), time.Unix(1_600_000_000, 0))
233+ if out, code := runCmd(t, uploadModes[mode](env, src.dir)); code != 0 {
234+ t.Fatalf("exit %d\n%s", code, out)
235+ }
236+ env.feature.Data.FileMax = 10
237+ src.write("page.html", []byte(strings.Repeat("x", 100)), time.Unix(1_600_000_000, 0))
238+ src.write("new.html", []byte(strings.Repeat("y", 100)), time.Unix(1_600_000_000, 0))
239+ if out, code := runCmd(t, uploadModes[mode](env, src.dir)); code == 0 {
240+ t.Fatalf("oversized upload exited 0\n%s", out)
241+ }
242+ env.assertStored(map[string]string{"page.html": "original"})
243+ if _, _, ok := env.stored("/site/new.html"); ok {
244+ t.Error("rejected new file left an object in storage")
245+ }
246+ })
247+ }
248+}
249+
250+// TestSftpOpenFiles drives the calls sshfs makes for touch and editors.
251+func TestSftpOpenFiles(t *testing.T) {
252+ env := newRsyncEnv(t, rsyncStorageBackends()["memory"](t))
253+ conn, err := env.user.NewClientAddr(env.addr)
254+ if err != nil {
255+ t.Fatal(err)
256+ }
257+ defer func() { _ = conn.Close() }()
258+ client, err := sftp.NewClient(conn)
259+ if err != nil {
260+ t.Fatal(err)
261+ }
262+ defer func() { _ = client.Close() }()
263+ _ = client.Mkdir("/site")
264+
265+ put := func(name, contents string) {
266+ t.Helper()
267+ f, err := client.Create(name)
268+ if err != nil {
269+ t.Fatal(err)
270+ }
271+ if _, err := f.Write([]byte(contents)); err != nil {
272+ t.Fatal(err)
273+ }
274+ if err := f.Close(); err != nil {
275+ t.Fatal(err)
276+ }
277+ }
278+
279+ t.Run("touch keeps an existing file", func(t *testing.T) {
280+ put("/site/a.txt", "hello")
281+ f, err := client.OpenFile("/site/a.txt", os.O_WRONLY|os.O_CREATE)
282+ if err != nil {
283+ t.Fatal(err)
284+ }
285+ if err := f.Close(); err != nil {
286+ t.Fatal(err)
287+ }
288+ env.assertStored(map[string]string{"a.txt": "hello"})
289+ })
290+
291+ t.Run("touch creates an empty file", func(t *testing.T) {
292+ f, err := client.OpenFile("/site/empty.txt", os.O_WRONLY|os.O_CREATE)
293+ if err != nil {
294+ t.Fatal(err)
295+ }
296+ if _, err := client.Stat("/site/empty.txt"); err != nil {
297+ t.Errorf("open file not visible to stat: %v", err)
298+ }
299+ if _, _, ok := env.stored("/site/empty.txt"); ok {
300+ t.Error("file stored before close")
301+ }
302+ if err := f.Close(); err != nil {
303+ t.Fatal(err)
304+ }
305+ env.assertStored(map[string]string{"empty.txt": ""})
306+ })
307+
308+ t.Run("open file is listed and readable", func(t *testing.T) {
309+ f, err := client.OpenFile("/site/draft.txt", os.O_WRONLY|os.O_CREATE|os.O_TRUNC)
310+ if err != nil {
311+ t.Fatal(err)
312+ }
313+ if _, err := f.Write([]byte("draft")); err != nil {
314+ t.Fatal(err)
315+ }
316+ infos, err := client.ReadDir("/site")
317+ if err != nil {
318+ t.Fatal(err)
319+ }
320+ listed := false
321+ for _, info := range infos {
322+ if info.Name() == "draft.txt" && info.Size() == 5 {
323+ listed = true
324+ }
325+ }
326+ if !listed {
327+ t.Error("open file missing from the listing")
328+ }
329+ r, err := client.Open("/site/draft.txt")
330+ if err != nil {
331+ t.Fatal(err)
332+ }
333+ got, _ := io.ReadAll(r)
334+ _ = r.Close()
335+ if string(got) != "draft" {
336+ t.Errorf("read %q from the open file", got)
337+ }
338+ if err := f.Close(); err != nil {
339+ t.Fatal(err)
340+ }
341+ env.assertStored(map[string]string{"draft.txt": "draft"})
342+ })
343+
344+ t.Run("write at an offset keeps the rest", func(t *testing.T) {
345+ put("/site/b.txt", "hello world")
346+ f, err := client.OpenFile("/site/b.txt", os.O_WRONLY)
347+ if err != nil {
348+ t.Fatal(err)
349+ }
350+ if _, err := f.WriteAt([]byte("WORLD"), 6); err != nil {
351+ t.Fatal(err)
352+ }
353+ if err := f.Close(); err != nil {
354+ t.Fatal(err)
355+ }
356+ env.assertStored(map[string]string{"b.txt": "hello WORLD"})
357+ })
358+
359+ t.Run("truncate keeps nothing", func(t *testing.T) {
360+ put("/site/c.txt", "old")
361+ f, err := client.OpenFile("/site/c.txt", os.O_WRONLY|os.O_TRUNC)
362+ if err != nil {
363+ t.Fatal(err)
364+ }
365+ if err := f.Close(); err != nil {
366+ t.Fatal(err)
367+ }
368+ env.assertStored(map[string]string{"c.txt": ""})
369+ })
370+
371+ t.Run("rename", func(t *testing.T) {
372+ put("/site/r1.txt", "one")
373+ put("/site/r2.txt", "two")
374+ if err := client.Rename("/site/r1.txt", "/site/r2.txt"); err == nil {
375+ t.Error("rename replaced an existing file")
376+ }
377+ if err := client.PosixRename("/site/r1.txt", "/site/r2.txt"); err != nil {
378+ t.Fatal(err)
379+ }
380+ env.assertStored(map[string]string{"r2.txt": "one"})
381+ if _, _, ok := env.stored("/site/r1.txt"); ok {
382+ t.Error("source survived the rename")
383+ }
384+ if err := client.Rename("/site/r2.txt", "/site/.r2"); err == nil {
385+ t.Error("rename to an ignored name succeeded")
386+ }
387+ env.assertStored(map[string]string{"r2.txt": "one"})
388+
389+ f, err := client.OpenFile("/site/r2.txt", os.O_WRONLY)
390+ if err != nil {
391+ t.Fatal(err)
392+ }
393+ if _, err := f.WriteAt([]byte("ONE"), 0); err != nil {
394+ t.Fatal(err)
395+ }
396+ if err := client.Rename("/site/r2.txt", "/site/r3.txt"); err != nil {
397+ t.Fatal(err)
398+ }
399+ if err := f.Close(); err != nil {
400+ t.Fatal(err)
401+ }
402+ env.assertStored(map[string]string{"r3.txt": "ONE"})
403+ if _, _, ok := env.stored("/site/r2.txt"); ok {
404+ t.Error("open file's old path is still stored after it closed")
405+ }
406+ })
407+
408+ t.Run("truncate then write past the end leaves zeros", func(t *testing.T) {
409+ put("/site/z.txt", "secret")
410+ f, err := client.OpenFile("/site/z.txt", os.O_WRONLY)
411+ if err != nil {
412+ t.Fatal(err)
413+ }
414+ if err := f.Truncate(0); err != nil {
415+ t.Fatal(err)
416+ }
417+ if _, err := f.WriteAt([]byte("x"), 5); err != nil {
418+ t.Fatal(err)
419+ }
420+ if err := f.Close(); err != nil {
421+ t.Fatal(err)
422+ }
423+ env.assertStored(map[string]string{"z.txt": "\x00\x00\x00\x00\x00x"})
424+ })
425+
426+ t.Run("renaming an open file keeps the source until it is stored", func(t *testing.T) {
427+ openAndWrite := func(name, contents string) *sftp.File {
428+ t.Helper()
429+ f, err := client.OpenFile(name, os.O_WRONLY)
430+ if err != nil {
431+ t.Fatal(err)
432+ }
433+ if _, err := f.WriteAt([]byte(contents), 0); err != nil {
434+ t.Fatal(err)
435+ }
436+ return f
437+ }
438+
439+ put("/site/k.txt", "keep")
440+ f := openAndWrite("/site/k.txt", "KEEP")
441+ if err := client.PosixRename("/site/k.txt", "/site/.ignored"); err == nil {
442+ t.Error("renamed an open file to an ignored name")
443+ }
444+ _ = f.Close()
445+ env.assertStored(map[string]string{"k.txt": "KEEP"})
446+
447+ put("/site/k.txt", "keep")
448+ f = openAndWrite("/site/k.txt", strings.Repeat("x", 200))
449+ if err := client.PosixRename("/site/k.txt", "/site/big.txt"); err != nil {
450+ t.Fatal(err)
451+ }
452+ if _, err := client.Stat("/site/k.txt"); !errors.Is(err, os.ErrNotExist) {
453+ t.Errorf("renamed source still visible: %v", err)
454+ }
455+ env.feature.Data.FileMax = 100
456+ if err := f.Close(); err == nil {
457+ t.Error("oversized close succeeded")
458+ }
459+ env.feature.Data.FileMax = rsyncFileMax
460+ env.assertStored(map[string]string{"k.txt": "keep"})
461+ if _, _, ok := env.stored("/site/big.txt"); ok {
462+ t.Error("rejected destination was stored")
463+ }
464+
465+ f = openAndWrite("/site/k.txt", "NEW!")
466+ if err := client.PosixRename("/site/k.txt", "/site/moved.txt"); err != nil {
467+ t.Fatal(err)
468+ }
469+ env.assertStored(map[string]string{"k.txt": "keep"})
470+ put("/site/k.txt", "replacement")
471+ if err := f.Close(); err != nil {
472+ t.Fatal(err)
473+ }
474+ env.assertStored(map[string]string{"moved.txt": "NEW!", "k.txt": "replacement"})
475+
476+ put("/site/k2.txt", "k2")
477+ f = openAndWrite("/site/k2.txt", "K2")
478+ if err := client.PosixRename("/site/k2.txt", "/site/moved2.txt"); err != nil {
479+ t.Fatal(err)
480+ }
481+ if err := f.Close(); err != nil {
482+ t.Fatal(err)
483+ }
484+ env.assertStored(map[string]string{"moved2.txt": "K2"})
485+ if _, _, ok := env.stored("/site/k2.txt"); ok {
486+ t.Error("source kept after the renamed file was stored")
487+ }
488+ })
489+
490+ t.Run("renamed source replaced by another session survives", func(t *testing.T) {
491+ conn2, err := env.user.NewClientAddr(env.addr)
492+ if err != nil {
493+ t.Fatal(err)
494+ }
495+ defer func() { _ = conn2.Close() }()
496+ other, err := sftp.NewClient(conn2)
497+ if err != nil {
498+ t.Fatal(err)
499+ }
500+ defer func() { _ = other.Close() }()
501+
502+ put("/site/s1.txt", "orig")
503+ f, err := client.OpenFile("/site/s1.txt", os.O_WRONLY)
504+ if err != nil {
505+ t.Fatal(err)
506+ }
507+ if _, err := f.WriteAt([]byte("ORIG"), 0); err != nil {
508+ t.Fatal(err)
509+ }
510+ if err := client.PosixRename("/site/s1.txt", "/site/s2.txt"); err != nil {
511+ t.Fatal(err)
512+ }
513+ g, err := other.Create("/site/s1.txt")
514+ if err != nil {
515+ t.Fatal(err)
516+ }
517+ if _, err := g.Write([]byte("from another session")); err != nil {
518+ t.Fatal(err)
519+ }
520+ if err := g.Close(); err != nil {
521+ t.Fatal(err)
522+ }
523+ if err := f.Close(); err != nil {
524+ t.Fatal(err)
525+ }
526+ env.assertStored(map[string]string{"s1.txt": "from another session", "s2.txt": "ORIG"})
527+ })
528+
529+ t.Run("removing a file renamed over a stored one removes both", func(t *testing.T) {
530+ put("/site/old.txt", "old")
531+ put("/site/src.txt", "src")
532+ f, err := client.OpenFile("/site/src.txt", os.O_WRONLY)
533+ if err != nil {
534+ t.Fatal(err)
535+ }
536+ if _, err := f.WriteAt([]byte("SRC"), 0); err != nil {
537+ t.Fatal(err)
538+ }
539+ if err := client.PosixRename("/site/src.txt", "/site/old.txt"); err != nil {
540+ t.Fatal(err)
541+ }
542+ if err := client.Remove("/site/old.txt"); err != nil {
543+ t.Fatal(err)
544+ }
545+ if err := f.Close(); err != nil {
546+ t.Fatal(err)
547+ }
548+ for _, name := range []string{"old.txt", "src.txt"} {
549+ if _, _, ok := env.stored("/site/" + name); ok {
550+ t.Errorf("%s is still stored", name)
551+ }
552+ }
553+ })
554+
555+ t.Run("removed while open is not stored", func(t *testing.T) {
556+ f, err := client.OpenFile("/site/swap.txt", os.O_WRONLY|os.O_CREATE|os.O_TRUNC)
557+ if err != nil {
558+ t.Fatal(err)
559+ }
560+ if _, err := f.Write([]byte("swap")); err != nil {
561+ t.Fatal(err)
562+ }
563+ if err := client.Remove("/site/swap.txt"); err != nil {
564+ t.Fatal(err)
565+ }
566+ if err := f.Close(); err != nil {
567+ t.Fatal(err)
568+ }
569+ if _, _, ok := env.stored("/site/swap.txt"); ok {
570+ t.Error("file removed while open was stored")
571+ }
572+ })
573+}
574+
575+// TestSshfs mounts pgs with sshfs and edits files the way shells and
576+// editors do.
577+func TestSshfs(t *testing.T) {
578+ if _, err := exec.LookPath("sshfs"); err != nil {
579+ t.Skip("sshfs not installed")
580+ }
581+ for name, newStorage := range rsyncStorageBackends() {
582+ t.Run(name, func(t *testing.T) { testSshfs(t, newRsyncEnv(t, newStorage(t))) })
583+ }
584+}
585+
586+func testSshfs(t *testing.T, env *rsyncEnv) {
587+ mnt := t.TempDir()
588+ opts := []string{"-p", env.sshArgs[1], "-o", "IdentityFile=" + env.sshArgs[3], "-o", "IdentitiesOnly=yes",
589+ "-o", "StrictHostKeyChecking=no", "-o", "UserKnownHostsFile=/dev/null", "-o", "LogLevel=ERROR"}
590+ if out, err := exec.Command("sshfs", append(opts, "localhost:/", mnt)...).CombinedOutput(); err != nil {
591+ t.Skipf("sshfs mount failed: %v\n%s", err, out)
592+ }
593+ t.Cleanup(func() { _ = exec.Command("fusermount3", "-u", "-z", mnt).Run() })
594+
595+ home := t.TempDir()
596+ steps := []struct {
597+ cmd string
598+ fails bool
599+ stored map[string]string
600+ }{
601+ {cmd: "mkdir site && echo hello > site/a.txt", stored: map[string]string{"a.txt": "hello\n"}},
602+ {cmd: "touch site/a.txt", stored: map[string]string{"a.txt": "hello\n"}},
603+ {cmd: "touch site/new.txt", stored: map[string]string{"new.txt": ""}},
604+ {cmd: "vim -n -es -c 'normal! ihi' -c wq site/vim.txt", stored: map[string]string{"vim.txt": "hi\n"}},
605+ {cmd: "vim -es -c '%s/hello/bye/' -c wq site/a.txt", stored: map[string]string{"a.txt": "bye\n"}},
606+ {cmd: "echo more >> site/a.txt", stored: map[string]string{"a.txt": "bye\nmore\n"}},
607+ {cmd: "printf abc > site/t.txt && truncate -s 1 site/t.txt", stored: map[string]string{"t.txt": "a"}},
608+ {cmd: "rm site/new.txt", stored: map[string]string{"a.txt": "bye\nmore\n"}},
609+ {cmd: "echo secret > site/.dot", fails: true},
610+ {cmd: "mv site/t.txt site/moved.txt", stored: map[string]string{"moved.txt": "a"}},
611+ {cmd: "mv site/moved.txt site/a.txt", stored: map[string]string{"a.txt": "a"}},
612+ {cmd: "mv site/a.txt site/.dot", fails: true, stored: map[string]string{"a.txt": "a"}},
613+ {cmd: "mkdir -p site/dir/sub site/empty && echo y > site/dir/sub/y.txt && mv site/dir site/dir2 && mv site/a.txt site/dir2/", stored: map[string]string{"dir2/sub/y.txt": "y\n", "dir2/a.txt": "a"}},
614+ {cmd: "mv site/empty site/empty2 && test -d site/empty2 && ! test -e site/empty && ! test -e site/dir"},
615+ {cmd: "echo q > site/q.txt && mv site/q.txt site/r.txt && cat site/r.txt", stored: map[string]string{"r.txt": "q\n"}},
616+ }
617+ for _, step := range steps {
618+ cmd := exec.Command("bash", "-c", step.cmd)
619+ cmd.Dir = mnt
620+ cmd.Env = append(os.Environ(), "HOME="+home)
621+ out, err := cmd.CombinedOutput()
622+ if (err != nil) != step.fails {
623+ t.Fatalf("%s: err=%v\n%s", step.cmd, err, out)
624+ }
625+ env.waitStored(step.stored)
626+ }
627+ for _, name := range []string{"new.txt", ".dot", "t.txt", "moved.txt", "a.txt", "dir/sub/y.txt", "q.txt"} {
628+ if _, _, ok := env.stored("/site/" + name); ok {
629+ t.Errorf("%s is stored", name)
630+ }
631+ }
632+}
M pkg/apps/pgs/uploader.go
+18, -5
......@@ -265,6 +265,19 @@ func mtimeToTime(entry *sendutils.FileEntry) time.Time {
265265 }
266266
267267 func (h *UploadAssetHandler) Write(s *pssh.SSHServerConnSession, entry *sendutils.FileEntry) (string, error) {
268+ return h.write(s, entry, false)
269+}
270+
271+// ValidateWrite runs the checks Write makes before reading the file.
272+func (h *UploadAssetHandler) ValidateWrite(s *pssh.SSHServerConnSession, entry *sendutils.FileEntry) error {
273+ if entry.Mode.IsDir() {
274+ return nil
275+ }
276+ _, err := h.write(s, entry, true)
277+ return err
278+}
279+
280+func (h *UploadAssetHandler) write(s *pssh.SSHServerConnSession, entry *sendutils.FileEntry, validateOnly bool) (string, error) {
268281 logger := pssh.GetLogger(s)
269282 user := pssh.GetUser(s)
270283
......@@ -339,12 +352,11 @@ func (h *UploadAssetHandler) Write(s *pssh.SSHServerConnSession, entry *sendutil
339352 remaining := int64(storageMax) - curStorageSize
340353 sizeRemaining := min(remaining+curFileSize, fileMax)
341354 if sizeRemaining <= 0 {
342- _, _ = fmt.Fprintln(s.Stderr(), "storage quota reached")
343- _, _ = fmt.Fprintf(s.Stderr(), "\r")
344- _ = s.Exit(1)
345- _ = s.Close()
346355 return "", fmt.Errorf("storage quota reached")
347356 }
357+ if validateOnly {
358+ return "", nil
359+ }
348360 logger = logger.With(
349361 "storageMax", storageMax,
350362 "currentStorageMax", curStorageSize,
......@@ -467,7 +479,8 @@ func (h *UploadAssetHandler) validateAsset(data *FileData) (bool, error) {
467479 fpath := strings.Replace(data.Filepath, "/"+projectName, "", 1)
468480 if data.DenyList.MatchesPath(fpath) {
469481 err := fmt.Errorf(
470- "ERROR: (%s) file rejected, https://pico.sh/pgs#-pgs-ignore",
482+ "%w: (%s), https://pico.sh/pgs#-pgs-ignore",
483+ sendutils.ErrFileIgnored,
471484 data.Filepath,
472485 )
473486 return false, err
M pkg/filehandlers/post_handler.go
+25, -3
......@@ -81,6 +81,28 @@ func (h *ScpUploadHandler) Read(s *pssh.SSHServerConnSession, entry *sendutils.F
8181 return fileInfo, reader, nil
8282 }
8383
84+// ValidateWrite runs the hooks' validation on the file before its
85+// contents arrive.
86+func (h *ScpUploadHandler) ValidateWrite(s *pssh.SSHServerConnSession, entry *sendutils.FileEntry) error {
87+ user := pssh.GetUser(s)
88+ if user == nil {
89+ return fmt.Errorf("could not get user from ctx")
90+ }
91+ if entry.Mode.IsDir() {
92+ return fmt.Errorf("file entry is directory, but only files are supported: %s", entry.Filepath)
93+ }
94+ filename := filepath.Base(entry.Filepath)
95+ metadata := PostMetaData{
96+ Post: &db.Post{Filename: filename, Slug: shared.SanitizeFileExt(filename)},
97+ User: user,
98+ FileEntry: entry,
99+ }
100+ if valid, err := h.Hooks.FileValidate(s, &metadata); !valid {
101+ return err
102+ }
103+ return nil
104+}
105+
84106 func (h *ScpUploadHandler) Write(s *pssh.SSHServerConnSession, entry *sendutils.FileEntry) (string, error) {
85107 logger := pssh.GetLogger(s)
86108 user := pssh.GetUser(s)
......@@ -102,9 +124,9 @@ func (h *ScpUploadHandler) Write(s *pssh.SSHServerConnSession, entry *sendutils.
102124 return "", fmt.Errorf("file entry is directory, but only files are supported: %s", filename)
103125 }
104126
105- var origText []byte
106- if b, err := io.ReadAll(entry.Reader); err == nil {
107- origText = b
127+ origText, err := io.ReadAll(entry.Reader)
128+ if err != nil {
129+ return "", fmt.Errorf("could not read file (%s): %w", filename, err)
108130 }
109131
110132 mimeType := http.DetectContentType(origText)
A pkg/filehandlers/post_handler_test.go
+71, -0
......@@ -0,0 +1,71 @@
1+package filehandlers
2+
3+import (
4+ "context"
5+ "errors"
6+ "io"
7+ "log/slog"
8+ "strings"
9+ "testing"
10+ "testing/iotest"
11+
12+ "github.com/picosh/pico/pkg/db"
13+ "github.com/picosh/pico/pkg/db/stub"
14+ "github.com/picosh/pico/pkg/pssh"
15+ sendutils "github.com/picosh/pico/pkg/send/utils"
16+ "github.com/picosh/pico/pkg/shared"
17+ "golang.org/x/crypto/ssh"
18+)
19+
20+type uploadHooks struct {
21+ validated bool
22+}
23+
24+func (h *uploadHooks) FileValidate(*pssh.SSHServerConnSession, *PostMetaData) (bool, error) {
25+ h.validated = true
26+ return false, errors.New("unexpected validation after a failed read")
27+}
28+
29+func (h *uploadHooks) FileMeta(*pssh.SSHServerConnSession, *PostMetaData) error {
30+ return nil
31+}
32+
33+func TestPostUploadReadError(t *testing.T) {
34+ checksumErr := errors.New("rsync: file failed verification")
35+ tests := []struct {
36+ name string
37+ text string
38+ err error
39+ }{
40+ {name: "before contents", err: io.ErrClosedPipe},
41+ {name: "partial contents", text: "partial upload", err: io.ErrUnexpectedEOF},
42+ {name: "failed verification", text: "complete upload", err: checksumErr},
43+ }
44+ for _, tt := range tests {
45+ t.Run(tt.name, func(t *testing.T) {
46+ logger := slog.New(slog.DiscardHandler)
47+ session := &pssh.SSHServerConnSession{
48+ SSHServerConn: &pssh.SSHServerConn{
49+ Conn: &ssh.ServerConn{Permissions: &ssh.Permissions{Extensions: map[string]string{}}},
50+ Logger: logger,
51+ },
52+ Ctx: context.Background(),
53+ }
54+ pssh.SetUser(session, &db.User{ID: "user-1", Name: "tester"})
55+ hooks := &uploadHooks{}
56+ handler := NewScpPostHandler(stub.NewStubDB(logger), &shared.ConfigSite{Space: "pastes"}, hooks)
57+ reader := io.MultiReader(strings.NewReader(tt.text), iotest.ErrReader(tt.err))
58+
59+ message, err := handler.Write(session, &sendutils.FileEntry{Filepath: "/test.txt", Reader: reader})
60+ if !errors.Is(err, tt.err) {
61+ t.Fatalf("error = %v, want %v", err, tt.err)
62+ }
63+ if message != "" {
64+ t.Fatalf("failed upload returned a success message: %q", message)
65+ }
66+ if hooks.validated {
67+ t.Fatal("failed upload reached validation")
68+ }
69+ })
70+ }
71+}
M pkg/filehandlers/router_handler.go
+11, -0
......@@ -65,6 +65,17 @@ func (r *FileHandlerRouter) Write(s *pssh.SSHServerConnSession, entry *utils.Fil
6565 return handler.Write(s, entry)
6666 }
6767
68+func (r *FileHandlerRouter) ValidateWrite(s *pssh.SSHServerConnSession, entry *utils.FileEntry) error {
69+ handler, err := r.findHandler(entry.Filepath)
70+ if err != nil {
71+ return err
72+ }
73+ if v, ok := handler.(utils.WriteValidator); ok {
74+ return v.ValidateWrite(s, entry)
75+ }
76+ return nil
77+}
78+
6879 func (r *FileHandlerRouter) Delete(s *pssh.SSHServerConnSession, entry *utils.FileEntry) error {
6980 handler, err := r.findHandler(entry.Filepath)
7081 if err != nil {
M pkg/rsync/filter.go
+500, -77
......@@ -1,7 +1,11 @@
11 package rsync
22
33 import (
4+ "errors"
45 "fmt"
6+ "io/fs"
7+ "path"
8+ "slices"
59 "strings"
610 )
711
......@@ -30,18 +34,60 @@ const (
3034 ruleXattr
3135 )
3236
37+// containerFlags are the flags rules in a merge file take from the merge
38+// rule (FILTRULES_FROM_CONTAINER).
39+const containerFlags = ruleAbsPath | ruleInclude | ruleDirectory | ruleNegate | rulePerishable
40+
41+const sideFlags = ruleSenderSide | ruleReceiverSide
42+
43+// maxMergeFileSize bounds a per-directory merge file read from storage.
44+const maxMergeFileSize = 1 << 20
45+
3346 type filterRule struct {
3447 pattern string
3548 flags int
3649 slashCnt int
50+ // merge indexes filterList.merges for a per-directory merge rule.
51+ merge int
3752 }
3853
3954 // filterList holds the rules the client sent. The client only sends the
40-// rules that apply to our side of the transfer.
55+// rules that apply to our side of the transfer. A per-directory merge rule
56+// stands for the rules read from that file in each directory, which a
57+// filterScope holds.
4158 type filterList struct {
42- rules []*filterRule
59+ rules []*filterRule
60+ merges []*filterRule
61+ // otherSide marks merge-file rules that only apply to the client.
62+ otherSide int
63+ deleteExcluded bool
64+ cvsPerishable bool
65+}
66+
67+// filterScope is the per-directory merge-file rules in effect in one
68+// directory, each list holding the directory's own rules before the
69+// inherited ones.
70+type filterScope struct {
71+ // root is the storage path of the transfer root, which absolute
72+ // patterns are matched from.
73+ root string
74+ lists [][]*filterRule
4375 }
4476
77+// mergeFileError is a per-directory merge file that could not be used.
78+// Skipping its rules could delete or send files they cover, so callers
79+// stop instead.
80+type mergeFileError struct {
81+ name string
82+ err error
83+}
84+
85+func (e *mergeFileError) Error() string {
86+ return fmt.Sprintf("per-directory filter file %q: %v", "/"+e.name, e.err)
87+}
88+
89+func (e *mergeFileError) Unwrap() error { return e.err }
90+
4591 // defaultCVSIgnore is the list rsync's -C adds.
4692 const defaultCVSIgnore = "RCS SCCS CVS CVS.adm RCSLOG cvslog.* tags TAGS" +
4793 " .make.state .nse_depinfo *~ #* .#* ,* _$* *$" +
......@@ -50,32 +96,68 @@ const defaultCVSIgnore = "RCS SCCS CVS CVS.adm RCSLOG cvslog.* tags TAGS" +
5096 " *.Z *.elc *.ln core" +
5197 " .svn/ .git/ .hg/ .bzr/"
5298
53-// parse adds the rules in one line received from the client. Before
54-// protocol 29 only the "- " and "+ " prefixes exist.
55-func (l *filterList) parse(line string, oldPrefixes bool) error {
56- rule := &filterRule{}
57- s := line
99+var longRuleNames = map[byte]struct {
100+ word string
101+ ch byte
102+}{
103+ 'c': {"clear", '!'},
104+ 'd': {"dir-merge", ':'},
105+ 'e': {"exclude", '-'},
106+ 'h': {"hide", 'H'},
107+ 'i': {"include", '+'},
108+ 'm': {"merge", '.'},
109+ 'p': {"protect", 'P'},
110+ 'r': {"risk", 'R'},
111+ 's': {"show", 'S'},
112+}
113+
114+func isSpace(c byte) bool { return c == ' ' || (c >= '\t' && c <= '\r') }
115+
116+// parseRuleTok parses one rule (parse_rule_tok). tmpl is the flags of the
117+// merge rule whose file the rule was read from, or 0 for the client's
118+// rules. It returns a nil rule for an empty one.
119+func parseRuleTok(s string, tmpl int, oldPrefixes bool) (*filterRule, error) {
120+ if tmpl&ruleWordSplit != 0 {
121+ s = strings.TrimLeftFunc(s, func(r rune) bool { return r < 0x80 && isSpace(byte(r)) })
122+ }
123+ if s == "" {
124+ return nil, nil
125+ }
126+ rule := &filterRule{flags: tmpl & containerFlags}
58127 switch {
128+ case tmpl&ruleNoPrefixes != 0:
129+ if s[0] == '!' && tmpl&ruleCVSIgnore != 0 {
130+ rule.flags |= ruleClearList
131+ }
59132 case oldPrefixes:
60133 switch {
61134 case strings.HasPrefix(s, "- "):
135+ rule.flags &^= ruleInclude
62136 s = s[2:]
63137 case strings.HasPrefix(s, "+ "):
64138 rule.flags |= ruleInclude
65139 s = s[2:]
66- case strings.HasPrefix(s, "!"):
140+ case s[0] == '!':
67141 rule.flags |= ruleClearList
68142 }
69143 default:
70- // Clients send the short form built by get_rule_prefix: a rule
71- // character, optional modifiers, then a space or underscore.
72- if s == "" {
73- return nil
74- }
75- ch := s[0]
76- i := 1
77- if len(s) > 1 && s[1] == ',' {
78- i = 2
144+ // p ends on the last character of the rule name, as the modifier
145+ // loop starts by stepping past it.
146+ ch, p := s[0], 0
147+ if long, ok := longRuleNames[s[0]]; ok {
148+ rest, found := strings.CutPrefix(s, long.word)
149+ switch {
150+ case !found:
151+ return nil, errors.New("unknown filter rule")
152+ case rest == "" || rest[0] == '_' || isSpace(rest[0]):
153+ ch, p = long.ch, len(long.word)-1
154+ case rest[0] == ',':
155+ ch, p = long.ch, len(long.word)
156+ default:
157+ return nil, errors.New("unknown filter rule")
158+ }
159+ } else if len(s) > 1 && s[1] == ',' {
160+ p = 1
79161 }
80162 sideFromPrefix := false
81163 switch ch {
......@@ -101,75 +183,161 @@ func (l *filterList) parse(line string, oldPrefixes bool) error {
101183 case '!':
102184 rule.flags |= ruleClearList
103185 default:
104- return fmt.Errorf("unknown filter rule: %q", line)
105- }
106- if ch != '!' {
107- for ; i < len(s) && s[i] != ' ' && s[i] != '_'; i++ {
108- switch s[i] {
109- case '-':
110- rule.flags |= ruleNoPrefixes
111- case '+':
112- rule.flags |= ruleNoPrefixes | ruleInclude
113- case '/':
114- rule.flags |= ruleAbsPath
115- case '!':
116- rule.flags |= ruleNegate
117- case 'C':
118- if sideFromPrefix {
119- return fmt.Errorf("invalid modifier in filter rule: %q", line)
120- }
121- rule.flags |= ruleNoPrefixes | ruleWordSplit | ruleNoInherit | ruleCVSIgnore
122- case 'e':
123- rule.flags |= ruleExcludeSelf
124- case 'n':
125- rule.flags |= ruleNoInherit
126- case 'p':
127- rule.flags |= rulePerishable
128- case 'r':
129- rule.flags |= ruleReceiverSide
130- case 's':
131- rule.flags |= ruleSenderSide
132- case 'w':
133- rule.flags |= ruleWordSplit
134- case 'x':
135- rule.flags |= ruleXattr
136- default:
137- return fmt.Errorf("invalid modifier in filter rule: %q", line)
138- }
186+ return nil, errors.New("unknown filter rule")
187+ }
188+ merge := rule.flags&ruleMergeFile != 0
189+ for ch != '!' {
190+ p++
191+ if p >= len(s) || s[p] == ' ' || s[p] == '_' {
192+ break
193+ }
194+ if tmpl&ruleWordSplit != 0 && isSpace(s[p]) {
195+ p--
196+ break
139197 }
140- if i < len(s) {
141- i++
198+ valid := true
199+ switch s[p] {
200+ case '-':
201+ valid = merge && rule.flags&ruleNoPrefixes == 0
202+ rule.flags |= ruleNoPrefixes
203+ case '+':
204+ valid = merge && rule.flags&ruleNoPrefixes == 0
205+ rule.flags |= ruleNoPrefixes | ruleInclude
206+ case '/':
207+ rule.flags |= ruleAbsPath
208+ case '!':
209+ valid = !merge
210+ rule.flags |= ruleNegate
211+ case 'C':
212+ valid = rule.flags&ruleNoPrefixes == 0 && !sideFromPrefix
213+ rule.flags |= ruleNoPrefixes | ruleWordSplit | ruleNoInherit | ruleCVSIgnore
214+ case 'e':
215+ valid = merge
216+ rule.flags |= ruleExcludeSelf
217+ case 'n':
218+ valid = merge
219+ rule.flags |= ruleNoInherit
220+ case 'p':
221+ rule.flags |= rulePerishable
222+ case 'r':
223+ valid = !sideFromPrefix
224+ rule.flags |= ruleReceiverSide
225+ case 's':
226+ valid = !sideFromPrefix
227+ rule.flags |= ruleSenderSide
228+ case 'w':
229+ valid = merge
230+ rule.flags |= ruleWordSplit
231+ case 'x':
232+ rule.flags |= ruleXattr
233+ default:
234+ valid = false
142235 }
236+ if !valid {
237+ return nil, fmt.Errorf("invalid modifier %q in filter rule", s[p])
238+ }
239+ }
240+ if p < len(s) {
241+ p++
143242 }
144- s = s[min(i, len(s)):]
243+ s = s[min(p, len(s)):]
244+ }
245+ if tmpl&sideFlags != 0 {
246+ if rule.flags&sideFlags != 0 {
247+ return nil, errors.New("specified-side merge file contains specified-side filter")
248+ }
249+ rule.flags |= tmpl & sideFlags
145250 }
146251
252+ if tmpl&ruleWordSplit != 0 {
253+ if i := strings.IndexFunc(s, func(r rune) bool { return r < 0x80 && isSpace(byte(r)) }); i >= 0 {
254+ s = s[:i]
255+ }
256+ }
147257 if rule.flags&ruleClearList != 0 {
148- if len(s) <= 1 {
149- l.rules = nil
150- return nil
258+ if tmpl&ruleNoPrefixes == 0 && !oldPrefixes && s != "" {
259+ return nil, errors.New("'!' rule has trailing characters")
151260 }
152- rule.flags &^= ruleClearList
261+ if len(s) > 1 {
262+ rule.flags &^= ruleClearList
263+ }
264+ } else if s == "" && rule.flags&ruleCVSIgnore == 0 {
265+ return nil, errors.New("unexpected end of filter rule")
266+ }
267+ if len(s) >= maxPathLen {
268+ return nil, errors.New("filter rule too long")
153269 }
270+ rule.pattern = s
271+ return rule, nil
272+}
154273
274+// parse adds the rules in one line received from the client. Before
275+// protocol 29 only the "- " and "+ " prefixes exist.
276+func (l *filterList) parse(line string, oldPrefixes bool) error {
277+ rule, err := parseRuleTok(line, 0, oldPrefixes)
155278 switch {
156- case rule.flags&ruleMergeFile != 0:
157- // Merge files would have to be read from storage on our side.
158- // They can only add rules for files we don't have, so skip them.
279+ case err != nil:
280+ return fmt.Errorf("%w: %q", err, line)
281+ case rule == nil:
159282 return nil
160- case rule.flags&ruleCVSIgnore != 0:
161- for _, pat := range strings.Fields(defaultCVSIgnore) {
162- l.add(pat, ruleNoPrefixes)
283+ case rule.flags&ruleClearList != 0:
284+ l.rules, l.merges = nil, nil
285+ return nil
286+ case rule.flags&ruleMergeFile != 0:
287+ if rule.flags&rulePerdirMerge == 0 {
288+ // Clients read merge files themselves and send their rules.
289+ return fmt.Errorf("merge-file filter rules are not supported: %q", line)
163290 }
291+ return l.addMerge(rule)
292+ case rule.flags&ruleCVSIgnore != 0:
293+ l.rules = append(l.rules, l.cvsIgnoreRules()...)
164294 return nil
165- case s == "":
166- return fmt.Errorf("unexpected end of filter rule: %q", line)
167295 }
168- l.add(s, rule.flags)
296+ l.add(rule.pattern, rule.flags)
169297 return nil
170298 }
171299
172-func (l *filterList) add(pat string, flags int) {
300+// addMerge adds a per-directory merge rule. Only the file's base name is
301+// read in each directory; a directory part names where the scan of parent
302+// directories starts (setup_merge_file).
303+func (l *filterList) addMerge(rule *filterRule) error {
304+ if rule.pattern == "" {
305+ rule.pattern = ".cvsignore"
306+ }
307+ base := path.Base(rule.pattern)
308+ if strings.HasSuffix(rule.pattern, "/") || base == "." || base == ".." {
309+ return fmt.Errorf("invalid merge-file name: %q", rule.pattern)
310+ }
311+ if rule.flags&ruleExcludeSelf != 0 {
312+ l.add(base, 0)
313+ rule.flags &^= ruleExcludeSelf
314+ }
315+ for _, m := range l.merges {
316+ if path.Base(m.pattern) == base {
317+ return nil
318+ }
319+ }
320+ rule.merge = len(l.merges)
321+ l.rules = append(l.rules, rule)
322+ l.merges = append(l.merges, rule)
323+ return nil
324+}
325+
326+// cvsIgnoreRules returns rsync's default CVS excludes, which are
327+// perishable from protocol 30 (get_cvs_excludes).
328+func (l *filterList) cvsIgnoreRules() []*filterRule {
329+ flags := ruleNoPrefixes
330+ if l.cvsPerishable {
331+ flags |= rulePerishable
332+ }
333+ var rules []*filterRule
334+ for _, pat := range strings.Fields(defaultCVSIgnore) {
335+ rules = append(rules, newRule(pat, flags))
336+ }
337+ return rules
338+}
339+
340+func newRule(pat string, flags int) *filterRule {
173341 rule := &filterRule{flags: flags}
174342 if len(pat) > 1 && strings.HasSuffix(pat, "/") {
175343 pat = pat[:len(pat)-1]
......@@ -189,14 +357,260 @@ func (l *filterList) add(pat string, flags int) {
189357 }
190358 }
191359 }
192- l.rules = append(l.rules, rule)
360+ return rule
361+}
362+
363+func (l *filterList) add(pat string, flags int) {
364+ l.rules = append(l.rules, newRule(pat, flags))
365+}
366+
367+// scopeTree builds the filter scope of each directory below a traversal
368+// root the first time it is needed.
369+type scopeTree struct {
370+ l *filterList
371+ root string
372+ dir string
373+ read func(string) ([]byte, error)
374+ scopes map[string]*filterScope
375+}
376+
377+// newScopeTree starts at dir, the storage path of the first directory
378+// traversed. root is the storage path of the transfer root.
379+func (l *filterList) newScopeTree(root, dir string, read func(string) ([]byte, error)) *scopeTree {
380+ return &scopeTree{l: l, root: root, dir: dir, read: read, scopes: map[string]*filterScope{}}
381+}
382+
383+// scope returns the scope for rel, a directory relative to the tree's
384+// first directory, which is ".".
385+func (t *scopeTree) scope(rel string) (*filterScope, error) {
386+ if sc, ok := t.scopes[rel]; ok {
387+ return sc, nil
388+ }
389+ var sc *filterScope
390+ if rel == "." {
391+ var err error
392+ if sc, err = t.l.newScope(t.root, t.dir, t.read); err != nil {
393+ return nil, err
394+ }
395+ } else {
396+ parent, err := t.scope(path.Dir(rel))
397+ if err != nil {
398+ return nil, err
399+ }
400+ if sc, err = t.l.enter(parent, path.Join(t.dir, rel), t.read); err != nil {
401+ return nil, err
402+ }
403+ }
404+ t.scopes[rel] = sc
405+ return sc, nil
406+}
407+
408+// mergeReader reads per-directory merge files from storage, returning nil
409+// for a missing one. files lists the regular files below tree, relative to
410+// it; other files are looked up in their directory, as not every backend
411+// reports a missing file from Open.
412+func (s *session) mergeReader(tree string, files map[string]bool) func(string) ([]byte, error) {
413+ return func(name string) ([]byte, error) {
414+ rel, under := name, tree == ""
415+ if !under {
416+ rel, under = strings.CutPrefix(name, tree+"/")
417+ }
418+ found := false
419+ if under && files != nil {
420+ found = files[rel]
421+ } else {
422+ dir := path.Dir(name)
423+ if dir == "." {
424+ dir = ""
425+ }
426+ entries, err := s.fs.ReadDir(dir, false)
427+ if err != nil && !errors.Is(err, fs.ErrNotExist) {
428+ return nil, err
429+ }
430+ base := path.Base(name)
431+ found = slices.ContainsFunc(entries, func(e FileInfo) bool {
432+ return !e.IsDir && strings.Trim(e.Name, "/") == base
433+ })
434+ }
435+ if !found {
436+ return nil, nil
437+ }
438+ f, info, err := s.fs.Open(name)
439+ if err != nil {
440+ return nil, err
441+ }
442+ defer func() { _ = f.Close() }()
443+ if info.Size > maxMergeFileSize {
444+ return nil, fmt.Errorf("larger than %d bytes", maxMergeFileSize)
445+ }
446+ buf := make([]byte, info.Size)
447+ if err := readFullAt(f, buf, 0); err != nil {
448+ return nil, err
449+ }
450+ return buf, nil
451+ }
452+}
453+
454+// newScope returns the scope for dir, the storage path of the first
455+// directory traversed. A merge rule whose name has a directory part that
456+// is an ancestor of dir first reads the file in that directory and each
457+// one below it down to dir's parent (setup_merge_file).
458+func (l *filterList) newScope(root, dir string, read func(string) ([]byte, error)) (*filterScope, error) {
459+ parent := &filterScope{root: root, lists: make([][]*filterRule, len(l.merges))}
460+ for i, m := range l.merges {
461+ if !strings.Contains(m.pattern, "/") {
462+ continue
463+ }
464+ from := path.Dir(m.pattern)
465+ if !strings.HasPrefix(from, "/") {
466+ from = path.Join("/"+dir, from)
467+ }
468+ from = strings.TrimPrefix(path.Clean(from), "/")
469+ if from == dir || (from != "" && !strings.HasPrefix(dir, from+"/")) {
470+ continue
471+ }
472+ rest := strings.Split(strings.TrimPrefix(dir, from), "/")
473+ d := from
474+ for _, elem := range rest {
475+ if elem == "" {
476+ continue
477+ }
478+ rules, cleared, err := l.readMerge(m, d, read)
479+ if err != nil {
480+ return nil, err
481+ }
482+ if cleared {
483+ parent.lists[i] = nil
484+ }
485+ if m.flags&ruleNoInherit == 0 {
486+ parent.lists[i] = append(rules, parent.lists[i]...)
487+ }
488+ d = path.Join(d, elem)
489+ }
490+ }
491+ return l.enter(parent, dir, read)
492+}
493+
494+// enter returns the scope for dir, a storage path, below parent's
495+// directory (push_local_filters).
496+func (l *filterList) enter(parent *filterScope, dir string, read func(string) ([]byte, error)) (*filterScope, error) {
497+ if len(l.merges) == 0 {
498+ return parent, nil
499+ }
500+ sc := &filterScope{root: parent.root, lists: make([][]*filterRule, len(l.merges))}
501+ for i, m := range l.merges {
502+ var inherited []*filterRule
503+ if m.flags&ruleNoInherit == 0 {
504+ inherited = parent.lists[i]
505+ }
506+ rules, cleared, err := l.readMerge(m, dir, read)
507+ if err != nil {
508+ return nil, err
509+ }
510+ if cleared {
511+ inherited = nil
512+ }
513+ sc.lists[i] = append(rules, inherited...)
514+ }
515+ return sc, nil
516+}
517+
518+// readMerge parses the merge file for m in dir (parse_filter_file). A "!"
519+// clears the rules before it, inherited ones included.
520+func (l *filterList) readMerge(m *filterRule, dir string, read func(string) ([]byte, error)) (rules []*filterRule, cleared bool, err error) {
521+ name := path.Join(dir, path.Base(m.pattern))
522+ data, err := read(name)
523+ if err != nil {
524+ return nil, false, &mergeFileError{name: name, err: err}
525+ }
526+ var toks []string
527+ if m.flags&ruleWordSplit != 0 {
528+ toks = strings.FieldsFunc(string(data), func(r rune) bool { return r < 0x80 && isSpace(byte(r)) })
529+ } else {
530+ text := strings.ReplaceAll(string(data), "\r\n", "\n")
531+ for line := range strings.FieldsFuncSeq(text, func(r rune) bool { return r == '\n' || r == '\r' }) {
532+ if line[0] != '#' && line[0] != ';' {
533+ toks = append(toks, line)
534+ }
535+ }
536+ }
537+ for _, tok := range toks {
538+ rule, err := parseRuleTok(tok, m.flags, false)
539+ if err != nil {
540+ return nil, false, &mergeFileError{name: name, err: err}
541+ }
542+ if rule == nil {
543+ continue
544+ }
545+ switch {
546+ case rule.flags&ruleClearList != 0:
547+ rules, cleared = nil, true
548+ continue
549+ case rule.flags&ruleMergeFile != 0:
550+ return nil, false, &mergeFileError{name: name, err: errors.New("merge rules inside a merge file are not supported")}
551+ case rule.flags&ruleCVSIgnore != 0:
552+ rules = append(rules, l.cvsIgnoreRules()...)
553+ continue
554+ }
555+ if l.deleteExcluded && rule.flags&sideFlags == 0 {
556+ rule.flags |= ruleSenderSide
557+ }
558+ if rule.flags&sideFlags == l.otherSide {
559+ continue
560+ }
561+ pat := rule.pattern
562+ if rule.flags&ruleAbsPath == 0 && strings.HasPrefix(pat, "/") {
563+ // Anchored patterns are relative to the file's directory.
564+ rule.flags |= ruleAbsPath
565+ if dir != "" {
566+ pat = "/" + dir + pat
567+ }
568+ }
569+ r := newRule(pat, rule.flags)
570+ rules = append(rules, r)
571+ }
572+ return rules, cleared, nil
193573 }
194574
195575 // check returns -1 if name is excluded, 1 if it is included and 0 when no
196-// rule matched. name is relative to the transfer root.
197-func (l *filterList) check(name string, isDir bool) int {
576+// rule matched. name is relative to the transfer root. sc holds the
577+// per-directory rules for name's directory, if any have been read.
578+func (l *filterList) check(name string, isDir bool, sc *filterScope) int {
579+ return l.checkPerishable(name, isDir, sc, false)
580+}
581+
582+// checkPerishable is check, skipping perishable rules when
583+// ignorePerishable is set (rsync's ignore_perishable).
584+func (l *filterList) checkPerishable(name string, isDir bool, sc *filterScope, ignorePerishable bool) int {
585+ root := ""
586+ if sc != nil {
587+ root = sc.root
588+ }
198589 for _, r := range l.rules {
199- if r.matches(name, isDir) {
590+ if ignorePerishable && r.flags&rulePerishable != 0 {
591+ continue
592+ }
593+ if r.flags&rulePerdirMerge != 0 {
594+ if sc != nil {
595+ if rc := checkRules(sc.lists[r.merge], name, isDir, root, ignorePerishable); rc != 0 {
596+ return rc
597+ }
598+ }
599+ continue
600+ }
601+ if rc := checkRules([]*filterRule{r}, name, isDir, root, ignorePerishable); rc != 0 {
602+ return rc
603+ }
604+ }
605+ return 0
606+}
607+
608+func checkRules(rules []*filterRule, name string, isDir bool, root string, ignorePerishable bool) int {
609+ for _, r := range rules {
610+ if ignorePerishable && r.flags&rulePerishable != 0 {
611+ continue
612+ }
613+ if r.matches(name, isDir, root) {
200614 if r.flags&ruleInclude != 0 {
201615 return 1
202616 }
......@@ -206,12 +620,19 @@ func (l *filterList) check(name string, isDir bool) int {
206620 return 0
207621 }
208622
209-func (l *filterList) excluded(name string, isDir bool) bool {
210- return l != nil && l.check(name, isDir) < 0
623+func (l *filterList) excluded(name string, isDir bool, sc *filterScope) bool {
624+ return l != nil && l.check(name, isDir, sc) < 0
625+}
626+
627+// protectsInDeletedDir reports whether a rule keeps name, which is inside a
628+// directory being deleted. Perishable rules don't, so the directory can go.
629+func (l *filterList) protectsInDeletedDir(name string, isDir bool, sc *filterScope) bool {
630+ return l != nil && l.checkPerishable(name, isDir, sc, true) < 0
211631 }
212632
213-// matches is rsync's rule_matches.
214-func (r *filterRule) matches(fname string, isDir bool) bool {
633+// matches is rsync's rule_matches. Absolute patterns match the storage
634+// path, which is name below root.
635+func (r *filterRule) matches(fname string, isDir bool, root string) bool {
215636 ret := r.flags&ruleNegate == 0
216637 name := strings.TrimPrefix(fname, "/")
217638 if name == "" || r.flags&ruleXattr != 0 {
......@@ -224,6 +645,8 @@ func (r *filterRule) matches(fname string, isDir bool) bool {
224645 if i := strings.LastIndexByte(name, '/'); i >= 0 {
225646 name = name[i+1:]
226647 }
648+ case r.flags&ruleAbsPath != 0 && !strings.HasPrefix(fname, "/") && root != "":
649+ strs = append(strs, root, "/")
227650 case r.flags&ruleWild2Prefix != 0 && !strings.HasPrefix(fname, "/"):
228651 strs = append(strs, "/")
229652 }
M pkg/rsync/fs.go
+6, -0
......@@ -1,10 +1,15 @@
11 package rsync
22
33 import (
4+ "errors"
45 "io"
56 "time"
67 )
78
9+// ErrFileIgnored lets Put skip a file without failing the transfer or
10+// acknowledging it for --remove-source-files.
11+var ErrFileIgnored = errors.New("file ignored")
12+
813 // FileInfo describes an entry in an FS.
914 type FileInfo struct {
1015 // Name is slash-separated and relative to the directory being listed.
......@@ -34,6 +39,7 @@ type FS interface {
3439 // Put stores a regular file with the data read from r. If the transfer
3540 // fails, r returns an error instead of io.EOF and Put must not keep
3641 // what it read. The returned message, if any, is shown to the user.
42+ // Return an error wrapping ErrFileIgnored for an intentional skip.
3743 Put(name string, info FileInfo, r io.Reader) (string, error)
3844 // Remove deletes a file or an empty directory.
3945 Remove(name string, isDir bool) error
M pkg/rsync/integration_test.go
+249, -1
......@@ -91,6 +91,18 @@ func (d dirFS) Put(name string, info FileInfo, r io.Reader) (string, error) {
9191
9292 func (d dirFS) Remove(name string, isDir bool) error { return os.Remove(d.full(name)) }
9393
94+type rejectingFS struct {
95+ dirFS
96+ err error
97+}
98+
99+func (d rejectingFS) Put(name string, info FileInfo, r io.Reader) (string, error) {
100+ if filepath.Base(name) == "skip.txt" {
101+ return "", d.err
102+ }
103+ return d.dirFS.Put(name, info, r)
104+}
105+
94106 // TestHelperProcess is the remote shell rsync runs in these tests. It serves
95107 // the directory in RSYNC_TEST_ROOT on stdin and stdout.
96108 func TestHelperProcess(t *testing.T) {
......@@ -111,7 +123,14 @@ func TestHelperProcess(t *testing.T) {
111123 io.Reader
112124 io.Writer
113125 }{os.Stdin, os.Stdout}
114- err := Serve(Config{FS: dirFS{root: root}, Stderr: os.Stderr}, rw, args)
126+ var fsys FS = dirFS{root: root}
127+ switch os.Getenv("RSYNC_TEST_PUT_RESULT") {
128+ case "ignored":
129+ fsys = rejectingFS{dirFS{root: root}, fmt.Errorf("policy: %w", ErrFileIgnored)}
130+ case "failed":
131+ fsys = rejectingFS{dirFS{root: root}, errors.New("storage backend failed")}
132+ }
133+ err := Serve(Config{FS: fsys, Stderr: os.Stderr}, rw, args)
115134 code := 0
116135 var exit *ExitError
117136 if errors.As(err, &exit) {
......@@ -670,6 +689,235 @@ func TestRsyncSkipOptions(t *testing.T) {
670689 }
671690 }
672691
692+func TestRsyncPutResults(t *testing.T) {
693+ for _, proto := range protocols {
694+ for _, result := range []string{"ignored", "failed"} {
695+ t.Run("protocol_"+proto+"/"+result, func(t *testing.T) {
696+ t.Setenv("RSYNC_TEST_PUT_RESULT", result)
697+ e := newRsyncEnv(t)
698+ for _, name := range []string{"a.txt", "skip.txt", "z.txt"} {
699+ writeFile(t, filepath.Join(e.local, name), textBytes(100_000, name))
700+ }
701+ writeFile(t, filepath.Join(e.remote, "site/skip.txt"), []byte("original"))
702+ out, err := e.run(append(protoArgs(proto), "-r", "--remove-source-files", e.local+"/", "host:site")...)
703+ if result == "ignored" {
704+ if err != nil {
705+ t.Fatalf("ignored file failed the batch: %v\n%s", err, out)
706+ }
707+ } else {
708+ var exit *exec.ExitError
709+ if !errors.As(err, &exit) || exit.ExitCode() != 23 {
710+ t.Fatalf("want exit 23, got %v\n%s", err, out)
711+ }
712+ }
713+ got := tree(t, filepath.Join(e.remote, "site"))
714+ if len(got) != 3 || got["skip.txt"] != "original" {
715+ t.Fatalf("rejected file changed: %v", got)
716+ }
717+ for _, name := range []string{"a.txt", "z.txt"} {
718+ if got[name] != string(textBytes(100_000, name)) {
719+ t.Errorf("accepted file %s not stored", name)
720+ }
721+ if _, err := os.Stat(filepath.Join(e.local, name)); !errors.Is(err, os.ErrNotExist) {
722+ t.Errorf("accepted file %s not removed from source: %v", name, err)
723+ }
724+ }
725+ if data, err := os.ReadFile(filepath.Join(e.local, "skip.txt")); err != nil || !bytes.Equal(data, textBytes(100_000, "skip.txt")) {
726+ t.Fatalf("rejected source file changed: %v", err)
727+ }
728+ })
729+ }
730+ }
731+}
732+
733+// writeTree writes files, named relative to root.
734+func writeTree(t *testing.T, root string, files map[string]string) {
735+ t.Helper()
736+ for name, contents := range files {
737+ writeFile(t, filepath.Join(root, name), []byte(contents))
738+ }
739+}
740+
741+// mergeFilterTree has per-directory merge files covering inheritance,
742+// overrides, anchoring, clearing and protect rules.
743+var mergeFilterTree = map[string]string{
744+ ".rsync-filter": "- keep.log\nP precious/\n- *.tmp\n-p *.perish\n",
745+ "keep.log": "keep",
746+ "stale.txt": "stale",
747+ "x.tmp": "tmp",
748+ "precious/a.txt": "precious",
749+ "sub/.rsync-filter": "+ sub-only.tmp\n- /anchored.txt\n",
750+ "sub/a.tmp": "tmp",
751+ "sub/sub-only.tmp": "kept by include",
752+ "sub/anchored.txt": "anchored",
753+ "sub/deeper/anchored.txt": "not anchored here",
754+ "sub/deeper/keep.log": "inherited",
755+ "cleared/.rsync-filter": "!\n- only.txt\n",
756+ "cleared/keep.log": "no longer excluded",
757+ "cleared/only.txt": "only",
758+ "gone/.rsync-filter": "- inside.log\n",
759+ "gone/inside.log": "protects its directory",
760+ "gone/other.txt": "other",
761+ ".cvsignore": "precious.dat\n",
762+ "precious.dat": "cvs",
763+ "obj.o": "default cvs pattern",
764+ "nested/.cvsignore": "nested.dat",
765+ "nested/nested.dat": "nested",
766+ "nested/deeper/nested.dat": "cvsignore does not inherit",
767+ "nested/deeper/precious.dat": "nor does the parent's",
768+ "top.perish": "perishable rules protect a kept directory's entries",
769+ "perish/keep.perish": "but not a deleted directory's",
770+ "perish/deeper/keep.perish": "at any depth",
771+ "objdir/x.o": "default CVS patterns are perishable from protocol 30",
772+}
773+
774+// TestRsyncMergeFiltersMatchRsync runs the same transfer against real rsync
775+// and against our server, each from a copy of mergeFilterTree, and expects
776+// the same result.
777+func TestRsyncMergeFiltersMatchRsync(t *testing.T) {
778+ local := map[string]string{"index.html": "new", "sub/new.txt": "new"}
779+ pushes := [][]string{
780+ {"-F", "--delete"},
781+ {"-F", "--delete-before"},
782+ {"-F", "--delete-during"},
783+ {"-F", "--delete-after"},
784+ {"-F", "--delete-delay"},
785+ {"-FF", "--delete"},
786+ {"-F", "--delete", "--delete-excluded"},
787+ {"-C", "--delete"},
788+ {"-CF", "--delete"},
789+ {"--filter=:e .rsync-filter", "--delete"},
790+ {"--filter=:n- .rsync-filter", "--delete"},
791+ {"--filter=dir-merge,- .rsync-filter", "--delete"},
792+ }
793+ for _, proto := range []string{"29", "30", "31", ""} {
794+ for _, flags := range pushes {
795+ t.Run(fmt.Sprintf("push/proto=%s/%s", proto, strings.Join(flags, " ")), func(t *testing.T) {
796+ if proto == "29" && slices.Contains(flags, "--delete-delay") {
797+ t.Skip("needs protocol 30")
798+ }
799+ e := newRsyncEnv(t)
800+ writeTree(t, e.local, local)
801+ ref := filepath.Join(t.TempDir(), "site")
802+ writeTree(t, ref, mergeFilterTree)
803+ writeTree(t, filepath.Join(e.remote, "site"), mergeFilterTree)
804+ args := append(protoArgs(proto), "-r")
805+ args = append(args, flags...)
806+
807+ refCmd := exec.Command("rsync", append(slices.Clone(args), e.local+"/", ref)...)
808+ if out, err := refCmd.CombinedOutput(); err != nil {
809+ t.Fatalf("reference rsync: %v\n%s", err, out)
810+ }
811+ e.mustRun(append(args, e.local+"/", "host:site")...)
812+ if slices.Contains(flags, "--delete-delay") {
813+ // rsync drops inherited per-directory rules before its
814+ // delayed deletions, so it removes this file. We keep it.
815+ inherited := filepath.Join(e.remote, "site/sub/deeper/keep.log")
816+ if err := os.Remove(inherited); err != nil {
817+ t.Fatalf("file protected by an inherited rule was deleted: %v", err)
818+ }
819+ }
820+ assertSameTree(t, ref, filepath.Join(e.remote, "site"))
821+ })
822+ }
823+ }
824+
825+ pulls := [][]string{{"-F"}, {"-FF"}, {"-C"}, {"--filter=:n .rsync-filter"}}
826+ for _, proto := range []string{"29", "30", "31", ""} {
827+ for _, flags := range pulls {
828+ for _, src := range []string{"site/", "site"} {
829+ t.Run(fmt.Sprintf("pull/proto=%s/%s/%s", proto, strings.Join(flags, " "), src), func(t *testing.T) {
830+ e := newRsyncEnv(t)
831+ writeTree(t, filepath.Join(e.remote, "site"), mergeFilterTree)
832+ refSrc := t.TempDir()
833+ writeTree(t, filepath.Join(refSrc, "site"), mergeFilterTree)
834+ ref := t.TempDir()
835+ args := append(protoArgs(proto), "-r")
836+ args = append(args, flags...)
837+
838+ refCmd := exec.Command("rsync", append(slices.Clone(args), filepath.Join(refSrc, src), ref+"/")...)
839+ if strings.HasSuffix(src, "/") {
840+ refCmd.Args[len(refCmd.Args)-2] += "/"
841+ }
842+ if out, err := refCmd.CombinedOutput(); err != nil {
843+ t.Fatalf("reference rsync: %v\n%s", err, out)
844+ }
845+ e.mustRun(append(args, "host:"+src, e.local+"/")...)
846+ assertSameTree(t, ref, e.local)
847+ })
848+ }
849+ }
850+ }
851+}
852+
853+// TestRsyncMergeFilterParentScan checks that "-F" reads .rsync-filter in
854+// the directories above the transfer, as rsync does up to "/".
855+func TestRsyncMergeFilterParentScan(t *testing.T) {
856+ e := newRsyncEnv(t)
857+ writeFile(t, filepath.Join(e.local, "index.html"), []byte("new"))
858+ writeTree(t, e.remote, map[string]string{
859+ ".rsync-filter": "- /site/scanned.txt\n",
860+ "site/scanned.txt": "protected from above",
861+ "site/stale.txt": "stale",
862+ "other/scanned.txt": "outside the transfer",
863+ })
864+ e.mustRun("-r", "-F", "--delete", e.local+"/", "host:site")
865+ got := tree(t, filepath.Join(e.remote, "site"))
866+ if _, ok := got["scanned.txt"]; !ok {
867+ t.Error("rule from the parent directory's merge file was ignored")
868+ }
869+ if _, ok := got["stale.txt"]; ok {
870+ t.Error("unprotected file survived --delete")
871+ }
872+}
873+
874+// TestRsyncMergeFilterErrors checks that a merge file we can't use stops
875+// deletions on upload and the whole transfer on download.
876+func TestRsyncMergeFilterErrors(t *testing.T) {
877+ for _, contents := range []string{"merge other\n", "bogus rule\n"} {
878+ t.Run("push/"+contents, func(t *testing.T) {
879+ e := newRsyncEnv(t)
880+ writeFile(t, filepath.Join(e.local, "index.html"), []byte("new"))
881+ writeTree(t, filepath.Join(e.remote, "site"), map[string]string{
882+ "sub/.rsync-filter": contents,
883+ "sub/stale.txt": "stale",
884+ "stale.txt": "stale",
885+ })
886+ out, err := e.run("-r", "-F", "--delete-after", e.local+"/", "host:site")
887+ var exit *exec.ExitError
888+ if !errors.As(err, &exit) || exit.ExitCode() != exitPartial {
889+ t.Fatalf("want exit %d, got %v\n%s", exitPartial, err, out)
890+ }
891+ if !strings.Contains(out, "skipping file deletion") {
892+ t.Errorf("no explanation for skipped deletion\n%s", out)
893+ }
894+ got := tree(t, filepath.Join(e.remote, "site"))
895+ if got["index.html"] != "new" {
896+ t.Error("file was not transferred")
897+ }
898+ for _, name := range []string{"stale.txt", "sub/stale.txt"} {
899+ if _, ok := got[name]; !ok {
900+ t.Errorf("%s was deleted despite the unusable merge file", name)
901+ }
902+ }
903+ })
904+ t.Run("pull/"+contents, func(t *testing.T) {
905+ e := newRsyncEnv(t)
906+ writeTree(t, filepath.Join(e.remote, "site"), map[string]string{
907+ "sub/.rsync-filter": contents,
908+ "index.html": "index",
909+ })
910+ out, err := e.run("-r", "-F", "host:site/", e.local+"/")
911+ if err == nil || !strings.Contains(out, "per-directory filter file") {
912+ t.Fatalf("want a merge-file error, got %v\n%s", err, out)
913+ }
914+ if got := tree(t, e.local); len(got) != 0 {
915+ t.Errorf("files transferred despite the error: %v", got)
916+ }
917+ })
918+ }
919+}
920+
673921 func TestRsyncUnsupportedOptions(t *testing.T) {
674922 e := newRsyncEnv(t)
675923 populate(t, e.local)
M pkg/rsync/receiver.go
+27, -7
......@@ -508,6 +508,14 @@ func (g *generator) deletePass() error {
508508 return f.flags&flagContentDir != 0
509509 }
510510
511+ files := make(map[string]bool, len(existing))
512+ for _, e := range existing {
513+ if name, err := cleanSourcePath(e.Name); err == nil && !e.IsDir {
514+ files[name] = true
515+ }
516+ }
517+ scopes := s.filters.newScopeTree(g.dest.dir, g.dest.dir, s.mergeReader(g.dest.dir, files))
518+
511519 type victim struct {
512520 name string
513521 isDir bool
......@@ -538,11 +546,21 @@ func (g *generator) deletePass() error {
538546 if !contentDir(anc) {
539547 continue
540548 }
549+ // Walk down from the ancestor so merge files are only read in
550+ // directories rsync would enter. Below it everything is inside a
551+ // directory being deleted.
541552 protected := false
542- for i, p := range chain {
543- if s.filters.excluded(p, i > 0 || e.IsDir) {
544- protected = true
545- break
553+ for i := len(chain) - 1; i >= 0 && !protected; i-- {
554+ sc, err := scopes.scope(path.Dir(chain[i]))
555+ if err != nil {
556+ s.xferError("%v -- skipping file deletion", err)
557+ return nil
558+ }
559+ isDir := i > 0 || e.IsDir
560+ if i == len(chain)-1 {
561+ protected = s.filters.excluded(chain[i], isDir, sc)
562+ } else {
563+ protected = s.filters.protectsInDeletedDir(chain[i], isDir, sc)
546564 }
547565 }
548566 if protected {
......@@ -702,9 +720,11 @@ func (r *receiver) run() error {
702720 if !errors.As(err, &putErr) {
703721 return err
704722 }
705- // Storage refusing a file (pgs rejecting a dotfile, say) is
706- // reported without failing the whole run.
707- s.warning("rsync: [receiver] %s not stored: %v", f.name(), putErr.err)
723+ if errors.Is(putErr.err, ErrFileIgnored) {
724+ s.warning("rsync: [receiver] %s skipped: %v", f.name(), putErr.err)
725+ } else {
726+ s.xferError("%s not stored: %v", f.name(), putErr.err)
727+ }
708728 }
709729 }
710730 if err := r.notify(recvEvent{done: true}); err != nil {
M pkg/rsync/sender.go
+18, -6
......@@ -69,7 +69,7 @@ func (s *session) buildFileList(args []string) ([]*fileEntry, error) {
6969 }
7070 continue
7171 }
72- if nameType != nameDotDir && s.filters.excluded(fn, info.IsDir) {
72+ if nameType != nameDotDir && s.filters.excluded(fn, info.IsDir, nil) {
7373 continue
7474 }
7575 if info.IsDir && !s.opts.xferDirs {
......@@ -81,7 +81,11 @@ func (s *session) buildFileList(args []string) ([]*fileEntry, error) {
8181 top := s.newEntry(fn, src, info, flagTopDir|flagContentDir|flags)
8282 files = append(files, top)
8383 if info.IsDir {
84- children, err := s.listDir(fn, src, flags)
84+ children, err := s.listDir(fn, src, dir, flags)
85+ var mergeErr *mergeFileError
86+ if errors.As(err, &mergeErr) {
87+ return nil, err
88+ }
8589 if err != nil {
8690 s.ioError |= ioErrGeneral
8791 s.xferError("opendir %q failed: %v", "/"+src, err)
......@@ -98,8 +102,9 @@ func (s *session) buildFileList(args []string) ([]*fileEntry, error) {
98102
99103 // listDir returns the entries below a source directory, with names under
100104 // prefix. Directories missing from the listing are implied by their
101-// contents, as object stores don't keep them.
102-func (s *session) listDir(prefix, src string, flags int) ([]*fileEntry, error) {
105+// contents, as object stores don't keep them. root is the storage path of
106+// the transfer root.
107+func (s *session) listDir(prefix, src, root string, flags int) ([]*fileEntry, error) {
103108 entries, err := s.fs.ReadDir(src, s.opts.recurse)
104109 if err != nil {
105110 return nil, err
......@@ -132,10 +137,13 @@ func (s *session) listDir(prefix, src string, flags int) ([]*fileEntry, error) {
132137 }
133138 }
134139 names := make([]string, 0, len(byName))
135- for name := range byName {
140+ regular := make(map[string]bool, len(byName))
141+ for name, e := range byName {
136142 names = append(names, name)
143+ regular[name] = !e.IsDir
137144 }
138145 slices.Sort(names)
146+ scopes := s.filters.newScopeTree(root, src, s.mergeReader(src, regular))
139147
140148 var files []*fileEntry
141149 var excludedDirs []string
......@@ -145,7 +153,11 @@ func (s *session) listDir(prefix, src string, flags int) ([]*fileEntry, error) {
145153 continue
146154 }
147155 full := join(name)
148- if s.filters.excluded(full, e.IsDir) {
156+ sc, err := scopes.scope(path.Dir(name))
157+ if err != nil {
158+ return nil, err
159+ }
160+ if s.filters.excluded(full, e.IsDir, sc) {
149161 if e.IsDir {
150162 excludedDirs = append(excludedDirs, name)
151163 }
M pkg/rsync/server.go
+9, -2
......@@ -191,6 +191,11 @@ func Serve(cfg Config, rw io.ReadWriter, args []string) error {
191191 infoDel: o.Info(rsyncopts.INFO_DEL),
192192 infoNonreg: o.Info(rsyncopts.INFO_NONREG),
193193 }
194+ s.filters.deleteExcluded = s.opts.deleteExcluded
195+ s.filters.otherSide = ruleSenderSide
196+ if s.opts.sender {
197+ s.filters.otherSide = ruleReceiverSide
198+ }
194199
195200 if err := s.setupProtocol(o); err != nil {
196201 var exit *ExitError
......@@ -199,6 +204,7 @@ func Serve(cfg Config, rw io.ReadWriter, args []string) error {
199204 }
200205 return early(exitProtocol, err)
201206 }
207+ s.filters.cvsPerishable = s.c.protocol >= 30
202208
203209 if err := c.w.setMultiplex(true); err != nil {
204210 return &ExitError{Code: exitStreamIO, Err: err}
......@@ -663,9 +669,10 @@ func (s *session) recvFilterList() error {
663669 }
664670 }
665671 if s.opts.cvsExclude && (s.opts.sender || s.c.protocol < 29) {
666- for _, pat := range strings.Fields(defaultCVSIgnore) {
667- s.filters.add(pat, ruleNoPrefixes)
672+ if err := s.filters.parse(":C", false); err != nil {
673+ return s.fatal(exitSyntax, err)
668674 }
675+ s.filters.rules = append(s.filters.rules, s.filters.cvsIgnoreRules()...)
669676 }
670677 return nil
671678 }
M pkg/rsync/wire_test.go
+160, -2
......@@ -2,8 +2,11 @@ package rsync
22
33 import (
44 "bytes"
5+ "errors"
56 "io"
7+ "path"
68 "slices"
9+ "strings"
710 "testing"
811 "time"
912 )
......@@ -297,6 +300,161 @@ func TestWildmatch(t *testing.T) {
297300 }
298301 }
299302
303+func mapReader(files map[string]string) func(string) ([]byte, error) {
304+ return func(name string) ([]byte, error) {
305+ data, ok := files[name]
306+ if !ok {
307+ return nil, nil
308+ }
309+ return []byte(data), nil
310+ }
311+}
312+
313+func TestPerDirMergeRules(t *testing.T) {
314+ files := map[string]string{
315+ ".rsync-filter": "- /site/scanned.txt\n- /outside.txt\n",
316+ "site/.rsync-filter": "# comment\n; comment\n- *.log\nP protected.txt\nH hidden.txt\nexclude long.txt\r\n",
317+ "site/sub/.rsync-filter": "+ important.log\n- /anchored.txt\n",
318+ "site/clear/.rsync-filter": "!\n- z\n",
319+ "site/.cvsignore": "a.o b.o\n",
320+ "site/sub/.cvsignore": "! c.o",
321+ }
322+ tests := []struct {
323+ otherSide int
324+ name string
325+ want int
326+ }{
327+ {ruleSenderSide, "a.log", -1},
328+ {ruleSenderSide, "sub/x.log", -1},
329+ {ruleSenderSide, "sub/important.log", 1},
330+ {ruleSenderSide, "long.txt", -1},
331+ {ruleSenderSide, "scanned.txt", -1},
332+ {ruleSenderSide, "outside.txt", 0},
333+ {ruleSenderSide, "sub/anchored.txt", -1},
334+ {ruleSenderSide, "anchored.txt", 0},
335+ {ruleSenderSide, "sub/deeper/anchored.txt", 0},
336+ {ruleSenderSide, "clear/a.log", 0},
337+ {ruleSenderSide, "clear/z", -1},
338+ {ruleSenderSide, ".rsync-filter", -1},
339+ {ruleSenderSide, "a.o", -1},
340+ {ruleSenderSide, "sub/a.o", 0},
341+ {ruleSenderSide, "sub/c.o", -1},
342+ {ruleSenderSide, "protected.txt", -1},
343+ {ruleSenderSide, "hidden.txt", 0},
344+ {ruleReceiverSide, "protected.txt", 0},
345+ {ruleReceiverSide, "hidden.txt", -1},
346+ }
347+ for _, tt := range tests {
348+ l := filterList{otherSide: tt.otherSide}
349+ for _, rule := range []string{":e /.rsync-filter", ": .rsync-filter", ":C"} {
350+ if err := l.parse(rule, false); err != nil {
351+ t.Fatalf("parse %q: %v", rule, err)
352+ }
353+ }
354+ if len(l.merges) != 2 {
355+ t.Fatalf("duplicate merge rule kept: %d merges", len(l.merges))
356+ }
357+ sc, err := l.newScopeTree("site", "site", mapReader(files)).scope(path.Dir(tt.name))
358+ if err != nil {
359+ t.Fatal(err)
360+ }
361+ if got := l.check(tt.name, false, sc); got != tt.want {
362+ t.Errorf("otherSide=%d check(%q) = %d, want %d", tt.otherSide, tt.name, got, tt.want)
363+ }
364+ }
365+}
366+
367+func TestPerDirMergeDeleteExcluded(t *testing.T) {
368+ l := filterList{otherSide: ruleSenderSide, deleteExcluded: true}
369+ if err := l.parse(": .rsync-filter", false); err != nil {
370+ t.Fatal(err)
371+ }
372+ files := map[string]string{"site/.rsync-filter": "- excluded.txt\nP protected.txt\n"}
373+ sc, err := l.newScopeTree("site", "site", mapReader(files)).scope(".")
374+ if err != nil {
375+ t.Fatal(err)
376+ }
377+ if l.excluded("excluded.txt", false, sc) {
378+ t.Error("--delete-excluded should let the receiver delete excluded files")
379+ }
380+ if !l.excluded("protected.txt", false, sc) {
381+ t.Error("protect rules still apply with --delete-excluded")
382+ }
383+}
384+
385+func TestPerDirMergeNoInherit(t *testing.T) {
386+ l := filterList{otherSide: ruleSenderSide}
387+ if err := l.parse(":n /.rsync-filter", false); err != nil {
388+ t.Fatal(err)
389+ }
390+ files := map[string]string{
391+ ".rsync-filter": "- *.root\n",
392+ "site/.rsync-filter": "- *.tmp\n",
393+ }
394+ scopes := l.newScopeTree("site", "site", mapReader(files))
395+ for _, tt := range []struct {
396+ name string
397+ want int
398+ }{{"x.tmp", -1}, {"sub/x.tmp", 0}, {"x.root", 0}} {
399+ sc, err := scopes.scope(path.Dir(tt.name))
400+ if err != nil {
401+ t.Fatal(err)
402+ }
403+ if got := l.check(tt.name, false, sc); got != tt.want {
404+ t.Errorf("check(%q) = %d, want %d", tt.name, got, tt.want)
405+ }
406+ }
407+}
408+
409+func TestPerDirMergeErrors(t *testing.T) {
410+ for _, rule := range []string{". rules", "merge rules", ":! .rsync-filter", ": x/"} {
411+ var l filterList
412+ if err := l.parse(rule, false); err == nil {
413+ t.Errorf("parse %q: want an error", rule)
414+ }
415+ }
416+ for _, contents := range []string{
417+ "merge other\n",
418+ ": .nested\n",
419+ "dir-merge .nested\n",
420+ "bogus\n",
421+ "-foo\n",
422+ "! trailing\n",
423+ "- " + strings.Repeat("x", maxPathLen) + "\n",
424+ } {
425+ l := filterList{otherSide: ruleSenderSide}
426+ if err := l.parse(": .rsync-filter", false); err != nil {
427+ t.Fatal(err)
428+ }
429+ files := map[string]string{"site/.rsync-filter": contents}
430+ _, err := l.newScopeTree("site", "site", mapReader(files)).scope(".")
431+ var mergeErr *mergeFileError
432+ if !errors.As(err, &mergeErr) {
433+ t.Errorf("merge file %q: want a mergeFileError, got %v", contents, err)
434+ }
435+ }
436+
437+ l := filterList{otherSide: ruleSenderSide}
438+ if err := l.parse(": .rsync-filter", false); err != nil {
439+ t.Fatal(err)
440+ }
441+ readErr := errors.New("backend down")
442+ _, err := l.newScopeTree("site", "site", func(string) ([]byte, error) { return nil, readErr }).scope(".")
443+ if !errors.Is(err, readErr) {
444+ t.Errorf("read error not reported: %v", err)
445+ }
446+
447+ for _, pattern := range []string{". rules", ": rules"} {
448+ var l filterList
449+ if err := l.parse(pattern, true); err != nil {
450+ t.Fatal(err)
451+ }
452+ if l.check(pattern, false, nil) != -1 {
453+ t.Errorf("protocol 28 pattern %q should remain an exclusion", pattern)
454+ }
455+ }
456+}
457+
300458 func TestFilterRules(t *testing.T) {
301459 var l filterList
302460 for _, rule := range []string{"+ keep.log", "- *.log", "- /build/", "- node_modules/", "- docs/*.tmp", "-! *.html"} {
......@@ -321,7 +479,7 @@ func TestFilterRules(t *testing.T) {
321479 {"index.html", false, 0},
322480 }
323481 for _, tt := range tests {
324- if got := l.check(tt.name, tt.isDir); got != tt.want {
482+ if got := l.check(tt.name, tt.isDir, nil); got != tt.want {
325483 t.Errorf("check(%q, dir=%v) = %d, want %d", tt.name, tt.isDir, got, tt.want)
326484 }
327485 }
......@@ -329,7 +487,7 @@ func TestFilterRules(t *testing.T) {
329487 var old filterList
330488 _ = old.parse("*.o", true)
331489 _ = old.parse("+ keep.o", true)
332- if old.check("x.o", false) != -1 {
490+ if old.check("x.o", false, nil) != -1 {
333491 t.Error("protocol 28 bare pattern is an exclude")
334492 }
335493 _ = old.parse("!", true)
M pkg/send/protocols/rsync/rsync.go
+13, -1
......@@ -90,7 +90,7 @@ func (s *storageFS) Open(name string) (rsync.File, rsync.FileInfo, error) {
9090 }
9191
9292 func (s *storageFS) Put(name string, info rsync.FileInfo, r io.Reader) (string, error) {
93- return s.handler.Write(s.session, &utils.FileEntry{
93+ msg, err := s.handler.Write(s.session, &utils.FileEntry{
9494 Filepath: abs(name),
9595 Mode: fs.FileMode(0600),
9696 Size: info.Size,
......@@ -98,8 +98,20 @@ func (s *storageFS) Put(name string, info rsync.FileInfo, r io.Reader) (string,
9898 Atime: info.ModTime.Unix(),
9999 Reader: r,
100100 })
101+ if errors.Is(err, utils.ErrFileIgnored) {
102+ return msg, ignoredError{err}
103+ }
104+ return msg, err
101105 }
102106
107+// ignoredError marks a handler's ignored file for rsync, keeping the
108+// handler's message.
109+type ignoredError struct{ error }
110+
111+func (e ignoredError) Is(target error) bool { return target == rsync.ErrFileIgnored }
112+
113+func (e ignoredError) Unwrap() error { return e.error }
114+
103115 func (s *storageFS) Remove(name string, isDir bool) error {
104116 return s.handler.Delete(s.session, &utils.FileEntry{Filepath: abs(name)})
105117 }
M pkg/send/protocols/scp/copy_from_client.go
+19, -3
......@@ -9,6 +9,7 @@ import (
99 "path/filepath"
1010 "regexp"
1111 "strconv"
12+ "strings"
1213
1314 "github.com/picosh/pico/pkg/pssh"
1415 "github.com/picosh/pico/pkg/send/utils"
......@@ -85,19 +86,27 @@ func copyFromClient(session *pssh.SSHServerConnSession, info Info, handler utils
8586 // accepts the header
8687 _, _ = session.Write(utils.NULL)
8788
89+ reader := utils.NewLimitReader(r, int(size))
8890 result, err := handler.Write(session, &utils.FileEntry{
8991 Filepath: filepath.Join(path, name),
9092 Mode: fs.FileMode(mode),
9193 Size: size,
9294 Mtime: mtime,
9395 Atime: atime,
94- Reader: utils.NewLimitReader(r, int(size)),
96+ Reader: reader,
9597 })
98+ // A handler that rejects a file may not read all of it. Skip
99+ // the rest so the next header is read from the right place.
100+ if _, derr := io.Copy(io.Discard, reader); derr != nil {
101+ return fmt.Errorf("failed to read file: %q: %w", name, derr)
102+ }
96103
97- if err == nil {
104+ switch {
105+ case err == nil:
98106 writeSuccess = append(writeSuccess, result)
99- } else {
107+ case errors.Is(err, utils.ErrFileIgnored):
100108 writeErrors = append(writeErrors, err)
109+ default:
101110 fmt.Printf("failed to write file: %q: %v\n", name, err)
102111 }
103112
......@@ -106,6 +115,13 @@ func copyFromClient(session *pssh.SSHServerConnSession, info Info, handler utils
106115
107116 mtime = 0
108117 atime = 0
118+ if err != nil && !errors.Is(err, utils.ErrFileIgnored) {
119+ // A warning reply makes the client report the file and
120+ // exit non-zero once the rest are sent.
121+ msg := strings.ReplaceAll(err.Error(), "\n", " ")
122+ _, _ = fmt.Fprintf(session, "\x01%s\n", msg)
123+ continue
124+ }
109125 // says 'hey im done'
110126 _, _ = session.Write(utils.NULL)
111127 continue
M pkg/send/protocols/sftp/handler.go
+421, -9
......@@ -2,14 +2,17 @@ package sftp
22
33 import (
44 "bytes"
5+ "crypto/sha256"
56 "errors"
67 "fmt"
78 "io"
89 "io/fs"
910 "os"
1011 "path/filepath"
11-
1212 "slices"
13+ "strings"
14+ "sync"
15+ "time"
1316
1417 "github.com/picosh/pico/pkg/pssh"
1518 "github.com/picosh/pico/pkg/send/utils"
......@@ -33,6 +36,41 @@ func (f listerat) ListAt(ls []os.FileInfo, offset int64) (int, error) {
3336 type handler struct {
3437 session *pssh.SSHServerConnSession
3538 writeHandler utils.CopyFromClientHandler
39+
40+ mu sync.Mutex
41+ pending map[string]*pendingFile
42+ // ignored counts open handles to files the handler ignores. A stat of
43+ // one is refused so a mounted filesystem reports that it wasn't saved.
44+ ignored map[string]int
45+ // hidden holds stored files renamed away while open, by the open file
46+ // that will replace them. They look deleted to this session.
47+ hidden map[string]*pendingFile
48+}
49+
50+func (f *handler) pendingFile(name string) *pendingFile {
51+ f.mu.Lock()
52+ defer f.mu.Unlock()
53+ return f.pending[name]
54+}
55+
56+func (f *handler) isHidden(name string) bool {
57+ f.mu.Lock()
58+ defer f.mu.Unlock()
59+ return f.hidden[name] != nil
60+}
61+
62+// release returns the files p was renamed from that are still waiting on
63+// it, and shows them again. The caller holds f.mu.
64+func (f *handler) release(p *pendingFile) []renamedFile {
65+ var out []renamedFile
66+ for _, old := range p.replaces {
67+ if f.hidden[old.path] == p {
68+ delete(f.hidden, old.path)
69+ out = append(out, old)
70+ }
71+ }
72+ p.replaces = nil
73+ return out
3674 }
3775
3876 func (f *handler) Filecmd(r *sftp.Request) error {
......@@ -42,8 +80,34 @@ func (f *handler) Filecmd(r *sftp.Request) error {
4280
4381 if r.Method == "Rmdir" {
4482 entry.Mode = os.ModeDir
83+ return f.writeHandler.Delete(f.session, entry)
4584 }
4685
86+ f.mu.Lock()
87+ hidden := f.hidden[r.Filepath] != nil
88+ p, ok := f.pending[r.Filepath]
89+ var replaced []renamedFile
90+ if ok {
91+ delete(f.pending, r.Filepath)
92+ replaced = f.release(p)
93+ }
94+ f.mu.Unlock()
95+ if hidden && !ok {
96+ return os.ErrNotExist
97+ }
98+ // Removing a renamed file removes what it was renamed from.
99+ for _, old := range replaced {
100+ if err := f.deleteIfUnchanged(old.path, old.sum); err != nil {
101+ return err
102+ }
103+ }
104+ if ok {
105+ // An open file may have been renamed over a stored one.
106+ stored, _, err := f.stored(entry, false)
107+ if err != nil || !stored {
108+ return err
109+ }
110+ }
47111 return f.writeHandler.Delete(f.session, entry)
48112 case "Mkdir":
49113 entry := toFileEntry(r)
......@@ -53,7 +117,17 @@ func (f *handler) Filecmd(r *sftp.Request) error {
53117 _, err := f.writeHandler.Write(f.session, entry)
54118
55119 return err
120+ case "Rename":
121+ return f.rename(r.Filepath, r.Target, false)
56122 case "Setstat":
123+ // Only an open file's size can change, as when an editor
124+ // truncates what it is writing.
125+ if p := f.pendingFile(r.Filepath); p != nil && r.AttrFlags().Size {
126+ f.mu.Lock()
127+ p.dirty = true
128+ f.mu.Unlock()
129+ p.buf.truncate(int64(r.Attributes().Size))
130+ }
57131 return nil
58132 }
59133 return errors.New("unsupported")
......@@ -64,10 +138,29 @@ func (f *handler) Filelist(r *sftp.Request) (sftp.ListerAt, error) {
64138 case "List", "Stat":
65139 list := r.Method == "List"
66140
141+ if !list {
142+ f.mu.Lock()
143+ p, ignored := f.pending[r.Filepath], f.ignored[r.Filepath] > 0
144+ hidden := f.hidden[r.Filepath] != nil
145+ f.mu.Unlock()
146+ if ignored {
147+ return nil, os.ErrPermission
148+ }
149+ if hidden && p == nil {
150+ return nil, os.ErrNotExist
151+ }
152+ if p != nil {
153+ return listerat{p.info(filepath.Base(r.Filepath))}, nil
154+ }
155+ }
156+
67157 listData, err := f.writeHandler.List(f.session, r.Filepath, list, false)
68158 if err != nil {
69159 return nil, err
70160 }
161+ if list {
162+ listData = f.withPending(r.Filepath, listData)
163+ }
71164
72165 // an empty string or exact match from filepath base name is what we want
73166 if !list {
......@@ -92,6 +185,276 @@ func (f *handler) Filelist(r *sftp.Request) (sftp.ListerAt, error) {
92185 return nil, errors.New("unsupported")
93186 }
94187
188+// PosixRename is a rename that replaces an existing file at the target.
189+func (f *handler) PosixRename(r *sftp.Request) error {
190+ return f.rename(r.Filepath, r.Target, true)
191+}
192+
193+// rename moves a file or a directory. Storage has no move, so each file
194+// is written to its new path through the handler, which applies its usual
195+// checks, and the sources are deleted once every write succeeded.
196+func (f *handler) rename(from, to string, replace bool) error {
197+ if from == to {
198+ return nil
199+ }
200+ if strings.HasPrefix(to, from+"/") {
201+ return fmt.Errorf("cannot move %s into itself", from)
202+ }
203+ if moved, err := f.renamePending(from, to, replace); moved || err != nil {
204+ return err
205+ }
206+ if f.isHidden(from) {
207+ return os.ErrNotExist
208+ }
209+ toHidden := f.isHidden(to)
210+
211+ isFile, _, err := f.stored(&utils.FileEntry{Filepath: from}, false)
212+ if err != nil {
213+ return err
214+ }
215+ files, dirs := []string{"."}, []string(nil)
216+ if !isFile {
217+ if files, dirs, err = f.tree(from); err != nil {
218+ return err
219+ }
220+ if len(files)+len(dirs) == 0 {
221+ if ok, err := f.isDir(from); err != nil || !ok {
222+ return errors.Join(os.ErrNotExist, err)
223+ }
224+ }
225+ }
226+ if !replace || !isFile {
227+ exists, _, err := f.stored(&utils.FileEntry{Filepath: to}, false)
228+ if err != nil {
229+ return err
230+ }
231+ exists = exists && !toHidden
232+ belowFiles, belowDirs, err := f.tree(to)
233+ if err != nil {
234+ return err
235+ }
236+ if exists || len(belowFiles)+len(belowDirs) > 0 {
237+ return os.ErrExist
238+ }
239+ }
240+
241+ if !isFile {
242+ for _, d := range append([]string{"."}, dirs...) {
243+ if _, err := f.writeHandler.Write(f.session, &utils.FileEntry{Filepath: filepath.Join(to, d), Mode: os.ModeDir}); err != nil {
244+ return err
245+ }
246+ }
247+ }
248+ var written []string
249+ sums := make([][]byte, len(files))
250+ for i, name := range files {
251+ dst := filepath.Join(to, name)
252+ sum, err := f.copyFile(filepath.Join(from, name), dst)
253+ if err != nil {
254+ for _, w := range written {
255+ _ = f.writeHandler.Delete(f.session, &utils.FileEntry{Filepath: w})
256+ }
257+ return err
258+ }
259+ sums[i] = sum
260+ written = append(written, dst)
261+ }
262+ for i, name := range files {
263+ if err := f.deleteIfUnchanged(filepath.Join(from, name), sums[i]); err != nil {
264+ return err
265+ }
266+ }
267+ if !isFile {
268+ // Children before parents.
269+ slices.SortFunc(dirs, func(a, b string) int { return strings.Count(b, "/") - strings.Count(a, "/") })
270+ for _, d := range append(dirs, ".") {
271+ _ = f.writeHandler.Delete(f.session, &utils.FileEntry{Filepath: filepath.Join(from, d), Mode: os.ModeDir})
272+ }
273+ }
274+ f.mu.Lock()
275+ delete(f.hidden, to)
276+ f.mu.Unlock()
277+ return nil
278+}
279+
280+// renamePending moves a file that is open for writing, which happens when
281+// sshfs delivers the close after a rename. Its contents are stored at the
282+// new path when it closes, and a stored source is kept until then.
283+func (f *handler) renamePending(from, to string, replace bool) (bool, error) {
284+ f.mu.Lock()
285+ p, ok := f.pending[from]
286+ if !ok {
287+ defer f.mu.Unlock()
288+ for name := range f.pending {
289+ if strings.HasPrefix(name, from+"/") {
290+ return false, fmt.Errorf("cannot move %s while %s is open for writing", from, name)
291+ }
292+ }
293+ return false, nil
294+ }
295+ _, toOpen := f.pending[to]
296+ toHidden := f.hidden[to] != nil
297+ stored := p.stored
298+ f.mu.Unlock()
299+ if v, ok := f.writeHandler.(utils.WriteValidator); ok {
300+ if err := v.ValidateWrite(f.session, &utils.FileEntry{Filepath: to, Mode: p.entry.Mode}); err != nil {
301+ return true, err
302+ }
303+ }
304+ if !replace {
305+ exists, _, err := f.stored(&utils.FileEntry{Filepath: to}, false)
306+ if err != nil {
307+ return true, err
308+ }
309+ if (exists && !toHidden) || toOpen {
310+ return true, os.ErrExist
311+ }
312+ }
313+
314+ var sum []byte
315+ if stored {
316+ var err error
317+ if sum, stored, err = f.checksum(from); err != nil {
318+ return true, err
319+ }
320+ }
321+
322+ f.mu.Lock()
323+ defer f.mu.Unlock()
324+ delete(f.pending, from)
325+ delete(f.hidden, to)
326+ f.pending[to] = p
327+ p.path = to
328+ p.dirty = true
329+ if stored {
330+ p.replaces = append(p.replaces, renamedFile{path: from, sum: sum})
331+ f.hidden[from] = p
332+ }
333+ p.stored = false
334+ return true, nil
335+}
336+
337+// copyFile copies a stored file and returns the checksum of what it copied.
338+func (f *handler) copyFile(from, to string) ([]byte, error) {
339+ info, r, err := f.writeHandler.Read(f.session, &utils.FileEntry{Filepath: from})
340+ if err != nil {
341+ return nil, err
342+ }
343+ defer func() { _ = r.Close() }()
344+ h := sha256.New()
345+ msg, err := f.writeHandler.Write(f.session, &utils.FileEntry{
346+ Filepath: to,
347+ Mode: info.Mode().Perm(),
348+ Size: info.Size(),
349+ Mtime: info.ModTime().Unix(),
350+ Atime: info.ModTime().Unix(),
351+ Reader: io.TeeReader(r, h),
352+ })
353+ if err != nil {
354+ return nil, fmt.Errorf("%s: %w", to, err)
355+ }
356+ if msg != "" {
357+ _, _ = fmt.Fprintf(f.session.Stderr(), "%s\r\n", msg)
358+ }
359+ return h.Sum(nil), nil
360+}
361+
362+// checksum hashes the stored file at name.
363+func (f *handler) checksum(name string) ([]byte, bool, error) {
364+ stored, data, err := f.stored(&utils.FileEntry{Filepath: name}, true)
365+ if err != nil || !stored {
366+ return nil, stored, err
367+ }
368+ sum := sha256.Sum256(data)
369+ return sum[:], true, nil
370+}
371+
372+// deleteIfUnchanged deletes a stored file unless another session replaced
373+// it since it had sum. Storage has no conditional delete, so a write
374+// between the check and the delete can still be lost.
375+func (f *handler) deleteIfUnchanged(name string, sum []byte) error {
376+ cur, stored, err := f.checksum(name)
377+ if err != nil || !stored || !bytes.Equal(cur, sum) {
378+ return err
379+ }
380+ return f.writeHandler.Delete(f.session, &utils.FileEntry{Filepath: name})
381+}
382+
383+// tree returns the files and directories below dir, relative to it.
384+func (f *handler) tree(dir string) (files, dirs []string, err error) {
385+ entries, err := f.writeHandler.List(f.session, dir, true, true)
386+ if err != nil {
387+ return nil, nil, err
388+ }
389+ for _, e := range entries {
390+ name := strings.Trim(e.Name(), "/")
391+ switch {
392+ case name == "" || name == ".":
393+ case e.IsDir():
394+ dirs = append(dirs, name)
395+ default:
396+ files = append(files, name)
397+ }
398+ }
399+ return files, dirs, nil
400+}
401+
402+func (f *handler) isDir(name string) (bool, error) {
403+ entries, err := f.writeHandler.List(f.session, filepath.Dir(name), true, false)
404+ if err != nil {
405+ return false, err
406+ }
407+ base := filepath.Base(name)
408+ return slices.ContainsFunc(entries, func(e os.FileInfo) bool {
409+ return e.IsDir() && strings.Trim(e.Name(), "/") == base
410+ }), nil
411+}
412+
413+// withPending adds the files open for writing in dir to its listing and
414+// drops the ones renamed away.
415+func (f *handler) withPending(dir string, listData []os.FileInfo) []os.FileInfo {
416+ f.mu.Lock()
417+ defer f.mu.Unlock()
418+ for name := range f.hidden {
419+ if filepath.Dir(name) == filepath.Clean(dir) {
420+ base := filepath.Base(name)
421+ listData = slices.DeleteFunc(listData, func(info os.FileInfo) bool { return info.Name() == base })
422+ }
423+ }
424+ for name, p := range f.pending {
425+ if filepath.Dir(name) != filepath.Clean(dir) {
426+ continue
427+ }
428+ base := filepath.Base(name)
429+ listData = slices.DeleteFunc(listData, func(info os.FileInfo) bool { return info.Name() == base })
430+ listData = append(listData, p.info(base))
431+ }
432+ return listData
433+}
434+
435+// stored returns the contents of the file at name if storage has it.
436+// Contents are only read when withContents is set.
437+func (f *handler) stored(entry *utils.FileEntry, withContents bool) (bool, []byte, error) {
438+ list, err := f.writeHandler.List(f.session, entry.Filepath, false, false)
439+ if err != nil {
440+ return false, nil, err
441+ }
442+ base := filepath.Base(entry.Filepath)
443+ if !slices.ContainsFunc(list, func(info os.FileInfo) bool { return !info.IsDir() && info.Name() == base }) {
444+ return false, nil, nil
445+ }
446+ if !withContents {
447+ return true, nil, nil
448+ }
449+ _, r, err := f.writeHandler.Read(f.session, entry)
450+ if err != nil {
451+ return false, nil, err
452+ }
453+ defer func() { _ = r.Close() }()
454+ data, err := io.ReadAll(r)
455+ return true, data, err
456+}
457+
95458 func toFileEntry(r *sftp.Request) *utils.FileEntry {
96459 attrs := r.Attributes()
97460 var size int64 = 0
......@@ -115,19 +478,54 @@ func toFileEntry(r *sftp.Request) *utils.FileEntry {
115478 return entry
116479 }
117480
481+// Filewrite opens a file for writing. Nothing is stored until the last
482+// handle to it closes, so a rejected upload leaves storage untouched.
118483 func (f *handler) Filewrite(r *sftp.Request) (io.WriterAt, error) {
119484 entry := toFileEntry(r)
120- entry.Reader = bytes.NewReader([]byte{})
121-
122- _, err := f.writeHandler.Write(f.session, entry)
123- if err != nil {
124- return nil, err
485+ if v, ok := f.writeHandler.(utils.WriteValidator); ok {
486+ if err := v.ValidateWrite(f.session, entry); err != nil {
487+ if errors.Is(err, utils.ErrFileIgnored) {
488+ _, _ = fmt.Fprintf(f.session.Stderr(), "%s\r\n", err)
489+ f.mu.Lock()
490+ f.ignored[r.Filepath]++
491+ f.mu.Unlock()
492+ return ignoredWriter{handler: f, path: r.Filepath}, nil
493+ }
494+ return nil, err
495+ }
125496 }
126497
127- buf := &buffer{}
128- entry.Reader = buf
498+ trunc := r.Pflags().Trunc
499+ f.mu.Lock()
500+ defer f.mu.Unlock()
501+ if p, ok := f.pending[r.Filepath]; ok {
502+ p.handles++
503+ if trunc {
504+ p.buf.truncate(0)
505+ p.dirty = true
506+ }
507+ return &pendingWriter{handler: f, file: p}, nil
508+ }
129509
130- return fakeWrite{fileEntry: entry, buf: buf, handler: f}, nil
510+ // A file renamed away while open is gone, and this one takes its path.
511+ var stored bool
512+ var data []byte
513+ if f.hidden[r.Filepath] != nil {
514+ delete(f.hidden, r.Filepath)
515+ trunc = true
516+ } else {
517+ // Opening without truncating keeps what is stored, as touch does.
518+ var err error
519+ if stored, data, err = f.stored(entry, !trunc); err != nil {
520+ return nil, err
521+ }
522+ }
523+ p := &pendingFile{path: r.Filepath, entry: entry, buf: &buffer{}, modTime: time.Now(), handles: 1, stored: stored, dirty: trunc}
524+ if _, err := p.buf.WriteAt(data, 0); err != nil {
525+ return nil, err
526+ }
527+ f.pending[r.Filepath] = p
528+ return &pendingWriter{handler: f, file: p}, nil
131529 }
132530
133531 func (f *handler) Fileread(r *sftp.Request) (io.ReaderAt, error) {
......@@ -135,6 +533,13 @@ func (f *handler) Fileread(r *sftp.Request) (io.ReaderAt, error) {
135533 return nil, os.ErrInvalid
136534 }
137535
536+ if p := f.pendingFile(r.Filepath); p != nil {
537+ return bytes.NewReader(p.buf.bytes()), nil
538+ }
539+ if f.isHidden(r.Filepath) {
540+ return nil, os.ErrNotExist
541+ }
542+
138543 fileEntry := toFileEntry(r)
139544 _, reader, err := f.writeHandler.Read(f.session, fileEntry)
140545
......@@ -152,6 +557,13 @@ func (f *handlererr) Filecmd(r *sftp.Request) error {
152557 }
153558 return err
154559 }
560+func (f *handlererr) PosixRename(r *sftp.Request) error {
561+ err := f.Handler.PosixRename(r)
562+ if err != nil {
563+ _, _ = fmt.Fprintln(f.Handler.session.Stderr(), err)
564+ }
565+ return err
566+}
155567 func (f *handlererr) Filelist(r *sftp.Request) (sftp.ListerAt, error) {
156568 result, err := f.Handler.Filelist(r)
157569 if err != nil {
M pkg/send/protocols/sftp/sftp.go
+3, -0
......@@ -45,6 +45,9 @@ func Middleware(writeHandler utils.CopyFromClientHandler) pssh.SSHServerMiddlewa
4545 Handler: &handler{
4646 session: session,
4747 writeHandler: writeHandler,
48+ pending: map[string]*pendingFile{},
49+ ignored: map[string]int{},
50+ hidden: map[string]*pendingFile{},
4851 },
4952 }
5053
M pkg/send/protocols/sftp/writer.go
+107, -29
......@@ -1,9 +1,10 @@
11 package sftp
22
33 import (
4+ "bytes"
45 "fmt"
5- "io"
66 "sync"
7+ "time"
78
89 "github.com/picosh/pico/pkg/send/utils"
910 )
......@@ -11,7 +12,6 @@ import (
1112 type buffer struct {
1213 buf []byte
1314 m sync.Mutex
14- off int
1515 }
1616
1717 func (b *buffer) WriteAt(p []byte, pos int64) (n int, err error) {
......@@ -20,58 +20,136 @@ func (b *buffer) WriteAt(p []byte, pos int64) (n int, err error) {
2020 b.m.Lock()
2121 defer b.m.Unlock()
2222 if int64(len(b.buf)) < expLen {
23+ oldLen := len(b.buf)
2324 if int64(cap(b.buf)) < expLen {
2425 newBuf := make([]byte, expLen)
2526 copy(newBuf, b.buf)
2627 b.buf = newBuf
2728 }
29+ // A truncated buffer keeps its old bytes past its length.
2830 b.buf = b.buf[:expLen]
31+ clear(b.buf[oldLen:])
2932 }
3033 copy(b.buf[pos:], p)
3134 return pLen, nil
3235 }
3336
34-func (b *buffer) Read(p []byte) (n int, err error) {
37+func (b *buffer) truncate(size int64) {
3538 b.m.Lock()
3639 defer b.m.Unlock()
37- if len(b.buf) <= b.off {
38- if len(p) == 0 {
39- return 0, nil
40- }
41- return 0, io.EOF
40+ if size < int64(len(b.buf)) {
41+ b.buf = b.buf[:size]
42+ return
4243 }
43- n = copy(p, b.buf[b.off:])
44- b.off += n
45- return n, nil
44+ b.buf = append(b.buf, make([]byte, size-int64(len(b.buf)))...)
4645 }
4746
48-func (b *buffer) Close() error {
49- b.buf = []byte{}
50- return nil
47+func (b *buffer) bytes() []byte {
48+ b.m.Lock()
49+ defer b.m.Unlock()
50+ return bytes.Clone(b.buf)
5151 }
5252
53-type fakeWrite struct {
54- fileEntry *utils.FileEntry
55- handler *handler
56- buf *buffer
53+// pendingFile is a file open for writing. Editors over sshfs stat a file
54+// right after creating it, so it is listed from here until the last handle
55+// closes and its contents are written.
56+type pendingFile struct {
57+ path string
58+ entry *utils.FileEntry
59+ buf *buffer
60+ modTime time.Time
61+ handles int
62+ // stored is set when storage had the file at open. Unless a handle
63+ // truncated or wrote to it, it's left as it is.
64+ stored bool
65+ dirty bool
66+ // replaces holds stored files this one was renamed from while open.
67+ // They are deleted once it is stored at its new path.
68+ replaces []renamedFile
5769 }
5870
59-func (f fakeWrite) WriteAt(p []byte, off int64) (int, error) {
60- return f.buf.WriteAt(p, off)
71+// renamedFile is a stored file renamed away, with the checksum it had then.
72+// Another session may replace it before it's deleted.
73+type renamedFile struct {
74+ path string
75+ sum []byte
6176 }
6277
63-func (f fakeWrite) Close() error {
64- msg, err := f.handler.writeHandler.Write(f.handler.session, f.fileEntry)
65- if err != nil {
66- errMsg := fmt.Sprintf("%s\r\n", err.Error())
67- _, err = f.handler.session.Stderr().Write([]byte(errMsg))
78+func (p *pendingFile) info(name string) *utils.VirtualFile {
79+ return &utils.VirtualFile{
80+ FName: name,
81+ FSize: int64(len(p.buf.bytes())),
82+ FModTime: p.modTime,
6883 }
69- if msg != "" {
70- nMsg := fmt.Sprintf("%s\r\n", msg)
71- _, err = f.handler.session.Stderr().Write([]byte(nMsg))
84+}
85+
86+type pendingWriter struct {
87+ handler *handler
88+ file *pendingFile
89+}
90+
91+func (w *pendingWriter) WriteAt(p []byte, off int64) (int, error) {
92+ w.handler.mu.Lock()
93+ w.file.dirty = true
94+ w.handler.mu.Unlock()
95+ return w.file.buf.WriteAt(p, off)
96+}
97+
98+func (w *pendingWriter) Close() error {
99+ h, f := w.handler, w.file
100+ h.mu.Lock()
101+ f.handles--
102+ // A file removed while open is dropped.
103+ commit := f.handles == 0 && h.pending[f.path] == f
104+ if commit {
105+ delete(h.pending, f.path)
106+ }
107+ name := f.path
108+ h.mu.Unlock()
109+ if !commit || (f.stored && !f.dirty) {
110+ return nil
72111 }
112+
113+ data := f.buf.bytes()
114+ entry := *f.entry
115+ entry.Filepath = name
116+ entry.Reader = bytes.NewReader(data)
117+ entry.Size = int64(len(data))
118+ msg, err := h.writeHandler.Write(h.session, &entry)
119+ h.mu.Lock()
120+ replaced := h.release(f)
121+ h.mu.Unlock()
73122 if err != nil {
123+ _, _ = fmt.Fprintf(h.session.Stderr(), "%s\r\n", err)
74124 return err
75125 }
76- return f.buf.Close()
126+ if msg != "" {
127+ _, _ = fmt.Fprintf(h.session.Stderr(), "%s\r\n", msg)
128+ }
129+ for _, old := range replaced {
130+ if err := h.deleteIfUnchanged(old.path, old.sum); err != nil {
131+ _, _ = fmt.Fprintf(h.session.Stderr(), "%s\r\n", err)
132+ return err
133+ }
134+ }
135+ return nil
136+}
137+
138+// ignoredWriter accepts a file the handler ignores by policy, so a
139+// transfer moves on to the next one.
140+type ignoredWriter struct {
141+ handler *handler
142+ path string
143+}
144+
145+func (ignoredWriter) WriteAt(p []byte, _ int64) (int, error) { return len(p), nil }
146+
147+func (w ignoredWriter) Close() error {
148+ w.handler.mu.Lock()
149+ defer w.handler.mu.Unlock()
150+ w.handler.ignored[w.path]--
151+ if w.handler.ignored[w.path] <= 0 {
152+ delete(w.handler.ignored, w.path)
153+ }
154+ return nil
77155 }
M pkg/send/utils/utils.go
+10, -0
......@@ -2,6 +2,7 @@ package utils
22
33 import (
44 "encoding/base64"
5+ "errors"
56 "fmt"
67 "io"
78 "io/fs"
......@@ -16,6 +17,9 @@ import (
1617 // NULL is an array with a single NULL byte.
1718 var NULL = []byte{'\x00'}
1819
20+// ErrFileIgnored marks a file deliberately excluded by upload policy.
21+var ErrFileIgnored = errors.New("file ignored")
22+
1923 // FileEntry is an Entry that reads from a Reader, defining a file and
2024 // its contents.
2125 type FileEntry struct {
......@@ -66,6 +70,12 @@ type CopyFromClientHandler interface {
6670 Validate(*pssh.SSHServerConnSession) error
6771 }
6872
73+// WriteValidator is implemented by handlers that can check a write before
74+// its contents arrive, so SFTP can refuse a file when it is opened.
75+type WriteValidator interface {
76+ ValidateWrite(*pssh.SSHServerConnSession, *FileEntry) error
77+}
78+
6979 func KeyText(session *pssh.SSHServerConnSession) (string, error) {
7080 if session.PublicKey() == nil {
7181 return "", fmt.Errorf("session doesn't have public key")