| 1 | package bot |
| 2 | |
| 3 | import ( |
| 4 | "errors" |
| 5 | "fmt" |
| 6 | "os" |
| 7 | "strings" |
| 8 | "sync" |
| 9 | "time" |
| 10 | |
| 11 | "lchat/integrations" |
| 12 | "lchat/internal/e2ee" |
| 13 | ) |
| 14 | |
| 15 | // Connected-services capability ops (docs/connected-services.md): the Go port |
| 16 | // of runtime.mjs's handleConnect / handleOAuthExchange. These are protocol |
| 17 | // plumbing, not chat behaviour — they run regardless of the bot's recipe or |
| 18 | // whether it has a model brain, so the runtime routes connect.* payloads here |
| 19 | // directly (before the recipe Handler ever sees them). |
| 20 | // |
| 21 | // The trust story is the whole point: a member seals a time-boxed grant to |
| 22 | // THIS bot's public key; we open it with our own private key (a grant sealed |
| 23 | // to a different bot won't open, so we stay silent), act once, and drop the |
| 24 | // token. The lchat server only ever relays the sealed blobs. |
| 25 | |
| 26 | // connectConfig is read from the same env the Node sidecar uses. The broker |
| 27 | // holds DNSimple's client_secret; unset, the OAuth exchange can't run (but |
| 28 | // grant-based ops still can, since those carry their own token). |
| 29 | type connectConfig struct { |
| 30 | brokerURL string |
| 31 | brokerToken string |
| 32 | dnsimple *dnsimpleClient |
| 33 | refreshes *refreshGroup // serializes token refreshes per credential |
| 34 | } |
| 35 | |
| 36 | // defaultBrokerURL is the shared connect broker; an instance sets only its |
| 37 | // per-instance LCHAT_CONNECT_BROKER_TOKEN, so the URL defaults here (matching |
| 38 | // the Node runtime, which defaults it inline). Override with |
| 39 | // LCHAT_CONNECT_BROKER_URL for a self-hosted broker. |
| 40 | const defaultBrokerURL = "https://connect.elevenmessenger.com" |
| 41 | |
| 42 | // maxOpResultBytes caps one registry-op result: it must fit inside a single |
| 43 | // encrypted message (maxBlobBytes, 64 KB ciphertext), and base64 costs 4/3 — |
| 44 | // 40 KB of result is ~54 KB encrypted, leaving envelope headroom. Manifests |
| 45 | // keep real results below this via `pick` (a 100-repo GitHub page projects to |
| 46 | // ~26 KB). |
| 47 | const maxOpResultBytes = 40 << 10 |
| 48 | |
| 49 | func loadConnectConfig() *connectConfig { |
| 50 | brokerURL := os.Getenv("LCHAT_CONNECT_BROKER_URL") |
| 51 | if brokerURL == "" { |
| 52 | brokerURL = defaultBrokerURL |
| 53 | } |
| 54 | return &connectConfig{ |
| 55 | brokerURL: brokerURL, |
| 56 | brokerToken: os.Getenv("LCHAT_CONNECT_BROKER_TOKEN"), |
| 57 | dnsimple: newDNSimpleClient(os.Getenv("LCHAT_DNSIMPLE_BASE")), |
| 58 | refreshes: newRefreshGroup(), |
| 59 | } |
| 60 | } |
| 61 | |
| 62 | // connectPayload is the decrypted shape of a connect.op / connect.oauth message. |
| 63 | type connectPayload struct { |
| 64 | Type string `json:"type"` |
| 65 | Service string `json:"service"` |
| 66 | Op string `json:"op"` |
| 67 | Domain string `json:"domain"` |
| 68 | Args map[string]string `json:"args"` // registry-op arguments (integrations.Execute validates them) |
| 69 | Grant string `json:"grant"` // sealed to this bot: {service, token, exp, access} |
| 70 | RID string `json:"rid"` // request id for a requestData round-trip |
| 71 | Code string `json:"code"` // OAuth |
| 72 | State string `json:"state"` // OAuth |
| 73 | UID string `json:"uid"` // sender's user id (to seal the token back) |
| 74 | Name string `json:"name"` |
| 75 | Src string `json:"src"` // "chip" = a tapped bot proposal (its failure gets a spoken explanation) |
| 76 | } |
| 77 | |
| 78 | // grant is what a member sealed to the bot for one capability op. |
| 79 | type grant struct { |
| 80 | Service string `json:"service"` |
| 81 | Token string `json:"token"` |
| 82 | Exp int64 `json:"exp"` // unix millis; 0 = no expiry |
| 83 | Access string `json:"access"` // "read" | "write"; empty = legacy grant, read-only |
| 84 | // Expiring-token providers (GitHub Apps): the refresh token and the access |
| 85 | // token's own expiry ride along so the bot can rotate a stale pair mid-op |
| 86 | // and hand the new pair back (sealed to the member) on the result. |
| 87 | Refresh string `json:"refresh"` |
| 88 | TokenExp int64 `json:"token_expires_at"` // unix millis; 0 = non-expiring |
| 89 | // Kind is how the credential was born ("pat" | "oauth"; empty = legacy). |
| 90 | // Mixed-scheme providers need it: Linear's pasted keys go in the header |
| 91 | // bare, its OAuth tokens as Bearer — ExecuteAs picks per token. |
| 92 | Kind string `json:"kind"` |
| 93 | } |
| 94 | |
| 95 | // handleConnect runs a capability op against a sealed grant. It replies ONLY |
| 96 | // with a non-rendering connect.result (matched by rid) — the DNSimple app |
| 97 | // renders the data itself, so a capability op is never visible as a bot chat |
| 98 | // message. A rid-less op (fire-and-forget) gets no reply. |
| 99 | func (r *Runtime) handleConnect(threadID string, key []byte, p *connectPayload) { |
| 100 | var g grant |
| 101 | if e2ee.OpenSealed(p.Grant, r.Session.UserPriv, &g) != nil { |
| 102 | return // not sealed to us — silently ignore (it isn't ours) |
| 103 | } |
| 104 | reply := func(result map[string]any) { |
| 105 | if p.RID == "" { |
| 106 | return |
| 107 | } |
| 108 | out := map[string]any{"type": "connect.result", "service": p.Service, "op": p.Op, "rid": p.RID, "ts": time.Now().UnixMilli()} |
| 109 | for k, v := range result { |
| 110 | out[k] = v |
| 111 | } |
| 112 | if err := r.sendPayload(threadID, key, out); err != nil { |
| 113 | r.logf("connect.result: %v", err) |
| 114 | } |
| 115 | } |
| 116 | // A chip is this bot's own proposal, so its failure is answered in the |
| 117 | // bot's voice: the raw error still goes back on the result (the instant |
| 118 | // notice), and the brain follows with a plain-language explanation in the |
| 119 | // same thread. Chip ops only — an app surface (the GitHub app's Refresh) |
| 120 | // renders its own inline errors, and a bot message there would be noise. |
| 121 | fail := func(msg string) { |
| 122 | reply(map[string]any{"error": msg}) |
| 123 | if p.Src == "chip" { |
| 124 | go r.explainConnectError(threadID, key, p, msg) |
| 125 | } |
| 126 | } |
| 127 | if g.Exp != 0 && g.Exp < time.Now().UnixMilli() { |
| 128 | reply(map[string]any{"error": "That authorization expired — try again."}) |
| 129 | return |
| 130 | } |
| 131 | // Registry-driven path: any service/op pair the integrations registry |
| 132 | // declares runs through the pinned executor. Anything else falls through |
| 133 | // to the legacy dnsimple-only path below — deliberately including the |
| 134 | // legacy op id "list", which the dnsimple manifest does NOT declare (its |
| 135 | // op is "domains.list"), so existing DNSimple-app clients keep working. |
| 136 | if m := integrations.MustRegistry().Get(p.Service); m != nil { |
| 137 | if op := m.Op(p.Op); op != nil { |
| 138 | if g.Service != p.Service { |
| 139 | reply(map[string]any{"error": "That authorization is for \"" + g.Service + "\", not " + p.Service + "."}) |
| 140 | return |
| 141 | } |
| 142 | // A read grant cannot run a write op: the member scoped what they |
| 143 | // consented to, and an empty access is a legacy (read) grant. |
| 144 | if op.Access == "write" && g.Access != "write" { |
| 145 | reply(map[string]any{"error": "That authorization is read-only."}) |
| 146 | return |
| 147 | } |
| 148 | // Expiring-token providers: rotate a stale pair before calling |
| 149 | // (with a little slack), and hand the fresh pair back on the |
| 150 | // result so the member's stored credential stays alive. Rotation |
| 151 | // is serialized per credential — providers' refresh tokens are |
| 152 | // single-use, so a concurrent second op shares the first refresh |
| 153 | // instead of burning a dead token. |
| 154 | token := g.Token |
| 155 | var renewed *tokenPair |
| 156 | if g.Refresh != "" && g.TokenExp != 0 && time.Now().UnixMilli() > g.TokenExp-60_000 { |
| 157 | pair, err := r.refreshGrant(&g, p.UID) |
| 158 | if err != nil { |
| 159 | fail(err.Error()) |
| 160 | return |
| 161 | } |
| 162 | token, renewed = pair.Access, pair |
| 163 | } |
| 164 | data, err := integrations.ExecuteAs(m, p.Op, p.Args, token, g.Kind) |
| 165 | // A 401 on an un-refreshed expiring pair means the expiry stamp |
| 166 | // lied (revoked-and-reissued, clock skew): rotate once and retry. |
| 167 | var se *integrations.StatusError |
| 168 | if err != nil && errors.As(err, &se) && se.Code == 401 && g.Refresh != "" && renewed == nil { |
| 169 | pair, rerr := r.refreshGrant(&g, p.UID) |
| 170 | if rerr == nil { |
| 171 | token, renewed = pair.Access, pair |
| 172 | data, err = integrations.ExecuteAs(m, p.Op, p.Args, token, g.Kind) |
| 173 | } |
| 174 | } |
| 175 | if err != nil { |
| 176 | fail(err.Error()) |
| 177 | return |
| 178 | } |
| 179 | // An op result rides inside ONE encrypted message (64 KB blob cap, |
| 180 | // hub.go), so cap what one op can push into a thread. Truncating |
| 181 | // would hand the app broken JSON — refuse honestly instead; ops |
| 182 | // bound their responses with the manifest's `pick` projection. |
| 183 | if len(data) > maxOpResultBytes { |
| 184 | fail("That result is too large to bring into the chat — narrow the request.") |
| 185 | return |
| 186 | } |
| 187 | out := map[string]any{"data": data} |
| 188 | // The renewed pair rides on the result, sealed to the MEMBER (never |
| 189 | // stored here) — the requesting device re-stores it, so the next op |
| 190 | // starts from the fresh pair. |
| 191 | if renewed != nil { |
| 192 | if pub := r.senderPublicKey(p.UID); pub != "" { |
| 193 | if sealed, err := e2ee.SealForPub(pub, renewed.sealable()); err == nil { |
| 194 | out["renewed"] = sealed |
| 195 | } |
| 196 | } |
| 197 | } |
| 198 | reply(out) |
| 199 | return |
| 200 | } |
| 201 | } |
| 202 | if g.Service != "dnsimple" { |
| 203 | reply(map[string]any{"error": "Can't act on \"" + g.Service + "\" yet."}) |
| 204 | return |
| 205 | } |
| 206 | if p.Op != "list" { |
| 207 | reply(map[string]any{"error": "Unknown action \"" + p.Op + "\"."}) |
| 208 | return |
| 209 | } |
| 210 | domains, err := r.connect.dnsimple.listDomains(g.Token) |
| 211 | if err != nil { |
| 212 | reply(map[string]any{"error": err.Error()}) |
| 213 | return |
| 214 | } |
| 215 | reply(map[string]any{"domains": domains}) |
| 216 | } |
| 217 | |
| 218 | // handleOAuthExchange finishes the confidential OAuth dance for any service |
| 219 | // the broker has an app for: run the broker exchange (no secret on this |
| 220 | // host), seal the token to the MEMBER's public key so only they can open it, |
| 221 | // hand it straight back, and keep no copy. The service must be a registry |
| 222 | // service (or legacy dnsimple) — the broker refuses anything it has no app |
| 223 | // for regardless. |
| 224 | func (r *Runtime) handleOAuthExchange(threadID string, key []byte, p *connectPayload) { |
| 225 | if p.Service == "" { |
| 226 | return |
| 227 | } |
| 228 | name := p.Service |
| 229 | if m := integrations.MustRegistry().Get(p.Service); m != nil { |
| 230 | name = m.Name |
| 231 | } else if p.Service != "dnsimple" { |
| 232 | return // not a service we know — stay silent (it isn't ours) |
| 233 | } |
| 234 | pair, err := exchangeViaBroker(p.Service, p.Code, p.State, r.connect.brokerURL, r.connect.brokerToken) |
| 235 | if err != nil { |
| 236 | r.logf("connect: %s exchange failed: %v", p.Service, err) |
| 237 | r.postReply(threadID, key, Reply{Text: "Couldn't finish connecting " + name + ": " + err.Error()}) |
| 238 | return |
| 239 | } |
| 240 | pub := r.senderPublicKey(p.UID) |
| 241 | if pub == "" { |
| 242 | r.postReply(threadID, key, Reply{Text: "Connected, but I couldn't find your key to hand it back — try again."}) |
| 243 | return |
| 244 | } |
| 245 | // The whole pair (access + refresh + expiry, when the provider issues |
| 246 | // them) is sealed to the member; a non-expiring provider keeps the legacy |
| 247 | // {secret} shape. |
| 248 | sealed, err := e2ee.SealForPub(pub, pair.sealable()) |
| 249 | if err != nil { |
| 250 | r.logf("connect: seal token: %v", err) |
| 251 | return |
| 252 | } |
| 253 | if err := r.sendPayload(threadID, key, map[string]any{ |
| 254 | "type": "connect.token", "service": p.Service, "sealed": sealed, "name": r.Session.Name, "ts": time.Now().UnixMilli(), |
| 255 | }); err != nil { |
| 256 | r.logf("connect.token: %v", err) |
| 257 | } |
| 258 | } |
| 259 | |
| 260 | // explainConnectError has the brain own a failed chip op conversationally: |
| 261 | // the chip was this bot's proposal, so its failure gets a short plain-language |
| 262 | // reply in-thread — what went wrong and the concrete fix (the knowledge base |
| 263 | // carries the decoder for the common cases, e.g. GitHub's 404-means-not- |
| 264 | // installed). The raw error already went back on the connect.result; a bot |
| 265 | // with no brain just leaves it at that. |
| 266 | func (r *Runtime) explainConnectError(threadID string, key []byte, p *connectPayload, errMsg string) { |
| 267 | if r.ai == nil { |
| 268 | return |
| 269 | } |
| 270 | resp, err := r.ai.create(anthRequest{ |
| 271 | Model: r.model, |
| 272 | MaxTokens: 1024, |
| 273 | System: r.systemBlocks(), |
| 274 | Messages: []anthMessage{{Role: "user", Content: "[automatic system note — not written by the member] " + |
| 275 | "The member just tapped your \"" + p.Op + "\" chip for " + p.Service + " and the operation FAILED. " + |
| 276 | "The raw error (already shown to them as a notice): " + errMsg + " — " + |
| 277 | "In one short, friendly message: acknowledge it didn't work, explain the likely cause in plain " + |
| 278 | "language, and give the concrete fix. No tool calls, no new chips, no apologising twice."}}, |
| 279 | }) |
| 280 | if err != nil { |
| 281 | r.logf("connect explain: %v", err) |
| 282 | return |
| 283 | } |
| 284 | if text := strings.TrimSpace(resp.text()); text != "" { |
| 285 | r.postReply(threadID, key, Reply{Text: text}) |
| 286 | } |
| 287 | } |
| 288 | |
| 289 | // refreshGrant rotates the grant's expiring pair via the broker. It refuses |
| 290 | // to rotate when the fresh pair couldn't be handed back (no resolvable member |
| 291 | // key) — rotation consumes the stored refresh token, so losing the result |
| 292 | // would sever the connection outright. |
| 293 | func (r *Runtime) refreshGrant(g *grant, uid string) (*tokenPair, error) { |
| 294 | if uid == "" || r.senderPublicKey(uid) == "" { |
| 295 | return nil, fmt.Errorf("that %s connection's token expired and I couldn't hand a fresh one back — reconnect it in Integrations", g.Service) |
| 296 | } |
| 297 | return r.connect.refreshes.do(g.Refresh, func() (*tokenPair, error) { |
| 298 | return refreshViaBroker(g.Service, g.Refresh, r.connect.brokerURL, r.connect.brokerToken) |
| 299 | }) |
| 300 | } |
| 301 | |
| 302 | // refreshGroup coalesces concurrent refreshes of the SAME credential: refresh |
| 303 | // tokens are single-use (GitHub rotates one on every call), so two |
| 304 | // near-simultaneous ops must share one rotation rather than the loser burning |
| 305 | // a dead token. Nothing is retained once a flight lands — the fresh pair |
| 306 | // lives only in the results handed to the waiting ops. |
| 307 | type refreshGroup struct { |
| 308 | mu sync.Mutex |
| 309 | calls map[string]*refreshCall |
| 310 | } |
| 311 | |
| 312 | type refreshCall struct { |
| 313 | done chan struct{} |
| 314 | pair *tokenPair |
| 315 | err error |
| 316 | } |
| 317 | |
| 318 | func newRefreshGroup() *refreshGroup { return &refreshGroup{calls: map[string]*refreshCall{}} } |
| 319 | |
| 320 | func (rg *refreshGroup) do(key string, fn func() (*tokenPair, error)) (*tokenPair, error) { |
| 321 | rg.mu.Lock() |
| 322 | if c, ok := rg.calls[key]; ok { |
| 323 | rg.mu.Unlock() |
| 324 | <-c.done |
| 325 | return c.pair, c.err |
| 326 | } |
| 327 | c := &refreshCall{done: make(chan struct{})} |
| 328 | rg.calls[key] = c |
| 329 | rg.mu.Unlock() |
| 330 | c.pair, c.err = fn() |
| 331 | close(c.done) |
| 332 | rg.mu.Lock() |
| 333 | delete(rg.calls, key) |
| 334 | rg.mu.Unlock() |
| 335 | return c.pair, c.err |
| 336 | } |
| 337 | |
| 338 | // senderPublicKey looks up a roster member's public key by user id (to seal the |
| 339 | // OAuth token back to them). |
| 340 | func (r *Runtime) senderPublicKey(uid string) string { |
| 341 | if uid == "" { |
| 342 | return "" |
| 343 | } |
| 344 | var rl struct { |
| 345 | Roster []struct { |
| 346 | UserID string `json:"user_id"` |
| 347 | PublicKey string `json:"public_key"` |
| 348 | } `json:"roster"` |
| 349 | } |
| 350 | if err := r.get("/api/roster", &rl); err != nil { |
| 351 | r.logf("connect: roster: %v", err) |
| 352 | return "" |
| 353 | } |
| 354 | for _, e := range rl.Roster { |
| 355 | if e.UserID == uid { |
| 356 | return e.PublicKey |
| 357 | } |
| 358 | } |
| 359 | return "" |
| 360 | } |
| 361 | |
amadan