1package daemon23// This file contains the concrete adapters that keep replication tenant4// scoped. The replication package stays independent from daemon ownership,5// filesystem layout, and community implementation details.67import (8 "bytes"9 "context"10 "crypto/sha256"11 "database/sql"12 "encoding/base64"13 "encoding/hex"14 "encoding/json"15 "errors"16 "fmt"17 "os"18 "path/filepath"19 "strconv"20 "strings"21 "time"2223 "github.com/FelineStateMachine/tinyrelay/internal/configport"24 "github.com/FelineStateMachine/tinyrelay/internal/event"25 "github.com/FelineStateMachine/tinyrelay/internal/policy"26 "github.com/FelineStateMachine/tinyrelay/internal/records"27 "github.com/FelineStateMachine/tinyrelay/internal/relay"28 "github.com/FelineStateMachine/tinyrelay/internal/replication"29 "github.com/FelineStateMachine/tinyrelay/internal/storage"30 "github.com/FelineStateMachine/tinyrelay/internal/work"31 gitrelay "github.com/FelineStateMachine/tinyrelay/tinygit"32)3334type tenantReplicationProvider struct{ tenant *Tenant }3536func (t *Tenant) ReplicationBackupProvider() replication.BackupProvider {37 return tenantReplicationProvider{tenant: t}38}3940func (p tenantReplicationProvider) Snapshot(ctx context.Context) (replication.BackupState, error) {41 if p.tenant == nil {42 return replication.BackupState{}, errors.New("daemon: nil tenant")43 }44 ctx, done, admissionErr := p.tenant.beginMaintenance(ctx)45 if admissionErr != nil {46 return replication.BackupState{}, admissionErr47 }48 defer done()49 state := replication.BackupState{}50 policy, err := json.Marshal(p.tenant.Policy())51 if err != nil {52 return state, err53 }54 state.Config = policy55 pubkey := ""56 if p.tenant.records != nil {57 pubkey = p.tenant.records.PublicKey()58 }59 identity, err := json.Marshal(map[string]string{"pubkey": pubkey, "relay": p.tenant.RelayURL()})60 if err != nil {61 return state, err62 }63 state.Identity = identity64 state.Database, err = snapshotTables(ctx, p.tenant.store)65 if err != nil {66 return state, err67 }68 state.Blobs, err = snapshotFiles(filepath.Join(p.tenant.meta.Paths.Root, "blobs"))69 if err != nil {70 return state, err71 }72 state.Git, err = snapshotFiles(p.tenant.meta.Paths.Git)73 if err != nil {74 return state, err75 }76 return state, nil77}7879func (p tenantReplicationProvider) SnapshotArchive(ctx context.Context, now int64) (replication.BackupArchive, error) {80 state, err := p.Snapshot(ctx)81 if err != nil {82 return replication.BackupArchive{}, err83 }84 archive := replication.BackupArchive{Format: replication.BackupFormat, Created: now, State: state}85 for _, table := range state.Database {86 if table.Name != "events" {87 continue88 }89 for _, row := range table.Rows {90 for i, column := range table.Columns {91 if column != "raw" || i >= len(row) {92 continue93 }94 var raw string95 if err := json.Unmarshal(row[i], &raw); err != nil {96 return replication.BackupArchive{}, err97 }98 var item event.Event99 if err := json.Unmarshal([]byte(raw), &item); err != nil {100 return replication.BackupArchive{}, err101 }102 archive.Events = append(archive.Events, item)103 }104 }105 }106 return replication.SealBackup(archive)107}108109func (p tenantReplicationProvider) Restore(ctx context.Context, state replication.BackupState) error {110 if p.tenant == nil {111 return errors.New("daemon: nil tenant")112 }113 ctx, done, admissionErr := p.tenant.beginMaintenance(ctx)114 if admissionErr != nil {115 return admissionErr116 }117 defer done()118 oldBlobs, err := snapshotFiles(filepath.Join(p.tenant.meta.Paths.Root, "blobs"))119 if err != nil {120 return err121 }122 oldGit, err := snapshotFiles(p.tenant.meta.Paths.Git)123 if err != nil {124 return err125 }126 oldTables, err := snapshotTables(ctx, p.tenant.store)127 if err != nil {128 return err129 }130 restored := false131 defer func() {132 if restored {133 return134 }135 // Restore is a single maintenance operation. Any failure after the136 // first write must leave both SQLite and tenant files as they were.137 _ = restoreTables(context.Background(), p.tenant.store, oldTables)138 _ = restoreFiles(filepath.Join(p.tenant.meta.Paths.Root, "blobs"), oldBlobs)139 _ = restoreFiles(p.tenant.meta.Paths.Git, oldGit)140 if p.tenant.git != nil {141 _ = p.tenant.git.Reload(context.WithoutCancel(ctx))142 }143 }()144 if err := restoreTables(ctx, p.tenant.store, state.Database); err != nil {145 return err146 }147 if err := restoreFiles(filepath.Join(p.tenant.meta.Paths.Root, "blobs"), state.Blobs); err != nil {148 _ = restoreTables(ctx, p.tenant.store, oldTables)149 _ = restoreFiles(filepath.Join(p.tenant.meta.Paths.Root, "blobs"), oldBlobs)150 return err151 }152 if err := restoreFiles(p.tenant.meta.Paths.Git, state.Git); err != nil {153 _ = restoreTables(ctx, p.tenant.store, oldTables)154 _ = restoreFiles(filepath.Join(p.tenant.meta.Paths.Root, "blobs"), oldBlobs)155 _ = restoreFiles(p.tenant.meta.Paths.Git, oldGit)156 return err157 }158 if len(state.GitSQL) > 0 && string(state.GitSQL) != "null" {159 if p.tenant.git == nil {160 _ = restoreTables(ctx, p.tenant.store, oldTables)161 _ = restoreFiles(filepath.Join(p.tenant.meta.Paths.Root, "blobs"), oldBlobs)162 _ = restoreFiles(filepath.Join(p.tenant.meta.Paths.Git), oldGit)163 return errors.New("daemon: Git backup migration unavailable")164 }165 if _, err := gitrelay.ValidateGitBackup(state.GitSQL); err != nil {166 _ = restoreTables(ctx, p.tenant.store, oldTables)167 _ = restoreFiles(filepath.Join(p.tenant.meta.Paths.Root, "blobs"), oldBlobs)168 _ = restoreFiles(filepath.Join(p.tenant.meta.Paths.Git), oldGit)169 return fmt.Errorf("daemon: validate legacy Git backup: %w", err)170 }171 if err := p.tenant.git.ImportGitBackup(ctx, state.GitSQL); err != nil {172 _ = restoreTables(ctx, p.tenant.store, oldTables)173 _ = restoreFiles(filepath.Join(p.tenant.meta.Paths.Root, "blobs"), oldBlobs)174 _ = restoreFiles(filepath.Join(p.tenant.meta.Paths.Git), oldGit)175 return fmt.Errorf("daemon: import legacy Git backup: %w", err)176 }177 }178 if len(state.Config) > 0 {179 var envelope struct {180 Format string `json:"format"`181 }182 if json.Unmarshal(state.Config, &envelope) == nil && strings.HasPrefix(envelope.Format, "bind.ws/relay-config/") {183 legacyConfig := true184 if p.tenant.config == nil {185 return errors.New("daemon: config importer unavailable for legacy backup")186 }187 parsed, err := configport.Parse(state.Config)188 if err != nil {189 return fmt.Errorf("daemon: legacy config: %w", err)190 }191 var identity struct {192 Owner string `json:"owner"`193 }194 if len(state.Identity) > 0 {195 if err := json.Unmarshal(state.Identity, &identity); err != nil {196 return fmt.Errorf("daemon: legacy identity: %w", err)197 }198 }199 if _, err := p.tenant.config.ApplyWithOptions(ctx, parsed, configport.ApplyOptions{MigrationOwner: identity.Owner}); err != nil {200 return fmt.Errorf("daemon: legacy config apply: %w", err)201 }202 fresh, err := records.New(ctx, records.Config{Store: p.tenant.store, Community: p.tenant.community, Policy: p.tenant.Policy, RelayURL: p.tenant.RelayURL(), GroupID: p.tenant.meta.Name, OnGenerated: p.tenant.generatedRecord, DeliverNotification: p.tenant.deliverNotification, PushNotification: p.tenant.enqueuePush, EventVisible: func(ctx context.Context, e event.Event, a policy.Access) bool {203 return p.tenant.gate.CanSee(ctx, e, relay.Session{PubKeys: a.PubKeys, RelayURL: p.tenant.RelayURL()}, nil)204 }, SetPolicy: func(next policy.Policy) error { return p.tenant.applyPolicy(context.Background(), next) }})205 if err != nil {206 return fmt.Errorf("daemon: refresh records after legacy config: %w", err)207 }208 if legacyConfig {209 p.tenant.records = fresh210 if p.tenant.customViews != nil {211 p.tenant.customViews.records = fresh212 }213 }214 } else {215 var next policy.Policy216 if err := json.Unmarshal(state.Config, &next); err != nil {217 return err218 }219 if next.Owner == "" {220 var envelope struct {221 Policy policy.Policy `json:"policy"`222 }223 if err := json.Unmarshal(state.Config, &envelope); err != nil {224 return err225 }226 next = envelope.Policy227 if next.Owner == "" {228 next.Owner = p.tenant.Policy().Owner229 }230 }231 p.tenant.community.SetOwner(next.Owner)232 if p.tenant.router != nil {233 p.tenant.replacePolicy(next)234 } else {235 p.tenant.mu.Lock()236 p.tenant.policy = next237 p.tenant.mu.Unlock()238 }239 fresh, err := records.New(ctx, records.Config{Store: p.tenant.store, Community: p.tenant.community, Policy: p.tenant.Policy, RelayURL: p.tenant.RelayURL(), GroupID: p.tenant.meta.Name, OnGenerated: p.tenant.generatedRecord, DeliverNotification: p.tenant.deliverNotification, PushNotification: p.tenant.enqueuePush, EventVisible: func(ctx context.Context, e event.Event, a policy.Access) bool {240 return p.tenant.gate.CanSee(ctx, e, relay.Session{PubKeys: a.PubKeys, RelayURL: p.tenant.RelayURL()}, nil)241 }, SetPolicy: func(next policy.Policy) error { return p.tenant.applyPolicy(context.Background(), next) }})242 if err != nil {243 return fmt.Errorf("daemon: refresh records identity: %w", err)244 }245 p.tenant.records = fresh246 if p.tenant.customViews != nil {247 p.tenant.customViews.records = fresh248 }249 }250 }251 if len(state.Blobs) > 0 {252 for _, object := range state.Blobs {253 if object.Type == "" {254 continue255 }256 if _, err := p.tenant.store.DB().ExecContext(ctx, `INSERT OR REPLACE INTO blobs(sha256,size,type,uploader,uploaded) VALUES(?,?,?,?,?)`, object.SHA256, len(object.Content), object.Type, object.Uploader, object.Uploaded); err != nil {257 return err258 }259 }260 }261 if len(state.Community) > 0 {262 if err := restoreRetainedState(ctx, p.tenant.store, state.Community); err != nil {263 return fmt.Errorf("daemon: restore retained state: %w", err)264 }265 }266 if p.tenant.git != nil {267 if err := p.tenant.git.Reload(ctx); err != nil {268 return fmt.Errorf("daemon: reload Git after restore: %w", err)269 }270 }271 restored = true272 return nil273}274275func (p tenantReplicationProvider) FinalizeRestore(ctx context.Context) error {276 if p.tenant == nil {277 return errors.New("daemon: nil tenant")278 }279 if err := p.tenant.store.WithTx(ctx, func(tx *sql.Tx) error {280 now := time.Now().Unix()281 for _, statement := range []string{282 `CREATE TABLE IF NOT EXISTS replication_pending_until(id TEXT PRIMARY KEY NOT NULL, until INTEGER NOT NULL)`,283 `CREATE TABLE IF NOT EXISTS replication_pr_refs(repo TEXT NOT NULL, ref TEXT NOT NULL, until INTEGER NOT NULL, PRIMARY KEY(repo,ref))`,284 } {285 if _, err := tx.ExecContext(ctx, statement); err != nil {286 return err287 }288 }289 if _, err := tx.ExecContext(ctx, `DELETE FROM pending_events WHERE id IN (SELECT id FROM replication_pending_until WHERE until<=?)`, now); err != nil {290 return err291 }292 if _, err := tx.ExecContext(ctx, `DELETE FROM replication_pending_until WHERE until<=?`, now); err != nil {293 return err294 }295 if _, err := tx.ExecContext(ctx, `DELETE FROM replication_pr_refs WHERE until>0 AND until<=?`, now); err != nil {296 return err297 }298 _, err := tx.ExecContext(ctx, `INSERT OR IGNORE INTO pending_events(id,reason) SELECT p.id,'backup' FROM replication_pending_until p JOIN events e ON e.id=p.id WHERE p.until>?`, now)299 return err300 }); err != nil {301 return err302 }303 if p.tenant.git != nil {304 return p.tenant.git.Reload(ctx)305 }306 return nil307}308309// restoreRetainedState keeps the bind.ws relay's auxiliary state even when310// the current schema has no first-class feature for one of those records. The311// small native tables are deliberately explicit: a restore must either write312// every record or return an error, never silently discard a family.313func restoreRetainedState(ctx context.Context, store *storage.Store, rawState []byte) error {314 var retained struct {315 ListHistory []string `json:"listHistory"`316 Hidden []string `json:"hidden"`317 Hosted []string `json:"hosted"`318 Pending []struct {319 ID string `json:"id"`320 Until int64 `json:"until"`321 } `json:"pending"`322 PRRefs []struct {323 Repo string `json:"repo"`324 Ref string `json:"ref"`325 Until int64 `json:"until"`326 } `json:"prRefs"`327 }328 if err := json.Unmarshal(rawState, &retained); err != nil {329 return fmt.Errorf("invalid state: %w", err)330 }331 now := time.Now().Unix()332 return store.WithTx(ctx, func(tx *sql.Tx) error {333 if _, err := tx.ExecContext(ctx, `CREATE TABLE IF NOT EXISTS replication_hosted_events(id TEXT PRIMARY KEY NOT NULL)`); err != nil {334 return err335 }336 if _, err := tx.ExecContext(ctx, `CREATE TABLE IF NOT EXISTS replication_pending_until(id TEXT PRIMARY KEY NOT NULL, until INTEGER NOT NULL)`); err != nil {337 return err338 }339 if _, err := tx.ExecContext(ctx, `CREATE TABLE IF NOT EXISTS replication_pr_refs(repo TEXT NOT NULL, ref TEXT NOT NULL, until INTEGER NOT NULL, PRIMARY KEY(repo,ref))`); err != nil {340 return err341 }342 for _, encoded := range retained.ListHistory {343 var item event.Event344 if err := json.Unmarshal([]byte(encoded), &item); err != nil {345 return fmt.Errorf("list history event: %w", err)346 }347 if err := event.Validate(item); err != nil {348 return fmt.Errorf("list history event validation: %w", err)349 }350 if _, err := tx.ExecContext(ctx, `INSERT OR IGNORE INTO list_history(owner,kind,d,event_id,created_at,saved_at,expires,raw) VALUES(?,?,?,?,?,?,?,?)`, item.PubKey, item.Kind, event.Tag(item, "d"), item.ID, item.CreatedAt, now, event.Expiration(item), encoded); err != nil {351 return err352 }353 }354 for _, id := range retained.Hidden {355 if id == "" {356 return errors.New("hidden event id is empty")357 }358 if _, err := tx.ExecContext(ctx, `INSERT OR IGNORE INTO hidden_events(id,reason) VALUES(?,?)`, id, "backup"); err != nil {359 return err360 }361 }362 for _, id := range retained.Hosted {363 if id == "" {364 return errors.New("hosted event id is empty")365 }366 if _, err := tx.ExecContext(ctx, `INSERT OR IGNORE INTO replication_hosted_events(id) VALUES(?)`, id); err != nil {367 return err368 }369 }370 for _, row := range retained.Pending {371 if row.ID == "" || row.Until < 0 {372 return errors.New("invalid pending event state")373 }374 if _, err := tx.ExecContext(ctx, `INSERT OR REPLACE INTO replication_pending_until(id,until) VALUES(?,?)`, row.ID, row.Until); err != nil {375 return err376 }377 // pending_events references events, so retain the marker only when the378 // event is already present. The durable until table above preserves it379 // for the subsequent event import.380 if _, err := tx.ExecContext(ctx, `INSERT OR IGNORE INTO pending_events(id,reason) SELECT id,'backup' FROM events WHERE id=?`, row.ID); err != nil {381 return err382 }383 }384 for _, row := range retained.PRRefs {385 if row.Repo == "" || row.Ref == "" || row.Until < 0 {386 return errors.New("invalid pull request reference state")387 }388 if _, err := tx.ExecContext(ctx, `INSERT OR REPLACE INTO replication_pr_refs(repo,ref,until) VALUES(?,?,?)`, row.Repo, row.Ref, row.Until); err != nil {389 return err390 }391 }392 return nil393 })394}395396func snapshotTables(ctx context.Context, store *storage.Store) ([]replication.BackupTable, error) {397 var out []replication.BackupTable398 err := store.WithTx(ctx, func(tx *sql.Tx) error {399 rows, err := tx.QueryContext(ctx, `SELECT name,sql FROM sqlite_schema WHERE type='table' AND name NOT LIKE 'sqlite_%' ORDER BY name`)400 if err != nil {401 return err402 }403 defer rows.Close()404 for rows.Next() {405 var name, definition string406 if err := rows.Scan(&name, &definition); err != nil {407 return err408 }409 if excludedSnapshotTable(name, definition) {410 continue411 }412 table, err := snapshotTableTx(ctx, tx, name)413 if err != nil {414 return err415 }416 out = append(out, table)417 }418 return rows.Err()419 })420 return out, err421}422423func excludedSnapshotTable(name, definition string) bool {424 lower := strings.ToLower(name + " " + definition)425 for _, part := range []string{"work_intents", "replication_job_cursors", "browser_sessions", "auth_challenges", "push_queue", "push_delivered", "delivery_queue"} {426 if strings.Contains(lower, part) {427 return true428 }429 }430 return strings.Contains(lower, "virtual table") || strings.HasSuffix(name, "_content") || strings.HasSuffix(name, "_data") || strings.HasSuffix(name, "_idx") || strings.HasSuffix(name, "_docsize") || strings.HasSuffix(name, "_config")431}432433func snapshotTable(ctx context.Context, db *sql.DB, name string) (replication.BackupTable, error) {434 return snapshotTableDB(ctx, db, name)435}436437func snapshotTableTx(ctx context.Context, tx *sql.Tx, name string) (replication.BackupTable, error) {438 if !safeIdentifier(name) {439 return replication.BackupTable{}, errors.New("daemon: unsafe table name")440 }441 rows, err := tx.QueryContext(ctx, `SELECT * FROM "`+name+`"`)442 if err != nil {443 return replication.BackupTable{}, err444 }445 defer rows.Close()446 columns, err := rows.Columns()447 if err != nil {448 return replication.BackupTable{}, err449 }450 table := replication.BackupTable{Name: name, Columns: columns}451 for rows.Next() {452 values := make([]any, len(columns))453 destinations := make([]any, len(columns))454 for i := range values {455 destinations[i] = &values[i]456 }457 if err := rows.Scan(destinations...); err != nil {458 return table, err459 }460 encoded := make([]json.RawMessage, len(values))461 for i, value := range values {462 raw, err := marshalDBValue(value)463 if err != nil {464 return table, err465 }466 encoded[i] = raw467 }468 table.Rows = append(table.Rows, encoded)469 }470 return table, rows.Err()471}472473func snapshotTableDB(ctx context.Context, db *sql.DB, name string) (replication.BackupTable, error) {474 if !safeIdentifier(name) {475 return replication.BackupTable{}, errors.New("daemon: unsafe table name")476 }477 rows, err := db.QueryContext(ctx, "SELECT * FROM \""+name+"\"")478 if err != nil {479 return replication.BackupTable{}, err480 }481 defer rows.Close()482 cols, err := rows.Columns()483 if err != nil {484 return replication.BackupTable{}, err485 }486 out := replication.BackupTable{Name: name, Columns: cols}487 for rows.Next() {488 values := make([]any, len(cols))489 dest := make([]any, len(cols))490 for i := range values {491 dest[i] = &values[i]492 }493 if err := rows.Scan(dest...); err != nil {494 return out, err495 }496 encoded := make([]json.RawMessage, len(values))497 for i, value := range values {498 raw, err := json.Marshal(value)499 if err != nil {500 return out, err501 }502 encoded[i] = raw503 }504 out.Rows = append(out.Rows, encoded)505 }506 return out, rows.Err()507}508509func restoreTables(ctx context.Context, store *storage.Store, tables []replication.BackupTable) error {510 return store.WithTx(ctx, func(tx *sql.Tx) error {511 if _, err := tx.ExecContext(ctx, `PRAGMA defer_foreign_keys=ON`); err != nil {512 return err513 }514 for _, table := range tables {515 if !safeIdentifier(table.Name) {516 return errors.New("daemon: unsafe table name")517 }518 if _, err := tx.ExecContext(ctx, "DELETE FROM \""+table.Name+"\""); err != nil {519 return err520 }521 for _, row := range table.Rows {522 if len(row) != len(table.Columns) {523 return errors.New("daemon: malformed table row")524 }525 columns := make([]string, len(table.Columns))526 marks := make([]string, len(table.Columns))527 args := make([]any, len(row))528 for i, column := range table.Columns {529 if !safeIdentifier(column) {530 return errors.New("daemon: unsafe column name")531 }532 columns[i] = "\"" + column + "\""533 marks[i] = "?"534 value, err := unmarshalDBValue(row[i])535 if err != nil {536 return err537 }538 args[i] = value539 }540 query := "INSERT INTO \"" + table.Name + "\" (" + strings.Join(columns, ",") + ") VALUES (" + strings.Join(marks, ",") + ")"541 if _, err := tx.ExecContext(ctx, query, args...); err != nil {542 return fmt.Errorf("restore %s: %w", table.Name, err)543 }544 }545 }546 if _, err := tx.ExecContext(ctx, `DELETE FROM search`); err == nil {547 if _, err := tx.ExecContext(ctx, `INSERT INTO search(rowid,content) SELECT seq,json_extract(raw,'$.content') FROM events WHERE json_extract(raw,'$.content')<>''`); err != nil {548 return err549 }550 }551 if _, err := tx.ExecContext(ctx, `UPDATE replication_jobs SET state='pending',phase='pending',error='',stored=0,sent=0,skipped=0,refused=0,cursor=0,rounds=0,started_at=0,finished_at=0,updated_at=strftime('%s','now')`); err != nil {552 // Older tenant databases may not have the replication job table yet.553 if !strings.Contains(err.Error(), "no such table") {554 return err555 }556 }557 return nil558 })559}560561func marshalDBValue(value any) (json.RawMessage, error) {562 if bytesValue, ok := value.([]byte); ok {563 return json.Marshal(map[string]string{"$bytes": base64.StdEncoding.EncodeToString(bytesValue)})564 }565 return json.Marshal(value)566}567568func unmarshalDBValue(raw json.RawMessage) (any, error) {569 var marker map[string]string570 if err := json.Unmarshal(raw, &marker); err == nil && marker["$bytes"] != "" {571 return base64.StdEncoding.DecodeString(marker["$bytes"])572 }573 decoder := json.NewDecoder(bytes.NewReader(raw))574 decoder.UseNumber()575 var value any576 if err := decoder.Decode(&value); err != nil {577 return nil, err578 }579 if number, ok := value.(json.Number); ok {580 if integer, err := strconv.ParseInt(string(number), 10, 64); err == nil {581 return integer, nil582 }583 return strconv.ParseFloat(string(number), 64)584 }585 return value, nil586}587588func snapshotFiles(root string) ([]replication.BackupObject, error) {589 if root == "" {590 return nil, nil591 }592 var out []replication.BackupObject593 err := filepath.Walk(root, func(path string, info os.FileInfo, err error) error {594 if err != nil {595 return err596 }597 if info.IsDir() {598 return nil599 }600 body, err := os.ReadFile(path)601 if err != nil {602 return err603 }604 sum := sha256.Sum256(body)605 rel, err := filepath.Rel(root, path)606 if err != nil {607 return err608 }609 out = append(out, replication.BackupObject{Name: rel, SHA256: hex.EncodeToString(sum[:]), Content: body})610 return nil611 })612 if os.IsNotExist(err) {613 return nil, nil614 }615 return out, err616}617func restoreFiles(root string, objects []replication.BackupObject) error {618 allowed := make(map[string]struct{}, len(objects))619 for _, object := range objects {620 allowed[filepath.Clean(object.Name)] = struct{}{}621 }622 if err := filepath.Walk(root, func(path string, info os.FileInfo, walkErr error) error {623 if walkErr != nil {624 if os.IsNotExist(walkErr) {625 return nil626 }627 return walkErr628 }629 if info.IsDir() || path == root {630 return nil631 }632 rel, err := filepath.Rel(root, path)633 if err != nil {634 return err635 }636 if _, ok := allowed[filepath.Clean(rel)]; !ok {637 return os.Remove(path)638 }639 return nil640 }); err != nil && !os.IsNotExist(err) {641 return err642 }643 for _, object := range objects {644 if object.Name == "" || filepath.IsAbs(object.Name) || strings.HasPrefix(filepath.Clean(object.Name), ".."+string(filepath.Separator)) {645 return errors.New("daemon: unsafe backup file")646 }647 sum := sha256.Sum256(object.Content)648 if hex.EncodeToString(sum[:]) != object.SHA256 {649 return errors.New("daemon: backup file checksum mismatch")650 }651 path := filepath.Join(root, object.Name)652 if err := os.MkdirAll(filepath.Dir(path), 0700); err != nil {653 return err654 }655 if err := os.WriteFile(path, object.Content, 0600); err != nil {656 return err657 }658 }659 return nil660}661func safeIdentifier(value string) bool {662 if value == "" {663 return false664 }665 for _, r := range value {666 if r != '_' && (r < 'a' || r > 'z') && (r < 'A' || r > 'Z') && (r < '0' || r > '9') {667 return false668 }669 }670 return true671}672673type tenantDiscovery struct {674 tenant *Tenant675 transport *replication.NostrTransport676 seeds []string677}678679func (t *Tenant) ReplicationDiscovery(seeds []string) replication.Discovery {680 dialer := replication.WebsocketDialer{}681 if t != nil && t.app != nil {682 dialer.AllowPrivate = t.app.cfg.AllowPrivateRelays683 dialer.MaxMessageBytes = t.app.cfg.MaxMessageBytes684 }685 return tenantDiscovery{tenant: t, transport: &replication.NostrTransport{Dialer: dialer}, seeds: append([]string(nil), seeds...)}686}687func (d tenantDiscovery) DiscoverRelays(ctx context.Context, pubkey string) ([]string, error) {688 return d.discover(ctx, pubkey, false)689}690691func (d tenantDiscovery) DiscoverReadRelays(ctx context.Context, pubkey string) ([]string, error) {692 return d.discover(ctx, pubkey, true)693}694695func (d tenantDiscovery) discover(ctx context.Context, pubkey string, read bool) ([]string, error) {696 if d.tenant == nil {697 return nil, errors.New("daemon: nil tenant")698 }699 directory := localDirectory{store: d.tenant.store}700 local := directory.WriteRelays(pubkey)701 if read {702 local = directory.ReadRelays(pubkey)703 }704 if len(local) > 0 {705 return local, nil706 }707 if d.transport == nil || len(d.seeds) == 0 {708 return nil, nil709 }710 for _, seed := range d.seeds {711 events, err := d.transport.Query(ctx, seed, event.Filter{Authors: []string{pubkey}, Kinds: []int{10002}, Tags: map[string][]string{}, Limit: intPtr(1)})712 if err != nil {713 continue714 }715 for _, item := range events {716 if err := event.Validate(item); err != nil {717 continue718 }719 if _, err := d.tenant.store.Save(ctx, item, storage.SaveOptions{Now: time.Now().Unix()}); err != nil && err != storage.ErrDuplicate && err != storage.ErrReplaced {720 continue721 }722 stored := localDirectory{store: d.tenant.store}723 if read {724 return stored.ReadRelays(pubkey), nil725 }726 return stored.WriteRelays(pubkey), nil727 }728 }729 return nil, nil730}731func intPtr(value int) *int { return &value }732733func (t *Tenant) PrepareReplicationCallbacks(ctx context.Context, e event.Event) ([]storage.Intent, error) {734 if t == nil || !t.Policy().Features.Push {735 return nil, nil736 }737 approved := t.callbackPolicy()738 rows, err := t.store.DB().QueryContext(ctx, `SELECT raw FROM events WHERE kind=? ORDER BY created_at DESC`, event.KIND_PUSH_REGISTRATION)739 if err != nil {740 return nil, err741 }742 defer rows.Close()743 registrations := make([]replication.PushRegistration, 0)744 for rows.Next() {745 var raw string746 if err := rows.Scan(&raw); err != nil {747 return nil, err748 }749 var registration event.Event750 if err := json.Unmarshal([]byte(raw), ®istration); err != nil {751 continue752 }753 parsed, parseErr := replication.ParsePushRegistration(registration, approved, t.RelayURL())754 if parseErr == nil {755 registrations = append(registrations, parsed)756 }757 }758 return replication.PrepareCallbackIntents(ctx, e, registrations), rows.Err()759}760761// callbackPolicy keeps host approval separate from the tenant owner's762// registration list. A tenant can opt in an origin only after the operator has763// placed it in PushCallbackOrigins as well.764func (t *Tenant) callbackPolicy() replication.CallbackPolicy {765 p := t.Policy()766 return replication.CallbackPolicy{HostOrigins: append([]string(nil), t.app.cfg.PushCallbackOrigins...), OwnerOrigins: append([]string(nil), p.PushCallbacks...)}767}768769// validatePushRegistration applies the source relay's admission rules before770// the registration is persisted. Callback workers revalidate later, but an771// invalid registration must never become durable state.772func (t *Tenant) validatePushRegistration(ctx context.Context, e event.Event, s relay.Session) error {773 if e.Kind != event.KIND_PUSH_REGISTRATION {774 return nil775 }776 if !t.Policy().Features.Push {777 return errors.New("unsupported: relay push is switched off")778 }779 if !containsString(s.PubKeys, e.PubKey) {780 return errors.New("auth-required: push registration must be authenticated as its author")781 }782 if role, err := t.community.Role(ctx, e.PubKey); err != nil {783 return err784 } else if role == "" {785 return errors.New("restricted: push registrations are limited to relay members")786 }787 if len(event.TagValues(e, "d")) != 1 || event.Tag(e, "d") == "" || len(event.Tag(e, "d")) > 256 {788 return errors.New("invalid: push registration needs one d tag")789 }790 if len(event.TagValues(e, "relay")) == 0 || len(event.TagValues(e, "relay")) > 4 || len(event.TagValues(e, "callback")) != 1 {791 return errors.New("invalid: push registration needs relay and callback tags")792 }793 for _, name := range []string{"filter", "ignore"} {794 values := event.TagValues(e, name)795 for _, raw := range values {796 var object map[string]json.RawMessage797 if err := json.Unmarshal([]byte(raw), &object); err != nil {798 return errors.New("invalid: push filter is not JSON")799 }800 if object == nil {801 return errors.New("invalid: push filter must be an object")802 }803 for key := range object {804 if !pushFilterField(key) {805 return errors.New("invalid: unknown push filter field")806 }807 }808 }809 }810 parsed, err := replication.ParsePushRegistration(e, t.callbackPolicy(), t.RelayURL())811 if err != nil {812 return err813 }814 for _, filter := range append(parsed.Filters, parsed.Ignore...) {815 for _, kind := range filter.Kinds {816 if kind < 0 || kind > 65535 {817 return errors.New("invalid: push filter value out of range")818 }819 }820 if filter.Since != nil && *filter.Since < 0 || filter.Until != nil && *filter.Until < 0 {821 return errors.New("invalid: push filter value out of range")822 }823 }824 return nil825}826827func pushFilterField(key string) bool {828 if key == "ids" || key == "authors" || key == "kinds" || key == "since" || key == "until" {829 return true830 }831 return len(key) == 2 && key[0] == '#'832}833834func (t *Tenant) ReplicationCallbackHandler() work.Handler {835 return replication.NewCallbackHandlerWithEventVisibility(t.store, replication.CallbackClient{}, func() replication.CallbackPolicy {836 return t.callbackPolicy()837 }, func() bool { return t.Policy().Features.Push }, func(ctx context.Context, pubkey string) bool {838 return t.Policy().Reads == "open" || func() bool { role, err := t.community.Role(ctx, pubkey); return err == nil && role != "" }()839 }, func(ctx context.Context, pubkey string, delivered event.Event) bool {840 return t.gate.CanSee(ctx, delivered, relay.Session{PubKeys: []string{pubkey}, RelayURL: t.RelayURL()}, nil)841 })842}