tinyrelayfile
tinyrelay/internal/blob/service.go/
internal/blob/service.go raw
1package blob
2
3import (
4 "context"
5 "database/sql"
6 "errors"
7 "fmt"
8 "github.com/FelineStateMachine/tinyrelay/internal/storage"
9 "io"
10 "net"
11 "net/http"
12 "os"
13 "path/filepath"
14 "regexp"
15 "strings"
16 "sync"
17)
18
19// Action identifies the file-door operation being authorized.
20type Action string
21
22const (
23 ActionUpload Action = "upload"
24 ActionGet Action = "get"
25 ActionDelete Action = "delete"
26 ActionList Action = "list"
27 ActionMirror Action = "mirror"
28 ActionReport Action = "report"
29)
30
31// Authorize authenticates an HTTP request for an action and returns the
32// caller's pubkey.
33type Authorize func(request *http.Request, action Action) (pubkey string, err error)
34
35// CanRead reports whether any supplied reader pubkey may read the blob.
36type CanRead func(ctx context.Context, blobSHA string, readerPubkeys []string) bool
37
38// ResolveIP resolves a host for mirror URL policy checks.
39type ResolveIP func(ctx context.Context, host string) ([]net.IP, error)
40
41// ReportRecorder records a report in the host moderation transaction. The
42// targetKind and reportType values describe the reported object and report;
43// blob reports currently pass "blob" for both.
44type ReportRecorder func(ctx context.Context, tx *sql.Tx, reporter string, targetBlob string, targetKind string, reportType string, content string) error
45
46// Config supplies durable storage, HTTP policy and host integration hooks.
47type Config struct {
48 Root string
49 PublicURL string
50 Store *storage.Store
51 Authorize Authorize
52 CanRead CanRead
53 HTTPClient *http.Client
54 ResolveIP ResolveIP
55 IsOwner func(string) bool
56 ValidateUpload func(*http.Request, string) error
57 // ResolveServers returns the ordered BUD-03 servers for an authorized
58 // caller. It is used only for Blossom URI mirror fallback.
59 ResolveServers func(context.Context, string) ([]string, error)
60 // RecordReport joins native blob reporting to the host moderation ledger.
61 // It runs inside the blob report transaction.
62 RecordReport ReportRecorder
63 // Limits is read for each upload so policy changes apply without reopening
64 // the blob service. Zero values mean unlimited.
65 Limits func() Limits
66 // UploadTerms returns what an uploader's standing imposes on its uploads:
67 // an agent grant with a sites ttl gives its claims an expiry, and one with
68 // the encrypted flag refuses plain uploads. Zero values mean no terms.
69 UploadTerms func(context.Context, string) (UploadTerms, error)
70}
71
72// Limits are applied when each upload starts. Zero leaves that limit open.
73type Limits struct {
74 MaxFileBytes int64
75 UserStorageBytes int64
76}
77
78// UploadTerms is what an uploader's grant imposes on the blobs it stores.
79// ExpiresAt is the unix time its claims lapse, 0 for never; Encrypted
80// requires the stored bytes to be an encrypted blob or manifest.
81type UploadTerms struct {
82 ExpiresAt int64
83 Encrypted bool
84}
85
86// encryptedTypes are the content types the browser uploader gives encrypted
87// objects: AES-GCM ciphertext for files and chunks, and the BUD-16 directory
88// type for encrypted manifests. Anything the relay recognizes as a page,
89// image, video, audio or document is plain.
90var encryptedTypes = map[string]bool{"application/octet-stream": true, "application/vnd.blossom.directory+msgpack": true}
91
92// Blob is the metadata for one content-addressed object.
93type Blob struct {
94 SHA256 string `json:"sha256"`
95 Size int64 `json:"size"`
96 Type string `json:"type"`
97 Uploader string `json:"uploader"`
98 Uploaded int64 `json:"uploaded"`
99}
100
101// BlobClaimMetadata names a blob for one uploader. Names are claim-scoped so
102// deduplicated blobs can have different names for different uploaders.
103type BlobClaimMetadata struct {
104 SHA256 string `json:"sha256"`
105 Uploader string `json:"uploader"`
106 Name string `json:"name"`
107 Path string `json:"path"`
108 Purpose string `json:"purpose,omitempty"`
109 Size int64 `json:"size,omitempty"`
110 Type string `json:"type,omitempty"`
111 UpdatedAt int64 `json:"updated_at"`
112}
113
114// PutOptions describes a blob upload and its optional metadata transaction.
115type PutOptions struct {
116 Reader io.Reader
117 Type string
118 Uploader string
119 Hash string
120 Name string
121 Path string
122 // Access controls who may retrieve the stored bytes. Empty means public.
123 Access string
124 // Purpose identifies logical encrypted roots or their internal chunks.
125 Purpose string
126 // Commit records ownership metadata in the blob transaction. Its isNew
127 // argument reports whether the blob was newly installed. An error rolls
128 // back all claims.
129 // New bytes remain unavailable until their ownership metadata is durable.
130 Commit func(ctx context.Context, tx *sql.Tx, blob Blob, isNew bool) error
131}
132
133// Service owns blob files and their metadata operations.
134type Service struct {
135 config Config
136 root string
137 store *storage.Store
138 resolve ResolveIP
139 quotaMu sync.Mutex
140 // multipartMu protects partial files and their range metadata. Lock it
141 // before quotaMu; network bodies are never read while either is held.
142 multipartMu sync.Mutex
143 activeParts map[string]int
144 // multipartFinalizing prevents another chunk from changing a complete
145 // session while its assembled file is being hashed. Hashing is deliberately
146 // outside multipartMu because the upload may be as large as 1 GiB.
147 multipartFinalizing map[string]bool
148 // verifyMultipartHook is used by package tests to pause finalization
149 // without allocating a GiB fixture. It is nil in production.
150 verifyMultipartHook func(context.Context, string, string, int64) error
151}
152
153var ErrBlocked = errors.New("blocked: this blob was removed by a moderator")
154var ErrHashMismatch = errors.New("conflict: mirrored blob hash does not match the requested hash")
155var ErrFileTooLarge = errors.New("invalid: blob exceeds the per-file size limit")
156var ErrQuotaExceeded = errors.New("restricted: uploader storage quota exceeded")
157var ErrMultipartConflict = errors.New("conflict: upload metadata differs from existing session")
158var ErrNotEncrypted = errors.New("restricted: agent grant requires encrypted uploads")
159var ErrAccessConflict = errors.New("conflict: the blob already has a different access policy")
160var ErrInvalidAccess = errors.New("invalid: access must be public or members")
161
162const (
163 AccessPublic = "public"
164 AccessMembers = "members"
165)
166
167var shaPattern = regexp.MustCompile(`^[0-9a-f]{64}$`)
168var blobPathPattern = regexp.MustCompile(`^/([0-9a-f]{64})(?:\.[a-z0-9]{1,8})?$`)
169
170// HandlesPath keeps the daemon mount in sync with every supported file door.
171func HandlesPath(path string) bool {
172 return path == "/upload" || path == "/mirror" || path == "/report" ||
173 path == "/files/metadata" ||
174 path == "/nip96" || path == "/.well-known/nostr/nip96.json" ||
175 strings.HasPrefix(path, "/nip96/") || strings.HasPrefix(path, "/list/") ||
176 blobPathPattern.MatchString(path)
177}
178
179var extensions = map[string]string{
180 "image/png": "png", "image/jpeg": "jpg", "image/gif": "gif", "image/webp": "webp", "image/avif": "avif", "image/svg+xml": "svg",
181 "video/mp4": "mp4", "video/webm": "webm", "video/quicktime": "mov", "audio/mpeg": "mp3", "audio/ogg": "ogg", "audio/wav": "wav", "audio/mp4": "m4a",
182 "application/pdf": "pdf", "application/json": "json", "text/plain": "txt", "text/markdown": "md",
183 "application/vnd.blossom.directory+msgpack": "bdir",
184}
185
186func New(ctx context.Context, config Config) (*Service, error) {
187 if config.Root == "" || config.Store == nil {
188 return nil, errors.New("blob root and store are required")
189 }
190 if err := os.MkdirAll(filepath.Join(config.Root, "blobs"), 0o700); err != nil {
191 return nil, fmt.Errorf("create blob root: %w", err)
192 }
193 if _, err := config.Store.DB().ExecContext(ctx, `CREATE TABLE IF NOT EXISTS blobs (
194 sha256 TEXT PRIMARY KEY, size INTEGER NOT NULL, type TEXT NOT NULL,
195 uploader TEXT NOT NULL, uploaded INTEGER NOT NULL, access TEXT NOT NULL DEFAULT 'public'
196 ); CREATE TABLE IF NOT EXISTS blob_claims (
197 sha256 TEXT NOT NULL, uploader TEXT NOT NULL, claimed_at INTEGER NOT NULL,
198 PRIMARY KEY (sha256, uploader), FOREIGN KEY (sha256) REFERENCES blobs(sha256) ON DELETE CASCADE
199 ); CREATE INDEX IF NOT EXISTS blobs_uploader ON blobs(uploader, uploaded DESC);
200 CREATE INDEX IF NOT EXISTS blob_claims_uploader ON blob_claims(uploader, claimed_at DESC, sha256 DESC);
201 CREATE TABLE IF NOT EXISTS blob_claim_metadata (
202 sha256 TEXT NOT NULL, uploader TEXT NOT NULL, name TEXT NOT NULL, path TEXT NOT NULL,
203 purpose TEXT NOT NULL DEFAULT '', logical_size INTEGER NOT NULL DEFAULT 0, logical_type TEXT NOT NULL DEFAULT '',
204 updated_at INTEGER NOT NULL, PRIMARY KEY (sha256,uploader,path),
205 FOREIGN KEY (sha256,uploader) REFERENCES blob_claims(sha256,uploader) ON DELETE CASCADE
206 );
207 CREATE INDEX IF NOT EXISTS blob_claim_metadata_uploader ON blob_claim_metadata(uploader,updated_at DESC,sha256 DESC);
208 CREATE TABLE IF NOT EXISTS blob_reports (
209 id TEXT PRIMARY KEY, reporter TEXT NOT NULL, target_pubkey TEXT NOT NULL DEFAULT '',
210 target_blob TEXT NOT NULL, type TEXT NOT NULL, content TEXT NOT NULL, at INTEGER NOT NULL,
211 status TEXT NOT NULL DEFAULT 'open', resolved_by TEXT NOT NULL DEFAULT '',
212 resolved_at INTEGER NOT NULL DEFAULT 0, action TEXT NOT NULL DEFAULT '',
213 UNIQUE(reporter,target_blob));
214 CREATE TRIGGER IF NOT EXISTS blob_claim_on_insert AFTER INSERT ON blobs
215 WHEN NEW.uploader != '' BEGIN INSERT OR IGNORE INTO blob_claims(sha256,uploader,claimed_at) VALUES(NEW.sha256,NEW.uploader,NEW.uploaded); END;
216 CREATE TABLE IF NOT EXISTS blob_blocks (sha256 TEXT PRIMARY KEY, reason TEXT NOT NULL DEFAULT '', blocked_at INTEGER NOT NULL DEFAULT 0);
217 CREATE TABLE IF NOT EXISTS blob_tombstones (sha256 TEXT PRIMARY KEY, deleted_at INTEGER NOT NULL);
218 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, name TEXT NOT NULL DEFAULT '', path TEXT NOT NULL DEFAULT '', access TEXT NOT NULL DEFAULT 'public', purpose TEXT NOT NULL DEFAULT '');
219 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);
220 INSERT OR IGNORE INTO blob_claims(sha256,uploader,claimed_at)
221 SELECT sha256,uploader,uploaded FROM blobs WHERE uploader != ''`); err != nil {
222 return nil, fmt.Errorf("initialize blob metadata: %w", err)
223 }
224 if _, err := config.Store.DB().ExecContext(ctx, "ALTER TABLE blobs ADD COLUMN access TEXT NOT NULL DEFAULT 'public'"); err != nil && !strings.Contains(err.Error(), "duplicate column") {
225 return nil, fmt.Errorf("add blob access: %w", err)
226 }
227 for _, column := range []string{"purpose TEXT NOT NULL DEFAULT ''", "logical_size INTEGER NOT NULL DEFAULT 0", "logical_type TEXT NOT NULL DEFAULT ''"} {
228 if _, err := config.Store.DB().ExecContext(ctx, "ALTER TABLE blob_claim_metadata ADD COLUMN "+column); err != nil && !strings.Contains(err.Error(), "duplicate column") {
229 return nil, fmt.Errorf("add blob claim metadata field: %w", err)
230 }
231 }
232 // Claims made under an agent grant with a ttl carry an expiry; every
233 // earlier claim keeps 0, which never lapses.
234 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") {
235 return nil, fmt.Errorf("add blob claim expiry: %w", err)
236 }
237 for _, column := range []string{"name TEXT NOT NULL DEFAULT ''", "path TEXT NOT NULL DEFAULT ''"} {
238 if _, err := config.Store.DB().ExecContext(ctx, "ALTER TABLE multipart_uploads ADD COLUMN "+column); err != nil && !strings.Contains(err.Error(), "duplicate column") {
239 return nil, fmt.Errorf("add multipart metadata: %w", err)
240 }
241 }
242 if _, err := config.Store.DB().ExecContext(ctx, "ALTER TABLE multipart_uploads ADD COLUMN access TEXT NOT NULL DEFAULT 'public'"); err != nil && !strings.Contains(err.Error(), "duplicate column") {
243 return nil, fmt.Errorf("add multipart access: %w", err)
244 }
245 if _, err := config.Store.DB().ExecContext(ctx, "ALTER TABLE multipart_uploads ADD COLUMN purpose TEXT NOT NULL DEFAULT ''"); err != nil && !strings.Contains(err.Error(), "duplicate column") {
246 return nil, fmt.Errorf("add multipart purpose: %w", err)
247 }
248 if _, err := config.Store.DB().ExecContext(ctx, "CREATE INDEX IF NOT EXISTS blob_claims_expires ON blob_claims(expires) WHERE expires>0"); err != nil {
249 return nil, fmt.Errorf("index blob claim expiry: %w", err)
250 }
251 if err := reconcile(ctx, config.Store, filepath.Join(config.Root, "blobs")); err != nil {
252 return nil, err
253 }
254 resolve := config.ResolveIP
255 if resolve == nil {
256 resolve = func(ctx context.Context, host string) ([]net.IP, error) {
257 return net.DefaultResolver.LookupIP(ctx, "ip", host)
258 }
259 }
260 s := &Service{config: config, root: filepath.Join(config.Root, "blobs"), store: config.Store, resolve: resolve, activeParts: make(map[string]int), multipartFinalizing: make(map[string]bool)}
261 if err := s.reconcileMultipart(ctx); err != nil {
262 return nil, err
263 }
264 return s, nil
265}
266
267func (s *Service) Handler() http.HandlerFunc { return s.serveHTTP }

About

ownerc01503e8035f
visibilitypublic
headd730679639bf

Maintainers

  • c01503e8035f owner
  • ba1b5beed5b5 agent hermes

Clone

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

Refs