返回 DeepSeek-Reasonix
browser_broker.go
根目录 / desktop / browser_broker.go
1 package main
2
3 import (
4 "context"
5 "crypto/rand"
6 "encoding/hex"
7 "fmt"
8 "net"
9 "net/http"
10 "os"
11 "strings"
12 "sync"
13 "time"
14
15 "reasonix/internal/browser"
16 "reasonix/internal/remote/forward"
17 )
18
19 // The desktop browser broker is the local end of the remote browser channel:
20 // a loopback listener behind SSH reverse forwards, keyed by per-generation
21 // tokens that a reconnect revokes at once.
22
23 // browserBrokerForwardName prefixes the per-host reverse forward the remote
24 // serve's REASONIX_BROWSER_BROKER endpoint points at.
25 const browserBrokerForwardName = "browser-broker:"
26
27 // browserBrokerRoute binds one token to one host connection generation.
28 type browserBrokerRoute struct {
29 hostID string
30 gen *managedHost
31 ctx context.Context
32 cancel context.CancelFunc
33 }
34
35 // browserSessionResolution is what the broker resolves one request's session
36 // header into: the desktop executor bound to that session's tab and the
37 // workspace whose SFTP scratch area relays captures back.
38 type browserSessionResolution struct {
39 exec browser.Executor
40 workspace string
41 }
42
43 // browserSessionResolver maps (host, remote session path) to the desktop
44 // executor that owns it. An unknown or foreign session must fail with
45 // browser.ErrNoGrant so the wire handler answers 409 no_grant.
46 type browserSessionResolver func(hostID, sessionPath string) (browserSessionResolution, error)
47
48 type browserBroker struct {
49 lifecycleMu sync.Mutex
50 mu sync.Mutex
51 ln net.Listener
52 server *http.Server
53 port int
54 routes map[string]*browserBrokerRoute
55 byHost map[string]string
56 resolve browserSessionResolver
57 // current reports whether gen is still the live connection for hostID;
58 // a replaced generation's token stops authenticating immediately.
59 current func(hostID string, gen *managedHost) bool
60 // connFor returns the generation's SSH client for the capture relay.
61 connFor func(hostID string, gen *managedHost) sftpConn
62 // newRelay builds the capture relay for a connection; nil uses the SFTP
63 // relay. Tests substitute a fake.
64 newRelay func(conn sftpConn) FileRelay
65 onRevoke func(hostID string)
66 }
67
68 func newBrowserBroker(resolve browserSessionResolver, current func(string, *managedHost) bool, connFor func(string, *managedHost) sftpConn) *browserBroker {
69 return &browserBroker{
70 routes: map[string]*browserBrokerRoute{},
71 byHost: map[string]string{},
72 resolve: resolve,
73 current: current,
74 connFor: connFor,
75 }
76 }
77
78 // register mints a fresh token for (hostID, gen), replacing the host's
79 // previous token. Returns the token and the broker's loopback port.
80 func (b *browserBroker) register(hostID string, gen *managedHost) (string, int, error) {
81 b.lifecycleMu.Lock()
82 defer b.lifecycleMu.Unlock()
83 buf := make([]byte, 32)
84 if _, err := rand.Read(buf); err != nil {
85 return "", 0, fmt.Errorf("browser broker: mint token: %w", err)
86 }
87 token := hex.EncodeToString(buf)
88 b.mu.Lock()
89 if b.ln == nil {
90 b.mu.Unlock()
91 return "", 0, fmt.Errorf("browser broker: not running")
92 }
93 replaced := false
94 if old := b.byHost[hostID]; old != "" {
95 if route := b.routes[old]; route != nil && route.cancel != nil {
96 route.cancel()
97 }
98 delete(b.routes, old)
99 replaced = true
100 }
101 ctx, cancel := context.WithCancel(context.Background())
102 b.routes[token] = &browserBrokerRoute{hostID: hostID, gen: gen, ctx: ctx, cancel: cancel}
103 b.byHost[hostID] = token
104 port := b.port
105 b.mu.Unlock()
106 if replaced && b.onRevoke != nil {
107 b.onRevoke(hostID)
108 }
109 return token, port, nil
110 }
111
112 // revokeHost drops every token minted for hostID (serve stop, disconnect).
113 func (b *browserBroker) revokeHost(hostID string) {
114 b.lifecycleMu.Lock()
115 defer b.lifecycleMu.Unlock()
116 b.mu.Lock()
117 if token := b.byHost[hostID]; token != "" {
118 if route := b.routes[token]; route != nil && route.cancel != nil {
119 route.cancel()
120 }
121 delete(b.routes, token)
122 delete(b.byHost, hostID)
123 }
124 b.mu.Unlock()
125 if b.onRevoke != nil {
126 b.onRevoke(hostID)
127 }
128 }
129
130 func (b *browserBroker) close() {
131 b.lifecycleMu.Lock()
132 defer b.lifecycleMu.Unlock()
133 b.mu.Lock()
134 server, listener := b.server, b.ln
135 hosts := make([]string, 0, len(b.byHost))
136 for _, route := range b.routes {
137 if route.cancel != nil {
138 route.cancel()
139 }
140 hosts = append(hosts, route.hostID)
141 }
142 b.server, b.ln = nil, nil
143 b.routes = map[string]*browserBrokerRoute{}
144 b.byHost = map[string]string{}
145 b.mu.Unlock()
146 if b.onRevoke != nil {
147 for _, hostID := range hosts {
148 b.onRevoke(hostID)
149 }
150 }
151 if server != nil {
152 _ = server.Close()
153 }
154 if listener != nil {
155 _ = listener.Close()
156 }
157 }
158
159 func (b *browserBroker) ServeHTTP(w http.ResponseWriter, r *http.Request) {
160 // Unauthenticated liveness for the reverse-tunnel probe, mirroring the
161 // credential proxy: the listener is only reachable through the tunnel.
162 if r.URL.Path == "/healthz" {
163 w.WriteHeader(http.StatusNoContent)
164 return
165 }
166 token := bearerToken(r.Header.Get("Authorization"))
167 b.mu.Lock()
168 route := b.routes[token]
169 b.mu.Unlock()
170 if token == "" || route == nil || (b.current != nil && !b.current(route.hostID, route.gen)) {
171 w.Header().Set("WWW-Authenticate", `Bearer realm="reasonix-browser-broker"`)
172 http.Error(w, "invalid or stale browser broker token", http.StatusUnauthorized)
173 return
174 }
175 exec := &brokerSessionExecutor{broker: b, route: route}
176 if route.ctx != nil {
177 ctx, cancel := context.WithCancel(r.Context())
178 body := r.Body
179 stop := context.AfterFunc(route.ctx, func() { cancel(); _ = body.Close() })
180 defer stop()
181 defer cancel()
182 r = r.WithContext(ctx)
183 }
184 browser.NewHTTPHandler(exec, token).ServeHTTP(w, r)
185 }
186
187 // brokerSessionExecutor is the per-request executor the broker serves: every
188 // method resolves the request's session header to the desktop tab that owns
189 // it, so one host token can never drive another session's browser.
190 type brokerSessionExecutor struct {
191 broker *browserBroker
192 route *browserBrokerRoute
193 }
194
195 func (s *brokerSessionExecutor) resolve(ctx context.Context) (browserSessionResolution, error) {
196 if !s.current(ctx) {
197 return browserSessionResolution{}, browser.ErrNoGrant
198 }
199 res, err := s.broker.resolve(s.route.hostID, browser.SessionFromContext(ctx))
200 if err != nil {
201 return browserSessionResolution{}, err
202 }
203 if res.exec == nil || !s.current(ctx) {
204 return browserSessionResolution{}, browser.ErrNoGrant
205 }
206 return res, nil
207 }
208
209 func (s *brokerSessionExecutor) current(ctx context.Context) bool {
210 return ctx.Err() == nil && (s.route.ctx == nil || s.route.ctx.Err() == nil) && (s.broker.current == nil || s.broker.current(s.route.hostID, s.route.gen))
211 }
212
213 func (s *brokerSessionExecutor) Available(ctx context.Context) bool {
214 res, err := s.resolve(ctx)
215 if err != nil {
216 return false
217 }
218 if a, ok := res.exec.(browser.Availability); ok {
219 return a.Available(ctx)
220 }
221 return true
222 }
223
224 func (s *brokerSessionExecutor) Tabs(ctx context.Context) ([]browser.Tab, error) {
225 res, err := s.resolve(ctx)
226 if err != nil {
227 return nil, err
228 }
229 return res.exec.Tabs(ctx)
230 }
231
232 func (s *brokerSessionExecutor) Open(ctx context.Context, req browser.OpenRequest) (browser.Tab, error) {
233 res, err := s.resolve(ctx)
234 if err != nil {
235 return browser.Tab{}, err
236 }
237 return res.exec.Open(ctx, req)
238 }
239
240 func (s *brokerSessionExecutor) Navigate(ctx context.Context, req browser.NavigateRequest) (browser.Tab, error) {
241 res, err := s.resolve(ctx)
242 if err != nil {
243 return browser.Tab{}, err
244 }
245 return res.exec.Navigate(ctx, req)
246 }
247
248 func (s *brokerSessionExecutor) Snapshot(ctx context.Context, req browser.SnapshotRequest) (browser.Snapshot, error) {
249 res, err := s.resolve(ctx)
250 if err != nil {
251 return browser.Snapshot{}, err
252 }
253 return res.exec.Snapshot(ctx, req)
254 }
255
256 // Screenshot relays the capture file onto the remote host before answering:
257 // the path the serve receives must be local to the serve, never a desktop
258 // path it cannot read.
259 func (s *brokerSessionExecutor) Screenshot(ctx context.Context, req browser.ScreenshotRequest) (browser.Screenshot, error) {
260 res, err := s.resolve(ctx)
261 if err != nil {
262 return browser.Screenshot{}, err
263 }
264 shot, err := res.exec.Screenshot(ctx, req)
265 if err != nil {
266 return browser.Screenshot{}, err
267 }
268 shot.Path, err = s.relay(ctx, res.workspace, shot.Path)
269 if err != nil {
270 return browser.Screenshot{}, err
271 }
272 return shot, nil
273 }
274
275 func (s *brokerSessionExecutor) Downloads(ctx context.Context, req browser.DownloadsRequest) ([]browser.Download, error) {
276 res, err := s.resolve(ctx)
277 if err != nil {
278 return nil, err
279 }
280 downloads, err := res.exec.Downloads(ctx, req)
281 if err != nil {
282 return nil, err
283 }
284 for i, d := range downloads {
285 if strings.TrimSpace(d.Path) == "" {
286 continue
287 }
288 downloads[i].Path, err = s.relay(ctx, res.workspace, d.Path)
289 if err != nil {
290 return nil, err
291 }
292 }
293 return downloads, nil
294 }
295
296 func (s *brokerSessionExecutor) Act(ctx context.Context, req browser.ActRequest) (browser.ActResult, error) {
297 res, err := s.resolve(ctx)
298 if err != nil {
299 return browser.ActResult{}, err
300 }
301 if req.Action == browser.ActionUpload {
302 owner, ok := res.exec.(interface{ captureDir() (string, error) })
303 if !ok || s.broker.connFor == nil {
304 return browser.ActResult{}, fmt.Errorf("browser upload: no staging owner")
305 }
306 conn := s.broker.connFor(s.route.hostID, s.route.gen)
307 if conn == nil {
308 return browser.ActResult{}, browser.ErrNoGrant
309 }
310 newRelay := s.broker.newRelay
311 if newRelay == nil {
312 newRelay = func(c sftpConn) FileRelay { return sftpFileRelay{conn: c} }
313 }
314 relay, ok := newRelay(conn).(browserUploadRelay)
315 if !ok {
316 return browser.ActResult{}, fmt.Errorf("browser upload: relay cannot receive remote files")
317 }
318 scratch, err := owner.captureDir()
319 if err != nil {
320 return browser.ActResult{}, err
321 }
322 dir, err := os.MkdirTemp(scratch, "remote-upload-")
323 if err != nil {
324 return browser.ActResult{}, err
325 }
326 defer os.RemoveAll(dir)
327 files := make([]string, 0, len(req.Files))
328 for _, remote := range req.Files {
329 local, err := relay.Fetch(ctx, res.workspace, remote, dir)
330 if err != nil {
331 return browser.ActResult{}, err
332 }
333 files = append(files, local)
334 }
335 req.Files = files
336 }
337 if !s.current(ctx) {
338 return browser.ActResult{}, browser.ErrNoGrant
339 }
340 return res.exec.Act(ctx, req)
341 }
342
343 func (s *brokerSessionExecutor) Close(ctx context.Context, req browser.CloseRequest) error {
344 res, err := s.resolve(ctx)
345 if err != nil {
346 return err
347 }
348 return res.exec.Close(ctx, req)
349 }
350
351 // relay stages one desktop capture file onto the remote host through the
352 // connection generation's SFTP channel.
353 func (s *brokerSessionExecutor) relay(ctx context.Context, workspace, localPath string) (string, error) {
354 if strings.TrimSpace(localPath) == "" {
355 return "", nil
356 }
357 if s.broker.connFor == nil {
358 return "", fmt.Errorf("browser broker: no file relay for this connection")
359 }
360 conn := s.broker.connFor(s.route.hostID, s.route.gen)
361 if conn == nil {
362 return "", fmt.Errorf("browser broker: host %q connection is gone", s.route.hostID)
363 }
364 newRelay := s.broker.newRelay
365 if newRelay == nil {
366 newRelay = func(c sftpConn) FileRelay { return sftpFileRelay{conn: c} }
367 }
368 return newRelay(conn).Stage(ctx, workspace, localPath)
369 }
370
371 // browserBrokerPort returns the broker's loopback port, starting the listener
372 // on first use. The broker serves every remote host off one port; tokens keep
373 // the hosts apart.
374 func (a *App) browserBrokerPort() (int, error) {
375 a.browserBrokerMu.Lock()
376 defer a.browserBrokerMu.Unlock()
377 if a.browserBroker != nil {
378 return a.browserBroker.port, nil
379 }
380 if !a.hostMode() {
381 return 0, fmt.Errorf("browser broker: the desktop shell is not attached")
382 }
383 ln, err := net.Listen("tcp", "127.0.0.1:0")
384 if err != nil {
385 return 0, fmt.Errorf("browser broker: listen: %w", err)
386 }
387 b := newBrowserBroker(a.resolveRemoteBrowserSession, a.remoteHostGenerationCurrent, a.remoteHostGenerationClient)
388 b.onRevoke = a.revokeRemoteBrowserHost
389 b.ln = ln
390 b.port = ln.Addr().(*net.TCPAddr).Port
391 b.server = &http.Server{
392 Handler: b,
393 ReadHeaderTimeout: 10 * time.Second,
394 IdleTimeout: 2 * time.Minute,
395 MaxHeaderBytes: 1 << 20,
396 }
397 a.browserBroker = b
398 a.goSafe("browserBroker", func() { _ = b.server.Serve(ln) })
399 return b.port, nil
400 }
401
402 // registerBrowserBrokerRoute starts the broker if needed and mints the
403 // (host, generation) token a remote serve bootstrap hands over.
404 func (a *App) registerBrowserBrokerRoute(hostID string, gen *managedHost) (string, int, error) {
405 if _, err := a.browserBrokerPort(); err != nil {
406 return "", 0, err
407 }
408 a.browserBrokerMu.Lock()
409 b := a.browserBroker
410 a.browserBrokerMu.Unlock()
411 if b == nil {
412 return "", 0, fmt.Errorf("browser broker: not running")
413 }
414 return b.register(hostID, gen)
415 }
416
417 func (a *App) revokeBrowserBrokerRoutes(hostID string) {
418 a.browserBrokerMu.Lock()
419 b := a.browserBroker
420 a.browserBrokerMu.Unlock()
421 if b != nil {
422 b.revokeHost(hostID)
423 }
424 }
425
426 func (a *App) closeBrowserBroker() {
427 a.browserBrokerMu.Lock()
428 b := a.browserBroker
429 a.browserBroker = nil
430 a.browserBrokerMu.Unlock()
431 if b != nil {
432 b.close()
433 }
434 }
435
436 // closeRemoteBrokers tears down the loopback brokers that serve remote hosts.
437 func (a *App) closeRemoteBrokers() {
438 a.closeCredentialProxy()
439 a.closeBrowserBroker()
440 }
441
442 // remoteHostGenerationCurrent fences broker routes to their connection
443 // generation: once the manager swaps or drops the host, minted tokens die.
444 func (a *App) remoteHostGenerationCurrent(hostID string, gen *managedHost) bool {
445 a.remoteMu.Lock()
446 rt := a.remoteRuntime
447 a.remoteMu.Unlock()
448 m, ok := rt.(*desktopRemoteManager)
449 if !ok || m == nil {
450 return false
451 }
452 return m.isCurrent(hostID, gen)
453 }
454
455 func (a *App) remoteHostGenerationClient(hostID string, gen *managedHost) sftpConn {
456 a.remoteMu.Lock()
457 rt := a.remoteRuntime
458 a.remoteMu.Unlock()
459 m, ok := rt.(*desktopRemoteManager)
460 if !ok || m == nil {
461 return nil
462 }
463 m.mu.Lock()
464 defer m.mu.Unlock()
465 if m.hosts[hostID] != gen {
466 return nil
467 }
468 return gen.client
469 }
470
471 // resolveRemoteBrowserSession maps a remote serve's session path to the
472 // desktop executor of the remote tab displaying it. Any session this desktop
473 // does not show for that host is refused with browser.ErrNoGrant, so a token
474 // can never reach a foreign session's tabs.
475 func (a *App) resolveRemoteBrowserSession(hostID, sessionPath string) (browserSessionResolution, error) {
476 sessionPath = strings.TrimSpace(sessionPath)
477 if sessionPath == "" || !a.hostMode() {
478 return browserSessionResolution{}, browser.ErrNoGrant
479 }
480 a.remoteTabMu.Lock()
481 var tab *remoteTab
482 for _, t := range a.remoteTabs {
483 if t == nil || t.ref.HostID != hostID {
484 continue
485 }
486 t.sessionMu.Lock()
487 path := strings.TrimSpace(t.session.path)
488 t.sessionMu.Unlock()
489 if path != "" && path == sessionPath {
490 tab = t
491 break
492 }
493 }
494 a.remoteTabMu.Unlock()
495 if tab == nil {
496 return browserSessionResolution{}, fmt.Errorf("%w: no desktop tab serves session %s", browser.ErrNoGrant, sessionPath)
497 }
498 return browserSessionResolution{
499 exec: a.browserExecutorForRemoteTab(tab, sessionPath),
500 workspace: tab.ref.Workspace,
501 }, nil
502 }
503
504 // browserExecutorForRemoteTab returns the cached executor for one remote
505 // tab's browser surface; a session rotation re-scopes the grant.
506 func (a *App) browserExecutorForRemoteTab(tab *remoteTab, sessionPath string) browser.Executor {
507 if tab == nil || !a.hostMode() || a.browserControl.off() {
508 return nil
509 }
510 key := "remote/" + tab.id
511 a.browserExecMu.Lock()
512 defer a.browserExecMu.Unlock()
513 if a.browserExecutors == nil {
514 a.browserExecutors = map[string]*hostBrowserExecutor{}
515 }
516 if exec, ok := a.browserExecutors[key]; ok {
517 if exec.sessionKey == sessionPath {
518 return exec
519 }
520 // A session rotation creates a new immutable owner; mutating the old
521 // executor races in-flight calls and lets them inherit the new grant.
522 a.revokeBrowserExecutor(exec)
523 }
524 exec := &hostBrowserExecutor{
525 app: a, host: a.hostShell.server, tabID: tab.id,
526 grantID: newBrowserGrantID(), sessionKey: sessionPath,
527 }
528 a.browserExecutors[key] = exec
529 return exec
530 }
531
532 // ensureBrowserBrokerForward opens (idempotently) the reverse tunnel from the
533 // remote loopback to the desktop broker, mirroring the credential proxy's
534 // forward. Returns the actually bound remote port.
535 func ensureBrowserBrokerForward(c desktopSSHClient, hostID string, desktopPort int) (int, error) {
536 name := browserBrokerForwardName + hostID
537 target := fmt.Sprintf("127.0.0.1:%d", desktopPort)
538 for _, f := range c.Forwards().List() {
539 if f.Spec.Name == name && f.Spec.TargetAddr == target && f.Up {
540 if port, ok := portOfAddr(f.BoundAddr); ok {
541 return port, nil
542 }
543 }
544 }
545 bound, err := c.Forwards().Replace(forward.Spec{
546 Name: name,
547 Direction: forward.Remote,
548 BindAddr: "127.0.0.1:0",
549 TargetAddr: target,
550 })
551 if err != nil {
552 return 0, err
553 }
554 port, ok := portOfAddr(bound)
555 if !ok {
556 return 0, fmt.Errorf("browser broker: reverse tunnel bound unexpected address %q", bound)
557 }
558 return port, nil
559 }
560
560 lines GO