| 1 | package daemon |
| 2 | |
| 3 | import ( |
| 4 | "bytes" |
| 5 | "context" |
| 6 | "crypto/sha256" |
| 7 | "database/sql" |
| 8 | "encoding/hex" |
| 9 | "encoding/json" |
| 10 | "errors" |
| 11 | "fmt" |
| 12 | "io" |
| 13 | "net" |
| 14 | "net/http" |
| 15 | "net/url" |
| 16 | "strings" |
| 17 | "time" |
| 18 | |
| 19 | "github.com/FelineStateMachine/tinyrelay/internal/blob" |
| 20 | "github.com/FelineStateMachine/tinyrelay/internal/community" |
| 21 | "github.com/FelineStateMachine/tinyrelay/internal/configport" |
| 22 | "github.com/FelineStateMachine/tinyrelay/internal/event" |
| 23 | "github.com/FelineStateMachine/tinyrelay/internal/gitstore" |
| 24 | "github.com/FelineStateMachine/tinyrelay/internal/policy" |
| 25 | "github.com/FelineStateMachine/tinyrelay/internal/records" |
| 26 | "github.com/FelineStateMachine/tinyrelay/internal/relay" |
| 27 | "github.com/FelineStateMachine/tinyrelay/internal/replication" |
| 28 | "github.com/FelineStateMachine/tinyrelay/internal/sites" |
| 29 | "github.com/FelineStateMachine/tinyrelay/internal/storage" |
| 30 | "github.com/FelineStateMachine/tinyrelay/internal/views" |
| 31 | "github.com/FelineStateMachine/tinyrelay/internal/work" |
| 32 | "github.com/FelineStateMachine/tinyrelay/protocol/auth" |
| 33 | webui "github.com/FelineStateMachine/tinyrelay/tinyclient" |
| 34 | gitrelay "github.com/FelineStateMachine/tinyrelay/tinygit" |
| 35 | ) |
| 36 | |
| 37 | func gitrelayPolicy(p policy.Policy) gitrelay.Policy { |
| 38 | return gitrelay.Policy{Owner: p.Owner, Reads: p.Reads, PrivatePeers: append([]string(nil), p.PrivatePeers...), Features: gitrelay.PolicyFeatures{Grasp: p.Features.Grasp, Grasp02: p.Features.Grasp02, Grasp03: p.Features.Grasp03, Grasp05: p.Features.Grasp05, Grasp06: p.Features.Grasp06, Grasp08: p.Features.Grasp08}} |
| 39 | } |
| 40 | |
| 41 | // initServices constructs feature services around the tenant store and policy |
| 42 | // reader. Custom-view storage is initialized before Git can promote events; |
| 43 | // the resulting Git service is bound before handlers and workers start. |
| 44 | func (t *Tenant) initServices(ctx context.Context) error { |
| 45 | var err error |
| 46 | if err := t.initSessions(ctx); err != nil { |
| 47 | return err |
| 48 | } |
| 49 | if err := t.initPush(ctx); err != nil { |
| 50 | return err |
| 51 | } |
| 52 | t.callbacks = newCallbackService(callbackServiceConfig{Store: t.store, Community: t.community, Gate: t.gate, Policy: t.Policy, PublicURL: t.publicURL, RelayURL: t.RelayURL(), Tenant: t.meta, Telemetry: t.app.telemetry}) |
| 53 | if err := t.initCallbacks(ctx); err != nil { |
| 54 | return err |
| 55 | } |
| 56 | t.blobs, err = blob.New(ctx, blob.Config{ |
| 57 | Root: t.meta.Paths.Root, PublicURL: t.publicURL, Store: t.store, Authorize: t.authorizeBlob, UploadTerms: t.blobUploadTerms, |
| 58 | Limits: func() blob.Limits { |
| 59 | limits := t.Policy().FileLimits |
| 60 | return blob.Limits{MaxFileBytes: limits.MaxFileBytes, UserStorageBytes: limits.UserStorageBytes} |
| 61 | }, |
| 62 | ValidateUpload: t.validateBlobUpload, |
| 63 | ResolveServers: func(ctx context.Context, pubkey string) ([]string, error) { |
| 64 | 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}) |
| 65 | if err != nil { |
| 66 | return nil, fmt.Errorf("lookup Blossom server list: %w" , err) |
| 67 | } |
| 68 | if len(result.Events) == 0 { |
| 69 | return nil, nil |
| 70 | } |
| 71 | return blob.BlossomServerList(result.Events[0]) |
| 72 | }, |
| 73 | RecordReport: func(ctx context.Context, tx *sql.Tx, reporter, target, targetType, reportType, content string) error { |
| 74 | return t.community.SubmitReportTx(ctx, tx, reporter, target, targetType, reportType, content, t.Policy().ReportThreshold) |
| 75 | }, |
| 76 | CanRead: func(ctx context.Context, hash string, pubkeys []string) bool { |
| 77 | access, accessErr := t.blobs.BlobAccess(ctx, hash) |
| 78 | if accessErr != nil { |
| 79 | return false |
| 80 | } |
| 81 | if access == blob.AccessMembers && len(pubkeys) == 0 { |
| 82 | return false |
| 83 | } |
| 84 | keys := []string{} |
| 85 | for _, pubkey := range pubkeys { |
| 86 | if pubkey == "" { |
| 87 | continue |
| 88 | } |
| 89 | banned, err := t.community.IsBanned(ctx, pubkey) |
| 90 | if err != nil || banned { |
| 91 | return false |
| 92 | } |
| 93 | if access == blob.AccessMembers { |
| 94 | role, roleErr := t.community.Role(ctx, pubkey) |
| 95 | if roleErr != nil || !t.memberBlobReader(ctx, pubkey, role) { |
| 96 | return false |
| 97 | } |
| 98 | } |
| 99 | keys = append(keys, pubkey) |
| 100 | } |
| 101 | if access == blob.AccessMembers && len(keys) == 0 { |
| 102 | return false |
| 103 | } |
| 104 | _, err := t.gate.Read(ctx, []event.Filter{{}}, relay.Session{PubKeys: keys, RelayURL: t.RelayURL()}) |
| 105 | return err == nil |
| 106 | }, IsOwner: func(pubkey string) bool { return pubkey == t.Policy().Owner }, |
| 107 | }) |
| 108 | if err != nil { |
| 109 | return err |
| 110 | } |
| 111 | if err := t.initRoomAttachments(ctx); err != nil { |
| 112 | return err |
| 113 | } |
| 114 | t.community.ConfigurePolicy(t.Policy) |
| 115 | t.sites, err = sites.New(sites.Config{ |
| 116 | Store: t.store, |
| 117 | GetBlob: func(ctx context.Context, hash string) (sites.Blob, error) { |
| 118 | if access, accessErr := t.blobs.BlobAccess(ctx, hash); accessErr != nil { |
| 119 | return sites.Blob{}, accessErr |
| 120 | } else if access == blob.AccessMembers { |
| 121 | return sites.Blob{}, errors.New("restricted: member files cannot be published as sites" ) |
| 122 | } |
| 123 | rooms, err := t.roomAttachmentRooms(ctx, hash) |
| 124 | if err != nil { |
| 125 | return sites.Blob{}, err |
| 126 | } |
| 127 | if len(rooms) > 0 { |
| 128 | return sites.Blob{}, errors.New("restricted: room attachment cannot be served by a site" ) |
| 129 | } |
| 130 | entry, body, err := t.blobs.Get(ctx, hash) |
| 131 | if err != nil { |
| 132 | return sites.Blob{}, err |
| 133 | } |
| 134 | return sites.Blob{Body: body, Type: entry.Type, Size: entry.Size}, nil |
| 135 | }, |
| 136 | PutBlob: func(ctx context.Context, hash, typ string, body io.Reader) error { |
| 137 | _, err := t.blobs.Put(ctx, blob.PutOptions{Reader: body, Type: typ, Uploader: t.Policy().Owner, Hash: hash}) |
| 138 | return err |
| 139 | }, Policy: t.Policy, BaseDomain: t.siteDomain(), RelayURL: func(*http.Request) string { return t.RelayURL() }, ResolveHost: t.siteHostResolver, ReadAccess: func(r *http.Request) bool { return t.siteReadAccess(r) }, |
| 140 | }) |
| 141 | if err != nil { |
| 142 | return err |
| 143 | } |
| 144 | t.records, err = records.New(ctx, records.Config{Store: t.store, Community: t.community, Policy: t.Policy, RelayURL: t.RelayURL(), GroupID: t.meta.Name, OnGenerated: t.generatedRecord, DeliverNotification: t.deliverNotification, PushNotification: t.enqueuePush, SetPolicy: func(next policy.Policy) error { return t.applyPolicy(context.Background(), next) }}) |
| 145 | if err != nil { |
| 146 | return err |
| 147 | } |
| 148 | t.config = configport.New(configport.ConfigStore{Store: t.store, Community: t.community, Policy: t.Policy, ValidatePolicy: t.validatePolicyTransition, OnApplied: func(p policy.Policy) { t.replacePolicy(p) }}) |
| 149 | transport := &replication.NostrTransport{Dialer: replication.WebsocketDialer{AllowPrivate: t.app.cfg.AllowPrivateRelays, MaxMessageBytes: t.app.cfg.MaxMessageBytes}} |
| 150 | t.replication, err = replication.NewService(replication.Config{Store: t.store, Owner: func() string { return t.Policy().Owner }, Policy: replication.Policy{Enabled: t.Policy().Delivery.Enabled, SelfPubKey: t.records.PublicKey()}, CurrentPolicy: func() replication.Policy { |
| 151 | p := t.Policy() |
| 152 | return replication.Policy{Enabled: p.Delivery.Enabled, ReadMembersOnly: p.Reads != "open" , SelfPubKey: t.records.PublicKey(), SelfRelayURL: t.RelayURL()} |
| 153 | }, Directory: localDirectory{store: t.store}, Discovery: t.ReplicationDiscovery(t.app.cfg.DiscoveryRelays), Delivery: transport, Pull: transport, Push: transport, DataDir: t.meta.Paths.Root, Ingest: t.ingest, BackupProvider: t.ReplicationBackupProvider()}) |
| 154 | if err != nil { |
| 155 | return err |
| 156 | } |
| 157 | currentPolicy := t.Policy() |
| 158 | if err := t.reconcileAutomaticInbox(ctx, currentPolicy, currentPolicy); err != nil { |
| 159 | return err |
| 160 | } |
| 161 | t.customViews = newCustomViewService(customViewServiceConfig{Store: t.store, Community: t.community, Records: t.records, Git: t.git, Gate: t.gate, Telemetry: t.app.telemetry, TenantName: t.meta.Name, PublicURL: t.publicURL, Loopback: func() bool { return loopbackPublicURL(t.publicURL) }, ResolveActor: t.resolveUIActor, BrowseRead: t.browseRead}) |
| 162 | if err := t.customViews.initCustomViews(ctx); err != nil { |
| 163 | return err |
| 164 | } |
| 165 | t.followups = &eventFollowups{store: t.store, callbacks: t.callbacks, views: t.customViews, telemetry: t.app.telemetry, push: t.planReplicationPushFollowup} |
| 166 | t.git, err = gitrelay.New(gitrelay.Config{Store: gitstore.Wrap(t.store), Root: t.meta.Paths.Git, Policy: func() gitrelay.Policy { return gitrelayPolicy(t.Policy()) }, PublicURL: t.publicURL, AllowPrivateRelays: t.app.cfg.AllowPrivateRelays, PrivatePeers: t.Policy().PrivatePeers, HTTPAuth: t.privateHTTPAuth, GitSync: t.gitSync, EventSync: t.gitEventSync, AuthorizeHTTP: t.authorizeGit, Maintainers: t, OnPromote: func(ctx context.Context, id string, _ gitrelay.Repository) error { |
| 167 | return t.releaseGit(ctx, id) |
| 168 | }}) |
| 169 | if err != nil { |
| 170 | return err |
| 171 | } |
| 172 | t.customViews.git = t.git |
| 173 | t.ui, err = webui.New(backend{tenant: t}, webui.Options{Actor: t.resolveUIActor, Version: t.app.cfg.Version, Revision: t.app.cfg.Revision}) |
| 174 | if err != nil { |
| 175 | return err |
| 176 | } |
| 177 | if err := t.initMCP(); err != nil { |
| 178 | return err |
| 179 | } |
| 180 | workerRun := t.workRunner() |
| 181 | t.workCtx, t.workCancel = context.WithCancel(context.Background()) |
| 182 | t.workWG.Add(1) |
| 183 | go func() { |
| 184 | defer t.workWG.Done() |
| 185 | if err := workerRun(t.workCtx); err != nil && !errors.Is(err, context.Canceled) { |
| 186 | t.app.telemetry.Logger().Error("tenant worker stopped" , "tenant" , t.meta.Name, "error" , err) |
| 187 | t.workErrMu.Lock() |
| 188 | t.workErr = err |
| 189 | t.workErrMu.Unlock() |
| 190 | } |
| 191 | }() |
| 192 | t.workWG.Add(1) |
| 193 | go func() { defer t.workWG.Done(); t.runScheduler() }() |
| 194 | t.workWG.Add(1) |
| 195 | go func() { defer t.workWG.Done(); t.runGitLive(t.workCtx) }() |
| 196 | t.workWG.Add(1) |
| 197 | go func() { defer t.workWG.Done(); t.runGitHistory(t.workCtx) }() |
| 198 | return nil |
| 199 | } |
| 200 | |
| 201 | func (t *Tenant) memberBlobReader(ctx context.Context, pubkey, role string) bool { |
| 202 | if role == "owner" || role == "moderator" || role == "member" { |
| 203 | return true |
| 204 | } |
| 205 | if role != "agent" { |
| 206 | return false |
| 207 | } |
| 208 | _, err := t.agentUploadGrant(ctx, pubkey) |
| 209 | return err == nil |
| 210 | } |
| 211 | |
| 212 | func (t *Tenant) fileReadAccess(ctx context.Context, actor, hash string) error { |
| 213 | access, err := t.blobs.BlobAccess(ctx, hash) |
| 214 | if err != nil { |
| 215 | return err |
| 216 | } |
| 217 | if access != blob.AccessMembers { |
| 218 | return nil |
| 219 | } |
| 220 | if actor == "" { |
| 221 | return errors.New("auth-required: member file requires sign in" ) |
| 222 | } |
| 223 | banned, err := t.community.IsBanned(ctx, actor) |
| 224 | if err != nil { |
| 225 | return fmt.Errorf("check file reader ban: %w" , err) |
| 226 | } |
| 227 | if banned { |
| 228 | return errors.New("blocked: file reader is banned" ) |
| 229 | } |
| 230 | role, err := t.community.Role(ctx, actor) |
| 231 | if err != nil { |
| 232 | return fmt.Errorf("check file reader role: %w" , err) |
| 233 | } |
| 234 | if !t.memberBlobReader(ctx, actor, role) { |
| 235 | return errors.New("restricted: relay membership required to read this file" ) |
| 236 | } |
| 237 | return nil |
| 238 | } |
| 239 | |
| 240 | // backend adapts the management method signature to webui's form-oriented API. |
| 241 | type backend struct{ tenant *Tenant } |
| 242 | |
| 243 | type localDirectory struct{ store *storage.Store } |
| 244 | |
| 245 | func (d localDirectory) WriteRelays(pubkey string) []string { |
| 246 | return d.relaysForDirection(pubkey, true) |
| 247 | } |
| 248 | func (d localDirectory) ReadRelays(pubkey string) []string { |
| 249 | return d.relaysForDirection(pubkey, false) |
| 250 | } |
| 251 | func (d localDirectory) RelayList(pubkey string) []string { return d.relays(pubkey, false, true) } |
| 252 | func (d localDirectory) relaysForDirection(pubkey string, write bool) []string { |
| 253 | return d.relays(pubkey, write, false) |
| 254 | } |
| 255 | func (d localDirectory) relays(pubkey string, write, all bool) []string { |
| 256 | rows, err := d.store.Query(context.Background(), event.Filter{Authors: []string{pubkey}, Kinds: []int{10002}, Tags: map[string][]string{}}, storage.QueryOptions{Now: time.Now().Unix(), Access: storage.Access{All: true}, Limit: 1}) |
| 257 | if err != nil || len(rows.Events) == 0 { |
| 258 | return nil |
| 259 | } |
| 260 | result := make([]string, 0) |
| 261 | for _, tag := range rows.Events[0].Tags { |
| 262 | if len(tag) < 2 || tag[0] != "r" || (!all && !write && len(tag) > 2 && tag[2] != "read" ) || (!all && write && len(tag) > 2 && tag[2] == "read" ) { |
| 263 | continue |
| 264 | } |
| 265 | result = append(result, tag[1]) |
| 266 | } |
| 267 | return result |
| 268 | } |
| 269 | |
| 270 | func (b backend) Query(ctx context.Context, method string, params []json.RawMessage, actor string) (any, error) { |
| 271 | return b.tenant.Execute(ctx, actor, method, params) |
| 272 | } |
| 273 | func (b backend) Policy() policy.Policy { return b.tenant.Policy() } |
| 274 | func (b backend) URL() string { return b.tenant.publicURL } |
| 275 | func (b backend) Slug() string { return b.tenant.meta.Name } |
| 276 | func (b backend) Identity() string { return b.tenant.records.PublicKey() } |
| 277 | |
| 278 | // CustomViews gives the page renderers the enabled custom views' names and |
| 279 | // languages, never a transform or a secret. |
| 280 | func (b backend) CustomViews() []views.View { return b.tenant.customViews.CustomViews() } |
| 281 | |
| 282 | // ReadAllowed is consumed by the web UI private boundary. It deliberately |
| 283 | // delegates to the same uncached membership check used by private Git and |
| 284 | // NIP-42, so a revoked browser session cannot retain page access. |
| 285 | func (b backend) ReadAllowed(ctx context.Context, actor string) error { |
| 286 | if !b.tenant.PrivateServiceEnabled() { |
| 287 | return nil |
| 288 | } |
| 289 | return b.tenant.requirePrivateAccess(ctx, actor) |
| 290 | } |
| 291 | |
| 292 | func (b backend) SharePresence(ctx context.Context, actor string) bool { |
| 293 | return b.tenant.chatSharePresence(ctx, actor) |
| 294 | } |
| 295 | |
| 296 | // closeServices cancels and joins the tenant's worker, scheduler and Git |
| 297 | // loops, then returns the worker's terminal queue error, if any. Store and |
| 298 | // relay connection cleanup belong to Tenant.Close. |
| 299 | func (t *Tenant) closeServices(ctx context.Context) error { |
| 300 | if t.workCancel != nil { |
| 301 | t.workCancel() |
| 302 | } |
| 303 | t.workWG.Wait() |
| 304 | t.workErrMu.Lock() |
| 305 | defer t.workErrMu.Unlock() |
| 306 | return t.workErr |
| 307 | } |
| 308 | |
| 309 | func hostName(raw string) string { |
| 310 | u, err := url.Parse(raw) |
| 311 | if err != nil { |
| 312 | return "localhost" |
| 313 | } |
| 314 | if u.Hostname() == "" { |
| 315 | return "localhost" |
| 316 | } |
| 317 | return u.Hostname() |
| 318 | } |
| 319 | func (t *Tenant) siteReadAccess(r *http.Request) bool { |
| 320 | actor, err := t.resolveUIActor(r) |
| 321 | if err != nil { |
| 322 | return false |
| 323 | } |
| 324 | keys := []string{} |
| 325 | if actor != "" { |
| 326 | keys = append(keys, actor) |
| 327 | } |
| 328 | _, err = t.gate.Read(r.Context(), []event.Filter{{}}, relay.Session{PubKeys: keys, RelayURL: t.RelayURL()}) |
| 329 | return err == nil |
| 330 | } |
| 331 | func (t *Tenant) authorizeBlob(r *http.Request, action blob.Action) (string, error) { |
| 332 | hash := "" |
| 333 | if action == blob.ActionGet || action == blob.ActionDelete || (action == blob.ActionUpload && (r.Method == http.MethodPut || r.Method == http.MethodPatch)) { |
| 334 | path := strings.TrimPrefix(r.URL.Path, "/" ) |
| 335 | path = strings.TrimPrefix(path, "nip96/" ) |
| 336 | path = strings.TrimPrefix(path, "media/" ) |
| 337 | hash = strings.Split(path, "." )[0] |
| 338 | if len(hash) != 64 || strings.Trim(hash, "0123456789abcdef" ) != "" { |
| 339 | hash = "" |
| 340 | } |
| 341 | } |
| 342 | body, bodyErr := blobAuthBody(r, action) |
| 343 | if bodyErr != nil { |
| 344 | return "" , bodyErr |
| 345 | } |
| 346 | pubkey := "" |
| 347 | var err error |
| 348 | if action == blob.ActionUpload && r.Method == http.MethodPost && r.URL.Path == "/nip96" { |
| 349 | var token event.Event |
| 350 | token, err = t.auth.VerifyNIP98(r.Header.Get("Authorization" ), t.requestURL(r), r.Method, body) |
| 351 | pubkey = token.PubKey |
| 352 | } else { |
| 353 | if auth.IsBlossomAuthorization(r.Header.Get("Authorization" )) { |
| 354 | blossomAction := string(action) |
| 355 | if action == blob.ActionMirror { |
| 356 | // BUD-04 mirror uses the upload authorization verb; the fetched |
| 357 | // bytes are the resulting blob and are scoped by x below. |
| 358 | blossomAction = string(blob.ActionUpload) |
| 359 | } |
| 360 | blobHash := hash |
| 361 | if blobHash == "" && action == blob.ActionUpload { |
| 362 | blobHash = strings.ToLower(strings.TrimSpace(r.Header.Get("x-sha-256" ))) |
| 363 | } |
| 364 | var token event.Event |
| 365 | token, err = t.auth.VerifyBlossomRequest(r.Header.Get("Authorization" ), blossomAction, t.siteDomain(), blobHash) |
| 366 | pubkey = token.PubKey |
| 367 | } else { |
| 368 | if isBUD13RemoteUpload(r) { |
| 369 | var token event.Event |
| 370 | token, err = t.auth.VerifyNIP98(r.Header.Get("Authorization" ), t.requestURL(r), r.Method, "" ) |
| 371 | if err == nil && event.Tag(token, "payload" ) != "" && event.Tag(token, "payload" ) != emptyPayloadHash() { |
| 372 | err = errors.New("auth-required: remote upload token payload must hash an empty body" ) |
| 373 | } |
| 374 | pubkey = token.PubKey |
| 375 | } else { |
| 376 | var keys []string |
| 377 | keys, err = t.auth.WhoAsks(r.Header.Get("Authorization" ), t.requestURL(r), r.Method, body, string(action), hash) |
| 378 | pubkey = first(keys) |
| 379 | } |
| 380 | } |
| 381 | } |
| 382 | if err != nil { |
| 383 | return "" , err |
| 384 | } |
| 385 | if action == blob.ActionGet { |
| 386 | if err := t.roomAttachmentAccess(r.Context(), pubkey, hash); err != nil { |
| 387 | return "" , err |
| 388 | } |
| 389 | } |
| 390 | if pubkey == "" { |
| 391 | if action == blob.ActionGet { |
| 392 | return "" , nil |
| 393 | } |
| 394 | return "" , errors.New("auth-required: blob authorization required" ) |
| 395 | } |
| 396 | if banned, banErr := t.community.IsBanned(r.Context(), pubkey); banErr != nil { |
| 397 | return "" , fmt.Errorf("check blob uploader ban: %w" , banErr) |
| 398 | } else if banned { |
| 399 | return "" , errors.New("blocked: this pubkey is banned from this relay" ) |
| 400 | } |
| 401 | if action == blob.ActionUpload || action == blob.ActionMirror { |
| 402 | if !t.Policy().Features.Files { |
| 403 | return "" , errors.New("restricted: file uploads are disabled" ) |
| 404 | } |
| 405 | role, roleErr := t.community.Role(r.Context(), pubkey) |
| 406 | if roleErr != nil { |
| 407 | return "" , fmt.Errorf("check blob uploader role: %w" , roleErr) |
| 408 | } |
| 409 | allowed := t.Policy().Writes == "open" || (t.Policy().Writes == "owner" && role == "owner" ) || (t.Policy().Writes != "owner" && role != "" ) |
| 410 | if !allowed { |
| 411 | return "" , errors.New("restricted: file upload is not allowed" ) |
| 412 | } |
| 413 | // An agent's standing is its grant: an active one makes it a member |
| 414 | // for uploads, a paused, revoked or expired one stops them. |
| 415 | if role == "agent" { |
| 416 | if _, err := t.agentUploadGrant(r.Context(), pubkey); err != nil { |
| 417 | return "" , err |
| 418 | } |
| 419 | } |
| 420 | } |
| 421 | return pubkey, nil |
| 422 | } |
| 423 | |
| 424 | // agentUploadGrant returns the grant behind an agent key's uploads when it |
| 425 | // currently allows publishing. |
| 426 | func (t *Tenant) agentUploadGrant(ctx context.Context, pubkey string) (community.AgentGrant, error) { |
| 427 | grant, ok, err := t.community.AgentGrant(ctx, pubkey) |
| 428 | if err != nil { |
| 429 | return community.AgentGrant{}, fmt.Errorf("check agent upload grant: %w" , err) |
| 430 | } |
| 431 | if !ok { |
| 432 | return community.AgentGrant{}, errors.New("restricted: agent grant is missing" ) |
| 433 | } |
| 434 | switch { |
| 435 | case grant.RevokedAt > 0: |
| 436 | return community.AgentGrant{}, errors.New("restricted: agent grant is revoked" ) |
| 437 | case grant.Paused: |
| 438 | return community.AgentGrant{}, errors.New("restricted: agent grant is paused" ) |
| 439 | case grant.ExpiresAt <= time.Now().Unix(): |
| 440 | return community.AgentGrant{}, errors.New("restricted: agent grant has expired" ) |
| 441 | } |
| 442 | return grant, nil |
| 443 | } |
| 444 | |
| 445 | // blobUploadTerms is what the blob service asks before it stores an upload: |
| 446 | // a key whose only standing is an agent grant takes the grant's sites ttl as |
| 447 | // the expiry of its claims and its encrypted flag as a requirement. Keys |
| 448 | // with a human role, and agents without a sites tag, upload on no terms. |
| 449 | func (t *Tenant) blobUploadTerms(ctx context.Context, pubkey string) (blob.UploadTerms, error) { |
| 450 | role, err := t.community.Role(ctx, pubkey) |
| 451 | if err != nil { |
| 452 | return blob.UploadTerms{}, fmt.Errorf("check uploader role: %w" , err) |
| 453 | } |
| 454 | if role != "agent" { |
| 455 | return blob.UploadTerms{}, nil |
| 456 | } |
| 457 | grant, err := t.agentUploadGrant(ctx, pubkey) |
| 458 | if err != nil { |
| 459 | return blob.UploadTerms{}, err |
| 460 | } |
| 461 | terms := blob.UploadTerms{Encrypted: grant.UploadsEncrypted()} |
| 462 | if days := grant.UploadTTLDays(); days > 0 { |
| 463 | terms.ExpiresAt = time.Now().Unix() + int64(days)*86400 |
| 464 | } |
| 465 | return terms, nil |
| 466 | } |
| 467 | |
| 468 | // sweepExpiredBlobs retires uploads whose agent ttl has lapsed. The step is |
| 469 | // recorded as the blob-expiry storage operation with a bounded outcome and |
| 470 | // a count, never a hash or a key. |
| 471 | func (t *Tenant) sweepExpiredBlobs(ctx context.Context, now int64) error { |
| 472 | if t.blobs == nil { |
| 473 | return nil |
| 474 | } |
| 475 | started := time.Now() |
| 476 | removed, err := t.blobs.SweepExpired(ctx, now) |
| 477 | outcome := "ok" |
| 478 | if err != nil { |
| 479 | outcome = "error" |
| 480 | } |
| 481 | t.app.telemetry.ObserveStorage("blob-expiry" , outcome, time.Since(started)) |
| 482 | if removed > 0 { |
| 483 | t.app.telemetry.Logger().Info("expired uploads removed" , "tenant" , t.meta.Name, "blobs" , removed) |
| 484 | } |
| 485 | return err |
| 486 | } |
| 487 | |
| 488 | func (t *Tenant) validateBlobUpload(r *http.Request, hash string) error { |
| 489 | // PATCH carries only one chunk. Blossom scopes the final path hash, while |
| 490 | // NIP-98 binds the bytes in this individual request body. |
| 491 | if r.Method == http.MethodPatch && auth.IsBlossomAuthorization(r.Header.Get("Authorization" )) { |
| 492 | finalHash := strings.Split(strings.TrimPrefix(r.URL.Path, "/" ), "." )[0] |
| 493 | return t.auth.ValidateBlossomPayload(r.Header.Get("Authorization" ), string(blob.ActionUpload), finalHash) |
| 494 | } |
| 495 | // Mirror proofs bind the HTTP request body before fetching. Blossom |
| 496 | // authorization additionally names the fetched content through its x tag. |
| 497 | if r.URL.Path == "/mirror" || isBUD13RemoteUpload(r) { |
| 498 | if auth.IsBlossomAuthorization(r.Header.Get("Authorization" )) { |
| 499 | return t.auth.ValidateBlossomPayload(r.Header.Get("Authorization" ), string(blob.ActionUpload), hash) |
| 500 | } |
| 501 | return nil |
| 502 | } |
| 503 | return t.auth.ValidateBlobPayload(r.Header.Get("Authorization" ), hash) |
| 504 | } |
| 505 | |
| 506 | // NIP-98 binds its payload to the complete HTTP body. The streaming Blossom |
| 507 | // upload remains unbuffered and is checked against its resulting SHA-256 by |
| 508 | // ValidateUpload; the JSON and multipart doors are buffered only so their |
| 509 | // signed request can be verified before parsing or fetching any content. |
| 510 | func blobAuthBody(r *http.Request, action blob.Action) (string, error) { |
| 511 | buffer := action == blob.ActionMirror || action == blob.ActionReport || |
| 512 | (action == blob.ActionUpload && r.Method == http.MethodPost && (r.URL.Path == "/nip96" || r.URL.Path == "/files/metadata" )) |
| 513 | if !buffer { |
| 514 | return "" , nil |
| 515 | } |
| 516 | var source io.Reader = r.Body |
| 517 | if r.URL.Path == "/files/metadata" { |
| 518 | source = io.LimitReader(r.Body, 16*1024+1) |
| 519 | } |
| 520 | body, err := io.ReadAll(source) |
| 521 | if err != nil { |
| 522 | return "" , fmt.Errorf("read blob authorization body: %w" , err) |
| 523 | } |
| 524 | if r.URL.Path == "/files/metadata" && len(body) > 16*1024 { |
| 525 | return "" , errors.New("invalid: file metadata exceeds 16 KiB" ) |
| 526 | } |
| 527 | r.Body = io.NopCloser(bytes.NewReader(body)) |
| 528 | return string(body), nil |
| 529 | } |
| 530 | |
| 531 | func (t *Tenant) siteDomain() string { |
| 532 | if t.meta.Name != t.app.cfg.DefaultTenant { |
| 533 | return t.app.siteBase(t.meta) |
| 534 | } |
| 535 | host := hostName(t.publicURL) |
| 536 | if ip := net.ParseIP(host); ip != nil && ip.IsLoopback() { |
| 537 | return "localhost" |
| 538 | } |
| 539 | return host |
| 540 | } |
| 541 | func (t *Tenant) authorizeGit(ctx context.Context, r *http.Request, repo gitrelay.Repository) error { |
| 542 | if !t.Policy().Features.Grasp { |
| 543 | return errors.New("restricted: Git hosting is disabled" ) |
| 544 | } |
| 545 | // Authenticate before reading a potentially large pack. The body binding |
| 546 | // is checked below before any bytes reach git-receive-pack. |
| 547 | e, ok := privateGitProof(r) |
| 548 | var err error |
| 549 | if !ok { |
| 550 | e, err = t.auth.VerifyNIP98(r.Header.Get("Authorization" ), t.requestURL(r), r.Method, "" ) |
| 551 | if err != nil { |
| 552 | return err |
| 553 | } |
| 554 | } |
| 555 | if banned, banErr := t.community.IsBanned(ctx, e.PubKey); banErr != nil { |
| 556 | return fmt.Errorf("check Git author ban: %w" , banErr) |
| 557 | } else if banned { |
| 558 | return errors.New("blocked: this pubkey is banned from this relay" ) |
| 559 | } |
| 560 | if privatePolicy(t.Policy()) { |
| 561 | // Tenant membership is the private-service boundary. A repository |
| 562 | // maintainer or owner who is not a member must not learn whether a |
| 563 | // private repository exists through Smart HTTP. |
| 564 | if err := t.requirePrivateAccess(ctx, e.PubKey); err != nil { |
| 565 | return err |
| 566 | } |
| 567 | } else if !t.IsMaintainer(ctx, repo, e.PubKey) { |
| 568 | role, roleErr := t.community.Role(ctx, e.PubKey) |
| 569 | if roleErr != nil || (role != "owner" && role != "moderator" && role != "member" ) { |
| 570 | return errors.New("restricted: repository authorization required" ) |
| 571 | } |
| 572 | } |
| 573 | if _, checked := r.Context().Value(privateGitPayloadCheckedKey{}).(struct{}); checked { |
| 574 | return nil |
| 575 | } |
| 576 | return spoolGitPayload(r, e) |
| 577 | } |
| 578 | func first(v []string) string { |
| 579 | if len(v) == 0 { |
| 580 | return "" |
| 581 | } |
| 582 | return v[0] |
| 583 | } |
| 584 | |
| 585 | func isBUD13RemoteUpload(r *http.Request) bool { |
| 586 | if r == nil || r.Method != http.MethodPut || r.URL.Query().Get("url" ) == "" { |
| 587 | return false |
| 588 | } |
| 589 | path := strings.TrimPrefix(r.URL.Path, "/" ) |
| 590 | hash := strings.Split(path, "." )[0] |
| 591 | return len(hash) == 64 && strings.Trim(hash, "0123456789abcdefABCDEF" ) == "" |
| 592 | } |
| 593 | |
| 594 | func emptyPayloadHash() string { |
| 595 | sum := sha256.Sum256(nil) |
| 596 | return hex.EncodeToString(sum[:]) |
| 597 | } |
| 598 | |
| 599 | func (t *Tenant) resolveUIActor(r *http.Request) (string, error) { |
| 600 | if r.Header.Get("Authorization" ) == "" { |
| 601 | if r.Method != http.MethodGet && r.Method != http.MethodHead { |
| 602 | return "" , errors.New("auth-required: sign this request" ) |
| 603 | } |
| 604 | return t.cookieActor(r) |
| 605 | } |
| 606 | if r.Body == nil { |
| 607 | r.Body = http.NoBody |
| 608 | } |
| 609 | body, err := io.ReadAll(r.Body) |
| 610 | if err != nil { |
| 611 | return "" , err |
| 612 | } |
| 613 | r.Body = io.NopCloser(bytes.NewReader(body)) |
| 614 | e, err := t.auth.VerifyNIP98(r.Header.Get("Authorization" ), t.requestURL(r), r.Method, string(body)) |
| 615 | if err != nil { |
| 616 | return "" , err |
| 617 | } |
| 618 | return e.PubKey, nil |
| 619 | } |
| 620 | |
| 621 | func (t *Tenant) workHandlers() map[string]work.Handler { |
| 622 | handlers := map[string]work.Handler{ |
| 623 | "view-publish" : func(ctx context.Context, intent work.Intent) error { |
| 624 | if err := t.records.MarkView(ctx, intent.Target, time.Now().Unix()); err != nil { |
| 625 | return err |
| 626 | } |
| 627 | t.scheduleViews() |
| 628 | return nil |
| 629 | }, |
| 630 | "git-metadata" : func(ctx context.Context, intent work.Intent) error { |
| 631 | var raw string |
| 632 | err := t.store.DB().QueryRowContext(ctx, "SELECT raw FROM events WHERE id=?" , intent.EventID).Scan(&raw) |
| 633 | if errors.Is(err, sql.ErrNoRows) { |
| 634 | return nil |
| 635 | } |
| 636 | if err != nil { |
| 637 | return err |
| 638 | } |
| 639 | e, err := event.Parse([]byte(raw)) |
| 640 | if err != nil { |
| 641 | return err |
| 642 | } |
| 643 | var repo gitrelay.Repository |
| 644 | if err := json.Unmarshal([]byte(intent.Payload), &repo); err != nil { |
| 645 | return err |
| 646 | } |
| 647 | if err := t.git.CommitAfterStore(ctx, e, repo); err != nil { |
| 648 | return err |
| 649 | } |
| 650 | if !t.git.IsPending(e.ID) { |
| 651 | return t.releaseGit(ctx, e.ID) |
| 652 | } |
| 653 | return nil |
| 654 | }, |
| 655 | "catalog-owner" : func(ctx context.Context, intent work.Intent) error { |
| 656 | return t.app.catalog.UpdateOwner(ctx, t.meta.ID, t.Policy().Owner) |
| 657 | }, |
| 658 | "records-projection" : func(ctx context.Context, intent work.Intent) error { |
| 659 | var payload struct { |
| 660 | Action string `json:"action"` |
| 661 | Room string `json:"room"` |
| 662 | PubKey string `json:"pubkey"` |
| 663 | Role string `json:"role"` |
| 664 | Removed int `json:"removed"` |
| 665 | Deleted int `json:"deleted"` |
| 666 | } |
| 667 | if err := json.Unmarshal([]byte(intent.Payload), &payload); err != nil { |
| 668 | return err |
| 669 | } |
| 670 | if payload.Action == "room" { |
| 671 | return t.projectRoom(ctx, payload.Room, payload.PubKey, payload.Removed != 0, payload.Deleted != 0) |
| 672 | } |
| 673 | var changes []records.MembershipChange |
| 674 | if payload.PubKey != "" && (payload.Role != "" || payload.Removed != 0) { |
| 675 | added := payload.Removed == 0 |
| 676 | changes = []records.MembershipChange{{PubKey: payload.PubKey, Added: &added, Roles: []string{payload.Role}}} |
| 677 | } |
| 678 | _, err := t.records.PublishMembershipChanges(ctx, changes, time.Now().Unix()) |
| 679 | return err |
| 680 | }, |
| 681 | "site-mirror" : func(ctx context.Context, intent work.Intent) error { |
| 682 | return t.sites.RunMirrorIntent(ctx, storage.Intent{Kind: intent.Kind, EventID: intent.EventID, Target: intent.Target, Payload: intent.Payload}) |
| 683 | }, |
| 684 | "callback" : t.ReplicationCallbackHandler(), |
| 685 | } |
| 686 | for kind, handler := range t.notificationHandlers() { |
| 687 | handlers[kind] = handler |
| 688 | } |
| 689 | handlers[notificationPush] = t.handleNotificationPush |
| 690 | handlers[callbackDelivery] = t.handleCallbackDelivery |
| 691 | handlers[viewTransform] = t.customViews.handleViewTransform |
| 692 | handlers[eventFollowupKind] = t.followups.handle |
| 693 | return handlers |
| 694 | } |
| 695 | |
| 696 | // releaseGit publishes a repository state after Git verifies its objects. It |
| 697 | // removes the pending marker and persists optional planning in one transaction |
| 698 | // under the router's publication lock. Push planning retains the registrations |
| 699 | // captured at client acceptance; webhook and view matching occurs at release. |
| 700 | func (t *Tenant) releaseGit(ctx context.Context, id string) error { |
| 701 | var followup *storage.Intent |
| 702 | if t.followups != nil { |
| 703 | var raw []byte |
| 704 | err := t.store.DB().QueryRowContext(ctx, "SELECT raw FROM events WHERE id=?" , id).Scan(&raw) |
| 705 | if err != nil && !errors.Is(err, sql.ErrNoRows) { |
| 706 | return err |
| 707 | } |
| 708 | if err == nil { |
| 709 | e, err := event.Parse(raw) |
| 710 | if err != nil { |
| 711 | return err |
| 712 | } |
| 713 | followup = t.followups.prepare(ctx, e, time.Now().Unix(), true) |
| 714 | } |
| 715 | } |
| 716 | released, err := t.router.CommitGenerated(ctx, func(ctx context.Context) (event.Event, error) { |
| 717 | var promoted event.Event |
| 718 | err := t.store.WithTx(ctx, func(tx *sql.Tx) error { |
| 719 | result, err := tx.ExecContext(ctx, "DELETE FROM pending_events WHERE id=?" , id) |
| 720 | if err != nil { |
| 721 | return err |
| 722 | } |
| 723 | n, err := result.RowsAffected() |
| 724 | if err != nil || n == 0 { |
| 725 | return err |
| 726 | } |
| 727 | if _, err := tx.ExecContext(ctx, "DELETE FROM replication_pending_until WHERE id=?" , id); err != nil && !isMissingTable(err) { |
| 728 | return err |
| 729 | } |
| 730 | var raw []byte |
| 731 | err = tx.QueryRowContext(ctx, `SELECT raw FROM events WHERE id=? AND (expires=0 OR expires>?) AND NOT EXISTS(SELECT 1 FROM hidden_events WHERE hidden_events.id=events.id)` , id, time.Now().Unix()).Scan(&raw) |
| 732 | if errors.Is(err, sql.ErrNoRows) { |
| 733 | return nil |
| 734 | } |
| 735 | if err != nil { |
| 736 | return err |
| 737 | } |
| 738 | promoted, err = event.Parse(raw) |
| 739 | if err != nil || followup == nil { |
| 740 | return err |
| 741 | } |
| 742 | return storage.AddIntents(ctx, tx, []storage.Intent{*followup}, time.Now().Unix()) |
| 743 | }) |
| 744 | return promoted, err |
| 745 | }) |
| 746 | if err == nil && released.ID != "" { |
| 747 | // A repository state is hidden until its objects arrive, so agents |
| 748 | // that wait for pushes are woken here rather than at publish time. |
| 749 | if t.customViews != nil && t.customViews.git != nil { |
| 750 | t.followups.try(ctx, followup) |
| 751 | } |
| 752 | } |
| 753 | return err |
| 754 | } |
| 755 | |
| 756 | func (t *Tenant) sweep(ctx context.Context, now int64) error { |
| 757 | if _, err := t.store.SweepExpired(ctx, now); err != nil { |
| 758 | return err |
| 759 | } |
| 760 | opts := storage.RetentionOptions{Now: now, Identity: t.records.PublicKey()} |
| 761 | rules, err := t.community.RetentionRules(ctx) |
| 762 | if err != nil { |
| 763 | return err |
| 764 | } |
| 765 | for _, rule := range rules { |
| 766 | kind := -1 |
| 767 | if rule.Kind != nil { |
| 768 | kind = *rule.Kind |
| 769 | } |
| 770 | opts.Rules = append(opts.Rules, storage.RetentionRule{Kind: kind, Days: rule.Days}) |
| 771 | } |
| 772 | members, err := t.community.MembersForRetention(ctx) |
| 773 | if err != nil { |
| 774 | return err |
| 775 | } |
| 776 | for _, member := range members { |
| 777 | opts.Members = append(opts.Members, storage.MemberRetention{PubKey: member.PubKey, Days: member.Days}) |
| 778 | } |
| 779 | if _, err := t.store.SweepRetention(ctx, opts); err != nil { |
| 780 | return err |
| 781 | } |
| 782 | return t.customViews.pruneAllViewArtifacts(ctx) |
| 783 | } |
| 784 | func (t *Tenant) replacePolicy(p policy.Policy) { |
| 785 | t.mu.Lock() |
| 786 | previous := t.policy |
| 787 | t.policy = p |
| 788 | t.mu.Unlock() |
| 789 | if community.RoomAccessFor(previous.Reads) != community.RoomAccessFor(p.Reads) { |
| 790 | if err := t.community.EnsureSlugRoom(context.Background(), t.meta.Name, community.RoomAccessFor(p.Reads)); err != nil { |
| 791 | t.app.telemetry.Logger().Error("refresh main room" , "tenant" , t.meta.Name, "error" , err) |
| 792 | } |
| 793 | } |
| 794 | if err := t.reconcileAutomaticInbox(context.Background(), previous, p); err != nil { |
| 795 | t.app.telemetry.Logger().Error("reconcile automatic inbox job" , "tenant" , t.meta.Name, "error" , err) |
| 796 | } |
| 797 | t.router.CloseSubscriptions("blocked: relay policy changed; subscribe again" ) |
| 798 | t.scheduleViews() |
| 799 | } |
| 800 | func (t *Tenant) generatedRecord(ctx context.Context, e event.Event) error { |
| 801 | ctx, done, admissionErr := t.beginOperation(ctx) |
| 802 | if admissionErr != nil { |
| 803 | return admissionErr |
| 804 | } |
| 805 | defer done() |
| 806 | _, err := t.router.CommitGenerated(ctx, func(ctx context.Context) (event.Event, error) { |
| 807 | _, err := t.store.Save(ctx, e, storage.SaveOptions{Now: e.CreatedAt, SearchMode: t.Policy().Features.Search}) |
| 808 | return e, err |
| 809 | }) |
| 810 | return err |
| 811 | } |
| 812 | func (t *Tenant) ingest(ctx context.Context, e event.Event, _ replication.Origin) error { |
| 813 | ctx, done, admissionErr := t.beginOperation(ctx) |
| 814 | if admissionErr != nil { |
| 815 | return admissionErr |
| 816 | } |
| 817 | defer done() |
| 818 | _, err := t.router.CommitGenerated(ctx, func(ctx context.Context) (event.Event, error) { |
| 819 | return e, t.commitImported(ctx, e, replication.OriginImport) |
| 820 | }) |
| 821 | if errors.Is(err, storage.ErrDuplicate) || errors.Is(err, storage.ErrReplaced) { |
| 822 | return nil |
| 823 | } |
| 824 | return err |
| 825 | } |
| 826 | |
| 827 | // commitImported applies import admission and commits the shared projections |
| 828 | // and optional planning for an event. It returns persistence errors to the |
| 829 | // caller, which decides how to handle duplicate and superseded imports. |
| 830 | // origin supplies the replication source for planning. |
| 831 | func (t *Tenant) commitImported(ctx context.Context, e event.Event, origin replication.Origin) error { |
| 832 | if err := t.gate.Import(ctx, e, time.Now().Unix()); err != nil { |
| 833 | return err |
| 834 | } |
| 835 | if err := t.validateStoredEvent(ctx, e); err != nil { |
| 836 | return err |
| 837 | } |
| 838 | plan, err := t.prepareStoredEvent(ctx, e, origin, time.Now().Unix()) |
| 839 | if err != nil { |
| 840 | return err |
| 841 | } |
| 842 | _, err = t.store.Save(ctx, e, plan.options) |
| 843 | if err == nil { |
| 844 | t.afterStoredEvent(ctx, e, plan.followups) |
| 845 | } |
| 846 | return err |
| 847 | } |