Antonio Mika
·
2026-10-11
1package pgs
2
3import (
4 "bytes"
5 "crypto/rand"
6 "encoding/pem"
7 "errors"
8 "fmt"
9 "io"
10 "log/slog"
11 "net"
12 "os"
13 "os/exec"
14 "path/filepath"
15 "regexp"
16 "strconv"
17 "strings"
18 "testing"
19 "time"
20
21 pgsdb "github.com/picosh/pico/pkg/apps/pgs/db"
22 "github.com/picosh/pico/pkg/db"
23 "github.com/picosh/pico/pkg/pssh"
24 "github.com/picosh/pico/pkg/shared"
25 "github.com/picosh/pico/pkg/storage"
26 "github.com/picosh/pico/pkg/storage/storagetest"
27)
28
29// rsyncFileMax lifts pgs's per-file cap so the test can move files larger
30// than the default allows.
31const rsyncFileMax = 24 << 20
32
33// rsyncEnv is a pgs SSH server backed by one storage implementation, and a
34// real rsync client set up to talk to it.
35type rsyncEnv struct {
36 t *testing.T
37 st storage.StorageServe
38 bucket storage.Bucket
39 rsh string
40 feature *db.FeatureFlag
41 // sshArgs are the client options scp and sftp need to reach the server.
42 sshArgs []string
43 addr string
44 user UserSSH
45}
46
47func newRsyncEnv(t *testing.T, st storage.StorageServe) *rsyncEnv {
48 if _, err := exec.LookPath("rsync"); err != nil {
49 t.Skip("rsync not installed")
50 }
51 logger := slog.New(slog.DiscardHandler)
52 dbpool := pgsdb.NewDBMemory(logger)
53 dbpool.SetupTestData()
54 dbpool.Feature.Data.FileMax = rsyncFileMax
55 dbpool.Feature.Data.StorageMax = 256 << 20
56 bucket := storagetest.Bucket(t, st, shared.GetAssetBucketName(dbpool.Users[0].ID))
57
58 t.Setenv("PGS_SSH_PORT", "0")
59 cfg := NewConfig(logger, dbpool, st, discardPubsub{})
60 cfg.Domain = "pgs.sh"
61 cfg.WebProtocol = "https"
62 done := make(chan error)
63 readyCh := make(chan *pssh.SSHServer)
64 go StartSshServerForTesting(cfg, done, readyCh)
65 t.Cleanup(func() { close(done) })
66
67 server := <-readyCh
68 if server == nil {
69 t.Fatal("failed to create ssh server")
70 }
71 var addr string
72 for range 100 {
73 server.Mu.Lock()
74 if server.Listener != nil {
75 addr = server.Listener.Addr().String()
76 }
77 server.Mu.Unlock()
78 if addr != "" {
79 break
80 }
81 time.Sleep(10 * time.Millisecond)
82 }
83 if addr == "" {
84 t.Fatal("server listener not ready")
85 }
86
87 user := GenerateUser()
88 dbpool.Pubkeys = append(dbpool.Pubkeys, &db.PublicKey{
89 ID: "rsync-pubkey",
90 UserID: dbpool.Users[0].ID,
91 Key: shared.KeyForKeyText(user.signer.PublicKey()),
92 })
93 keyFile := filepath.Join(t.TempDir(), "id_ed25519")
94 block := &pem.Block{Type: "OPENSSH PRIVATE KEY", Bytes: user.privateKey}
95 if err := os.WriteFile(keyFile, pem.EncodeToMemory(block), 0o600); err != nil {
96 t.Fatal(err)
97 }
98 _, port, _ := net.SplitHostPort(addr)
99 rsh := fmt.Sprintf(
100 "ssh -p %s -i %s -o IdentitiesOnly=yes -o StrictHostKeyChecking=no -o UserKnownHostsFile=/dev/null -o LogLevel=ERROR",
101 port, keyFile,
102 )
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}
105}
106
107// discardPubsub drops cache purges, which pgs sends from goroutines that can
108// outlive the test.
109type discardPubsub struct{}
110
111func (discardPubsub) Read([]byte) (int, error) { return 0, io.EOF }
112func (discardPubsub) Write(b []byte) (int, error) { return len(b), nil }
113func (discardPubsub) Close() error { return nil }
114
115type rsyncStats struct {
116 output string
117 transferred int
118 deleted int
119 literal int
120 matched int
121}
122
123var statPatterns = map[string]*regexp.Regexp{
124 "transferred": regexp.MustCompile(`Number of regular files transferred: ([\d,]+)`),
125 "deleted": regexp.MustCompile(`Number of deleted files: ([\d,]+)`),
126 "literal": regexp.MustCompile(`Literal data: ([\d,]+)`),
127 "matched": regexp.MustCompile(`Matched data: ([\d,]+)`),
128}
129
130func (e *rsyncEnv) rsync(args ...string) rsyncStats {
131 e.t.Helper()
132 return e.rsyncExit(0, args...)
133}
134
135func (e *rsyncEnv) rsyncExit(want int, args ...string) rsyncStats {
136 e.t.Helper()
137 args = append([]string{"-e", e.rsh, "--stats"}, args...)
138 out, err := exec.Command("rsync", args...).CombinedOutput()
139 code := 0
140 if err != nil {
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)
149 }
150 stat := func(name string) int {
151 m := statPatterns[name].FindSubmatch(out)
152 if m == nil {
153 e.t.Fatalf("no %s in rsync output:\n%s", name, out)
154 }
155 n, _ := strconv.Atoi(strings.ReplaceAll(string(m[1]), ",", ""))
156 return n
157 }
158 return rsyncStats{
159 output: string(out),
160 transferred: stat("transferred"),
161 deleted: stat("deleted"),
162 literal: stat("literal"),
163 matched: stat("matched"),
164 }
165}
166
167func (e *rsyncEnv) stored(name string) ([]byte, time.Time, bool) {
168 e.t.Helper()
169 r, info, err := e.st.GetObject(e.bucket, name)
170 if err != nil {
171 return nil, time.Time{}, false
172 }
173 defer func() { _ = r.Close() }()
174 data, err := io.ReadAll(r)
175 if err != nil {
176 e.t.Fatalf("read %s: %v", name, err)
177 }
178 return data, info.LastModified, true
179}
180
181// tree is a local directory of files with fixed contents and mtimes.
182type tree struct {
183 t *testing.T
184 dir string
185}
186
187func (tr tree) write(name string, data []byte, mtime time.Time) {
188 tr.t.Helper()
189 p := filepath.Join(tr.dir, name)
190 if err := os.MkdirAll(filepath.Dir(p), 0o755); err != nil {
191 tr.t.Fatal(err)
192 }
193 if err := os.WriteFile(p, data, 0o644); err != nil {
194 tr.t.Fatal(err)
195 }
196 tr.touch(name, mtime)
197}
198
199func (tr tree) touch(name string, mtime time.Time) {
200 tr.t.Helper()
201 if err := os.Chtimes(filepath.Join(tr.dir, name), mtime, mtime); err != nil {
202 tr.t.Fatal(err)
203 }
204}
205
206func (tr tree) read(name string) ([]byte, time.Time) {
207 tr.t.Helper()
208 p := filepath.Join(tr.dir, name)
209 data, err := os.ReadFile(p)
210 if err != nil {
211 tr.t.Fatal(err)
212 }
213 info, err := os.Stat(p)
214 if err != nil {
215 tr.t.Fatal(err)
216 }
217 return data, info.ModTime()
218}
219
220func randomBytes(n int) []byte {
221 b := make([]byte, n)
222 _, _ = rand.Read(b)
223 return b
224}
225
226func rsyncStorageBackends() map[string]func(t *testing.T) storage.StorageServe {
227 return map[string]func(t *testing.T) storage.StorageServe{
228 "fs": func(t *testing.T) storage.StorageServe {
229 st, err := storage.NewStorageFS(slog.New(slog.DiscardHandler), t.TempDir())
230 if err != nil {
231 t.Fatal(err)
232 }
233 return st
234 },
235 "memory": func(t *testing.T) storage.StorageServe {
236 st, err := storage.NewStorageMemory(map[string]map[string]string{})
237 if err != nil {
238 t.Fatal(err)
239 }
240 return st
241 },
242 }
243}
244
245// TestRsyncStorage runs a real rsync client against pgs on each storage
246// backend and checks that quick checks, deltas, checksums and deletes are
247// driven by what storage holds.
248func TestRsyncStorage(t *testing.T) {
249 for name, newStorage := range rsyncStorageBackends() {
250 t.Run(name, func(t *testing.T) {
251 env := newRsyncEnv(t, newStorage(t))
252 src := tree{t, t.TempDir()}
253 remote := "localhost:/site"
254
255 t0 := time.Unix(1_600_000_000, 0)
256 t1 := time.Unix(1_650_000_000, 0)
257 big := randomBytes(20 << 20)
258 src.write("index.html", []byte("<html>index</html>"), t0)
259 src.write("big.bin", big, t0)
260 src.write("css/site.css", []byte("body{}"), t0)
261 src.write("css/deep/x.css", []byte("a{}"), t1)
262 src.write("img/logo.svg", []byte("<svg/>"), t1)
263 src.write("keep.txt", []byte("excluded from deletes"), t0)
264
265 s := env.rsync("-rtz", src.dir+"/", remote)
266 if s.transferred != 6 {
267 t.Fatalf("initial upload transferred %d files\n%s", s.transferred, s.output)
268 }
269 for name, mtime := range map[string]time.Time{"index.html": t0, "big.bin": t0, "css/deep/x.css": t1} {
270 data, stored, ok := env.stored("/site/" + name)
271 local, _ := src.read(name)
272 if !ok || !bytes.Equal(data, local) {
273 t.Fatalf("%s not stored correctly\n%s", name, s.output)
274 }
275 if !stored.Equal(mtime) {
276 t.Errorf("%s stored with mtime %v, want %v", name, stored, mtime)
277 }
278 }
279
280 t.Run("unchanged upload sends nothing", func(t *testing.T) {
281 s := env.rsync("-rtz", src.dir+"/", remote)
282 if s.transferred != 0 || s.literal != 0 {
283 t.Fatalf("transferred %d files, %d literal bytes\n%s", s.transferred, s.literal, s.output)
284 }
285 })
286
287 t.Run("changed file is sent as a delta", func(t *testing.T) {
288 copy(big[1<<20:], "a small change in the middle of the file")
289 src.write("big.bin", big, t1)
290 s := env.rsync("-rtz", src.dir+"/", remote)
291 if s.transferred != 1 || s.literal > 16<<10 || s.matched < len(big)-16<<10 {
292 t.Fatalf("transferred %d, literal %d, matched %d\n%s", s.transferred, s.literal, s.matched, s.output)
293 }
294 data, mtime, _ := env.stored("/site/big.bin")
295 if !bytes.Equal(data, big) || !mtime.Equal(t1) {
296 t.Fatalf("stored big.bin wrong after delta (mtime %v)", mtime)
297 }
298 })
299
300 t.Run("grown and shrunk files", func(t *testing.T) {
301 src.write("index.html", []byte("<html>index, now with more content</html>"), t1)
302 src.write("css/site.css", []byte("b{}"), t1)
303 env.rsync("-rtz", src.dir+"/", remote)
304 for _, name := range []string{"index.html", "css/site.css"} {
305 data, _, _ := env.stored("/site/" + name)
306 local, _ := src.read(name)
307 if !bytes.Equal(data, local) {
308 t.Errorf("%s stored as %q, want %q", name, data, local)
309 }
310 }
311 })
312
313 t.Run("checksum mode ignores mtimes", func(t *testing.T) {
314 src.touch("css/deep/x.css", t0)
315 s := env.rsync("-rcz", src.dir+"/", remote)
316 if s.transferred != 0 {
317 t.Fatalf("-c transferred %d files with unchanged contents\n%s", s.transferred, s.output)
318 }
319 s = env.rsync("-rtz", src.dir+"/", remote)
320 if s.transferred != 1 || s.literal != 0 {
321 t.Fatalf("mtime-only change transferred %d files, %d literal bytes\n%s", s.transferred, s.literal, s.output)
322 }
323 if _, mtime, _ := env.stored("/site/css/deep/x.css"); !mtime.Equal(t0) {
324 t.Fatalf("mtime-only change stored mtime %v, want %v", mtime, t0)
325 }
326 })
327
328 t.Run("oversized file is rejected", func(t *testing.T) {
329 src.write("huge.bin", randomBytes(rsyncFileMax+1), t0)
330 s := env.rsyncExit(23, "-rt", src.dir+"/", remote)
331 if !strings.Contains(s.output, "huge.bin") {
332 t.Fatalf("no warning about huge.bin\n%s", s.output)
333 }
334 if _, _, ok := env.stored("/site/huge.bin"); ok {
335 t.Fatal("rejected file was stored")
336 }
337 if err := os.Remove(filepath.Join(src.dir, "huge.bin")); err != nil {
338 t.Fatal(err)
339 }
340 })
341
342 t.Run("delete removes what the source lacks", func(t *testing.T) {
343 if err := os.RemoveAll(filepath.Join(src.dir, "css/deep")); err != nil {
344 t.Fatal(err)
345 }
346 if err := os.Remove(filepath.Join(src.dir, "img/logo.svg")); err != nil {
347 t.Fatal(err)
348 }
349 if err := os.Remove(filepath.Join(src.dir, "keep.txt")); err != nil {
350 t.Fatal(err)
351 }
352 s := env.rsync("-rt", "--delete", "--exclude=keep.txt", src.dir+"/", remote)
353 if s.deleted < 2 {
354 t.Fatalf("deleted %d files\n%s", s.deleted, s.output)
355 }
356 for name, want := range map[string]bool{
357 "css/deep/x.css": false,
358 "img/logo.svg": false,
359 "keep.txt": true,
360 "css/site.css": true,
361 "big.bin": true,
362 } {
363 if _, _, ok := env.stored("/site/" + name); ok != want {
364 t.Errorf("%s stored=%v after delete, want %v", name, ok, want)
365 }
366 }
367 })
368
369 dst := tree{t, t.TempDir()}
370
371 t.Run("download", func(t *testing.T) {
372 s := env.rsync("-rtz", remote+"/", dst.dir+"/")
373 if s.transferred != 4 {
374 t.Fatalf("download transferred %d files\n%s", s.transferred, s.output)
375 }
376 for _, name := range []string{"index.html", "big.bin", "css/site.css", "keep.txt"} {
377 got, mtime := dst.read(name)
378 want, storedMtime, _ := env.stored("/site/" + name)
379 if !bytes.Equal(got, want) {
380 t.Errorf("downloaded %s differs from storage", name)
381 }
382 if !mtime.Equal(storedMtime) {
383 t.Errorf("downloaded %s has mtime %v, storage has %v", name, mtime, storedMtime)
384 }
385 }
386 })
387
388 t.Run("unchanged download sends nothing", func(t *testing.T) {
389 s := env.rsync("-rtz", remote+"/", dst.dir+"/")
390 if s.transferred != 0 || s.literal != 0 {
391 t.Fatalf("transferred %d files, %d literal bytes\n%s", s.transferred, s.literal, s.output)
392 }
393 })
394
395 t.Run("download delta against a local copy", func(t *testing.T) {
396 local, _ := dst.read("big.bin")
397 copy(local[2<<20:], "local edit")
398 dst.write("big.bin", local, t0)
399 s := env.rsync("-rtz", remote+"/", dst.dir+"/")
400 if s.transferred != 1 || s.literal > 16<<10 || s.matched < len(big)-16<<10 {
401 t.Fatalf("transferred %d, literal %d, matched %d\n%s", s.transferred, s.literal, s.matched, s.output)
402 }
403 if got, _ := dst.read("big.bin"); !bytes.Equal(got, big) {
404 t.Fatal("big.bin wrong after delta download")
405 }
406 })
407
408 t.Run("download single file", func(t *testing.T) {
409 one := tree{t, t.TempDir()}
410 env.rsync("-t", remote+"/css/site.css", one.dir+"/")
411 if got, _ := one.read("site.css"); string(got) != "b{}" {
412 t.Fatalf("site.css is %q", got)
413 }
414 })
415 })
416 }
417}
418
419func 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
473// TestRsyncStorageQuota checks that shrinking and deleting files frees quota
474// for files later in the same run.
475func TestRsyncStorageQuota(t *testing.T) {
476 for name, newStorage := range rsyncStorageBackends() {
477 t.Run(name, func(t *testing.T) {
478 env := newRsyncEnv(t, newStorage(t))
479 env.feature.Data.StorageMax = 3 << 20
480 src := tree{t, t.TempDir()}
481 remote := "localhost:/site"
482 t0 := time.Unix(1_600_000_000, 0)
483
484 src.write("a.bin", randomBytes(2<<20), t0)
485 env.rsync("-rt", src.dir+"/", remote)
486
487 // a.bin is sent before b.bin, which only fits once a.bin's
488 // shrink is counted.
489 src.write("a.bin", randomBytes(512<<10), t0)
490 src.write("b.bin", randomBytes(2<<20), t0)
491 if s := env.rsync("-rt", src.dir+"/", remote); s.transferred != 2 {
492 t.Fatalf("transferred %d files, want 2\n%s", s.transferred, s.output)
493 }
494
495 if err := os.Remove(filepath.Join(src.dir, "a.bin")); err != nil {
496 t.Fatal(err)
497 }
498 src.write("c.bin", randomBytes(768<<10), t0)
499 if s := env.rsync("-rt", "--delete-before", src.dir+"/", remote); s.deleted != 1 || s.transferred != 1 {
500 t.Fatalf("deleted %d and transferred %d files, want 1 and 1\n%s", s.deleted, s.transferred, s.output)
501 }
502 if _, _, ok := env.stored("/site/c.bin"); !ok {
503 t.Fatal("c.bin was not stored")
504 }
505 })
506 }
507}
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.
512func 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}