diff --git a/go/chat/attachment_httpsrv.go b/go/chat/attachment_httpsrv.go index e4d49e735369..505d8595c88b 100644 --- a/go/chat/attachment_httpsrv.go +++ b/go/chat/attachment_httpsrv.go @@ -121,13 +121,11 @@ func (r *AttachmentHTTPSrv) genURLKey(prefix string, payload any) (string, error } func (r *AttachmentHTTPSrv) getURL(ctx context.Context, prefix string, payload any) string { - if !r.httpSrv.Active() { - r.Debug(ctx, "getURL: http server failed to start earlier") - return "" - } + // Addr fails only before the server first binds; while it is stopped it + // returns where the server comes back. addr, err := r.httpSrv.Addr() if err != nil { - r.Debug(ctx, "getURL: failed to get HTTP server address: %s", err) + r.Debug(ctx, "getURL: no HTTP server address: %s", err) return "" } key, err := r.genURLKey(prefix, payload) @@ -149,6 +147,10 @@ func (r *AttachmentHTTPSrv) GetURL(ctx context.Context, convID chat1.Conversatio ConvID: convID, MsgID: msgID, }) + if url == "" { + // Without a server there is no URL; the query alone would be a garbage one. + return "" + } url += fmt.Sprintf("&prev=%v&noanim=%v&isemoji=%v", preview, noAnim, isEmoji) r.Debug(ctx, "GetURL: handler URL: convID: %s msgID: %d %s", convID, msgID, url) return url diff --git a/go/chat/attachment_httpsrv_appstate_test.go b/go/chat/attachment_httpsrv_appstate_test.go new file mode 100644 index 000000000000..c8317469b618 --- /dev/null +++ b/go/chat/attachment_httpsrv_appstate_test.go @@ -0,0 +1,90 @@ +package chat + +import ( + "context" + "net" + "strings" + "testing" + "time" + + "github.com/keybase/client/go/chat/globals" + "github.com/keybase/client/go/chat/types" + "github.com/keybase/client/go/externalstest" + "github.com/keybase/client/go/kbhttp/manager" + "github.com/keybase/client/go/libkb" + "github.com/keybase/client/go/protocol/chat1" + "github.com/keybase/client/go/protocol/keybase1" + "github.com/stretchr/testify/require" +) + +type startOnlyAttachmentFetcher struct { + types.AttachmentFetcher +} + +func (startOnlyAttachmentFetcher) OnStart(libkb.MetaContext) {} + +// requireSrvServing waits until the server does or does not accept connections +// at the address it hands out. +func requireSrvServing(t *testing.T, srv *manager.Srv, serving bool) { + t.Helper() + require.Eventually(t, func() bool { + addr, err := srv.Addr() + if err != nil { + return false + } + conn, err := net.DialTimeout("tcp", addr, time.Second) + if err == nil { + conn.Close() + } + return (err == nil) == serving + }, 10*time.Second, time.Millisecond, "server serving != %v", serving) +} + +func TestGetURLWhileStoppedUsesLastAddress(t *testing.T) { + tc := externalstest.SetupTest(t, "attachment-url-stopped", 0) + defer tc.Cleanup() + tc.G.ConnectionManager = libkb.NewConnectionManager() + tc.G.MobileAppState.Update(keybase1.MobileAppState_FOREGROUND) + g := globals.NewContext(tc.G, &globals.ChatContext{}) + httpSrv := manager.NewSrv(tc.G) + srv := NewAttachmentHTTPSrv(g, httpSrv, startOnlyAttachmentFetcher{}, nil) + g.AttachmentURLSrv = srv + emoji := NewDevConvEmojiSource(g, nil) + ctx := context.TODO() + convID := chat1.ConversationID([]byte{1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16}) + msg := chat1.EmojiMessage{ConvID: convID, MsgID: 3} + + type urls struct { + full, preview, emoji, emojiNoAnim, emojiNoAnimOnly string + } + get := func() urls { + var res urls + res.full = srv.GetURL(ctx, msg.ConvID, msg.MsgID, false, false, false) + res.preview = srv.GetURL(ctx, msg.ConvID, msg.MsgID, true, false, false) + source, noAnimSource, err := emoji.RemoteToLocalSource(ctx, chat1.NewEmojiRemoteSourceWithMessage(msg), false) + require.NoError(t, err) + res.emoji, res.emojiNoAnim = source.Httpsrv(), noAnimSource.Httpsrv() + source, _, err = emoji.RemoteToLocalSource(ctx, chat1.NewEmojiRemoteSourceWithMessage(msg), true) + require.NoError(t, err) + res.emojiNoAnimOnly = source.Httpsrv() + return res + } + + requireSrvServing(t, httpSrv, true) + addr, err := httpSrv.Addr() + require.NoError(t, err) + prefix := "http://" + addr + "/" + up := get() + for _, url := range []string{up.full, up.preview, up.emoji, up.emojiNoAnim, up.emojiNoAnimOnly} { + require.True(t, strings.HasPrefix(url, prefix), "url %q while serving", url) + } + require.Contains(t, up.preview, "&prev=true") + + tc.G.MobileAppState.Update(keybase1.MobileAppState_BACKGROUND) + requireSrvServing(t, httpSrv, false) + require.Equal(t, up, get()) + + tc.G.MobileAppState.Update(keybase1.MobileAppState_FOREGROUND) + requireSrvServing(t, httpSrv, true) + require.Equal(t, up, get()) +} diff --git a/go/kbfs/libhttpserver/server.go b/go/kbfs/libhttpserver/server.go index fb711fb8233a..35470bbb2d2b 100644 --- a/go/kbfs/libhttpserver/server.go +++ b/go/kbfs/libhttpserver/server.go @@ -8,10 +8,10 @@ import ( "context" "crypto/rand" "encoding/base64" - "errors" "io" "net/http" "path" + "runtime" "strings" "sync" "time" @@ -24,6 +24,7 @@ import ( "github.com/keybase/client/go/kbfs/libmime" "github.com/keybase/client/go/kbfs/tlf" "github.com/keybase/client/go/kbhttp" + "github.com/keybase/client/go/kbhttp/manager" "github.com/keybase/client/go/libkb" "github.com/keybase/client/go/logger" "github.com/keybase/client/go/protocol/keybase1" @@ -33,11 +34,9 @@ const fsCacheSize = 64 // Server is a local HTTP server for serving KBFS content over HTTP. type Server struct { - config libkbfs.Config - logger logger.Logger - vlog *libkb.VDebugLog - appStateUpdater env.AppStateUpdater - cancel func() + config libkbfs.Config + logger logger.Logger + vlog *libkb.VDebugLog tokenLock sync.RWMutex token string @@ -45,8 +44,7 @@ type Server struct { fs *lru.Cache - serverLock sync.RWMutex - server *kbhttp.Srv + server *manager.Srv } const ( @@ -221,67 +219,15 @@ const ( requestPathRoot = "/files/" ) -func (s *Server) restart() (err error) { - s.serverLock.Lock() - defer s.serverLock.Unlock() - if s.server != nil { - s.server.Stop() - err = s.server.Start() - } - if s.server == nil || - // If pinned port is in use, just pick a new one like we never had a - // server before. - errors.Is(err, kbhttp.ErrPinnedPortInUse) { - s.server = kbhttp.NewSrv(s.logger, - kbhttp.NewRandomPortRangeListenerSource(portStart, portEnd)) - err = s.server.Start() - } - if err != nil { - return err - } - // Have to start this first to populate the ServeMux object. - s.server.Handle(requestPathRoot, - http.StripPrefix(requestPathRoot, http.HandlerFunc(s.serve))) - return nil -} - -func (s *Server) monitorAppState(ctx context.Context) { - state := keybase1.MobileAppState_FOREGROUND - for { - select { - case <-ctx.Done(): - return - case <-s.appStateUpdater.NextAppStateUpdate(state): - state = s.appStateUpdater.AppState() - // Due to the way NextUpdate is designed, it's possible we miss an - // update if processing the last update takes too long. So it's - // possible to get consecutive FOREGROUND updates even if there are - // other states in-between. Since libkb/appstate.go already - // deduplicates, it'll never actually send consecutive identical - // states to us. In addition, apart from FOREGROUND/BACKGROUND, - // there are other possible states too, and potentially more in the - // future. So, we just restart the server under FOREGROUND instead - // of trying to listen on all state updates. - if state != keybase1.MobileAppState_FOREGROUND { - continue - } - if err := s.restart(); err != nil { - s.logger.Error("(Re)starting server failed: %v", err) - } - } - } -} - // New creates and starts a new server. func New(appStateUpdater env.AppStateUpdater, config libkbfs.Config) ( s *Server, err error, ) { logger := config.MakeLogger("HTTP") s = &Server{ - appStateUpdater: appStateUpdater, - config: config, - logger: logger, - vlog: config.MakeVLogger(logger), + config: config, + logger: logger, + vlog: config.MakeVLogger(logger), } s.fs, err = lru.NewWithEvict(fsCacheSize, func(_ any, value any) { if e, ok := value.(obsoleteTrackingFS); ok && e.unsubscribe != nil { @@ -291,30 +237,33 @@ func New(appStateUpdater env.AppStateUpdater, config libkbfs.Config) ( if err != nil { return nil, err } - if err = s.restart(); err != nil { + // A failed first start is fatal here: the retry rides on app state changes, + // and on desktop -- which runs this server too -- the app state never moves. + s.server, err = manager.New("kbfsHTTP", logger, appStateUpdater.AppState, appStateUpdater.NextAppStateUpdate, + func() kbhttp.ListenerSource { + return kbhttp.NewRandomPortRangeListenerSource(portStart, portEnd) + }, runtime.GOOS != "android", func(context.Context, keybase1.HttpSrvInfo) {}) + if err != nil { + s.server.Shutdown() return nil, err } - ctx, cancel := context.WithCancel(context.Background()) - go s.monitorAppState(ctx) - s.cancel = cancel + // The token is checked in serve. No one has the address before New + // returns, so registering after the first start answers no request with a 404. + s.server.HandleFunc(strings.TrimPrefix(requestPathRoot, "/"), manager.SrvTokenModeUnchecked, + http.StripPrefix(requestPathRoot, http.HandlerFunc(s.serve)).ServeHTTP) libmime.Patch(additionalMimeTypes) return s, nil } // Address returns the address that the server is listening on. func (s *Server) Address() (string, error) { - s.serverLock.RLock() - defer s.serverLock.RUnlock() return s.server.Addr() } // Shutdown shuts down the server. func (s *Server) Shutdown() { - s.serverLock.Lock() - defer s.serverLock.Unlock() - s.server.Stop() + s.server.Shutdown() // Purge the LRU so its evict callback runs and unsubscribes any // folder-branch observers still held by cached entries. s.fs.Purge() - s.cancel() } diff --git a/go/kbhttp/manager/manager.go b/go/kbhttp/manager/manager.go index a5db6f19e570..5a6b529e16bc 100644 --- a/go/kbhttp/manager/manager.go +++ b/go/kbhttp/manager/manager.go @@ -8,9 +8,11 @@ import ( "net/http" "runtime" "sync" + "sync/atomic" "github.com/keybase/client/go/kbhttp" "github.com/keybase/client/go/libkb" + "github.com/keybase/client/go/logger" "github.com/keybase/client/go/protocol/keybase1" ) @@ -26,119 +28,237 @@ type srvEndpoint struct { serve func(w http.ResponseWriter, req *http.Request) } +type handlerRequest struct { + endpoint string + desc srvEndpoint + done chan struct{} +} + +// Srv runs a local HTTP server. One goroutine, run, owns it: only run starts +// and stops it, reacting to app state changes, handler registrations and +// shutdown. type Srv struct { - libkb.Contextified + name string // prefixes every log line, so each server's lines are told apart + log logger.Logger + // appState reads the current app state and nextAppState waits for the next + // change, as libkb.MobileAppState and kbfs's env.AppStateUpdater spell them. + appState func() keybase1.MobileAppState + nextAppState func(lastState keybase1.MobileAppState) <-chan struct{} + // token is set once and kept across restarts, so URLs handed out before a restart keep working. + token string + listenerSource func() kbhttp.ListenerSource + stopInBackground bool // false on Android, where the server stays up in every state + // notify runs on run, so it must not call HandleFunc. It runs only when the + // bound address changes, the first bind included. + notify func(context.Context, keybase1.HttpSrvInfo) + + // status is the last address the server bound, kept while it is stopped so + // URLs built then point where it comes back; empty only until the first + // bind. Readers never wait on run. + status atomic.Pointer[keybase1.HttpSrvInfo] + handlers chan handlerRequest + shutdownOnce sync.Once + shutdownCh chan struct{} + done chan struct{} + // Owned by run. httpSrv *kbhttp.Srv endpoints map[string]srvEndpoint - token string - startMu sync.Mutex + state keybase1.MobileAppState } +// NewSrv runs the service's HTTP server until the service shuts down. It reads +// g.NotifyRouter once, now, to announce every address it binds. func NewSrv(g *libkb.GlobalContext) *Srv { - h := &Srv{ - Contextified: libkb.NewContextified(g), - endpoints: make(map[string]srvEndpoint), - } - h.initHTTPSrv() - h.startHTTPSrv() - g.PushShutdownHook(func(mctx libkb.MetaContext) error { - h.httpSrv.Stop() + listenerSource := func() kbhttp.ListenerSource { + return kbhttp.NewRandomPortRangeListenerSource(g.GetEnv().GetAttachmentHTTPStartPort(), 18000) + } + notifyRouter := g.NotifyRouter + // A failed start is logged, and the next app state change tries again. + r, _ := New("Srv", g.GetLog(), g.MobileAppState.State, g.MobileAppState.NextUpdate, listenerSource, + runtime.GOOS != "android", func(ctx context.Context, info keybase1.HttpSrvInfo) { + // e2e tests match this line; only this server logs it. + g.GetLog().CDebugf(ctx, "Srv: start: addr: %s token: %s", info.Address, TokenPrefix(info.Token)) + notifyRouter.HandleHTTPSrvInfoUpdate(ctx, info) + }) + g.PushShutdownHook(func(libkb.MetaContext) error { + r.Shutdown() return nil }) - go h.monitorAppState() - return h + return r +} + +// New returns a server that has acted on the current app state, with the error +// of that first start, if any. The server runs until Shutdown either way, and +// the next app state change tries again -- but only where the app state moves, +// which is mobile, so a caller on desktop decides for itself whether a failed +// first start is fatal. +func New(name string, log logger.Logger, appState func() keybase1.MobileAppState, + nextAppState func(lastState keybase1.MobileAppState) <-chan struct{}, + listenerSource func() kbhttp.ListenerSource, stopInBackground bool, + notify func(context.Context, keybase1.HttpSrvInfo), +) (*Srv, error) { + token, _ := libkb.RandHexString("", 32) + r := &Srv{ + name: name, + log: log, + appState: appState, + nextAppState: nextAppState, + token: token, + listenerSource: listenerSource, + stopInBackground: stopInBackground, + notify: notify, + handlers: make(chan handlerRequest), + shutdownCh: make(chan struct{}), + done: make(chan struct{}), + endpoints: make(map[string]srvEndpoint), + } + // Publish an empty status before run can be observed, so readers never dereference nil. + r.status.Store(&keybase1.HttpSrvInfo{}) + r.httpSrv = kbhttp.NewSrv(r.log, r.listenerSource()) + ready := make(chan error) + go r.run(ready) + return r, <-ready } func (r *Srv) debug(ctx context.Context, msg string, args ...any) { - r.G().Log.CDebugf(ctx, "Srv: %s", fmt.Sprintf(msg, args...)) + r.log.CDebugf(ctx, "%s: %s", r.name, fmt.Sprintf(msg, args...)) +} + +// TokenPrefix shortens a token for logging. +func TokenPrefix(token string) string { + if len(token) > 8 { + return token[:8] + "..." + } + return token } -func (r *Srv) initHTTPSrv() { - startPort := r.G().GetEnv().GetAttachmentHTTPStartPort() - r.httpSrv = kbhttp.NewSrv(r.G().GetLog(), kbhttp.NewRandomPortRangeListenerSource(startPort, 18000)) +func (r *Srv) wantUp(state keybase1.MobileAppState) bool { + return !r.stopInBackground || state != keybase1.MobileAppState_BACKGROUND } -func (r *Srv) startHTTPSrv() { - r.startMu.Lock() - defer r.startMu.Unlock() +// run owns the server. ready takes the first start's error, once run has acted +// on the app state it started in and published the result. +func (r *Srv) run(ready chan<- error) { + defer close(r.done) ctx := context.Background() - token, _ := libkb.RandHexString("", 32) - maxTries := 2 - success := false - for range maxTries { - if err := r.httpSrv.Start(); err != nil { - if errors.Is(err, kbhttp.ErrPinnedPortInUse) { - // If we hit this, just try again and get a different port. - // The advantage is that backing in and out of the thread will restore attachments, - // whereas if we do nothing you need to bkg/foreground. - r.debug(ctx, "startHTTPSrv: pinned port taken error, re-initializing and trying again") - r.initHTTPSrv() - continue + r.state = r.appState() + r.debug(ctx, "run: starting up in %v", r.state) + ready <- r.reconcile(ctx) + for { + select { + case <-r.nextAppState(r.state): + prev := r.state + r.state = r.appState() + if r.leavingBackground(prev) { + r.debug(ctx, "run: rebinding on %v -> %v", prev, r.state) + r.httpSrv.Stop() + } + _ = r.reconcile(ctx) + case req := <-r.handlers: + r.endpoints[req.endpoint] = req.desc + // A stopped server has no mux; start registers every endpoint. + if r.httpSrv.Active() { + r.httpSrv.HandleFunc("/"+req.endpoint, r.checkToken(req.desc.tokenMode, req.desc.serve)) } - r.debug(ctx, "startHTTPSrv: failed to start HTTP server: %s", err) - break + close(req.done) + case <-r.shutdownCh: + <-r.httpSrv.Stop() + return } - success = true - break } - if !success { - r.debug(ctx, "startHTTPSrv: exhausted attempts to start HTTP server, giving up") - return +} + +// leavingBackground reports a move from BACKGROUND or BACKGROUNDACTIVE to +// FOREGROUND or INACTIVE where the server stops in the background. The OS can +// reclaim a suspended app's listening socket without the app reaching +// BACKGROUND, leaving a server that looks up but never accepts, so the server +// is rebound on the way back rather than trusted. +func (r *Srv) leavingBackground(prev keybase1.MobileAppState) bool { + if !r.stopInBackground { + return false + } + switch prev { + case keybase1.MobileAppState_BACKGROUND, keybase1.MobileAppState_BACKGROUNDACTIVE: + default: + return false + } + return r.state == keybase1.MobileAppState_FOREGROUND || r.state == keybase1.MobileAppState_INACTIVE +} + +// reconcile tears the server down only in BACKGROUND, and only where +// stopInBackground. INACTIVE (Control Center, system alerts, the app +// switcher) keeps it up, and every other state starts it if it isn't serving. +func (r *Srv) reconcile(ctx context.Context) error { + if !r.wantUp(r.state) { + r.httpSrv.Stop() + return nil + } + return r.start(ctx) +} + +func (r *Srv) start(ctx context.Context) error { + if r.httpSrv.Active() { + return nil + } + err := r.httpSrv.StartWithHandlers(r.registerEndpoints) + if errors.Is(err, kbhttp.ErrPinnedPortInUse) { + // Try again on a different port. Backing in and out of a thread then restores + // attachments; doing nothing would need a background/foreground. + r.debug(ctx, "start: pinned port taken, trying a new one") + r.httpSrv = kbhttp.NewSrv(r.log, r.listenerSource()) + err = r.httpSrv.StartWithHandlers(r.registerEndpoints) } - for endpoint, serveDesc := range r.endpoints { - r.HandleFunc(endpoint, serveDesc.tokenMode, serveDesc.serve) + if err != nil { + r.log.CWarningf(ctx, "%s: start: failed to start HTTP server: %s", r.name, err) + return err } addr, err := r.httpSrv.Addr() if err != nil { - r.debug(ctx, "startHTTPSrv: failed to get address after start?: %s", err) - } else { - r.debug(ctx, "startHTTPSrv: start success: addr: %s", addr) - } - r.token = token - tokenPrefix := r.token - if len(tokenPrefix) > 8 { - tokenPrefix = tokenPrefix[:8] + "..." - } - r.debug(ctx, "startHTTPSrv: addr: %s token: %s", addr, tokenPrefix) - r.G().NotifyRouter.HandleHTTPSrvInfoUpdate(ctx, keybase1.HttpSrvInfo{ - Address: addr, - Token: r.token, - }) + return err + } + if addr == r.status.Load().Address { + return nil + } + info := keybase1.HttpSrvInfo{Address: addr, Token: r.token} + // Publish before notifying, so a listener reading Info gets the address it is told about. + r.status.Store(&info) + r.notify(ctx, info) + return nil } -func (r *Srv) monitorAppState() { - ctx := context.Background() - r.debug(ctx, "monitorAppState: starting up") - state := keybase1.MobileAppState_FOREGROUND - // We don't need this on Android - if runtime.GOOS == "android" { - return - } - for { - <-r.G().MobileAppState.NextUpdate(state) - state = r.G().MobileAppState.State() - switch state { - // INACTIVE means the UI is on screen without receiving events, so the - // server has to stay up; only BACKGROUND takes it down. - case keybase1.MobileAppState_FOREGROUND, keybase1.MobileAppState_BACKGROUNDACTIVE, - keybase1.MobileAppState_INACTIVE: - r.startHTTPSrv() - case keybase1.MobileAppState_BACKGROUND: - r.httpSrv.Stop() - } +func (r *Srv) registerEndpoints(mux *http.ServeMux) { + for endpoint, desc := range r.endpoints { + mux.HandleFunc("/"+endpoint, r.checkToken(desc.tokenMode, desc.serve)) } } +// Shutdown stops the server for good and waits for run to exit. +func (r *Srv) Shutdown() { + r.shutdownOnce.Do(func() { close(r.shutdownCh) }) + <-r.done +} + func (r *Srv) HandleFunc(endpoint string, tokenMode SrvTokenMode, serve func(w http.ResponseWriter, req *http.Request), ) { - r.httpSrv.HandleFunc("/"+endpoint, func(w http.ResponseWriter, req *http.Request) { + req := handlerRequest{endpoint: endpoint, desc: srvEndpoint{tokenMode: tokenMode, serve: serve}, done: make(chan struct{})} + select { + case r.handlers <- req: + <-req.done + case <-r.done: + } +} + +func (r *Srv) checkToken(tokenMode SrvTokenMode, + serve func(w http.ResponseWriter, req *http.Request), +) http.HandlerFunc { + return func(w http.ResponseWriter, req *http.Request) { switch tokenMode { case SrvTokenModeDefault: if !hmac.Equal([]byte(req.URL.Query().Get("token")), []byte(r.token)) { r.debug(context.Background(), "HandleFunc: token failed: %s != %s", - req.URL.Query().Get("token"), r.token) + TokenPrefix(req.URL.Query().Get("token")), TokenPrefix(r.token)) w.WriteHeader(http.StatusForbidden) return } @@ -146,25 +266,23 @@ func (r *Srv) HandleFunc(endpoint string, tokenMode SrvTokenMode, // serve needs to authenticate on its own } serve(w, req) - }) - r.endpoints[endpoint] = srvEndpoint{ - tokenMode: tokenMode, - serve: serve, } } -func (r *Srv) Active() bool { - return r.httpSrv.Active() -} - func (r *Srv) Addr() (string, error) { - r.startMu.Lock() - defer r.startMu.Unlock() - return r.httpSrv.Addr() + info, err := r.Info() + return info.Address, err } -func (r *Srv) Token() string { - r.startMu.Lock() - defer r.startMu.Unlock() - return r.token +func (r *Srv) Token() string { return r.token } + +// Info returns the address and token together, for handing both to a client. +// While the server is stopped it returns where it last bound; it errors only +// if the server has never bound. +func (r *Srv) Info() (keybase1.HttpSrvInfo, error) { + info := *r.status.Load() + if info.Address == "" { + return keybase1.HttpSrvInfo{}, errors.New("server has never bound") + } + return info, nil } diff --git a/go/kbhttp/manager/manager_test.go b/go/kbhttp/manager/manager_test.go new file mode 100644 index 000000000000..05e8f7baf6b9 --- /dev/null +++ b/go/kbhttp/manager/manager_test.go @@ -0,0 +1,807 @@ +package manager + +import ( + "context" + "errors" + "fmt" + "io" + "math/rand" + "net" + "net/http" + "runtime" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/keybase/client/go/kbhttp" + "github.com/keybase/client/go/libkb" + "github.com/keybase/client/go/libkb/lifecycle/lifecycletest" + "github.com/keybase/client/go/protocol/keybase1" + "github.com/stretchr/testify/require" +) + +// listeners hands out pinned random-port listener sources, as NewSrv does, +// counts the listeners made, and remembers the last one so a test can kill it +// underneath the server. +type listeners struct { + sync.Mutex + calls int + last net.Listener +} + +type trackedSource struct { + l *listeners + src kbhttp.ListenerSource +} + +func (s trackedSource) GetListener() (net.Listener, string, error) { + listener, address, err := s.src.GetListener() + s.l.Lock() + defer s.l.Unlock() + s.l.calls++ + if err == nil { + s.l.last = listener + } + return listener, address, err +} + +func (l *listeners) source() kbhttp.ListenerSource { + return trackedSource{l: l, src: kbhttp.NewRandomPortRangeListenerSource(20000, 60000)} +} + +func (l *listeners) Calls() int { + l.Lock() + defer l.Unlock() + return l.calls +} + +func (l *listeners) kill(t *testing.T) { + l.Lock() + defer l.Unlock() + require.NoError(t, l.last.Close()) +} + +var client = &http.Client{ + Timeout: 5 * time.Second, + Transport: &http.Transport{DisableKeepAlives: true}, +} + +// appState records the wait run asks for once per turn, after acting on the +// state it read. +type appState struct { + *libkb.MobileAppState + mu sync.Mutex + wait <-chan struct{} +} + +func (a *appState) NextUpdate(last keybase1.MobileAppState) <-chan struct{} { + wait := a.MobileAppState.NextUpdate(last) + a.mu.Lock() + defer a.mu.Unlock() + a.wait = wait + return wait +} + +// apps holds each test server's app state, since Srv now takes its two +// functions rather than an interface it could be read back off. +var apps sync.Map + +func app(srv *Srv) *appState { + a, ok := apps.Load(srv) + if !ok { + // Not t.Fatal: some callers are on worker goroutines. + panic("no appState registered for this Srv") + } + return a.(*appState) +} + +// serving reports whether anything answers HTTP at the address srv hands out. +func serving(srv *Srv) bool { + info, err := srv.Info() + if err != nil { + return false + } + status, _ := fetch(info) + return status != 0 +} + +func setup(t *testing.T, state keybase1.MobileAppState, stopInBackground bool) (*Srv, *listeners) { + return setupWithNotify(t, state, stopInBackground, func(context.Context, keybase1.HttpSrvInfo) {}) +} + +func setupWithNotify(t *testing.T, state keybase1.MobileAppState, stopInBackground bool, + notify func(context.Context, keybase1.HttpSrvInfo), +) (*Srv, *listeners) { + tc := libkb.SetupTest(t, "kbhttp", 2) + t.Cleanup(tc.Cleanup) + tc.G.MobileAppState.Update(state) + l := &listeners{} + as := &appState{MobileAppState: tc.G.MobileAppState} + srv, err := New("Srv", tc.G.Log, as.State, as.NextUpdate, l.source, stopInBackground, notify) + require.NoError(t, err) + apps.Store(srv, as) + t.Cleanup(func() { apps.Delete(srv) }) + t.Cleanup(srv.Shutdown) + // New returns having acted on the launch state; HandleFunc below would wait for run anyway. + require.Equal(t, srv.wantUp(state), serving(srv), "launch state not applied when New returned") + srv.HandleFunc("test", SrvTokenModeDefault, func(w http.ResponseWriter, req *http.Request) { + fmt.Fprint(w, "ok") + }) + return srv, l +} + +func fetch(info keybase1.HttpSrvInfo) (int, error) { + return fetchPath(info, "test") +} + +// fetchPath returns the HTTP status, or 0 with an error when no response came +// back. +func fetchPath(info keybase1.HttpSrvInfo, endpoint string) (int, error) { + resp, err := client.Get(fmt.Sprintf("http://%s/%s?token=%s", info.Address, endpoint, info.Token)) + if err != nil { + return 0, err + } + defer resp.Body.Close() + body, err := io.ReadAll(resp.Body) + if err != nil { + return 0, err + } + if resp.StatusCode != http.StatusOK || string(body) != "ok" { + return resp.StatusCode, fmt.Errorf("status %d body %q", resp.StatusCode, body) + } + return resp.StatusCode, nil +} + +// waitLoop waits until run has acted on the current app state and waits for +// its next change. Handler requests are synchronous, so no event a caller made +// is still pending. +func waitLoop(t *testing.T, srv *Srv) { + t.Helper() + require.Eventually(t, func() bool { + a := app(srv) + a.mu.Lock() + wait := a.wait + a.mu.Unlock() + if wait == nil { + return false + } + select { + case <-wait: + return false + default: + return true + } + }, 10*time.Second, time.Millisecond, "run did not catch up") +} + +func requireServing(t *testing.T, srv *Srv) keybase1.HttpSrvInfo { + t.Helper() + info, err := srv.Info() + require.NoError(t, err) + _, err = fetch(info) + require.NoError(t, err) + return info +} + +func requireStopped(t *testing.T, srv *Srv) { + t.Helper() + require.False(t, serving(srv), "server still serving") +} + +func requireNeverBound(t *testing.T, srv *Srv) { + t.Helper() + _, err := srv.Info() + require.Error(t, err) +} + +// leavesBackground is a move the server rebinds on where it stops in the +// background. +func leavesBackground(from, to keybase1.MobileAppState) bool { + return (from == keybase1.MobileAppState_BACKGROUND || from == keybase1.MobileAppState_BACKGROUNDACTIVE) && + (to == keybase1.MobileAppState_FOREGROUND || to == keybase1.MobileAppState_INACTIVE) +} + +// A stopped server keeps handing out where it last bound, so URLs built while +// it is down point where it comes back. +func TestInfoKeepsLastAddressWhileStopped(t *testing.T) { + srv, _ := setup(t, keybase1.MobileAppState_FOREGROUND, true) + waitLoop(t, srv) + first := requireServing(t, srv) + + app(srv).Update(keybase1.MobileAppState_BACKGROUND) + waitLoop(t, srv) + requireStopped(t, srv) + info, err := srv.Info() + require.NoError(t, err) + require.Equal(t, first, info) + addr, err := srv.Addr() + require.NoError(t, err) + require.Equal(t, first.Address, addr) + + app(srv).Update(keybase1.MobileAppState_FOREGROUND) + waitLoop(t, srv) + require.Equal(t, first, requireServing(t, srv)) +} + +func TestNotifyOnlyOnAddressChange(t *testing.T) { + var mu sync.Mutex + var notified []keybase1.HttpSrvInfo + srv, _ := setupWithNotify(t, keybase1.MobileAppState_FOREGROUND, true, func(_ context.Context, info keybase1.HttpSrvInfo) { + mu.Lock() + defer mu.Unlock() + notified = append(notified, info) + }) + sent := func() []keybase1.HttpSrvInfo { + mu.Lock() + defer mu.Unlock() + return append([]keybase1.HttpSrvInfo(nil), notified...) + } + waitLoop(t, srv) + first := requireServing(t, srv) + require.Equal(t, []keybase1.HttpSrvInfo{first}, sent(), "first bind not announced once") + + // Every way back up on the pinned port binds the same address. + for _, next := range []keybase1.MobileAppState{ + keybase1.MobileAppState_BACKGROUND, + keybase1.MobileAppState_FOREGROUND, + keybase1.MobileAppState_BACKGROUNDACTIVE, + keybase1.MobileAppState_INACTIVE, + keybase1.MobileAppState_BACKGROUND, + keybase1.MobileAppState_BACKGROUNDACTIVE, + keybase1.MobileAppState_FOREGROUND, + } { + app(srv).Update(next) + waitLoop(t, srv) + } + require.Equal(t, first, requireServing(t, srv)) + require.Equal(t, []keybase1.HttpSrvInfo{first}, sent(), "same address announced again") + + app(srv).Update(keybase1.MobileAppState_BACKGROUND) + waitLoop(t, srv) + squatter, err := net.Listen("tcp", first.Address) + require.NoError(t, err) + defer squatter.Close() + app(srv).Update(keybase1.MobileAppState_FOREGROUND) + waitLoop(t, srv) + again := requireServing(t, srv) + require.NotEqual(t, first.Address, again.Address) + require.Equal(t, []keybase1.HttpSrvInfo{first, again}, sent(), "new address not announced once") +} + +// Leaving BACKGROUNDACTIVE stops and starts the server on its pinned port, so a +// listener the OS reclaimed while the app was suspended comes back. +func TestRebindOnLeavingBackground(t *testing.T) { + for _, to := range []keybase1.MobileAppState{ + keybase1.MobileAppState_FOREGROUND, + keybase1.MobileAppState_INACTIVE, + } { + t.Run(to.String(), func(t *testing.T) { + srv, l := setup(t, keybase1.MobileAppState_FOREGROUND, true) + waitLoop(t, srv) + first := requireServing(t, srv) + + app(srv).Update(keybase1.MobileAppState_BACKGROUNDACTIVE) + waitLoop(t, srv) + require.Equal(t, first, requireServing(t, srv)) + require.Equal(t, 1, l.Calls(), "BACKGROUNDACTIVE restarted the server") + + l.kill(t) + requireStopped(t, srv) + app(srv).Update(to) + waitLoop(t, srv) + require.Equal(t, 2, l.Calls(), "leaving BACKGROUNDACTIVE did not rebind") + require.Equal(t, first, requireServing(t, srv)) + + // Moving between up states is not leaving the background. + for _, next := range []keybase1.MobileAppState{ + keybase1.MobileAppState_FOREGROUND, + keybase1.MobileAppState_INACTIVE, + keybase1.MobileAppState_FOREGROUND, + } { + app(srv).Update(next) + waitLoop(t, srv) + } + require.Equal(t, 2, l.Calls(), "server rebound without leaving the background") + }) + } +} + +func TestNothingStartsAfterShutdown(t *testing.T) { + srv, l := setup(t, keybase1.MobileAppState_FOREGROUND, true) + waitLoop(t, srv) + requireServing(t, srv) + srv.Shutdown() + requireStopped(t, srv) + calls := l.Calls() + app(srv).Update(keybase1.MobileAppState_BACKGROUND) + app(srv).Update(keybase1.MobileAppState_FOREGROUND) + registered := make(chan struct{}) + go func() { + srv.HandleFunc("late", SrvTokenModeDefault, func(http.ResponseWriter, *http.Request) {}) + close(registered) + }() + select { + case <-registered: + case <-time.After(10 * time.Second): + require.Fail(t, "HandleFunc hung after Shutdown") + } + require.Never(t, func() bool { return serving(srv) || l.Calls() != calls }, 200*time.Millisecond, 10*time.Millisecond) +} + +// notify must see the address it announces, so a client reading Info right away gets it. +func TestInfoUpdateAnnouncesAPublishedAddress(t *testing.T) { + var srv *Srv + seen := make(chan error, 10) + srv, _ = setupWithNotify(t, keybase1.MobileAppState_BACKGROUND, true, func(_ context.Context, info keybase1.HttpSrvInfo) { + got, err := srv.Info() + if err == nil && got != info { + err = fmt.Errorf("Info %v while announcing %v", got, info) + } + seen <- err + }) + app(srv).Update(keybase1.MobileAppState_FOREGROUND) + select { + case err := <-seen: + require.NoError(t, err) + case <-time.After(10 * time.Second): + require.Fail(t, "no HTTPSrvInfoUpdate") + } +} + +func TestHandlerAddedWhileServingAnswers(t *testing.T) { + srv, _ := setup(t, keybase1.MobileAppState_FOREGROUND, true) + waitLoop(t, srv) + srv.HandleFunc("late", SrvTokenModeDefault, func(w http.ResponseWriter, _ *http.Request) { fmt.Fprint(w, "ok") }) + info, err := srv.Info() + require.NoError(t, err) + resp, err := client.Get(fmt.Sprintf("http://%s/late?token=%s", info.Address, info.Token)) + require.NoError(t, err) + defer resp.Body.Close() + require.Equal(t, http.StatusOK, resp.StatusCode) +} + +func TestInactiveKeepsServingBackgroundStops(t *testing.T) { + srv, l := setup(t, keybase1.MobileAppState_FOREGROUND, true) + waitLoop(t, srv) + first := requireServing(t, srv) + require.Equal(t, 1, l.Calls()) + + app(srv).Update(keybase1.MobileAppState_INACTIVE) + waitLoop(t, srv) + require.Equal(t, first, requireServing(t, srv)) + require.Equal(t, 1, l.Calls(), "INACTIVE restarted the server") + + app(srv).Update(keybase1.MobileAppState_BACKGROUND) + waitLoop(t, srv) + requireStopped(t, srv) + _, err := fetch(first) + require.Error(t, err) + + app(srv).Update(keybase1.MobileAppState_FOREGROUND) + waitLoop(t, srv) + again := requireServing(t, srv) + // Usually 2; another process may take the pinned port while stopped. + require.GreaterOrEqual(t, l.Calls(), 2) + require.Equal(t, first.Token, again.Token, "token changed across a restart") + require.Equal(t, first.Token, srv.Token()) + _, err = fetch(keybase1.HttpSrvInfo{Address: again.Address, Token: first.Token}) + require.NoError(t, err) +} + +func TestBackgroundLaunchStartsOnlyWhenLeavingBackground(t *testing.T) { + srv, l := setup(t, keybase1.MobileAppState_BACKGROUND, true) + require.Equal(t, 0, l.Calls(), "server started during a background launch") + requireNeverBound(t, srv) + waitLoop(t, srv) + require.Equal(t, 0, l.Calls()) + + app(srv).Update(keybase1.MobileAppState_BACKGROUNDACTIVE) + waitLoop(t, srv) + requireServing(t, srv) +} + +var allStates = []keybase1.MobileAppState{ + keybase1.MobileAppState_FOREGROUND, + keybase1.MobileAppState_INACTIVE, + keybase1.MobileAppState_BACKGROUNDACTIVE, + keybase1.MobileAppState_BACKGROUND, +} + +func TestUpUnlessBackground(t *testing.T) { + for _, initial := range allStates { + t.Run(initial.String(), func(t *testing.T) { + srv, _ := setup(t, initial, true) + for range 2 { + for _, next := range allStates { + app(srv).Update(next) + waitLoop(t, srv) + if next == keybase1.MobileAppState_BACKGROUND { + requireStopped(t, srv) + } else { + requireServing(t, srv) + } + } + } + }) + } +} + +// An INACTIVE blip neither restarts the server nor breaks a request in flight, +// and neither does a BACKGROUNDACTIVE one where the server stays up in the +// background. +func TestBlipKeepsRequestInFlight(t *testing.T) { + for _, c := range []struct { + blip keybase1.MobileAppState + stopInBackground bool + }{ + {keybase1.MobileAppState_INACTIVE, true}, + {keybase1.MobileAppState_INACTIVE, false}, + {keybase1.MobileAppState_BACKGROUNDACTIVE, false}, + } { + blip := c.blip + t.Run(fmt.Sprintf("%v-stopInBackground=%v", blip, c.stopInBackground), func(t *testing.T) { + srv, l := setup(t, keybase1.MobileAppState_FOREGROUND, c.stopInBackground) + entered, hold := make(chan struct{}), make(chan struct{}) + srv.HandleFunc("hold", SrvTokenModeDefault, func(w http.ResponseWriter, _ *http.Request) { + close(entered) + <-hold + fmt.Fprint(w, "ok") + }) + waitLoop(t, srv) + info := requireServing(t, srv) + res := make(chan error, 1) + go func() { + _, err := fetchPath(info, "hold") + res <- err + }() + select { + case <-entered: + case <-time.After(10 * time.Second): + require.Fail(t, "request did not arrive") + } + for _, state := range []keybase1.MobileAppState{blip, keybase1.MobileAppState_FOREGROUND} { + app(srv).Update(state) + waitLoop(t, srv) + } + close(hold) + require.NoError(t, <-res, "in-flight request broke across %v", blip) + require.Equal(t, info, requireServing(t, srv)) + require.Equal(t, 1, l.Calls(), "server restarted across %v", blip) + }) + } +} + +// Without stopping in the background (Android), the server serves in every +// state and is never rebound. +func TestNotStoppingInBackgroundStaysUp(t *testing.T) { + srv, l := setup(t, keybase1.MobileAppState_BACKGROUND, false) + waitLoop(t, srv) + requireServing(t, srv) + for _, next := range []keybase1.MobileAppState{ + keybase1.MobileAppState_BACKGROUNDACTIVE, + keybase1.MobileAppState_BACKGROUND, + keybase1.MobileAppState_FOREGROUND, + keybase1.MobileAppState_INACTIVE, + keybase1.MobileAppState_BACKGROUND, + } { + app(srv).Update(next) + waitLoop(t, srv) + requireServing(t, srv) + } + require.Equal(t, 1, l.Calls(), "server restarted") +} + +// brokenSource makes no listener while it is broken. +type brokenSource struct { + broken *atomic.Bool + src kbhttp.ListenerSource +} + +func (s brokenSource) GetListener() (net.Listener, string, error) { + if s.broken.Load() { + return nil, "", errors.New("no listener") + } + return s.src.GetListener() +} + +// New reports a failed first start, which kbfs treats as fatal since desktop +// has no app state change to retry on. The server is left running, and where +// the app state does move it comes up at the next change. +func TestNewReportsFirstStartErrorAndRetries(t *testing.T) { + tc := libkb.SetupTest(t, "kbhttp", 2) + defer tc.Cleanup() + tc.G.MobileAppState.Update(keybase1.MobileAppState_FOREGROUND) + broken := &atomic.Bool{} + broken.Store(true) + srv, err := New("Srv", tc.G.Log, tc.G.MobileAppState.State, tc.G.MobileAppState.NextUpdate, + func() kbhttp.ListenerSource { + return brokenSource{broken: broken, src: kbhttp.NewRandomPortRangeListenerSource(20000, 60000)} + }, true, func(context.Context, keybase1.HttpSrvInfo) {}) + require.Error(t, err) + t.Cleanup(srv.Shutdown) + requireNeverBound(t, srv) + + broken.Store(false) + tc.G.MobileAppState.Update(keybase1.MobileAppState_INACTIVE) + require.Eventually(t, func() bool { return serving(srv) }, 10*time.Second, time.Millisecond, + "server did not start on the next app state change") +} + +func TestScenarioReplay(t *testing.T) { + for _, sc := range lifecycletest.Scenarios { + t.Run(sc.Name, func(t *testing.T) { + stopInBackground := sc.Platform == lifecycletest.IOS + srv, l := setup(t, lifecycletest.InitialState, stopInBackground) + lifecycletest.Play(t, app(srv).MobileAppState, sc, func(h *lifecycletest.Harness, i int, step lifecycletest.Step) { + waitLoop(t, srv) + if !srv.wantUp(step.Want) { + if serving(srv) { + t.Fatalf("step %d %v: server up in BACKGROUND", i, step.Do) + } + return + } + info, err := srv.Info() + require.NoError(t, err) + if _, err := fetch(info); err != nil { + t.Fatalf("step %d %v: server down in %v: %v", i, step.Do, step.Want, err) + } + // Leave the server dead before a step that leaves the + // background, whose rebind must bring it back. + if stopInBackground && i+1 < len(sc.Steps) && leavesBackground(step.Want, sc.Steps[i+1].Want) { + l.kill(t) + } + }) + }) + } +} + +func TestPinnedPortTakenPicksNewAddress(t *testing.T) { + srv, l := setup(t, keybase1.MobileAppState_FOREGROUND, true) + waitLoop(t, srv) + first := requireServing(t, srv) + + // Readers race run replacing the server, for the race detector. + stop := make(chan struct{}) + var readers sync.WaitGroup + for _, read := range []func(){ + func() { _, _ = srv.Addr() }, + func() { _, _ = srv.Info() }, + } { + readers.Add(1) + go func() { + defer readers.Done() + for { + select { + case <-stop: + return + default: + } + read() + runtime.Gosched() + } + }() + } + + app(srv).Update(keybase1.MobileAppState_BACKGROUND) + waitLoop(t, srv) + requireStopped(t, srv) + squatter, err := net.Listen("tcp", first.Address) + require.NoError(t, err) + defer squatter.Close() + + app(srv).Update(keybase1.MobileAppState_FOREGROUND) + waitLoop(t, srv) + close(stop) + readers.Wait() + + again := requireServing(t, srv) + require.NotEqual(t, first.Address, again.Address) + require.Equal(t, first.Token, again.Token) + require.Equal(t, 3, l.Calls()) +} + +// requestWorker fetches until stop closes. A request racing a restart may +// fail to connect, but any response it gets must be a good one. +func requestWorker(srv *Srv, stale keybase1.HttpSrvInfo, stop chan struct{}, ok *atomic.Int64, bad chan error) { + for i := 0; ; i++ { + select { + case <-stop: + return + default: + } + info := stale + if i%2 == 0 { + var err error + if info, err = srv.Info(); err != nil { + runtime.Gosched() + continue + } + } + status, err := fetch(info) + switch { + case err == nil: + ok.Add(1) + case status != 0: + select { + case bad <- err: + default: + } + } + } +} + +func TestConcurrentRequestsDuringRestart(t *testing.T) { + srv, l := setup(t, keybase1.MobileAppState_FOREGROUND, true) + waitLoop(t, srv) + first := requireServing(t, srv) + + stop := make(chan struct{}) + bad := make(chan error, 1) + var ok atomic.Int64 + var wg sync.WaitGroup + for range 8 { + wg.Add(1) + go func() { + defer wg.Done() + requestWorker(srv, first, stop, &ok, bad) + }() + } + wg.Add(1) + go func() { + defer wg.Done() + for i := range 100 { + srv.HandleFunc(fmt.Sprintf("extra%d", i), SrvTokenModeUnchecked, func(http.ResponseWriter, *http.Request) {}) + } + }() + + for range 50 { + app(srv).Update(keybase1.MobileAppState_BACKGROUND) + waitLoop(t, srv) + app(srv).Update(keybase1.MobileAppState_FOREGROUND) + waitLoop(t, srv) + time.Sleep(time.Millisecond) + } + close(stop) + wg.Wait() + + select { + case err := <-bad: + t.Fatalf("bad response during restarts: %v", err) + default: + } + require.Positive(t, ok.Load()) + require.GreaterOrEqual(t, l.Calls(), 51) + require.Equal(t, first.Token, requireServing(t, srv).Token) +} + +// Transitions, listener deaths, handler registrations and requests racing +// each other leave a working server and no goroutines after Shutdown. +func TestStressTransitionsAndRequests(t *testing.T) { + tc := libkb.SetupTest(t, "kbhttp", 1) + defer tc.Cleanup() + baseline := runtime.NumGoroutine() + + l := &listeners{} + as := &appState{MobileAppState: tc.G.MobileAppState} + srv, err := New("Srv", tc.G.Log, as.State, as.NextUpdate, l.source, true, + func(context.Context, keybase1.HttpSrvInfo) {}) + require.NoError(t, err) + apps.Store(srv, as) + t.Cleanup(func() { apps.Delete(srv) }) + srv.HandleFunc("test", SrvTokenModeDefault, func(w http.ResponseWriter, req *http.Request) { + fmt.Fprint(w, "ok") + }) + waitLoop(t, srv) + first := requireServing(t, srv) + token := first.Token + states := []keybase1.MobileAppState{ + keybase1.MobileAppState_FOREGROUND, + keybase1.MobileAppState_BACKGROUND, + keybase1.MobileAppState_INACTIVE, + keybase1.MobileAppState_BACKGROUNDACTIVE, + } + + stop := make(chan struct{}) + bad := make(chan error, 1) + var ok atomic.Int64 + var workers, writers sync.WaitGroup + for range 4 { + workers.Add(1) + go func() { + defer workers.Done() + requestWorker(srv, first, stop, &ok, bad) + }() + } + workers.Add(1) + go func() { + defer workers.Done() + for i := 0; ; i++ { + select { + case <-stop: + return + default: + } + if i < 200 { + srv.HandleFunc(fmt.Sprintf("extra%d", i), SrvTokenModeUnchecked, func(http.ResponseWriter, *http.Request) {}) + } + _, _ = srv.Addr() + if info, err := srv.Info(); err == nil && info.Token != token { + select { + case bad <- fmt.Errorf("token changed to %s", info.Token): + default: + } + } + runtime.Gosched() + } + }() + workers.Add(1) + go func() { + defer workers.Done() + for { + select { + case <-stop: + return + case <-time.After(5 * time.Millisecond): + } + l.Lock() + if l.last != nil { + _ = l.last.Close() + } + l.Unlock() + } + }() + for w := range 4 { + writers.Add(1) + go func() { + defer writers.Done() + rng := rand.New(rand.NewSource(int64(w))) + for range 300 { + app(srv).Update(states[rng.Intn(len(states))]) + if rng.Intn(4) == 0 { + time.Sleep(time.Duration(rng.Intn(200)) * time.Microsecond) + } + } + }() + } + + done := make(chan struct{}) + go func() { + writers.Wait() + close(stop) + workers.Wait() + close(done) + }() + select { + case <-done: + case <-time.After(60 * time.Second): + t.Fatal("deadlock: transitions and requests did not finish") + } + select { + case err := <-bad: + t.Fatalf("bad response during transitions: %v", err) + default: + } + + // Leaving BACKGROUND rebinds whatever listener the killer left dead. + for _, state := range []keybase1.MobileAppState{ + keybase1.MobileAppState_BACKGROUND, + keybase1.MobileAppState_FOREGROUND, + } { + app(srv).Update(state) + waitLoop(t, srv) + } + require.Equal(t, token, requireServing(t, srv).Token) + app(srv).Update(keybase1.MobileAppState_BACKGROUND) + waitLoop(t, srv) + requireStopped(t, srv) + t.Logf("%d good responses, %d listeners", ok.Load(), l.Calls()) + + srv.Shutdown() + + deadline := time.Now().Add(10 * time.Second) + for runtime.NumGoroutine() > baseline+5 && time.Now().Before(deadline) { + time.Sleep(10 * time.Millisecond) + } + require.LessOrEqual(t, runtime.NumGoroutine(), baseline+5, "leaked goroutines") +} diff --git a/go/kbhttp/srv.go b/go/kbhttp/srv.go index 5e4b61236e1b..9c3b765e12e0 100644 --- a/go/kbhttp/srv.go +++ b/go/kbhttp/srv.go @@ -161,6 +161,13 @@ func NewSrv(log logger.Logger, listenerSource ListenerSource) *Srv { // Start starts listening on the server's listener source. func (h *Srv) Start() (err error) { + return h.StartWithHandlers(nil) +} + +// StartWithHandlers starts listening like Start, but first lets register add +// handlers to the new ServeMux, so no request can reach the server before +// they exist. +func (h *Srv) StartWithHandlers(register func(mux *http.ServeMux)) (err error) { h.Lock() defer h.Unlock() if h.server != nil { @@ -174,6 +181,9 @@ func (h *Srv) Start() (err error) { h.log.Debug("kbhttp.Srv: failed to get a listener: %s", err) return err } + if register != nil { + register(h.ServeMux) + } h.server = &http.Server{ Addr: address, Handler: h.ServeMux, diff --git a/go/kbhttp/srv_test.go b/go/kbhttp/srv_test.go index e25ff18a782a..4292d6da0f0c 100644 --- a/go/kbhttp/srv_test.go +++ b/go/kbhttp/srv_test.go @@ -6,7 +6,9 @@ package kbhttp import ( "fmt" "io" + "net" "net/http" + "sync" "testing" "github.com/keybase/client/go/logger" @@ -37,3 +39,65 @@ func TestSrv(t *testing.T) { test(NewPortRangeListenerSource(7000, 8000)) test(NewRandomPortRangeListenerSource(7000, 8000)) } + +type capturingListenerSource struct { + sync.Mutex + listener net.Listener +} + +func (c *capturingListenerSource) GetListener() (net.Listener, string, error) { + listener, address, err := NewAutoPortListenerSource().GetListener() + c.Lock() + defer c.Unlock() + c.listener = listener + return listener, address, err +} + +func (c *capturingListenerSource) kill() { + c.Lock() + defer c.Unlock() + _ = c.listener.Close() +} + +// A server whose listener died underneath it still counts as running, so the +// manager rebinds it with Stop and then Start. +func TestSrvStopStartAfterListenerDies(t *testing.T) { + source := &capturingListenerSource{} + srv := NewSrv(logger.NewTestLogger(t), source) + client := &http.Client{Transport: &http.Transport{DisableKeepAlives: true}} + get := func() error { + addr, err := srv.Addr() + if err != nil { + return err + } + resp, err := client.Get(fmt.Sprintf("http://%s/test", addr)) + if err != nil { + return err + } + defer resp.Body.Close() + out, err := io.ReadAll(resp.Body) + if err != nil { + return err + } + if string(out) != "success" { + return fmt.Errorf("unexpected body %q", out) + } + return nil + } + register := func(mux *http.ServeMux) { + mux.HandleFunc("/test", func(resp http.ResponseWriter, req *http.Request) { + fmt.Fprintf(resp, "success") + }) + } + + require.NoError(t, srv.StartWithHandlers(register)) + require.NoError(t, get()) + + source.kill() + require.Error(t, get()) + <-srv.Stop() + require.NoError(t, srv.StartWithHandlers(register)) + require.NoError(t, get()) + <-srv.Stop() + require.False(t, srv.Active()) +} diff --git a/go/service/config.go b/go/service/config.go index 70be37b8f22e..5e67f484bf51 100644 --- a/go/service/config.go +++ b/go/service/config.go @@ -15,6 +15,7 @@ import ( "github.com/keybase/client/go/engine" "github.com/keybase/client/go/install" + "github.com/keybase/client/go/kbhttp/manager" "github.com/keybase/client/go/libkb" keybase1 "github.com/keybase/client/go/protocol/keybase1" "github.com/keybase/client/go/status" @@ -360,23 +361,15 @@ func (h ConfigHandler) GetBootstrapStatus(ctx context.Context, sessionID int) (r return res, err } res = eng.Status() - m.Debug("GetBootstrapStatus: attempting to get HTTP server address") - for range 40 { // wait at most 2 seconds - addr, addrErr := h.svc.httpSrv.Addr() - if addrErr != nil { - m.Debug("GetBootstrapStatus: failed to get HTTP server address: %s", addrErr) - } else { - m.Debug("GetBootstrapStatus: http server: addr: %s token: %s", addr, h.svc.httpSrv.Token()) - res.HttpSrvInfo = &keybase1.HttpSrvInfo{ - Address: addr, - Token: h.svc.httpSrv.Token(), - } - break - } - time.Sleep(50 * time.Millisecond) - } - if res.HttpSrvInfo == nil { - m.Debug("GetBootstrapStatus: failed to get HTTP srv info after max attempts") + // Not waited on: a client learns the address from clientState and from + // HTTPSrvInfoUpdate, which the server sends whenever its address changes. + // This field is left as a convenience for a status read once the server has + // bound. + if info, infoErr := h.svc.httpSrv.Info(); infoErr != nil { + m.Debug("GetBootstrapStatus: no HTTP server address: %s", infoErr) + } else { + m.Debug("GetBootstrapStatus: http server: addr: %s token: %s", info.Address, manager.TokenPrefix(info.Token)) + res.HttpSrvInfo = &info } return res, nil } diff --git a/go/service/main.go b/go/service/main.go index 5e40dda69d5c..59f7ac42c459 100644 --- a/go/service/main.go +++ b/go/service/main.go @@ -119,7 +119,6 @@ func NewService(g *libkb.GlobalContext, isDaemon bool) *Service { teamUpgrader: teams.NewUpgrader(), walletState: stellar.NewWalletState(g, remote.NewRemoteNet(g)), offlineRPCCache: offline.NewRPCCache(g), - httpSrv: manager.NewSrv(g), initialLoginAttemptDone: make(chan struct{}), } @@ -349,6 +348,11 @@ func (d *Service) Run() (err error) { func (d *Service) SetupCriticalSubServices() error { allG := globals.NewContext(d.G(), d.ChatG()) mctx := d.MetaContext(context.TODO()) + // Not in NewService: the service sets up NotifyRouter after that, and the + // server reads it once, when created. A standalone client never sets one + // up, so both see a nil router, which announces nothing -- and nothing + // subscribes to it anyway. + d.httpSrv = manager.NewSrv(d.G()) d.G().RuntimeStats = runtimestats.NewRunner(allG) teams.ServiceInit(d.G()) stellar.ServiceInit(d.G(), d.walletState, d.badger) diff --git a/shared/chat/conversation/thread-message-state.test.tsx b/shared/chat/conversation/thread-message-state.test.tsx index 8ba2bfb62629..76def478788f 100644 --- a/shared/chat/conversation/thread-message-state.test.tsx +++ b/shared/chat/conversation/thread-message-state.test.tsx @@ -673,3 +673,74 @@ describe('addMessagesToThreadState', () => { expect(merged?.type === 'text' && merged.text.stringValue()).toBe('edited') }) }) + +// The service keeps its last-bound address in Info()/getURL once it has bound (go/kbhttp/manager +// Srv.Info), so an update takes whatever URL the service sent. +describe('local server urls', () => { + const textAt = (ord: number, override?: Omit, 'text'>) => + makeTextMessage({ + id: T.Chat.numberToMessageID(ord), + ordinal: T.Chat.numberToOrdinal(ord), + outboxID: undefined, + ...override, + }) + const attachmentOrdinal = T.Chat.numberToOrdinal(201) + + test('a new non-empty url replaces the old one', () => { + const state = makeThreadState([]) + addMessagesToThreadState(state, [makeAttachmentMessage({fileURL: 'http://127.0.0.1:5000/f'})], {}) + addMessagesToThreadState(state, [makeAttachmentMessage({fileURL: 'http://127.0.0.1:6000/f'})], {}) + expect((state.messageMap.get(attachmentOrdinal) as T.Chat.MessageAttachment).fileURL).toBe( + 'http://127.0.0.1:6000/f' + ) + }) + + test('an empty url in an update overwrites an existing one', () => { + const state = makeThreadState([]) + addMessagesToThreadState( + state, + [makeAttachmentMessage({fileURL: 'http://127.0.0.1:5000/f', previewURL: 'http://127.0.0.1:5000/p'})], + {} + ) + addMessagesToThreadState( + state, + [makeAttachmentMessage({fileURL: '', previewURL: '', title: 'renamed'})], + {} + ) + const m = state.messageMap.get(attachmentOrdinal) as T.Chat.MessageAttachment + expect(m.fileURL).toBe('') + expect(m.previewURL).toBe('') + expect(m.title).toBe('renamed') + }) + + test('reactions take the incoming decoration on a merge and on a reaction update', () => { + const reaction = (decorated: string, users: Array): T.Chat.ReactionDesc => ({ + decorated, + users: users.map((username, i) => ({timestamp: i + 1, username})), + }) + const state = makeThreadState([]) + addMessagesToThreadState( + state, + [textAt(10, {reactions: new Map([[':party:', reaction(':party:', ['testuser'])]])})], + {} + ) + addMessagesToThreadState( + state, + [textAt(10, {reactions: new Map([[':party:', reaction('', ['testuser', 'testuser-mac'])]])})], + {} + ) + const merged = (state.messageMap.get(T.Chat.numberToOrdinal(10)) as T.Chat.MessageText).reactions?.get(':party:') + expect(merged?.decorated).toBe('') + expect(merged?.users.map(u => u.username)).toEqual(['testuser', 'testuser-mac']) + + updateReactionsInThreadState(state, [ + { + reactions: new Map([[':party:', reaction(':party:', ['testuser'])]]), + targetMsgID: T.Chat.numberToMessageID(10), + }, + ]) + const updated = (state.messageMap.get(T.Chat.numberToOrdinal(10)) as T.Chat.MessageText).reactions?.get(':party:') + expect(updated?.decorated).toBe(':party:') + expect(updated?.users.map(u => u.username)).toEqual(['testuser']) + }) +}) diff --git a/shared/chat/conversation/thread-message-state.tsx b/shared/chat/conversation/thread-message-state.tsx index a2c575bb253b..bcda97f9e89b 100644 --- a/shared/chat/conversation/thread-message-state.tsx +++ b/shared/chat/conversation/thread-message-state.tsx @@ -573,8 +573,9 @@ export const updateReactionsInThreadState = ( ) const newReactions = new Map() for (const emoji of existingOrder) { - if (reactions.has(emoji)) { - newReactions.set(emoji, reactions.get(emoji)!) + const incoming = reactions.get(emoji) + if (incoming) { + newReactions.set(emoji, incoming) } } const remainingEmojis = [...reactions.keys()].filter(emoji => !newReactions.has(emoji)) diff --git a/shared/common-adapters/image.tsx b/shared/common-adapters/image.tsx index 26d22748be6b..3004aed5151e 100644 --- a/shared/common-adapters/image.tsx +++ b/shared/common-adapters/image.tsx @@ -3,6 +3,7 @@ import * as Styles from '@/styles' import type {ImageLoadEventData, ImageErrorEventData} from 'expo-image' import {Image as ExpoImage} from 'expo-image' import LoadingStateView from './loading-state-view' +import {isLocalhostSrc, retryLocalhostSrc} from './localhost-src' import type {StylesCrossPlatform} from '@/styles' import {useConfigState} from '@/stores/config' import {useShellState} from '@/stores/shell' @@ -48,13 +49,9 @@ const DesktopImage = (p: Props) => { ) } -// Srcs served by the local service http server can fail transiently: iOS stops that server -// on background/inactive and restarts it (new token, possibly new port) on foreground, so a -// load racing the restart gets connection refused. Those are worth retrying; remote srcs keep -// the old fail-once behavior. -const isLocalhostSrc = (src: Props['src']): src is string => - typeof src === 'string' && src.startsWith('http://127.0.0.1:') - +// Srcs served by the local service http server can fail transiently: iOS stops that server in +// the background and restarts it, possibly on a new port, so a load racing the restart gets +// connection refused. Those are worth retrying; remote srcs keep the old fail-once behavior. const maxRetries = 3 const NativeImage = (p: Props) => { @@ -63,6 +60,7 @@ const NativeImage = (p: Props) => { const [lastSrc, setLastSrc] = React.useState(src) const [attempt, setAttempt] = React.useState(0) const retryable = isLocalhostSrc(src) + const httpSrv = useConfigState(s => s.httpSrv) const failedRef = React.useRef(false) const triesRef = React.useRef(0) const timerRef = React.useRef>(undefined) @@ -106,8 +104,8 @@ const NativeImage = (p: Props) => { if (!retryable) return const maybeHeal = () => { if (!failedRef.current) return - // server is stopped while inactive/backgrounded; the active flip will land here again - if (useShellState.getState().mobileAppState !== 'active') return + // the server is stopped in the background; becoming active lands here again + if (useShellState.getState().mobileAppState === 'background') return failedRef.current = false triesRef.current = 0 setLoading(true) @@ -130,9 +128,8 @@ const NativeImage = (p: Props) => { } }, [retryable]) - // cache-buster forces expo-image to actually refetch; recyclingKey stays on the original - // src so the view isn't blanked by retries - const srcToUse = retryable && attempt > 0 ? `${src}${src.includes('?') ? '&' : '?'}kbRetry=${attempt}` : src + // recyclingKey stays on the original src so the view isn't blanked by retries + const srcToUse = retryable && attempt > 0 ? retryLocalhostSrc(src, attempt, httpSrv) : src const recyclingKey = typeof src === 'string' ? src : Array.isArray(src) ? src[0]?.uri : String(src) return ( diff --git a/shared/common-adapters/localhost-src.test.ts b/shared/common-adapters/localhost-src.test.ts new file mode 100644 index 000000000000..c0d324d5d1dc --- /dev/null +++ b/shared/common-adapters/localhost-src.test.ts @@ -0,0 +1,44 @@ +/// +import {isLocalhostSrc, retryLocalhostSrc} from './localhost-src' + +const httpSrv = {address: '127.0.0.1:61234', token: 'newtoken'} + +test('only local service srcs are retryable', () => { + expect(isLocalhostSrc('http://127.0.0.1:5000/av?name=testuser')).toBe(true) + expect(isLocalhostSrc('https://keybase.io/images/testuser.png')).toBe(false) + expect(isLocalhostSrc(3)).toBe(false) +}) + +test('a retry points a baked attachment url at the current server port', () => { + const src = 'http://127.0.0.1:5000/at?key=abc&prev=true&noanim=false&isemoji=false' + expect(retryLocalhostSrc(src, 1, httpSrv)).toBe( + 'http://127.0.0.1:61234/at?key=abc&prev=true&noanim=false&isemoji=false&kbRetry=1' + ) +}) + +test('a retry replaces the token param when there is one', () => { + const src = 'http://127.0.0.1:5000/av?typ=user&name=testuser&token=oldtoken&count=0' + expect(retryLocalhostSrc(src, 2, httpSrv)).toBe( + 'http://127.0.0.1:61234/av?typ=user&name=testuser&token=newtoken&count=0&kbRetry=2' + ) +}) + +test('a service restart on the same port still carries the new token', () => { + const src = 'http://127.0.0.1:61234/av?typ=user&name=testuser&token=oldtoken&count=0' + expect(retryLocalhostSrc(src, 1, {address: '127.0.0.1:61234', token: 'newtoken'})).toBe( + 'http://127.0.0.1:61234/av?typ=user&name=testuser&token=newtoken&count=0&kbRetry=1' + ) +}) + +test('a retry keeps the baked address when the current one is unknown', () => { + const src = 'http://127.0.0.1:5000/at?key=abc' + expect(retryLocalhostSrc(src, 1, {address: '', token: ''})).toBe('http://127.0.0.1:5000/at?key=abc&kbRetry=1') +}) + +test('a kbfs src keeps its own server and token', () => { + // kbfs runs a second local server on its own port with its own token + const src = 'http://127.0.0.1:7000/files/private/testuser/cat.png?token=kbfstoken&viewTypeInvariance=1' + expect(retryLocalhostSrc(src, 1, httpSrv)).toBe( + 'http://127.0.0.1:7000/files/private/testuser/cat.png?token=kbfstoken&viewTypeInvariance=1&kbRetry=1' + ) +}) diff --git a/shared/common-adapters/localhost-src.tsx b/shared/common-adapters/localhost-src.tsx new file mode 100644 index 000000000000..f2420f422777 --- /dev/null +++ b/shared/common-adapters/localhost-src.tsx @@ -0,0 +1,32 @@ +const localhostPrefix = /^http:\/\/127\.0\.0\.1:\d+/ + +// The service's own endpoints: "at" (go/chat/attachment_httpsrv.go), "av" (go/avatars/srv.go) and +// "map" (go/chat/maps/srv.go). KBFS serves /files/ from a second local server with its own port and +// its own token, so a src has to be matched against these before anything is repointed. +const serviceSrc = /^http:\/\/127\.0\.0\.1:\d+\/(?:at|av|map)(?=[?#]|$)/ + +export const isLocalhostSrc = (src: unknown): src is string => + typeof src === 'string' && localhostPrefix.test(src) + +// The service can restart its http server on a new port, but chat bakes the address into +// attachment and emoji URLs, so a retry points the src at wherever the server is now. A service +// process restart draws a fresh port and mints a new per-process token (see +// go/kbhttp/manager/manager.go), and a reconnect alone doesn't refetch already-rendered thread +// data, so the token also needs rewriting or a stale token= keeps failing forever. The +// cache-buster forces expo-image to actually refetch, and is all a non-service src gets. +export const retryLocalhostSrc = ( + src: string, + attempt: number, + httpSrv: {address: string; token: string} +) => { + let next = src + if (serviceSrc.test(src)) { + if (httpSrv.address) { + next = next.replace(localhostPrefix, `http://${httpSrv.address}`) + } + if (httpSrv.token) { + next = next.replace(/([?&]token=)[^&#]*/, `$1${httpSrv.token}`) + } + } + return `${next}${next.includes('?') ? '&' : '?'}kbRetry=${attempt}` +}