blossom.go raw
1 // Package blossom implements the Blossom blob protocol (BUD-01, BUD-02,
2 // BUD-06 and BUD-12) on top of a content-addressed directory.
3 //
4 // Blobs live at blobs/<first two hash bytes>/<sha256>; an append-only index log
5 // records ownership and content type so that listing by pubkey does not have to
6 // read every blob file.
7 package blossom
8
9 import (
10 "encoding/binary"
11 "encoding/json"
12 "errors"
13 "fmt"
14 "io"
15 "os"
16 "path/filepath"
17 "strings"
18 "sync"
19 "time"
20
21 "git.smesh.lol/torly/pkg/nostr"
22 )
23
24 // Request is one HTTP request, independent of any server implementation.
25 type Request struct {
26 Method string
27 Path string
28 Query string
29 Header map[string]string
30 Body []byte
31 }
32
33 // Response is one HTTP response produced by Serve.
34 type Response struct {
35 Status int
36 Header [][2]string
37 Body []byte
38 }
39
40 // respWriter collects handler output into a Response.
41 type respWriter struct {
42 status int
43 header [][2]string
44 body []byte
45 }
46
47 func (w *respWriter) SetHeader(name, value string) {
48 for i := range w.header {
49 if strings.EqualFold(w.header[i][0], name) {
50 w.header[i][1] = value
51 return
52 }
53 }
54 w.header = append(w.header, [2]string{name, value})
55 }
56
57 func (w *respWriter) WriteHeader(status int) {
58 if w.status == 0 {
59 w.status = status
60 }
61 }
62
63 func (w *respWriter) Write(p []byte) (int, error) {
64 if w.status == 0 {
65 w.status = 200
66 }
67 w.body = append(w.body, p...)
68 return len(p), nil
69 }
70
71 func (w *respWriter) WriteString(v string) (int, error) { return w.Write([]byte(v)) }
72
73 // Blob describes one stored blob.
74 type Blob struct {
75 Hash string `json:"sha256"`
76 PubKey string `json:"pubkey,omitempty"`
77 Size int64 `json:"size"`
78 Type string `json:"type"`
79 Created int64 `json:"uploaded"`
80 }
81
82 // Descriptor is the BUD-02 response body.
83 type Descriptor struct {
84 URL string `json:"url"`
85 SHA256 string `json:"sha256"`
86 Size int64 `json:"size"`
87 Type string `json:"type"`
88 Uploaded int64 `json:"uploaded"`
89 }
90
91 // Server is a Blossom store rooted at dir.
92 type Server struct {
93 dir string
94 maxBlob int64
95
96 mu sync.Mutex
97 blobs map[string]*Blob
98 byPub map[string]map[string]*Blob
99 log *os.File
100 }
101
102 const (
103 indexName = "index.log"
104 blobDirName = "blobs"
105 indexCompact = 4 << 20
106 // record header: op(1) | hash(32) | pubkey(32) | size(8) | created(8) | typeLen(2)
107 indexHeaderLen = 1 + 32 + 32 + 8 + 8 + 2
108 )
109
110 // Open opens or creates a Blossom store.
111 func Open(dir string, maxBlob int64) (*Server, error) {
112 if maxBlob <= 0 {
113 maxBlob = 50 << 20
114 }
115 if err := os.MkdirAll(filepath.Join(dir, blobDirName), 0o755); err != nil {
116 return nil, err
117 }
118 s := &Server{
119 dir: dir,
120 maxBlob: maxBlob,
121 blobs: make(map[string]*Blob),
122 byPub: make(map[string]map[string]*Blob),
123 }
124 if err := s.loadIndex(); err != nil {
125 return nil, err
126 }
127 f, err := os.OpenFile(filepath.Join(dir, indexName), os.O_RDWR|os.O_CREATE|os.O_APPEND, 0o644)
128 if err != nil {
129 return nil, err
130 }
131 s.log = f
132 if info, err := f.Stat(); err == nil && info.Size() > indexCompact {
133 if err := s.compact(); err != nil {
134 return nil, err
135 }
136 }
137 return s, nil
138 }
139
140 func (s *Server) loadIndex() error {
141 f, err := os.Open(filepath.Join(s.dir, indexName))
142 if err != nil {
143 if os.IsNotExist(err) {
144 return nil
145 }
146 return err
147 }
148 defer f.Close()
149 hdr := make([]byte, indexHeaderLen)
150 for {
151 if _, err := io.ReadFull(f, hdr); err != nil {
152 if err == io.EOF || err == io.ErrUnexpectedEOF {
153 return nil
154 }
155 return err
156 }
157 op := hdr[0]
158 hash := nostr.Hex(hdr[1:33])
159 size := int64(binary.BigEndian.Uint64(hdr[65:73]))
160 created := int64(binary.BigEndian.Uint64(hdr[73:81]))
161 typeLen := int(binary.BigEndian.Uint16(hdr[81:83]))
162 typ := ""
163 if typeLen > 0 {
164 if typeLen > 1024 {
165 return errors.New("blossom: corrupt index type length")
166 }
167 buf := make([]byte, typeLen)
168 if _, err := io.ReadFull(f, buf); err != nil {
169 return nil
170 }
171 typ = string(buf)
172 }
173 switch op {
174 case 'p':
175 pub := nostr.Hex(hdr[33:65])
176 b := &Blob{Hash: hash, PubKey: pub, Size: size, Type: typ, Created: created}
177 s.putMem(b)
178 case 'd':
179 s.delMem(hash)
180 default:
181 return errors.New("blossom: corrupt index record")
182 }
183 }
184 }
185
186 func (s *Server) putMem(b *Blob) {
187 s.blobs[b.Hash] = b
188 m := s.byPub[b.PubKey]
189 if m == nil {
190 m = make(map[string]*Blob)
191 s.byPub[b.PubKey] = m
192 }
193 m[b.Hash] = b
194 }
195
196 func (s *Server) delMem(hash string) {
197 b := s.blobs[hash]
198 if b == nil {
199 return
200 }
201 delete(s.blobs, hash)
202 if m := s.byPub[b.PubKey]; m != nil {
203 delete(m, hash)
204 if len(m) == 0 {
205 delete(s.byPub, b.PubKey)
206 }
207 }
208 }
209
210 func (s *Server) appendRecord(op byte, b *Blob) error {
211 typ := b.Type
212 if len(typ) > 1024 {
213 typ = typ[:1024]
214 }
215 rec := make([]byte, indexHeaderLen+len(typ))
216 rec[0] = op
217 copy(rec[1:33], mustDecode(b.Hash))
218 if b.PubKey != "" {
219 copy(rec[33:65], mustDecode(b.PubKey))
220 }
221 binary.BigEndian.PutUint64(rec[65:73], uint64(b.Size))
222 binary.BigEndian.PutUint64(rec[73:81], uint64(b.Created))
223 binary.BigEndian.PutUint16(rec[81:83], uint16(len(typ)))
224 copy(rec[indexHeaderLen:], typ)
225 if _, err := s.log.Write(rec); err != nil {
226 return err
227 }
228 return s.log.Sync()
229 }
230
231 func mustDecode(hexStr string) []byte {
232 var out [32]byte
233 nostr.DecodeRaw(hexStr, out[:])
234 return out[:]
235 }
236
237 // compact rewrites the index from the in-memory map.
238 func (s *Server) compact() error {
239 tmp := filepath.Join(s.dir, indexName+".tmp")
240 f, err := os.OpenFile(tmp, os.O_RDWR|os.O_CREATE|os.O_TRUNC, 0o644)
241 if err != nil {
242 return err
243 }
244 s.mu.Lock()
245 for _, b := range s.blobs {
246 if err := writeRecord(f, 'p', b); err != nil {
247 s.mu.Unlock()
248 f.Close()
249 return err
250 }
251 }
252 s.mu.Unlock()
253 if err := f.Sync(); err != nil {
254 f.Close()
255 return err
256 }
257 if err := f.Close(); err != nil {
258 return err
259 }
260 if err := s.log.Close(); err != nil {
261 return err
262 }
263 if err := os.Rename(tmp, filepath.Join(s.dir, indexName)); err != nil {
264 return err
265 }
266 nf, err := os.OpenFile(filepath.Join(s.dir, indexName), os.O_RDWR|os.O_APPEND, 0o644)
267 if err != nil {
268 return err
269 }
270 s.log = nf
271 return nil
272 }
273
274 func writeRecord(f *os.File, op byte, b *Blob) error {
275 typ := b.Type
276 if len(typ) > 1024 {
277 typ = typ[:1024]
278 }
279 rec := make([]byte, indexHeaderLen+len(typ))
280 rec[0] = op
281 copy(rec[1:33], mustDecode(b.Hash))
282 copy(rec[33:65], mustDecode(b.PubKey))
283 binary.BigEndian.PutUint64(rec[65:73], uint64(b.Size))
284 binary.BigEndian.PutUint64(rec[73:81], uint64(b.Created))
285 binary.BigEndian.PutUint16(rec[81:83], uint16(len(typ)))
286 copy(rec[indexHeaderLen:], typ)
287 _, err := f.Write(rec)
288 return err
289 }
290
291 // Close closes the index log.
292 func (s *Server) Close() error {
293 s.mu.Lock()
294 defer s.mu.Unlock()
295 if s.log == nil {
296 return nil
297 }
298 err := s.log.Close()
299 s.log = nil
300 return err
301 }
302
303 // Dir returns the Blossom store root.
304 func (s *Server) Dir() string { return s.dir }
305
306 func (s *Server) blobPath(hash string) string {
307 return filepath.Join(s.dir, blobDirName, hash[:2], hash)
308 }
309
310 // Serve routes a Blossom request and returns the response. The boolean is
311 // false when the path does not belong to Blossom.
312 func (s *Server) Serve(req *Request) (Response, bool) {
313 w := &respWriter{}
314 handled := s.handle(w, req)
315 status := w.status
316 if status == 0 {
317 status = 200
318 }
319 return Response{Status: status, Header: w.header, Body: w.body}, handled
320 }
321
322 // normalizePath accepts both the root Blossom layout and the common /blossom
323 // mount used by web clients and reverse proxies.
324 func normalizePath(path string) string {
325 switch {
326 case path == "/blossom":
327 return "/"
328 case strings.HasPrefix(path, "/blossom/"):
329 return strings.TrimPrefix(path, "/blossom")
330 default:
331 return path
332 }
333 }
334
335 // blobHash accepts /<sha256> and /<sha256>.<extension>, the BUD-01 GET shape.
336 func blobHash(path string) (string, bool) {
337 name := strings.TrimPrefix(path, "/")
338 if len(name) < 64 {
339 return "", false
340 }
341 hash := name[:64]
342 if len(name) > 64 && name[64] != '.' {
343 return "", false
344 }
345 if !nostr.IsHex64(hash) {
346 return "", false
347 }
348 return hash, true
349 }
350
351 func (s *Server) handle(w *respWriter, req *Request) bool {
352 path := normalizePath(req.Path)
353 req.Path = path
354 if req.Method == "OPTIONS" {
355 if path == "/upload" || strings.HasPrefix(path, "/list/") {
356 w.WriteHeader(204)
357 return true
358 }
359 if _, ok := blobHash(path); ok {
360 w.WriteHeader(204)
361 return true
362 }
363 return false
364 }
365 switch {
366 case path == "/upload":
367 s.handleUpload(w, req)
368 return true
369 case strings.HasPrefix(path, "/list/"):
370 s.handleList(w, req)
371 return true
372 default:
373 hash, ok := blobHash(path)
374 if !ok {
375 return false
376 }
377 s.handleBlob(w, req, hash)
378 return true
379 }
380 }
381
382 func (s *Server) handleBlob(w *respWriter, req *Request, hash string) {
383 switch req.Method {
384 case "GET", "HEAD":
385 s.mu.Lock()
386 b := s.blobs[hash]
387 s.mu.Unlock()
388 if b == nil {
389 s.err(w, 404, "blob not found")
390 return
391 }
392 data, err := os.ReadFile(s.blobPath(hash))
393 if err != nil {
394 s.err(w, 404, "blob not found")
395 return
396 }
397 w.SetHeader("content-type", b.Type)
398 w.SetHeader("content-length", fmt.Sprint(len(data)))
399 w.SetHeader("etag", `"`+hash+`"`)
400 w.SetHeader("cache-control", "public, max-age=31536000, immutable")
401 w.WriteHeader(200)
402 if req.Method == "GET" {
403 w.Write(data)
404 }
405 case "DELETE":
406 pub, err := s.authorize(req, "delete", hash)
407 if err != nil {
408 s.err(w, 401, err.Error())
409 return
410 }
411 s.mu.Lock()
412 b := s.blobs[hash]
413 owned := b != nil && b.PubKey == nostr.Hex(pub[:])
414 s.mu.Unlock()
415 if b == nil {
416 s.err(w, 404, "blob not found")
417 return
418 }
419 if !owned {
420 s.err(w, 403, "not the blob owner")
421 return
422 }
423 os.Remove(s.blobPath(hash))
424 s.mu.Lock()
425 s.delMem(hash)
426 err = s.appendRecord('d', b)
427 s.mu.Unlock()
428 if err != nil {
429 s.err(w, 500, "could not update index")
430 return
431 }
432 w.WriteHeader(204)
433 default:
434 s.err(w, 405, "method not allowed")
435 }
436 }
437
438 func (s *Server) handleUpload(w *respWriter, req *Request) {
439 switch req.Method {
440 case "HEAD":
441 s.handleUploadCheck(w, req)
442 case "PUT":
443 s.handlePut(w, req)
444 default:
445 s.err(w, 405, "method not allowed")
446 }
447 }
448
449 func (s *Server) handleUploadCheck(w *respWriter, req *Request) {
450 hash := req.Header["x-sha-256"]
451 pub, err := s.authorize(req, "upload", hash)
452 _ = pub
453 if err != nil {
454 s.err(w, 401, err.Error())
455 return
456 }
457 if hash == "" {
458 s.err(w, 400, "missing X-SHA-256")
459 return
460 }
461 if !nostr.IsHex64(hash) {
462 s.err(w, 400, "malformed X-SHA-256")
463 return
464 }
465 s.mu.Lock()
466 b := s.blobs[hash]
467 s.mu.Unlock()
468 if b == nil {
469 s.err(w, 404, "blob not found")
470 return
471 }
472 w.SetHeader("x-sha-256", hash)
473 w.SetHeader("content-length", fmt.Sprint(b.Size))
474 w.SetHeader("content-type", b.Type)
475 w.WriteHeader(200)
476 }
477
478 func (s *Server) handlePut(w *respWriter, req *Request) {
479 if int64(len(req.Body)) > s.maxBlob {
480 s.err(w, 413, "blob is too large")
481 return
482 }
483 sum := sha256Bytes(req.Body)
484 hash := nostr.Hex(sum[:])
485 pub, err := s.authorize(req, "upload", hash)
486 if err != nil {
487 s.err(w, 401, err.Error())
488 return
489 }
490 ct := req.Header["content-type"]
491 if ct == "" {
492 ct = "application/octet-stream"
493 }
494 if len(ct) > 1024 {
495 ct = ct[:1024]
496 }
497 now := time.Now().Unix()
498
499 s.mu.Lock()
500 existing := s.blobs[hash]
501 if existing != nil {
502 desc := s.descriptor(req, existing)
503 s.mu.Unlock()
504 s.writeJSON(w, 200, desc)
505 return
506 }
507 dir := filepath.Dir(s.blobPath(hash))
508 if err := os.MkdirAll(dir, 0o755); err != nil {
509 s.mu.Unlock()
510 s.err(w, 500, "could not create blob directory")
511 return
512 }
513 tmp := s.blobPath(hash) + ".tmp"
514 if err := os.WriteFile(tmp, req.Body, 0o644); err != nil {
515 s.mu.Unlock()
516 s.err(w, 500, "could not write blob")
517 return
518 }
519 if err := os.Rename(tmp, s.blobPath(hash)); err != nil {
520 s.mu.Unlock()
521 s.err(w, 500, "could not commit blob")
522 return
523 }
524 b := &Blob{Hash: hash, PubKey: nostr.Hex(pub[:]), Size: int64(len(req.Body)), Type: ct, Created: now}
525 s.putMem(b)
526 if err := s.appendRecord('p', b); err != nil {
527 s.mu.Unlock()
528 s.err(w, 500, "could not update index")
529 return
530 }
531 desc := s.descriptor(req, b)
532 s.mu.Unlock()
533 s.writeJSON(w, 200, desc)
534 }
535
536 func (s *Server) handleList(w *respWriter, req *Request) {
537 if req.Method != "GET" && req.Method != "HEAD" {
538 s.err(w, 405, "method not allowed")
539 return
540 }
541 pubHex := strings.TrimPrefix(req.Path, "/list/")
542 if !nostr.IsHex64(pubHex) {
543 s.err(w, 400, "malformed pubkey")
544 return
545 }
546 if authHeader(req) != "" {
547 pub, err := s.authorize(req, "list", "")
548 if err != nil {
549 s.err(w, 401, err.Error())
550 return
551 }
552 if nostr.Hex(pub[:]) != pubHex {
553 s.err(w, 403, "cannot list another pubkey's blobs")
554 return
555 }
556 }
557 s.mu.Lock()
558 var descs []Descriptor
559 for _, b := range s.byPub[pubHex] {
560 descs = append(descs, s.descriptor(req, b))
561 }
562 s.mu.Unlock()
563 // Stable order for clients and tests.
564 for i := 1; i < len(descs); i++ {
565 for j := i; j > 0 && descs[j].Uploaded < descs[j-1].Uploaded; j-- {
566 descs[j], descs[j-1] = descs[j-1], descs[j]
567 }
568 }
569 body, _ := json.Marshal(descs)
570 w.SetHeader("content-type", "application/json")
571 w.SetHeader("content-length", fmt.Sprint(len(body)))
572 w.WriteHeader(200)
573 if req.Method == "GET" {
574 w.Write(body)
575 }
576 }
577
578 // extensionForType returns the conventional file extension for a MIME type.
579 // Blossom clients expect the descriptor URL to carry it so the URL can be used
580 // directly as a media source.
581 func extensionForType(typ string) string {
582 if i := strings.IndexByte(typ, ';'); i >= 0 {
583 typ = typ[:i]
584 }
585 switch strings.ToLower(strings.TrimSpace(typ)) {
586 case "image/jpeg":
587 return ".jpg"
588 case "image/png":
589 return ".png"
590 case "image/gif":
591 return ".gif"
592 case "image/webp":
593 return ".webp"
594 case "image/avif":
595 return ".avif"
596 case "image/svg+xml":
597 return ".svg"
598 case "video/mp4":
599 return ".mp4"
600 case "video/webm":
601 return ".webm"
602 case "video/quicktime":
603 return ".mov"
604 case "audio/mpeg":
605 return ".mp3"
606 case "audio/mp4":
607 return ".m4a"
608 case "audio/wav", "audio/x-wav":
609 return ".wav"
610 case "audio/ogg":
611 return ".ogg"
612 case "application/pdf":
613 return ".pdf"
614 case "application/json":
615 return ".json"
616 case "application/zip":
617 return ".zip"
618 case "application/gzip":
619 return ".gz"
620 case "application/x-tar":
621 return ".tar"
622 case "text/plain":
623 return ".txt"
624 case "text/html":
625 return ".html"
626 case "text/css":
627 return ".css"
628 case "application/javascript", "text/javascript":
629 return ".js"
630 case "application/xml", "text/xml":
631 return ".xml"
632 default:
633 return ""
634 }
635 }
636
637 func (s *Server) descriptor(req *Request, b *Blob) Descriptor {
638 scheme := "http"
639 if strings.EqualFold(req.Header["x-forwarded-proto"], "https") {
640 scheme = "https"
641 }
642 host := req.Header["host"]
643 if host == "" {
644 host = "localhost"
645 }
646 return Descriptor{
647 URL: fmt.Sprintf("%s://%s/%s%s", scheme, host, b.Hash, extensionForType(b.Type)),
648 SHA256: b.Hash,
649 Size: b.Size,
650 Type: b.Type,
651 Uploaded: b.Created,
652 }
653 }
654
655 func (s *Server) authorize(req *Request, action, hash string) ([32]byte, error) {
656 return verifyAuth(authHeader(req), action, hash, time.Now().Unix())
657 }
658
659 func authHeader(req *Request) string {
660 return strings.TrimSpace(req.Header["authorization"])
661 }
662
663 func (s *Server) writeJSON(w *respWriter, status int, v any) {
664 body, err := json.Marshal(v)
665 if err != nil {
666 s.err(w, 500, "could not encode response")
667 return
668 }
669 w.SetHeader("content-type", "application/json")
670 w.SetHeader("content-length", fmt.Sprint(len(body)))
671 w.WriteHeader(status)
672 w.Write(body)
673 }
674
675 func (s *Server) err(w *respWriter, status int, msg string) {
676 body, _ := json.Marshal(map[string]string{"error": msg})
677 w.SetHeader("content-type", "application/json")
678 w.SetHeader("content-length", fmt.Sprint(len(body)))
679 w.WriteHeader(status)
680 w.Write(body)
681 }
682