1package daemon23import (4 "context"5 "crypto/hmac"6 "crypto/sha256"7 "encoding/hex"8 "encoding/json"9 "io"10 "net/http"11 "net/http/httptest"12 "strconv"13 "strings"14 "sync"15 "testing"16 "time"1718 "github.com/FelineStateMachine/tinyrelay/internal/community"19 "github.com/FelineStateMachine/tinyrelay/internal/event"20 "github.com/FelineStateMachine/tinyrelay/internal/work"21)2223// loopbackTenant is a tenant whose public URL is a loopback address, which24// lets a callback point at a local test server.25func loopbackTenant(t *testing.T) (*App, *Tenant) {26 t.Helper()27 ctx := context.Background()28 a, err := New(ctx, Config{DataDir: t.TempDir(), DefaultTenant: "main"})29 if err != nil {30 t.Fatal(err)31 }32 t.Cleanup(func() {33 if err := a.Close(ctx); err != nil {34 t.Error(err)35 }36 })37 owner, _ := event.PublicKey(testOwnerSecret)38 meta, err := a.Create(ctx, CreateOptions{Name: "main", Owner: owner, Template: "default"})39 if err != nil {40 t.Fatal(err)41 }42 tenant, err := a.tenant(ctx, meta, "http://127.0.0.1:8080")43 if err != nil {44 t.Fatal(err)45 }46 // Fence the production worker so the tests drive each delivery intent47 // themselves.48 tenant.workCancel()49 tenant.workWG.Wait()50 return a, tenant51}5253func setRole(t *testing.T, tenant *Tenant, pubkey, role string) {54 t.Helper()55 raw, _ := json.Marshal(map[string]string{"role": role})56 if _, err := tenant.Execute(context.Background(), tenant.Policy().Owner, "setmember", []json.RawMessage{json.RawMessage(strconv.Quote(pubkey)), raw}); err != nil {57 t.Fatal(err)58 }59}6061func callbackParams(url string, filter string, extra ...string) []json.RawMessage {62 options := map[string]any{"url": url, "filter": json.RawMessage(filter)}63 if len(extra) > 0 {64 options["secret"] = extra[0]65 }66 raw, _ := json.Marshal(options)67 return []json.RawMessage{raw}68}6970func addCallback(t *testing.T, tenant *Tenant, actor, url, filter string) map[string]any {71 t.Helper()72 result, err := tenant.Execute(context.Background(), actor, "addcallback", callbackParams(url, filter))73 if err != nil {74 t.Fatalf("addcallback: %v", err)75 }76 return result.(map[string]any)77}7879func TestCallbackRegistrationValidatesURLFilterScopeAndCap(t *testing.T) {80 _, tenant := testTenant(t)81 ctx := context.Background()82 owner := tenant.Policy().Owner83 member, _ := event.PublicKey(testMemberSecret)84 agent, _ := event.PublicKey(testAgentSecret)85 setRole(t, tenant, member, "member")86 now := time.Now().Unix()87 if err := publishAs(t, tenant, agentGrantEvent(t, agent, now, now+3600, []string{"k", "1"}, []string{"room", "build"}, []string{"repo", owner + ":tinyrelay:read"})); err != nil {88 t.Fatal(err)89 }90 ids := make([]string, 9)91 for i := range ids {92 ids[i] = strings.Repeat(strconv.Itoa(i), 64)93 }94 nineIDs, _ := json.Marshal(ids)95 for _, tc := range []struct {96 name, actor, url, filter, want string97 }{98 {"guest", strings.Repeat("e", 64), "https://hooks.example/wake", `{"kinds":[1]}`, "restricted:"},99 {"http url", member, "http://hooks.example/wake", `{"kinds":[1]}`, "invalid: url must be https"},100 {"credentials", member, "https://user:pw@hooks.example/wake", `{"kinds":[1]}`, "invalid: url must be https"},101 {"private address", member, "https://10.0.0.1/wake", `{"kinds":[1]}`, "invalid: url must not name a private"},102 {"loopback address", member, "https://127.0.0.1/wake", `{"kinds":[1]}`, "invalid: url must not name a private"},103 {"loopback ipv6", member, "https://[::1]/wake", `{"kinds":[1]}`, "invalid: url must not name a private"},104 {"shared address", member, "https://100.64.0.9/wake", `{"kinds":[1]}`, "invalid: url must not name a private"},105 {"localhost", member, "https://localhost/wake", `{"kinds":[1]}`, "invalid: url must name a public host"},106 {"bare host", member, "https://relay/wake", `{"kinds":[1]}`, "invalid: url must name a public host"},107 {"no kinds", member, "https://hooks.example/wake", `{"#p":["` + member + `"]}`, "invalid: filter must name at least one kind"},108 {"unsupported key", member, "https://hooks.example/wake", `{"kinds":[1],"ids":["` + strings.Repeat("a", 64) + `"]}`, "invalid: filter key \"ids\""},109 {"unsupported tag", member, "https://hooks.example/wake", `{"kinds":[1],"#t":["x"]}`, "invalid: filter key \"#t\""},110 {"too many values", member, "https://hooks.example/wake", `{"kinds":[1],"#e":` + string(nineIDs) + `}`, "invalid: #e must list 1 to 8 values"},111 {"bad pubkey", member, "https://hooks.example/wake", `{"kinds":[1],"#p":["nope"]}`, "invalid: #p values"},112 {"bad coordinate", member, "https://hooks.example/wake", `{"kinds":[1621],"#a":["tinyrelay"]}`, "invalid: #a values"},113 {"bad room", member, "https://hooks.example/wake", `{"kinds":[9],"#h":["Not A Room"]}`, "invalid: #h values"},114 {"not an object", member, "https://hooks.example/wake", `[1]`, "invalid: filter must be a JSON object"},115 {"agent room outside grant", agent, "https://hooks.example/wake", `{"kinds":[9],"#h":["general"]}`, "restricted: agent grant does not cover room general"},116 {"agent repo outside grant", agent, "https://hooks.example/wake", `{"kinds":[1621],"#a":["30617:` + owner + `:other"]}`, "restricted: agent grant does not cover repository other"},117 } {118 _, err := tenant.Execute(ctx, tc.actor, "addcallback", callbackParams(tc.url, tc.filter))119 if err == nil || !strings.HasPrefix(err.Error(), tc.want) {120 t.Errorf("%s: err = %v, want prefix %q", tc.name, err, tc.want)121 }122 }123 if _, err := tenant.Execute(ctx, member, "addcallback", callbackParams("https://hooks.example/wake", `{"kinds":[1]}`, "short")); err == nil || !strings.HasPrefix(err.Error(), "invalid: secret must be") {124 t.Fatalf("short secret: %v", err)125 }126 // An agent may watch the room and repository its grant covers, and any127 // kind, including ones it may not publish.128 inside := addCallback(t, tenant, agent, "https://hooks.example/wake", `{"kinds":[7,9,1621],"#h":["build"],"#a":["30617:`+owner+`:tinyrelay"],"since":1}`)129 if inside["owner"] != agent || inside["host"] != "hooks.example" || inside["paused"] != false {130 t.Fatalf("agent callback = %v", inside)131 }132 secret, _ := inside["secret"].(string)133 if len(secret) != 64 {134 t.Fatalf("generated secret = %q", secret)135 }136 filter, _ := json.Marshal(inside["filter"])137 if string(filter) != `{"#a":["30617:`+owner+`:tinyrelay"],"#h":["build"],"kinds":[7,9,1621]}` {138 t.Fatalf("stored filter = %s", filter)139 }140 // A member gets four callbacks; the owner has no cap.141 for i := 0; i < 4; i++ {142 addCallback(t, tenant, member, "https://hooks.example/wake/"+strconv.Itoa(i), `{"kinds":[1]}`)143 }144 if _, err := tenant.Execute(ctx, member, "addcallback", callbackParams("https://hooks.example/wake/5", `{"kinds":[1]}`)); err == nil || err.Error() != "invalid: at most 4 callbacks per key" {145 t.Fatalf("fifth callback: %v", err)146 }147 for i := 0; i < 6; i++ {148 addCallback(t, tenant, owner, "https://hooks.example/owner/"+strconv.Itoa(i), `{"kinds":[1]}`)149 }150 // A public address literal is a host like any other.151 addCallback(t, tenant, owner, "https://[2001:db8::10]/wake", `{"kinds":[1]}`)152 // The secret shows once: listings and the table row never repeat it.153 listed, err := tenant.Execute(ctx, agent, "listcallbacks", nil)154 if err != nil {155 t.Fatal(err)156 }157 rows := listed.([]map[string]any)158 if len(rows) != 1 || rows[0]["id"] != inside["id"] || rows[0]["url"] != "https://hooks.example/wake" {159 t.Fatalf("agent listing = %v", rows)160 }161 if _, present := rows[0]["secret"]; present {162 t.Fatal("listing repeats the secret")163 }164 encoded, _ := json.Marshal(rows[0])165 if strings.Contains(string(encoded), secret) {166 t.Fatal("listing carries the secret")167 }168 // Setting the policy allowance to zero closes registration for169 // everyone but the owner.170 if _, err := tenant.Execute(ctx, owner, "setpolicy", []json.RawMessage{json.RawMessage(`{"callbacks":0}`)}); err != nil {171 t.Fatal(err)172 }173 if _, err := tenant.Execute(ctx, agent, "addcallback", callbackParams("https://hooks.example/again", `{"kinds":[1]}`)); err == nil || err.Error() != "invalid: at most 0 callbacks per key" {174 t.Fatalf("closed registration: %v", err)175 }176 if got := tenant.Policy().Callbacks; got != 0 {177 t.Fatalf("policy callbacks = %d", got)178 }179}180181func TestCallbackManagementPermissionMatrix(t *testing.T) {182 _, tenant := testTenant(t)183 ctx := context.Background()184 owner := tenant.Policy().Owner185 member, _ := event.PublicKey(testMemberSecret)186 moderator, _ := event.PublicKey(testModSecret)187 agent, _ := event.PublicKey(testAgentSecret)188 guest := strings.Repeat("e", 64)189 setRole(t, tenant, member, "member")190 setRole(t, tenant, moderator, "moderator")191 now := time.Now().Unix()192 if err := publishAs(t, tenant, agentGrantEvent(t, agent, now, now+3600, []string{"k", "1"})); err != nil {193 t.Fatal(err)194 }195 methods, err := tenant.Execute(ctx, owner, "supportedmethods", nil)196 if err != nil {197 t.Fatal(err)198 }199 for _, name := range []string{"listcallbacks", "addcallback", "removecallback", "pausecallback", "resumecallback"} {200 if !containsString(methods.([]string), name) {201 t.Fatalf("supportedmethods lacks %s", name)202 }203 }204 mine := addCallback(t, tenant, member, "https://hooks.example/member?token=abc", `{"kinds":[1]}`)205 theirs := addCallback(t, tenant, agent, "https://hooks.example/agent", `{"kinds":[1]}`)206 id := func(values map[string]any) json.RawMessage {207 return json.RawMessage(strconv.Quote(values["id"].(string)))208 }209 for _, method := range []string{"listcallbacks", "addcallback", "removecallback", "pausecallback", "resumecallback"} {210 if _, err := tenant.Execute(ctx, guest, method, []json.RawMessage{id(mine)}); err == nil || !strings.HasPrefix(err.Error(), "restricted:") {211 t.Fatalf("%s by guest: %v", method, err)212 }213 }214 // A member sees and controls only its own callbacks.215 listed, _ := tenant.Execute(ctx, member, "listcallbacks", nil)216 if rows := listed.([]map[string]any); len(rows) != 1 || rows[0]["id"] != mine["id"] {217 t.Fatalf("member listing = %v", rows)218 }219 for _, method := range []string{"pausecallback", "resumecallback", "removecallback"} {220 if _, err := tenant.Execute(ctx, member, method, []json.RawMessage{id(theirs)}); err == nil || !strings.HasPrefix(err.Error(), "restricted:") {221 t.Fatalf("%s on another key's callback: %v", method, err)222 }223 }224 paused, err := tenant.Execute(ctx, member, "pausecallback", []json.RawMessage{json.RawMessage(`{"id":"` + mine["id"].(string) + `"}`)})225 if err != nil || paused.(map[string]any)["paused"] != true {226 t.Fatalf("pause own: %v %v", paused, err)227 }228 if resumed, err := tenant.Execute(ctx, member, "resumecallback", []json.RawMessage{id(mine)}); err != nil || resumed.(map[string]any)["paused"] != false {229 t.Fatalf("resume own: %v %v", resumed, err)230 }231 // Moderators and the owner see everything, with the host but not the232 // path of other keys' URLs, and may control any callback.233 for _, actor := range []string{moderator, owner} {234 listed, err := tenant.Execute(ctx, actor, "listcallbacks", nil)235 if err != nil {236 t.Fatalf("listcallbacks by %s: %v", actor, err)237 }238 rows := listed.([]map[string]any)239 byOwner := map[string]map[string]any{}240 for _, row := range rows {241 byOwner[row["owner"].(string)] = row242 }243 if len(rows) != 2 || byOwner[member]["host"] != "hooks.example" || byOwner[agent]["id"] != theirs["id"] {244 t.Fatalf("operator listing = %v", rows)245 }246 if _, present := byOwner[member]["url"]; present {247 t.Fatalf("operator listing carries another key's URL path: %v", byOwner[member])248 }249 paused, err := tenant.Execute(ctx, actor, "pausecallback", []json.RawMessage{id(theirs)})250 if err != nil {251 t.Fatalf("pause by %s: %v", actor, err)252 }253 if _, present := paused.(map[string]any)["url"]; present {254 t.Fatalf("pause response carries another key's URL path: %v", paused)255 }256 resumed, err := tenant.Execute(ctx, actor, "resumecallback", []json.RawMessage{id(theirs)})257 if err != nil {258 t.Fatalf("resume by %s: %v", actor, err)259 }260 if _, present := resumed.(map[string]any)["url"]; present {261 t.Fatalf("resume response carries another key's URL path: %v", resumed)262 }263 }264 if _, err := tenant.Execute(ctx, owner, "pausecallback", []json.RawMessage{json.RawMessage(`"missing"`)}); err == nil || !strings.HasPrefix(err.Error(), "not found:") {265 t.Fatalf("unknown id: %v", err)266 }267 if _, err := tenant.Execute(ctx, owner, "pausecallback", nil); err == nil || !strings.HasPrefix(err.Error(), "invalid:") {268 t.Fatalf("missing id: %v", err)269 }270 removed, err := tenant.Execute(ctx, moderator, "removecallback", []json.RawMessage{id(theirs)})271 if err != nil || removed.(map[string]any)["removed"] != true {272 t.Fatalf("remove by moderator: %v %v", removed, err)273 }274 if _, err := tenant.Execute(ctx, agent, "removecallback", []json.RawMessage{id(mine)}); err == nil || !strings.HasPrefix(err.Error(), "restricted:") {275 t.Fatalf("agent removing a member's callback: %v", err)276 }277 if _, err := tenant.Execute(ctx, member, "removecallback", []json.RawMessage{id(mine)}); err != nil {278 t.Fatalf("remove own: %v", err)279 }280 audit, err := tenant.Execute(ctx, owner, "listaudit", nil)281 if err != nil {282 t.Fatal(err)283 }284 seen := map[string]bool{}285 for _, row := range audit.([]community.AuditRow) {286 seen[row.Action] = true287 if strings.Contains(row.Detail, "hooks.example") || strings.Contains(row.Target, "hooks.example") {288 t.Fatalf("audit carries a callback URL: %+v", row)289 }290 }291 for _, action := range []string{"addcallback", "pausecallback", "resumecallback", "removecallback"} {292 if !seen[action] {293 t.Fatalf("audit lacks %s: %v", action, seen)294 }295 }296}297298func TestCallbackMatchesFilterAndKeepsPrivateEventsBehindTheGate(t *testing.T) {299 _, tenant := loopbackTenant(t)300 ctx := context.Background()301 member, _ := event.PublicKey(testMemberSecret)302 moderator, _ := event.PublicKey(testModSecret)303 setRole(t, tenant, member, "member")304 setRole(t, tenant, moderator, "moderator")305 now := time.Now().Unix()306 notes := addCallback(t, tenant, member, "https://127.0.0.1:1/notes", `{"kinds":[1],"#p":["`+member+`"]}`)307 messages := addCallback(t, tenant, member, "https://127.0.0.1:1/messages", `{"kinds":[4]}`)308 intents := func() map[string]int {309 rows, err := tenant.store.DB().QueryContext(ctx, "SELECT target,count(*) FROM work_intents WHERE kind=? GROUP BY target", callbackDelivery)310 if err != nil {311 t.Fatal(err)312 }313 defer rows.Close()314 counts := map[string]int{}315 for rows.Next() {316 var target string317 var count int318 if err := rows.Scan(&target, &count); err != nil {319 t.Fatal(err)320 }321 counts[target] = count322 }323 return counts324 }325 // Kind and tag both have to match.326 if err := publishAs(t, tenant, signedEvent(t, testOwnerSecret, 1, now, [][]string{{"p", moderator}}, "not for the member")); err != nil {327 t.Fatal(err)328 }329 if err := publishAs(t, tenant, signedEvent(t, testOwnerSecret, 7, now, [][]string{{"p", member}}, "+")); err != nil {330 t.Fatal(err)331 }332 if counts := intents(); len(counts) != 0 {333 t.Fatalf("intents after unmatched events = %v", counts)334 }335 if err := publishAs(t, tenant, signedEvent(t, testOwnerSecret, 1, now+1, [][]string{{"p", member}}, "for the member")); err != nil {336 t.Fatal(err)337 }338 if counts := intents(); counts[notes["id"].(string)] != 1 {339 t.Fatalf("intents after matched note = %v", counts)340 }341 // A private message the key may not read is matched by the filter but342 // stopped by the gate; one addressed to the key goes through.343 if err := publishAs(t, tenant, signedEvent(t, testOwnerSecret, 4, now+2, [][]string{{"p", moderator}}, "secret")); err != nil {344 t.Fatal(err)345 }346 if counts := intents(); counts[messages["id"].(string)] != 0 {347 t.Fatalf("private message leaked to the callback: %v", counts)348 }349 if err := publishAs(t, tenant, signedEvent(t, testOwnerSecret, 4, now+3, [][]string{{"p", member}}, "for you")); err != nil {350 t.Fatal(err)351 }352 if counts := intents(); counts[messages["id"].(string)] != 1 {353 t.Fatalf("addressed message not queued: %v", counts)354 }355 // A paused callback leaves the index at once.356 if _, err := tenant.Execute(ctx, member, "pausecallback", []json.RawMessage{json.RawMessage(strconv.Quote(notes["id"].(string)))}); err != nil {357 t.Fatal(err)358 }359 if err := publishAs(t, tenant, signedEvent(t, testOwnerSecret, 1, now+4, [][]string{{"p", member}}, "while paused")); err != nil {360 t.Fatal(err)361 }362 if counts := intents(); counts[notes["id"].(string)] != 1 {363 t.Fatalf("paused callback still queued: %v", counts)364 }365 if got := tenant.Policy().Callbacks; got != 4 {366 t.Fatalf("default callbacks allowance = %d", got)367 }368}369370type callbackReceiver struct {371 mu sync.Mutex372 status int373 headers []http.Header374 bodies [][]byte375}376377func (r *callbackReceiver) ServeHTTP(w http.ResponseWriter, req *http.Request) {378 body, _ := io.ReadAll(req.Body)379 r.mu.Lock()380 r.headers = append(r.headers, req.Header.Clone())381 r.bodies = append(r.bodies, body)382 status := r.status383 r.mu.Unlock()384 w.WriteHeader(status)385}386387func (r *callbackReceiver) count() int {388 r.mu.Lock()389 defer r.mu.Unlock()390 return len(r.bodies)391}392393// pendingCallbackIntents lists queued deliveries of one event to one394// callback, first attempt first.395func pendingCallbackIntents(t *testing.T, tenant *Tenant, id, eventID string) []work.Intent {396 t.Helper()397 rows, err := tenant.store.DB().QueryContext(context.Background(), "SELECT id,event_id,payload,next_at FROM work_intents WHERE kind=? AND target=? AND state='pending' AND (event_id=? OR event_id LIKE ?) ORDER BY next_at,created_at,id", callbackDelivery, id, eventID, eventID+"#%")398 if err != nil {399 t.Fatal(err)400 }401 defer rows.Close()402 var out []work.Intent403 for rows.Next() {404 var item work.Intent405 var next int64406 if err := rows.Scan(&item.ID, &item.EventID, &item.Payload, &next); err != nil {407 t.Fatal(err)408 }409 item.Kind, item.Target, item.NextAt = callbackDelivery, id, time.Unix(next, 0)410 out = append(out, item)411 }412 return out413}414415func TestCallbackDeliverySignsBodyRetriesAndPauses(t *testing.T) {416 _, tenant := loopbackTenant(t)417 ctx := context.Background()418 member, _ := event.PublicKey(testMemberSecret)419 setRole(t, tenant, member, "member")420 receiver := &callbackReceiver{status: http.StatusAccepted}421 service := httptest.NewTLSServer(receiver)422 defer service.Close()423 tenant.callbacks.callbackClient = service.Client()424 registered := addCallback(t, tenant, member, service.URL+"/wake?token=t0k3n", `{"kinds":[1621]}`)425 id, secret := registered["id"].(string), registered["secret"].(string)426 now := time.Now().Unix()427 issue := signedEvent(t, testOwnerSecret, 1621, now, [][]string{{"a", "30617:" + tenant.Policy().Owner + ":tinyrelay"}, {"subject", "Broken build"}}, "The build fails.")428 if err := publishAs(t, tenant, issue); err != nil {429 t.Fatal(err)430 }431 queued := pendingCallbackIntents(t, tenant, id, issue.ID)432 if len(queued) != 1 || queued[0].EventID != issue.ID {433 t.Fatalf("queued = %+v", queued)434 }435 if err := tenant.handleCallbackDelivery(ctx, queued[0]); err != nil {436 t.Fatal(err)437 }438 if receiver.count() != 1 {439 t.Fatalf("deliveries = %d", receiver.count())440 }441 headers, body := receiver.headers[0], receiver.bodies[0]442 mac := hmac.New(sha256.New, []byte(secret))443 mac.Write(body)444 if headers.Get("Content-Type") != "application/json" || headers.Get("X-Tiny-Callback") != id || headers.Get("X-Tiny-Relay") != "http://127.0.0.1:8080" || headers.Get("X-Tiny-Signature") != "sha256="+hex.EncodeToString(mac.Sum(nil)) {445 t.Fatalf("headers = %v", headers)446 }447 delivered, err := event.Parse(body)448 if err != nil || delivered.ID != issue.ID || delivered.Content != issue.Content {449 t.Fatalf("body = %s (%v)", body, err)450 }451 var lastAt int64452 var lastStatus string453 var failures, paused int454 state := func() {455 t.Helper()456 if err := tenant.store.DB().QueryRowContext(ctx, "SELECT last_delivery_at,last_status,failures,paused FROM callbacks WHERE id=?", id).Scan(&lastAt, &lastStatus, &failures, &paused); err != nil {457 t.Fatal(err)458 }459 }460 state()461 if lastAt < now || lastStatus != "ok" || failures != 0 || paused != 0 {462 t.Fatalf("after delivery: at=%d status=%q failures=%d paused=%d", lastAt, lastStatus, failures, paused)463 }464 listed, _ := tenant.Execute(ctx, member, "listcallbacks", nil)465 if rows := listed.([]map[string]any); rows[0]["lastStatus"] != "ok" || rows[0]["lastDelivery"].(int64) < now {466 t.Fatalf("listing after delivery = %v", rows)467 }468469 // A failing receiver: attempt one queues attempt two a minute out,470 // attempt two queues attempt three five minutes out, and attempt three471 // is the last.472 receiver.status = http.StatusInternalServerError473 second := signedEvent(t, testOwnerSecret, 1621, now+1, [][]string{{"a", "30617:" + tenant.Policy().Owner + ":tinyrelay"}, {"subject", "Again"}}, "Again.")474 if err := publishAs(t, tenant, second); err != nil {475 t.Fatal(err)476 }477 queued = pendingCallbackIntents(t, tenant, id, second.ID)478 if len(queued) != 1 {479 t.Fatalf("queued = %+v", queued)480 }481 if err := tenant.handleCallbackDelivery(ctx, queued[0]); err != nil {482 t.Fatalf("attempt 1 should not fail the intent: %v", err)483 }484 state()485 if failures != 1 || lastStatus != "HTTP 500" || paused != 0 {486 t.Fatalf("after attempt 1: status=%q failures=%d paused=%d", lastStatus, failures, paused)487 }488 retries := pendingCallbackIntents(t, tenant, id, second.ID)489 if len(retries) != 2 || retries[1].EventID != second.ID+"#2" || retries[1].NextAt.Unix() < now+55 || retries[1].NextAt.Unix() > now+70 {490 t.Fatalf("retry after attempt 1 = %+v", retries)491 }492 var payload callbackPayload493 if err := json.Unmarshal([]byte(retries[1].Payload), &payload); err != nil || payload.Attempt != 2 || payload.Event.ID != second.ID {494 t.Fatalf("retry payload = %s", retries[1].Payload)495 }496 if err := tenant.handleCallbackDelivery(ctx, retries[1]); err != nil {497 t.Fatal(err)498 }499 retries = pendingCallbackIntents(t, tenant, id, second.ID)500 if len(retries) != 3 || retries[2].EventID != second.ID+"#3" || retries[2].NextAt.Unix() < now+295 || retries[2].NextAt.Unix() > now+310 {501 t.Fatalf("retry after attempt 2 = %+v", retries)502 }503 if err := tenant.handleCallbackDelivery(ctx, retries[2]); err != nil {504 t.Fatal(err)505 }506 if again := pendingCallbackIntents(t, tenant, id, second.ID); len(again) != 3 {507 t.Fatalf("attempt 3 queued another attempt: %+v", again)508 }509 state()510 if failures != 3 || receiver.count() != 4 {511 t.Fatalf("after three attempts: failures=%d deliveries=%d", failures, receiver.count())512 }513 // Twenty failures in a row pause the callback with the reason, after514 // which nothing more is queued or sent until it is resumed.515 for failures < callbackPauseFailures {516 if err := tenant.handleCallbackDelivery(ctx, retries[2]); err != nil {517 t.Fatal(err)518 }519 state()520 }521 if paused != 1 || !strings.HasPrefix(lastStatus, "paused after 20 failures: HTTP 500") {522 t.Fatalf("after 20 failures: status=%q paused=%d", lastStatus, paused)523 }524 sent := receiver.count()525 third := signedEvent(t, testOwnerSecret, 1621, now+2, [][]string{{"a", "30617:" + tenant.Policy().Owner + ":tinyrelay"}, {"subject", "Third"}}, "Third.")526 if err := publishAs(t, tenant, third); err != nil {527 t.Fatal(err)528 }529 if queued := pendingCallbackIntents(t, tenant, id, third.ID); len(queued) != 0 {530 t.Fatalf("paused callback queued a delivery: %+v", queued)531 }532 if err := tenant.handleCallbackDelivery(ctx, retries[2]); err != nil || receiver.count() != sent {533 t.Fatalf("paused callback delivered: %v %d", err, receiver.count())534 }535 receiver.status = http.StatusOK536 if _, err := tenant.Execute(ctx, member, "resumecallback", []json.RawMessage{json.RawMessage(strconv.Quote(id))}); err != nil {537 t.Fatal(err)538 }539 state()540 if paused != 0 || failures != 0 {541 t.Fatalf("after resume: failures=%d paused=%d", failures, paused)542 }543 fourth := signedEvent(t, testOwnerSecret, 1621, now+3, [][]string{{"a", "30617:" + tenant.Policy().Owner + ":tinyrelay"}, {"subject", "Fourth"}}, "Fourth.")544 if err := publishAs(t, tenant, fourth); err != nil {545 t.Fatal(err)546 }547 queued = pendingCallbackIntents(t, tenant, id, fourth.ID)548 if len(queued) != 1 {549 t.Fatalf("queued after resume = %+v", queued)550 }551 if err := tenant.handleCallbackDelivery(ctx, queued[0]); err != nil || receiver.count() != sent+1 {552 t.Fatalf("delivery after resume: %v %d", err, receiver.count())553 }554 // A removed callback consumes its intents without a POST.555 if _, err := tenant.Execute(ctx, member, "removecallback", []json.RawMessage{json.RawMessage(strconv.Quote(id))}); err != nil {556 t.Fatal(err)557 }558 if err := tenant.handleCallbackDelivery(ctx, queued[0]); err != nil || receiver.count() != sent+1 {559 t.Fatalf("removed callback delivered: %v %d", err, receiver.count())560 }561}562563func TestCallbackDeliveryRechecksTheGateAndMembership(t *testing.T) {564 _, tenant := loopbackTenant(t)565 ctx := context.Background()566 owner := tenant.Policy().Owner567 member, _ := event.PublicKey(testMemberSecret)568 setRole(t, tenant, member, "member")569 receiver := &callbackReceiver{status: http.StatusOK}570 service := httptest.NewTLSServer(receiver)571 defer service.Close()572 tenant.callbacks.callbackClient = service.Client()573 registered := addCallback(t, tenant, member, service.URL+"/wake", `{"kinds":[1]}`)574 id := registered["id"].(string)575 now := time.Now().Unix()576 note := signedEvent(t, testOwnerSecret, 1, now, nil, "hidden later")577 if err := publishAs(t, tenant, note); err != nil {578 t.Fatal(err)579 }580 queued := pendingCallbackIntents(t, tenant, id, note.ID)581 if len(queued) != 1 {582 t.Fatalf("queued = %+v", queued)583 }584 // Hidden between arrival and delivery: nothing is sent.585 if _, err := tenant.Execute(ctx, owner, "banevent", []json.RawMessage{json.RawMessage(strconv.Quote(note.ID))}); err != nil {586 t.Fatal(err)587 }588 if err := tenant.handleCallbackDelivery(ctx, queued[0]); err != nil || receiver.count() != 0 {589 t.Fatalf("hidden event delivered: %v %d", err, receiver.count())590 }591 // A key that lost its membership has its callback paused.592 if _, err := tenant.Execute(ctx, owner, "removemember", []json.RawMessage{json.RawMessage(strconv.Quote(member))}); err != nil {593 t.Fatal(err)594 }595 if err := tenant.handleCallbackDelivery(ctx, queued[0]); err != nil || receiver.count() != 0 {596 t.Fatalf("former member delivered: %v %d", err, receiver.count())597 }598 var paused int599 var status string600 if err := tenant.store.DB().QueryRowContext(ctx, "SELECT paused,last_status FROM callbacks WHERE id=?", id).Scan(&paused, &status); err != nil || paused != 1 || status != "paused: the key is no longer a member" {601 t.Fatalf("former member callback: paused=%d status=%q %v", paused, status, err)602 }603}604605func TestCallbackDeliveryStopsForInactiveAgentGrant(t *testing.T) {606 _, tenant := loopbackTenant(t)607 ctx := context.Background()608 owner := tenant.Policy().Owner609 agent, _ := event.PublicKey(testAgentSecret)610 now := time.Now().Unix()611 grant := agentGrantEvent(t, agent, now-1, now+3600, []string{"k", "1"})612 if err := publishAs(t, tenant, grant); err != nil {613 t.Fatal(err)614 }615 receiver := &callbackReceiver{status: http.StatusOK}616 service := httptest.NewTLSServer(receiver)617 defer service.Close()618 tenant.callbacks.callbackClient = service.Client()619 registered := addCallback(t, tenant, agent, service.URL+"/wake", `{"kinds":[1]}`)620 id := registered["id"].(string)621 note := signedEvent(t, testOwnerSecret, 1, now, nil, "grant pause")622 if err := publishAs(t, tenant, note); err != nil {623 t.Fatal(err)624 }625 queued := pendingCallbackIntents(t, tenant, id, note.ID)626 if len(queued) != 1 {627 t.Fatalf("queued = %+v", queued)628 }629 if _, err := tenant.Execute(ctx, owner, "pauseagent", []json.RawMessage{json.RawMessage(strconv.Quote(agent))}); err != nil {630 t.Fatal(err)631 }632 if err := tenant.handleCallbackDelivery(ctx, queued[0]); err != nil {633 t.Fatal(err)634 }635 if receiver.count() != 0 {636 t.Fatal("paused agent callback delivered")637 }638 var paused int639 if err := tenant.store.DB().QueryRowContext(ctx, "SELECT paused FROM callbacks WHERE id=?", id).Scan(&paused); err != nil || paused != 1 {640 t.Fatalf("inactive grant did not pause callback: paused=%d err=%v", paused, err)641 }642}643644func TestCallbackDeliverySkipsDeletedSource(t *testing.T) {645 _, tenant := loopbackTenant(t)646 ctx := context.Background()647 member, _ := event.PublicKey(testMemberSecret)648 setRole(t, tenant, member, "member")649 receiver := &callbackReceiver{status: http.StatusOK}650 service := httptest.NewTLSServer(receiver)651 defer service.Close()652 tenant.callbacks.callbackClient = service.Client()653 registered := addCallback(t, tenant, member, service.URL+"/wake", `{"kinds":[1]}`)654 id := registered["id"].(string)655 note := signedEvent(t, testOwnerSecret, 1, time.Now().Unix(), nil, "deleted source")656 if err := publishAs(t, tenant, note); err != nil {657 t.Fatal(err)658 }659 queued := pendingCallbackIntents(t, tenant, id, note.ID)660 if len(queued) != 1 {661 t.Fatalf("queued = %+v", queued)662 }663 if _, err := tenant.store.DB().ExecContext(ctx, "DELETE FROM events WHERE id=?", note.ID); err != nil {664 t.Fatal(err)665 }666 if err := tenant.handleCallbackDelivery(ctx, queued[0]); err != nil {667 t.Fatal(err)668 }669 if receiver.count() != 0 {670 t.Fatal("deleted source callback delivered")671 }672}673674func TestCallbackDeliverySkipsExpiredSource(t *testing.T) {675 _, tenant := loopbackTenant(t)676 ctx := context.Background()677 member, _ := event.PublicKey(testMemberSecret)678 setRole(t, tenant, member, "member")679 receiver := &callbackReceiver{status: http.StatusOK}680 service := httptest.NewTLSServer(receiver)681 defer service.Close()682 tenant.callbacks.callbackClient = service.Client()683 registered := addCallback(t, tenant, member, service.URL+"/wake", `{"kinds":[1]}`)684 id := registered["id"].(string)685 note := signedEvent(t, testOwnerSecret, 1, time.Now().Unix(), nil, "expired source")686 if err := publishAs(t, tenant, note); err != nil {687 t.Fatal(err)688 }689 queued := pendingCallbackIntents(t, tenant, id, note.ID)690 if len(queued) != 1 {691 t.Fatalf("queued = %+v", queued)692 }693 if _, err := tenant.store.DB().ExecContext(ctx, "UPDATE events SET expires=? WHERE id=?", time.Now().Unix()-1, note.ID); err != nil {694 t.Fatal(err)695 }696 if err := tenant.handleCallbackDelivery(ctx, queued[0]); err != nil {697 t.Fatal(err)698 }699 if receiver.count() != 0 {700 t.Fatal("expired source callback delivered")701 }702}703704func TestCallbackDeliveryRejectsForgedEphemeralProvenance(t *testing.T) {705 _, tenant := loopbackTenant(t)706 ctx := context.Background()707 member, _ := event.PublicKey(testMemberSecret)708 setRole(t, tenant, member, "member")709 receiver := &callbackReceiver{status: http.StatusOK}710 service := httptest.NewTLSServer(receiver)711 defer service.Close()712 tenant.callbacks.callbackClient = service.Client()713 registered := addCallback(t, tenant, member, service.URL+"/wake", `{"kinds":[1]}`)714 id := registered["id"].(string)715 note := signedEvent(t, testOwnerSecret, 1, time.Now().Unix(), nil, "normal event")716 if err := publishAs(t, tenant, note); err != nil {717 t.Fatal(err)718 }719 queued := pendingCallbackIntents(t, tenant, id, note.ID)720 if len(queued) != 1 {721 t.Fatalf("queued = %+v", queued)722 }723 var payload map[string]json.RawMessage724 if err := json.Unmarshal([]byte(queued[0].Payload), &payload); err != nil {725 t.Fatal(err)726 }727 payload["ephemeral"] = json.RawMessage("true")728 encoded, err := json.Marshal(payload)729 if err != nil {730 t.Fatal(err)731 }732 if _, err := tenant.store.DB().ExecContext(ctx, "UPDATE work_intents SET payload=? WHERE id=?", string(encoded), queued[0].ID); err != nil {733 t.Fatal(err)734 }735 if _, err := tenant.store.DB().ExecContext(ctx, "DELETE FROM events WHERE id=?", note.ID); err != nil {736 t.Fatal(err)737 }738 queued[0].Payload = string(encoded)739 if err := tenant.handleCallbackDelivery(ctx, queued[0]); err != nil || receiver.count() != 0 {740 t.Fatalf("forged ephemeral provenance: err=%v deliveries=%d", err, receiver.count())741 }742}743744func TestCallbackDeliveryStopsWhenAgentScopeNarrows(t *testing.T) {745 _, tenant := loopbackTenant(t)746 ctx := context.Background()747 agent, _ := event.PublicKey(testAgentSecret)748 owner := tenant.Policy().Owner749 now := time.Now().Unix()750 grant := agentGrantEvent(t, agent, now-1, now+3600, []string{"k", "1"}, []string{"repo", owner + ":repo:read"})751 if err := publishAs(t, tenant, grant); err != nil {752 t.Fatal(err)753 }754 receiver := &callbackReceiver{status: http.StatusOK}755 service := httptest.NewTLSServer(receiver)756 defer service.Close()757 tenant.callbacks.callbackClient = service.Client()758 filter := `{"kinds":[1],"#a":["30617:` + owner + `:repo"]}`759 registered := addCallback(t, tenant, agent, service.URL+"/wake", filter)760 id := registered["id"].(string)761 note := signedEvent(t, testOwnerSecret, 1, now, [][]string{{"a", "30617:" + owner + ":repo"}}, "narrowed scope")762 if err := publishAs(t, tenant, note); err != nil {763 t.Fatal(err)764 }765 queued := pendingCallbackIntents(t, tenant, id, note.ID)766 if len(queued) != 1 {767 t.Fatalf("queued = %+v", queued)768 }769 if _, err := tenant.store.DB().ExecContext(ctx, `UPDATE agent_grants SET scope=? WHERE agent=?`, `{"kinds":[1],"rooms":[],"repos":[],"rate":60}`, agent); err != nil {770 t.Fatal(err)771 }772 if err := tenant.handleCallbackDelivery(ctx, queued[0]); err != nil {773 t.Fatal(err)774 }775 if receiver.count() != 0 {776 t.Fatal("scope-narrowed agent callback delivered")777 }778 var paused int779 if err := tenant.store.DB().QueryRowContext(ctx, "SELECT paused FROM callbacks WHERE id=?", id).Scan(&paused); err != nil || paused != 1 {780 t.Fatalf("scope-narrowed callback not paused: paused=%d err=%v", paused, err)781 }782}783784func TestCallbackDeliveryAllowsActiveAgentWithStoredSource(t *testing.T) {785 _, tenant := loopbackTenant(t)786 ctx := context.Background()787 agent, _ := event.PublicKey(testAgentSecret)788 now := time.Now().Unix()789 grant := agentGrantEvent(t, agent, now-1, now+3600, []string{"k", "1"})790 if err := publishAs(t, tenant, grant); err != nil {791 t.Fatal(err)792 }793 receiver := &callbackReceiver{status: http.StatusOK}794 service := httptest.NewTLSServer(receiver)795 defer service.Close()796 tenant.callbacks.callbackClient = service.Client()797 registered := addCallback(t, tenant, agent, service.URL+"/wake", `{"kinds":[1]}`)798 id := registered["id"].(string)799 note := signedEvent(t, testOwnerSecret, 1, now, nil, "active grant")800 if err := publishAs(t, tenant, note); err != nil {801 t.Fatal(err)802 }803 queued := pendingCallbackIntents(t, tenant, id, note.ID)804 if len(queued) != 1 {805 t.Fatalf("queued = %+v", queued)806 }807 if err := tenant.handleCallbackDelivery(ctx, queued[0]); err != nil {808 t.Fatal(err)809 }810 if receiver.count() != 1 {811 t.Fatalf("active agent callback deliveries=%d, want 1", receiver.count())812 }813}814815func TestEphemeralCallbackDeliveryUsesKindProvenance(t *testing.T) {816 _, tenant := loopbackTenant(t)817 ctx := context.Background()818 member, _ := event.PublicKey(testMemberSecret)819 setRole(t, tenant, member, "member")820 receiver := &callbackReceiver{status: http.StatusOK}821 service := httptest.NewTLSServer(receiver)822 defer service.Close()823 tenant.callbacks.callbackClient = service.Client()824 registered := addCallback(t, tenant, member, service.URL+"/wake", `{"kinds":[20001]}`)825 id := registered["id"].(string)826 presence := signedEvent(t, testOwnerSecret, 20001, time.Now().Unix(), nil, "ephemeral")827 tenant.notifyCallbacks(ctx, presence)828 queued := pendingCallbackIntents(t, tenant, id, presence.ID)829 if len(queued) != 1 {830 t.Fatalf("queued = %+v", queued)831 }832 if err := tenant.handleCallbackDelivery(ctx, queued[0]); err != nil {833 t.Fatal(err)834 }835 if receiver.count() != 1 {836 t.Fatalf("ephemeral callback count=%d, want 1", receiver.count())837 }838}839840func TestCallbackDeliveryStopsForRevokedOrExpiredAgentGrant(t *testing.T) {841 for _, action := range []string{"revoked", "expired"} {842 t.Run(action, func(t *testing.T) {843 _, tenant := loopbackTenant(t)844 ctx := context.Background()845 agent, _ := event.PublicKey(testAgentSecret)846 now := time.Now().Unix()847 grant := agentGrantEvent(t, agent, now-1, now+3600, []string{"k", "1"})848 if err := publishAs(t, tenant, grant); err != nil {849 t.Fatal(err)850 }851 receiver := &callbackReceiver{status: http.StatusOK}852 service := httptest.NewTLSServer(receiver)853 defer service.Close()854 tenant.callbacks.callbackClient = service.Client()855 registered := addCallback(t, tenant, agent, service.URL+"/wake", `{"kinds":[1]}`)856 id := registered["id"].(string)857 note := signedEvent(t, testOwnerSecret, 1, now, nil, action)858 if err := publishAs(t, tenant, note); err != nil {859 t.Fatal(err)860 }861 queued := pendingCallbackIntents(t, tenant, id, note.ID)862 if len(queued) != 1 {863 t.Fatalf("queued = %+v", queued)864 }865 column, value := "revoked_at", now866 if action == "expired" {867 column, value = "expires_at", now-1868 }869 if _, err := tenant.store.DB().ExecContext(ctx, "UPDATE agent_grants SET "+column+"=? WHERE agent=?", value, agent); err != nil {870 t.Fatal(err)871 }872 if err := tenant.handleCallbackDelivery(ctx, queued[0]); err != nil {873 t.Fatal(err)874 }875 if receiver.count() != 0 {876 t.Fatalf("%s agent callback delivered", action)877 }878 })879 }880}881882func TestMCPCallbackTools(t *testing.T) {883 app, tenant := testTenant(t)884 member, _ := event.PublicKey(testMemberSecret)885 setRole(t, tenant, member, "member")886 call := func(name string, arguments map[string]any, secret string) (map[string]any, bool) {887 t.Helper()888 w, response := mcpCall{method: "tools/call", name: name, arguments: arguments, sign: true, secret: secret}.do(t, app, "/mcp")889 if w.Code != http.StatusOK {890 t.Fatalf("%s: %d %s", name, w.Code, w.Body.String())891 }892 return mcpToolResult(t, response)893 }894 structured := func(result map[string]any) map[string]any {895 value, _ := result["structuredContent"].(map[string]any)896 inner, _ := value["result"].(map[string]any)897 if inner == nil {898 return value899 }900 return inner901 }902 text := func(result map[string]any) string {903 content, _ := result["content"].([]any)904 if len(content) == 0 {905 return ""906 }907 return content[0].(map[string]any)["text"].(string)908 }909 _, response := mcpCall{method: "tools/list", sign: true}.do(t, app, "/mcp")910 names := map[string]bool{}911 for _, tool := range response.Result.(map[string]any)["tools"].([]any) {912 names[tool.(map[string]any)["name"].(string)] = true913 }914 for _, want := range []string{"list_callbacks", "add_callback", "remove_callback", "pause_callback", "resume_callback"} {915 if !names[want] {916 t.Fatalf("missing tool %s", want)917 }918 }919 filter := map[string]any{"kinds": []any{1621.0, 1111.0}, "#p": []any{member}}920 result, isError := call("add_callback", map[string]any{"url": "https://hooks.example/wake", "filter": filter}, strings.Repeat("7", 64))921 if !isError || !strings.Contains(text(result), "restricted") {922 t.Fatalf("add_callback as guest: %v %s", isError, text(result))923 }924 result, isError = call("add_callback", map[string]any{"url": "http://hooks.example/wake", "filter": filter}, testMemberSecret)925 if !isError || !strings.Contains(text(result), "https") {926 t.Fatalf("add_callback with http: %v %s", isError, text(result))927 }928 result, isError = call("add_callback", map[string]any{"url": "https://hooks.example/wake", "filter": filter, "secret": "a-shared-secret-of-my-own"}, testMemberSecret)929 added := structured(result)930 if isError || added["secret"] != "a-shared-secret-of-my-own" || added["host"] != "hooks.example" || added["owner"] != member {931 t.Fatalf("add_callback: %s", text(result))932 }933 id, _ := added["id"].(string)934 result, isError = call("list_callbacks", map[string]any{}, testMemberSecret)935 listed, _ := result["structuredContent"].(map[string]any)["result"].([]any)936 if isError || len(listed) != 1 || listed[0].(map[string]any)["id"] != id || listed[0].(map[string]any)["paused"] != false {937 t.Fatalf("list_callbacks: %s", text(result))938 }939 if strings.Contains(text(result), "a-shared-secret-of-my-own") {940 t.Fatal("list_callbacks repeats the secret")941 }942 result, isError = call("pause_callback", map[string]any{"id": id}, "")943 if isError || structured(result)["paused"] != true {944 t.Fatalf("pause_callback as owner: %s", text(result))945 }946 result, isError = call("resume_callback", map[string]any{"id": id}, testMemberSecret)947 if isError || structured(result)["paused"] != false {948 t.Fatalf("resume_callback: %s", text(result))949 }950 result, isError = call("remove_callback", map[string]any{"id": id}, strings.Repeat("7", 64))951 if !isError {952 t.Fatalf("remove_callback as guest: %s", text(result))953 }954 result, isError = call("remove_callback", map[string]any{"id": id}, testMemberSecret)955 if isError || structured(result)["removed"] != true {956 t.Fatalf("remove_callback: %s", text(result))957 }958 result, _ = call("list_callbacks", map[string]any{}, "")959 if listed, _ := result["structuredContent"].(map[string]any)["result"].([]any); len(listed) != 0 {960 t.Fatalf("list_callbacks after remove: %s", text(result))961 }962 r := httptest.NewRequest(http.MethodGet, "http://relay.test/llms.txt", nil)963 w := httptest.NewRecorder()964 app.ServeHTTP(w, r)965 if !strings.Contains(w.Body.String(), "list_callbacks") || !strings.Contains(w.Body.String(), "add_callback") {966 t.Fatalf("llms.txt lacks the callback tools: %s", w.Body.String())967 }968}