| 1 | diff --git a/docs/agents.md b/docs/agents.md |
| 2 | index ecd5a1c..abb3fb4 100644 |
| 3 | --- a/docs/agents.md |
| 4 | +++ b/docs/agents.md |
| 5 | @@ -354,7 +354,7 @@ export function verify(body, header, secret) { |
| 6 | } |
| 7 | ``` |
| 8 | |
| 9 | -Before each POST, the relay checks that the callback exists and is not paused, that its owner still has membership and that the read gate permits the queued event. Delivery does not separately recheck whether an agent grant is active or reload the source event. Pause the callback itself when delivery must stop; pausing a grant or deleting an event is not a guaranteed cancellation of already queued delivery. An in-flight request cannot be recalled. |
| 10 | +Before each POST, the relay checks the callback, its owner's membership and current read access. An agent must still have an active grant covering the callback's room and repository filters. Pausing, revoking or expiring its grant stops further delivery and pauses the callback. Stored events must still exist and be unexpired; deleting or hiding one stops its queued delivery. Ephemeral events travel in the queue because they have no stored source. An in-flight request cannot be recalled. |
| 11 | |
| 12 | Each callback has at most one in-process delivery at a time. Retries can repeat delivery, so receivers must handle events idempotently. A repository state is delivered once its objects have arrived. See the [callback contract](extensions/callbacks.md) for exact payloads, management responses, retries and audited limits. |
| 13 | |
| 14 | diff --git a/docs/extensions/callbacks.md b/docs/extensions/callbacks.md |
| 15 | index 394a6f7..282762a 100644 |
| 16 | --- a/docs/extensions/callbacks.md |
| 17 | +++ b/docs/extensions/callbacks.md |
| 18 | @@ -30,7 +30,7 @@ Members and agents manage their own registrations. Owner/moderator roles may lis |
| 19 | |
| 20 | A supplied nonempty secret must be 16 to 128 bytes of ASCII `!` through `~`: embedded spaces are rejected despite the error's “printable ASCII” wording. An omitted/blank secret generates 32 random bytes encoded as 64 lowercase hex characters. Registration IDs are 16 random bytes encoded as 32 lowercase hex characters. Treat an ID as an opaque registration identifier, not authentication. |
| 21 | |
| 22 | -Registration returns a summary plus `secret` once. Summary fields are `id`, `owner`, `url`, `host`, `filter`, `created`, `lastDelivery`, `lastStatus`, `failures` and `paused`. Times are Unix seconds. `host` includes an explicit port. List responses omit `url` for registrations owned by someone other than the caller, and never include secrets. Pause/resume return a summary; removal returns `{"id":"...","removed":true}`. **Audit gap:** pause/resume currently return `record.summary()` without the listing's cross-owner URL redaction. Do not interpret the redacted list as a guarantee that an operator cannot retrieve a path/query through a mutation response. |
| 23 | +Registration returns a summary plus `secret` once. Summary fields are `id`, `owner`, `url`, `host`, `filter`, `created`, `lastDelivery`, `lastStatus`, `failures` and `paused`. Times are Unix seconds. `host` includes an explicit port. List, pause and resume responses omit `url` for registrations owned by someone other than the caller and never include secrets. Removal returns `{"id":"...","removed":true}`. |
| 24 | |
| 25 | ## Filter contract |
| 26 | |
| 27 | @@ -69,14 +69,14 @@ The receiver acknowledges with any 2xx status within 10 seconds. Response conten |
| 28 | |
| 29 | - Stored event acceptance captures eligible registrations for later work. Optional local planning failure does not reject a valid source event. Recovery is bounded by a 15-minute age limit and three background worker planning attempts, separately from delivery retries. Successfully planned child work retains its identity. Failed discovery recovery excludes registrations created at or after the acceptance second, so uncertain same-second registrations can be skipped. |
| 30 | - Repository-state callbacks are planned after object promotion, rather than waking a receiver while objects are missing. Ephemeral events use a separate immediate planning path whose intent carries the event; they do not gain durable source history or the stored-event planning recovery guarantee. |
| 31 | -- Before each POST the service rereads the registration, skips missing/paused registrations, checks current membership and applies `Gate.CanSee` to the queued event. A key with no remaining membership has its callback paused. Gate checks include bans, held events, host read policy, rooms and recipients. |
| 32 | +- Before each POST the service rereads the registration and skips missing or paused callbacks. It checks current membership and, for agents, requires an active grant covering the callback's room and repository filters. Losing membership or that grant coverage pauses the callback. Stored events are reloaded and must still exist, be unexpired and pass current read checks, including bans, held events, host read policy, rooms and recipients. Ephemeral events use their signed kind to identify the queued payload as their source. |
| 33 | - Each callback has at most one in-process POST at a time. A failed event gets two further scheduled attempts, after one minute and then five minutes. The third is final. Each failed HTTP attempt increments consecutive failures; success resets them. At 20 failures the callback pauses. Resumption does not promise replay of skipped or exhausted events. |
| 34 | |
| 35 | Outbound URLs use the shared [`NewPinnedClient`](../../internal/replication/callback.go), which dials resolved public addresses directly and refuses redirects. The lexical registration check rejects private/loopback IPs and local-style hostnames. A relay configured with a private/loopback public URL deliberately relaxes private-target restrictions for local use; HTTPS and redirect refusal remain. This is a deployment trust switch, not an exemption clients can request. DNS filtering is the implementation's explicit address-deny list, not a claim of a complete SSRF audit. |
| 36 | |
| 37 | Callbacks intentionally disclose the entire readable signed event to an external service. Secrets and URL tokens require protection in storage and at the receiver. Revocation cannot recall an already sent request. |
| 38 | |
| 39 | -**Read-state limits:** Delivery does not rerun `checkCallbackScope` or `AgentGrant.Active`; `Gate.CanSee` is a read gate, not a write-grant check. Paused/expired agents can retain membership/read access, so pausing a grant alone is not a guaranteed webhook kill switch. Pause the callback to stop it. Delivery checks the queued event payload, not a fresh source lookup; deletion/expiration alone is not a guaranteed cancellation of already planned deliveries. Do not broaden the guide's read-recheck wording into guarantees the implementation does not enforce. |
| 40 | +Restoring a grant does not automatically resume a paused callback. Resume it after its membership and grant scope are valid. Delivery checks apply to each attempt; an already dispatched request cannot be recalled. |
| 41 | |
| 42 | ## Errors, discovery and evidence |
| 43 | |
| 44 | @@ -89,5 +89,3 @@ Evidence: |
| 45 | - [`callbacks_test.go`](../../internal/daemon/callbacks_test.go): `TestCallbackRegistrationValidatesURLFilterScopeAndCap`, `TestCallbackManagementPermissionMatrix`, `TestCallbackMatchesFilterAndKeepsPrivateEventsBehindTheGate`, `TestCallbackDeliverySignsBodyRetriesAndPauses`, `TestCallbackDeliveryRechecksTheGateAndMembership`, `TestMCPCallbackTools`. |
| 46 | - [`callback_followups_test.go`](../../internal/daemon/callback_followups_test.go): planning failure, preserved completed work, paused/removed/later registrations, exhaustion and independent view failure. |
| 47 | - [`replication/callback.go`](../../internal/replication/callback.go) supplies the shared pinned HTTP client; this audit found no direct pinned-HTTP-client test in the replication test files. The callback restart test in [`callback_restart_acceptance_test.go`](../../internal/daemon/callback_restart_acceptance_test.go) exercises **replication** callback work, not proof of this HMAC webhook's end-to-end restart delivery. |
| 48 | - |
| 49 | -The URL-redaction, grant-state and queued-source limits above are unresolved implementation boundaries. This specification neither fixes them nor claims tests for stronger guarantees. |
| 50 | diff --git a/internal/daemon/callbacks.go b/internal/daemon/callbacks.go |
| 51 | index f6cf681..c18b34e 100644 |
| 52 | --- a/internal/daemon/callbacks.go |
| 53 | +++ b/internal/daemon/callbacks.go |
| 54 | @@ -446,7 +446,11 @@ func (s *callbackService) changeCallback(ctx context.Context, actor string, oper |
| 55 | if method == "removecallback" { |
| 56 | return map[string]any{"id": id, "removed": true}, nil |
| 57 | } |
| 58 | - return record.summary(), nil |
| 59 | + result := record.summary() |
| 60 | + if operator && record.Owner != actor { |
| 61 | + delete(result, "url") |
| 62 | + } |
| 63 | + return result, nil |
| 64 | } |
| 65 | |
| 66 | // callbackIDParam reads the id from a bare string or from {"id": ...}. |
| 67 | diff --git a/internal/daemon/callbacks_delivery.go b/internal/daemon/callbacks_delivery.go |
| 68 | index 95aed6b..3408b67 100644 |
| 69 | --- a/internal/daemon/callbacks_delivery.go |
| 70 | +++ b/internal/daemon/callbacks_delivery.go |
| 71 | @@ -27,8 +27,8 @@ import ( |
| 72 | "github.com/FelineStateMachine/tinyrelay/internal/work" |
| 73 | ) |
| 74 | |
| 75 | -// callbackPayload is the work intent body: the event itself, so delivery |
| 76 | -// does not depend on the event still being stored, and the attempt number. |
| 77 | +// callbackPayload carries an event and its delivery attempt. Stored events |
| 78 | +// are reloaded before delivery; ephemeral events use this signed payload. |
| 79 | type callbackPayload struct { |
| 80 | Event event.Event `json:"event"` |
| 81 | Attempt int `json:"attempt"` |
| 82 | @@ -84,6 +84,10 @@ func (s *callbackService) handleCallbackDelivery(ctx context.Context, intent wor |
| 83 | if payload.Attempt < 1 { |
| 84 | payload.Attempt = 1 |
| 85 | } |
| 86 | + if err := event.Validate(payload.Event); err != nil { |
| 87 | + outcome = "invalid" |
| 88 | + return errors.New("callback-delivery: invalid event") |
| 89 | + } |
| 90 | lock := s.callbackLock(intent.Target) |
| 91 | lock.Lock() |
| 92 | defer lock.Unlock() |
| 93 | @@ -99,6 +103,23 @@ func (s *callbackService) handleCallbackDelivery(ctx context.Context, intent wor |
| 94 | outcome = "paused" |
| 95 | return nil |
| 96 | } |
| 97 | + source := payload.Event |
| 98 | + if !event.IsEphemeral(payload.Event.Kind) { |
| 99 | + var raw string |
| 100 | + 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) |
| 101 | + if errors.Is(err, sql.ErrNoRows) { |
| 102 | + outcome = "unauthorized" |
| 103 | + return nil |
| 104 | + } |
| 105 | + if err != nil { |
| 106 | + return err |
| 107 | + } |
| 108 | + parsed, err := event.Parse([]byte(raw)) |
| 109 | + if err != nil { |
| 110 | + return err |
| 111 | + } |
| 112 | + source = parsed |
| 113 | + } |
| 114 | role, err := s.community.Role(ctx, record.Owner) |
| 115 | if err != nil { |
| 116 | return err |
| 117 | @@ -109,11 +130,25 @@ func (s *callbackService) handleCallbackDelivery(ctx context.Context, intent wor |
| 118 | outcome = "unauthorized" |
| 119 | return s.pauseCallback(ctx, record.ID, record.Failures, "paused: the key is no longer a member") |
| 120 | } |
| 121 | - if !s.gate.CanSee(ctx, payload.Event, relay.Session{PubKeys: []string{record.Owner}, RelayURL: s.relayURL}, nil) { |
| 122 | + if role == "agent" { |
| 123 | + grant, ok, err := s.community.AgentGrant(ctx, record.Owner) |
| 124 | + if err != nil { |
| 125 | + return err |
| 126 | + } |
| 127 | + if !ok || !grant.Active(time.Now().Unix()) { |
| 128 | + outcome = "unauthorized" |
| 129 | + return s.pauseCallback(ctx, record.ID, record.Failures, "paused: the agent grant is no longer active") |
| 130 | + } |
| 131 | + if err := s.checkCallbackScope(ctx, record.Owner, record.Filter); err != nil { |
| 132 | + outcome = "unauthorized" |
| 133 | + return s.pauseCallback(ctx, record.ID, record.Failures, "paused: the agent grant no longer covers this callback") |
| 134 | + } |
| 135 | + } |
| 136 | + if !s.gate.CanSee(ctx, source, relay.Session{PubKeys: []string{record.Owner}, RelayURL: s.relayURL}, nil) { |
| 137 | outcome = "unauthorized" |
| 138 | return nil |
| 139 | } |
| 140 | - status, postErr := s.postCallback(ctx, record, payload.Event) |
| 141 | + status, postErr := s.postCallback(ctx, record, source) |
| 142 | now := time.Now().Unix() |
| 143 | if postErr == nil { |
| 144 | outcome = "ok" |
| 145 | diff --git a/internal/daemon/callbacks_test.go b/internal/daemon/callbacks_test.go |
| 146 | index c1ecd51..ec64c78 100644 |
| 147 | --- a/internal/daemon/callbacks_test.go |
| 148 | +++ b/internal/daemon/callbacks_test.go |
| 149 | @@ -246,12 +246,20 @@ func TestCallbackManagementPermissionMatrix(t *testing.T) { |
| 150 | if _, present := byOwner[member]["url"]; present { |
| 151 | t.Fatalf("operator listing carries another key's URL path: %v", byOwner[member]) |
| 152 | } |
| 153 | - if _, err := tenant.Execute(ctx, actor, "pausecallback", []json.RawMessage{id(theirs)}); err != nil { |
| 154 | + paused, err := tenant.Execute(ctx, actor, "pausecallback", []json.RawMessage{id(theirs)}) |
| 155 | + if err != nil { |
| 156 | t.Fatalf("pause by %s: %v", actor, err) |
| 157 | } |
| 158 | - if _, err := tenant.Execute(ctx, actor, "resumecallback", []json.RawMessage{id(theirs)}); err != nil { |
| 159 | + if _, present := paused.(map[string]any)["url"]; present { |
| 160 | + t.Fatalf("pause response carries another key's URL path: %v", paused) |
| 161 | + } |
| 162 | + resumed, err := tenant.Execute(ctx, actor, "resumecallback", []json.RawMessage{id(theirs)}) |
| 163 | + if err != nil { |
| 164 | t.Fatalf("resume by %s: %v", actor, err) |
| 165 | } |
| 166 | + if _, present := resumed.(map[string]any)["url"]; present { |
| 167 | + t.Fatalf("resume response carries another key's URL path: %v", resumed) |
| 168 | + } |
| 169 | } |
| 170 | if _, err := tenant.Execute(ctx, owner, "pausecallback", []json.RawMessage{json.RawMessage(`"missing"`)}); err == nil || !strings.HasPrefix(err.Error(), "not found:") { |
| 171 | t.Fatalf("unknown id: %v", err) |
| 172 | @@ -594,6 +602,283 @@ func TestCallbackDeliveryRechecksTheGateAndMembership(t *testing.T) { |
| 173 | } |
| 174 | } |
| 175 | |
| 176 | +func TestCallbackDeliveryStopsForInactiveAgentGrant(t *testing.T) { |
| 177 | + _, tenant := loopbackTenant(t) |
| 178 | + ctx := context.Background() |
| 179 | + owner := tenant.Policy().Owner |
| 180 | + agent, _ := event.PublicKey(testAgentSecret) |
| 181 | + now := time.Now().Unix() |
| 182 | + grant := agentGrantEvent(t, agent, now-1, now+3600, []string{"k", "1"}) |
| 183 | + if err := publishAs(t, tenant, grant); err != nil { |
| 184 | + t.Fatal(err) |
| 185 | + } |
| 186 | + receiver := &callbackReceiver{status: http.StatusOK} |
| 187 | + service := httptest.NewTLSServer(receiver) |
| 188 | + defer service.Close() |
| 189 | + tenant.callbacks.callbackClient = service.Client() |
| 190 | + registered := addCallback(t, tenant, agent, service.URL+"/wake", `{"kinds":[1]}`) |
| 191 | + id := registered["id"].(string) |
| 192 | + note := signedEvent(t, testOwnerSecret, 1, now, nil, "grant pause") |
| 193 | + if err := publishAs(t, tenant, note); err != nil { |
| 194 | + t.Fatal(err) |
| 195 | + } |
| 196 | + queued := pendingCallbackIntents(t, tenant, id, note.ID) |
| 197 | + if len(queued) != 1 { |
| 198 | + t.Fatalf("queued = %+v", queued) |
| 199 | + } |
| 200 | + if _, err := tenant.Execute(ctx, owner, "pauseagent", []json.RawMessage{json.RawMessage(strconv.Quote(agent))}); err != nil { |
| 201 | + t.Fatal(err) |
| 202 | + } |
| 203 | + if err := tenant.handleCallbackDelivery(ctx, queued[0]); err != nil { |
| 204 | + t.Fatal(err) |
| 205 | + } |
| 206 | + if receiver.count() != 0 { |
| 207 | + t.Fatal("paused agent callback delivered") |
| 208 | + } |
| 209 | + var paused int |
| 210 | + if err := tenant.store.DB().QueryRowContext(ctx, "SELECT paused FROM callbacks WHERE id=?", id).Scan(&paused); err != nil || paused != 1 { |
| 211 | + t.Fatalf("inactive grant did not pause callback: paused=%d err=%v", paused, err) |
| 212 | + } |
| 213 | +} |
| 214 | + |
| 215 | +func TestCallbackDeliverySkipsDeletedSource(t *testing.T) { |
| 216 | + _, tenant := loopbackTenant(t) |
| 217 | + ctx := context.Background() |
| 218 | + member, _ := event.PublicKey(testMemberSecret) |
| 219 | + setRole(t, tenant, member, "member") |
| 220 | + receiver := &callbackReceiver{status: http.StatusOK} |
| 221 | + service := httptest.NewTLSServer(receiver) |
| 222 | + defer service.Close() |
| 223 | + tenant.callbacks.callbackClient = service.Client() |
| 224 | + registered := addCallback(t, tenant, member, service.URL+"/wake", `{"kinds":[1]}`) |
| 225 | + id := registered["id"].(string) |
| 226 | + note := signedEvent(t, testOwnerSecret, 1, time.Now().Unix(), nil, "deleted source") |
| 227 | + if err := publishAs(t, tenant, note); err != nil { |
| 228 | + t.Fatal(err) |
| 229 | + } |
| 230 | + queued := pendingCallbackIntents(t, tenant, id, note.ID) |
| 231 | + if len(queued) != 1 { |
| 232 | + t.Fatalf("queued = %+v", queued) |
| 233 | + } |
| 234 | + if _, err := tenant.store.DB().ExecContext(ctx, "DELETE FROM events WHERE id=?", note.ID); err != nil { |
| 235 | + t.Fatal(err) |
| 236 | + } |
| 237 | + if err := tenant.handleCallbackDelivery(ctx, queued[0]); err != nil { |
| 238 | + t.Fatal(err) |
| 239 | + } |
| 240 | + if receiver.count() != 0 { |
| 241 | + t.Fatal("deleted source callback delivered") |
| 242 | + } |
| 243 | +} |
| 244 | + |
| 245 | +func TestCallbackDeliverySkipsExpiredSource(t *testing.T) { |
| 246 | + _, tenant := loopbackTenant(t) |
| 247 | + ctx := context.Background() |
| 248 | + member, _ := event.PublicKey(testMemberSecret) |
| 249 | + setRole(t, tenant, member, "member") |
| 250 | + receiver := &callbackReceiver{status: http.StatusOK} |
| 251 | + service := httptest.NewTLSServer(receiver) |
| 252 | + defer service.Close() |
| 253 | + tenant.callbacks.callbackClient = service.Client() |
| 254 | + registered := addCallback(t, tenant, member, service.URL+"/wake", `{"kinds":[1]}`) |
| 255 | + id := registered["id"].(string) |
| 256 | + note := signedEvent(t, testOwnerSecret, 1, time.Now().Unix(), nil, "expired source") |
| 257 | + if err := publishAs(t, tenant, note); err != nil { |
| 258 | + t.Fatal(err) |
| 259 | + } |
| 260 | + queued := pendingCallbackIntents(t, tenant, id, note.ID) |
| 261 | + if len(queued) != 1 { |
| 262 | + t.Fatalf("queued = %+v", queued) |
| 263 | + } |
| 264 | + if _, err := tenant.store.DB().ExecContext(ctx, "UPDATE events SET expires=? WHERE id=?", time.Now().Unix()-1, note.ID); err != nil { |
| 265 | + t.Fatal(err) |
| 266 | + } |
| 267 | + if err := tenant.handleCallbackDelivery(ctx, queued[0]); err != nil { |
| 268 | + t.Fatal(err) |
| 269 | + } |
| 270 | + if receiver.count() != 0 { |
| 271 | + t.Fatal("expired source callback delivered") |
| 272 | + } |
| 273 | +} |
| 274 | + |
| 275 | +func TestCallbackDeliveryRejectsForgedEphemeralProvenance(t *testing.T) { |
| 276 | + _, tenant := loopbackTenant(t) |
| 277 | + ctx := context.Background() |
| 278 | + member, _ := event.PublicKey(testMemberSecret) |
| 279 | + setRole(t, tenant, member, "member") |
| 280 | + receiver := &callbackReceiver{status: http.StatusOK} |
| 281 | + service := httptest.NewTLSServer(receiver) |
| 282 | + defer service.Close() |
| 283 | + tenant.callbacks.callbackClient = service.Client() |
| 284 | + registered := addCallback(t, tenant, member, service.URL+"/wake", `{"kinds":[1]}`) |
| 285 | + id := registered["id"].(string) |
| 286 | + note := signedEvent(t, testOwnerSecret, 1, time.Now().Unix(), nil, "normal event") |
| 287 | + if err := publishAs(t, tenant, note); err != nil { |
| 288 | + t.Fatal(err) |
| 289 | + } |
| 290 | + queued := pendingCallbackIntents(t, tenant, id, note.ID) |
| 291 | + if len(queued) != 1 { |
| 292 | + t.Fatalf("queued = %+v", queued) |
| 293 | + } |
| 294 | + var payload map[string]json.RawMessage |
| 295 | + if err := json.Unmarshal([]byte(queued[0].Payload), &payload); err != nil { |
| 296 | + t.Fatal(err) |
| 297 | + } |
| 298 | + payload["ephemeral"] = json.RawMessage("true") |
| 299 | + encoded, err := json.Marshal(payload) |
| 300 | + if err != nil { |
| 301 | + t.Fatal(err) |
| 302 | + } |
| 303 | + if _, err := tenant.store.DB().ExecContext(ctx, "UPDATE work_intents SET payload=? WHERE id=?", string(encoded), queued[0].ID); err != nil { |
| 304 | + t.Fatal(err) |
| 305 | + } |
| 306 | + if _, err := tenant.store.DB().ExecContext(ctx, "DELETE FROM events WHERE id=?", note.ID); err != nil { |
| 307 | + t.Fatal(err) |
| 308 | + } |
| 309 | + queued[0].Payload = string(encoded) |
| 310 | + if err := tenant.handleCallbackDelivery(ctx, queued[0]); err != nil || receiver.count() != 0 { |
| 311 | + t.Fatalf("forged ephemeral provenance: err=%v deliveries=%d", err, receiver.count()) |
| 312 | + } |
| 313 | +} |
| 314 | + |
| 315 | +func TestCallbackDeliveryStopsWhenAgentScopeNarrows(t *testing.T) { |
| 316 | + _, tenant := loopbackTenant(t) |
| 317 | + ctx := context.Background() |
| 318 | + agent, _ := event.PublicKey(testAgentSecret) |
| 319 | + owner := tenant.Policy().Owner |
| 320 | + now := time.Now().Unix() |
| 321 | + grant := agentGrantEvent(t, agent, now-1, now+3600, []string{"k", "1"}, []string{"repo", owner + ":repo:read"}) |
| 322 | + if err := publishAs(t, tenant, grant); err != nil { |
| 323 | + t.Fatal(err) |
| 324 | + } |
| 325 | + receiver := &callbackReceiver{status: http.StatusOK} |
| 326 | + service := httptest.NewTLSServer(receiver) |
| 327 | + defer service.Close() |
| 328 | + tenant.callbacks.callbackClient = service.Client() |
| 329 | + filter := `{"kinds":[1],"#a":["30617:` + owner + `:repo"]}` |
| 330 | + registered := addCallback(t, tenant, agent, service.URL+"/wake", filter) |
| 331 | + id := registered["id"].(string) |
| 332 | + note := signedEvent(t, testOwnerSecret, 1, now, [][]string{{"a", "30617:" + owner + ":repo"}}, "narrowed scope") |
| 333 | + if err := publishAs(t, tenant, note); err != nil { |
| 334 | + t.Fatal(err) |
| 335 | + } |
| 336 | + queued := pendingCallbackIntents(t, tenant, id, note.ID) |
| 337 | + if len(queued) != 1 { |
| 338 | + t.Fatalf("queued = %+v", queued) |
| 339 | + } |
| 340 | + if _, err := tenant.store.DB().ExecContext(ctx, `UPDATE agent_grants SET scope=? WHERE agent=?`, `{"kinds":[1],"rooms":[],"repos":[],"rate":60}`, agent); err != nil { |
| 341 | + t.Fatal(err) |
| 342 | + } |
| 343 | + if err := tenant.handleCallbackDelivery(ctx, queued[0]); err != nil { |
| 344 | + t.Fatal(err) |
| 345 | + } |
| 346 | + if receiver.count() != 0 { |
| 347 | + t.Fatal("scope-narrowed agent callback delivered") |
| 348 | + } |
| 349 | + var paused int |
| 350 | + if err := tenant.store.DB().QueryRowContext(ctx, "SELECT paused FROM callbacks WHERE id=?", id).Scan(&paused); err != nil || paused != 1 { |
| 351 | + t.Fatalf("scope-narrowed callback not paused: paused=%d err=%v", paused, err) |
| 352 | + } |
| 353 | +} |
| 354 | + |
| 355 | +func TestCallbackDeliveryAllowsActiveAgentWithStoredSource(t *testing.T) { |
| 356 | + _, tenant := loopbackTenant(t) |
| 357 | + ctx := context.Background() |
| 358 | + agent, _ := event.PublicKey(testAgentSecret) |
| 359 | + now := time.Now().Unix() |
| 360 | + grant := agentGrantEvent(t, agent, now-1, now+3600, []string{"k", "1"}) |
| 361 | + if err := publishAs(t, tenant, grant); err != nil { |
| 362 | + t.Fatal(err) |
| 363 | + } |
| 364 | + receiver := &callbackReceiver{status: http.StatusOK} |
| 365 | + service := httptest.NewTLSServer(receiver) |
| 366 | + defer service.Close() |
| 367 | + tenant.callbacks.callbackClient = service.Client() |
| 368 | + registered := addCallback(t, tenant, agent, service.URL+"/wake", `{"kinds":[1]}`) |
| 369 | + id := registered["id"].(string) |
| 370 | + note := signedEvent(t, testOwnerSecret, 1, now, nil, "active grant") |
| 371 | + if err := publishAs(t, tenant, note); err != nil { |
| 372 | + t.Fatal(err) |
| 373 | + } |
| 374 | + queued := pendingCallbackIntents(t, tenant, id, note.ID) |
| 375 | + if len(queued) != 1 { |
| 376 | + t.Fatalf("queued = %+v", queued) |
| 377 | + } |
| 378 | + if err := tenant.handleCallbackDelivery(ctx, queued[0]); err != nil { |
| 379 | + t.Fatal(err) |
| 380 | + } |
| 381 | + if receiver.count() != 1 { |
| 382 | + t.Fatalf("active agent callback deliveries=%d, want 1", receiver.count()) |
| 383 | + } |
| 384 | +} |
| 385 | + |
| 386 | +func TestEphemeralCallbackDeliveryUsesKindProvenance(t *testing.T) { |
| 387 | + _, tenant := loopbackTenant(t) |
| 388 | + ctx := context.Background() |
| 389 | + member, _ := event.PublicKey(testMemberSecret) |
| 390 | + setRole(t, tenant, member, "member") |
| 391 | + receiver := &callbackReceiver{status: http.StatusOK} |
| 392 | + service := httptest.NewTLSServer(receiver) |
| 393 | + defer service.Close() |
| 394 | + tenant.callbacks.callbackClient = service.Client() |
| 395 | + registered := addCallback(t, tenant, member, service.URL+"/wake", `{"kinds":[20001]}`) |
| 396 | + id := registered["id"].(string) |
| 397 | + presence := signedEvent(t, testOwnerSecret, 20001, time.Now().Unix(), nil, "ephemeral") |
| 398 | + tenant.notifyCallbacks(ctx, presence) |
| 399 | + queued := pendingCallbackIntents(t, tenant, id, presence.ID) |
| 400 | + if len(queued) != 1 { |
| 401 | + t.Fatalf("queued = %+v", queued) |
| 402 | + } |
| 403 | + if err := tenant.handleCallbackDelivery(ctx, queued[0]); err != nil { |
| 404 | + t.Fatal(err) |
| 405 | + } |
| 406 | + if receiver.count() != 1 { |
| 407 | + t.Fatalf("ephemeral callback count=%d, want 1", receiver.count()) |
| 408 | + } |
| 409 | +} |
| 410 | + |
| 411 | +func TestCallbackDeliveryStopsForRevokedOrExpiredAgentGrant(t *testing.T) { |
| 412 | + for _, action := range []string{"revoked", "expired"} { |
| 413 | + t.Run(action, func(t *testing.T) { |
| 414 | + _, tenant := loopbackTenant(t) |
| 415 | + ctx := context.Background() |
| 416 | + agent, _ := event.PublicKey(testAgentSecret) |
| 417 | + now := time.Now().Unix() |
| 418 | + grant := agentGrantEvent(t, agent, now-1, now+3600, []string{"k", "1"}) |
| 419 | + if err := publishAs(t, tenant, grant); err != nil { |
| 420 | + t.Fatal(err) |
| 421 | + } |
| 422 | + receiver := &callbackReceiver{status: http.StatusOK} |
| 423 | + service := httptest.NewTLSServer(receiver) |
| 424 | + defer service.Close() |
| 425 | + tenant.callbacks.callbackClient = service.Client() |
| 426 | + registered := addCallback(t, tenant, agent, service.URL+"/wake", `{"kinds":[1]}`) |
| 427 | + id := registered["id"].(string) |
| 428 | + note := signedEvent(t, testOwnerSecret, 1, now, nil, action) |
| 429 | + if err := publishAs(t, tenant, note); err != nil { |
| 430 | + t.Fatal(err) |
| 431 | + } |
| 432 | + queued := pendingCallbackIntents(t, tenant, id, note.ID) |
| 433 | + if len(queued) != 1 { |
| 434 | + t.Fatalf("queued = %+v", queued) |
| 435 | + } |
| 436 | + column, value := "revoked_at", now |
| 437 | + if action == "expired" { |
| 438 | + column, value = "expires_at", now-1 |
| 439 | + } |
| 440 | + if _, err := tenant.store.DB().ExecContext(ctx, "UPDATE agent_grants SET "+column+"=? WHERE agent=?", value, agent); err != nil { |
| 441 | + t.Fatal(err) |
| 442 | + } |
| 443 | + if err := tenant.handleCallbackDelivery(ctx, queued[0]); err != nil { |
| 444 | + t.Fatal(err) |
| 445 | + } |
| 446 | + if receiver.count() != 0 { |
| 447 | + t.Fatalf("%s agent callback delivered", action) |
| 448 | + } |
| 449 | + }) |
| 450 | + } |
| 451 | +} |
| 452 | + |
| 453 | func TestMCPCallbackTools(t *testing.T) { |
| 454 | app, tenant := testTenant(t) |
| 455 | member, _ := event.PublicKey(testMemberSecret) |