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, SetPolicy: func(next policy.Policy) error { return t.applyPolicy(context.Background(), next) }})145 if err != nil {146 return err147 }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 err156 }157 currentPolicy := t.Policy()158 if err := t.reconcileAutomaticInbox(ctx, currentPolicy, currentPolicy); err != nil {159 return err160 }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 err164 }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 err171 }172 t.customViews.git = t.git173 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 err176 }177 if err := t.initMCP(); err != nil {178 return err179 }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 = err189 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 nil199}200201func (t *Tenant) memberBlobReader(ctx context.Context, pubkey, role string) bool {202 if role == "owner" || role == "moderator" || role == "member" {203 return true204 }205 if role != "agent" {206 return false207 }208 _, err := t.agentUploadGrant(ctx, pubkey)209 return err == nil210}211212func (t *Tenant) fileReadAccess(ctx context.Context, actor, hash string) error {213 access, err := t.blobs.BlobAccess(ctx, hash)214 if err != nil {215 return err216 }217 if access != blob.AccessMembers {218 return nil219 }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 nil238}239240// backend adapts the management method signature to webui's form-oriented API.241type backend struct{ tenant *Tenant }242243type localDirectory struct{ store *storage.Store }244245func (d localDirectory) WriteRelays(pubkey string) []string {246 return d.relaysForDirection(pubkey, true)247}248func (d localDirectory) ReadRelays(pubkey string) []string {249 return d.relaysForDirection(pubkey, false)250}251func (d localDirectory) RelayList(pubkey string) []string { return d.relays(pubkey, false, true) }252func (d localDirectory) relaysForDirection(pubkey string, write bool) []string {253 return d.relays(pubkey, write, false)254}255func (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 nil259 }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 continue264 }265 result = append(result, tag[1])266 }267 return result268}269270func (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}273func (b backend) Policy() policy.Policy { return b.tenant.Policy() }274func (b backend) URL() string { return b.tenant.publicURL }275func (b backend) Slug() string { return b.tenant.meta.Name }276func (b backend) Identity() string { return b.tenant.records.PublicKey() }277278// CustomViews gives the page renderers the enabled custom views' names and279// languages, never a transform or a secret.280func (b backend) CustomViews() []views.View { return b.tenant.customViews.CustomViews() }281282// ReadAllowed is consumed by the web UI private boundary. It deliberately283// delegates to the same uncached membership check used by private Git and284// NIP-42, so a revoked browser session cannot retain page access.285func (b backend) ReadAllowed(ctx context.Context, actor string) error {286 if !b.tenant.PrivateServiceEnabled() {287 return nil288 }289 return b.tenant.requirePrivateAccess(ctx, actor)290}291292func (b backend) SharePresence(ctx context.Context, actor string) bool {293 return b.tenant.chatSharePresence(ctx, actor)294}295296// closeServices cancels and joins the tenant's worker, scheduler and Git297// loops, then returns the worker's terminal queue error, if any. Store and298// relay connection cleanup belong to Tenant.Close.299func (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.workErr307}308309func 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}319func (t *Tenant) siteReadAccess(r *http.Request) bool {320 actor, err := t.resolveUIActor(r)321 if err != nil {322 return false323 }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 == nil330}331func (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 "", bodyErr345 }346 pubkey := ""347 var err error348 if action == blob.ActionUpload && r.Method == http.MethodPost && r.URL.Path == "/nip96" {349 var token event.Event350 token, err = t.auth.VerifyNIP98(r.Header.Get("Authorization"), t.requestURL(r), r.Method, body)351 pubkey = token.PubKey352 } 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 fetched357 // bytes are the resulting blob and are scoped by x below.358 blossomAction = string(blob.ActionUpload)359 }360 blobHash := hash361 if blobHash == "" && action == blob.ActionUpload {362 blobHash = strings.ToLower(strings.TrimSpace(r.Header.Get("x-sha-256")))363 }364 var token event.Event365 token, err = t.auth.VerifyBlossomRequest(r.Header.Get("Authorization"), blossomAction, t.siteDomain(), blobHash)366 pubkey = token.PubKey367 } else {368 if isBUD13RemoteUpload(r) {369 var token event.Event370 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.PubKey375 } else if action == blob.ActionUpload && (r.Method == http.MethodPut || r.Method == http.MethodPatch) {376 var token event.Event377 token, err = t.auth.VerifyNIP98Deferred(r.Header.Get("Authorization"), t.requestURL(r), r.Method)378 pubkey = token.PubKey379 } else {380 var keys []string381 keys, err = t.auth.WhoAsks(r.Header.Get("Authorization"), t.requestURL(r), r.Method, body, string(action), hash)382 pubkey = first(keys)383 }384 }385 }386 if err != nil {387 return "", err388 }389 if action == blob.ActionGet {390 if err := t.roomAttachmentAccess(r.Context(), pubkey, hash); err != nil {391 return "", err392 }393 }394 if pubkey == "" {395 if action == blob.ActionGet {396 return "", nil397 }398 return "", errors.New("auth-required: blob authorization required")399 }400 if banned, banErr := t.community.IsBanned(r.Context(), pubkey); banErr != nil {401 return "", fmt.Errorf("check blob uploader ban: %w", banErr)402 } else if banned {403 return "", errors.New("blocked: this pubkey is banned from this relay")404 }405 if action == blob.ActionUpload || action == blob.ActionMirror {406 if !t.Policy().Features.Files {407 return "", errors.New("restricted: file uploads are disabled")408 }409 role, roleErr := t.community.Role(r.Context(), pubkey)410 if roleErr != nil {411 return "", fmt.Errorf("check blob uploader role: %w", roleErr)412 }413 allowed := t.Policy().Writes == "open" || (t.Policy().Writes == "owner" && role == "owner") || (t.Policy().Writes != "owner" && role != "")414 if !allowed {415 return "", errors.New("restricted: file upload is not allowed")416 }417 // An agent's standing is its grant: an active one makes it a member418 // for uploads, a paused, revoked or expired one stops them.419 if role == "agent" {420 if _, err := t.agentUploadGrant(r.Context(), pubkey); err != nil {421 return "", err422 }423 }424 }425 return pubkey, nil426}427428// agentUploadGrant returns the grant behind an agent key's uploads when it429// currently allows publishing.430func (t *Tenant) agentUploadGrant(ctx context.Context, pubkey string) (community.AgentGrant, error) {431 grant, ok, err := t.community.AgentGrant(ctx, pubkey)432 if err != nil {433 return community.AgentGrant{}, fmt.Errorf("check agent upload grant: %w", err)434 }435 if !ok {436 return community.AgentGrant{}, errors.New("restricted: agent grant is missing")437 }438 switch {439 case grant.RevokedAt > 0:440 return community.AgentGrant{}, errors.New("restricted: agent grant is revoked")441 case grant.Paused:442 return community.AgentGrant{}, errors.New("restricted: agent grant is paused")443 case grant.ExpiresAt <= time.Now().Unix():444 return community.AgentGrant{}, errors.New("restricted: agent grant has expired")445 }446 return grant, nil447}448449// blobUploadTerms is what the blob service asks before it stores an upload:450// a key whose only standing is an agent grant takes the grant's sites ttl as451// the expiry of its claims and its encrypted flag as a requirement. Keys452// with a human role, and agents without a sites tag, upload on no terms.453func (t *Tenant) blobUploadTerms(ctx context.Context, pubkey string) (blob.UploadTerms, error) {454 role, err := t.community.Role(ctx, pubkey)455 if err != nil {456 return blob.UploadTerms{}, fmt.Errorf("check uploader role: %w", err)457 }458 if role != "agent" {459 return blob.UploadTerms{}, nil460 }461 grant, err := t.agentUploadGrant(ctx, pubkey)462 if err != nil {463 return blob.UploadTerms{}, err464 }465 terms := blob.UploadTerms{Encrypted: grant.UploadsEncrypted()}466 if days := grant.UploadTTLDays(); days > 0 {467 terms.ExpiresAt = time.Now().Unix() + int64(days)*86400468 }469 return terms, nil470}471472// sweepExpiredBlobs retires uploads whose agent ttl has lapsed. The step is473// recorded as the blob-expiry storage operation with a bounded outcome and474// a count, never a hash or a key.475func (t *Tenant) sweepExpiredBlobs(ctx context.Context, now int64) error {476 if t.blobs == nil {477 return nil478 }479 started := time.Now()480 removed, err := t.blobs.SweepExpired(ctx, now)481 outcome := "ok"482 if err != nil {483 outcome = "error"484 }485 t.app.telemetry.ObserveStorage("blob-expiry", outcome, time.Since(started))486 if removed > 0 {487 t.app.telemetry.Logger().Info("expired uploads removed", "tenant", t.meta.Name, "blobs", removed)488 }489 return err490}491492func (t *Tenant) validateBlobUpload(r *http.Request, hash string) error {493 // PATCH carries only one chunk. Blossom scopes the final path hash, while494 // NIP-98 binds the bytes in this individual request body.495 if r.Method == http.MethodPatch && auth.IsBlossomAuthorization(r.Header.Get("Authorization")) {496 finalHash := strings.Split(strings.TrimPrefix(r.URL.Path, "/"), ".")[0]497 return t.auth.ValidateBlossomPayload(r.Header.Get("Authorization"), string(blob.ActionUpload), finalHash)498 }499 // Mirror proofs bind the HTTP request body before fetching. Blossom500 // authorization additionally names the fetched content through its x tag.501 if r.URL.Path == "/mirror" || isBUD13RemoteUpload(r) {502 if auth.IsBlossomAuthorization(r.Header.Get("Authorization")) {503 return t.auth.ValidateBlossomPayload(r.Header.Get("Authorization"), string(blob.ActionUpload), hash)504 }505 return nil506 }507 return t.auth.ValidateBlobPayload(r.Header.Get("Authorization"), hash)508}509510// NIP-98 binds its payload to the complete HTTP body. The streaming Blossom511// upload remains unbuffered and is checked against its resulting SHA-256 by512// ValidateUpload; the JSON and multipart doors are buffered only so their513// signed request can be verified before parsing or fetching any content.514func blobAuthBody(r *http.Request, action blob.Action) (string, error) {515 buffer := action == blob.ActionMirror || action == blob.ActionReport ||516 (action == blob.ActionUpload && r.Method == http.MethodPost && (r.URL.Path == "/nip96" || r.URL.Path == "/files/metadata"))517 if !buffer {518 return "", nil519 }520 var source io.Reader = r.Body521 if r.URL.Path == "/files/metadata" {522 source = io.LimitReader(r.Body, 16*1024+1)523 }524 body, err := io.ReadAll(source)525 if err != nil {526 return "", fmt.Errorf("read blob authorization body: %w", err)527 }528 if r.URL.Path == "/files/metadata" && len(body) > 16*1024 {529 return "", errors.New("invalid: file metadata exceeds 16 KiB")530 }531 r.Body = io.NopCloser(bytes.NewReader(body))532 return string(body), nil533}534535func (t *Tenant) siteDomain() string {536 if t.meta.Name != t.app.cfg.DefaultTenant {537 return t.app.siteBase(t.meta)538 }539 host := hostName(t.publicURL)540 if ip := net.ParseIP(host); ip != nil && ip.IsLoopback() {541 return "localhost"542 }543 return host544}545func (t *Tenant) authorizeGit(ctx context.Context, r *http.Request, repo gitrelay.Repository) error {546 if !t.Policy().Features.Grasp {547 return errors.New("restricted: Git hosting is disabled")548 }549 // Authenticate before reading a potentially large pack. The body binding550 // is checked below before any bytes reach git-receive-pack.551 e, ok := privateGitProof(r)552 var err error553 if !ok {554 e, err = t.auth.VerifyNIP98Deferred(r.Header.Get("Authorization"), t.requestURL(r), r.Method)555 if err != nil {556 return err557 }558 }559 if banned, banErr := t.community.IsBanned(ctx, e.PubKey); banErr != nil {560 return fmt.Errorf("check Git author ban: %w", banErr)561 } else if banned {562 return errors.New("blocked: this pubkey is banned from this relay")563 }564 if privatePolicy(t.Policy()) {565 // Tenant membership is the private-service boundary. A repository566 // maintainer or owner who is not a member must not learn whether a567 // private repository exists through Smart HTTP.568 if err := t.requirePrivateAccess(ctx, e.PubKey); err != nil {569 return err570 }571 } else if !t.IsMaintainer(ctx, repo, e.PubKey) {572 role, roleErr := t.community.Role(ctx, e.PubKey)573 if roleErr != nil || (role != "owner" && role != "moderator" && role != "member") {574 return errors.New("restricted: repository authorization required")575 }576 }577 if _, checked := r.Context().Value(privateGitPayloadCheckedKey{}).(struct{}); checked {578 return nil579 }580 return spoolGitPayload(r, e)581}582func first(v []string) string {583 if len(v) == 0 {584 return ""585 }586 return v[0]587}588589func isBUD13RemoteUpload(r *http.Request) bool {590 if r == nil || r.Method != http.MethodPut || r.URL.Query().Get("url") == "" {591 return false592 }593 path := strings.TrimPrefix(r.URL.Path, "/")594 hash := strings.Split(path, ".")[0]595 return len(hash) == 64 && strings.Trim(hash, "0123456789abcdefABCDEF") == ""596}597598func emptyPayloadHash() string {599 sum := sha256.Sum256(nil)600 return hex.EncodeToString(sum[:])601}602603func (t *Tenant) resolveUIActor(r *http.Request) (string, error) {604 if r.Header.Get("Authorization") == "" {605 if r.Method != http.MethodGet && r.Method != http.MethodHead {606 return "", errors.New("auth-required: sign this request")607 }608 return t.cookieActor(r)609 }610 if r.Body == nil {611 r.Body = http.NoBody612 }613 body, err := io.ReadAll(r.Body)614 if err != nil {615 return "", err616 }617 r.Body = io.NopCloser(bytes.NewReader(body))618 e, err := t.auth.VerifyNIP98(r.Header.Get("Authorization"), t.requestURL(r), r.Method, string(body))619 if err != nil {620 return "", err621 }622 return e.PubKey, nil623}624625func (t *Tenant) workHandlers() map[string]work.Handler {626 handlers := map[string]work.Handler{627 "view-publish": func(ctx context.Context, intent work.Intent) error {628 if err := t.records.MarkView(ctx, intent.Target, time.Now().Unix()); err != nil {629 return err630 }631 t.scheduleViews()632 return nil633 },634 "git-metadata": func(ctx context.Context, intent work.Intent) error {635 var raw string636 err := t.store.DB().QueryRowContext(ctx, "SELECT raw FROM events WHERE id=?", intent.EventID).Scan(&raw)637 if errors.Is(err, sql.ErrNoRows) {638 return nil639 }640 if err != nil {641 return err642 }643 e, err := event.Parse([]byte(raw))644 if err != nil {645 return err646 }647 var repo gitrelay.Repository648 if err := json.Unmarshal([]byte(intent.Payload), &repo); err != nil {649 return err650 }651 if err := t.git.CommitAfterStore(ctx, e, repo); err != nil {652 return err653 }654 if !t.git.IsPending(e.ID) {655 return t.releaseGit(ctx, e.ID)656 }657 return nil658 },659 "catalog-owner": func(ctx context.Context, intent work.Intent) error {660 return t.app.catalog.UpdateOwner(ctx, t.meta.ID, t.Policy().Owner)661 },662 "records-projection": func(ctx context.Context, intent work.Intent) error {663 var payload struct {664 Action string `json:"action"`665 Room string `json:"room"`666 PubKey string `json:"pubkey"`667 Role string `json:"role"`668 Removed int `json:"removed"`669 Deleted int `json:"deleted"`670 }671 if err := json.Unmarshal([]byte(intent.Payload), &payload); err != nil {672 return err673 }674 if payload.Action == "room" {675 return t.projectRoom(ctx, payload.Room, payload.PubKey, payload.Removed != 0, payload.Deleted != 0)676 }677 var changes []records.MembershipChange678 if payload.PubKey != "" && (payload.Role != "" || payload.Removed != 0) {679 added := payload.Removed == 0680 changes = []records.MembershipChange{{PubKey: payload.PubKey, Added: &added, Roles: []string{payload.Role}}}681 }682 _, err := t.records.PublishMembershipChanges(ctx, changes, time.Now().Unix())683 return err684 },685 "site-mirror": func(ctx context.Context, intent work.Intent) error {686 return t.sites.RunMirrorIntent(ctx, storage.Intent{Kind: intent.Kind, EventID: intent.EventID, Target: intent.Target, Payload: intent.Payload})687 },688 "callback": t.ReplicationCallbackHandler(),689 }690 for kind, handler := range t.notificationHandlers() {691 handlers[kind] = handler692 }693 handlers[notificationPush] = t.handleNotificationPush694 handlers[callbackDelivery] = t.handleCallbackDelivery695 handlers[viewTransform] = t.customViews.handleViewTransform696 handlers[eventFollowupKind] = t.followups.handle697 return handlers698}699700// releaseGit publishes a repository state after Git verifies its objects. It701// removes the pending marker and persists optional planning in one transaction702// under the router's publication lock. Push planning retains the registrations703// captured at client acceptance; webhook and view matching occurs at release.704func (t *Tenant) releaseGit(ctx context.Context, id string) error {705 var followup *storage.Intent706 if t.followups != nil {707 var raw []byte708 err := t.store.DB().QueryRowContext(ctx, "SELECT raw FROM events WHERE id=?", id).Scan(&raw)709 if err != nil && !errors.Is(err, sql.ErrNoRows) {710 return err711 }712 if err == nil {713 e, err := event.Parse(raw)714 if err != nil {715 return err716 }717 followup = t.followups.prepare(ctx, e, time.Now().Unix(), true)718 }719 }720 released, err := t.router.CommitGenerated(ctx, func(ctx context.Context) (event.Event, error) {721 var promoted event.Event722 err := t.store.WithTx(ctx, func(tx *sql.Tx) error {723 result, err := tx.ExecContext(ctx, "DELETE FROM pending_events WHERE id=?", id)724 if err != nil {725 return err726 }727 n, err := result.RowsAffected()728 if err != nil || n == 0 {729 return err730 }731 if _, err := tx.ExecContext(ctx, "DELETE FROM replication_pending_until WHERE id=?", id); err != nil && !isMissingTable(err) {732 return err733 }734 var raw []byte735 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)736 if errors.Is(err, sql.ErrNoRows) {737 return nil738 }739 if err != nil {740 return err741 }742 promoted, err = event.Parse(raw)743 if err != nil || followup == nil {744 return err745 }746 return storage.AddIntents(ctx, tx, []storage.Intent{*followup}, time.Now().Unix())747 })748 return promoted, err749 })750 if err == nil && released.ID != "" {751 // A repository state is hidden until its objects arrive, so agents752 // that wait for pushes are woken here rather than at publish time.753 if t.customViews != nil && t.customViews.git != nil {754 t.followups.try(ctx, followup)755 }756 }757 return err758}759760func (t *Tenant) sweep(ctx context.Context, now int64) error {761 if _, err := t.store.SweepExpired(ctx, now); err != nil {762 return err763 }764 opts := storage.RetentionOptions{Now: now, Identity: t.records.PublicKey()}765 rules, err := t.community.RetentionRules(ctx)766 if err != nil {767 return err768 }769 for _, rule := range rules {770 kind := -1771 if rule.Kind != nil {772 kind = *rule.Kind773 }774 opts.Rules = append(opts.Rules, storage.RetentionRule{Kind: kind, Days: rule.Days})775 }776 members, err := t.community.MembersForRetention(ctx)777 if err != nil {778 return err779 }780 for _, member := range members {781 opts.Members = append(opts.Members, storage.MemberRetention{PubKey: member.PubKey, Days: member.Days})782 }783 if _, err := t.store.SweepRetention(ctx, opts); err != nil {784 return err785 }786 return t.customViews.pruneAllViewArtifacts(ctx)787}788func (t *Tenant) replacePolicy(p policy.Policy) {789 t.mu.Lock()790 previous := t.policy791 t.policy = p792 t.mu.Unlock()793 if community.RoomAccessFor(previous.Reads) != community.RoomAccessFor(p.Reads) {794 if err := t.community.EnsureSlugRoom(context.Background(), t.meta.Name, community.RoomAccessFor(p.Reads)); err != nil {795 t.app.telemetry.Logger().Error("refresh main room", "tenant", t.meta.Name, "error", err)796 }797 }798 if err := t.reconcileAutomaticInbox(context.Background(), previous, p); err != nil {799 t.app.telemetry.Logger().Error("reconcile automatic inbox job", "tenant", t.meta.Name, "error", err)800 }801 t.router.CloseSubscriptions("blocked: relay policy changed; subscribe again")802 t.scheduleViews()803}804func (t *Tenant) generatedRecord(ctx context.Context, e event.Event) error {805 ctx, done, admissionErr := t.beginOperation(ctx)806 if admissionErr != nil {807 return admissionErr808 }809 defer done()810 _, err := t.router.CommitGenerated(ctx, func(ctx context.Context) (event.Event, error) {811 _, err := t.store.Save(ctx, e, storage.SaveOptions{Now: e.CreatedAt, SearchMode: t.Policy().Features.Search})812 return e, err813 })814 return err815}816func (t *Tenant) ingest(ctx context.Context, e event.Event, _ replication.Origin) error {817 ctx, done, admissionErr := t.beginOperation(ctx)818 if admissionErr != nil {819 return admissionErr820 }821 defer done()822 _, err := t.router.CommitGenerated(ctx, func(ctx context.Context) (event.Event, error) {823 return e, t.commitImported(ctx, e, replication.OriginImport)824 })825 if errors.Is(err, storage.ErrDuplicate) || errors.Is(err, storage.ErrReplaced) {826 return nil827 }828 return err829}830831// commitImported applies import admission and commits the shared projections832// and optional planning for an event. It returns persistence errors to the833// caller, which decides how to handle duplicate and superseded imports.834// origin supplies the replication source for planning.835func (t *Tenant) commitImported(ctx context.Context, e event.Event, origin replication.Origin) error {836 if err := t.gate.Import(ctx, e, time.Now().Unix()); err != nil {837 return err838 }839 if err := t.validateStoredEvent(ctx, e); err != nil {840 return err841 }842 plan, err := t.prepareStoredEvent(ctx, e, origin, time.Now().Unix())843 if err != nil {844 return err845 }846 _, err = t.store.Save(ctx, e, plan.options)847 if err == nil {848 t.afterStoredEvent(ctx, e, plan.followups)849 }850 return err851}