tinyrelayfile
tinyrelay/internal/blob/blob_test.go/
internal/blob/blob_test.go raw
1package blob
2
3import (
4 "context"
5 "crypto/sha256"
6 "encoding/hex"
7 "errors"
8 "io"
9 "net"
10 "net/http"
11 "net/http/httptest"
12 "os"
13 "path/filepath"
14 "strings"
15 "sync"
16 "testing"
17 "time"
18
19 "github.com/FelineStateMachine/tinyrelay/internal/storage"
20)
21
22func TestUploadDeduplicatesAndServesByHash(t *testing.T) {
23 service := testService(t)
24 body := []byte("hello blob")
25 hash := sha256.Sum256(body)
26 digest := hex.EncodeToString(hash[:])
27 req := httptest.NewRequest(http.MethodPut, "https://relay.test/upload", strings.NewReader(string(body)))
28 req.Header.Set("Content-Type", "text/plain")
29 first := service.Handler().ServeHTTP
30 response := record(first, req)
31 if response.Code != http.StatusOK {
32 t.Fatalf("upload status = %d, want 200 (%s)", response.Code, response.Body.String())
33 }
34 if !strings.Contains(response.Body.String(), digest) {
35 t.Fatalf("upload descriptor lacks hash: %s", response.Body.String())
36 }
37 req = httptest.NewRequest(http.MethodPut, "https://relay.test/upload", strings.NewReader(string(body)))
38 req.Header.Set("Content-Type", "text/plain")
39 response = record(first, req)
40 if response.Code != http.StatusOK {
41 t.Fatalf("duplicate upload status = %d, want 200", response.Code)
42 }
43 response = record(first, httptest.NewRequest(http.MethodGet, "https://relay.test/"+digest+".txt", nil))
44 if response.Code != http.StatusOK || response.Body.String() != string(body) {
45 t.Fatalf("download = %d %q", response.Code, response.Body.String())
46 }
47 if response.Header().Get("ETag") != `"`+digest+`"` {
48 t.Fatalf("etag = %q", response.Header().Get("ETag"))
49 }
50}
51
52func TestUploadEnforcesFileAndUploaderLimits(t *testing.T) {
53 service := testService(t)
54 service.config.Limits = func() Limits { return Limits{MaxFileBytes: 4, UserStorageBytes: 6} }
55 if _, err := service.Put(context.Background(), PutOptions{Reader: strings.NewReader("12345"), Type: "text/plain", Uploader: testUploader}); !errors.Is(err, ErrFileTooLarge) {
56 t.Fatalf("oversized upload error = %v", err)
57 }
58 if _, err := service.Put(context.Background(), PutOptions{Reader: strings.NewReader("1234"), Type: "text/plain", Uploader: testUploader}); err != nil {
59 t.Fatalf("first limited upload: %v", err)
60 }
61 if _, err := service.Put(context.Background(), PutOptions{Reader: strings.NewReader("5678"), Type: "text/plain", Uploader: testUploader}); !errors.Is(err, ErrQuotaExceeded) {
62 t.Fatalf("quota upload error = %v", err)
63 }
64 // A deduplicated claim does not consume additional physical storage.
65 if _, err := service.Put(context.Background(), PutOptions{Reader: strings.NewReader("1234"), Type: "text/plain", Uploader: strings.Repeat("b", 64)}); err != nil {
66 t.Fatalf("deduplicated upload: %v", err)
67 }
68 second := strings.Repeat("b", 64)
69 if _, err := service.Put(context.Background(), PutOptions{Reader: strings.NewReader("abc"), Type: "text/plain", Uploader: strings.Repeat("c", 64)}); err != nil {
70 t.Fatal(err)
71 }
72 if _, err := service.Put(context.Background(), PutOptions{Reader: strings.NewReader("56"), Type: "text/plain", Uploader: second}); err != nil {
73 t.Fatalf("second uploader upload: %v", err)
74 }
75 if _, err := service.Put(context.Background(), PutOptions{Reader: strings.NewReader("xyz"), Type: "text/plain", Uploader: second}); !errors.Is(err, ErrQuotaExceeded) {
76 t.Fatalf("second uploader remaining quota error = %v", err)
77 }
78 if _, err := service.Put(context.Background(), PutOptions{Reader: strings.NewReader("abc"), Type: "text/plain", Uploader: second}); !errors.Is(err, ErrQuotaExceeded) {
79 t.Fatalf("existing blob claim bypassed remaining quota: %v", err)
80 }
81 digest := sha256.Sum256([]byte("56"))
82 sha := hex.EncodeToString(digest[:])
83 if err := service.DeleteForUploader(context.Background(), sha, second); err != nil {
84 t.Fatalf("remove second uploader claim: %v", err)
85 }
86 if _, err := service.Put(context.Background(), PutOptions{Reader: strings.NewReader("ab"), Type: "text/plain", Uploader: second}); err != nil {
87 t.Fatalf("quota after claim deletion: %v", err)
88 }
89}
90
91func TestConcurrentUploadsRespectUploaderQuota(t *testing.T) {
92 service := testService(t)
93 service.config.Limits = func() Limits { return Limits{UserStorageBytes: 4} }
94 var wg sync.WaitGroup
95 errs := make(chan error, 2)
96 for _, body := range []string{"aaaa", "bbbb"} {
97 wg.Add(1)
98 go func(body string) {
99 defer wg.Done()
100 _, err := service.Put(context.Background(), PutOptions{Reader: strings.NewReader(body), Type: "text/plain", Uploader: testUploader})
101 errs <- err
102 }(body)
103 }
104 wg.Wait()
105 close(errs)
106 var success, quota int
107 for err := range errs {
108 if err == nil {
109 success++
110 } else if errors.Is(err, ErrQuotaExceeded) {
111 quota++
112 }
113 }
114 if success != 1 || quota != 1 {
115 t.Fatalf("concurrent quota results success=%d quota=%d", success, quota)
116 }
117}
118
119func TestQuotaOnlyLimitReportsQuotaBeforeHashMismatch(t *testing.T) {
120 service := testService(t)
121 service.config.Limits = func() Limits { return Limits{UserStorageBytes: 4} }
122 _, err := service.Put(context.Background(), PutOptions{Reader: strings.NewReader("12345"), Type: "text/plain", Uploader: testUploader, Hash: strings.Repeat("a", 64)})
123 if !errors.Is(err, ErrQuotaExceeded) {
124 t.Fatalf("quota-only oversized upload error = %v", err)
125 }
126}
127
128func TestSlowUploadDoesNotHoldQuotaLockWhileReading(t *testing.T) {
129 service := testService(t)
130 reader, writer := io.Pipe()
131 done := make(chan error, 1)
132 go func() {
133 _, err := service.Put(context.Background(), PutOptions{Reader: reader, Type: "text/plain", Uploader: testUploader})
134 done <- err
135 }()
136 select {
137 case <-done:
138 t.Fatal("stalled upload returned before receiving data")
139 case <-time.After(25 * time.Millisecond):
140 }
141 second := strings.Repeat("b", 64)
142 finished := make(chan error, 1)
143 go func() {
144 _, err := service.Put(context.Background(), PutOptions{Reader: strings.NewReader("fast"), Type: "text/plain", Uploader: second})
145 finished <- err
146 }()
147 select {
148 case err := <-finished:
149 if err != nil {
150 t.Fatalf("second upload: %v", err)
151 }
152 case <-time.After(time.Second):
153 t.Fatal("second upload blocked behind stalled stream")
154 }
155 _, _ = writer.Write([]byte("slow"))
156 _ = writer.Close()
157 if err := <-done; err != nil {
158 t.Fatalf("stalled upload completion: %v", err)
159 }
160}
161
162func TestPrivateBlobRequiresReadAuthorization(t *testing.T) {
163 service := testService(t)
164 service.config.CanRead = func(context.Context, string, []string) bool { return false }
165 service.config.Authorize = func(*http.Request, Action) (string, error) { return "", os.ErrPermission }
166 response := record(service.Handler().ServeHTTP, httptest.NewRequest(http.MethodGet, "https://relay.test/"+strings.Repeat("a", 64), nil))
167 if response.Code != http.StatusUnauthorized {
168 t.Fatalf("private blob status = %d, want 401", response.Code)
169 }
170}
171
172func TestMirrorURLRejectsPrivateAndNonHTTPS(t *testing.T) {
173 for _, raw := range []string{"http://example.com/a", "https://127.0.0.1/a", "https://[::1]/a", "https://localhost/a"} {
174 if err := validateMirrorURL(raw, func(context.Context, string) ([]net.IP, error) { return nil, nil }); err == nil {
175 t.Errorf("validateMirrorURL(%q) succeeded", raw)
176 }
177 }
178}
179
180func TestMirrorBlossomURIFallsBackAndVerifiesHash(t *testing.T) {
181 service := testService(t)
182 body := []byte("fallback body")
183 hash := sha256.Sum256(body)
184 digest := hex.EncodeToString(hash[:])
185 first := "https://first.example/" + digest
186 author := strings.Repeat("c", 64)
187 service.config.ResolveIP = func(context.Context, string) ([]net.IP, error) { return []net.IP{net.ParseIP("8.8.8.8")}, nil }
188 service.resolve = service.config.ResolveIP
189 service.config.ResolveServers = func(_ context.Context, pubkey string) ([]string, error) {
190 if pubkey != author {
191 t.Fatalf("fallback resolved uploader %q, want URI author", pubkey)
192 }
193 return []string{"https://second.example"}, nil
194 }
195 service.config.HTTPClient = &http.Client{Transport: roundTripFunc(func(req *http.Request) (*http.Response, error) {
196 if req.URL.String() == first {
197 return &http.Response{StatusCode: http.StatusBadGateway, Body: http.NoBody, Header: make(http.Header), Request: req}, nil
198 }
199 return &http.Response{StatusCode: http.StatusOK, Body: io.NopCloser(strings.NewReader(string(body))), Header: http.Header{"Content-Type": []string{"text/plain"}}, Request: req}, nil
200 })}
201 uri := "blossom:" + digest + "?xs=first.example&as=" + author
202 request := httptest.NewRequest(http.MethodPut, "https://relay.test/mirror", strings.NewReader(`{"url":"`+uri+`"}`))
203 response := record(service.Handler().ServeHTTP, request)
204 if response.Code != http.StatusCreated || !strings.Contains(response.Body.String(), digest) {
205 t.Fatalf("fallback mirror = %d %s", response.Code, response.Body.String())
206 }
207}
208
209func TestMirrorBlossomURIRejectsWrongHash(t *testing.T) {
210 service := testService(t)
211 wanted := strings.Repeat("a", 64)
212 service.config.ResolveIP = func(context.Context, string) ([]net.IP, error) { return []net.IP{net.ParseIP("8.8.8.8")}, nil }
213 service.resolve = service.config.ResolveIP
214 service.config.ResolveServers = func(context.Context, string) ([]string, error) { return []string{"https://origin.example"}, nil }
215 service.config.HTTPClient = &http.Client{Transport: roundTripFunc(func(req *http.Request) (*http.Response, error) {
216 return &http.Response{StatusCode: http.StatusOK, Body: io.NopCloser(strings.NewReader("wrong")), Header: make(http.Header), Request: req}, nil
217 })}
218 request := httptest.NewRequest(http.MethodPut, "https://relay.test/mirror", strings.NewReader(`{"url":"blossom:`+wanted+`"}`))
219 response := record(service.Handler().ServeHTTP, request)
220 if response.Code != http.StatusConflict {
221 t.Fatalf("wrong hash status = %d, want 409", response.Code)
222 }
223 if _, _, err := service.Get(context.Background(), wanted); err == nil {
224 t.Fatal("wrong hash was stored")
225 }
226}
227
228type roundTripFunc func(*http.Request) (*http.Response, error)
229
230func (f roundTripFunc) RoundTrip(r *http.Request) (*http.Response, error) { return f(r) }
231
232func TestDeleteTombstonePreventsOrphanResurrection(t *testing.T) {
233 service := testService(t)
234 body := []byte("delete me")
235 digest := sha256.Sum256(body)
236 sha := hex.EncodeToString(digest[:])
237 if _, err := service.Put(context.Background(), PutOptions{Reader: strings.NewReader(string(body)), Type: "text/plain", Uploader: testUploader}); err != nil {
238 t.Fatal(err)
239 }
240 if err := service.Delete(context.Background(), sha); err != nil {
241 t.Fatal(err)
242 }
243 if err := os.WriteFile(filepath.Join(service.root, sha), body, 0o600); err != nil {
244 t.Fatal(err)
245 }
246 if _, err := New(context.Background(), service.config); err != nil {
247 t.Fatal(err)
248 }
249 if _, _, err := service.Get(context.Background(), sha); err == nil {
250 t.Fatal("deleted orphan was resurrected")
251 }
252}
253
254func TestBlobClaimsPreserveDeduplicatedOwnershipAndCursor(t *testing.T) {
255 service := testService(t)
256 body := "shared blob"
257 first, err := service.Put(context.Background(), PutOptions{Reader: strings.NewReader(body), Type: "text/plain", Uploader: testUploader})
258 if err != nil {
259 t.Fatal(err)
260 }
261 secondUploader := strings.Repeat("b", 64)
262 second, err := service.Put(context.Background(), PutOptions{Reader: strings.NewReader(body), Type: "text/plain", Uploader: secondUploader})
263 if err != nil {
264 t.Fatal(err)
265 }
266 if first.SHA256 != second.SHA256 {
267 t.Fatalf("deduplicated hashes differ: %q and %q", first.SHA256, second.SHA256)
268 }
269 if _, err := service.Put(context.Background(), PutOptions{Reader: strings.NewReader("another blob"), Type: "text/plain", Uploader: testUploader}); err != nil {
270 t.Fatal(err)
271 }
272 page, next, err := service.ListPage(context.Background(), testUploader, 1, "")
273 if err != nil || len(page) != 1 || next == "" {
274 t.Fatalf("first page = %#v next=%q err=%v", page, next, err)
275 }
276 page, next, err = service.ListPage(context.Background(), secondUploader, 1, "")
277 if err != nil || len(page) != 1 || next != "" {
278 t.Fatalf("second owner page = %#v next=%q err=%v", page, next, err)
279 }
280 if err := service.DeleteForUploader(context.Background(), first.SHA256, testUploader); err != nil {
281 t.Fatal(err)
282 }
283 if _, _, err := service.Get(context.Background(), second.SHA256); err != nil {
284 t.Fatalf("shared content removed with one claim: %v", err)
285 }
286 if err := service.DeleteForUploader(context.Background(), second.SHA256, secondUploader); err != nil {
287 t.Fatal(err)
288 }
289 if _, _, err := service.Get(context.Background(), second.SHA256); err == nil {
290 t.Fatal("final claim deletion retained blob")
291 }
292}
293
294// TestAgentUploadsExpireUnlessAPersonClaimsThem covers the ttl an agent
295// grant puts on uploads: claims carry the expiry, the sweep removes blobs
296// whose every claim lapsed and keeps those a person holds.
297func TestAgentUploadsExpireUnlessAPersonClaimsThem(t *testing.T) {
298 service := testService(t)
299 agent := strings.Repeat("a", 64)
300 person := strings.Repeat("b", 64)
301 now := time.Now().Unix()
302 service.config.UploadTerms = func(_ context.Context, uploader string) (UploadTerms, error) {
303 if uploader == agent {
304 return UploadTerms{ExpiresAt: now + 10}, nil
305 }
306 return UploadTerms{}, nil
307 }
308 ctx := context.Background()
309 shared, err := service.Put(ctx, PutOptions{Reader: strings.NewReader("shared site file"), Type: "text/html", Uploader: agent})
310 if err != nil {
311 t.Fatal(err)
312 }
313 alone, err := service.Put(ctx, PutOptions{Reader: strings.NewReader("agent only file"), Type: "text/html", Uploader: agent})
314 if err != nil {
315 t.Fatal(err)
316 }
317 var expires int64
318 if err := service.store.DB().QueryRow("SELECT expires FROM blob_claims WHERE sha256=? AND uploader=?", alone.SHA256, agent).Scan(&expires); err != nil || expires != now+10 {
319 t.Fatalf("agent claim expires=%d err=%v, want %d", expires, err, now+10)
320 }
321 if _, err := service.Put(ctx, PutOptions{Reader: strings.NewReader("shared site file"), Type: "text/html", Uploader: person}); err != nil {
322 t.Fatal(err)
323 }
324 if err := service.store.DB().QueryRow("SELECT expires FROM blob_claims WHERE sha256=? AND uploader=?", shared.SHA256, person).Scan(&expires); err != nil || expires != 0 {
325 t.Fatalf("person claim expires=%d err=%v, want 0", expires, err)
326 }
327 if removed, err := service.SweepExpired(ctx, now+9); err != nil || removed != 0 {
328 t.Fatalf("early sweep removed %d err=%v", removed, err)
329 }
330 if removed, err := service.SweepExpired(ctx, now+10); err != nil || removed != 1 {
331 t.Fatalf("sweep removed %d err=%v, want 1", removed, err)
332 }
333 if _, _, err := service.Get(ctx, alone.SHA256); err == nil {
334 t.Fatal("expired unclaimed blob still served")
335 }
336 if _, err := os.Stat(filepath.Join(service.root, alone.SHA256)); !errors.Is(err, os.ErrNotExist) {
337 t.Fatalf("expired blob file remains: %v", err)
338 }
339 if _, _, err := service.Get(ctx, shared.SHA256); err != nil {
340 t.Fatalf("blob a person claimed was removed: %v", err)
341 }
342 mine, err := service.ListMetadata(ctx, agent)
343 if err != nil || len(mine) != 0 {
344 t.Fatalf("agent still lists %d blobs after expiry: %v", len(mine), err)
345 }
346 theirs, err := service.ListMetadata(ctx, person)
347 if err != nil || len(theirs) != 1 {
348 t.Fatalf("person lists %d blobs: %v", len(theirs), err)
349 }
350 // A restart keeps the expiry column and the earlier rows.
351 if _, err := New(ctx, service.config); err != nil {
352 t.Fatal(err)
353 }
354}
355
356func TestEncryptedOnlyTermsRefusePlainUploads(t *testing.T) {
357 service := testService(t)
358 agent := strings.Repeat("a", 64)
359 service.config.UploadTerms = func(_ context.Context, uploader string) (UploadTerms, error) {
360 return UploadTerms{Encrypted: uploader == agent}, nil
361 }
362 ctx := context.Background()
363 png := "\x89PNG\r\n\x1a\n" + strings.Repeat("\x00", 64)
364 for name, options := range map[string]PutOptions{
365 "declared page": {Reader: strings.NewReader("<html>plain</html>"), Type: "text/html", Uploader: agent},
366 "sniffed image": {Reader: strings.NewReader(png), Type: "application/octet-stream", Uploader: agent},
367 "declared manifest": {Reader: strings.NewReader("{}"), Type: "application/json", Uploader: agent},
368 } {
369 if _, err := service.Put(ctx, options); !errors.Is(err, ErrNotEncrypted) {
370 t.Fatalf("%s: err=%v, want ErrNotEncrypted", name, err)
371 }
372 }
373 if entries, _ := os.ReadDir(service.root); len(entries) != 0 {
374 t.Fatalf("refused uploads left %d files", len(entries))
375 }
376 cipher := "\x8f\x1a\x9c\x03" + strings.Repeat("\xe2\x71\x05\x9b\x4c", 20)
377 if _, err := service.Put(ctx, PutOptions{Reader: strings.NewReader(cipher), Type: "application/octet-stream", Uploader: agent}); err != nil {
378 t.Fatalf("ciphertext refused: %v", err)
379 }
380 if _, err := service.Put(ctx, PutOptions{Reader: strings.NewReader("\x82\xa1t\x01"), Type: "application/vnd.blossom.directory+msgpack", Uploader: agent}); err != nil {
381 t.Fatalf("encrypted manifest refused: %v", err)
382 }
383 // Claiming a plain blob someone else stored is refused too.
384 if _, err := service.Put(ctx, PutOptions{Reader: strings.NewReader("<html>theirs</html>"), Type: "text/html", Uploader: strings.Repeat("b", 64)}); err != nil {
385 t.Fatal(err)
386 }
387 if _, err := service.Put(ctx, PutOptions{Reader: strings.NewReader("<html>theirs</html>"), Type: "text/html", Uploader: agent}); !errors.Is(err, ErrNotEncrypted) {
388 t.Fatalf("claim of a plain blob: err=%v", err)
389 }
390 // The door answers 403 with the reason.
391 service.config.Authorize = func(*http.Request, Action) (string, error) { return agent, nil }
392 req := httptest.NewRequest(http.MethodPut, "https://relay.test/upload", strings.NewReader("<html>plain</html>"))
393 req.Header.Set("Content-Type", "text/html")
394 response := record(service.Handler().ServeHTTP, req)
395 if response.Code != http.StatusForbidden || !strings.Contains(response.Body.String(), "requires encrypted uploads") {
396 t.Fatalf("plain upload = %d %s", response.Code, response.Body.String())
397 }
398}
399
400const testUploader = "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef"
401
402func testService(t *testing.T) *Service {
403 t.Helper()
404 root := t.TempDir()
405 store, err := storage.Open(context.Background(), filepath.Join(root, "relay.db"))
406 if err != nil {
407 t.Fatal(err)
408 }
409 t.Cleanup(func() { _ = store.Close() })
410 service, err := New(context.Background(), Config{Root: root, Store: store, Authorize: func(*http.Request, Action) (string, error) { return "uploader", nil }})
411 if err != nil {
412 t.Fatal(err)
413 }
414 return service
415}
416
417func record(handler http.HandlerFunc, req *http.Request) *httptest.ResponseRecorder {
418 response := httptest.NewRecorder()
419 handler(response, req)
420 return response
421}

About

ownerc01503e8035f
visibilitypublic
headd730679639bf

Maintainers

  • c01503e8035f owner
  • ba1b5beed5b5 agent hermes

Clone

  • https://012.run/npub1cq2s86qrtadf36eqq3epm38wmhdynrrcwxf84eldk3f94dkv7ttsrulrz5/tinyrelay.git

Refs