1package daemon23import (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"1819 "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)3637func 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}4041// initServices constructs feature services around the tenant store and policy42// reader. Custom-view storage is initialized before Git can promote events;43// the resulting Git service is bound before handlers and workers start.44func (t *Tenant) initServices(ctx context.Context) error {45 var err error46 if err := t.initSessions(ctx); err != nil {47 return err48 }49 if err := t.initPush(ctx); err != nil {50 return err51 }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 err55 }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().FileLimits60 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, nil70 }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 false80 }81 if access == blob.AccessMembers && len(pubkeys) == 0 {82 return false83 }84 keys := []string{}85 for _, pubkey := range pubkeys {86 if pubkey == "" {87 continue88 }89 banned, err := t.community.IsBanned(ctx, pubkey)90 if err != nil || banned {91 return false92 }93 if access == blob.AccessMembers {94 role, roleErr := t.community.Role(ctx, pubkey)95 if roleErr != nil || !t.memberBlobReader(ctx, pubkey, role) {96 return false97 }98 }99 keys = append(keys, pubkey)100 }101 if access == blob.AccessMembers && len(keys) == 0 {102 return false103 }104 _, err := t.gate.Read(ctx, []event.Filter{{}}, relay.Session{PubKeys: keys, RelayURL: t.RelayURL()})105 return err == nil106 }, IsOwner: func(pubkey string) bool { return pubkey == t.Policy().Owner },107 })108 if err != nil {109 return err110 }111 if err := t.initRoomAttachments(ctx); err != nil {112 return err113 }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{}, accessErr120 } 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{}, err126 }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{}, err133 }134 return sites.Blob{Body: body, Type: entry.Type, Size: entry.Size}, nil135 },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 err139 }, 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 err143 }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, EventVisible: func(ctx context.Context, e event.Event, a policy.Access) bool {145 return t.gate.CanSee(ctx, e, relay.Session{PubKeys: a.PubKeys, RelayURL: t.RelayURL()}, nil)146 }, SetPolicy: func(next policy.Policy) error { return t.applyPolicy(context.Background(), next) }})147 if err != nil {148 return err149 }150 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) }})151 transport := &replication.NostrTransport{Dialer: replication.WebsocketDialer{AllowPrivate: t.app.cfg.AllowPrivateRelays, MaxMessageBytes: t.app.cfg.MaxMessageBytes}}152 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 {153 p := t.Policy()154 return replication.Policy{Enabled: p.Delivery.Enabled, ReadMembersOnly: p.Reads != "open", SelfPubKey: t.records.PublicKey(), SelfRelayURL: t.RelayURL()}155 }, 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()})156 if err != nil {157 return err158 }159 currentPolicy := t.Policy()160 if err := t.reconcileAutomaticInbox(ctx, currentPolicy, currentPolicy); err != nil {161 return err162 }163 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})164 if err := t.customViews.initCustomViews(ctx); err != nil {165 return err166 }167 t.followups = &eventFollowups{store: t.store, callbacks: t.callbacks, views: t.customViews, telemetry: t.app.telemetry, push: t.planReplicationPushFollowup}168 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 {169 return t.releaseGit(ctx, id)170 }})171 if err != nil {172 return err173 }174 t.customViews.git = t.git175 t.ui, err = webui.New(backend{tenant: t}, webui.Options{Actor: t.resolveUIActor, Version: t.app.cfg.Version, Revision: t.app.cfg.Revision})176 if err != nil {177 return err178 }179 if err := t.initMCP(); err != nil {180 return err181 }182 workerRun := t.workRunner()183 t.workCtx, t.workCancel = context.WithCancel(context.Background())184 t.workWG.Add(1)185 go func() {186 defer t.workWG.Done()187 if err := workerRun(t.workCtx); err != nil && !errors.Is(err, context.Canceled) {188 t.app.telemetry.Logger().Error("tenant worker stopped", "tenant", t.meta.Name, "error", err)189 t.workErrMu.Lock()190 t.workErr = err191 t.workErrMu.Unlock()192 }193 }()194 t.workWG.Add(1)195 go func() { defer t.workWG.Done(); t.runScheduler() }()196 t.workWG.Add(1)197 go func() { defer t.workWG.Done(); t.runGitLive(t.workCtx) }()198 t.workWG.Add(1)199 go func() { defer t.workWG.Done(); t.runGitHistory(t.workCtx) }()200 return nil201}202203func (t *Tenant) memberBlobReader(ctx context.Context, pubkey, role string) bool {204 if role == "owner" || role == "moderator" || role == "member" {205 return true206 }207 if role != "agent" {208 return false209 }210 _, err := t.agentUploadGrant(ctx, pubkey)211 return err == nil212}213214func (t *Tenant) fileReadAccess(ctx context.Context, actor, hash string) error {215 access, err := t.blobs.BlobAccess(ctx, hash)216 if err != nil {217 return err218 }219 if access != blob.AccessMembers {220 return nil221 }222 if actor == "" {223 return errors.New("auth-required: member file requires sign in")224 }225 banned, err := t.community.IsBanned(ctx, actor)226 if err != nil {227 return fmt.Errorf("check file reader ban: %w", err)228 }229 if banned {230 return errors.New("blocked: file reader is banned")231 }232 role, err := t.community.Role(ctx, actor)233 if err != nil {234 return fmt.Errorf("check file reader role: %w", err)235 }236 if !t.memberBlobReader(ctx, actor, role) {237 return errors.New("restricted: relay membership required to read this file")238 }239 return nil240}241242// backend adapts the management method signature to webui's form-oriented API.243type backend struct{ tenant *Tenant }244245type localDirectory struct{ store *storage.Store }246247func (d localDirectory) WriteRelays(pubkey string) []string {248 return d.relaysForDirection(pubkey, true)249}250func (d localDirectory) ReadRelays(pubkey string) []string {251 return d.relaysForDirection(pubkey, false)252}253func (d localDirectory) RelayList(pubkey string) []string { return d.relays(pubkey, false, true) }254func (d localDirectory) relaysForDirection(pubkey string, write bool) []string {255 return d.relays(pubkey, write, false)256}257func (d localDirectory) relays(pubkey string, write, all bool) []string {258 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})259 if err != nil || len(rows.Events) == 0 {260 return nil261 }262 result := make([]string, 0)263 for _, tag := range rows.Events[0].Tags {264 if len(tag) < 2 || tag[0] != "r" || (!all && !write && len(tag) > 2 && tag[2] != "read") || (!all && write && len(tag) > 2 && tag[2] == "read") {265 continue266 }267 result = append(result, tag[1])268 }269 return result270}271272func (b backend) Query(ctx context.Context, method string, params []json.RawMessage, actor string) (any, error) {273 return b.tenant.Execute(ctx, actor, method, params)274}275func (b backend) Policy() policy.Policy { return b.tenant.Policy() }276func (b backend) URL() string { return b.tenant.publicURL }277func (b backend) Slug() string { return b.tenant.meta.Name }278func (b backend) Identity() string { return b.tenant.records.PublicKey() }279280// CustomViews gives the page renderers the enabled custom views' names and281// languages, never a transform or a secret.282func (b backend) CustomViews() []views.View { return b.tenant.customViews.CustomViews() }283284// NameFor and PubKeyFor let web pages write and read people by member name.285func (b backend) NameFor(ctx context.Context, pubkey string) string {286 if b.tenant.community == nil {287 return ""288 }289 name, err := b.tenant.community.MemberName(ctx, pubkey)290 if err != nil {291 return ""292 }293 return name294}295func (b backend) PubKeyFor(ctx context.Context, name string) string {296 if b.tenant.community == nil {297 return ""298 }299 member, err := b.tenant.community.MemberByName(ctx, name)300 if err != nil {301 return ""302 }303 return member.PubKey304}305306// ReadAllowed is consumed by the web UI private boundary. It deliberately307// delegates to the same uncached membership check used by private Git and308// NIP-42, so a revoked browser session cannot retain page access.309func (b backend) ReadAllowed(ctx context.Context, actor string) error {310 if !b.tenant.PrivateServiceEnabled() {311 return nil312 }313 return b.tenant.requirePrivateAccess(ctx, actor)314}315316func (b backend) SharePresence(ctx context.Context, actor string) bool {317 return b.tenant.chatSharePresence(ctx, actor)318}319320// closeServices cancels and joins the tenant's worker, scheduler and Git321// loops, then returns the worker's terminal queue error, if any. Store and322// relay connection cleanup belong to Tenant.Close.323func (t *Tenant) closeServices(ctx context.Context) error {324 if t.workCancel != nil {325 t.workCancel()326 }327 t.workWG.Wait()328 t.workErrMu.Lock()329 defer t.workErrMu.Unlock()330 return t.workErr331}332333func hostName(raw string) string {334 u, err := url.Parse(raw)335 if err != nil {336 return "localhost"337 }338 if u.Hostname() == "" {339 return "localhost"340 }341 return u.Hostname()342}343func (t *Tenant) siteReadAccess(r *http.Request) bool {344 actor, err := t.resolveUIActor(r)345 if err != nil {346 return false347 }348 keys := []string{}349 if actor != "" {350 keys = append(keys, actor)351 }352 _, err = t.gate.Read(r.Context(), []event.Filter{{}}, relay.Session{PubKeys: keys, RelayURL: t.RelayURL()})353 return err == nil354}355func (t *Tenant) authorizeBlob(r *http.Request, action blob.Action) (string, error) {356 hash := ""357 if action == blob.ActionGet || action == blob.ActionDelete || (action == blob.ActionUpload && (r.Method == http.MethodPut || r.Method == http.MethodPatch)) {358 path := strings.TrimPrefix(r.URL.Path, "/")359 path = strings.TrimPrefix(path, "nip96/")360 path = strings.TrimPrefix(path, "media/")361 hash = strings.Split(path, ".")[0]362 if len(hash) != 64 || strings.Trim(hash, "0123456789abcdef") != "" {363 hash = ""364 }365 }366 body, bodyErr := blobAuthBody(r, action)367 if bodyErr != nil {368 return "", bodyErr369 }370 pubkey := ""371 var err error372 if action == blob.ActionUpload && r.Method == http.MethodPost && r.URL.Path == "/nip96" {373 var token event.Event374 token, err = t.auth.VerifyNIP98(r.Header.Get("Authorization"), t.requestURL(r), r.Method, body)375 pubkey = token.PubKey376 } else {377 if auth.IsBlossomAuthorization(r.Header.Get("Authorization")) {378 blossomAction := string(action)379 if action == blob.ActionMirror {380 // BUD-04 mirror uses the upload authorization verb; the fetched381 // bytes are the resulting blob and are scoped by x below.382 blossomAction = string(blob.ActionUpload)383 }384 blobHash := hash385 if blobHash == "" && action == blob.ActionUpload {386 blobHash = strings.ToLower(strings.TrimSpace(r.Header.Get("x-sha-256")))387 }388 var token event.Event389 token, err = t.auth.VerifyBlossomRequest(r.Header.Get("Authorization"), blossomAction, t.siteDomain(), blobHash)390 pubkey = token.PubKey391 } else {392 if isBUD13RemoteUpload(r) {393 var token event.Event394 token, err = t.auth.VerifyNIP98(r.Header.Get("Authorization"), t.requestURL(r), r.Method, "")395 if err == nil && event.Tag(token, "payload") != "" && event.Tag(token, "payload") != emptyPayloadHash() {396 err = errors.New("auth-required: remote upload token payload must hash an empty body")397 }398 pubkey = token.PubKey399 } else if action == blob.ActionUpload && (r.Method == http.MethodPut || r.Method == http.MethodPatch) {400 var token event.Event401 token, err = t.auth.VerifyNIP98Deferred(r.Header.Get("Authorization"), t.requestURL(r), r.Method)402 pubkey = token.PubKey403 } else {404 var keys []string405 keys, err = t.auth.WhoAsks(r.Header.Get("Authorization"), t.requestURL(r), r.Method, body, string(action), hash)406 pubkey = first(keys)407 }408 }409 }410 if err != nil {411 return "", err412 }413 if action == blob.ActionGet {414 if err := t.roomAttachmentAccess(r.Context(), pubkey, hash); err != nil {415 return "", err416 }417 }418 if pubkey == "" {419 if action == blob.ActionGet {420 return "", nil421 }422 return "", errors.New("auth-required: blob authorization required")423 }424 if banned, banErr := t.community.IsBanned(r.Context(), pubkey); banErr != nil {425 return "", fmt.Errorf("check blob uploader ban: %w", banErr)426 } else if banned {427 return "", errors.New("blocked: this pubkey is banned from this relay")428 }429 if action == blob.ActionUpload || action == blob.ActionMirror {430 if !t.Policy().Features.Files {431 return "", errors.New("restricted: file uploads are disabled")432 }433 role, roleErr := t.community.Role(r.Context(), pubkey)434 if roleErr != nil {435 return "", fmt.Errorf("check blob uploader role: %w", roleErr)436 }437 allowed := t.Policy().Writes == "open" || (t.Policy().Writes == "owner" && role == "owner") || (t.Policy().Writes != "owner" && role != "")438 if !allowed {439 return "", errors.New("restricted: file upload is not allowed")440 }441 // An agent's standing is its grant: an active one makes it a member442 // for uploads, a paused, revoked or expired one stops them.443 if role == "agent" {444 if _, err := t.agentUploadGrant(r.Context(), pubkey); err != nil {445 return "", err446 }447 }448 }449 return pubkey, nil450}451452// agentUploadGrant returns the grant behind an agent key's uploads when it453// currently allows publishing.454func (t *Tenant) agentUploadGrant(ctx context.Context, pubkey string) (community.AgentGrant, error) {455 grant, ok, err := t.community.AgentGrant(ctx, pubkey)456 if err != nil {457 return community.AgentGrant{}, fmt.Errorf("check agent upload grant: %w", err)458 }459 if !ok {460 return community.AgentGrant{}, errors.New("restricted: agent grant is missing")461 }462 switch {463 case grant.RevokedAt > 0:464 return community.AgentGrant{}, errors.New("restricted: agent grant is revoked")465 case grant.Paused:466 return community.AgentGrant{}, errors.New("restricted: agent grant is paused")467 case grant.ExpiresAt <= time.Now().Unix():468 return community.AgentGrant{}, errors.New("restricted: agent grant has expired")469 }470 return grant, nil471}472473// blobUploadTerms is what the blob service asks before it stores an upload:474// a key whose only standing is an agent grant takes the grant's sites ttl as475// the expiry of its claims and its encrypted flag as a requirement. Keys476// with a human role, and agents without a sites tag, upload on no terms.477func (t *Tenant) blobUploadTerms(ctx context.Context, pubkey string) (blob.UploadTerms, error) {478 role, err := t.community.Role(ctx, pubkey)479 if err != nil {480 return blob.UploadTerms{}, fmt.Errorf("check uploader role: %w", err)481 }482 if role != "agent" {483 return blob.UploadTerms{}, nil484 }485 grant, err := t.agentUploadGrant(ctx, pubkey)486 if err != nil {487 return blob.UploadTerms{}, err488 }489 terms := blob.UploadTerms{Encrypted: grant.UploadsEncrypted()}490 if days := grant.UploadTTLDays(); days > 0 {491 terms.ExpiresAt = time.Now().Unix() + int64(days)*86400492 }493 return terms, nil494}495496// sweepExpiredBlobs retires uploads whose agent ttl has lapsed. The step is497// recorded as the blob-expiry storage operation with a bounded outcome and498// a count, never a hash or a key.499func (t *Tenant) sweepExpiredBlobs(ctx context.Context, now int64) error {500 if t.blobs == nil {501 return nil502 }503 started := time.Now()504 removed, err := t.blobs.SweepExpired(ctx, now)505 outcome := "ok"506 if err != nil {507 outcome = "error"508 }509 t.app.telemetry.ObserveStorage("blob-expiry", outcome, time.Since(started))510 if removed > 0 {511 t.app.telemetry.Logger().Info("expired uploads removed", "tenant", t.meta.Name, "blobs", removed)512 }513 return err514}515516func (t *Tenant) validateBlobUpload(r *http.Request, hash string) error {517 // PATCH carries only one chunk. Blossom scopes the final path hash, while518 // NIP-98 binds the bytes in this individual request body.519 if r.Method == http.MethodPatch && auth.IsBlossomAuthorization(r.Header.Get("Authorization")) {520 finalHash := strings.Split(strings.TrimPrefix(r.URL.Path, "/"), ".")[0]521 return t.auth.ValidateBlossomPayload(r.Header.Get("Authorization"), string(blob.ActionUpload), finalHash)522 }523 // Mirror proofs bind the HTTP request body before fetching. Blossom524 // authorization additionally names the fetched content through its x tag.525 if r.URL.Path == "/mirror" || isBUD13RemoteUpload(r) {526 if auth.IsBlossomAuthorization(r.Header.Get("Authorization")) {527 return t.auth.ValidateBlossomPayload(r.Header.Get("Authorization"), string(blob.ActionUpload), hash)528 }529 return nil530 }531 return t.auth.ValidateBlobPayload(r.Header.Get("Authorization"), hash)532}533534// NIP-98 binds its payload to the complete HTTP body. The streaming Blossom535// upload remains unbuffered and is checked against its resulting SHA-256 by536// ValidateUpload; the JSON and multipart doors are buffered only so their537// signed request can be verified before parsing or fetching any content.538func blobAuthBody(r *http.Request, action blob.Action) (string, error) {539 buffer := action == blob.ActionMirror || action == blob.ActionReport ||540 (action == blob.ActionUpload && r.Method == http.MethodPost && (r.URL.Path == "/nip96" || r.URL.Path == "/files/metadata"))541 if !buffer {542 return "", nil543 }544 var source io.Reader = r.Body545 if r.URL.Path == "/files/metadata" {546 source = io.LimitReader(r.Body, 16*1024+1)547 }548 body, err := io.ReadAll(source)549 if err != nil {550 return "", fmt.Errorf("read blob authorization body: %w", err)551 }552 if r.URL.Path == "/files/metadata" && len(body) > 16*1024 {553 return "", errors.New("invalid: file metadata exceeds 16 KiB")554 }555 r.Body = io.NopCloser(bytes.NewReader(body))556 return string(body), nil557}558559func (t *Tenant) siteDomain() string {560 if t.meta.Name != t.app.cfg.DefaultTenant {561 return t.app.siteBase(t.meta)562 }563 host := hostName(t.publicURL)564 if ip := net.ParseIP(host); ip != nil && ip.IsLoopback() {565 return "localhost"566 }567 return host568}569func (t *Tenant) authorizeGit(ctx context.Context, r *http.Request, repo gitrelay.Repository) error {570 if !t.Policy().Features.Grasp {571 return errors.New("restricted: Git hosting is disabled")572 }573 // Authenticate before reading a potentially large pack. The body binding574 // is checked below before any bytes reach git-receive-pack.575 e, ok := privateGitProof(r)576 var err error577 if !ok {578 e, err = t.auth.VerifyNIP98Deferred(r.Header.Get("Authorization"), t.requestURL(r), r.Method)579 if err != nil {580 return err581 }582 }583 if banned, banErr := t.community.IsBanned(ctx, e.PubKey); banErr != nil {584 return fmt.Errorf("check Git author ban: %w", banErr)585 } else if banned {586 return errors.New("blocked: this pubkey is banned from this relay")587 }588 if privatePolicy(t.Policy()) {589 // Tenant membership is the private-service boundary. A repository590 // maintainer or owner who is not a member must not learn whether a591 // private repository exists through Smart HTTP.592 if err := t.requirePrivateAccess(ctx, e.PubKey); err != nil {593 return err594 }595 } else if !t.IsMaintainer(ctx, repo, e.PubKey) {596 role, roleErr := t.community.Role(ctx, e.PubKey)597 if roleErr != nil || (role != "owner" && role != "moderator" && role != "member") {598 return errors.New("restricted: repository authorization required")599 }600 }601 if _, checked := r.Context().Value(privateGitPayloadCheckedKey{}).(struct{}); checked {602 return nil603 }604 return spoolGitPayload(r, e)605}606func first(v []string) string {607 if len(v) == 0 {608 return ""609 }610 return v[0]611}612613func isBUD13RemoteUpload(r *http.Request) bool {614 if r == nil || r.Method != http.MethodPut || r.URL.Query().Get("url") == "" {615 return false616 }617 path := strings.TrimPrefix(r.URL.Path, "/")618 hash := strings.Split(path, ".")[0]619 return len(hash) == 64 && strings.Trim(hash, "0123456789abcdefABCDEF") == ""620}621622func emptyPayloadHash() string {623 sum := sha256.Sum256(nil)624 return hex.EncodeToString(sum[:])625}626627func (t *Tenant) resolveUIActor(r *http.Request) (string, error) {628 if r.Header.Get("Authorization") == "" {629 if r.Method != http.MethodGet && r.Method != http.MethodHead {630 return "", errors.New("auth-required: sign this request")631 }632 return t.cookieActor(r)633 }634 if r.Body == nil {635 r.Body = http.NoBody636 }637 body, err := io.ReadAll(r.Body)638 if err != nil {639 return "", err640 }641 r.Body = io.NopCloser(bytes.NewReader(body))642 e, err := t.auth.VerifyNIP98(r.Header.Get("Authorization"), t.requestURL(r), r.Method, string(body))643 if err != nil {644 return "", err645 }646 return e.PubKey, nil647}648649func (t *Tenant) workHandlers() map[string]work.Handler {650 handlers := map[string]work.Handler{651 "view-publish": func(ctx context.Context, intent work.Intent) error {652 if err := t.records.MarkView(ctx, intent.Target, time.Now().Unix()); err != nil {653 return err654 }655 t.scheduleViews()656 return nil657 },658 "git-metadata": func(ctx context.Context, intent work.Intent) error {659 var raw string660 err := t.store.DB().QueryRowContext(ctx, "SELECT raw FROM events WHERE id=?", intent.EventID).Scan(&raw)661 if errors.Is(err, sql.ErrNoRows) {662 return nil663 }664 if err != nil {665 return err666 }667 e, err := event.Parse([]byte(raw))668 if err != nil {669 return err670 }671 var repo gitrelay.Repository672 if err := json.Unmarshal([]byte(intent.Payload), &repo); err != nil {673 return err674 }675 if err := t.git.CommitAfterStore(ctx, e, repo); err != nil {676 return err677 }678 if !t.git.IsPending(e.ID) {679 return t.releaseGit(ctx, e.ID)680 }681 return nil682 },683 "catalog-owner": func(ctx context.Context, intent work.Intent) error {684 return t.app.catalog.UpdateOwner(ctx, t.meta.ID, t.Policy().Owner)685 },686 "records-projection": func(ctx context.Context, intent work.Intent) error {687 var payload struct {688 Action string `json:"action"`689 Room string `json:"room"`690 PubKey string `json:"pubkey"`691 Role string `json:"role"`692 Removed int `json:"removed"`693 Deleted int `json:"deleted"`694 }695 if err := json.Unmarshal([]byte(intent.Payload), &payload); err != nil {696 return err697 }698 if payload.Action == "room" {699 return t.projectRoom(ctx, payload.Room, payload.PubKey, payload.Removed != 0, payload.Deleted != 0)700 }701 var changes []records.MembershipChange702 if payload.PubKey != "" && (payload.Role != "" || payload.Removed != 0) {703 added := payload.Removed == 0704 changes = []records.MembershipChange{{PubKey: payload.PubKey, Added: &added, Roles: []string{payload.Role}}}705 }706 _, err := t.records.PublishMembershipChanges(ctx, changes, time.Now().Unix())707 return err708 },709 "site-mirror": func(ctx context.Context, intent work.Intent) error {710 return t.sites.RunMirrorIntent(ctx, storage.Intent{Kind: intent.Kind, EventID: intent.EventID, Target: intent.Target, Payload: intent.Payload})711 },712 "callback": t.ReplicationCallbackHandler(),713 }714 for kind, handler := range t.notificationHandlers() {715 handlers[kind] = handler716 }717 handlers[notificationPush] = t.handleNotificationPush718 handlers[callbackDelivery] = t.handleCallbackDelivery719 handlers[viewTransform] = t.customViews.handleViewTransform720 handlers[eventFollowupKind] = t.followups.handle721 return handlers722}723724// releaseGit publishes a repository state after Git verifies its objects. It725// removes the pending marker and persists optional planning in one transaction726// under the router's publication lock. Push planning retains the registrations727// captured at client acceptance; webhook and view matching occurs at release.728func (t *Tenant) releaseGit(ctx context.Context, id string) error {729 var followup *storage.Intent730 if t.followups != nil {731 var raw []byte732 err := t.store.DB().QueryRowContext(ctx, "SELECT raw FROM events WHERE id=?", id).Scan(&raw)733 if err != nil && !errors.Is(err, sql.ErrNoRows) {734 return err735 }736 if err == nil {737 e, err := event.Parse(raw)738 if err != nil {739 return err740 }741 followup = t.followups.prepare(ctx, e, time.Now().Unix(), true)742 }743 }744 released, err := t.router.CommitGenerated(ctx, func(ctx context.Context) (event.Event, error) {745 var promoted event.Event746 err := t.store.WithTx(ctx, func(tx *sql.Tx) error {747 result, err := tx.ExecContext(ctx, "DELETE FROM pending_events WHERE id=?", id)748 if err != nil {749 return err750 }751 n, err := result.RowsAffected()752 if err != nil || n == 0 {753 return err754 }755 if _, err := tx.ExecContext(ctx, "DELETE FROM replication_pending_until WHERE id=?", id); err != nil && !isMissingTable(err) {756 return err757 }758 var raw []byte759 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)760 if errors.Is(err, sql.ErrNoRows) {761 return nil762 }763 if err != nil {764 return err765 }766 promoted, err = event.Parse(raw)767 if err != nil || followup == nil {768 return err769 }770 return storage.AddIntents(ctx, tx, []storage.Intent{*followup}, time.Now().Unix())771 })772 return promoted, err773 })774 if err == nil && released.ID != "" {775 // A repository state is hidden until its objects arrive, so agents776 // that wait for pushes are woken here rather than at publish time.777 if t.customViews != nil && t.customViews.git != nil {778 t.followups.try(ctx, followup)779 }780 }781 return err782}783784func (t *Tenant) sweep(ctx context.Context, now int64) error {785 if _, err := t.store.SweepExpired(ctx, now); err != nil {786 return err787 }788 opts := storage.RetentionOptions{Now: now, Identity: t.records.PublicKey()}789 rules, err := t.community.RetentionRules(ctx)790 if err != nil {791 return err792 }793 for _, rule := range rules {794 kind := -1795 if rule.Kind != nil {796 kind = *rule.Kind797 }798 opts.Rules = append(opts.Rules, storage.RetentionRule{Kind: kind, Days: rule.Days})799 }800 members, err := t.community.MembersForRetention(ctx)801 if err != nil {802 return err803 }804 for _, member := range members {805 opts.Members = append(opts.Members, storage.MemberRetention{PubKey: member.PubKey, Days: member.Days})806 }807 if _, err := t.store.SweepRetention(ctx, opts); err != nil {808 return err809 }810 return t.customViews.pruneAllViewArtifacts(ctx)811}812func (t *Tenant) replacePolicy(p policy.Policy) {813 t.mu.Lock()814 previous := t.policy815 t.policy = p816 t.mu.Unlock()817 if community.RoomAccessFor(previous.Reads) != community.RoomAccessFor(p.Reads) {818 if err := t.community.EnsureSlugRoom(context.Background(), t.meta.Name, community.RoomAccessFor(p.Reads)); err != nil {819 t.app.telemetry.Logger().Error("refresh main room", "tenant", t.meta.Name, "error", err)820 }821 }822 if err := t.reconcileAutomaticInbox(context.Background(), previous, p); err != nil {823 t.app.telemetry.Logger().Error("reconcile automatic inbox job", "tenant", t.meta.Name, "error", err)824 }825 t.router.CloseSubscriptions("blocked: relay policy changed; subscribe again")826 t.scheduleViews()827}828func (t *Tenant) generatedRecord(ctx context.Context, e event.Event) error {829 ctx, done, admissionErr := t.beginOperation(ctx)830 if admissionErr != nil {831 return admissionErr832 }833 defer done()834 _, err := t.router.CommitGenerated(ctx, func(ctx context.Context) (event.Event, error) {835 _, err := t.store.Save(ctx, e, storage.SaveOptions{Now: e.CreatedAt, SearchMode: t.Policy().Features.Search})836 return e, err837 })838 return err839}840func (t *Tenant) ingest(ctx context.Context, e event.Event, _ replication.Origin) error {841 ctx, done, admissionErr := t.beginOperation(ctx)842 if admissionErr != nil {843 return admissionErr844 }845 defer done()846 _, err := t.router.CommitGenerated(ctx, func(ctx context.Context) (event.Event, error) {847 return e, t.commitImported(ctx, e, replication.OriginImport)848 })849 if errors.Is(err, storage.ErrDuplicate) || errors.Is(err, storage.ErrReplaced) {850 return nil851 }852 return err853}854855// commitImported applies import admission and commits the shared projections856// and optional planning for an event. It returns persistence errors to the857// caller, which decides how to handle duplicate and superseded imports.858// origin supplies the replication source for planning.859func (t *Tenant) commitImported(ctx context.Context, e event.Event, origin replication.Origin) error {860 if err := t.gate.Import(ctx, e, time.Now().Unix()); err != nil {861 return err862 }863 if err := t.validateStoredEvent(ctx, e); err != nil {864 return err865 }866 plan, err := t.prepareStoredEvent(ctx, e, origin, time.Now().Unix())867 if err != nil {868 return err869 }870 _, err = t.store.Save(ctx, e, plan.options)871 if err == nil {872 t.afterStoredEvent(ctx, e, plan.followups)873 }874 return err875}