1package daemon23// Callbacks wake agents that cannot hold a relay connection open. A key4// registers an https URL with a filter, and the relay POSTs each matching5// event the key may see to that URL as it arrives. Registration is a6// management method; delivery is durable work with bounded retries. This7// file holds storage, validation, matching and the management methods;8// callbacks_delivery.go holds the POST.910import (11 "context"12 "crypto/rand"13 "database/sql"14 "encoding/hex"15 "encoding/json"16 "errors"17 "fmt"18 "net"19 "net/url"20 "sort"21 "strings"22 "time"2324 "github.com/FelineStateMachine/tinyrelay/internal/community"25 "github.com/FelineStateMachine/tinyrelay/internal/event"26 "github.com/FelineStateMachine/tinyrelay/internal/relay"27 "github.com/FelineStateMachine/tinyrelay/internal/storage"28)2930const (31 callbackDelivery = "callback-delivery"32 callbackTagMax = 833 callbackKindMax = 3234 callbackURLMax = 204835 callbackSecretMin = 1636 callbackSecretMax = 12837 callbackTimeout = 10 * time.Second38 callbackMaxAttempts = 339 callbackPauseFailures = 2040)4142// callbackBackoff is the wait before the second and third attempt.43var callbackBackoff = []time.Duration{time.Minute, 5 * time.Minute}4445const callbackSchema = `CREATE TABLE IF NOT EXISTS callbacks(id TEXT PRIMARY KEY, owner TEXT NOT NULL, url TEXT NOT NULL, filter TEXT NOT NULL, secret TEXT NOT NULL, created_at INTEGER NOT NULL, last_delivery_at INTEGER NOT NULL DEFAULT 0, last_status TEXT NOT NULL DEFAULT '', failures INTEGER NOT NULL DEFAULT 0, paused INTEGER NOT NULL DEFAULT 0);46CREATE INDEX IF NOT EXISTS callbacks_owner ON callbacks(owner);`4748// callbackFilter is the NIP-01 subset a callback may use: kinds, authors and49// the a, e, p and h tags. since is accepted and ignored.50type callbackFilter struct {51 Kinds []int `json:"kinds"`52 Authors []string `json:"authors,omitempty"`53 Tags map[string][]string `json:"-"`54}5556var callbackFilterTags = []string{"a", "e", "p", "h"}5758// MarshalJSON writes the filter in NIP-01 form, with tag lists as #a, #e,59// #p and #h.60func (f callbackFilter) MarshalJSON() ([]byte, error) {61 out := map[string]any{"kinds": f.Kinds}62 if len(f.Authors) > 0 {63 out["authors"] = f.Authors64 }65 for _, name := range callbackFilterTags {66 if values := f.Tags[name]; len(values) > 0 {67 out["#"+name] = values68 }69 }70 return json.Marshal(out)71}7273// eventFilter is the storage filter that matches the same events.74func (f callbackFilter) eventFilter() event.Filter {75 tags := map[string][]string{}76 for name, values := range f.Tags {77 tags[name] = values78 }79 return event.Filter{Kinds: f.Kinds, Authors: f.Authors, Tags: tags}80}8182// parseCallbackFilter validates the filter subset. Unknown keys, empty83// lists, malformed values and lists longer than the cap are refused so a84// registration that will never match is caught when it is made.85func parseCallbackFilter(raw json.RawMessage) (callbackFilter, error) {86 var values map[string]json.RawMessage87 if len(raw) == 0 || json.Unmarshal(raw, &values) != nil {88 // A filter may arrive as a JSON string from a plain form.89 var text string90 if json.Unmarshal(raw, &text) != nil || json.Unmarshal([]byte(text), &values) != nil {91 return callbackFilter{}, errors.New("invalid: filter must be a JSON object")92 }93 }94 f := callbackFilter{Tags: map[string][]string{}}95 for key, value := range values {96 switch {97 case key == "kinds":98 if json.Unmarshal(value, &f.Kinds) != nil || len(f.Kinds) == 0 || len(f.Kinds) > callbackKindMax {99 return callbackFilter{}, fmt.Errorf("invalid: kinds must list 1 to %d kinds", callbackKindMax)100 }101 for _, kind := range f.Kinds {102 if kind < 0 || kind > 65535 {103 return callbackFilter{}, errors.New("invalid: kinds must be between 0 and 65535")104 }105 }106 case key == "authors":107 if json.Unmarshal(value, &f.Authors) != nil || len(f.Authors) == 0 || len(f.Authors) > callbackTagMax {108 return callbackFilter{}, fmt.Errorf("invalid: authors must list 1 to %d public keys", callbackTagMax)109 }110 for _, author := range f.Authors {111 if !hex64(author) {112 return callbackFilter{}, errors.New("invalid: authors must be 64 lowercase hex characters")113 }114 }115 case key == "since":116 // Accepted for compatibility with clients that send whole117 // filters; a callback only sees events that arrive after it.118 case len(key) == 2 && key[0] == '#' && containsString(callbackFilterTags, key[1:]):119 var list []string120 if json.Unmarshal(value, &list) != nil || len(list) == 0 || len(list) > callbackTagMax {121 return callbackFilter{}, fmt.Errorf("invalid: %s must list 1 to %d values", key, callbackTagMax)122 }123 for _, item := range list {124 if err := checkCallbackTagValue(key[1:], item); err != nil {125 return callbackFilter{}, err126 }127 }128 f.Tags[key[1:]] = list129 default:130 return callbackFilter{}, fmt.Errorf("invalid: filter key %q is not supported; use kinds, authors, #a, #e, #p or #h", key)131 }132 }133 if len(f.Kinds) == 0 {134 return callbackFilter{}, errors.New("invalid: filter must name at least one kind")135 }136 sort.Ints(f.Kinds)137 return f, nil138}139140func checkCallbackTagValue(name, value string) error {141 switch name {142 case "e", "p":143 if !hex64(value) {144 return fmt.Errorf("invalid: #%s values must be 64 lowercase hex characters", name)145 }146 case "a":147 parts := strings.SplitN(value, ":", 3)148 if len(parts) != 3 || !hex64(parts[1]) || strings.Trim(parts[0], "0123456789") != "" || parts[0] == "" {149 return errors.New("invalid: #a values must be <kind>:<pubkey>:<identifier>")150 }151 case "h":152 if !community.ValidRoomID(value) {153 return errors.New("invalid: #h values must be room ids")154 }155 }156 return nil157}158159// callbackRecord is one stored callback. The secret never leaves the160// tenant except once, in the addcallback answer.161type callbackRecord struct {162 ID string `json:"id"`163 Owner string `json:"owner"`164 URL string `json:"url"`165 Filter callbackFilter `json:"filter"`166 CreatedAt int64 `json:"created"`167 LastAt int64 `json:"lastDelivery"`168 LastStatus string `json:"lastStatus"`169 Failures int `json:"failures"`170 Paused bool `json:"paused"`171 secret string172}173174// summary is the management view: the host stands in for the URL so a175// listing shown to moderators does not expose the path.176func (c callbackRecord) summary() map[string]any {177 return map[string]any{"id": c.ID, "owner": c.Owner, "url": c.URL, "host": callbackHost(c.URL), "filter": c.Filter, "created": c.CreatedAt, "lastDelivery": c.LastAt, "lastStatus": c.LastStatus, "failures": c.Failures, "paused": c.Paused}178}179180func callbackHost(raw string) string {181 parsed, err := url.Parse(raw)182 if err != nil {183 return ""184 }185 return parsed.Host186}187188func (s *callbackService) initCallbacks(ctx context.Context) error {189 _, err := s.store.DB().ExecContext(ctx, callbackSchema)190 return err191}192193// validateWebhookURL accepts https URLs to public hosts. When the relay194// itself runs on a loopback address, as it does in tests, private targets are195// allowed so a local receiver can be exercised.196func validateWebhookURL(raw, publicURL string) error {197 if len(raw) > callbackURLMax {198 return errors.New("invalid: url is too long")199 }200 parsed, err := url.Parse(raw)201 if err != nil || parsed.Scheme != "https" || parsed.Hostname() == "" || parsed.User != nil || parsed.Fragment != "" {202 return errors.New("invalid: url must be https without credentials or a fragment")203 }204 if loopbackPublicURL(publicURL) {205 return nil206 }207 host := strings.ToLower(parsed.Hostname())208 if ip := net.ParseIP(host); ip != nil {209 if privateAddress(ip) {210 return errors.New("invalid: url must not name a private or loopback address")211 }212 return nil213 }214 if host == "localhost" || strings.HasSuffix(host, ".localhost") || strings.HasSuffix(host, ".local") || strings.HasSuffix(host, ".internal") || !strings.Contains(host, ".") {215 return errors.New("invalid: url must name a public host")216 }217 return nil218}219220func privateAddress(ip net.IP) bool {221 if ip.IsLoopback() || ip.IsPrivate() || ip.IsLinkLocalUnicast() || ip.IsLinkLocalMulticast() || ip.IsMulticast() || ip.IsUnspecified() {222 return true223 }224 _, shared, _ := net.ParseCIDR("100.64.0.0/10")225 return shared.Contains(ip)226}227228// loopbackPublicURL reports whether the relay's own public URL is a loopback229// or private address, which only happens in tests and local trials.230func loopbackPublicURL(publicURL string) bool {231 parsed, err := url.Parse(publicURL)232 if err != nil {233 return false234 }235 host := strings.ToLower(parsed.Hostname())236 if host == "localhost" {237 return true238 }239 ip := net.ParseIP(host)240 return ip != nil && privateAddress(ip)241}242243func callbackMethod(method string) bool {244 return containsString([]string{"listcallbacks", "addcallback", "removecallback", "pausecallback", "resumecallback"}, method)245}246247// callbackExecute serves the callback management methods. Members and agents248// manage callbacks for their own key; the owner and moderators see and249// control every callback. Each call runs under the callback operation.250func (s *callbackService) callbackExecute(ctx context.Context, actor, method string, params []json.RawMessage) (result any, err error) {251 ctx, finish := s.telemetry.Start(ctx, "callback")252 outcome := "error"253 defer func() {254 if err != nil {255 outcome = callbackOutcome(err)256 }257 finish(outcome)258 }()259 role, err := s.community.Role(ctx, actor)260 if err != nil {261 return nil, err262 }263 if role == "" {264 return nil, fmt.Errorf("restricted: guest may not %s", method)265 }266 operator := role == "owner" || role == "moderator"267 switch method {268 case "listcallbacks":269 var rows []callbackRecord270 if operator {271 rows, err = s.callbackRows(ctx, "")272 } else {273 rows, err = s.callbackRows(ctx, actor)274 }275 if err != nil {276 return nil, err277 }278 out := make([]map[string]any, 0, len(rows))279 for _, row := range rows {280 item := row.summary()281 if row.Owner != actor {282 // Operators see whose callback it is and where it goes,283 // not the path, which may carry a token.284 delete(item, "url")285 }286 out = append(out, item)287 }288 outcome = "ok"289 return out, nil290 case "addcallback":291 result, err = s.addCallback(ctx, actor, role, params)292 default:293 result, err = s.changeCallback(ctx, actor, operator, method, params)294 }295 if err == nil {296 outcome = "ok"297 }298 return result, err299}300301func callbackOutcome(err error) string {302 switch {303 case strings.HasPrefix(err.Error(), "invalid:"), strings.HasPrefix(err.Error(), "not found:"):304 return "invalid"305 case strings.HasPrefix(err.Error(), "restricted:"):306 return "unauthorized"307 default:308 return "error"309 }310}311312func (s *callbackService) addCallback(ctx context.Context, actor, role string, params []json.RawMessage) (any, error) {313 if len(params) != 1 {314 return nil, errors.New("invalid: addcallback expects one object with url, filter and an optional secret")315 }316 var options struct {317 URL string `json:"url"`318 Filter json.RawMessage `json:"filter"`319 Secret string `json:"secret"`320 }321 if err := json.Unmarshal(params[0], &options); err != nil {322 return nil, errors.New("invalid: addcallback expects one object with url, filter and an optional secret")323 }324 options.URL = strings.TrimSpace(options.URL)325 if err := validateWebhookURL(options.URL, s.publicURL); err != nil {326 return nil, err327 }328 filter, err := parseCallbackFilter(options.Filter)329 if err != nil {330 return nil, err331 }332 if role == "agent" {333 if err := s.checkCallbackScope(ctx, actor, filter); err != nil {334 return nil, err335 }336 }337 generated := false338 secret := strings.TrimSpace(options.Secret)339 if secret == "" {340 secret, err = randomHex(32)341 if err != nil {342 return nil, err343 }344 generated = true345 } else if len(secret) < callbackSecretMin || len(secret) > callbackSecretMax || strings.Trim(secret, "!\"#$%&'()*+,-./0123456789:;<=>?@ABCDEFGHIJKLMNOPQRSTUVWXYZ[\\]^_`abcdefghijklmnopqrstuvwxyz{|}~") != "" {346 return nil, fmt.Errorf("invalid: secret must be %d to %d printable ASCII characters", callbackSecretMin, callbackSecretMax)347 }348 id, err := randomHex(16)349 if err != nil {350 return nil, err351 }352 encoded, err := json.Marshal(filter)353 if err != nil {354 return nil, err355 }356 limit := s.policy().Callbacks357 now := time.Now().Unix()358 err = s.store.WithTx(ctx, func(tx *sql.Tx) error {359 if role != "owner" {360 var count int361 if err := tx.QueryRowContext(ctx, `SELECT count(*) FROM callbacks WHERE owner=?`, actor).Scan(&count); err != nil {362 return err363 }364 if count >= limit {365 return fmt.Errorf("invalid: at most %d callbacks per key", limit)366 }367 }368 _, err := tx.ExecContext(ctx, `INSERT INTO callbacks(id,owner,url,filter,secret,created_at) VALUES(?,?,?,?,?,?)`, id, actor, options.URL, string(encoded), secret, now)369 return err370 })371 if err != nil {372 return nil, err373 }374 s.invalidateCallbacks()375 if err := s.community.Record(ctx, actor, "addcallback", id, ""); err != nil {376 return nil, err377 }378 // Counts only: the host, path and secret stay out of the log.379 s.telemetry.Logger().Info("callback registered", "tenant", s.tenant.Name, "kinds", len(filter.Kinds), "generated", generated)380 record := callbackRecord{ID: id, Owner: actor, URL: options.URL, Filter: filter, CreatedAt: now}381 out := record.summary()382 out["secret"] = secret383 return out, nil384}385386// checkCallbackScope keeps an agent's callback inside its grant: a room it387// names must be one it may post in and a repository it names must be one388// the grant covers. Kinds are not restricted, so an agent can listen for389// answers it may not publish itself.390func (s *callbackService) checkCallbackScope(ctx context.Context, actor string, filter callbackFilter) error {391 grant, ok, err := s.community.AgentGrant(ctx, actor)392 if err != nil {393 return err394 }395 if !ok || !grant.Active(time.Now().Unix()) {396 return errors.New("restricted: agent grant is not active")397 }398 for _, room := range filter.Tags["h"] {399 if !grant.AllowsRoom(room) {400 return fmt.Errorf("restricted: agent grant does not cover room %s", room)401 }402 }403 for _, coordinate := range filter.Tags["a"] {404 owner, identifier, ok := repoCoordinate(coordinate)405 if !ok {406 continue407 }408 if grant.RepoLevel(owner, identifier) == "" {409 return fmt.Errorf("restricted: agent grant does not cover repository %s", identifier)410 }411 }412 return nil413}414415func (s *callbackService) changeCallback(ctx context.Context, actor string, operator bool, method string, params []json.RawMessage) (any, error) {416 id := callbackIDParam(params)417 if id == "" {418 return nil, errors.New("invalid: callback id required")419 }420 record, err := s.callback(ctx, id)421 if err != nil {422 return nil, err423 }424 if !operator && record.Owner != actor {425 return nil, fmt.Errorf("restricted: only the callback's owner, the relay owner or a moderator may %s", method)426 }427 switch method {428 case "removecallback":429 _, err = s.store.DB().ExecContext(ctx, `DELETE FROM callbacks WHERE id=?`, id)430 case "pausecallback":431 _, err = s.store.DB().ExecContext(ctx, `UPDATE callbacks SET paused=1, last_status='paused' WHERE id=?`, id)432 record.Paused, record.LastStatus = true, "paused"433 case "resumecallback":434 _, err = s.store.DB().ExecContext(ctx, `UPDATE callbacks SET paused=0, failures=0, last_status='' WHERE id=?`, id)435 record.Paused, record.Failures, record.LastStatus = false, 0, ""436 default:437 return nil, fmt.Errorf("unsupported: management method %q", method)438 }439 if err != nil {440 return nil, err441 }442 s.invalidateCallbacks()443 if err := s.community.Record(ctx, actor, method, id, ""); err != nil {444 return nil, err445 }446 if method == "removecallback" {447 return map[string]any{"id": id, "removed": true}, nil448 }449 result := record.summary()450 if operator && record.Owner != actor {451 delete(result, "url")452 }453 return result, nil454}455456// callbackIDParam reads the id from a bare string or from {"id": ...}.457func callbackIDParam(params []json.RawMessage) string {458 if len(params) == 0 {459 return ""460 }461 if id := stringParam(params, 0); id != "" {462 return id463 }464 var options struct {465 ID string `json:"id"`466 }467 _ = json.Unmarshal(params[0], &options)468 return strings.TrimSpace(options.ID)469}470471func (s *callbackService) callback(ctx context.Context, id string) (callbackRecord, error) {472 rows, err := s.callbackRows(ctx, "", id)473 if err != nil {474 return callbackRecord{}, err475 }476 if len(rows) == 0 {477 return callbackRecord{}, errors.New("not found: callback")478 }479 return rows[0], nil480}481482// callbackRows lists callbacks, narrowed to an owner or to one id.483func (s *callbackService) callbackRows(ctx context.Context, owner string, ids ...string) ([]callbackRecord, error) {484 query := `SELECT id,owner,url,filter,secret,created_at,last_delivery_at,last_status,failures,paused FROM callbacks`485 args := []any{}486 switch {487 case len(ids) > 0:488 query += ` WHERE id=?`489 args = append(args, ids[0])490 case owner != "":491 query += ` WHERE owner=?`492 args = append(args, owner)493 }494 query += ` ORDER BY created_at,id`495 rows, err := s.store.DB().QueryContext(ctx, query, args...)496 if err != nil {497 return nil, fmt.Errorf("list callbacks: %w", err)498 }499 defer rows.Close()500 var out []callbackRecord501 for rows.Next() {502 var item callbackRecord503 var filter string504 var paused int505 if err := rows.Scan(&item.ID, &item.Owner, &item.URL, &filter, &item.secret, &item.CreatedAt, &item.LastAt, &item.LastStatus, &item.Failures, &paused); err != nil {506 return nil, err507 }508 item.Paused = paused != 0509 if item.Filter, err = parseCallbackFilter(json.RawMessage(filter)); err != nil {510 return nil, fmt.Errorf("callback filter: %w", err)511 }512 out = append(out, item)513 }514 return out, rows.Err()515}516517// callbackRules returns the active callbacks indexed by kind. The index is518// built from the table on first use and dropped whenever a callback519// changes, so matching an event costs one map lookup.520func (s *callbackService) callbackRules(ctx context.Context) map[int][]callbackRecord {521 s.callbackMu.Lock()522 defer s.callbackMu.Unlock()523 if s.callbackIndex != nil {524 return s.callbackIndex525 }526 index := map[int][]callbackRecord{}527 rows, err := s.callbackRows(ctx, "")528 if err != nil {529 s.telemetry.Logger().Error("load callbacks", "tenant", s.tenant.Name, "error", err)530 return index531 }532 for _, row := range rows {533 if row.Paused {534 continue535 }536 for _, kind := range row.Filter.Kinds {537 index[kind] = append(index[kind], row)538 }539 }540 s.callbackIndex = index541 return index542}543544func (s *callbackService) invalidateCallbacks() {545 s.callbackMu.Lock()546 s.callbackIndex = nil547 s.callbackMu.Unlock()548}549550// CandidateIDs captures the active callback registrations that match an551// event. The caller must persist these IDs with the event so later planning552// cannot include callbacks registered after the event was accepted.553func (s *callbackService) CandidateIDs(ctx context.Context, e event.Event) ([]string, error) {554 rows, err := s.callbackRows(ctx, "")555 if err != nil {556 return nil, err557 }558 ids := make([]string, 0, len(rows))559 for _, candidate := range rows {560 if candidate.Paused || !event.Matches(candidate.Filter.eventFilter(), e) {561 continue562 }563 ids = append(ids, candidate.ID)564 }565 return ids, nil566}567568// PrepareFor creates delivery intents for captured registration IDs. Each569// registration is re-read so deletion, pausing, membership changes and570// visibility changes take effect before an intent is created.571func (s *callbackService) PrepareFor(ctx context.Context, e event.Event, ids []string) ([]storage.Intent, error) {572 if len(ids) == 0 {573 return nil, nil574 }575 intents := make([]storage.Intent, 0, len(ids))576 seen := make(map[string]struct{}, len(ids))577 for _, id := range ids {578 if id == "" {579 continue580 }581 if _, ok := seen[id]; ok {582 continue583 }584 seen[id] = struct{}{}585 candidate, err := s.callback(ctx, id)586 if err != nil {587 if strings.HasPrefix(err.Error(), "not found:") {588 continue589 }590 return nil, err591 }592 if candidate.Paused || !event.Matches(candidate.Filter.eventFilter(), e) {593 continue594 }595 if !s.callbackOwnerCanSee(ctx, candidate.Owner, e) {596 continue597 }598 payload, err := json.Marshal(callbackPayload{Event: e, Attempt: 1})599 if err != nil {600 return nil, fmt.Errorf("encode callback delivery: %w", err)601 }602 intents = append(intents, storage.Intent{Kind: callbackDelivery, EventID: e.ID, Target: candidate.ID, Payload: string(payload)})603 }604 return intents, nil605}606607func (s *callbackService) callbackOwnerCanSee(ctx context.Context, owner string, e event.Event) bool {608 return s.gate.CanSee(ctx, e, relay.Session{PubKeys: []string{owner}, RelayURL: s.relayURL}, nil)609}610611// Prepare matches current callback registrations and rechecks their access612// before returning delivery intents. The caller persists the returned work.613func (s *callbackService) Prepare(ctx context.Context, e event.Event) ([]storage.Intent, error) {614 ids, err := s.CandidateIDs(ctx, e)615 if err != nil {616 return nil, err617 }618 return s.PrepareFor(ctx, e, ids)619}620621// notifyCallbacks plans and queues delivery in its own transaction, logging622// planning or storage failures. Ephemeral publication uses this path because623// its event is carried by the delivery payload.624func (s *callbackService) notifyCallbacks(ctx context.Context, e event.Event) {625 intents, err := s.Prepare(ctx, e)626 if err != nil {627 s.telemetry.Logger().Error("callback deliveries not prepared", "tenant", s.tenant.Name, "error", err)628 return629 }630 if len(intents) == 0 {631 return632 }633 err = s.store.WithTx(ctx, func(tx *sql.Tx) error {634 return storage.AddIntents(ctx, tx, intents, time.Now().Unix())635 })636 if err != nil {637 s.telemetry.Logger().Error("callback deliveries not queued", "tenant", s.tenant.Name, "error", err)638 return639 }640 s.telemetry.Logger().Debug("callback deliveries queued", "tenant", s.tenant.Name, "queued", len(intents))641}642643func randomHex(size int) (string, error) {644 buffer := make([]byte, size)645 if _, err := rand.Read(buffer); err != nil {646 return "", fmt.Errorf("callback secret: %w", err)647 }648 return hex.EncodeToString(buffer), nil649}650651// Tenant wrappers keep management and work dispatch independent of the652// callback service's storage and delivery state.653func (t *Tenant) initCallbacks(ctx context.Context) error {654 return t.callbacks.initCallbacks(ctx)655}656657func (t *Tenant) callbackExecute(ctx context.Context, actor, method string, params []json.RawMessage) (any, error) {658 return t.callbacks.callbackExecute(ctx, actor, method, params)659}660661func (t *Tenant) notifyCallbacks(ctx context.Context, e event.Event) {662 t.callbacks.notifyCallbacks(ctx, e)663}