| 1 | package calls |
| 2 | |
| 3 | import ( |
| 4 | "bytes" |
| 5 | "crypto/sha256" |
| 6 | "encoding/hex" |
| 7 | "encoding/json" |
| 8 | "fmt" |
| 9 | "net/http" |
| 10 | "strings" |
| 11 | "time" |
| 12 | |
| 13 | "github.com/go-chi/chi/v5" |
| 14 | |
| 15 | "amadan.net/rastrillo/rastrillo/sessions" |
| 16 | |
| 17 | "amadan.net/oficina/calls/internal/calls/hub" |
| 18 | ) |
| 19 | |
| 20 | const ( |
| 21 | // defaultHeartbeat and defaultStreamTTL are Docs' live-sync numbers |
| 22 | // (spec §4.1): a comment line every fifteen seconds so a dead stream |
| 23 | // is noticed, and a five-minute life so hibernation is never held |
| 24 | // off by one. The browser resumes; nothing is lost at the TTL. |
| 25 | defaultHeartbeat = 15 * time.Second |
| 26 | defaultStreamTTL = 5 * time.Minute |
| 27 | streamWriteWindow = 30 * time.Second |
| 28 | // minStreamTiming is the floor under both, so a zero or negative |
| 29 | // value from SetStreamTimings cannot panic a ticker in a request |
| 30 | // goroutine. |
| 31 | minStreamTiming = 10 * time.Millisecond |
| 32 | // maxLastQuery bounds the resume's ?last= before it is parsed. |
| 33 | maxLastQuery = 8 << 10 |
| 34 | ) |
| 35 | |
| 36 | // sessionKey is the session's identity for the hub (spec §4.2): the |
| 37 | // session cookie this request presented, keyed through a salt this |
| 38 | // process chose at boot, so what the hub holds is neither the token nor |
| 39 | // the hash rastrillo stores. Two tabs of one browser share it, which is |
| 40 | // what one peer per room and its takeover need; two browsers of one |
| 41 | // person do not, which §4.4's four streams per person allows. |
| 42 | // |
| 43 | // The plan expected the subject plus the session's creation instant. |
| 44 | // That is not an identity: two sign-ins of one person within one second |
| 45 | // collide, and the second browser is then refused the room. |
| 46 | func (a *app) sessionKey(r *http.Request) (string, bool) { |
| 47 | s, ok := sessions.Current(r) |
| 48 | if !ok || s.Subject == "" { |
| 49 | return "", false |
| 50 | } |
| 51 | c, err := r.Cookie(a.auth.SessionCookie()) |
| 52 | if err != nil || c.Value == "" { |
| 53 | return "", false |
| 54 | } |
| 55 | return hubSessionKey(a.hubSalt, s.Subject, c.Value), true |
| 56 | } |
| 57 | |
| 58 | // hubSessionKey is the derivation itself, kept apart from the request so |
| 59 | // its three properties can be gated (review W-4): the subject is bound |
| 60 | // in as well as the cookie, so the key cannot survive the person it |
| 61 | // belongs to whatever a cookie value might come to be reused for; the |
| 62 | // per-boot salt means it is meaningless across a restart; and the |
| 63 | // result is neither the session token nor the hash rastrillo stores, so |
| 64 | // a hub key is not a credential. |
| 65 | func hubSessionKey(salt, subject, cookie string) string { |
| 66 | sum := sha256.Sum256([]byte(salt + "|" + subject + "|" + cookie)) |
| 67 | return hex.EncodeToString(sum[:]) |
| 68 | } |
| 69 | |
| 70 | // acceptsEventStream is ruling 15. The attaching stream is only ever |
| 71 | // opened by EventSource, which always asks for text/event-stream; a |
| 72 | // cross-site top-level navigation asks for HTML, and without this check |
| 73 | // it would take the session's one peer slot in the room and hold it for |
| 74 | // the resume window. Anything else is 406 with {"error":"accept"}. The |
| 75 | // probe forms are exempt: they attach nothing. |
| 76 | func acceptsEventStream(r *http.Request) bool { |
| 77 | return strings.Contains(strings.ToLower(r.Header.Get("Accept")), "text/event-stream") |
| 78 | } |
| 79 | |
| 80 | // testTTLHeader shortens one stream's life, and only in development. |
| 81 | // EventSource cannot set a header, so nothing a browser does on its own |
| 82 | // can reach this; the browser gate rewrites it on the way past, which is |
| 83 | // the only way to watch a renewal without waiting five minutes. Outside |
| 84 | // development the header is not read at all: a request may not shorten |
| 85 | // a stream on its own say-so. |
| 86 | const testTTLHeader = "X-Calls-Test-TTL" |
| 87 | |
| 88 | // streamLifeFor is streamLife with that override applied, through the |
| 89 | // same floor atLeast puts under every timing — a header cannot panic a |
| 90 | // timer in a request goroutine any more than SetStreamTimings can. |
| 91 | func (a *app) streamLifeFor(r *http.Request) time.Duration { |
| 92 | if !a.cfg.development() { |
| 93 | return a.streamLife() |
| 94 | } |
| 95 | raw := r.Header.Get(testTTLHeader) |
| 96 | if raw == "" { |
| 97 | return a.streamLife() |
| 98 | } |
| 99 | d, err := time.ParseDuration(raw) |
| 100 | if err != nil { |
| 101 | return a.streamLife() |
| 102 | } |
| 103 | return atLeast(int64(d)) |
| 104 | } |
| 105 | |
| 106 | // streamJSON is every refusal this endpoint makes. The room's client is |
| 107 | // a script, not a page: spec §4.5 fixes the browser's behaviour per |
| 108 | // status and body, so a refusal answers JSON rather than the error page |
| 109 | // a screen would get. |
| 110 | func streamJSON(w http.ResponseWriter, status int, body map[string]any) { |
| 111 | w.Header().Set("Content-Type", "application/json") |
| 112 | w.Header().Set("Cache-Control", "no-store") |
| 113 | w.WriteHeader(status) |
| 114 | if body != nil { |
| 115 | _ = json.NewEncoder(w).Encode(body) |
| 116 | } |
| 117 | } |
| 118 | |
| 119 | func (a *app) refuseResume(w http.ResponseWriter, re *hub.ResumeError) { |
| 120 | body := map[string]any{} |
| 121 | if re.Reason != "" { |
| 122 | body["reason"] = string(re.Reason) |
| 123 | } |
| 124 | if re.Boot != "" { |
| 125 | body["boot"] = re.Boot |
| 126 | } |
| 127 | streamJSON(w, re.Status, body) |
| 128 | } |
| 129 | |
| 130 | // stream is GET /m/{sqid}/stream: one SSE stream per peer (spec §4.1), |
| 131 | // opened fresh on standing or resumed on a participation bound to this |
| 132 | // session (§2.5). ?probe=1 answers the same statuses and stops there. |
| 133 | func (a *app) stream(w http.ResponseWriter, r *http.Request) { |
| 134 | sqid := chi.URLParam(r, "sqid") |
| 135 | person, ok := currentPerson(r) |
| 136 | if !ok { |
| 137 | a.logger.Error("calls: stream with no person in context", "path", r.URL.Path) |
| 138 | streamJSON(w, http.StatusInternalServerError, nil) |
| 139 | return |
| 140 | } |
| 141 | session, ok := a.sessionKey(r) |
| 142 | if !ok { |
| 143 | a.logger.Error("calls: stream with no session subject", "path", r.URL.Path) |
| 144 | streamJSON(w, http.StatusInternalServerError, nil) |
| 145 | return |
| 146 | } |
| 147 | probe := r.URL.Query().Get("probe") == "1" |
| 148 | if !probe && !acceptsEventStream(r) { |
| 149 | // Not a stream client, so nothing is looked up and nothing is |
| 150 | // allocated (ruling 15). This runs before any authorization, so |
| 151 | // the answer is identical for every sqid and every person and |
| 152 | // discloses nothing — which is why it can afford to be the |
| 153 | // truthful 406 rather than the room's 404, and why a |
| 154 | // mis-headered client is told to fix its header instead of |
| 155 | // being sent round the fresh-stream loop. |
| 156 | streamJSON(w, http.StatusNotAcceptable, map[string]any{"error": "accept"}) |
| 157 | return |
| 158 | } |
| 159 | |
| 160 | var sub *hub.Sub |
| 161 | var peer hub.PeerID |
| 162 | var first []hub.Message |
| 163 | |
| 164 | if resume := hub.PeerID(r.URL.Query().Get("peer")); resume != "" { |
| 165 | // A resume is authorized by the participation, not by standing |
| 166 | // (§2.5), but the meeting itself must still be live: End writes |
| 167 | // no headstone for a room nobody opened, so ended_at is the only |
| 168 | // authority a fresh process has. |
| 169 | m, found, err := a.store.meetingBySqid(r.Context(), sqid) |
| 170 | if err != nil { |
| 171 | a.logger.Error("calls: stream: load meeting", "err", err) |
| 172 | streamJSON(w, http.StatusInternalServerError, nil) |
| 173 | return |
| 174 | } |
| 175 | if !found || m.EndedAt != nil { |
| 176 | // The hub answers first when it holds a tombstone, so a peer |
| 177 | // that was in the room when it ended is told why (410 ended) |
| 178 | // rather than sent to open a fresh stream. |
| 179 | if re := a.hub.CanResume(r.Context(), sqid, session, resume); re != nil { |
| 180 | a.refuseResume(w, re) |
| 181 | return |
| 182 | } |
| 183 | a.refuseResume(w, &hub.ResumeError{Status: http.StatusNotFound, Boot: a.hub.Boot()}) |
| 184 | return |
| 185 | } |
| 186 | if probe { |
| 187 | if re := a.hub.CanResume(r.Context(), sqid, session, resume); re != nil { |
| 188 | a.refuseResume(w, re) |
| 189 | return |
| 190 | } |
| 191 | streamJSON(w, http.StatusNoContent, nil) |
| 192 | return |
| 193 | } |
| 194 | var last map[hub.PeerID]uint64 |
| 195 | if raw := r.URL.Query().Get("last"); raw != "" { |
| 196 | // One seq per peer, sixteen peers to a room: a document this |
| 197 | // far past that is not a browser's, and is refused before it |
| 198 | // is parsed. |
| 199 | if len(raw) > maxLastQuery { |
| 200 | streamJSON(w, http.StatusBadRequest, map[string]any{"error": "bad last"}) |
| 201 | return |
| 202 | } |
| 203 | if err := json.Unmarshal([]byte(raw), &last); err != nil { |
| 204 | streamJSON(w, http.StatusBadRequest, map[string]any{"error": "bad last"}) |
| 205 | return |
| 206 | } |
| 207 | } |
| 208 | res, err := a.hub.Resume(r.Context(), sqid, session, resume, last) |
| 209 | if err != nil { |
| 210 | re, ok := err.(*hub.ResumeError) |
| 211 | if !ok { |
| 212 | a.logger.Error("calls: stream: resume", "err", err) |
| 213 | streamJSON(w, http.StatusInternalServerError, nil) |
| 214 | return |
| 215 | } |
| 216 | a.refuseResume(w, re) |
| 217 | return |
| 218 | } |
| 219 | sub, peer = res.Sub, resume |
| 220 | // Order matters: the roster, then the replay, then the gaps, all |
| 221 | // written before anything the live subscriber carries. The |
| 222 | // subscriber is live from the moment Resume returned, and the |
| 223 | // browser discards what is at or below the seq it already has, |
| 224 | // so a fresh candidate written ahead of the replay would make |
| 225 | // the replay look stale. |
| 226 | first = append(first, hub.Message{Type: hub.TypeHello, Peer: peer, Boot: a.hub.Boot(), Report: res.ReportToken}) |
| 227 | if res.Admitted { |
| 228 | first = append(first, hub.Message{Type: hub.TypeRoster, Roster: res.Roster}) |
| 229 | first = append(first, res.Replay...) |
| 230 | for _, from := range res.Gap { |
| 231 | first = append(first, hub.Message{Type: hub.TypeGap, From: from}) |
| 232 | } |
| 233 | } else { |
| 234 | // Waiting or unjoined: nothing about the room (§4.5). |
| 235 | first = append(first, hub.Message{Type: hub.TypeState, Waiting: true}) |
| 236 | } |
| 237 | } else { |
| 238 | m, found, err := a.store.meetingBySqid(r.Context(), sqid) |
| 239 | if err != nil { |
| 240 | a.logger.Error("calls: stream: load meeting", "err", err) |
| 241 | streamJSON(w, http.StatusInternalServerError, nil) |
| 242 | return |
| 243 | } |
| 244 | if !found || m.EndedAt != nil { |
| 245 | streamJSON(w, http.StatusNotFound, map[string]any{"boot": a.hub.Boot()}) |
| 246 | return |
| 247 | } |
| 248 | st, err := a.standingFor(r.Context(), person, m) |
| 249 | if err != nil { |
| 250 | a.logger.Error("calls: stream: standing", "err", err) |
| 251 | streamJSON(w, http.StatusInternalServerError, nil) |
| 252 | return |
| 253 | } |
| 254 | if st == StandingNone { |
| 255 | // Not this person's meeting to enter: 404, never 403 (§2.5). |
| 256 | streamJSON(w, http.StatusNotFound, map[string]any{"boot": a.hub.Boot()}) |
| 257 | return |
| 258 | } |
| 259 | if probe { |
| 260 | if existing, live := a.hub.LiveForSession(sqid, session); live { |
| 261 | streamJSON(w, http.StatusConflict, map[string]any{"peer": string(existing)}) |
| 262 | return |
| 263 | } |
| 264 | streamJSON(w, http.StatusNoContent, nil) |
| 265 | return |
| 266 | } |
| 267 | kind := hub.KindGuest |
| 268 | if st == StandingMember { |
| 269 | kind = hub.KindMember |
| 270 | } |
| 271 | res, err := a.hub.Open(sqid, session, person.ID, person.Name, kind) |
| 272 | switch err { |
| 273 | case nil: |
| 274 | case hub.ErrSessionHasPeer: |
| 275 | streamJSON(w, http.StatusConflict, map[string]any{"peer": string(res.Existing)}) |
| 276 | return |
| 277 | case hub.ErrRoomFull, hub.ErrPersonStreams: |
| 278 | streamJSON(w, http.StatusServiceUnavailable, map[string]any{"error": "full"}) |
| 279 | return |
| 280 | default: |
| 281 | // ErrUnknownPeer here is the room's headstone: the meeting |
| 282 | // ended between the read above and the hub lock. |
| 283 | streamJSON(w, http.StatusNotFound, map[string]any{"boot": a.hub.Boot()}) |
| 284 | return |
| 285 | } |
| 286 | sub, peer = res.Sub, res.Peer |
| 287 | first = append(first, hub.Message{Type: hub.TypeHello, Peer: peer, Boot: a.hub.Boot(), Report: res.ReportToken}) |
| 288 | } |
| 289 | |
| 290 | // From here the participation exists and is this handler's to hold |
| 291 | // open. Every exit detaches it — this handler's own subscriber, so a |
| 292 | // stream displaced by a second one does not detach its replacement — |
| 293 | // which starts the thirty-second resume window (§4.5). |
| 294 | defer a.hub.Detach(sqid, peer, sub) |
| 295 | a.serveStream(w, r, sub, first) |
| 296 | } |
| 297 | |
| 298 | // serveStream writes the events already decided, then the subscriber's, |
| 299 | // until the client goes, the TTL expires, or the stream is lost. |
| 300 | func (a *app) serveStream(w http.ResponseWriter, r *http.Request, sub *hub.Sub, first []hub.Message) { |
| 301 | rc := http.NewResponseController(w) |
| 302 | w.Header().Set("Content-Type", "text/event-stream") |
| 303 | w.Header().Set("Cache-Control", "no-store") |
| 304 | // Nothing between this process and the browser should buffer an |
| 305 | // event stream; nginx and its imitators read this header. |
| 306 | w.Header().Set("X-Accel-Buffering", "no") |
| 307 | w.WriteHeader(http.StatusOK) |
| 308 | if err := rc.Flush(); err != nil { |
| 309 | return |
| 310 | } |
| 311 | for _, m := range first { |
| 312 | if !writeEvent(w, rc, m) { |
| 313 | return |
| 314 | } |
| 315 | } |
| 316 | heartbeat := time.NewTicker(a.heartbeatEvery()) |
| 317 | defer heartbeat.Stop() |
| 318 | ttl := time.NewTimer(a.streamLifeFor(r)) |
| 319 | defer ttl.Stop() |
| 320 | for { |
| 321 | select { |
| 322 | case <-r.Context().Done(): |
| 323 | return |
| 324 | case <-a.done: |
| 325 | // The process is going down. Shutdown waits for handlers and |
| 326 | // does not cancel their contexts, so a live stream would |
| 327 | // hold the whole ten-second deadline and turn a clean stop |
| 328 | // into a failed one. The browser resumes, as it does at the |
| 329 | // TTL. |
| 330 | return |
| 331 | case <-ttl.C: |
| 332 | return |
| 333 | case <-heartbeat.C: |
| 334 | if !writeRaw(w, rc, ": hb\n\n") { |
| 335 | return |
| 336 | } |
| 337 | case m, ok := <-sub.C(): |
| 338 | if !ok { |
| 339 | return |
| 340 | } |
| 341 | if !writeEvent(w, rc, m) { |
| 342 | return |
| 343 | } |
| 344 | if m.Type == hub.TypeLost || m.Type == hub.TypeEnded { |
| 345 | // Terminal for this stream either way: lost{buffer} is |
| 346 | // followed by a resume, the rest by a screen that says |
| 347 | // why (§4.5). |
| 348 | return |
| 349 | } |
| 350 | } |
| 351 | } |
| 352 | } |
| 353 | |
| 354 | // writeEvent puts one message on the wire as an SSE frame. |
| 355 | // |
| 356 | // The encoder, rather than json.Marshal, for one reason: Marshal |
| 357 | // HTML-escapes `<`, `>` and `&` into <, > and &, and it |
| 358 | // does so inside a json.RawMessage too. A chat of 2048 literal `<` — |
| 359 | // validated at 2059 raw bytes, inside §4.4's cap — left this process |
| 360 | // as 12,299 payload bytes, six times what was measured (Codex S4-2). |
| 361 | // SSE data is not HTML and no reader of this stream treats it as |
| 362 | // markup: room.js parses the JSON and puts every string it contains on |
| 363 | // the page with textContent. Nothing here depends on the escaping. |
| 364 | // |
| 365 | // What the escaping never protected against, and what actually matters |
| 366 | // on a line-oriented wire, is a payload ending its own frame: JSON |
| 367 | // escapes a newline inside a string as \n rather than emitting one, so |
| 368 | // no value can produce the blank line that dispatches an event. |
| 369 | // callstest's TestChatCannotForgeASecondSSEFrame is the gate on that. |
| 370 | func writeEvent(w http.ResponseWriter, rc *http.ResponseController, m hub.Message) bool { |
| 371 | var buf bytes.Buffer |
| 372 | enc := json.NewEncoder(&buf) |
| 373 | enc.SetEscapeHTML(false) |
| 374 | if err := enc.Encode(m); err != nil { |
| 375 | return false |
| 376 | } |
| 377 | // Encode writes its own trailing newline; the frame below supplies |
| 378 | // the two this wire format needs. |
| 379 | b := bytes.TrimRight(buf.Bytes(), "\n") |
| 380 | return writeRaw(w, rc, fmt.Sprintf("event: %s\ndata: %s\n\n", m.Type, b)) |
| 381 | } |
| 382 | |
| 383 | // writeRaw writes one SSE frame under a deadline, so a client that has |
| 384 | // stopped reading cannot hold a goroutine and a participation open. |
| 385 | func writeRaw(w http.ResponseWriter, rc *http.ResponseController, s string) bool { |
| 386 | _ = rc.SetWriteDeadline(time.Now().Add(streamWriteWindow)) |
| 387 | if _, err := fmt.Fprint(w, s); err != nil { |
| 388 | return false |
| 389 | } |
| 390 | return rc.Flush() == nil |
| 391 | } |
| 392 | |
amadan