| 1 | diff --git a/internal/blob/blob.go b/internal/blob/blob.go |
| 2 | index 5d24940..2a7f45e 100644 |
| 3 | --- a/internal/blob/blob.go |
| 4 | +++ b/internal/blob/blob.go |
| 5 | @@ -801,11 +801,8 @@ func (s *Service) uploadTerms(ctx context.Context, uploader string) (UploadTerms |
| 6 | if s.config.UploadTerms == nil || uploader == "" { |
| 7 | return UploadTerms{}, nil |
| 8 | } |
| 9 | - terms, err := s.config.UploadTerms(ctx, uploader) |
| 10 | - if err != nil { |
| 11 | - return UploadTerms{}, fmt.Errorf("check upload terms: %w", err) |
| 12 | - } |
| 13 | - return terms, nil |
| 14 | + // A refusal keeps its reason so the door answers with the right status. |
| 15 | + return s.config.UploadTerms(ctx, uploader) |
| 16 | } |
| 17 | |
| 18 | // SweepExpired removes the blobs whose every claim has lapsed and releases |
| 19 | @@ -1408,6 +1405,9 @@ func statusFor(err error) int { |
| 20 | if strings.HasPrefix(err.Error(), "invalid:") { |
| 21 | return http.StatusBadRequest |
| 22 | } |
| 23 | + if strings.HasPrefix(err.Error(), "restricted:") { |
| 24 | + return http.StatusForbidden |
| 25 | + } |
| 26 | return http.StatusInternalServerError |
| 27 | } |
| 28 | |
| 29 | diff --git a/internal/daemon/blob_authorization_test.go b/internal/daemon/blob_authorization_test.go |
| 30 | index d9d4d4c..4c49fae 100644 |
| 31 | --- a/internal/daemon/blob_authorization_test.go |
| 32 | +++ b/internal/daemon/blob_authorization_test.go |
| 33 | @@ -18,8 +18,99 @@ import ( |
| 34 | |
| 35 | "github.com/FelineStateMachine/tinyrelay/internal/blob" |
| 36 | "github.com/FelineStateMachine/tinyrelay/internal/event" |
| 37 | + "github.com/FelineStateMachine/tinyrelay/internal/sites" |
| 38 | ) |
| 39 | |
| 40 | +// agentUpload sends one Blossom PUT /upload signed by secret through the |
| 41 | +// tenant and returns the response. |
| 42 | +func agentUpload(t *testing.T, tenant *Tenant, secret, body, typ string) *httptest.ResponseRecorder { |
| 43 | + t.Helper() |
| 44 | + sum := sha256.Sum256([]byte(body)) |
| 45 | + token := event.Event{Kind: 24242, CreatedAt: time.Now().Unix(), Tags: [][]string{{"t", "upload"}, {"x", hex.EncodeToString(sum[:])}, {"expiration", strconv.FormatInt(time.Now().Unix()+60, 10)}}, Content: "Authorize file storage"} |
| 46 | + if err := event.Sign(&token, secret); err != nil { |
| 47 | + t.Fatal(err) |
| 48 | + } |
| 49 | + raw, err := json.Marshal(token) |
| 50 | + if err != nil { |
| 51 | + t.Fatal(err) |
| 52 | + } |
| 53 | + r := httptest.NewRequest(http.MethodPut, "http://relay.test/upload", strings.NewReader(body)) |
| 54 | + r.Header.Set("Authorization", "Nostr "+base64.RawURLEncoding.EncodeToString(raw)) |
| 55 | + r.Header.Set("Content-Type", typ) |
| 56 | + w := httptest.NewRecorder() |
| 57 | + tenant.ServeHTTP(w, r) |
| 58 | + return w |
| 59 | +} |
| 60 | + |
| 61 | +// TestAgentGrantGovernsUploads covers the blob door for agent keys: an |
| 62 | +// active grant uploads as a member, a paused or revoked one is refused, a |
| 63 | +// sites ttl stamps the claim and the maintenance sweep retires it, and an |
| 64 | +// encrypted grant refuses plain files. |
| 65 | +func TestAgentGrantGovernsUploads(t *testing.T) { |
| 66 | + _, tenant := testTenant(t) |
| 67 | + ctx := context.Background() |
| 68 | + owner := tenant.Policy().Owner |
| 69 | + agent, _ := event.PublicKey(testAgentSecret) |
| 70 | + if _, err := tenant.Execute(ctx, owner, "setpolicy", []json.RawMessage{json.RawMessage(`{"writes":"allowlist"}`)}); err != nil { |
| 71 | + t.Fatal(err) |
| 72 | + } |
| 73 | + if w := agentUpload(t, tenant, testAgentSecret, "stranger", "text/plain"); w.Code == http.StatusOK { |
| 74 | + t.Fatal("a key without a grant uploaded under allowlist writes") |
| 75 | + } |
| 76 | + now := time.Now().Unix() |
| 77 | + own := sites.SiteLabel(event.Event{Kind: sites.KindSite, PubKey: agent}) |
| 78 | + if err := publishAs(t, tenant, agentGrantEvent(t, agent, now, now+3600, []string{"sites", own, "ttl=2"})); err != nil { |
| 79 | + t.Fatal(err) |
| 80 | + } |
| 81 | + if w := agentUpload(t, tenant, testAgentSecret, "site file", "text/html"); w.Code != http.StatusOK { |
| 82 | + t.Fatalf("granted agent upload = %d %s", w.Code, w.Body.String()) |
| 83 | + } |
| 84 | + sum := sha256.Sum256([]byte("site file")) |
| 85 | + hash := hex.EncodeToString(sum[:]) |
| 86 | + var expires int64 |
| 87 | + if err := tenant.store.DB().QueryRow("SELECT expires FROM blob_claims WHERE sha256=? AND uploader=?", hash, agent).Scan(&expires); err != nil || expires < now+2*86400 || expires > now+2*86400+60 { |
| 88 | + t.Fatalf("claim expires=%d err=%v, want about %d", expires, err, now+2*86400) |
| 89 | + } |
| 90 | + if entries, err := tenant.blobs.ListMetadata(ctx, agent); err != nil || len(entries) != 1 { |
| 91 | + t.Fatalf("agent list = %d entries, err=%v", len(entries), err) |
| 92 | + } |
| 93 | + if _, err := tenant.Execute(ctx, owner, "pauseagent", []json.RawMessage{json.RawMessage(strconv.Quote(agent))}); err != nil { |
| 94 | + t.Fatal(err) |
| 95 | + } |
| 96 | + if w := agentUpload(t, tenant, testAgentSecret, "while paused", "text/html"); w.Code == http.StatusOK || !strings.Contains(w.Body.String(), "paused") { |
| 97 | + t.Fatalf("paused agent upload = %d %s", w.Code, w.Body.String()) |
| 98 | + } |
| 99 | + if _, err := tenant.Execute(ctx, owner, "resumeagent", []json.RawMessage{json.RawMessage(strconv.Quote(agent))}); err != nil { |
| 100 | + t.Fatal(err) |
| 101 | + } |
| 102 | + if w := agentUpload(t, tenant, testAgentSecret, "after resume", "text/html"); w.Code != http.StatusOK { |
| 103 | + t.Fatalf("resumed agent upload = %d %s", w.Code, w.Body.String()) |
| 104 | + } |
| 105 | + if _, err := tenant.Execute(ctx, owner, "revokeagent", []json.RawMessage{json.RawMessage(strconv.Quote(agent))}); err != nil { |
| 106 | + t.Fatal(err) |
| 107 | + } |
| 108 | + if w := agentUpload(t, tenant, testAgentSecret, "after revoke", "text/html"); w.Code == http.StatusOK { |
| 109 | + t.Fatalf("revoked agent upload = %d %s", w.Code, w.Body.String()) |
| 110 | + } |
| 111 | + // The sweep removes the agent's files once the ttl lapses. |
| 112 | + if err := tenant.sweepExpiredBlobs(ctx, now+2*86400+61); err != nil { |
| 113 | + t.Fatal(err) |
| 114 | + } |
| 115 | + if _, _, err := tenant.blobs.Get(ctx, hash); err == nil { |
| 116 | + t.Fatal("expired agent upload still stored") |
| 117 | + } |
| 118 | + // A fresh grant with the encrypted flag refuses plain uploads. |
| 119 | + if err := publishAs(t, tenant, agentGrantEvent(t, agent, now+1, now+3600, []string{"sites", "*", "ttl=1", "encrypted"})); err != nil { |
| 120 | + t.Fatal(err) |
| 121 | + } |
| 122 | + if w := agentUpload(t, tenant, testAgentSecret, "<html>plain</html>", "text/html"); w.Code != http.StatusForbidden || !strings.Contains(w.Body.String(), "requires encrypted uploads") { |
| 123 | + t.Fatalf("plain upload under encrypted grant = %d %s", w.Code, w.Body.String()) |
| 124 | + } |
| 125 | + if w := agentUpload(t, tenant, testAgentSecret, "\x8f\x1a\x9c\x03"+strings.Repeat("\xe2\x71\x05\x9b\x4c", 20), "application/octet-stream"); w.Code != http.StatusOK { |
| 126 | + t.Fatalf("ciphertext upload under encrypted grant = %d %s", w.Code, w.Body.String()) |
| 127 | + } |
| 128 | +} |
| 129 | + |
| 130 | type blobOriginTransport func(*http.Request) (*http.Response, error) |
| 131 | |
| 132 | func (f blobOriginTransport) RoundTrip(r *http.Request) (*http.Response, error) { return f(r) } |
| 133 | diff --git a/internal/daemon/scheduler.go b/internal/daemon/scheduler.go |
| 134 | index 2722aa4..596ab32 100644 |
| 135 | --- a/internal/daemon/scheduler.go |
| 136 | +++ b/internal/daemon/scheduler.go |
| 137 | @@ -48,6 +48,7 @@ func (t *Tenant) runScheduler() { |
| 138 | if t.blobs != nil { |
| 139 | report("partial uploads", t.blobs.CleanupMultipart(ctx)) |
| 140 | } |
| 141 | + report("expired uploads", t.sweepExpiredBlobs(ctx, now.Unix())) |
| 142 | report("retention", t.sweep(ctx, now.Unix())) |
| 143 | report("restore-state", t.sweepReplicationState(ctx, now.Unix())) |
| 144 | nextMaintenance = now.Add(time.Minute) |
| 145 | diff --git a/internal/daemon/services.go b/internal/daemon/services.go |
| 146 | index c4f6240..0e07ec5 100644 |
| 147 | --- a/internal/daemon/services.go |
| 148 | +++ b/internal/daemon/services.go |
| 149 | @@ -47,7 +47,7 @@ func (t *Tenant) initServices(ctx context.Context) error { |
| 150 | return err |
| 151 | } |
| 152 | t.blobs, err = blob.New(ctx, blob.Config{ |
| 153 | - Root: t.meta.Paths.Root, PublicURL: t.publicURL, Store: t.store, Authorize: t.authorizeBlob, |
| 154 | + Root: t.meta.Paths.Root, PublicURL: t.publicURL, Store: t.store, Authorize: t.authorizeBlob, UploadTerms: t.blobUploadTerms, |
| 155 | Limits: func() blob.Limits { |
| 156 | limits := t.Policy().FileLimits |
| 157 | return blob.Limits{MaxFileBytes: limits.MaxFileBytes, UserStorageBytes: limits.UserStorageBytes} |
| 158 | @@ -310,10 +310,81 @@ func (t *Tenant) authorizeBlob(r *http.Request, action blob.Action) (string, err |
| 159 | if !allowed { |
| 160 | return "", errors.New("restricted: file upload is not allowed") |
| 161 | } |
| 162 | + // An agent's standing is its grant: an active one makes it a member |
| 163 | + // for uploads, a paused, revoked or expired one stops them. |
| 164 | + if role == "agent" { |
| 165 | + if _, err := t.agentUploadGrant(r.Context(), pubkey); err != nil { |
| 166 | + return "", err |
| 167 | + } |
| 168 | + } |
| 169 | } |
| 170 | return pubkey, nil |
| 171 | } |
| 172 | |
| 173 | +// agentUploadGrant returns the grant behind an agent key's uploads when it |
| 174 | +// currently allows publishing. |
| 175 | +func (t *Tenant) agentUploadGrant(ctx context.Context, pubkey string) (community.AgentGrant, error) { |
| 176 | + grant, ok, err := t.community.AgentGrant(ctx, pubkey) |
| 177 | + if err != nil { |
| 178 | + return community.AgentGrant{}, fmt.Errorf("check agent upload grant: %w", err) |
| 179 | + } |
| 180 | + if !ok { |
| 181 | + return community.AgentGrant{}, errors.New("restricted: agent grant is missing") |
| 182 | + } |
| 183 | + switch { |
| 184 | + case grant.RevokedAt > 0: |
| 185 | + return community.AgentGrant{}, errors.New("restricted: agent grant is revoked") |
| 186 | + case grant.Paused: |
| 187 | + return community.AgentGrant{}, errors.New("restricted: agent grant is paused") |
| 188 | + case grant.ExpiresAt <= time.Now().Unix(): |
| 189 | + return community.AgentGrant{}, errors.New("restricted: agent grant has expired") |
| 190 | + } |
| 191 | + return grant, nil |
| 192 | +} |
| 193 | + |
| 194 | +// blobUploadTerms is what the blob service asks before it stores an upload: |
| 195 | +// a key whose only standing is an agent grant takes the grant's sites ttl as |
| 196 | +// the expiry of its claims and its encrypted flag as a requirement. Keys |
| 197 | +// with a human role, and agents without a sites tag, upload on no terms. |
| 198 | +func (t *Tenant) blobUploadTerms(ctx context.Context, pubkey string) (blob.UploadTerms, error) { |
| 199 | + role, err := t.community.Role(ctx, pubkey) |
| 200 | + if err != nil { |
| 201 | + return blob.UploadTerms{}, fmt.Errorf("check uploader role: %w", err) |
| 202 | + } |
| 203 | + if role != "agent" { |
| 204 | + return blob.UploadTerms{}, nil |
| 205 | + } |
| 206 | + grant, err := t.agentUploadGrant(ctx, pubkey) |
| 207 | + if err != nil { |
| 208 | + return blob.UploadTerms{}, err |
| 209 | + } |
| 210 | + terms := blob.UploadTerms{Encrypted: grant.UploadsEncrypted()} |
| 211 | + if days := grant.UploadTTLDays(); days > 0 { |
| 212 | + terms.ExpiresAt = time.Now().Unix() + int64(days)*86400 |
| 213 | + } |
| 214 | + return terms, nil |
| 215 | +} |
| 216 | + |
| 217 | +// sweepExpiredBlobs retires uploads whose agent ttl has lapsed. The step is |
| 218 | +// recorded as the blob-expiry storage operation with a bounded outcome and |
| 219 | +// a count, never a hash or a key. |
| 220 | +func (t *Tenant) sweepExpiredBlobs(ctx context.Context, now int64) error { |
| 221 | + if t.blobs == nil { |
| 222 | + return nil |
| 223 | + } |
| 224 | + started := time.Now() |
| 225 | + removed, err := t.blobs.SweepExpired(ctx, now) |
| 226 | + outcome := "ok" |
| 227 | + if err != nil { |
| 228 | + outcome = "error" |
| 229 | + } |
| 230 | + t.app.telemetry.ObserveStorage("blob-expiry", outcome, time.Since(started)) |
| 231 | + if removed > 0 { |
| 232 | + t.app.telemetry.Logger().Info("expired uploads removed", "tenant", t.meta.Name, "blobs", removed) |
| 233 | + } |
| 234 | + return err |
| 235 | +} |
| 236 | + |
| 237 | func (t *Tenant) validateBlobUpload(r *http.Request, hash string) error { |
| 238 | // PATCH carries only one chunk. Blossom scopes the final path hash, while |
| 239 | // NIP-98 binds the bytes in this individual request body. |
| 240 | diff --git a/internal/telemetry/telemetry.go b/internal/telemetry/telemetry.go |
| 241 | index 9f31c1b..c21e2cf 100644 |
| 242 | --- a/internal/telemetry/telemetry.go |
| 243 | +++ b/internal/telemetry/telemetry.go |
| 244 | @@ -236,7 +236,7 @@ func latencyBuckets() []float64 { |
| 245 | return []float64{.0005, .001, .005, .01, .025, .05, .1, .25, .5, 1, 2.5, 5, 10} |
| 246 | } |
| 247 | |
| 248 | -var operationNames = map[string]struct{}{"http": {}, "websocket": {}, "publish": {}, "subscribe": {}, "query": {}, "count": {}, "fanout": {}, "delivery": {}, "commit": {}, "save": {}, "parse": {}, "authorize": {}, "upgrade": {}, "callback": {}, "delivery-discovery": {}, "notification-broadcast": {}, "browserepos": {}, "browserepo": {}, "browsefiles": {}, "browsefile": {}, "browsestatus": {}, "browseissues": {}, "browsepulls": {}, "browseissue": {}, "browsepull": {}, "browse-download": {}, "push": {}, "mcp": {}, "rooms": {}} |
| 249 | +var operationNames = map[string]struct{}{"http": {}, "websocket": {}, "publish": {}, "subscribe": {}, "query": {}, "count": {}, "fanout": {}, "delivery": {}, "commit": {}, "save": {}, "parse": {}, "authorize": {}, "upgrade": {}, "callback": {}, "delivery-discovery": {}, "notification-broadcast": {}, "browserepos": {}, "browserepo": {}, "browsefiles": {}, "browsefile": {}, "browsestatus": {}, "browseissues": {}, "browsepulls": {}, "browseissue": {}, "browsepull": {}, "browse-download": {}, "push": {}, "mcp": {}, "rooms": {}, "blob-expiry": {}} |
| 250 | var outcomeNames = map[string]struct{}{"success": {}, "ok": {}, "invalid": {}, "unauthorized": {}, "timeout": {}, "busy": {}, "closed": {}, "error": {}, "paused": {}} |
| 251 | var queueNames = map[string]struct{}{"inbox": {}, "outbox": {}, "write": {}, "delivery": {}} |
| 252 | var workNames = map[string]struct{}{"delivery": {}, "push": {}, "replication": {}, "webhook": {}, "job": {}, "records-projection": {}, "site-mirror": {}, "git-metadata": {}, "catalog-owner": {}, "callback": {}, "delivery-discovery": {}, "notification-delivery": {}, "notification-broadcast": {}, "notification-push": {}, "notification": {}, "callback-delivery": {}} |