| 1 | package daemon |
| 2 | |
| 3 | // 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 one |
| 5 | // minute and then after five; after 20 failures in a row the callback is |
| 6 | // paused until its owner resumes it. |
| 7 | |
| 8 | import ( |
| 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" |
| 23 | |
| 24 | "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 | ) |
| 29 | |
| 30 | // callbackPayload carries an event and its delivery attempt. Stored events |
| 31 | // are reloaded before delivery; ephemeral events use this signed payload. |
| 32 | type callbackPayload struct { |
| 33 | Event event.Event `json:"event"` |
| 34 | Attempt int `json:"attempt"` |
| 35 | } |
| 36 | |
| 37 | // callbackSignature is the X-Tiny-Signature value for a body: sha256= and |
| 38 | // the hex HMAC-SHA256 of the body under the callback secret. |
| 39 | func 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 | } |
| 44 | |
| 45 | // callbackHTTPClient refuses private addresses and redirects unless the |
| 46 | // relay itself runs on a loopback address. |
| 47 | func (s *callbackService) callbackHTTPClient() *http.Client { |
| 48 | if s.callbackClient != nil { |
| 49 | return s.callbackClient |
| 50 | } |
| 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 | } |
| 56 | |
| 57 | // callbackLock bounds delivery to one POST per callback at a time. |
| 58 | func (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] = lock |
| 68 | } |
| 69 | return lock |
| 70 | } |
| 71 | |
| 72 | // handleCallbackDelivery serves one callback-delivery intent. It rechecks |
| 73 | // the callback and the key's access right before the POST, so a callback |
| 74 | // removed or paused after the event arrived delivers nothing. |
| 75 | func (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 callbackPayload |
| 80 | 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 = 1 |
| 86 | } |
| 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 nil |
| 99 | } |
| 100 | return err |
| 101 | } |
| 102 | if record.Paused { |
| 103 | outcome = "paused" |
| 104 | return nil |
| 105 | } |
| 106 | source := payload.Event |
| 107 | if !event.IsEphemeral(payload.Event.Kind) { |
| 108 | var raw string |
| 109 | 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 nil |
| 113 | } |
| 114 | if err != nil { |
| 115 | return err |
| 116 | } |
| 117 | parsed, err := event.Parse([]byte(raw)) |
| 118 | if err != nil { |
| 119 | return err |
| 120 | } |
| 121 | source = parsed |
| 122 | } |
| 123 | role, err := s.community.Role(ctx, record.Owner) |
| 124 | if err != nil { |
| 125 | return err |
| 126 | } |
| 127 | if role == "" && record.Owner != s.policy().Owner { |
| 128 | // The key lost its standing since it registered; stop until a |
| 129 | // 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 err |
| 137 | } |
| 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 nil |
| 150 | } |
| 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 err |
| 158 | } |
| 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 + 1 |
| 163 | paused := failures >= callbackPauseFailures |
| 164 | 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 err |
| 168 | } |
| 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 err |
| 171 | } |
| 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 nil |
| 175 | } |
| 176 | return s.retryCallback(ctx, intent, payload) |
| 177 | } |
| 178 | |
| 179 | // pauseCallback stops deliveries and records the count and reason, then |
| 180 | // drops the index so no new intents are queued for the callback. |
| 181 | func (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 err |
| 184 | } |
| 185 | s.invalidateCallbacks() |
| 186 | return nil |
| 187 | } |
| 188 | |
| 189 | // retryCallback queues the next attempt after its backoff. Each attempt is |
| 190 | // its own intent so the durable queue's ordering and fencing still apply. |
| 191 | func (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 err |
| 200 | } |
| 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 err |
| 206 | }) |
| 207 | } |
| 208 | |
| 209 | // postCallback sends the event and returns the HTTP status with a bounded |
| 210 | // reason on failure. |
| 211 | func (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, nil |
| 240 | } |
| 241 | |
| 242 | func (t *Tenant) handleCallbackDelivery(ctx context.Context, intent work.Intent) error { |
| 243 | return t.callbacks.handleCallbackDelivery(ctx, intent) |
| 244 | } |