| 1 | diff --git a/internal/blob/blob.go b/internal/blob/blob.go |
| 2 | index 588d035..5d24940 100644 |
| 3 | --- a/internal/blob/blob.go |
| 4 | +++ b/internal/blob/blob.go |
| 5 | @@ -61,6 +61,10 @@ type Config struct { |
| 6 | // Limits is read for each upload so policy changes apply without reopening |
| 7 | // the blob service. Zero values mean unlimited. |
| 8 | Limits func() Limits |
| 9 | + // UploadTerms returns what an uploader's standing imposes on its uploads: |
| 10 | + // an agent grant with a sites ttl gives its claims an expiry, and one with |
| 11 | + // the encrypted flag refuses plain uploads. Zero values mean no terms. |
| 12 | + UploadTerms func(context.Context, string) (UploadTerms, error) |
| 13 | } |
| 14 | |
| 15 | type Limits struct { |
| 16 | @@ -68,6 +72,20 @@ type Limits struct { |
| 17 | UserStorageBytes int64 |
| 18 | } |
| 19 | |
| 20 | +// UploadTerms is what an uploader's grant imposes on the blobs it stores. |
| 21 | +// ExpiresAt is the unix time its claims lapse, 0 for never; Encrypted |
| 22 | +// requires the stored bytes to be an encrypted blob or manifest. |
| 23 | +type UploadTerms struct { |
| 24 | + ExpiresAt int64 |
| 25 | + Encrypted bool |
| 26 | +} |
| 27 | + |
| 28 | +// encryptedTypes are the content types the browser uploader gives encrypted |
| 29 | +// objects: AES-GCM ciphertext for files and chunks, and the BUD-16 directory |
| 30 | +// type for encrypted manifests. Anything the relay recognizes as a page, |
| 31 | +// image, video, audio or document is plain. |
| 32 | +var encryptedTypes = map[string]bool{"application/octet-stream": true, "application/vnd.blossom.directory+msgpack": true} |
| 33 | + |
| 34 | type Blob struct { |
| 35 | SHA256 string `json:"sha256"` |
| 36 | Size int64 `json:"size"` |
| 37 | @@ -107,6 +125,7 @@ var ErrHashMismatch = errors.New("conflict: mirrored blob hash does not match th |
| 38 | var ErrFileTooLarge = errors.New("invalid: blob exceeds the per-file size limit") |
| 39 | var ErrQuotaExceeded = errors.New("restricted: uploader storage quota exceeded") |
| 40 | var ErrMultipartConflict = errors.New("conflict: upload metadata differs from existing session") |
| 41 | +var ErrNotEncrypted = errors.New("restricted: agent grant requires encrypted uploads") |
| 42 | |
| 43 | var shaPattern = regexp.MustCompile(`^[0-9a-f]{64}$`) |
| 44 | var blobPathPattern = regexp.MustCompile(`^/([0-9a-f]{64})(?:\.[a-z0-9]{1,8})?$`) |
| 45 | @@ -157,6 +176,14 @@ func New(ctx context.Context, config Config) (*Service, error) { |
| 46 | SELECT sha256,uploader,uploaded FROM blobs WHERE uploader != ''`); err != nil { |
| 47 | return nil, fmt.Errorf("initialize blob metadata: %w", err) |
| 48 | } |
| 49 | + // Claims made under an agent grant with a ttl carry an expiry; every |
| 50 | + // earlier claim keeps 0, which never lapses. |
| 51 | + if _, err := config.Store.DB().ExecContext(ctx, "ALTER TABLE blob_claims ADD COLUMN expires INTEGER NOT NULL DEFAULT 0"); err != nil && !strings.Contains(err.Error(), "duplicate column") { |
| 52 | + return nil, fmt.Errorf("add blob claim expiry: %w", err) |
| 53 | + } |
| 54 | + if _, err := config.Store.DB().ExecContext(ctx, "CREATE INDEX IF NOT EXISTS blob_claims_expires ON blob_claims(expires) WHERE expires>0"); err != nil { |
| 55 | + return nil, fmt.Errorf("index blob claim expiry: %w", err) |
| 56 | + } |
| 57 | if err := reconcile(ctx, config.Store, filepath.Join(config.Root, "blobs")); err != nil { |
| 58 | return nil, err |
| 59 | } |
| 60 | @@ -649,6 +676,10 @@ func (s *Service) installUpload(ctx context.Context, candidate uploadCandidate) |
| 61 | if s.blocked(ctx, sha) { |
| 62 | return Blob{}, false, ErrBlocked |
| 63 | } |
| 64 | + terms, err := s.uploadTerms(ctx, uploader) |
| 65 | + if err != nil { |
| 66 | + return Blob{}, false, err |
| 67 | + } |
| 68 | limits := s.currentLimits() |
| 69 | if limits.MaxFileBytes > 0 && count > limits.MaxFileBytes { |
| 70 | return Blob{}, false, ErrFileTooLarge |
| 71 | @@ -657,6 +688,9 @@ func (s *Service) installUpload(ctx context.Context, candidate uploadCandidate) |
| 72 | return Blob{}, false, ErrQuotaExceeded |
| 73 | } |
| 74 | if existing, err := s.lookup(ctx, sha); err == nil { |
| 75 | + if terms.Encrypted && !encryptedTypes[existing.Type] { |
| 76 | + return Blob{}, false, ErrNotEncrypted |
| 77 | + } |
| 78 | if uploader != "" { |
| 79 | if limits.UserStorageBytes > 0 && !s.hasClaim(ctx, sha, uploader) { |
| 80 | used, usageErr := s.quotaUsageExcept(ctx, uploader, candidate.reservation) |
| 81 | @@ -667,7 +701,7 @@ func (s *Service) installUpload(ctx context.Context, candidate uploadCandidate) |
| 82 | return Blob{}, false, ErrQuotaExceeded |
| 83 | } |
| 84 | } |
| 85 | - if err := s.claim(ctx, sha, uploader); err != nil { |
| 86 | + if err := s.claim(ctx, sha, uploader, terms.ExpiresAt); err != nil { |
| 87 | return Blob{}, false, err |
| 88 | } |
| 89 | } |
| 90 | @@ -689,6 +723,16 @@ func (s *Service) installUpload(ctx context.Context, candidate uploadCandidate) |
| 91 | return Blob{}, false, ErrQuotaExceeded |
| 92 | } |
| 93 | } |
| 94 | + // The type is settled before the bytes are installed so a plain upload |
| 95 | + // under an encrypted-only grant leaves nothing behind. A declared octet |
| 96 | + // stream is sniffed: encrypted bytes stay one, a page or an image does not. |
| 97 | + typ = contentType(typ) |
| 98 | + if typ == "application/octet-stream" { |
| 99 | + typ = detectFileType(candidate.path) |
| 100 | + } |
| 101 | + if terms.Encrypted && !encryptedTypes[typ] { |
| 102 | + return Blob{}, false, ErrNotEncrypted |
| 103 | + } |
| 104 | final := filepath.Join(s.root, sha) |
| 105 | if err := os.Rename(candidate.path, final); err != nil && !errors.Is(err, os.ErrExist) { |
| 106 | return Blob{}, false, fmt.Errorf("install blob: %w", err) |
| 107 | @@ -697,12 +741,8 @@ func (s *Service) installUpload(ctx context.Context, candidate uploadCandidate) |
| 108 | return Blob{}, false, fmt.Errorf("sync blob directory: %w", err) |
| 109 | } |
| 110 | now := time.Now().UTC().Unix() |
| 111 | - typ = contentType(typ) |
| 112 | - if typ == "application/octet-stream" { |
| 113 | - typ = detectFileType(final) |
| 114 | - } |
| 115 | entry := Blob{SHA256: sha, Size: count, Type: typ, Uploader: uploader, Uploaded: now} |
| 116 | - err := s.store.WithTx(ctx, func(tx *sql.Tx) error { |
| 117 | + err = s.store.WithTx(ctx, func(tx *sql.Tx) error { |
| 118 | if _, err := tx.ExecContext(ctx, "DELETE FROM blob_tombstones WHERE sha256=?", entry.SHA256); err != nil { |
| 119 | return fmt.Errorf("clear blob deletion tombstone: %w", err) |
| 120 | } |
| 121 | @@ -710,7 +750,7 @@ func (s *Service) installUpload(ctx context.Context, candidate uploadCandidate) |
| 122 | return fmt.Errorf("record blob: %w", err) |
| 123 | } |
| 124 | if entry.Uploader != "" { |
| 125 | - if _, err := tx.ExecContext(ctx, "INSERT OR IGNORE INTO blob_claims(sha256,uploader,claimed_at) VALUES(?,?,?)", entry.SHA256, entry.Uploader, entry.Uploaded); err != nil { |
| 126 | + if _, err := tx.ExecContext(ctx, claimSQL, entry.SHA256, entry.Uploader, entry.Uploaded, terms.ExpiresAt); err != nil { |
| 127 | return fmt.Errorf("record blob claim: %w", err) |
| 128 | } |
| 129 | } |
| 130 | @@ -745,14 +785,64 @@ func (s *Service) quotaUsageExcept(ctx context.Context, uploader, reservation st |
| 131 | return used.Int64, nil |
| 132 | } |
| 133 | |
| 134 | -func (s *Service) claim(ctx context.Context, sha, uploader string) error { |
| 135 | - _, err := s.store.DB().ExecContext(ctx, "INSERT OR IGNORE INTO blob_claims(sha256,uploader,claimed_at) VALUES(?,?,?)", sha, uploader, time.Now().UTC().Unix()) |
| 136 | +// claimSQL records a claim. The insert trigger on blobs has usually made the |
| 137 | +// row already, without an expiry, so the newest upload's terms replace it. |
| 138 | +const claimSQL = "INSERT INTO blob_claims(sha256,uploader,claimed_at,expires) VALUES(?,?,?,?) ON CONFLICT(sha256,uploader) DO UPDATE SET expires=excluded.expires" |
| 139 | + |
| 140 | +func (s *Service) claim(ctx context.Context, sha, uploader string, expires int64) error { |
| 141 | + _, err := s.store.DB().ExecContext(ctx, claimSQL, sha, uploader, time.Now().UTC().Unix(), expires) |
| 142 | if err != nil { |
| 143 | return fmt.Errorf("record blob claim: %w", err) |
| 144 | } |
| 145 | return nil |
| 146 | } |
| 147 | |
| 148 | +func (s *Service) uploadTerms(ctx context.Context, uploader string) (UploadTerms, error) { |
| 149 | + if s.config.UploadTerms == nil || uploader == "" { |
| 150 | + return UploadTerms{}, nil |
| 151 | + } |
| 152 | + terms, err := s.config.UploadTerms(ctx, uploader) |
| 153 | + if err != nil { |
| 154 | + return UploadTerms{}, fmt.Errorf("check upload terms: %w", err) |
| 155 | + } |
| 156 | + return terms, nil |
| 157 | +} |
| 158 | + |
| 159 | +// SweepExpired removes the blobs whose every claim has lapsed and releases |
| 160 | +// lapsed claims on blobs someone else still holds, so an agent's files leave |
| 161 | +// with its site while a person's claim keeps them. It returns how many blobs |
| 162 | +// were removed. |
| 163 | +func (s *Service) SweepExpired(ctx context.Context, now int64) (int, error) { |
| 164 | + s.quotaMu.Lock() |
| 165 | + defer s.quotaMu.Unlock() |
| 166 | + rows, err := s.store.DB().QueryContext(ctx, `SELECT DISTINCT sha256 FROM blob_claims c WHERE c.expires>0 AND c.expires<=? |
| 167 | + AND NOT EXISTS(SELECT 1 FROM blob_claims live WHERE live.sha256=c.sha256 AND (live.expires=0 OR live.expires>?))`, now, now) |
| 168 | + if err != nil { |
| 169 | + return 0, fmt.Errorf("list expired blobs: %w", err) |
| 170 | + } |
| 171 | + var expired []string |
| 172 | + for rows.Next() { |
| 173 | + var sha string |
| 174 | + if err := rows.Scan(&sha); err != nil { |
| 175 | + _ = rows.Close() |
| 176 | + return 0, err |
| 177 | + } |
| 178 | + expired = append(expired, sha) |
| 179 | + } |
| 180 | + if err := errors.Join(rows.Err(), rows.Close()); err != nil { |
| 181 | + return 0, err |
| 182 | + } |
| 183 | + for _, sha := range expired { |
| 184 | + if err := s.Delete(ctx, sha); err != nil { |
| 185 | + return 0, err |
| 186 | + } |
| 187 | + } |
| 188 | + if _, err := s.store.DB().ExecContext(ctx, "DELETE FROM blob_claims WHERE expires>0 AND expires<=?", now); err != nil { |
| 189 | + return 0, fmt.Errorf("release expired blob claims: %w", err) |
| 190 | + } |
| 191 | + return len(expired), nil |
| 192 | +} |
| 193 | + |
| 194 | func (s *Service) currentLimits() Limits { |
| 195 | if s.config.Limits == nil { |
| 196 | return Limits{} |
| 197 | @@ -1309,7 +1399,7 @@ func statusFor(err error) int { |
| 198 | if errors.Is(err, ErrQuotaExceeded) { |
| 199 | return http.StatusForbidden |
| 200 | } |
| 201 | - if errors.Is(err, ErrBlocked) { |
| 202 | + if errors.Is(err, ErrBlocked) || errors.Is(err, ErrNotEncrypted) { |
| 203 | return http.StatusForbidden |
| 204 | } |
| 205 | if strings.HasPrefix(err.Error(), "auth-required:") { |
| 206 | diff --git a/internal/blob/blob_test.go b/internal/blob/blob_test.go |
| 207 | index f524ebf..cdfcfce 100644 |
| 208 | --- a/internal/blob/blob_test.go |
| 209 | +++ b/internal/blob/blob_test.go |
| 210 | @@ -291,6 +291,112 @@ func TestBlobClaimsPreserveDeduplicatedOwnershipAndCursor(t *testing.T) { |
| 211 | } |
| 212 | } |
| 213 | |
| 214 | +// TestAgentUploadsExpireUnlessAPersonClaimsThem covers the ttl an agent |
| 215 | +// grant puts on uploads: claims carry the expiry, the sweep removes blobs |
| 216 | +// whose every claim lapsed and keeps those a person holds. |
| 217 | +func TestAgentUploadsExpireUnlessAPersonClaimsThem(t *testing.T) { |
| 218 | + service := testService(t) |
| 219 | + agent := strings.Repeat("a", 64) |
| 220 | + person := strings.Repeat("b", 64) |
| 221 | + now := time.Now().Unix() |
| 222 | + service.config.UploadTerms = func(_ context.Context, uploader string) (UploadTerms, error) { |
| 223 | + if uploader == agent { |
| 224 | + return UploadTerms{ExpiresAt: now + 10}, nil |
| 225 | + } |
| 226 | + return UploadTerms{}, nil |
| 227 | + } |
| 228 | + ctx := context.Background() |
| 229 | + shared, err := service.Put(ctx, PutOptions{Reader: strings.NewReader("shared site file"), Type: "text/html", Uploader: agent}) |
| 230 | + if err != nil { |
| 231 | + t.Fatal(err) |
| 232 | + } |
| 233 | + alone, err := service.Put(ctx, PutOptions{Reader: strings.NewReader("agent only file"), Type: "text/html", Uploader: agent}) |
| 234 | + if err != nil { |
| 235 | + t.Fatal(err) |
| 236 | + } |
| 237 | + var expires int64 |
| 238 | + if err := service.store.DB().QueryRow("SELECT expires FROM blob_claims WHERE sha256=? AND uploader=?", alone.SHA256, agent).Scan(&expires); err != nil || expires != now+10 { |
| 239 | + t.Fatalf("agent claim expires=%d err=%v, want %d", expires, err, now+10) |
| 240 | + } |
| 241 | + if _, err := service.Put(ctx, PutOptions{Reader: strings.NewReader("shared site file"), Type: "text/html", Uploader: person}); err != nil { |
| 242 | + t.Fatal(err) |
| 243 | + } |
| 244 | + if err := service.store.DB().QueryRow("SELECT expires FROM blob_claims WHERE sha256=? AND uploader=?", shared.SHA256, person).Scan(&expires); err != nil || expires != 0 { |
| 245 | + t.Fatalf("person claim expires=%d err=%v, want 0", expires, err) |
| 246 | + } |
| 247 | + if removed, err := service.SweepExpired(ctx, now+9); err != nil || removed != 0 { |
| 248 | + t.Fatalf("early sweep removed %d err=%v", removed, err) |
| 249 | + } |
| 250 | + if removed, err := service.SweepExpired(ctx, now+10); err != nil || removed != 1 { |
| 251 | + t.Fatalf("sweep removed %d err=%v, want 1", removed, err) |
| 252 | + } |
| 253 | + if _, _, err := service.Get(ctx, alone.SHA256); err == nil { |
| 254 | + t.Fatal("expired unclaimed blob still served") |
| 255 | + } |
| 256 | + if _, err := os.Stat(filepath.Join(service.root, alone.SHA256)); !errors.Is(err, os.ErrNotExist) { |
| 257 | + t.Fatalf("expired blob file remains: %v", err) |
| 258 | + } |
| 259 | + if _, _, err := service.Get(ctx, shared.SHA256); err != nil { |
| 260 | + t.Fatalf("blob a person claimed was removed: %v", err) |
| 261 | + } |
| 262 | + mine, err := service.ListMetadata(ctx, agent) |
| 263 | + if err != nil || len(mine) != 0 { |
| 264 | + t.Fatalf("agent still lists %d blobs after expiry: %v", len(mine), err) |
| 265 | + } |
| 266 | + theirs, err := service.ListMetadata(ctx, person) |
| 267 | + if err != nil || len(theirs) != 1 { |
| 268 | + t.Fatalf("person lists %d blobs: %v", len(theirs), err) |
| 269 | + } |
| 270 | + // A restart keeps the expiry column and the earlier rows. |
| 271 | + if _, err := New(ctx, service.config); err != nil { |
| 272 | + t.Fatal(err) |
| 273 | + } |
| 274 | +} |
| 275 | + |
| 276 | +func TestEncryptedOnlyTermsRefusePlainUploads(t *testing.T) { |
| 277 | + service := testService(t) |
| 278 | + agent := strings.Repeat("a", 64) |
| 279 | + service.config.UploadTerms = func(_ context.Context, uploader string) (UploadTerms, error) { |
| 280 | + return UploadTerms{Encrypted: uploader == agent}, nil |
| 281 | + } |
| 282 | + ctx := context.Background() |
| 283 | + png := "\x89PNG\r\n\x1a\n" + strings.Repeat("\x00", 64) |
| 284 | + for name, options := range map[string]PutOptions{ |
| 285 | + "declared page": {Reader: strings.NewReader("<html>plain</html>"), Type: "text/html", Uploader: agent}, |
| 286 | + "sniffed image": {Reader: strings.NewReader(png), Type: "application/octet-stream", Uploader: agent}, |
| 287 | + "declared manifest": {Reader: strings.NewReader("{}"), Type: "application/json", Uploader: agent}, |
| 288 | + } { |
| 289 | + if _, err := service.Put(ctx, options); !errors.Is(err, ErrNotEncrypted) { |
| 290 | + t.Fatalf("%s: err=%v, want ErrNotEncrypted", name, err) |
| 291 | + } |
| 292 | + } |
| 293 | + if entries, _ := os.ReadDir(service.root); len(entries) != 0 { |
| 294 | + t.Fatalf("refused uploads left %d files", len(entries)) |
| 295 | + } |
| 296 | + cipher := "\x8f\x1a\x9c\x03" + strings.Repeat("\xe2\x71\x05\x9b\x4c", 20) |
| 297 | + if _, err := service.Put(ctx, PutOptions{Reader: strings.NewReader(cipher), Type: "application/octet-stream", Uploader: agent}); err != nil { |
| 298 | + t.Fatalf("ciphertext refused: %v", err) |
| 299 | + } |
| 300 | + if _, err := service.Put(ctx, PutOptions{Reader: strings.NewReader("\x82\xa1t\x01"), Type: "application/vnd.blossom.directory+msgpack", Uploader: agent}); err != nil { |
| 301 | + t.Fatalf("encrypted manifest refused: %v", err) |
| 302 | + } |
| 303 | + // Claiming a plain blob someone else stored is refused too. |
| 304 | + if _, err := service.Put(ctx, PutOptions{Reader: strings.NewReader("<html>theirs</html>"), Type: "text/html", Uploader: strings.Repeat("b", 64)}); err != nil { |
| 305 | + t.Fatal(err) |
| 306 | + } |
| 307 | + if _, err := service.Put(ctx, PutOptions{Reader: strings.NewReader("<html>theirs</html>"), Type: "text/html", Uploader: agent}); !errors.Is(err, ErrNotEncrypted) { |
| 308 | + t.Fatalf("claim of a plain blob: err=%v", err) |
| 309 | + } |
| 310 | + // The door answers 403 with the reason. |
| 311 | + service.config.Authorize = func(*http.Request, Action) (string, error) { return agent, nil } |
| 312 | + req := httptest.NewRequest(http.MethodPut, "https://relay.test/upload", strings.NewReader("<html>plain</html>")) |
| 313 | + req.Header.Set("Content-Type", "text/html") |
| 314 | + response := record(service.Handler().ServeHTTP, req) |
| 315 | + if response.Code != http.StatusForbidden || !strings.Contains(response.Body.String(), "requires encrypted uploads") { |
| 316 | + t.Fatalf("plain upload = %d %s", response.Code, response.Body.String()) |
| 317 | + } |
| 318 | +} |
| 319 | + |
| 320 | const testUploader = "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef" |
| 321 | |
| 322 | func testService(t *testing.T) *Service { |