1package daemon23// Callback delivery. Each matching event is one POST to the callback URL,4// signed with the callback's secret. A failed POST is tried again after one5// minute and then after five; after 20 failures in a row the callback is6// paused until its owner resumes it.78import (9 "context"10 "crypto/hmac"11 "crypto/sha256"12 "crypto/tls"13 "database/sql"14 "encoding/hex"15 "encoding/json"16 "errors"17 "fmt"18 "io"19 "net/http"20 "strings"21 "sync"22 "time"2324 "github.com/FelineStateMachine/tinyrelay/internal/event"25 "github.com/FelineStateMachine/tinyrelay/internal/relay"26 "github.com/FelineStateMachine/tinyrelay/internal/replication"27 "github.com/FelineStateMachine/tinyrelay/internal/work"28)2930// callbackPayload carries an event and its delivery attempt. Stored events31// are reloaded before delivery; ephemeral events use this signed payload.32type callbackPayload struct {33 Event event.Event `json:"event"`34 Attempt int `json:"attempt"`35}3637// callbackSignature is the X-Tiny-Signature value for a body: sha256= and38// the hex HMAC-SHA256 of the body under the callback secret.39func callbackSignature(secret string, body []byte) string {40 mac := hmac.New(sha256.New, []byte(secret))41 mac.Write(body)42 return "sha256=" + hex.EncodeToString(mac.Sum(nil))43}4445// callbackHTTPClient refuses private addresses and redirects unless the46// relay itself runs on a loopback address.47func (s *callbackService) callbackHTTPClient() *http.Client {48 if s.callbackClient != nil {49 return s.callbackClient50 }51 if loopbackPublicURL(s.publicURL) {52 return &http.Client{Timeout: callbackTimeout, Transport: &http.Transport{TLSClientConfig: &tls.Config{MinVersion: tls.VersionTLS12}}, CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse }}53 }54 return replication.NewPinnedClient(callbackTimeout)55}5657// callbackLock bounds delivery to one POST per callback at a time.58func (s *callbackService) callbackLock(id string) *sync.Mutex {59 s.callbackMu.Lock()60 defer s.callbackMu.Unlock()61 if s.callbackLocks == nil {62 s.callbackLocks = map[string]*sync.Mutex{}63 }64 lock := s.callbackLocks[id]65 if lock == nil {66 lock = &sync.Mutex{}67 s.callbackLocks[id] = lock68 }69 return lock70}7172// handleCallbackDelivery serves one callback-delivery intent. It rechecks73// the callback and the key's access right before the POST, so a callback74// removed or paused after the event arrived delivers nothing.75func (s *callbackService) handleCallbackDelivery(ctx context.Context, intent work.Intent) (err error) {76 ctx, finish := s.telemetry.Start(ctx, "callback")77 outcome := "error"78 defer func() { finish(outcome) }()79 var payload callbackPayload80 if err := json.Unmarshal([]byte(intent.Payload), &payload); err != nil || payload.Event.ID == "" {81 outcome = "invalid"82 return errors.New("callback-delivery: invalid payload")83 }84 if payload.Attempt < 1 {85 payload.Attempt = 186 }87 if err := event.Validate(payload.Event); err != nil {88 outcome = "invalid"89 return errors.New("callback-delivery: invalid event")90 }91 lock := s.callbackLock(intent.Target)92 lock.Lock()93 defer lock.Unlock()94 record, err := s.callback(ctx, intent.Target)95 if err != nil {96 if strings.HasPrefix(err.Error(), "not found:") {97 outcome = "paused"98 return nil99 }100 return err101 }102 if record.Paused {103 outcome = "paused"104 return nil105 }106 source := payload.Event107 if !event.IsEphemeral(payload.Event.Kind) {108 var raw string109 err := s.store.DB().QueryRowContext(ctx, `SELECT raw FROM events WHERE id=? AND (expires=0 OR expires>?)`, payload.Event.ID, time.Now().Unix()).Scan(&raw)110 if errors.Is(err, sql.ErrNoRows) {111 outcome = "unauthorized"112 return nil113 }114 if err != nil {115 return err116 }117 parsed, err := event.Parse([]byte(raw))118 if err != nil {119 return err120 }121 source = parsed122 }123 role, err := s.community.Role(ctx, record.Owner)124 if err != nil {125 return err126 }127 if role == "" && record.Owner != s.policy().Owner {128 // The key lost its standing since it registered; stop until a129 // person resumes the callback.130 outcome = "unauthorized"131 return s.pauseCallback(ctx, record.ID, record.Failures, "paused: the key is no longer a member")132 }133 if role == "agent" {134 grant, ok, err := s.community.AgentGrant(ctx, record.Owner)135 if err != nil {136 return err137 }138 if !ok || !grant.Active(time.Now().Unix()) {139 outcome = "unauthorized"140 return s.pauseCallback(ctx, record.ID, record.Failures, "paused: the agent grant is no longer active")141 }142 if err := s.checkCallbackScope(ctx, record.Owner, record.Filter); err != nil {143 outcome = "unauthorized"144 return s.pauseCallback(ctx, record.ID, record.Failures, "paused: the agent grant no longer covers this callback")145 }146 }147 if !s.gate.CanSee(ctx, source, relay.Session{PubKeys: []string{record.Owner}, RelayURL: s.relayURL}, nil) {148 outcome = "unauthorized"149 return nil150 }151 status, postErr := s.postCallback(ctx, record, source)152 now := time.Now().Unix()153 if postErr == nil {154 outcome = "ok"155 _, err = s.store.DB().ExecContext(ctx, `UPDATE callbacks SET last_delivery_at=?, last_status='ok', failures=0 WHERE id=?`, now, record.ID)156 s.telemetry.Logger().Info("callback delivered", "tenant", s.tenant.Name, "attempt", payload.Attempt, "status", status)157 return err158 }159 // Counts only: the reason names the HTTP status or the failure class,160 // never the URL.161 reason := postErr.Error()162 failures := record.Failures + 1163 paused := failures >= callbackPauseFailures164 if paused {165 outcome = "paused"166 if err := s.pauseCallback(ctx, record.ID, failures, fmt.Sprintf("paused after %d failures: %s", failures, reason)); err != nil {167 return err168 }169 } else if _, err := s.store.DB().ExecContext(ctx, `UPDATE callbacks SET failures=?, last_status=? WHERE id=?`, failures, reason, record.ID); err != nil {170 return err171 }172 s.telemetry.Logger().Info("callback delivery failed", "tenant", s.tenant.Name, "attempt", payload.Attempt, "status", status, "failures", failures, "paused", paused)173 if paused || payload.Attempt >= callbackMaxAttempts {174 return nil175 }176 return s.retryCallback(ctx, intent, payload)177}178179// pauseCallback stops deliveries and records the count and reason, then180// drops the index so no new intents are queued for the callback.181func (s *callbackService) pauseCallback(ctx context.Context, id string, failures int, status string) error {182 if _, err := s.store.DB().ExecContext(ctx, `UPDATE callbacks SET paused=1, failures=?, last_status=? WHERE id=?`, failures, status, id); err != nil {183 return err184 }185 s.invalidateCallbacks()186 return nil187}188189// retryCallback queues the next attempt after its backoff. Each attempt is190// its own intent so the durable queue's ordering and fencing still apply.191func (s *callbackService) retryCallback(ctx context.Context, intent work.Intent, payload callbackPayload) error {192 delay := callbackBackoff[len(callbackBackoff)-1]193 if payload.Attempt-1 < len(callbackBackoff) {194 delay = callbackBackoff[payload.Attempt-1]195 }196 payload.Attempt++197 encoded, err := json.Marshal(payload)198 if err != nil {199 return err200 }201 now := time.Now().Unix()202 key := sha256.Sum256([]byte(callbackDelivery + "\x00" + payload.Event.ID + "\x00" + intent.Target + "\x00" + fmt.Sprint(payload.Attempt)))203 return s.store.WithTx(ctx, func(tx *sql.Tx) error {204 _, err := tx.ExecContext(ctx, `INSERT OR IGNORE INTO work_intents(id,kind,event_id,target,payload,next_at,created_at,updated_at) VALUES(?,?,?,?,?,?,?,?)`, hex.EncodeToString(key[:]), callbackDelivery, payload.Event.ID+"#"+fmt.Sprint(payload.Attempt), intent.Target, string(encoded), now+int64(delay/time.Second), now, now)205 return err206 })207}208209// postCallback sends the event and returns the HTTP status with a bounded210// reason on failure.211func (s *callbackService) postCallback(ctx context.Context, record callbackRecord, e event.Event) (int, error) {212 body, err := json.Marshal(e)213 if err != nil {214 return 0, errors.New("encode event")215 }216 ctx, cancel := context.WithTimeout(ctx, callbackTimeout)217 defer cancel()218 request, err := http.NewRequestWithContext(ctx, http.MethodPost, record.URL, strings.NewReader(string(body)))219 if err != nil {220 return 0, errors.New("bad url")221 }222 request.Header.Set("Content-Type", "application/json")223 request.Header.Set("X-Tiny-Callback", record.ID)224 request.Header.Set("X-Tiny-Signature", callbackSignature(record.secret, body))225 request.Header.Set("X-Tiny-Relay", s.publicURL)226 request.Header.Set("User-Agent", "tinyrelay")227 response, err := s.callbackHTTPClient().Do(request)228 if err != nil {229 if errors.Is(err, context.DeadlineExceeded) || ctx.Err() != nil {230 return 0, errors.New("timeout")231 }232 return 0, errors.New("connection failed")233 }234 defer response.Body.Close()235 _, _ = io.Copy(io.Discard, io.LimitReader(response.Body, 4096))236 if response.StatusCode < 200 || response.StatusCode >= 300 {237 return response.StatusCode, fmt.Errorf("HTTP %d", response.StatusCode)238 }239 return response.StatusCode, nil240}241242func (t *Tenant) handleCallbackDelivery(ctx context.Context, intent work.Intent) error {243 return t.callbacks.handleCallbackDelivery(ctx, intent)244}