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// ReadAllowed is consumed by the web UI private boundary. It deliberately285// delegates to the same uncached membership check used by private Git and286// NIP-42, so a revoked browser session cannot retain page access.287func (b backend) ReadAllowed(ctx context.Context, actor string) error {288 if !b.tenant.PrivateServiceEnabled() {289 return nil290 }291 return b.tenant.requirePrivateAccess(ctx, actor)292}293294func (b backend) SharePresence(ctx context.Context, actor string) bool {295 return b.tenant.chatSharePresence(ctx, actor)296}297298// closeServices cancels and joins the tenant's worker, scheduler and Git299// loops, then returns the worker's terminal queue error, if any. Store and300// relay connection cleanup belong to Tenant.Close.301func (t *Tenant) closeServices(ctx context.Context) error {302 if t.workCancel != nil {303 t.workCancel()304 }305 t.workWG.Wait()306 t.workErrMu.Lock()307 defer t.workErrMu.Unlock()308 return t.workErr309}310311func hostName(raw string) string {312 u, err := url.Parse(raw)313 if err != nil {314 return "localhost"315 }316 if u.Hostname() == "" {317 return "localhost"318 }319 return u.Hostname()320}321func (t *Tenant) siteReadAccess(r *http.Request) bool {322 actor, err := t.resolveUIActor(r)323 if err != nil {324 return false325 }326 keys := []string{}327 if actor != "" {328 keys = append(keys, actor)329 }330 _, err = t.gate.Read(r.Context(), []event.Filter{{}}, relay.Session{PubKeys: keys, RelayURL: t.RelayURL()})331 return err == nil332}333func (t *Tenant) authorizeBlob(r *http.Request, action blob.Action) (string, error) {334 hash := ""335 if action == blob.ActionGet || action == blob.ActionDelete || (action == blob.ActionUpload && (r.Method == http.MethodPut || r.Method == http.MethodPatch)) {336 path := strings.TrimPrefix(r.URL.Path, "/")337 path = strings.TrimPrefix(path, "nip96/")338 path = strings.TrimPrefix(path, "media/")339 hash = strings.Split(path, ".")[0]340 if len(hash) != 64 || strings.Trim(hash, "0123456789abcdef") != "" {341 hash = ""342 }343 }344 body, bodyErr := blobAuthBody(r, action)345 if bodyErr != nil {346 return "", bodyErr347 }348 pubkey := ""349 var err error350 if action == blob.ActionUpload && r.Method == http.MethodPost && r.URL.Path == "/nip96" {351 var token event.Event352 token, err = t.auth.VerifyNIP98(r.Header.Get("Authorization"), t.requestURL(r), r.Method, body)353 pubkey = token.PubKey354 } else {355 if auth.IsBlossomAuthorization(r.Header.Get("Authorization")) {356 blossomAction := string(action)357 if action == blob.ActionMirror {358 // BUD-04 mirror uses the upload authorization verb; the fetched359 // bytes are the resulting blob and are scoped by x below.360 blossomAction = string(blob.ActionUpload)361 }362 blobHash := hash363 if blobHash == "" && action == blob.ActionUpload {364 blobHash = strings.ToLower(strings.TrimSpace(r.Header.Get("x-sha-256")))365 }366 var token event.Event367 token, err = t.auth.VerifyBlossomRequest(r.Header.Get("Authorization"), blossomAction, t.siteDomain(), blobHash)368 pubkey = token.PubKey369 } else {370 if isBUD13RemoteUpload(r) {371 var token event.Event372 token, err = t.auth.VerifyNIP98(r.Header.Get("Authorization"), t.requestURL(r), r.Method, "")373 if err == nil && event.Tag(token, "payload") != "" && event.Tag(token, "payload") != emptyPayloadHash() {374 err = errors.New("auth-required: remote upload token payload must hash an empty body")375 }376 pubkey = token.PubKey377 } else if action == blob.ActionUpload && (r.Method == http.MethodPut || r.Method == http.MethodPatch) {378 var token event.Event379 token, err = t.auth.VerifyNIP98Deferred(r.Header.Get("Authorization"), t.requestURL(r), r.Method)380 pubkey = token.PubKey381 } else {382 var keys []string383 keys, err = t.auth.WhoAsks(r.Header.Get("Authorization"), t.requestURL(r), r.Method, body, string(action), hash)384 pubkey = first(keys)385 }386 }387 }388 if err != nil {389 return "", err390 }391 if action == blob.ActionGet {392 if err := t.roomAttachmentAccess(r.Context(), pubkey, hash); err != nil {393 return "", err394 }395 }396 if pubkey == "" {397 if action == blob.ActionGet {398 return "", nil399 }400 return "", errors.New("auth-required: blob authorization required")401 }402 if banned, banErr := t.community.IsBanned(r.Context(), pubkey); banErr != nil {403 return "", fmt.Errorf("check blob uploader ban: %w", banErr)404 } else if banned {405 return "", errors.New("blocked: this pubkey is banned from this relay")406 }407 if action == blob.ActionUpload || action == blob.ActionMirror {408 if !t.Policy().Features.Files {409 return "", errors.New("restricted: file uploads are disabled")410 }411 role, roleErr := t.community.Role(r.Context(), pubkey)412 if roleErr != nil {413 return "", fmt.Errorf("check blob uploader role: %w", roleErr)414 }415 allowed := t.Policy().Writes == "open" || (t.Policy().Writes == "owner" && role == "owner") || (t.Policy().Writes != "owner" && role != "")416 if !allowed {417 return "", errors.New("restricted: file upload is not allowed")418 }419 // An agent's standing is its grant: an active one makes it a member420 // for uploads, a paused, revoked or expired one stops them.421 if role == "agent" {422 if _, err := t.agentUploadGrant(r.Context(), pubkey); err != nil {423 return "", err424 }425 }426 }427 return pubkey, nil428}429430// agentUploadGrant returns the grant behind an agent key's uploads when it431// currently allows publishing.432func (t *Tenant) agentUploadGrant(ctx context.Context, pubkey string) (community.AgentGrant, error) {433 grant, ok, err := t.community.AgentGrant(ctx, pubkey)434 if err != nil {435 return community.AgentGrant{}, fmt.Errorf("check agent upload grant: %w", err)436 }437 if !ok {438 return community.AgentGrant{}, errors.New("restricted: agent grant is missing")439 }440 switch {441 case grant.RevokedAt > 0:442 return community.AgentGrant{}, errors.New("restricted: agent grant is revoked")443 case grant.Paused:444 return community.AgentGrant{}, errors.New("restricted: agent grant is paused")445 case grant.ExpiresAt <= time.Now().Unix():446 return community.AgentGrant{}, errors.New("restricted: agent grant has expired")447 }448 return grant, nil449}450451// blobUploadTerms is what the blob service asks before it stores an upload:452// a key whose only standing is an agent grant takes the grant's sites ttl as453// the expiry of its claims and its encrypted flag as a requirement. Keys454// with a human role, and agents without a sites tag, upload on no terms.455func (t *Tenant) blobUploadTerms(ctx context.Context, pubkey string) (blob.UploadTerms, error) {456 role, err := t.community.Role(ctx, pubkey)457 if err != nil {458 return blob.UploadTerms{}, fmt.Errorf("check uploader role: %w", err)459 }460 if role != "agent" {461 return blob.UploadTerms{}, nil462 }463 grant, err := t.agentUploadGrant(ctx, pubkey)464 if err != nil {465 return blob.UploadTerms{}, err466 }467 terms := blob.UploadTerms{Encrypted: grant.UploadsEncrypted()}468 if days := grant.UploadTTLDays(); days > 0 {469 terms.ExpiresAt = time.Now().Unix() + int64(days)*86400470 }471 return terms, nil472}473474// sweepExpiredBlobs retires uploads whose agent ttl has lapsed. The step is475// recorded as the blob-expiry storage operation with a bounded outcome and476// a count, never a hash or a key.477func (t *Tenant) sweepExpiredBlobs(ctx context.Context, now int64) error {478 if t.blobs == nil {479 return nil480 }481 started := time.Now()482 removed, err := t.blobs.SweepExpired(ctx, now)483 outcome := "ok"484 if err != nil {485 outcome = "error"486 }487 t.app.telemetry.ObserveStorage("blob-expiry", outcome, time.Since(started))488 if removed > 0 {489 t.app.telemetry.Logger().Info("expired uploads removed", "tenant", t.meta.Name, "blobs", removed)490 }491 return err492}493494func (t *Tenant) validateBlobUpload(r *http.Request, hash string) error {495 // PATCH carries only one chunk. Blossom scopes the final path hash, while496 // NIP-98 binds the bytes in this individual request body.497 if r.Method == http.MethodPatch && auth.IsBlossomAuthorization(r.Header.Get("Authorization")) {498 finalHash := strings.Split(strings.TrimPrefix(r.URL.Path, "/"), ".")[0]499 return t.auth.ValidateBlossomPayload(r.Header.Get("Authorization"), string(blob.ActionUpload), finalHash)500 }501 // Mirror proofs bind the HTTP request body before fetching. Blossom502 // authorization additionally names the fetched content through its x tag.503 if r.URL.Path == "/mirror" || isBUD13RemoteUpload(r) {504 if auth.IsBlossomAuthorization(r.Header.Get("Authorization")) {505 return t.auth.ValidateBlossomPayload(r.Header.Get("Authorization"), string(blob.ActionUpload), hash)506 }507 return nil508 }509 return t.auth.ValidateBlobPayload(r.Header.Get("Authorization"), hash)510}511512// NIP-98 binds its payload to the complete HTTP body. The streaming Blossom513// upload remains unbuffered and is checked against its resulting SHA-256 by514// ValidateUpload; the JSON and multipart doors are buffered only so their515// signed request can be verified before parsing or fetching any content.516func blobAuthBody(r *http.Request, action blob.Action) (string, error) {517 buffer := action == blob.ActionMirror || action == blob.ActionReport ||518 (action == blob.ActionUpload && r.Method == http.MethodPost && (r.URL.Path == "/nip96" || r.URL.Path == "/files/metadata"))519 if !buffer {520 return "", nil521 }522 var source io.Reader = r.Body523 if r.URL.Path == "/files/metadata" {524 source = io.LimitReader(r.Body, 16*1024+1)525 }526 body, err := io.ReadAll(source)527 if err != nil {528 return "", fmt.Errorf("read blob authorization body: %w", err)529 }530 if r.URL.Path == "/files/metadata" && len(body) > 16*1024 {531 return "", errors.New("invalid: file metadata exceeds 16 KiB")532 }533 r.Body = io.NopCloser(bytes.NewReader(body))534 return string(body), nil535}536537func (t *Tenant) siteDomain() string {538 if t.meta.Name != t.app.cfg.DefaultTenant {539 return t.app.siteBase(t.meta)540 }541 host := hostName(t.publicURL)542 if ip := net.ParseIP(host); ip != nil && ip.IsLoopback() {543 return "localhost"544 }545 return host546}547func (t *Tenant) authorizeGit(ctx context.Context, r *http.Request, repo gitrelay.Repository) error {548 if !t.Policy().Features.Grasp {549 return errors.New("restricted: Git hosting is disabled")550 }551 // Authenticate before reading a potentially large pack. The body binding552 // is checked below before any bytes reach git-receive-pack.553 e, ok := privateGitProof(r)554 var err error555 if !ok {556 e, err = t.auth.VerifyNIP98Deferred(r.Header.Get("Authorization"), t.requestURL(r), r.Method)557 if err != nil {558 return err559 }560 }561 if banned, banErr := t.community.IsBanned(ctx, e.PubKey); banErr != nil {562 return fmt.Errorf("check Git author ban: %w", banErr)563 } else if banned {564 return errors.New("blocked: this pubkey is banned from this relay")565 }566 if privatePolicy(t.Policy()) {567 // Tenant membership is the private-service boundary. A repository568 // maintainer or owner who is not a member must not learn whether a569 // private repository exists through Smart HTTP.570 if err := t.requirePrivateAccess(ctx, e.PubKey); err != nil {571 return err572 }573 } else if !t.IsMaintainer(ctx, repo, e.PubKey) {574 role, roleErr := t.community.Role(ctx, e.PubKey)575 if roleErr != nil || (role != "owner" && role != "moderator" && role != "member") {576 return errors.New("restricted: repository authorization required")577 }578 }579 if _, checked := r.Context().Value(privateGitPayloadCheckedKey{}).(struct{}); checked {580 return nil581 }582 return spoolGitPayload(r, e)583}584func first(v []string) string {585 if len(v) == 0 {586 return ""587 }588 return v[0]589}590591func isBUD13RemoteUpload(r *http.Request) bool {592 if r == nil || r.Method != http.MethodPut || r.URL.Query().Get("url") == "" {593 return false594 }595 path := strings.TrimPrefix(r.URL.Path, "/")596 hash := strings.Split(path, ".")[0]597 return len(hash) == 64 && strings.Trim(hash, "0123456789abcdefABCDEF") == ""598}599600func emptyPayloadHash() string {601 sum := sha256.Sum256(nil)602 return hex.EncodeToString(sum[:])603}604605func (t *Tenant) resolveUIActor(r *http.Request) (string, error) {606 if r.Header.Get("Authorization") == "" {607 if r.Method != http.MethodGet && r.Method != http.MethodHead {608 return "", errors.New("auth-required: sign this request")609 }610 return t.cookieActor(r)611 }612 if r.Body == nil {613 r.Body = http.NoBody614 }615 body, err := io.ReadAll(r.Body)616 if err != nil {617 return "", err618 }619 r.Body = io.NopCloser(bytes.NewReader(body))620 e, err := t.auth.VerifyNIP98(r.Header.Get("Authorization"), t.requestURL(r), r.Method, string(body))621 if err != nil {622 return "", err623 }624 return e.PubKey, nil625}626627func (t *Tenant) workHandlers() map[string]work.Handler {628 handlers := map[string]work.Handler{629 "view-publish": func(ctx context.Context, intent work.Intent) error {630 if err := t.records.MarkView(ctx, intent.Target, time.Now().Unix()); err != nil {631 return err632 }633 t.scheduleViews()634 return nil635 },636 "git-metadata": func(ctx context.Context, intent work.Intent) error {637 var raw string638 err := t.store.DB().QueryRowContext(ctx, "SELECT raw FROM events WHERE id=?", intent.EventID).Scan(&raw)639 if errors.Is(err, sql.ErrNoRows) {640 return nil641 }642 if err != nil {643 return err644 }645 e, err := event.Parse([]byte(raw))646 if err != nil {647 return err648 }649 var repo gitrelay.Repository650 if err := json.Unmarshal([]byte(intent.Payload), &repo); err != nil {651 return err652 }653 if err := t.git.CommitAfterStore(ctx, e, repo); err != nil {654 return err655 }656 if !t.git.IsPending(e.ID) {657 return t.releaseGit(ctx, e.ID)658 }659 return nil660 },661 "catalog-owner": func(ctx context.Context, intent work.Intent) error {662 return t.app.catalog.UpdateOwner(ctx, t.meta.ID, t.Policy().Owner)663 },664 "records-projection": func(ctx context.Context, intent work.Intent) error {665 var payload struct {666 Action string `json:"action"`667 Room string `json:"room"`668 PubKey string `json:"pubkey"`669 Role string `json:"role"`670 Removed int `json:"removed"`671 Deleted int `json:"deleted"`672 }673 if err := json.Unmarshal([]byte(intent.Payload), &payload); err != nil {674 return err675 }676 if payload.Action == "room" {677 return t.projectRoom(ctx, payload.Room, payload.PubKey, payload.Removed != 0, payload.Deleted != 0)678 }679 var changes []records.MembershipChange680 if payload.PubKey != "" && (payload.Role != "" || payload.Removed != 0) {681 added := payload.Removed == 0682 changes = []records.MembershipChange{{PubKey: payload.PubKey, Added: &added, Roles: []string{payload.Role}}}683 }684 _, err := t.records.PublishMembershipChanges(ctx, changes, time.Now().Unix())685 return err686 },687 "site-mirror": func(ctx context.Context, intent work.Intent) error {688 return t.sites.RunMirrorIntent(ctx, storage.Intent{Kind: intent.Kind, EventID: intent.EventID, Target: intent.Target, Payload: intent.Payload})689 },690 "callback": t.ReplicationCallbackHandler(),691 }692 for kind, handler := range t.notificationHandlers() {693 handlers[kind] = handler694 }695 handlers[notificationPush] = t.handleNotificationPush696 handlers[callbackDelivery] = t.handleCallbackDelivery697 handlers[viewTransform] = t.customViews.handleViewTransform698 handlers[eventFollowupKind] = t.followups.handle699 return handlers700}701702// releaseGit publishes a repository state after Git verifies its objects. It703// removes the pending marker and persists optional planning in one transaction704// under the router's publication lock. Push planning retains the registrations705// captured at client acceptance; webhook and view matching occurs at release.706func (t *Tenant) releaseGit(ctx context.Context, id string) error {707 var followup *storage.Intent708 if t.followups != nil {709 var raw []byte710 err := t.store.DB().QueryRowContext(ctx, "SELECT raw FROM events WHERE id=?", id).Scan(&raw)711 if err != nil && !errors.Is(err, sql.ErrNoRows) {712 return err713 }714 if err == nil {715 e, err := event.Parse(raw)716 if err != nil {717 return err718 }719 followup = t.followups.prepare(ctx, e, time.Now().Unix(), true)720 }721 }722 released, err := t.router.CommitGenerated(ctx, func(ctx context.Context) (event.Event, error) {723 var promoted event.Event724 err := t.store.WithTx(ctx, func(tx *sql.Tx) error {725 result, err := tx.ExecContext(ctx, "DELETE FROM pending_events WHERE id=?", id)726 if err != nil {727 return err728 }729 n, err := result.RowsAffected()730 if err != nil || n == 0 {731 return err732 }733 if _, err := tx.ExecContext(ctx, "DELETE FROM replication_pending_until WHERE id=?", id); err != nil && !isMissingTable(err) {734 return err735 }736 var raw []byte737 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)738 if errors.Is(err, sql.ErrNoRows) {739 return nil740 }741 if err != nil {742 return err743 }744 promoted, err = event.Parse(raw)745 if err != nil || followup == nil {746 return err747 }748 return storage.AddIntents(ctx, tx, []storage.Intent{*followup}, time.Now().Unix())749 })750 return promoted, err751 })752 if err == nil && released.ID != "" {753 // A repository state is hidden until its objects arrive, so agents754 // that wait for pushes are woken here rather than at publish time.755 if t.customViews != nil && t.customViews.git != nil {756 t.followups.try(ctx, followup)757 }758 }759 return err760}761762func (t *Tenant) sweep(ctx context.Context, now int64) error {763 if _, err := t.store.SweepExpired(ctx, now); err != nil {764 return err765 }766 opts := storage.RetentionOptions{Now: now, Identity: t.records.PublicKey()}767 rules, err := t.community.RetentionRules(ctx)768 if err != nil {769 return err770 }771 for _, rule := range rules {772 kind := -1773 if rule.Kind != nil {774 kind = *rule.Kind775 }776 opts.Rules = append(opts.Rules, storage.RetentionRule{Kind: kind, Days: rule.Days})777 }778 members, err := t.community.MembersForRetention(ctx)779 if err != nil {780 return err781 }782 for _, member := range members {783 opts.Members = append(opts.Members, storage.MemberRetention{PubKey: member.PubKey, Days: member.Days})784 }785 if _, err := t.store.SweepRetention(ctx, opts); err != nil {786 return err787 }788 return t.customViews.pruneAllViewArtifacts(ctx)789}790func (t *Tenant) replacePolicy(p policy.Policy) {791 t.mu.Lock()792 previous := t.policy793 t.policy = p794 t.mu.Unlock()795 if community.RoomAccessFor(previous.Reads) != community.RoomAccessFor(p.Reads) {796 if err := t.community.EnsureSlugRoom(context.Background(), t.meta.Name, community.RoomAccessFor(p.Reads)); err != nil {797 t.app.telemetry.Logger().Error("refresh main room", "tenant", t.meta.Name, "error", err)798 }799 }800 if err := t.reconcileAutomaticInbox(context.Background(), previous, p); err != nil {801 t.app.telemetry.Logger().Error("reconcile automatic inbox job", "tenant", t.meta.Name, "error", err)802 }803 t.router.CloseSubscriptions("blocked: relay policy changed; subscribe again")804 t.scheduleViews()805}806func (t *Tenant) generatedRecord(ctx context.Context, e event.Event) error {807 ctx, done, admissionErr := t.beginOperation(ctx)808 if admissionErr != nil {809 return admissionErr810 }811 defer done()812 _, err := t.router.CommitGenerated(ctx, func(ctx context.Context) (event.Event, error) {813 _, err := t.store.Save(ctx, e, storage.SaveOptions{Now: e.CreatedAt, SearchMode: t.Policy().Features.Search})814 return e, err815 })816 return err817}818func (t *Tenant) ingest(ctx context.Context, e event.Event, _ replication.Origin) error {819 ctx, done, admissionErr := t.beginOperation(ctx)820 if admissionErr != nil {821 return admissionErr822 }823 defer done()824 _, err := t.router.CommitGenerated(ctx, func(ctx context.Context) (event.Event, error) {825 return e, t.commitImported(ctx, e, replication.OriginImport)826 })827 if errors.Is(err, storage.ErrDuplicate) || errors.Is(err, storage.ErrReplaced) {828 return nil829 }830 return err831}832833// commitImported applies import admission and commits the shared projections834// and optional planning for an event. It returns persistence errors to the835// caller, which decides how to handle duplicate and superseded imports.836// origin supplies the replication source for planning.837func (t *Tenant) commitImported(ctx context.Context, e event.Event, origin replication.Origin) error {838 if err := t.gate.Import(ctx, e, time.Now().Unix()); err != nil {839 return err840 }841 if err := t.validateStoredEvent(ctx, e); err != nil {842 return err843 }844 plan, err := t.prepareStoredEvent(ctx, e, origin, time.Now().Unix())845 if err != nil {846 return err847 }848 _, err = t.store.Save(ctx, e, plan.options)849 if err == nil {850 t.afterStoredEvent(ctx, e, plan.followups)851 }852 return err853}