- commit
- ac4fefc833a3a19b92c6644ebc1ed5b3f1c662c7
- parent
- 0aca966a569138bd3e705a7f9931fa95a1a348f3
- author
- Dami <dami.etoile@gmail.com>
| M | internal/ | 10 | +- |
| M | internal/ | 19 | + |
| M | internal/ | 187 | ++++++++++- |
| M | internal/ | 113 | ++++++ |
| A | internal/ | 525 | ++++++++++++++++++++++++++++++ |
| A | internal/ | 192 | +++++++++++ |
| A | internal/ | 171 | ++++++++++ |
| A | internal/ | 128 | +++++++ |
| A | internal/ | 131 | +++++++ |
| A | internal/ | 103 | ++++++ |
| A | internal/ | 125 | +++++++ |
| M | internal/ | 5 | +- |
| M | internal/ | 3 | + |
| M | internal/ | 48 | +++- |
| M | internal/ | 65 | ++++ |
| M | internal/ | 29 | ++ |
| M | internal/ | 9 | +- |
| M | internal/ | 3 | + |
| M | internal/ | 75 | ++++- |
| M | scripts/ | 175 | ++++++++++ |
@@ -105,16 +105,16 @@ func (v *Validator) verifyBlossom(header, action, server, blobHash string, stric
105105 if expiration == 0 || expiration <= now.Unix() {106106 return event.Event{}, authError("token expired")107107 }108- if e.CreatedAt > now.Unix()+int64(5*time.Minute/time.Second) {108+ // BUD-11 authorization timestamps may not be from the future. The109+ // expiration tag, rather than an arbitrary maximum age, controls how long110+ // an otherwise valid token remains usable.111+ if e.CreatedAt > now.Unix() {109112 return event.Event{}, authError("token is from the future")110113 }111- if strict && e.CreatedAt < now.Unix()-int64(5*time.Minute/time.Second) {112- return event.Event{}, authError("token is expired")113- }114114 if strict && requiresBlobScope(action) && len(event.TagValues(e, "x")) == 0 {115115 return event.Event{}, authError("token must include an x tag")116116 }117- if strict && blobHash != "" && !contains(event.TagValues(e, "x"), blobHash) {117+ if strict && blobHash != "" && len(event.TagValues(e, "x")) > 0 && !contains(event.TagValues(e, "x"), blobHash) {118118 return event.Event{}, authError("token x tag does not name this blob")119119 }120120 return e, nil@@ -122,6 +122,25 @@ func TestVerifyBlossomRequestRequiresScopeAndServer(t *testing.T) {
122122 }123123 }124124 125+func TestVerifyBlossomRequestAllowsUnscopedGet(t *testing.T) {126+ now := time.Unix(1_700_000_000, 0)127+ validator := NewValidator(func() time.Time { return now })128+ hash := sha256Hex("blob")129+ e := signedEvent(t, 24242, now.Add(-24*time.Hour).Unix(), "", [][]string{{"t", "get"}, {"expiration", "1700086400"}})130+ if _, err := validator.VerifyBlossomRequest(token(t, e), "get", "relay.example", hash); err != nil {131+ t.Fatalf("unscoped GET token rejected: %v", err)132+ }133+}134+135+func TestVerifyBlossomRequestRejectsFutureCreatedAt(t *testing.T) {136+ now := time.Unix(1_700_000_000, 0)137+ validator := NewValidator(func() time.Time { return now })138+ e := signedEvent(t, 24242, now.Add(time.Second).Unix(), "", [][]string{{"t", "get"}, {"expiration", "1700000300"}})139+ if _, err := validator.VerifyBlossomRequest(token(t, e), "get", "relay.example", ""); err == nil || !strings.Contains(err.Error(), "future") {140+ t.Fatalf("future token accepted: %v", err)141+ }142+}143+125144 func TestChallengeManagerBindsRelayAndConsumesChallenge(t *testing.T) {126145 now := time.Unix(1_700_000_000, 0)127146 m := NewChallengeManager("wss://relay.example/tenant", func() time.Time { return now })@@ -19,6 +19,7 @@ import (
1919 "sort"2020 "strconv"2121 "strings"22+ "sync"2223 "time"2324 2425 "github.com/FelineStateMachine/tinyrelay/internal/event"@@ -57,6 +58,14 @@ type Config struct {
5758 // RecordReport joins native blob reporting to the host moderation ledger.5859 // It runs inside the blob report transaction.5960 RecordReport ReportRecorder61+ // Limits is read for each upload so policy changes apply without reopening62+ // the blob service. Zero values mean unlimited.63+ Limits func() Limits64+}65+66+type Limits struct {67+ MaxFileBytes int6468+ UserStorageBytes int646069 }6170 6271 type Blob struct {@@ -79,10 +88,18 @@ type Service struct {
7988 root string8089 store *storage.Store8190 resolve ResolveIP91+ quotaMu sync.Mutex92+ // multipartMu protects partial files and their range metadata. Lock it93+ // before quotaMu; network bodies are never read while either is held.94+ multipartMu sync.Mutex95+ activeParts map[string]int8296 }8397 8498 var ErrBlocked = errors.New("blocked: this blob was removed by a moderator")8599 var ErrHashMismatch = errors.New("conflict: mirrored blob hash does not match the requested hash")100+var ErrFileTooLarge = errors.New("invalid: blob exceeds the per-file size limit")101+var ErrQuotaExceeded = errors.New("restricted: uploader storage quota exceeded")102+var ErrMultipartConflict = errors.New("conflict: upload metadata differs from existing session")86103 87104 var shaPattern = regexp.MustCompile(`^[0-9a-f]{64}$`)88105 var blobPathPattern = regexp.MustCompile(`^/([0-9a-f]{64})(?:\.[a-z0-9]{1,8})?$`)@@ -99,6 +116,7 @@ var extensions = map[string]string{
99116 "image/png": "png", "image/jpeg": "jpg", "image/gif": "gif", "image/webp": "webp", "image/avif": "avif", "image/svg+xml": "svg",100117 "video/mp4": "mp4", "video/webm": "webm", "video/quicktime": "mov", "audio/mpeg": "mp3", "audio/ogg": "ogg", "audio/wav": "wav", "audio/mp4": "m4a",101118 "application/pdf": "pdf", "application/json": "json", "text/plain": "txt", "text/markdown": "md",119+ "application/vnd.blossom.directory+msgpack": "bdir",102120 }103121 104122 func New(ctx context.Context, config Config) (*Service, error) {@@ -126,6 +144,8 @@ func New(ctx context.Context, config Config) (*Service, error) {
126144 WHEN NEW.uploader != '' BEGIN INSERT OR IGNORE INTO blob_claims(sha256,uploader,claimed_at) VALUES(NEW.sha256,NEW.uploader,NEW.uploaded); END;127145 CREATE TABLE IF NOT EXISTS blob_blocks (sha256 TEXT PRIMARY KEY, reason TEXT NOT NULL DEFAULT '', blocked_at INTEGER NOT NULL DEFAULT 0);128146 CREATE TABLE IF NOT EXISTS blob_tombstones (sha256 TEXT PRIMARY KEY, deleted_at INTEGER NOT NULL);147+ CREATE TABLE IF NOT EXISTS multipart_uploads (id TEXT PRIMARY KEY, sha256 TEXT NOT NULL, uploader TEXT NOT NULL, length INTEGER NOT NULL, type TEXT NOT NULL, created INTEGER NOT NULL, last_seen INTEGER NOT NULL);148+ CREATE TABLE IF NOT EXISTS multipart_parts (upload_id TEXT NOT NULL, offset INTEGER NOT NULL, length INTEGER NOT NULL, PRIMARY KEY(upload_id, offset), FOREIGN KEY(upload_id) REFERENCES multipart_uploads(id) ON DELETE CASCADE);129149 INSERT OR IGNORE INTO blob_claims(sha256,uploader,claimed_at)130150 SELECT sha256,uploader,uploaded FROM blobs WHERE uploader != ''`); err != nil {131151 return nil, fmt.Errorf("initialize blob metadata: %w", err)@@ -139,7 +159,11 @@ func New(ctx context.Context, config Config) (*Service, error) {
139159 return net.DefaultResolver.LookupIP(ctx, "ip", host)140160 }141161 }142- return &Service{config: config, root: filepath.Join(config.Root, "blobs"), store: config.Store, resolve: resolve}, nil162+ s := &Service{config: config, root: filepath.Join(config.Root, "blobs"), store: config.Store, resolve: resolve, activeParts: make(map[string]int)}163+ if err := s.reconcileMultipart(ctx); err != nil {164+ return nil, err165+ }166+ return s, nil143167 }144168 145169 func (s *Service) Handler() http.HandlerFunc { return s.serveHTTP }@@ -320,9 +344,13 @@ func (s *Service) Descriptor(baseURL string, entry Blob) map[string]any {
320344 321345 func (s *Service) serveHTTP(w http.ResponseWriter, r *http.Request) {322346 w.Header().Set("access-control-allow-origin", "*")323- w.Header().Set("access-control-allow-headers", "authorization, content-type, x-sha-256, x-content-length")324- w.Header().Set("access-control-allow-methods", "GET, HEAD, PUT, POST, DELETE, OPTIONS")347+ w.Header().Set("access-control-allow-headers", "authorization, content-type, x-sha-256, x-content-length, x-content-type, upload-type, upload-length, upload-offset")348+ w.Header().Set("access-control-allow-methods", "GET, HEAD, PUT, PATCH, POST, DELETE, OPTIONS")349+ w.Header().Set("Access-Control-Expose-Headers", "Allow, X-Reason")325350 if r.Method == http.MethodOptions {351+ if blobPathPattern.MatchString(r.URL.Path) {352+ w.Header().Set("Allow", "GET, HEAD, PUT, PATCH, DELETE, OPTIONS")353+ }326354 w.WriteHeader(http.StatusNoContent)327355 return328356 }@@ -342,6 +370,14 @@ func (s *Service) serveHTTP(w http.ResponseWriter, r *http.Request) {
342370 s.mirror(w, r)343371 return344372 }373+ if match := blobPathPattern.FindStringSubmatch(r.URL.Path); match != nil && (r.Method == http.MethodPut || r.Method == http.MethodPatch || r.Method == http.MethodOptions) {374+ if r.Method == http.MethodPatch || r.Method == http.MethodOptions {375+ s.multipart(w, r, match[1])376+ return377+ }378+ s.pathUpload(w, r, match[1])379+ return380+ }345381 if r.URL.Path == "/report" && r.Method == http.MethodPut {346382 s.report(w, r)347383 return@@ -388,12 +424,18 @@ func (s *Service) uploadHead(w http.ResponseWriter, r *http.Request) {
388424 s.fail(w, http.StatusBadRequest, errors.New("invalid: X-SHA-256 must be lowercase hex"))389425 return390426 }391- length, err := strconv.ParseInt(strings.TrimSpace(r.Header.Get("x-content-length")), 10, 64)427+ rawLength := strings.TrimSpace(r.Header.Get("x-content-length"))428+ if rawLength == "" {429+ s.fail(w, http.StatusLengthRequired, errors.New("invalid: X-Content-Length is required"))430+ return431+ }432+ length, err := strconv.ParseInt(rawLength, 10, 64)392433 if err != nil || length <= 0 {393434 s.fail(w, http.StatusBadRequest, errors.New("invalid: X-Content-Length must be a positive integer"))394435 return395436 }396- if _, err := s.authorize(r, ActionUpload); err != nil {437+ pubkey, err := s.authorize(r, ActionUpload)438+ if err != nil {397439 s.fail(w, http.StatusUnauthorized, err)398440 return399441 }@@ -405,6 +447,26 @@ func (s *Service) uploadHead(w http.ResponseWriter, r *http.Request) {
405447 s.fail(w, http.StatusBadRequest, errors.New("invalid: X-Content-Type is required"))406448 return407449 }450+ if existing, lookupErr := s.lookup(r.Context(), sha); lookupErr == nil && existing.SHA256 != "" && s.hasClaim(r.Context(), sha, pubkey) {451+ w.WriteHeader(http.StatusOK)452+ return453+ }454+ limits := s.currentLimits()455+ if limits.MaxFileBytes > 0 && length > limits.MaxFileBytes {456+ s.fail(w, http.StatusRequestEntityTooLarge, ErrFileTooLarge)457+ return458+ }459+ // HEAD is advisory for deduplicated uploads, whose final physical size is460+ // already present, but it should still reflect the current quota budget.461+ if limits.UserStorageBytes > 0 {462+ if used, quotaErr := s.quotaUsage(r.Context(), pubkey); quotaErr != nil {463+ s.fail(w, http.StatusInternalServerError, quotaErr)464+ return465+ } else if length > limits.UserStorageBytes-used {466+ s.fail(w, http.StatusForbidden, ErrQuotaExceeded)467+ return468+ }469+ }408470 if existing, err := s.lookup(r.Context(), sha); err == nil && existing.SHA256 != "" {409471 w.WriteHeader(http.StatusOK)410472 return@@ -516,6 +578,16 @@ func (s *Service) put(ctx context.Context, source io.Reader, typ, uploader, clai
516578 }517579 518580 func (s *Service) putValidated(ctx context.Context, source io.Reader, typ, uploader, claimed string, validate func(string) error) (Blob, bool, error) {581+ limits := s.currentLimits()582+ streamLimit := limits.MaxFileBytes583+ quotaStream := false584+ if streamLimit == 0 && limits.UserStorageBytes > 0 && uploader != "" {585+ streamLimit = limits.UserStorageBytes586+ quotaStream = true587+ }588+ if streamLimit > 0 && streamLimit < int64(^uint64(0)>>1) {589+ source = io.LimitReader(source, streamLimit+1)590+ }519591 tmp, err := os.CreateTemp(s.root, ".upload-")520592 if err != nil {521593 return Blob{}, false, fmt.Errorf("create upload temporary file: %w", err)@@ -528,9 +600,13 @@ func (s *Service) putValidated(ctx context.Context, source io.Reader, typ, uploa
528600 _ = tmp.Close()529601 return Blob{}, false, fmt.Errorf("read upload: %w", err)530602 }531- if count == 0 {603+ if limits.MaxFileBytes > 0 && count > limits.MaxFileBytes {532604 _ = tmp.Close()533- return Blob{}, false, errors.New("invalid: empty body")605+ return Blob{}, false, ErrFileTooLarge606+ }607+ if quotaStream && count > streamLimit {608+ _ = tmp.Close()609+ return Blob{}, false, ErrQuotaExceeded534610 }535611 if err := tmp.Sync(); err != nil {536612 _ = tmp.Close()@@ -548,21 +624,66 @@ func (s *Service) putValidated(ctx context.Context, source io.Reader, typ, uploa
548624 return Blob{}, false, err549625 }550626 }627+ return s.installUpload(ctx, uploadCandidate{path: tmpName, hash: sha, size: count, typ: typ, uploader: uploader})628+}629+630+type uploadCandidate struct {631+ path, hash, typ, uploader, reservation string632+ size int64633+}634+635+func (s *Service) installUpload(ctx context.Context, candidate uploadCandidate) (Blob, bool, error) {636+ sha, count, typ, uploader := candidate.hash, candidate.size, candidate.typ, candidate.uploader637+ // Serialize the lookup, quota check, installation and metadata transaction.638+ // This closes the race where concurrent uploads could each observe the same639+ // remaining quota before one of them installs a new physical object.640+ s.quotaMu.Lock()641+ defer s.quotaMu.Unlock()551642 if s.blocked(ctx, sha) {552643 return Blob{}, false, ErrBlocked553644 }645+ limits := s.currentLimits()646+ if limits.MaxFileBytes > 0 && count > limits.MaxFileBytes {647+ return Blob{}, false, ErrFileTooLarge648+ }649+ if limits.UserStorageBytes > 0 && uploader != "" && count > limits.UserStorageBytes {650+ return Blob{}, false, ErrQuotaExceeded651+ }554652 if existing, err := s.lookup(ctx, sha); err == nil {555653 if uploader != "" {654+ if limits.UserStorageBytes > 0 && !s.hasClaim(ctx, sha, uploader) {655+ used, usageErr := s.quotaUsageExcept(ctx, uploader, candidate.reservation)656+ if usageErr != nil {657+ return Blob{}, false, fmt.Errorf("check uploader storage quota: %w", usageErr)658+ }659+ if existing.Size > limits.UserStorageBytes-used {660+ return Blob{}, false, ErrQuotaExceeded661+ }662+ }556663 if err := s.claim(ctx, sha, uploader); err != nil {557664 return Blob{}, false, err558665 }559666 }667+ if candidate.reservation != "" {668+ if _, err := s.store.DB().ExecContext(ctx, "DELETE FROM multipart_uploads WHERE id=?", candidate.reservation); err != nil {669+ return Blob{}, false, err670+ }671+ }560672 return existing, false, nil561673 } else if !errors.Is(err, sql.ErrNoRows) {562674 return Blob{}, false, err563675 }676+ if limits.UserStorageBytes > 0 && uploader != "" {677+ used, err := s.quotaUsageExcept(ctx, uploader, candidate.reservation)678+ if err != nil {679+ return Blob{}, false, fmt.Errorf("check uploader storage quota: %w", err)680+ }681+ if count > limits.UserStorageBytes-used {682+ return Blob{}, false, ErrQuotaExceeded683+ }684+ }564685 final := filepath.Join(s.root, sha)565- if err := os.Rename(tmpName, final); err != nil && !errors.Is(err, os.ErrExist) {686+ if err := os.Rename(candidate.path, final); err != nil && !errors.Is(err, os.ErrExist) {566687 return Blob{}, false, fmt.Errorf("install blob: %w", err)567688 }568689 if err := syncDirectory(s.root); err != nil {@@ -574,7 +695,7 @@ func (s *Service) putValidated(ctx context.Context, source io.Reader, typ, uploa
574695 typ = detectFileType(final)575696 }576697 entry := Blob{SHA256: sha, Size: count, Type: typ, Uploader: uploader, Uploaded: now}577- err = s.store.WithTx(ctx, func(tx *sql.Tx) error {698+ err := s.store.WithTx(ctx, func(tx *sql.Tx) error {578699 if _, err := tx.ExecContext(ctx, "DELETE FROM blob_tombstones WHERE sha256=?", entry.SHA256); err != nil {579700 return fmt.Errorf("clear blob deletion tombstone: %w", err)580701 }@@ -586,6 +707,11 @@ func (s *Service) putValidated(ctx context.Context, source io.Reader, typ, uploa
586707 return fmt.Errorf("record blob claim: %w", err)587708 }588709 }710+ if candidate.reservation != "" {711+ if _, err := tx.ExecContext(ctx, "DELETE FROM multipart_uploads WHERE id=?", candidate.reservation); err != nil {712+ return fmt.Errorf("release upload reservation: %w", err)713+ }714+ }589715 return nil590716 })591717 if err != nil {@@ -597,6 +723,21 @@ func (s *Service) putValidated(ctx context.Context, source io.Reader, typ, uploa
597723 return entry, true, nil598724 }599725 726+func (s *Service) quotaUsage(ctx context.Context, uploader string) (int64, error) {727+ return s.quotaUsageExcept(ctx, uploader, "")728+}729+730+func (s *Service) quotaUsageExcept(ctx context.Context, uploader, reservation string) (int64, error) {731+ var used sql.NullInt64732+ err := s.store.DB().QueryRowContext(ctx, `SELECT COALESCE((SELECT SUM(b.size)733+ FROM blobs b JOIN blob_claims c ON c.sha256=b.sha256 WHERE c.uploader=?),0) +734+ COALESCE((SELECT SUM(length) FROM multipart_uploads WHERE uploader=? AND id != ?),0)`, uploader, uploader, reservation).Scan(&used)735+ if err != nil {736+ return 0, err737+ }738+ return used.Int64, nil739+}740+600741 func (s *Service) claim(ctx context.Context, sha, uploader string) error {601742 _, err := s.store.DB().ExecContext(ctx, "INSERT OR IGNORE INTO blob_claims(sha256,uploader,claimed_at) VALUES(?,?,?)", sha, uploader, time.Now().UTC().Unix())602743 if err != nil {@@ -605,6 +746,20 @@ func (s *Service) claim(ctx context.Context, sha, uploader string) error {
605746 return nil606747 }607748 749+func (s *Service) currentLimits() Limits {750+ if s.config.Limits == nil {751+ return Limits{}752+ }753+ limits := s.config.Limits()754+ if limits.MaxFileBytes < 0 {755+ limits.MaxFileBytes = 0756+ }757+ if limits.UserStorageBytes < 0 {758+ limits.UserStorageBytes = 0759+ }760+ return limits761+}762+608763 func (s *Service) hasClaim(ctx context.Context, sha, uploader string) bool {609764 var found int610765 err := s.store.DB().QueryRowContext(ctx, "SELECT 1 FROM blob_claims WHERE sha256=? AND uploader=?", sha, uploader).Scan(&found)@@ -634,7 +789,7 @@ func reconcile(ctx context.Context, store *storage.Store, root string) error {
634789 continue635790 }636791 name := entry.Name()637- if strings.HasPrefix(name, ".upload-") {792+ if strings.HasPrefix(name, ".upload-") || strings.HasPrefix(name, ".multipart-chunk-") {638793 if err := os.Remove(filepath.Join(root, name)); err != nil {639794 return fmt.Errorf("remove abandoned upload: %w", err)640795 }@@ -1138,9 +1293,21 @@ func statusFor(err error) int {
11381293 if errors.Is(err, ErrHashMismatch) {11391294 return http.StatusConflict11401295 }1296+ if errors.Is(err, ErrMultipartConflict) {1297+ return http.StatusConflict1298+ }1299+ if errors.Is(err, ErrFileTooLarge) {1300+ return http.StatusRequestEntityTooLarge1301+ }1302+ if errors.Is(err, ErrQuotaExceeded) {1303+ return http.StatusForbidden1304+ }11411305 if errors.Is(err, ErrBlocked) {11421306 return http.StatusForbidden11431307 }1308+ if strings.HasPrefix(err.Error(), "auth-required:") {1309+ return http.StatusUnauthorized1310+ }11441311 if strings.HasPrefix(err.Error(), "invalid:") {11451312 return http.StatusBadRequest11461313 }@@ -4,6 +4,7 @@ import (
44 "context"55 "crypto/sha256"66 "encoding/hex"7+ "errors"78 "io"89 "net"910 "net/http"@@ -11,7 +12,9 @@ import (
1112 "os"1213 "path/filepath"1314 "strings"15+ "sync"1416 "testing"17+ "time"1518 1619 "github.com/FelineStateMachine/tinyrelay/internal/storage"1720 )@@ -46,6 +49,116 @@ func TestUploadDeduplicatesAndServesByHash(t *testing.T) {
4649 }4750 }4851 52+func 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+91+func TestConcurrentUploadsRespectUploaderQuota(t *testing.T) {92+ service := testService(t)93+ service.config.Limits = func() Limits { return Limits{UserStorageBytes: 4} }94+ var wg sync.WaitGroup95+ 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 <- err102+ }(body)103+ }104+ wg.Wait()105+ close(errs)106+ var success, quota int107+ 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+119+func 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+128+func 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 <- err135+ }()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 <- err146+ }()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+49162 func TestPrivateBlobRequiresReadAuthorization(t *testing.T) {50163 service := testService(t)51164 service.config.CanRead = func(context.Context, string, []string) bool { return false }@@ -0,0 +1,525 @@
1+package blob2+3+import (4+ "context"5+ "crypto/sha256"6+ "database/sql"7+ "encoding/hex"8+ "errors"9+ "fmt"10+ "io"11+ "mime"12+ "net/http"13+ "os"14+ "path/filepath"15+ "sort"16+ "strconv"17+ "strings"18+ "time"19+)20+21+const (22+ multipartExpiry = 60 * time.Second23+ multipartChunkLimit = 8 << 2024+ multipartStorageLimit = 1 << 3025+ multipartSessionLimit = 6426+ multipartUserSessionLimit = 1627+ multipartActiveLimit = 3228+ multipartRangeLimit = 102429+)30+31+type multipartRequest struct {32+ id, hash, uploader, typ string33+ length, offset, count int6434+}35+36+type multipartError struct {37+ status int38+ message string39+}40+41+func (e *multipartError) Error() string { return e.message }42+func multipartFailure(status int, message string) error { return &multipartError{status, message} }43+func (s *Service) failMultipart(w http.ResponseWriter, err error) {44+ var failure *multipartError45+ if errors.As(err, &failure) {46+ s.fail(w, failure.status, err)47+ return48+ }49+ s.fail(w, statusFor(err), err)50+}51+52+func (s *Service) multipart(w http.ResponseWriter, r *http.Request, hash string) {53+ uploader, err := s.authorize(r, ActionUpload)54+ if err != nil {55+ s.fail(w, http.StatusUnauthorized, err)56+ return57+ }58+ part, err := parseMultipart(r, hash, uploader)59+ if err != nil {60+ s.failMultipart(w, err)61+ return62+ }63+ if err := s.reserveMultipart(r.Context(), part); err != nil {64+ s.failMultipart(w, err)65+ return66+ }67+ defer s.releaseMultipart(part.id)68+ chunk, err := s.stageMultipart(r, part)69+ if err != nil {70+ s.failMultipart(w, err)71+ return72+ }73+ defer os.Remove(chunk)74+ entry, created, complete, err := s.commitMultipart(r.Context(), part, chunk)75+ if err != nil {76+ s.failMultipart(w, err)77+ return78+ }79+ if !complete {80+ w.WriteHeader(http.StatusNoContent)81+ return82+ }83+ s.writeJSON(w, chooseStatus(created, http.StatusCreated), s.descriptor(r, entry))84+}85+86+func parseMultipart(r *http.Request, hash, uploader string) (multipartRequest, error) {87+ part := multipartRequest{id: multipartID(hash, uploader), hash: hash, uploader: uploader, count: r.ContentLength}88+ if r.ContentLength < 0 {89+ return part, multipartFailure(411, "Content-Length is required")90+ }91+ if r.Header.Get("Content-Type") != "application/octet-stream" {92+ return part, multipartFailure(415, "Content-Type must be application/octet-stream")93+ }94+ for _, item := range []struct {95+ name string96+ target *int6497+ }{{"Upload-Length", &part.length}, {"Upload-Offset", &part.offset}} {98+ raw := r.Header.Get(item.name)99+ value, err := strconv.ParseInt(raw, 10, 64)100+ if err != nil || value < 0 || raw == "" || strings.Trim(raw, "0123456789") != "" {101+ return part, multipartFailure(400, item.name+" must be a nonnegative integer")102+ }103+ *item.target = value104+ }105+ if part.offset > part.length || part.count > part.length-part.offset {106+ return part, multipartFailure(416, "chunk range exceeds Upload-Length")107+ }108+ if part.count == 0 && part.length != 0 {109+ return part, multipartFailure(400, "nonempty uploads require nonempty chunks")110+ }111+ if part.count > multipartChunkLimit || part.length > multipartStorageLimit {112+ return part, multipartFailure(413, "multipart upload exceeds temporary storage limits")113+ }114+ typ, _, err := mime.ParseMediaType(r.Header.Get("Upload-Type"))115+ if err != nil || !strings.Contains(typ, "/") {116+ return part, multipartFailure(400, "Upload-Type must be a MIME type")117+ }118+ part.typ = contentType(typ)119+ return part, nil120+}121+122+func multipartID(hash, uploader string) string {123+ digest := sha256.Sum256([]byte(uploader))124+ return hash + "-" + hex.EncodeToString(digest[:])125+}126+func (s *Service) multipartPath(id string) string { return filepath.Join(s.root, ".multipart-"+id) }127+128+// Admission reserves the final length before reading any request bytes. Each129+// active chunk is also bounded, so unlimited durable storage does not mean130+// unlimited temporary files or sessions.131+func (s *Service) reserveMultipart(ctx context.Context, part multipartRequest) error {132+ s.multipartMu.Lock()133+ defer s.multipartMu.Unlock()134+ s.quotaMu.Lock()135+ defer s.quotaMu.Unlock()136+ if err := s.cleanupMultipartLocked(ctx); err != nil {137+ return err138+ }139+ active := 0140+ for _, count := range s.activeParts {141+ active += count142+ }143+ if active >= multipartActiveLimit {144+ return multipartFailure(429, "too many active upload chunks")145+ }146+ if err := s.multipartAdmission(ctx, part); err != nil {147+ return err148+ }149+ s.activeParts[part.id]++150+ return nil151+}152+153+func (s *Service) multipartAdmission(ctx context.Context, part multipartRequest) error {154+ if s.blocked(ctx, part.hash) {155+ return ErrBlocked156+ }157+ limits := s.currentLimits()158+ if limits.MaxFileBytes > 0 && part.length > limits.MaxFileBytes {159+ return ErrFileTooLarge160+ }161+ existing, lookupErr := s.lookup(ctx, part.hash)162+ if lookupErr != nil && !errors.Is(lookupErr, sql.ErrNoRows) {163+ return lookupErr164+ }165+ if lookupErr == nil {166+ if existing.Size != part.length {167+ return ErrMultipartConflict168+ }169+ return s.checkMultipartQuota(ctx, part, limits)170+ }171+ var length int64172+ var typ string173+ err := s.store.DB().QueryRowContext(ctx, "SELECT length,type FROM multipart_uploads WHERE id=?", part.id).Scan(&length, &typ)174+ if err != nil && !errors.Is(err, sql.ErrNoRows) {175+ return err176+ }177+ if err == nil && (length != part.length || typ != part.typ) {178+ return ErrMultipartConflict179+ }180+ if err := s.checkMultipartQuota(ctx, part, limits); err != nil {181+ return err182+ }183+ if errors.Is(err, sql.ErrNoRows) {184+ return s.createMultipart(ctx, part)185+ }186+ _, err = s.store.DB().ExecContext(ctx, "UPDATE multipart_uploads SET last_seen=? WHERE id=?", time.Now().Unix(), part.id)187+ return err188+}189+190+func (s *Service) checkMultipartQuota(ctx context.Context, part multipartRequest, limits Limits) error {191+ if limits.UserStorageBytes == 0 || s.hasClaim(ctx, part.hash, part.uploader) {192+ return nil193+ }194+ used, err := s.quotaUsageExcept(ctx, part.uploader, part.id)195+ if err != nil {196+ return err197+ }198+ if part.length > limits.UserStorageBytes-used {199+ return ErrQuotaExceeded200+ }201+ return nil202+}203+204+func (s *Service) createMultipart(ctx context.Context, part multipartRequest) error {205+ var count, userCount int206+ var total int64207+ err := s.store.DB().QueryRowContext(ctx, `SELECT COUNT(*),COALESCE(SUM(length),0),COALESCE(SUM(uploader=?),0) FROM multipart_uploads`, part.uploader).Scan(&count, &total, &userCount)208+ if err != nil {209+ return err210+ }211+ if count >= multipartSessionLimit || userCount >= multipartUserSessionLimit || part.length > multipartStorageLimit-total {212+ return multipartFailure(429, "temporary upload capacity is full")213+ }214+ path := s.multipartPath(part.id)215+ file, err := os.OpenFile(path, os.O_CREATE|os.O_TRUNC|os.O_RDWR, 0o600)216+ if err != nil {217+ return err218+ }219+ err = errors.Join(file.Truncate(part.length), file.Sync(), file.Close())220+ if err != nil {221+ _ = os.Remove(path)222+ return err223+ }224+ now := time.Now().Unix()225+ _, err = s.store.DB().ExecContext(ctx, "INSERT INTO multipart_uploads(id,sha256,uploader,length,type,created,last_seen) VALUES(?,?,?,?,?,?,?)", part.id, part.hash, part.uploader, part.length, part.typ, now, now)226+ if err != nil {227+ _ = os.Remove(path)228+ }229+ return err230+}231+232+func (s *Service) stageMultipart(r *http.Request, part multipartRequest) (string, error) {233+ chunk, err := os.CreateTemp(s.root, ".multipart-chunk-")234+ if err != nil {235+ return "", err236+ }237+ path := chunk.Name()238+ keep := false239+ defer func() {240+ _ = chunk.Close()241+ if !keep {242+ _ = os.Remove(path)243+ }244+ }()245+ hasher := sha256.New()246+ if _, err := io.CopyN(io.MultiWriter(chunk, hasher), r.Body, part.count); err != nil {247+ return "", multipartFailure(400, "chunk body is shorter than Content-Length")248+ }249+ var extra [1]byte250+ if n, err := io.ReadFull(r.Body, extra[:]); n != 0 || (err != nil && !errors.Is(err, io.EOF)) {251+ return "", multipartFailure(400, "chunk body is longer than Content-Length")252+ }253+ if err := r.Context().Err(); err != nil {254+ return "", err255+ }256+ if s.config.ValidateUpload != nil {257+ if err := s.config.ValidateUpload(r, hex.EncodeToString(hasher.Sum(nil))); err != nil {258+ return "", err259+ }260+ }261+ if err := chunk.Close(); err != nil {262+ return "", err263+ }264+ keep = true265+ return path, nil266+}267+268+func (s *Service) releaseMultipart(id string) {269+ s.multipartMu.Lock()270+ defer s.multipartMu.Unlock()271+ s.activeParts[id]--272+ if s.activeParts[id] > 0 {273+ return274+ }275+ delete(s.activeParts, id)276+ ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)277+ defer cancel()278+ var parts int279+ if err := s.store.DB().QueryRowContext(ctx, "SELECT COUNT(*) FROM multipart_parts WHERE upload_id=?", id).Scan(&parts); err == nil && parts == 0 {280+ s.quotaMu.Lock()281+ defer s.quotaMu.Unlock()282+ _ = s.discardMultipart(ctx, id)283+ }284+}285+286+func (s *Service) commitMultipart(ctx context.Context, part multipartRequest, chunk string) (Blob, bool, bool, error) {287+ s.multipartMu.Lock()288+ defer s.multipartMu.Unlock()289+ // Recheck policy after streaming, including uploads whose next chunk does290+ // not complete the file. Policy may have changed while the body arrived.291+ s.quotaMu.Lock()292+ admissionErr := s.multipartAdmission(ctx, part)293+ s.quotaMu.Unlock()294+ if admissionErr != nil {295+ return Blob{}, false, false, admissionErr296+ }297+ candidate := uploadCandidate{hash: part.hash, size: part.length, typ: part.typ, uploader: part.uploader, reservation: part.id}298+ if existing, err := s.lookup(ctx, part.hash); err == nil {299+ if existing.Size != part.length {300+ return Blob{}, false, false, ErrMultipartConflict301+ }302+ entry, _, err := s.installUpload(ctx, candidate)303+ return entry, false, err == nil, err304+ } else if !errors.Is(err, sql.ErrNoRows) {305+ return Blob{}, false, false, err306+ }307+ ranges, err := s.multipartRanges(ctx, part)308+ if err != nil {309+ return Blob{}, false, false, err310+ }311+ candidate.path = s.multipartPath(part.id)312+ if err := writeMultipartChunk(candidate.path, chunk, part.offset); err != nil {313+ _ = s.discardMultipart(ctx, part.id)314+ return Blob{}, false, false, err315+ }316+ if err := s.recordMultipartRanges(ctx, part.id, ranges); err != nil {317+ _ = s.discardMultipart(ctx, part.id)318+ return Blob{}, false, false, err319+ }320+ if len(ranges) != 1 || ranges[0].start != 0 || ranges[0].end != part.length {321+ return Blob{}, false, false, nil322+ }323+ if err := verifyMultipart(candidate.path, part.hash, part.length); err != nil {324+ return Blob{}, false, false, errors.Join(err, s.discardMultipart(ctx, part.id))325+ }326+ entry, created, err := s.installUpload(ctx, candidate)327+ cleanupErr := s.discardMultipart(ctx, part.id)328+ return entry, created, err == nil && cleanupErr == nil, errors.Join(err, cleanupErr)329+}330+331+type multipartRange struct{ start, end int64 }332+333+func (s *Service) multipartRanges(ctx context.Context, part multipartRequest) ([]multipartRange, error) {334+ rows, err := s.store.DB().QueryContext(ctx, "SELECT offset,length FROM multipart_parts WHERE upload_id=? ORDER BY offset", part.id)335+ if err != nil {336+ return nil, err337+ }338+ defer rows.Close()339+ ranges := []multipartRange{{part.offset, part.offset + part.count}}340+ for rows.Next() {341+ var offset, length int64342+ if err := rows.Scan(&offset, &length); err != nil {343+ return nil, err344+ }345+ ranges = append(ranges, multipartRange{offset, offset + length})346+ }347+ if err := rows.Err(); err != nil {348+ return nil, err349+ }350+ sort.Slice(ranges, func(i, j int) bool { return ranges[i].start < ranges[j].start })351+ merged := []multipartRange{ranges[0]}352+ for _, span := range ranges[1:] {353+ last := &merged[len(merged)-1]354+ if span.start <= last.end {355+ last.end = max(last.end, span.end)356+ } else {357+ merged = append(merged, span)358+ }359+ }360+ if len(merged) > multipartRangeLimit {361+ return nil, multipartFailure(429, "too many disjoint upload ranges")362+ }363+ return merged, nil364+}365+366+func writeMultipartChunk(path, chunk string, offset int64) error {367+ source, err := os.Open(chunk)368+ if err != nil {369+ return err370+ }371+ defer source.Close()372+ target, err := os.OpenFile(path, os.O_RDWR, 0o600)373+ if err != nil {374+ return err375+ }376+ defer target.Close()377+ if _, err := target.Seek(offset, io.SeekStart); err != nil {378+ return err379+ }380+ if _, err := io.Copy(target, source); err != nil {381+ return err382+ }383+ return target.Sync()384+}385+func (s *Service) recordMultipartRanges(ctx context.Context, id string, ranges []multipartRange) error {386+ return s.store.WithTx(ctx, func(tx *sql.Tx) error {387+ if _, err := tx.ExecContext(ctx, "DELETE FROM multipart_parts WHERE upload_id=?", id); err != nil {388+ return err389+ }390+ for _, span := range ranges {391+ if _, err := tx.ExecContext(ctx, "INSERT INTO multipart_parts(upload_id,offset,length) VALUES(?,?,?)", id, span.start, span.end-span.start); err != nil {392+ return err393+ }394+ }395+ _, err := tx.ExecContext(ctx, "UPDATE multipart_uploads SET last_seen=? WHERE id=?", time.Now().Unix(), id)396+ return err397+ })398+}399+func verifyMultipart(path, hash string, size int64) error {400+ file, err := os.Open(path)401+ if err != nil {402+ return err403+ }404+ defer file.Close()405+ hasher := sha256.New()406+ count, err := io.Copy(hasher, file)407+ if err != nil {408+ return err409+ }410+ if count != size || hex.EncodeToString(hasher.Sum(nil)) != hash {411+ return ErrHashMismatch412+ }413+ return nil414+}415+func (s *Service) discardMultipart(ctx context.Context, id string) error {416+ _, err := s.store.DB().ExecContext(ctx, "DELETE FROM multipart_uploads WHERE id=?", id)417+ if err != nil {418+ return err419+ }420+ if err := os.Remove(s.multipartPath(id)); err != nil && !errors.Is(err, os.ErrNotExist) {421+ return err422+ }423+ return nil424+}425+426+// CleanupMultipart releases inactive reservations and removes partial bytes.427+// The tenant maintenance scheduler calls it even when no uploads arrive.428+func (s *Service) CleanupMultipart(ctx context.Context) error {429+ s.multipartMu.Lock()430+ defer s.multipartMu.Unlock()431+ s.quotaMu.Lock()432+ defer s.quotaMu.Unlock()433+ return s.cleanupMultipartLocked(ctx)434+}435+func (s *Service) cleanupMultipartLocked(ctx context.Context) error {436+ rows, err := s.store.DB().QueryContext(ctx, "SELECT id FROM multipart_uploads WHERE last_seen <= ?", time.Now().Add(-multipartExpiry).Unix())437+ if err != nil {438+ return err439+ }440+ var expired []string441+ for rows.Next() {442+ var id string443+ if err := rows.Scan(&id); err != nil {444+ _ = rows.Close()445+ return err446+ }447+ if s.activeParts[id] == 0 {448+ expired = append(expired, id)449+ }450+ }451+ err = errors.Join(rows.Err(), rows.Close())452+ if err != nil {453+ return err454+ }455+ for _, id := range expired {456+ if err := s.discardMultipart(ctx, id); err != nil {457+ return err458+ }459+ }460+ return nil461+}462+func (s *Service) reconcileMultipart(ctx context.Context) error {463+ if err := s.CleanupMultipart(ctx); err != nil {464+ return err465+ }466+ entries, err := os.ReadDir(s.root)467+ if err != nil {468+ return err469+ }470+ for _, entry := range entries {471+ if !strings.HasPrefix(entry.Name(), ".multipart-") || entry.IsDir() {472+ continue473+ }474+ id := strings.TrimPrefix(entry.Name(), ".multipart-")475+ var length int64476+ err := s.store.DB().QueryRowContext(ctx, "SELECT length FROM multipart_uploads WHERE id=?", id).Scan(&length)477+ if errors.Is(err, sql.ErrNoRows) {478+ if err := os.Remove(filepath.Join(s.root, entry.Name())); err != nil {479+ return err480+ }481+ continue482+ }483+ if err != nil {484+ return err485+ }486+ info, err := entry.Info()487+ if err != nil {488+ return err489+ }490+ if info.Size() != length {491+ if err := s.discardMultipart(ctx, id); err != nil {492+ return err493+ }494+ }495+ }496+ // A reservation with a missing file cannot represent acknowledged bytes.497+ rows, err := s.store.DB().QueryContext(ctx, "SELECT id FROM multipart_uploads")498+ if err != nil {499+ return err500+ }501+ var missing []string502+ for rows.Next() {503+ var id string504+ if err := rows.Scan(&id); err != nil {505+ _ = rows.Close()506+ return err507+ }508+ if _, err := os.Stat(s.multipartPath(id)); errors.Is(err, os.ErrNotExist) {509+ missing = append(missing, id)510+ } else if err != nil {511+ _ = rows.Close()512+ return err513+ }514+ }515+ err = errors.Join(rows.Err(), rows.Close())516+ if err != nil {517+ return err518+ }519+ for _, id := range missing {520+ if err := s.discardMultipart(ctx, id); err != nil {521+ return fmt.Errorf("reconcile multipart upload: %w", err)522+ }523+ }524+ return nil525+}@@ -0,0 +1,192 @@
1+package blob2+3+import (4+ "context"5+ "crypto/sha256"6+ "encoding/hex"7+ "errors"8+ "io"9+ "net/http"10+ "net/http/httptest"11+ "path/filepath"12+ "strings"13+ "sync"14+ "testing"15+ "time"16+)17+18+func multipartReviewRequest(sha string, length, offset int64, body string) *http.Request {19+ r := httptest.NewRequest(http.MethodPatch, "https://relay.test/"+sha, strings.NewReader(body))20+ r.Header.Set("Content-Type", "application/octet-stream")21+ r.Header.Set("Upload-Type", "application/octet-stream")22+ r.Header.Set("Upload-Length", formatInt(length))23+ r.Header.Set("Upload-Offset", formatInt(offset))24+ return r25+}26+27+func TestMultipartReviewConcurrentChunksProduceOneBlob(t *testing.T) {28+ s := testService(t)29+ body := []byte(strings.Repeat("a", 70000) + strings.Repeat("b", 70000))30+ digest := sha256.Sum256(body)31+ sha := hex.EncodeToString(digest[:])32+ parts := []struct {33+ offset int6434+ body string35+ }{36+ {70000, string(body[70000:])},37+ {0, string(body[:70000])},38+ }39+ responses := make([]*httptest.ResponseRecorder, len(parts))40+ var wg sync.WaitGroup41+ for i, part := range parts {42+ wg.Add(1)43+ go func() {44+ defer wg.Done()45+ responses[i] = record(s.Handler(), multipartReviewRequest(sha, int64(len(body)), part.offset, part.body))46+ }()47+ }48+ wg.Wait()49+ var created, partial int50+ for _, response := range responses {51+ switch response.Code {52+ case http.StatusCreated:53+ created++54+ case http.StatusNoContent:55+ partial++56+ default:57+ t.Fatalf("concurrent chunk status = %d: %s", response.Code, response.Body.String())58+ }59+ }60+ if created != 1 || partial != 1 {61+ t.Fatalf("concurrent chunk results created=%d partial=%d", created, partial)62+ }63+ entry, file, err := s.Get(context.Background(), sha)64+ if err != nil {65+ t.Fatal(err)66+ }67+ defer file.Close()68+ if entry.Size != int64(len(body)) {69+ t.Fatalf("stored size = %d, want %d", entry.Size, len(body))70+ }71+ got, err := io.ReadAll(file)72+ if err != nil || string(got) != string(body) {73+ t.Fatalf("stored body mismatch: %v", err)74+ }75+}76+77+func TestMultipartReviewStalledBodyDoesNotBlockOrdinaryUpload(t *testing.T) {78+ s := testService(t)79+ body := "stalled"80+ digest := sha256.Sum256([]byte(body))81+ sha := hex.EncodeToString(digest[:])82+ started := make(chan struct{})83+ release := make(chan struct{})84+ r := multipartReviewRequest(sha, int64(len(body)), 0, body)85+ r.Body = &multipartBlockingBody{Reader: strings.NewReader(body), started: started, release: release}86+ r.ContentLength = int64(len(body))87+ result := make(chan *httptest.ResponseRecorder, 1)88+ go func() { result <- record(s.Handler(), r) }()89+ select {90+ case <-started:91+ case <-time.After(time.Second):92+ t.Fatal("multipart body was not read")93+ }94+ if _, err := s.Put(context.Background(), PutOptions{Reader: strings.NewReader("ordinary"), Type: "text/plain", Uploader: "uploader"}); err != nil {95+ t.Fatalf("ordinary upload blocked by stalled multipart body: %v", err)96+ }97+ close(release)98+ select {99+ case response := <-result:100+ if response.Code != http.StatusCreated {101+ t.Fatalf("multipart completion status = %d", response.Code)102+ }103+ case <-time.After(time.Second):104+ t.Fatal("stalled multipart request did not finish")105+ }106+}107+108+type multipartBlockingBody struct {109+ io.Reader110+ started chan<- struct{}111+ release <-chan struct{}112+ once sync.Once113+}114+115+func (b *multipartBlockingBody) Read(p []byte) (int, error) {116+ b.once.Do(func() { close(b.started) })117+ <-b.release118+ return b.Reader.Read(p)119+}120+121+func (b *multipartBlockingBody) Close() error { return nil }122+123+func TestMultipartReviewReservationsBlockUploadsAndReleaseOnExpiry(t *testing.T) {124+ s := testService(t)125+ s.config.Limits = func() Limits { return Limits{UserStorageBytes: 8} }126+ full := "12345678"127+ digest := sha256.Sum256([]byte(full))128+ sha := hex.EncodeToString(digest[:])129+ partial := multipartReviewRequest(sha, int64(len(full)), 0, full[:4])130+ partial.ContentLength = 4131+ if response := record(s.Handler(), partial); response.Code != http.StatusNoContent {132+ t.Fatalf("partial reservation status = %d: %s", response.Code, response.Body.String())133+ }134+ if _, err := s.Put(context.Background(), PutOptions{Reader: strings.NewReader("1234"), Type: "text/plain", Uploader: "uploader"}); !errors.Is(err, ErrQuotaExceeded) {135+ t.Fatalf("ordinary upload after reservation = %v", err)136+ }137+ other := "abcdefgh"138+ otherDigest := sha256.Sum256([]byte(other))139+ second := multipartReviewRequest(hex.EncodeToString(otherDigest[:]), int64(len(other)), 0, other[:4])140+ second.ContentLength = 4141+ if response := record(s.Handler(), second); response.Code != http.StatusForbidden {142+ t.Fatalf("second reserved upload status = %d", response.Code)143+ }144+ if _, err := s.store.DB().Exec("UPDATE multipart_uploads SET last_seen=0"); err != nil {145+ t.Fatal(err)146+ }147+ if err := s.CleanupMultipart(context.Background()); err != nil {148+ t.Fatal(err)149+ }150+ matches, err := filepath.Glob(filepath.Join(s.root, ".multipart-*"))151+ if err != nil {152+ t.Fatal(err)153+ }154+ if len(matches) != 0 {155+ t.Fatalf("expired multipart files remain: %v", matches)156+ }157+ if _, err := s.Put(context.Background(), PutOptions{Reader: strings.NewReader("1234"), Type: "text/plain", Uploader: "uploader"}); err != nil {158+ t.Fatalf("ordinary upload after expiry: %v", err)159+ }160+}161+162+func TestMultipartReviewHashMismatchReleasesReservation(t *testing.T) {163+ s := testService(t)164+ s.config.Limits = func() Limits { return Limits{UserStorageBytes: 5} }165+ wrong := strings.Repeat("a", 64)166+ r := multipartReviewRequest(wrong, 5, 0, "wrong")167+ r.ContentLength = 5168+ if response := record(s.Handler(), r); response.Code != http.StatusConflict {169+ t.Fatalf("hash mismatch status = %d: %s", response.Code, response.Body.String())170+ }171+ if _, err := s.Put(context.Background(), PutOptions{Reader: strings.NewReader("right"), Type: "text/plain", Uploader: "uploader"}); err != nil {172+ t.Fatalf("upload after mismatch reservation: %v", err)173+ }174+}175+176+func TestMultipartReviewExistingBlobCannotBypassModerationBlock(t *testing.T) {177+ s := testService(t)178+ body := "blocked"179+ digest := sha256.Sum256([]byte(body))180+ sha := hex.EncodeToString(digest[:])181+ if _, err := s.Put(context.Background(), PutOptions{Reader: strings.NewReader(body), Type: "text/plain"}); err != nil {182+ t.Fatal(err)183+ }184+ if _, err := s.store.DB().Exec("INSERT INTO blob_blocks(sha256,reason,blocked_at) VALUES(?,?,?)", sha, "review", time.Now().Unix()); err != nil {185+ t.Fatal(err)186+ }187+ r := multipartReviewRequest(sha, int64(len(body)), 0, body)188+ r.ContentLength = int64(len(body))189+ if response := record(s.Handler(), r); response.Code != http.StatusForbidden {190+ t.Fatalf("blocked existing blob status = %d: %s", response.Code, response.Body.String())191+ }192+}@@ -0,0 +1,171 @@
1+package blob2+3+import (4+ "context"5+ "crypto/sha256"6+ "encoding/hex"7+ "io"8+ "net/http"9+ "net/http/httptest"10+ "strconv"11+ "strings"12+ "testing"13+)14+15+func TestMultipartAcceptsOutOfOrderLargeChunks(t *testing.T) {16+ s := testService(t)17+ body := []byte(strings.Repeat("a", 70000) + strings.Repeat("b", 70000))18+ digest := sha256.Sum256(body)19+ sha := hex.EncodeToString(digest[:])20+ patch := func(offset int64, data []byte) *httptest.ResponseRecorder {21+ r := httptest.NewRequest(http.MethodPatch, "https://relay.test/"+sha+".bin", strings.NewReader(string(data)))22+ r.Header.Set("Content-Type", "application/octet-stream")23+ r.Header.Set("Upload-Type", "application/octet-stream")24+ r.Header.Set("Upload-Length", "140000")25+ r.Header.Set("Upload-Offset", formatInt(offset))26+ return record(s.Handler(), r)27+ }28+ if got := patch(70000, body[70000:]); got.Code != http.StatusNoContent {29+ t.Fatalf("second chunk status = %d", got.Code)30+ }31+ if got := patch(0, body[:70000]); got.Code != http.StatusCreated {32+ t.Fatalf("first chunk status = %d: %s", got.Code, got.Body.String())33+ }34+ entry, file, err := s.Get(context.Background(), sha)35+ if err != nil || entry.Size != int64(len(body)) {36+ t.Fatalf("stored multipart blob = %#v, %v", entry, err)37+ }38+ defer file.Close()39+}40+41+func formatInt(value int64) string { return strconv.FormatInt(value, 10) }42+43+func TestMultipartRejectsMalformedRangesAndLength(t *testing.T) {44+ s := testService(t)45+ digest := sha256.Sum256([]byte("abc"))46+ sha := hex.EncodeToString(digest[:])47+ tests := []struct {48+ name string49+ contentLength int6450+ uploadLength string51+ offset string52+ want int53+ }{54+ {"missing content length", -1, "3", "0", http.StatusLengthRequired},55+ {"offset outside upload", 1, "3", "3", http.StatusRequestedRangeNotSatisfiable},56+ {"chunk crosses upload", 2, "3", "2", http.StatusRequestedRangeNotSatisfiable},57+ }58+ for _, tc := range tests {59+ t.Run(tc.name, func(t *testing.T) {60+ r := httptest.NewRequest(http.MethodPatch, "https://relay.test/"+sha, strings.NewReader("ab"))61+ r.ContentLength = tc.contentLength62+ r.Header.Set("Content-Type", "application/octet-stream")63+ r.Header.Set("Upload-Type", "text/plain")64+ r.Header.Set("Upload-Length", tc.uploadLength)65+ r.Header.Set("Upload-Offset", tc.offset)66+ if got := record(s.Handler(), r).Code; got != tc.want {67+ t.Fatalf("status = %d, want %d", got, tc.want)68+ }69+ })70+ }71+}72+73+func TestMultipartMismatchDoesNotPublishBlob(t *testing.T) {74+ s := testService(t)75+ want := strings.Repeat("0", 64)76+ r := httptest.NewRequest(http.MethodPatch, "https://relay.test/"+want, strings.NewReader("wrong"))77+ r.Header.Set("Content-Type", "application/octet-stream")78+ r.Header.Set("Upload-Type", "text/plain")79+ r.Header.Set("Upload-Length", "5")80+ r.Header.Set("Upload-Offset", "0")81+ if got := record(s.Handler(), r).Code; got != http.StatusConflict {82+ t.Fatalf("status = %d, want 409", got)83+ }84+ if _, _, err := s.Get(context.Background(), want); err == nil {85+ t.Fatal("mismatched multipart upload was published")86+ }87+}88+89+func TestMultipartOverlappingChunksReconstructExactBytes(t *testing.T) {90+ s := testService(t)91+ body := []byte("abcdefghij")92+ digest := sha256.Sum256(body)93+ sha := hex.EncodeToString(digest[:])94+ request := func(offset int, chunk string) int {95+ r := httptest.NewRequest(http.MethodPatch, "https://relay.test/"+sha, strings.NewReader(chunk))96+ r.Header.Set("Content-Type", "application/octet-stream")97+ r.Header.Set("Upload-Type", "text/plain")98+ r.Header.Set("Upload-Length", "10")99+ r.Header.Set("Upload-Offset", formatInt(int64(offset)))100+ return record(s.Handler(), r).Code101+ }102+ if got := request(0, "abcdef"); got != http.StatusNoContent {103+ t.Fatalf("first status = %d", got)104+ }105+ if got := request(4, "efghij"); got != http.StatusCreated {106+ t.Fatalf("overlap status = %d", got)107+ }108+ _, file, err := s.Get(context.Background(), sha)109+ if err != nil {110+ t.Fatal(err)111+ }112+ defer file.Close()113+ got, err := io.ReadAll(file)114+ if err != nil || string(got) != string(body) {115+ t.Fatalf("body = %q, %v", got, err)116+ }117+}118+119+func TestMultipartResumesAfterServiceRestart(t *testing.T) {120+ original := testService(t)121+ body := "survives restart"122+ digest := sha256.Sum256([]byte(body))123+ hash := hex.EncodeToString(digest[:])124+ patch := func(service *Service, offset int, chunk string) *httptest.ResponseRecorder {125+ request := httptest.NewRequest(http.MethodPatch, "https://relay.test/"+hash, strings.NewReader(chunk))126+ request.Header.Set("Content-Type", "application/octet-stream")127+ request.Header.Set("Upload-Type", "text/plain")128+ request.Header.Set("Upload-Length", strconv.Itoa(len(body)))129+ request.Header.Set("Upload-Offset", strconv.Itoa(offset))130+ return record(service.Handler(), request)131+ }132+ if result := patch(original, 0, body[:5]); result.Code != 204 {133+ t.Fatalf("partial: %d %s", result.Code, result.Body.String())134+ }135+ reopened, err := New(context.Background(), original.config)136+ if err != nil {137+ t.Fatal(err)138+ }139+ if result := patch(reopened, 5, body[5:]); result.Code != 201 {140+ t.Fatalf("resumed: %d %s", result.Code, result.Body.String())141+ }142+}143+144+func TestMultipartAcceptsEmptyFile(t *testing.T) {145+ service := testService(t)146+ digest := sha256.Sum256(nil)147+ request := httptest.NewRequest(http.MethodPatch, "https://relay.test/"+hex.EncodeToString(digest[:]), http.NoBody)148+ request.Header.Set("Content-Type", "application/octet-stream")149+ request.Header.Set("Upload-Type", "application/octet-stream")150+ request.Header.Set("Upload-Length", "0")151+ request.Header.Set("Upload-Offset", "0")152+ if result := record(service.Handler(), request); result.Code != 201 {153+ t.Fatalf("empty: %d %s", result.Code, result.Body.String())154+ }155+}156+157+func TestMultipartExistingBlobAddsUploaderClaim(t *testing.T) {158+ service := testService(t)159+ const content = "deduplicated"160+ entry, err := service.Put(context.Background(), PutOptions{Reader: strings.NewReader(content), Uploader: "first"})161+ if err != nil {162+ t.Fatal(err)163+ }164+ request := multipartReviewRequest(entry.SHA256, int64(len(content)), 0, content[:2])165+ if result := record(service.Handler(), request); result.Code != 200 {166+ t.Fatalf("duplicate: %d %s", result.Code, result.Body.String())167+ }168+ if !service.hasClaim(context.Background(), entry.SHA256, "uploader") {169+ t.Fatal("existing blob did not grant the uploader a claim")170+ }171+}@@ -0,0 +1,128 @@
1+package blob2+3+import (4+ "context"5+ "errors"6+ "fmt"7+ "io"8+ "net/http"9+ "net/url"10+)11+12+// maxPathUploadSources bounds the remote sources one BUD-13 upload may name.13+const maxPathUploadSources = 814+15+// pathUpload implements the draft BUD-13 path-addressed upload endpoint.16+// The route itself is dispatched by serveHTTP after validating the hash path.17+func (s *Service) pathUpload(w http.ResponseWriter, r *http.Request, expectedHash string) {18+ // Authenticate before inspecting the body. This is important for signed19+ // uploads: an invalid request must not consume a potentially large stream.20+ pubkey, err := s.authorize(r, ActionUpload)21+ if err != nil {22+ s.fail(w, http.StatusUnauthorized, err)23+ return24+ }25+26+ urls := r.URL.Query()["url"]27+ if len(urls) == 0 {28+ entry, created, putErr := s.putValidated(r.Context(), r.Body,29+ contentType(r.Header.Get("content-type")), pubkey, expectedHash,30+ func(actual string) error {31+ if s.config.ValidateUpload != nil {32+ return s.config.ValidateUpload(r, actual)33+ }34+ return nil35+ })36+ if putErr != nil {37+ s.fail(w, statusFor(putErr), putErr)38+ return39+ }40+ s.writeJSON(w, chooseStatus(created, http.StatusCreated), s.descriptor(r, entry))41+ return42+ }43+44+ // BUD-13 remote uploads must have an empty request body. Check this after45+ // authorization, including chunked requests whose ContentLength is unknown.46+ if r.Body != nil {47+ var probe [1]byte48+ n, readErr := io.ReadFull(r.Body, probe[:])49+ if n != 0 || (readErr != nil && !errors.Is(readErr, io.EOF)) {50+ s.fail(w, http.StatusBadRequest, errors.New("invalid: remote upload body must be empty"))51+ return52+ }53+ }54+ if len(urls) > maxPathUploadSources {55+ s.fail(w, http.StatusBadRequest, fmt.Errorf("invalid: at most %d remote sources per upload", maxPathUploadSources))56+ return57+ }58+ for _, raw := range urls {59+ if err := s.validatePathUploadURL(r.Context(), raw); err != nil {60+ s.fail(w, http.StatusBadRequest, err)61+ return62+ }63+ }64+65+ client := s.config.HTTPClient66+ if client == nil {67+ client = s.safeClient()68+ }69+ var mismatch bool70+ for _, raw := range urls {71+ fetchRequest, requestErr := http.NewRequestWithContext(r.Context(), http.MethodGet, raw, nil)72+ if requestErr != nil {73+ continue74+ }75+ response, fetchErr := client.Do(fetchRequest)76+ if fetchErr != nil {77+ continue78+ }79+ if response.StatusCode < 200 || response.StatusCode >= 300 {80+ _ = response.Body.Close()81+ continue82+ }83+ typ := contentType(response.Header.Get("content-type"))84+ if typ == "application/octet-stream" {85+ typ = typeForExtension(raw)86+ }87+ entry, created, putErr := s.putValidated(r.Context(), response.Body, typ,88+ pubkey, expectedHash, func(actual string) error {89+ if s.config.ValidateUpload != nil {90+ return s.config.ValidateUpload(r, actual)91+ }92+ return nil93+ })94+ _ = response.Body.Close()95+ if putErr == nil {96+ s.writeJSON(w, chooseStatus(created, http.StatusCreated), s.descriptor(r, entry))97+ return98+ }99+ if errors.Is(putErr, ErrHashMismatch) {100+ mismatch = true101+ continue102+ }103+ s.fail(w, statusFor(putErr), putErr)104+ return105+ }106+ if mismatch {107+ s.fail(w, http.StatusConflict, ErrHashMismatch)108+ return109+ }110+ s.fail(w, http.StatusBadGateway, errors.New("error: no usable remote source"))111+}112+113+func (s *Service) validatePathUploadURL(ctx context.Context, raw string) error {114+ u, err := url.Parse(raw)115+ if err != nil || u == nil || (u.Scheme != "http" && u.Scheme != "https") || u.Hostname() == "" || u.User != nil || u.Fragment != "" {116+ return errors.New("invalid: url must be an absolute http or https url")117+ }118+ ips, err := s.resolve(ctx, u.Hostname())119+ if err != nil || len(ips) == 0 {120+ return errors.New("invalid: origin hostname did not resolve")121+ }122+ for _, ip := range ips {123+ if !publicIP(ip) {124+ return errors.New("invalid: origin resolves to a private address")125+ }126+ }127+ return nil128+}@@ -0,0 +1,131 @@
1+package blob2+3+import (4+ "context"5+ "crypto/sha256"6+ "encoding/hex"7+ "errors"8+ "io"9+ "net"10+ "net/http"11+ "net/http/httptest"12+ "strings"13+ "testing"14+)15+16+func TestPathUploadStoresExactBytesAndReturnsCreatedThenOK(t *testing.T) {17+ s := testService(t)18+ body := []byte("BUD-13 bytes")19+ digest := sha256.Sum256(body)20+ sha := hex.EncodeToString(digest[:])21+ request := httptest.NewRequest(http.MethodPut, "https://relay.test/"+sha+".txt", strings.NewReader(string(body)))22+ request.Header.Set("Content-Type", "text/plain")23+ first := record(s.Handler(), request)24+ if first.Code != http.StatusCreated || !strings.Contains(first.Body.String(), sha) {25+ t.Fatalf("first path upload = %d %s", first.Code, first.Body.String())26+ }27+ second := record(s.Handler(), httptest.NewRequest(http.MethodPut, "https://relay.test/"+sha, strings.NewReader(string(body))))28+ if second.Code != http.StatusOK {29+ t.Fatalf("duplicate path upload = %d %s", second.Code, second.Body.String())30+ }31+}32+33+func TestPathUploadRejectsHashMismatchWithoutPersisting(t *testing.T) {34+ s := testService(t)35+ wanted := strings.Repeat("a", 64)36+ response := record(func(w http.ResponseWriter, r *http.Request) { s.pathUpload(w, r, wanted) }, httptest.NewRequest(http.MethodPut, "https://relay.test/"+wanted, strings.NewReader("wrong")))37+ if response.Code != http.StatusConflict {38+ t.Fatalf("mismatch status = %d, want 409", response.Code)39+ }40+ if _, _, err := s.Get(context.Background(), wanted); err == nil {41+ t.Fatal("mismatched path upload was persisted")42+ }43+}44+45+func TestPathUploadAuthenticatesBeforeReadingBody(t *testing.T) {46+ s := testService(t)47+ read := false48+ s.config.Authorize = func(*http.Request, Action) (string, error) { return "", errors.New("bad auth") }49+ body := &trackingReader{Reader: strings.NewReader("should not be read"), read: &read}50+ sha := strings.Repeat("b", 64)51+ response := record(func(w http.ResponseWriter, r *http.Request) { s.pathUpload(w, r, sha) }, httptest.NewRequest(http.MethodPut, "https://relay.test/"+sha, body))52+ if response.Code != http.StatusUnauthorized || read {53+ t.Fatalf("auth status/read = %d/%v", response.Code, read)54+ }55+}56+57+func TestPathUploadRejectsRemoteBodyAndMalformedURL(t *testing.T) {58+ s := testService(t)59+ sha := strings.Repeat("c", 64)60+ request := httptest.NewRequest(http.MethodPut, "https://relay.test/"+sha+"?url=https%3A%2F%2Fexample.com%2Fblob", strings.NewReader("body"))61+ response := record(func(w http.ResponseWriter, r *http.Request) { s.pathUpload(w, r, sha) }, request)62+ if response.Code != http.StatusBadRequest {63+ t.Fatalf("remote body status = %d, want 400", response.Code)64+ }65+ request = httptest.NewRequest(http.MethodPut, "https://relay.test/"+sha+"?url=file%3A%2F%2F%2Fetc%2Fpasswd", http.NoBody)66+ response = record(func(w http.ResponseWriter, r *http.Request) { s.pathUpload(w, r, sha) }, request)67+ if response.Code != http.StatusBadRequest {68+ t.Fatalf("malformed URL status = %d, want 400", response.Code)69+ }70+ request = httptest.NewRequest(http.MethodPut, "https://relay.test/"+sha+"?url=https%3A%2F%2Fexample.com%2Fx%23fragment", http.NoBody)71+ response = record(func(w http.ResponseWriter, r *http.Request) { s.pathUpload(w, r, sha) }, request)72+ if response.Code != http.StatusBadRequest {73+ t.Fatalf("fragment URL status = %d, want 400", response.Code)74+ }75+}76+77+func TestPathUploadRemoteSourceVerifiesHashAndSupportsHTTP(t *testing.T) {78+ s := testService(t)79+ body := []byte("remote BUD-13 bytes")80+ digest := sha256.Sum256(body)81+ sha := hex.EncodeToString(digest[:])82+ s.config.ResolveIP = func(context.Context, string) ([]net.IP, error) { return []net.IP{net.ParseIP("8.8.8.8")}, nil }83+ s.resolve = s.config.ResolveIP84+ s.config.HTTPClient = &http.Client{Transport: roundTripFunc(func(req *http.Request) (*http.Response, error) {85+ return &http.Response{StatusCode: http.StatusOK, Body: io.NopCloser(strings.NewReader(string(body))), Header: http.Header{"Content-Type": []string{"text/plain"}}, Request: req}, nil86+ })}87+ request := httptest.NewRequest(http.MethodPut, "https://relay.test/"+sha+"?url=http%3A%2F%2Forigin.example%2Ffile.txt", http.NoBody)88+ response := record(func(w http.ResponseWriter, r *http.Request) { s.pathUpload(w, r, sha) }, request)89+ if response.Code != http.StatusCreated || !strings.Contains(response.Body.String(), sha) {90+ t.Fatalf("remote upload = %d %s", response.Code, response.Body.String())91+ }92+}93+94+func TestPathUploadRemoteRejectsPrivateOrigin(t *testing.T) {95+ s := testService(t)96+ s.config.ResolveIP = func(context.Context, string) ([]net.IP, error) { return []net.IP{net.ParseIP("127.0.0.1")}, nil }97+ s.resolve = s.config.ResolveIP98+ response := record(func(w http.ResponseWriter, r *http.Request) { s.pathUpload(w, r, strings.Repeat("d", 64)) }, httptest.NewRequest(http.MethodPut, "https://relay.test/"+strings.Repeat("d", 64)+"?url=http%3A%2F%2Forigin.example%2Fx", http.NoBody))99+ if response.Code != http.StatusBadRequest {100+ t.Fatalf("private origin status = %d, want 400", response.Code)101+ }102+}103+104+func TestPathUploadRemoteUsesRequestContext(t *testing.T) {105+ s := testService(t)106+ sha := strings.Repeat("e", 64)107+ s.config.ResolveIP = func(ctx context.Context, _ string) ([]net.IP, error) {108+ return []net.IP{net.ParseIP("8.8.8.8")}, nil109+ }110+ s.resolve = s.config.ResolveIP111+ s.config.HTTPClient = &http.Client{Transport: roundTripFunc(func(req *http.Request) (*http.Response, error) {112+ return nil, req.Context().Err()113+ })}114+ ctx, cancel := context.WithCancel(context.Background())115+ cancel()116+ request := httptest.NewRequest(http.MethodPut, "https://relay.test/"+sha+"?url=https%3A%2F%2Forigin.example%2Ffile", http.NoBody).WithContext(ctx)117+ response := record(s.Handler(), request)118+ if response.Code != http.StatusBadGateway {119+ t.Fatalf("cancelled remote upload = %d, want 502", response.Code)120+ }121+}122+123+type trackingReader struct {124+ io.Reader125+ read *bool126+}127+128+func (r *trackingReader) Read(p []byte) (int, error) {129+ *r.read = true130+ return r.Reader.Read(p)131+}@@ -0,0 +1,103 @@
1+package daemon2+3+import (4+ "context"5+ "crypto/sha256"6+ "encoding/base64"7+ "encoding/hex"8+ "encoding/json"9+ "io"10+ "net"11+ "net/http"12+ "net/http/httptest"13+ "net/url"14+ "strconv"15+ "strings"16+ "testing"17+ "time"18+19+ "github.com/FelineStateMachine/tinyrelay/internal/blob"20+ "github.com/FelineStateMachine/tinyrelay/internal/event"21+)22+23+type blobOriginTransport func(*http.Request) (*http.Response, error)24+25+func (f blobOriginTransport) RoundTrip(r *http.Request) (*http.Response, error) { return f(r) }26+27+func TestBlobMirrorAuthenticationBindsRequestAndFetchedContent(t *testing.T) {28+ const content = "fetched encrypted bytes"29+ sum := sha256.Sum256([]byte(content))30+ hash := hex.EncodeToString(sum[:])31+ for _, tc := range []struct {32+ name, mode, action, scope string33+ pathUpload, tamper bool34+ want int35+ }{36+ {name: "mirror NIP98", mode: "nip98", want: 201},37+ {name: "mirror NIP98 tampered JSON", mode: "nip98", tamper: true, want: 401},38+ {name: "mirror Blossom upload verb", action: "upload", scope: hash, want: 201},39+ {name: "mirror Blossom wrong verb", action: "mirror", scope: hash, want: 401},40+ {name: "mirror Blossom wrong content", action: "upload", scope: strings.Repeat("a", 64), want: 401},41+ {name: "remote NIP98", mode: "nip98", pathUpload: true, want: 201},42+ {name: "remote Blossom", action: "upload", scope: hash, pathUpload: true, want: 201},43+ } {44+ t.Run(tc.name, func(t *testing.T) {45+ _, tenant := testTenant(t)46+ fetches := 047+ service, err := blob.New(context.Background(), blob.Config{48+ Root: t.TempDir(), Store: tenant.store, PublicURL: tenant.publicURL,49+ Authorize: tenant.authorizeBlob, ValidateUpload: tenant.validateBlobUpload,50+ ResolveIP: func(context.Context, string) ([]net.IP, error) { return []net.IP{net.ParseIP("8.8.8.8")}, nil },51+ HTTPClient: &http.Client{Transport: blobOriginTransport(func(*http.Request) (*http.Response, error) {52+ fetches++53+ return &http.Response{StatusCode: 200, Header: http.Header{"Content-Type": []string{"application/octet-stream"}}, Body: io.NopCloser(strings.NewReader(content))}, nil54+ })},55+ })56+ if err != nil {57+ t.Fatal(err)58+ }59+ body := `{"url":"https://files.example/source"}`60+ target := "http://relay.test/mirror"61+ if tc.pathUpload {62+ body = ""63+ target = "http://relay.test/" + hash + "?url=" + url.QueryEscape("https://files.example/source")64+ }65+ r := httptest.NewRequest(http.MethodPut, target, strings.NewReader(body))66+ if tc.mode == "nip98" {67+ signRequest(t, r, body)68+ } else {69+ token := event.Event{Kind: 24242, CreatedAt: time.Now().Unix(), Tags: [][]string{{"t", tc.action}, {"x", tc.scope}, {"expiration", strconv.FormatInt(time.Now().Unix()+60, 10)}}, Content: "Authorize file storage"}70+ if err := event.Sign(&token, strings.Repeat("0", 63)+"1"); err != nil {71+ t.Fatal(err)72+ }73+ raw, err := json.Marshal(token)74+ if err != nil {75+ t.Fatal(err)76+ }77+ r.Header.Set("Authorization", "Nostr "+base64.RawURLEncoding.EncodeToString(raw))78+ }79+ if tc.tamper {80+ r.Body = io.NopCloser(strings.NewReader(`{"url":"https://files.example/altered"}`))81+ }82+ w := httptest.NewRecorder()83+ service.Handler().ServeHTTP(w, r)84+ if w.Code != tc.want {85+ t.Fatalf("status=%d, want %d: %s", w.Code, tc.want, w.Body.String())86+ }87+ if (tc.tamper || tc.action == "mirror") && fetches != 0 {88+ t.Fatal("unauthorized request fetched an origin")89+ }90+ })91+ }92+}93+94+func TestUploadURLQueryCannotBypassNIP98PayloadValidation(t *testing.T) {95+ _, tenant := testTenant(t)96+ r := httptest.NewRequest(http.MethodPut, "http://relay.test/upload?url=https://files.example/ignored", strings.NewReader("tampered body"))97+ signRequest(t, r, "signed body")98+ w := httptest.NewRecorder()99+ tenant.ServeHTTP(w, r)100+ if w.Code != http.StatusUnauthorized {101+ t.Fatalf("tampered upload status=%d: %s", w.Code, w.Body.String())102+ }103+}@@ -0,0 +1,125 @@
1+package daemon2+3+import (4+ "crypto/sha256"5+ "encoding/base64"6+ "encoding/hex"7+ "encoding/json"8+ "net/http"9+ "net/http/httptest"10+ "strconv"11+ "strings"12+ "testing"13+ "time"14+15+ "github.com/FelineStateMachine/tinyrelay/internal/blob"16+ "github.com/FelineStateMachine/tinyrelay/internal/event"17+)18+19+func multipartAuthorization(t *testing.T, hash string) string {20+ t.Helper()21+ token := event.Event{Kind: 24242, CreatedAt: time.Now().Unix(), Tags: [][]string{22+ {"t", "upload"}, {"x", hash}, {"expiration", strconv.FormatInt(time.Now().Unix()+60, 10)},23+ }, Content: "Authorize multipart upload"}24+ if err := event.Sign(&token, strings.Repeat("0", 63)+"1"); err != nil {25+ t.Fatal(err)26+ }27+ raw, err := json.Marshal(token)28+ if err != nil {29+ t.Fatal(err)30+ }31+ return "Nostr " + base64.RawURLEncoding.EncodeToString(raw)32+}33+34+func TestMultipartAuthorizationBindsFinalHashBeforeReading(t *testing.T) {35+ _, tenant := testTenant(t)36+ finalHash := strings.Repeat("a", 64)37+ r := httptest.NewRequest(http.MethodPatch, "http://relay.test/"+finalHash+".txt", strings.NewReader("chunk"))38+ r.Header.Set("Authorization", multipartAuthorization(t, strings.Repeat("b", 64)))39+ if _, err := tenant.authorizeBlob(r, blob.ActionUpload); err == nil {40+ t.Fatal("multipart authorization accepted another final hash")41+ }42+ r.Header.Set("Authorization", multipartAuthorization(t, finalHash))43+ if _, err := tenant.authorizeBlob(r, blob.ActionUpload); err != nil {44+ t.Fatal(err)45+ }46+ chunkHash := sha256.Sum256([]byte("chunk"))47+ if err := tenant.validateBlobUpload(r, hex.EncodeToString(chunkHash[:])); err != nil {48+ t.Fatalf("Blossom proof incorrectly required a chunk hash: %v", err)49+ }50+}51+52+func TestMultipartNIP98BindsEachChunkBody(t *testing.T) {53+ _, tenant := testTenant(t)54+ r := httptest.NewRequest(http.MethodPatch, "http://relay.test/"+strings.Repeat("c", 64), strings.NewReader("chunk"))55+ signRequest(t, r, "chunk")56+ if _, err := tenant.authorizeBlob(r, blob.ActionUpload); err != nil {57+ t.Fatal(err)58+ }59+ chunkHash := sha256.Sum256([]byte("chunk"))60+ if err := tenant.validateBlobUpload(r, hex.EncodeToString(chunkHash[:])); err != nil {61+ t.Fatal(err)62+ }63+ if err := tenant.validateBlobUpload(r, strings.Repeat("c", 64)); err == nil {64+ t.Fatal("NIP-98 proof accepted final hash in place of chunk payload")65+ }66+}67+68+func TestMultipartPreflightAdvertisesMountedFileDoor(t *testing.T) {69+ _, tenant := testTenant(t)70+ r := httptest.NewRequest(http.MethodOptions, "http://relay.test/"+strings.Repeat("a", 64), nil)71+ w := httptest.NewRecorder()72+ tenant.ServeHTTP(w, r)73+ if w.Code != http.StatusNoContent || !strings.Contains(w.Header().Get("Allow"), "PATCH") {74+ t.Fatalf("multipart preflight: %d, Allow=%q", w.Code, w.Header().Get("Allow"))75+ }76+ for _, header := range []string{"upload-type", "upload-length", "upload-offset"} {77+ if !strings.Contains(strings.ToLower(w.Header().Get("Access-Control-Allow-Headers")), header) {78+ t.Fatalf("multipart CORS does not allow %s", header)79+ }80+ }81+ if !strings.Contains(w.Header().Get("Access-Control-Expose-Headers"), "Allow") {82+ t.Fatal("browser cannot read PATCH support")83+ }84+ for _, capability := range tenant.Capabilities(nil) {85+ if capability.ID == "BUD-14" && capability.Status == "enabled" {86+ return87+ }88+ }89+ t.Fatal("BUD-14 capability is missing")90+}91+92+func TestMultipartHTTPVerifiesEachNIP98ChunkBeforeAcceptingIt(t *testing.T) {93+ _, tenant := testTenant(t)94+ const body = "first and second"95+ digest := sha256.Sum256([]byte(body))96+ hash := hex.EncodeToString(digest[:])97+ patch := func(offset int, chunk, signed string) *httptest.ResponseRecorder {98+ r := httptest.NewRequest(http.MethodPatch, "http://relay.test/"+hash, strings.NewReader(chunk))99+ r.Header.Set("Content-Type", "application/octet-stream")100+ r.Header.Set("Upload-Type", "text/plain")101+ r.Header.Set("Upload-Length", strconv.Itoa(len(body)))102+ r.Header.Set("Upload-Offset", strconv.Itoa(offset))103+ signRequest(t, r, signed)104+ w := httptest.NewRecorder()105+ tenant.ServeHTTP(w, r)106+ return w107+ }108+ if w := patch(0, body[:6], "altered"); w.Code != http.StatusUnauthorized {109+ t.Fatalf("invalid chunk proof = %d: %s", w.Code, w.Body.String())110+ }111+ if w := patch(6, body[6:], body[6:]); w.Code != http.StatusNoContent {112+ t.Fatalf("out-of-order chunk = %d: %s", w.Code, w.Body.String())113+ }114+ if w := patch(0, body[:6], body[:6]); w.Code != http.StatusCreated {115+ t.Fatalf("complete upload = %d: %s", w.Code, w.Body.String())116+ }117+ if w := patch(0, body[:6], "still altered"); w.Code != http.StatusUnauthorized {118+ t.Fatalf("existing blob bypassed payload validation = %d: %s", w.Code, w.Body.String())119+ }120+ get := httptest.NewRecorder()121+ tenant.ServeHTTP(get, httptest.NewRequest(http.MethodGet, "http://relay.test/"+hash, nil))122+ if get.Code != http.StatusOK || get.Body.String() != body {123+ t.Fatalf("retrieved upload = %d: %s", get.Code, get.Body.String())124+ }125+}@@ -84,10 +84,13 @@ func CapabilityRegistryWithServices(p policy.Policy, gitAvailable, blobsAvailabl
8484 {ID: "BUD-01", Status: "enabled", Reason: "blob retrieval and basic server behavior"},8585 {ID: "BUD-02", Status: "enabled", Reason: "blob upload and management"},8686 {ID: "BUD-04", Status: "enabled", Reason: "blob mirroring"},87- {ID: "BUD-06", Status: "enabled", Reason: "blob existence and metadata checks"},87+ {ID: "BUD-06", Status: "enabled", Reason: "upload preflight and storage allowances"},88+ {ID: "BUD-08", Status: "enabled", Reason: "Nostr file metadata in blob descriptors"},8889 {ID: "BUD-09", Status: "enabled", Reason: "blob reports"},8990 {ID: "BUD-11", Status: "enabled", Reason: "Nostr authorization"},9091 {ID: "BUD-12", Status: "enabled", Reason: "cursor-paginated blob listing"},92+ {ID: "BUD-13", Status: "enabled", Reason: "draft hash-addressed uploads and remote sources"},93+ {ID: "BUD-14", Status: "enabled", Reason: "draft multipart uploads with bounded partial storage"},9194 {ID: "NIP-66", Status: "disabled", Reason: "operator peer liveness is not configured"},9295 }9396 if p.Features.Count {@@ -30,6 +30,9 @@ func TestCapabilityRegistryReportsHonestOptionalFeatures(t *testing.T) {
3030 if got["NIP-94"].Status != "enabled" || got["NIP-96"].Status != "enabled" {3131 t.Fatalf("file capabilities = %#v", got)3232 }33+ if got["BUD-08"].Status != "enabled" || got["BUD-13"].Status != "enabled" {34+ t.Fatalf("file metadata and upload capabilities = %#v", got)35+ }3336 }3437 3538 func TestPublishPeerStatusStoresSignedDiscoveryEvent(t *testing.T) {@@ -210,6 +210,29 @@ func (t *Tenant) browseFiles(ctx context.Context, actor string, q clientBrowseRe
210210 var next string211211 if role == "owner" || role == "moderator" {212212 entries, next, err = t.browseAllFiles(ctx, q)213+ } else if strings.TrimSpace(q.Query) != "" {214+ // ListPage is uploader scoped and cannot filter server-side. Walk its215+ // cursor until the requested number of matches is collected.216+ cursor := q.Cursor217+ for len(entries) < q.Limit {218+ var page []blob.Blob219+ page, next, err = t.blobs.ListPage(ctx, actor, q.Limit-len(entries), cursor)220+ if err != nil {221+ break222+ }223+ for _, entry := range page {224+ if blobMatchesQuery(entry, q.Query) {225+ entries = append(entries, entry)226+ if len(entries) == q.Limit {227+ break228+ }229+ }230+ }231+ if next == "" {232+ break233+ }234+ cursor = next235+ }213236 } else {214237 entries, next, err = t.blobs.ListPage(ctx, actor, q.Limit, q.Cursor)215238 }@@ -218,14 +241,25 @@ func (t *Tenant) browseFiles(ctx context.Context, actor string, q clientBrowseRe
218241 }219242 items := []map[string]any{}220243 for _, entry := range entries {244+ if q.Query != "" && !blobMatchesQuery(entry, q.Query) {245+ continue246+ }221247 items = append(items, t.browseBlobMetadata(entry))222248 }223249 return map[string]any{"items": items, "next_cursor": next}, nil224250 }225251 226252 func (t *Tenant) browseAllFiles(ctx context.Context, q clientBrowseRequest) ([]blob.Blob, string, error) {227- query := "SELECT sha256,size,type,uploader,uploaded FROM blobs WHERE sha256>? ORDER BY sha256 LIMIT ?"228- rows, err := t.store.DB().QueryContext(ctx, query, q.Cursor, q.Limit+1)253+ query := "SELECT sha256,size,type,uploader,uploaded FROM blobs WHERE sha256>?"254+ args := []any{q.Cursor}255+ if q.Query != "" {256+ query += " AND (instr(lower(sha256), ?) > 0 OR instr(lower(type), ?) > 0 OR instr(lower(uploader), ?) > 0)"257+ needle := strings.ToLower(strings.TrimSpace(q.Query))258+ args = append(args, needle, needle, needle)259+ }260+ query += " ORDER BY sha256 LIMIT ?"261+ args = append(args, q.Limit+1)262+ rows, err := t.store.DB().QueryContext(ctx, query, args...)229263 if err != nil {230264 return nil, "", err231265 }@@ -249,6 +283,16 @@ func (t *Tenant) browseAllFiles(ctx context.Context, q clientBrowseRequest) ([]b
249283 return entries, next, nil250284 }251285 286+func blobMatchesQuery(entry blob.Blob, query string) bool {287+ needle := strings.ToLower(strings.TrimSpace(query))288+ if needle == "" {289+ return true290+ }291+ return strings.Contains(strings.ToLower(entry.SHA256), needle) ||292+ strings.Contains(strings.ToLower(entry.Type), needle) ||293+ strings.Contains(strings.ToLower(entry.Uploader), needle)294+}295+252296 func (t *Tenant) browseBlobMetadata(entry blob.Blob) map[string]any {253297 portable := blob.BlossomURI{Hash: entry.SHA256, Extension: "bin", Author: entry.Uploader, Size: entry.Size}254298 if origin, err := url.Parse(t.publicURL); err == nil && strings.Trim(origin.Path, "/") == "" {@@ -3,6 +3,7 @@ package daemon
33 import (44 "context"55 "encoding/json"6+ "fmt"67 "net/http"78 "net/http/httptest"89 "strings"@@ -66,3 +67,67 @@ func TestClientStatusRequiresOwnerAndPreservesErrors(t *testing.T) {
6667 }6768 }6869 }70+71+func TestFileSearchPaginationPreservesEveryUploaderMatch(t *testing.T) {72+ _, tenant := testTenant(t)73+ ctx := context.Background()74+ uploader := strings.Repeat("b", 64)75+ var want []string76+ for i, typ := range []string{"text/plain", "image/png", "text/plain", "text/plain", "text/plain"} {77+ entry, err := tenant.blobs.Put(ctx, blob.PutOptions{Reader: strings.NewReader(fmt.Sprintf("file %d", i)), Type: typ, Uploader: uploader})78+ if err != nil {79+ t.Fatal(err)80+ }81+ if _, err := tenant.store.DB().ExecContext(ctx, "UPDATE blobs SET uploaded=? WHERE sha256=?", 100-i, entry.SHA256); err != nil {82+ t.Fatal(err)83+ }84+ if typ == "text/plain" {85+ want = append(want, entry.SHA256)86+ }87+ }88+ if _, err := tenant.blobs.Put(ctx, blob.PutOptions{Reader: strings.NewReader("another user's matching file"), Type: "text/plain", Uploader: tenant.Policy().Owner}); err != nil {89+ t.Fatal(err)90+ }91+ var got []string92+ cursor := ""93+ for page := 0; page < 5; page++ {94+ params, err := json.Marshal(map[string]any{"q": "text/plain", "limit": 2, "cursor": cursor})95+ if err != nil {96+ t.Fatal(err)97+ }98+ result, err := tenant.Execute(ctx, uploader, "browsefiles", []json.RawMessage{params})99+ if err != nil {100+ t.Fatal(err)101+ }102+ row := result.(map[string]any)103+ for _, item := range row["items"].([]map[string]any) {104+ got = append(got, item["sha256"].(string))105+ }106+ cursor = row["next_cursor"].(string)107+ if cursor == "" {108+ break109+ }110+ }111+ if fmt.Sprint(got) != fmt.Sprint(want) {112+ t.Fatalf("matching files = %v, want %v", got, want)113+ }114+}115+116+func TestOwnerFileSearchUsesLiteralTextBeforePagination(t *testing.T) {117+ _, tenant := testTenant(t)118+ ctx := context.Background()119+ for i, typ := range []string{"text/plain", "application/x_foo", "image/png"} {120+ if _, err := tenant.blobs.Put(ctx, blob.PutOptions{Reader: strings.NewReader(fmt.Sprintf("owner file %d", i)), Type: typ, Uploader: tenant.Policy().Owner}); err != nil {121+ t.Fatal(err)122+ }123+ }124+ result, err := tenant.Execute(ctx, tenant.Policy().Owner, "browsefiles", []json.RawMessage{json.RawMessage(`{"q":"_","limit":1}`)})125+ if err != nil {126+ t.Fatal(err)127+ }128+ row := result.(map[string]any)129+ items := row["items"].([]map[string]any)130+ if len(items) != 1 || items[0]["type"] != "application/x_foo" || row["next_cursor"] != "" {131+ t.Fatalf("literal search = %#v", row)132+ }133+}@@ -12,6 +12,7 @@ import (
1212 "testing"1313 "time"1414 15+ "github.com/FelineStateMachine/tinyrelay/internal/blob"1516 "github.com/FelineStateMachine/tinyrelay/internal/event"1617 )1718 @@ -82,6 +83,34 @@ func signRequest(t *testing.T, r *http.Request, body string) {
8283 r.Header.Set("Authorization", "Nostr "+base64.StdEncoding.EncodeToString(raw))8384 }8485 86+func TestBUD13RemoteNIP98AcceptsAbsentOrEmptyPayloadOnly(t *testing.T) {87+ _, tenant := testTenant(t)88+ sha := strings.Repeat("a", 64)89+ empty := sha256.Sum256(nil)90+ for name, payload := range map[string]string{"absent": "", "empty": hex.EncodeToString(empty[:]), "wrong": strings.Repeat("b", 64)} {91+ t.Run(name, func(t *testing.T) {92+ req := httptest.NewRequest(http.MethodPut, "http://relay.test/"+sha+"?url=https%3A%2F%2Forigin.example%2Fblob", http.NoBody)93+ tags := [][]string{{"u", req.URL.String()}, {"method", "PUT"}}94+ if payload != "" {95+ tags = append(tags, []string{"payload", payload})96+ }97+ e := event.Event{Kind: 27235, CreatedAt: time.Now().Unix(), Tags: tags}98+ if err := event.Sign(&e, strings.Repeat("0", 63)+"1"); err != nil {99+ t.Fatal(err)100+ }101+ raw, _ := json.Marshal(e)102+ req.Header.Set("Authorization", "Nostr "+base64.StdEncoding.EncodeToString(raw))103+ _, err := tenant.authorizeBlob(req, blob.ActionUpload)104+ if name == "wrong" && err == nil {105+ t.Fatal("accepted wrong empty-body payload")106+ }107+ if name != "wrong" && err != nil {108+ t.Fatalf("remote authorization: %v", err)109+ }110+ })111+ }112+}113+85114 func TestCreationRequiresOwnerAndInformationHasNoEconomy(t *testing.T) {86115 ctx := context.Background()87116 app, err := New(ctx, Config{DataDir: t.TempDir(), DefaultTenant: "main"})@@ -21,10 +21,15 @@ import (
2121 2222 func (t *Tenant) ServeHTTP(w http.ResponseWriter, r *http.Request) {2323 w.Header().Set("Access-Control-Allow-Origin", "*")24- w.Header().Set("Access-Control-Allow-Headers", "Authorization, Content-Type, X-SHA-256, X-Content-Length, X-Content-Type")25- w.Header().Set("Access-Control-Allow-Methods", "GET, HEAD, POST, PUT, DELETE, OPTIONS")24+ w.Header().Set("Access-Control-Allow-Headers", "Authorization, Content-Type, X-SHA-256, X-Content-Length, X-Content-Type, Upload-Type, Upload-Length, Upload-Offset")25+ w.Header().Set("Access-Control-Allow-Methods", "GET, HEAD, POST, PUT, PATCH, DELETE, OPTIONS")26+ w.Header().Set("Access-Control-Expose-Headers", "Allow, X-Reason")2627 w.Header().Set("X-Content-Type-Options", "nosniff")2728 if r.Method == http.MethodOptions {29+ if t.blobs != nil && t.Policy().Features.Files && blob.HandlesPath(r.URL.Path) {30+ t.blobs.Handler().ServeHTTP(w, r)31+ return32+ }2833 w.WriteHeader(http.StatusNoContent)2934 return3035 }@@ -45,6 +45,9 @@ func (t *Tenant) runScheduler() {
4545 if !wake {4646 report("records", t.records.Tick(ctx, now.Unix()))4747 if !now.Before(nextMaintenance) {48+ if t.blobs != nil {49+ report("partial uploads", t.blobs.CleanupMultipart(ctx))50+ }4851 report("retention", t.sweep(ctx, now.Unix()))4952 report("restore-state", t.sweepReplicationState(ctx, now.Unix()))5053 report("git-sync", t.git.GRASPService().Tick(ctx))@@ -3,7 +3,9 @@ package daemon
33 import (44 "bytes"55 "context"6+ "crypto/sha256"67 "database/sql"8+ "encoding/hex"79 "encoding/json"810 "errors"911 "fmt"@@ -39,15 +41,11 @@ func (t *Tenant) initServices(ctx context.Context) error {
3941 }4042 t.blobs, err = blob.New(ctx, blob.Config{4143 Root: t.meta.Paths.Root, PublicURL: t.publicURL, Store: t.store, Authorize: t.authorizeBlob,42- ValidateUpload: func(r *http.Request, hash string) error {43- // /mirror is authenticated against its complete JSON request body in44- // authorizeBlob. Its resulting blob hash is not the NIP-98 payload45- // hash; Blossom tokens still bind through their x tag.46- if r.URL.Path == "/mirror" {47- return t.auth.ValidateBlossomPayload(r.Header.Get("Authorization"), string(blob.ActionMirror), hash)48- }49- return t.auth.ValidateBlobPayload(r.Header.Get("Authorization"), hash)44+ Limits: func() blob.Limits {45+ limits := t.Policy().FileLimits46+ return blob.Limits{MaxFileBytes: limits.MaxFileBytes, UserStorageBytes: limits.UserStorageBytes}5047 },48+ ValidateUpload: t.validateBlobUpload,5149 ResolveServers: func(ctx context.Context, pubkey string) ([]string, error) {5250 result, err := t.store.Query(ctx, event.Filter{Authors: []string{pubkey}, Kinds: []int{10063}, Tags: map[string][]string{}}, storage.QueryOptions{Now: time.Now().Unix(), Access: storage.Access{All: true}, Limit: 1})5351 if err != nil {@@ -224,10 +222,13 @@ func (t *Tenant) siteReadAccess(r *http.Request) bool {
224222 }225223 func (t *Tenant) authorizeBlob(r *http.Request, action blob.Action) (string, error) {226224 hash := ""227- if action == blob.ActionGet || action == blob.ActionDelete {225+ if action == blob.ActionGet || action == blob.ActionDelete || (action == blob.ActionUpload && (r.Method == http.MethodPut || r.Method == http.MethodPatch)) {228226 path := strings.TrimPrefix(r.URL.Path, "/")229227 path = strings.TrimPrefix(path, "nip96/")230228 hash = strings.Split(path, ".")[0]229+ if len(hash) != 64 || strings.Trim(hash, "0123456789abcdef") != "" {230+ hash = ""231+ }231232 }232233 body, bodyErr := blobAuthBody(r, action)233234 if bodyErr != nil {@@ -241,17 +242,32 @@ func (t *Tenant) authorizeBlob(r *http.Request, action blob.Action) (string, err
241242 pubkey = token.PubKey242243 } else {243244 if auth.IsBlossomAuthorization(r.Header.Get("Authorization")) {245+ blossomAction := string(action)246+ if action == blob.ActionMirror {247+ // BUD-04 mirror uses the upload authorization verb; the fetched248+ // bytes are the resulting blob and are scoped by x below.249+ blossomAction = string(blob.ActionUpload)250+ }244251 blobHash := hash245252 if blobHash == "" && action == blob.ActionUpload {246253 blobHash = strings.ToLower(strings.TrimSpace(r.Header.Get("x-sha-256")))247254 }248255 var token event.Event249- token, err = t.auth.VerifyBlossomRequest(r.Header.Get("Authorization"), string(action), t.siteDomain(), blobHash)256+ token, err = t.auth.VerifyBlossomRequest(r.Header.Get("Authorization"), blossomAction, t.siteDomain(), blobHash)250257 pubkey = token.PubKey251258 } else {252- var keys []string253- keys, err = t.auth.WhoAsks(r.Header.Get("Authorization"), t.requestURL(r), r.Method, body, string(action), hash)254- pubkey = first(keys)259+ if isBUD13RemoteUpload(r) {260+ var token event.Event261+ token, err = t.auth.VerifyNIP98(r.Header.Get("Authorization"), t.requestURL(r), r.Method, "")262+ if err == nil && event.Tag(token, "payload") != "" && event.Tag(token, "payload") != emptyPayloadHash() {263+ err = errors.New("auth-required: remote upload token payload must hash an empty body")264+ }265+ pubkey = token.PubKey266+ } else {267+ var keys []string268+ keys, err = t.auth.WhoAsks(r.Header.Get("Authorization"), t.requestURL(r), r.Method, body, string(action), hash)269+ pubkey = first(keys)270+ }255271 }256272 }257273 if err != nil {@@ -284,6 +300,24 @@ func (t *Tenant) authorizeBlob(r *http.Request, action blob.Action) (string, err
284300 return pubkey, nil285301 }286302 303+func (t *Tenant) validateBlobUpload(r *http.Request, hash string) error {304+ // PATCH carries only one chunk. Blossom scopes the final path hash, while305+ // NIP-98 binds the bytes in this individual request body.306+ if r.Method == http.MethodPatch && auth.IsBlossomAuthorization(r.Header.Get("Authorization")) {307+ finalHash := strings.Split(strings.TrimPrefix(r.URL.Path, "/"), ".")[0]308+ return t.auth.ValidateBlossomPayload(r.Header.Get("Authorization"), string(blob.ActionUpload), finalHash)309+ }310+ // Mirror proofs bind the HTTP request body before fetching. Blossom311+ // authorization additionally names the fetched content through its x tag.312+ if r.URL.Path == "/mirror" || isBUD13RemoteUpload(r) {313+ if auth.IsBlossomAuthorization(r.Header.Get("Authorization")) {314+ return t.auth.ValidateBlossomPayload(r.Header.Get("Authorization"), string(blob.ActionUpload), hash)315+ }316+ return nil317+ }318+ return t.auth.ValidateBlobPayload(r.Header.Get("Authorization"), hash)319+}320+287321 // NIP-98 binds its payload to the complete HTTP body. The streaming Blossom288322 // upload remains unbuffered and is checked against its resulting SHA-256 by289323 // ValidateUpload; the JSON and multipart doors are buffered only so their@@ -355,6 +389,21 @@ func first(v []string) string {
355389 }356390 return v[0]357391 }392+393+func isBUD13RemoteUpload(r *http.Request) bool {394+ if r == nil || r.Method != http.MethodPut || r.URL.Query().Get("url") == "" {395+ return false396+ }397+ path := strings.TrimPrefix(r.URL.Path, "/")398+ hash := strings.Split(path, ".")[0]399+ return len(hash) == 64 && strings.Trim(hash, "0123456789abcdefABCDEF") == ""400+}401+402+func emptyPayloadHash() string {403+ sum := sha256.Sum256(nil)404+ return hex.EncodeToString(sum[:])405+}406+358407 func (t *Tenant) resolveUIActor(r *http.Request) (string, error) {359408 if r.Header.Get("Authorization") == "" {360409 if r.Method != http.MethodGet && r.Method != http.MethodHead {@@ -12,6 +12,181 @@ function token(sk: Uint8Array, verb: "upload" | "get" | "list" | "delete", sha?:
1212 }1313 1414 describe("Blossom media", () => {15+ it("advertises the BUD-06 browser and upload preflight contract", async () => {16+ const preflight = await fetch(`${HTTP_URL}/upload`, {17+ method: "OPTIONS",18+ headers: {19+ origin: "https://client.example",20+ "access-control-request-method": "PUT",21+ "access-control-request-headers": "authorization, content-type, x-content-length, x-content-type",22+ },23+ });24+ expect(preflight.status).toBe(204);25+ expect(preflight.headers.get("access-control-allow-origin")).toBe("*");26+ expect(preflight.headers.get("access-control-allow-methods")).toContain("PUT");27+ expect(preflight.headers.get("access-control-allow-headers")?.toLowerCase()).toContain("x-content-type");28+29+ const sk = newKey();30+ const body = new TextEncoder().encode(`preflight ${rand()}`);31+ const sha = hex(sha256(body));32+ const head = await fetch(`${HTTP_URL}/upload`, {33+ method: "HEAD",34+ headers: {35+ authorization: token(sk, "upload", sha),36+ "x-sha-256": sha,37+ "x-content-length": String(body.length),38+ "x-content-type": "text/plain",39+ },40+ });41+ expect(head.status).toBe(200);42+ expect(await head.text()).toBe("");43+ });44+45+ it("accepts a BUD-13 hash-addressed upload and rejects a mismatched hash", async () => {46+ const sk = newKey();47+ const body = new TextEncoder().encode(`path upload ${rand()}`);48+ const sha = hex(sha256(body));49+ const upload = await fetch(`${HTTP_URL}/${sha}.txt`, {50+ method: "PUT",51+ headers: { authorization: token(sk, "upload", sha), "content-type": "text/plain" },52+ body,53+ });54+ expect(upload.status, await upload.clone().text()).toBe(201);55+ expect((await upload.json<any>()).sha256).toBe(sha);56+57+ const mismatchHash = "0".repeat(64);58+ const mismatch = await fetch(`${HTTP_URL}/${mismatchHash}`, {59+ method: "PUT",60+ headers: { authorization: token(sk, "upload", mismatchHash), "content-type": "text/plain" },61+ body,62+ });63+ expect(mismatch.status).toBe(409);64+65+ const del = await fetch(`${HTTP_URL}/${sha}`, {66+ method: "DELETE",67+ headers: { authorization: token(sk, "delete", sha) },68+ });69+ expect(del.status).toBe(204);70+ });71+72+ it("advertises and completes BUD-14 uploads out of order with overlap", async () => {73+ const sk = newKey();74+ const body = new TextEncoder().encode(`bud14 ${rand()} multipart body`);75+ const sha = hex(sha256(body));76+ const split = Math.floor(body.length / 2);77+78+ const preflight = await fetch(`${HTTP_URL}/${sha}.bin`, {79+ method: "OPTIONS",80+ headers: {81+ origin: "https://client.example",82+ "access-control-request-method": "PATCH",83+ "access-control-request-headers": "authorization, content-type, upload-type, upload-length, upload-offset",84+ },85+ });86+ expect(preflight.status).toBe(204);87+ expect(preflight.headers.get("allow")).toContain("PATCH");88+ expect(preflight.headers.get("access-control-allow-origin")).toBe("*");89+ expect(preflight.headers.get("access-control-allow-methods")).toContain("PATCH");90+91+ // The authorization is scoped to the final hash. It is reused for both92+ // chunks; individual chunk hashes must not be required by BUD-14.93+ const auth = token(sk, "upload", sha);94+ const secondOffset = Math.max(1, split - 2);95+ const second = body.slice(secondOffset);96+ const first = body.slice(0, split + 2);97+ const send = (offset: number, chunk: Uint8Array) => fetch(`${HTTP_URL}/${sha}.bin`, {98+ method: "PATCH",99+ headers: {100+ authorization: auth,101+ "content-type": "application/octet-stream",102+ "upload-type": "text/plain",103+ "upload-length": String(body.length),104+ "upload-offset": String(offset),105+ "content-length": String(chunk.length),106+ },107+ body: chunk,108+ });109+110+ const chunkScoped = await fetch(`${HTTP_URL}/${sha}.bin`, {111+ method: "PATCH",112+ headers: {113+ authorization: token(sk, "upload", hex(sha256(second))),114+ "content-type": "application/octet-stream",115+ "upload-type": "text/plain",116+ "upload-length": String(body.length),117+ "upload-offset": String(secondOffset),118+ "content-length": String(second.length),119+ },120+ body: second,121+ });122+ expect(chunkScoped.status).toBe(401);123+124+ // Send the later range first, then an overlapping earlier range.125+ const later = await send(secondOffset, second);126+ expect(later.status).toBe(204);127+ expect((await fetch(`${HTTP_URL}/${sha}`)).status).toBe(404);128+ const complete = await send(0, first);129+ expect(complete.status, await complete.clone().text()).toBe(201);130+ expect((await fetch(`${HTTP_URL}/${sha}`)).status).toBe(200);131+ expect(new Uint8Array(await (await fetch(`${HTTP_URL}/${sha}`)).arrayBuffer())).toEqual(body);132+133+ const del = await fetch(`${HTTP_URL}/${sha}`, {134+ method: "DELETE",135+ headers: { authorization: token(sk, "delete", sha) },136+ });137+ expect(del.status).toBe(204);138+ });139+140+ it("rejects a complete BUD-14 upload whose reconstructed bytes have the wrong hash", async () => {141+ const sk = newKey();142+ const body = new TextEncoder().encode(`bud14 mismatch ${rand()}`);143+ const claimed = hex(sha256(new TextEncoder().encode(`${body} wrong`)));144+ const auth = token(sk, "upload", claimed);145+ const response = await fetch(`${HTTP_URL}/${claimed}`, {146+ method: "PATCH",147+ headers: {148+ authorization: auth,149+ "content-type": "application/octet-stream",150+ "upload-type": "text/plain",151+ "upload-length": String(body.length),152+ "upload-offset": "0",153+ "content-length": String(body.length),154+ },155+ body,156+ });157+ expect(response.status).toBe(409);158+ expect((await fetch(`${HTTP_URL}/${claimed}`)).status).toBe(404);159+ });160+161+ it("keeps repeated BUD-14 ranges idempotent while an upload is partial", async () => {162+ const sk = newKey();163+ const body = new TextEncoder().encode(`bud14 duplicate ${rand()}`);164+ const sha = hex(sha256(body));165+ const auth = token(sk, "upload", sha);166+ const send = (offset: number, chunk: Uint8Array) => fetch(`${HTTP_URL}/${sha}`, {167+ method: "PATCH",168+ headers: {169+ authorization: auth,170+ "content-type": "application/octet-stream",171+ "upload-type": "text/plain",172+ "upload-length": String(body.length),173+ "upload-offset": String(offset),174+ "content-length": String(chunk.length),175+ },176+ body: chunk,177+ });178+ const first = body.slice(0, Math.floor(body.length / 2));179+ expect((await send(0, first)).status).toBe(204);180+ expect((await send(0, first)).status).toBe(204);181+ const finish = await send(first.length, body.slice(first.length));182+ expect(finish.status, await finish.clone().text()).toBe(201);183+ expect((await fetch(`${HTTP_URL}/${sha}`)).status).toBe(200);184+ expect((await fetch(`${HTTP_URL}/${sha}`, {185+ method: "DELETE",186+ headers: { authorization: token(sk, "delete", sha) },187+ })).status).toBe(204);188+ });189+15190 it("uploads a file, serves it whole and by range, lists it, and deletes it", async () => {16191 const sk = newKey();17192 const body = new TextEncoder().encode(`a small file ${rand()} with enough bytes for a range`);