| 1 | // Package hooks is the webhook-ingest sidecar (`lchat hooks run`, issue #610, |
| 2 | // docs/webhooks.md) — the INVERSE of the bot sidecar. The bot sidecar is a |
| 3 | // pure-outbound client that must never listen; this one exists to listen. It |
| 4 | // serves `in.<host>` (a registry route behind the instance router) and turns |
| 5 | // each inbound HTTP request into an encrypted typed event posted into the one |
| 6 | // thread that webhook feeds. |
| 7 | // |
| 8 | // It is strictly write-only against the instance: |
| 9 | // |
| 10 | // - it never opens a websocket and never calls GET /api/messages — nothing |
| 11 | // here can read a conversation; |
| 12 | // - it never holds the instance key — webhook definitions are sealed |
| 13 | // directly to its service identity's pubkey (e2ee.OpenSealed); |
| 14 | // - per webhook it holds exactly that webhook's private key + device token |
| 15 | // (from the definition), which unwrap exactly one thread's key. |
| 16 | // |
| 17 | // The outbound path is the Announcer's: unwrap the wrapped thread key, encrypt |
| 18 | // the payload, POST /api/messages. Compromise of this process yields the |
| 19 | // threads its webhooks feed — never the space's other threads, never a bot's |
| 20 | // brain, never the Anthropic key (it has none). |
| 21 | package hooks |
| 22 | |
| 23 | import ( |
| 24 | "context" |
| 25 | "encoding/base64" |
| 26 | "encoding/json" |
| 27 | "fmt" |
| 28 | "io" |
| 29 | "net/http" |
| 30 | "net/url" |
| 31 | "strings" |
| 32 | "sync" |
| 33 | "time" |
| 34 | "unicode/utf8" |
| 35 | |
| 36 | "lchat/internal/bot" |
| 37 | "lchat/internal/client" |
| 38 | "lchat/internal/e2ee" |
| 39 | ) |
| 40 | |
| 41 | const ( |
| 42 | // maxRawBody caps an inbound request body. The server's message-blob cap is |
| 43 | // 64K; after JSON framing + encryption + base64 a 40K raw body is the safe |
| 44 | // bound. Oversize is rejected with 413 — never silently truncated (#610). |
| 45 | maxRawBody = 40 * 1024 |
| 46 | // Per-webhook rate limit: a token bucket refilled 1/s with a 60-deep burst. |
| 47 | // The server's content-blind message brake still backstops this. |
| 48 | rateBurst = 60 |
| 49 | ratePerSec = 1 |
| 50 | // DefaultPoll is how often the sidecar re-reads its definitions. |
| 51 | DefaultPoll = 10 * time.Second |
| 52 | ) |
| 53 | |
| 54 | // headerAllowlist is the ONLY request-header subset captured into events. |
| 55 | // Full headers routinely carry sender credentials (Authorization, cookies, |
| 56 | // bespoke API keys) that must never enter the log — even encrypted, every |
| 57 | // thread member could read them. Signature headers are fine: an HMAC over one |
| 58 | // body is evidence, not a reusable secret. |
| 59 | var headerAllowlist = []string{ |
| 60 | "Content-Type", |
| 61 | "User-Agent", |
| 62 | "X-Github-Event", |
| 63 | "X-Github-Delivery", |
| 64 | "X-Github-Hook-Id", |
| 65 | "X-Gitlab-Event", |
| 66 | "X-Gitlab-Instance", |
| 67 | "X-Hub-Signature-256", |
| 68 | "X-Event-Key", |
| 69 | "X-Request-Id", |
| 70 | "X-Slack-Signature", |
| 71 | "X-Slack-Request-Timestamp", |
| 72 | "X-Webhook-Event", |
| 73 | } |
| 74 | |
| 75 | // Definition is one webhook, as sealed into def_blob by the creating admin's |
| 76 | // browser (webhooks.js buildDefinition — the two must stay in step). |
| 77 | type Definition struct { |
| 78 | V int `json:"v"` |
| 79 | ID string `json:"id"` // webhook row id (wh-…) |
| 80 | UserID string `json:"user_id"` // the webhook's agent user (event attribution) |
| 81 | Name string `json:"name"` // display name (rides inside each event) |
| 82 | NS string `json:"ns"` // event namespace → type "webhooks.<ns>" |
| 83 | ThreadID string `json:"thread_id"` // the ONE thread this webhook feeds |
| 84 | Secret string `json:"secret"` // URL capability: /h/<secret> |
| 85 | Token string `json:"token"` // the webhook's device token (posting) |
| 86 | Priv string `json:"priv"` // the webhook's private key (thread-key unwrap) |
| 87 | } |
| 88 | |
| 89 | // route is one live webhook: its definition plus delivery state that survives |
| 90 | // definition re-polls (the unwrapped thread key and the rate bucket). |
| 91 | type route struct { |
| 92 | def Definition |
| 93 | |
| 94 | mu sync.Mutex |
| 95 | threadKey []byte // nil until first delivery fetches + unwraps it |
| 96 | tokens float64 // rate-limit bucket |
| 97 | lastFill time.Time |
| 98 | } |
| 99 | |
| 100 | // Sidecar is the running ingest process: an HTTP handler over a route table the |
| 101 | // definition poll rebuilds. |
| 102 | type Sidecar struct { |
| 103 | base string |
| 104 | sess *client.Session |
| 105 | logf func(string, ...any) |
| 106 | // Two clients, split by hibernation semantics (bot.SidecarHeader): |
| 107 | // polling is MARKED (recurring keepalive-ish traffic must never hold a |
| 108 | // hibernating instance awake — it just 503s while the instance sleeps and |
| 109 | // the stale route table keeps serving), while delivery is UNMARKED — an |
| 110 | // inbound webhook is real external activity and must WAKE a sleeping |
| 111 | // instance to land its event. |
| 112 | poll *http.Client |
| 113 | post *http.Client |
| 114 | |
| 115 | mu sync.RWMutex |
| 116 | routes map[string]*route // secret → route |
| 117 | polled bool // one successful poll has filled the table |
| 118 | warned map[string]bool // def ids we've already logged as unreadable |
| 119 | } |
| 120 | |
| 121 | // New builds a sidecar for base (the instance origin) with the provisioned |
| 122 | // service-identity session. |
| 123 | func New(base string, sess *client.Session, logf func(string, ...any)) *Sidecar { |
| 124 | if logf == nil { |
| 125 | logf = func(string, ...any) {} |
| 126 | } |
| 127 | return &Sidecar{ |
| 128 | base: strings.TrimRight(base, "/"), |
| 129 | sess: sess, |
| 130 | logf: logf, |
| 131 | poll: &http.Client{Timeout: 30 * time.Second, Transport: markTransport{}}, |
| 132 | post: &http.Client{Timeout: 30 * time.Second}, |
| 133 | routes: map[string]*route{}, |
| 134 | warned: map[string]bool{}, |
| 135 | } |
| 136 | } |
| 137 | |
| 138 | type markTransport struct{} |
| 139 | |
| 140 | func (markTransport) RoundTrip(req *http.Request) (*http.Response, error) { |
| 141 | req = req.Clone(req.Context()) |
| 142 | req.Header.Set(bot.SidecarHeader, "1") |
| 143 | return http.DefaultTransport.RoundTrip(req) |
| 144 | } |
| 145 | |
| 146 | // Run polls definitions until ctx ends. The HTTP listener is the caller's |
| 147 | // (hookscmd wires Handler() into its server); Run only keeps the table fresh. |
| 148 | func (s *Sidecar) Run(ctx context.Context, every time.Duration) { |
| 149 | if every <= 0 { |
| 150 | every = DefaultPoll |
| 151 | } |
| 152 | t := time.NewTicker(every) |
| 153 | defer t.Stop() |
| 154 | for { |
| 155 | if err := s.PollOnce(); err != nil { |
| 156 | s.logf("hooks: definition poll: %v", err) |
| 157 | } |
| 158 | select { |
| 159 | case <-ctx.Done(): |
| 160 | return |
| 161 | case <-t.C: |
| 162 | } |
| 163 | } |
| 164 | } |
| 165 | |
| 166 | // PollOnce re-reads the definition registry, rebuilds the route table (carrying |
| 167 | // over per-route delivery state), and acks any not-yet-deployed webhook. |
| 168 | func (s *Sidecar) PollOnce() error { |
| 169 | // A restarted sidecar has an EMPTY table until a poll succeeds — a |
| 170 | // delivery in that window would 404 the sender. So the very first poll is |
| 171 | // unmarked (it may wake a sleeping instance, once, at boot); the recurring |
| 172 | // ones are marked and simply fail while the instance sleeps, leaving the |
| 173 | // last-known table serving. |
| 174 | s.mu.RLock() |
| 175 | c := s.poll |
| 176 | if !s.polled { |
| 177 | c = s.post |
| 178 | } |
| 179 | s.mu.RUnlock() |
| 180 | resp, err := c.Get(s.base + "/api/hooks/defs?token=" + url.QueryEscape(s.sess.Token)) |
| 181 | if err != nil { |
| 182 | return err |
| 183 | } |
| 184 | defer resp.Body.Close() |
| 185 | if resp.StatusCode != http.StatusOK { |
| 186 | return fmt.Errorf("GET /api/hooks/defs: HTTP %d", resp.StatusCode) |
| 187 | } |
| 188 | var body struct { |
| 189 | Hooks []struct { |
| 190 | ID string `json:"id"` |
| 191 | DefBlob string `json:"def_blob"` |
| 192 | DeployedAt int64 `json:"deployed_at"` |
| 193 | } `json:"hooks"` |
| 194 | } |
| 195 | if err := json.NewDecoder(resp.Body).Decode(&body); err != nil { |
| 196 | return err |
| 197 | } |
| 198 | |
| 199 | next := make(map[string]*route, len(body.Hooks)) |
| 200 | var ack []string |
| 201 | s.mu.RLock() |
| 202 | prev := s.routes |
| 203 | s.mu.RUnlock() |
| 204 | prevByID := make(map[string]*route, len(prev)) |
| 205 | for _, rt := range prev { |
| 206 | prevByID[rt.def.ID] = rt |
| 207 | } |
| 208 | for _, h := range body.Hooks { |
| 209 | var def Definition |
| 210 | if err := e2ee.OpenSealed(h.DefBlob, s.sess.UserPriv, &def); err != nil { |
| 211 | // Sealed to a different service identity (e.g. re-provisioned |
| 212 | // sidecar) or corrupt — fail closed per row, loudly but once. |
| 213 | if !s.warned[h.ID] { |
| 214 | s.logf("hooks: definition %s is unreadable (sealed to another identity?): %v", h.ID, err) |
| 215 | } |
| 216 | s.warned[h.ID] = true |
| 217 | continue |
| 218 | } |
| 219 | if def.Secret == "" || def.ThreadID == "" || def.Token == "" || def.Priv == "" || !validNS(def.NS) { |
| 220 | if !s.warned[h.ID] { |
| 221 | s.logf("hooks: definition %s is malformed — skipping", h.ID) |
| 222 | } |
| 223 | s.warned[h.ID] = true |
| 224 | continue |
| 225 | } |
| 226 | rt := prevByID[def.ID] |
| 227 | if rt == nil || rt.def != def { |
| 228 | carried := &route{def: def, tokens: rateBurst, lastFill: time.Now()} |
| 229 | if rt != nil { |
| 230 | // Same webhook, changed definition: keep the bucket, drop the |
| 231 | // cached key only if the thread or identity changed. |
| 232 | rt.mu.Lock() |
| 233 | carried.tokens, carried.lastFill = rt.tokens, rt.lastFill |
| 234 | if rt.def.ThreadID == def.ThreadID && rt.def.Priv == def.Priv { |
| 235 | carried.threadKey = rt.threadKey |
| 236 | } |
| 237 | rt.mu.Unlock() |
| 238 | } |
| 239 | rt = carried |
| 240 | } |
| 241 | next[def.Secret] = rt |
| 242 | if h.DeployedAt == 0 { |
| 243 | ack = append(ack, def.ID) |
| 244 | } |
| 245 | } |
| 246 | s.mu.Lock() |
| 247 | s.routes = next |
| 248 | s.polled = true |
| 249 | s.mu.Unlock() |
| 250 | |
| 251 | for _, id := range ack { |
| 252 | if err := s.ackDeployed(id); err != nil { |
| 253 | s.logf("hooks: deploy ack %s: %v", id, err) |
| 254 | } else { |
| 255 | s.logf("hooks: webhook %s deployed", id) |
| 256 | } |
| 257 | } |
| 258 | return nil |
| 259 | } |
| 260 | |
| 261 | func (s *Sidecar) ackDeployed(id string) error { |
| 262 | raw, _ := json.Marshal(map[string]string{"token": s.sess.Token}) |
| 263 | resp, err := s.poll.Post(s.base+"/api/hooks/"+url.PathEscape(id)+"/deployed", "application/json", strings.NewReader(string(raw))) |
| 264 | if err != nil { |
| 265 | return err |
| 266 | } |
| 267 | defer resp.Body.Close() |
| 268 | if resp.StatusCode != http.StatusOK { |
| 269 | return fmt.Errorf("HTTP %d", resp.StatusCode) |
| 270 | } |
| 271 | return nil |
| 272 | } |
| 273 | |
| 274 | // Routes reports how many webhooks are currently live (logging/tests). |
| 275 | func (s *Sidecar) Routes() int { |
| 276 | s.mu.RLock() |
| 277 | defer s.mu.RUnlock() |
| 278 | return len(s.routes) |
| 279 | } |
| 280 | |
| 281 | // Handler is the ingest surface: ANY method on /h/<secret>[/<subpath>]. |
| 282 | // Unknown secrets 404 — the sidecar alone knows which webhooks exist; the |
| 283 | // router forwards blindly. |
| 284 | func (s *Sidecar) Handler() http.Handler { |
| 285 | mux := http.NewServeMux() |
| 286 | mux.HandleFunc("GET /healthz", func(w http.ResponseWriter, _ *http.Request) { |
| 287 | fmt.Fprintln(w, "ok") |
| 288 | }) |
| 289 | mux.HandleFunc("/h/", s.serveHook) |
| 290 | mux.HandleFunc("/", func(w http.ResponseWriter, _ *http.Request) { |
| 291 | http.Error(w, "not found", http.StatusNotFound) |
| 292 | }) |
| 293 | return mux |
| 294 | } |
| 295 | |
| 296 | func (s *Sidecar) serveHook(w http.ResponseWriter, r *http.Request) { |
| 297 | rest := strings.TrimPrefix(r.URL.Path, "/h/") |
| 298 | secret, subpath, _ := strings.Cut(rest, "/") |
| 299 | s.mu.RLock() |
| 300 | rt := s.routes[secret] |
| 301 | s.mu.RUnlock() |
| 302 | if secret == "" || rt == nil { |
| 303 | http.Error(w, "not found", http.StatusNotFound) |
| 304 | return |
| 305 | } |
| 306 | if !rt.allow() { |
| 307 | http.Error(w, "rate limited", http.StatusTooManyRequests) |
| 308 | return |
| 309 | } |
| 310 | // Read the capped body; MaxBytesReader makes an over-cap send a hard error |
| 311 | // rather than a silent prefix. |
| 312 | raw, err := io.ReadAll(http.MaxBytesReader(w, r.Body, maxRawBody)) |
| 313 | if err != nil { |
| 314 | http.Error(w, "body too large", http.StatusRequestEntityTooLarge) |
| 315 | return |
| 316 | } |
| 317 | |
| 318 | event := s.buildEvent(rt.def, r, subpath, raw) |
| 319 | if err := s.deliver(rt, event); err != nil { |
| 320 | s.logf("hooks: deliver %s: %v", rt.def.ID, err) |
| 321 | http.Error(w, "delivery failed", http.StatusBadGateway) |
| 322 | return |
| 323 | } |
| 324 | writeJSON(w, map[string]any{"ok": true}) |
| 325 | } |
| 326 | |
| 327 | // buildEvent shapes the typed event. Dumb ingest, smart render: the raw request |
| 328 | // rides inside the event forever, so client-side rendering can improve later |
| 329 | // without re-ingesting. Pinned by test/event-vectors/webhooks.json. |
| 330 | func (s *Sidecar) buildEvent(def Definition, r *http.Request, subpath string, raw []byte) map[string]any { |
| 331 | headers := map[string]string{} |
| 332 | for _, h := range headerAllowlist { |
| 333 | if v := r.Header.Get(h); v != "" { |
| 334 | headers[h] = v |
| 335 | } |
| 336 | } |
| 337 | query := map[string][]string(r.URL.Query()) |
| 338 | body, b64 := string(raw), false |
| 339 | if !utf8.ValidString(body) { |
| 340 | body, b64 = base64.RawURLEncoding.EncodeToString(raw), true |
| 341 | } |
| 342 | req := map[string]any{ |
| 343 | "method": r.Method, |
| 344 | "path": "/" + subpath, // after the secret segment; the secret itself never enters the log |
| 345 | "query": query, |
| 346 | "headers": headers, |
| 347 | "body": body, |
| 348 | } |
| 349 | if b64 { |
| 350 | req["body_b64"] = true |
| 351 | } |
| 352 | return map[string]any{ |
| 353 | "type": "webhooks." + def.NS, |
| 354 | "name": def.Name, |
| 355 | "uid": def.UserID, |
| 356 | "ts": time.Now().UnixMilli(), |
| 357 | "hook": map[string]any{"id": def.ID, "ns": def.NS}, |
| 358 | "req": req, |
| 359 | } |
| 360 | } |
| 361 | |
| 362 | // deliver encrypts the event to the webhook's thread key and posts it — the |
| 363 | // Announcer's outbound-only walk (unwrap → EncryptJSON → POST /api/messages). |
| 364 | func (s *Sidecar) deliver(rt *route, event map[string]any) error { |
| 365 | key, err := s.threadKey(rt) |
| 366 | if err != nil { |
| 367 | return err |
| 368 | } |
| 369 | blob, err := e2ee.EncryptJSON(key, event) |
| 370 | if err != nil { |
| 371 | return err |
| 372 | } |
| 373 | raw, _ := json.Marshal(map[string]string{"thread_id": rt.def.ThreadID, "blob": blob}) |
| 374 | resp, err := s.post.Post(s.base+"/api/messages?token="+url.QueryEscape(rt.def.Token), "application/json", strings.NewReader(string(raw))) |
| 375 | if err != nil { |
| 376 | return err |
| 377 | } |
| 378 | defer resp.Body.Close() |
| 379 | if resp.StatusCode != http.StatusOK { |
| 380 | return fmt.Errorf("POST /api/messages: HTTP %d", resp.StatusCode) |
| 381 | } |
| 382 | return nil |
| 383 | } |
| 384 | |
| 385 | // threadKey returns the webhook's unwrapped thread key, fetching it on first |
| 386 | // use: GET /api/threads under the WEBHOOK'S OWN token lists exactly the one |
| 387 | // thread it was added to, with the key wrapped to its pubkey at creation. |
| 388 | func (s *Sidecar) threadKey(rt *route) ([]byte, error) { |
| 389 | rt.mu.Lock() |
| 390 | cached := rt.threadKey |
| 391 | rt.mu.Unlock() |
| 392 | if cached != nil { |
| 393 | return cached, nil |
| 394 | } |
| 395 | // Unmarked like deliver: this fetch happens mid-delivery, so it too must |
| 396 | // be able to wake a sleeping instance. |
| 397 | resp, err := s.post.Get(s.base + "/api/threads?token=" + url.QueryEscape(rt.def.Token)) |
| 398 | if err != nil { |
| 399 | return nil, err |
| 400 | } |
| 401 | defer resp.Body.Close() |
| 402 | if resp.StatusCode != http.StatusOK { |
| 403 | return nil, fmt.Errorf("GET /api/threads: HTTP %d", resp.StatusCode) |
| 404 | } |
| 405 | var body struct { |
| 406 | Threads []struct { |
| 407 | ID string `json:"id"` |
| 408 | Wrapped string `json:"wrapped"` |
| 409 | } `json:"threads"` |
| 410 | } |
| 411 | if err := json.NewDecoder(resp.Body).Decode(&body); err != nil { |
| 412 | return nil, err |
| 413 | } |
| 414 | for _, t := range body.Threads { |
| 415 | if t.ID != rt.def.ThreadID || t.Wrapped == "" { |
| 416 | continue |
| 417 | } |
| 418 | keyB64, err := e2ee.UnwrapThreadKey(t.Wrapped, rt.def.Priv) |
| 419 | if err != nil { |
| 420 | return nil, fmt.Errorf("unwrap thread key: %w", err) |
| 421 | } |
| 422 | key, err := e2ee.ImportKey(keyB64) |
| 423 | if err != nil { |
| 424 | return nil, err |
| 425 | } |
| 426 | rt.mu.Lock() |
| 427 | rt.threadKey = key |
| 428 | rt.mu.Unlock() |
| 429 | return key, nil |
| 430 | } |
| 431 | return nil, fmt.Errorf("thread %s not visible to webhook %s (membership missing?)", rt.def.ThreadID, rt.def.ID) |
| 432 | } |
| 433 | |
| 434 | // allow is the per-webhook token bucket. |
| 435 | func (rt *route) allow() bool { |
| 436 | rt.mu.Lock() |
| 437 | defer rt.mu.Unlock() |
| 438 | now := time.Now() |
| 439 | rt.tokens += now.Sub(rt.lastFill).Seconds() * ratePerSec |
| 440 | if rt.tokens > rateBurst { |
| 441 | rt.tokens = rateBurst |
| 442 | } |
| 443 | rt.lastFill = now |
| 444 | if rt.tokens < 1 { |
| 445 | return false |
| 446 | } |
| 447 | rt.tokens-- |
| 448 | return true |
| 449 | } |
| 450 | |
| 451 | // validNS mirrors the client-side namespace rule: lowercase slug, so the event |
| 452 | // type "webhooks.<ns>" is always a well-formed app event type. |
| 453 | func validNS(ns string) bool { |
| 454 | if ns == "" || len(ns) > 32 { |
| 455 | return false |
| 456 | } |
| 457 | for i, c := range ns { |
| 458 | switch { |
| 459 | case c >= 'a' && c <= 'z', c >= '0' && c <= '9': |
| 460 | case c == '-' && i > 0: |
| 461 | default: |
| 462 | return false |
| 463 | } |
| 464 | } |
| 465 | return true |
| 466 | } |
| 467 | |
| 468 | func writeJSON(w http.ResponseWriter, v any) { |
| 469 | w.Header().Set("Content-Type", "application/json") |
| 470 | _ = json.NewEncoder(w).Encode(v) |
| 471 | } |
| 472 | |
amadan