ipn, cmd/tailscale/cli: add LocalAPI IPN bus watch, Start, convert CLI

Updates #6417
Updates tailscale/corp#8051

Change-Id: I1ca360730c45ffaa0261d8422877304277fc5625
Signed-off-by: Brad Fitzpatrick <bradfitz@tailscale.com>

ipn/ipnlocal/local.go

@@ -181,7 +181,8 @@ type LocalBackend struct {
 	loginFlags       controlclient.LoginFlags
 	incomingFiles    map[*incomingFile]bool
 	fileWaiters      map[*mapSetHandle]context.CancelFunc // handle => func to call on file received
-	lastStatusTime   time.Time                            // status.AsOf value of the last processed status update
+	notifyWatchers   map[*mapSetHandle]chan *ipn.Notify
+	lastStatusTime   time.Time // status.AsOf value of the last processed status update
 	// directFileRoot, if non-empty, means to write received files
 	// directly to this directory, without staging them in an
 	// intermediate buffered directory for "pick-up" later. If
@@ -1690,24 +1691,70 @@ func (b *LocalBackend) readPoller() {
 	}
 }
 
-// send delivers n to the connected frontend. If no frontend is
-// connected, the notification is dropped without being delivered.
+// WatchNotifications subscribes to the ipn.Notify message bus notification
+// messages.
+//
+// WatchNotifications blocks until ctx is done.
+//
+// The provided fn will only be called with non-nil pointers. The caller must
+// not modify roNotify. If fn returns false, the watch also stops.
+//
+// Failure to consume many notifications in a row will result in dropped
+// notifications. There is currently (2022-11-22) no mechanism provided to
+// detect when a message has been dropped.
+func (b *LocalBackend) WatchNotifications(ctx context.Context, fn func(roNotify *ipn.Notify) (keepGoing bool)) {
+	handle := new(mapSetHandle)
+	ch := make(chan *ipn.Notify, 128)
+
+	b.mu.Lock()
+	mak.Set(&b.notifyWatchers, handle, ch)
+	b.mu.Unlock()
+	defer func() {
+		b.mu.Lock()
+		delete(b.notifyWatchers, handle)
+		b.mu.Unlock()
+	}()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case n := <-ch:
+			if !fn(n) {
+				return
+			}
+		}
+	}
+}
+
+// send delivers n to the connected frontend and any API watchers from
+// LocalBackend.WatchNotifications (via the LocalAPI).
+//
+// If no frontend is connected or API watchers are backed up, the notification
+// is dropped without being delivered.
 func (b *LocalBackend) send(n ipn.Notify) {
+	n.Version = version.Long
+
 	b.mu.Lock()
 	notifyFunc := b.notify
 	apiSrv := b.peerAPIServer
-	b.mu.Unlock()
-
-	if notifyFunc == nil {
-		return
-	}
-
 	if apiSrv.hasFilesWaiting() {
 		n.FilesWaiting = &empty.Message{}
 	}
 
-	n.Version = version.Long
-	notifyFunc(n)
+	for _, ch := range b.notifyWatchers {
+		select {
+		case ch <- &n:
+		default:
+			// Drop the notification if the channel is full.
+		}
+	}
+
+	b.mu.Unlock()
+
+	if notifyFunc != nil {
+		notifyFunc(n)
+	}
 }
 
 func (b *LocalBackend) sendFileNotify() {

ipn/localapi/localapi.go

@@ -80,6 +80,7 @@ var handler = map[string]localAPIHandler{
 	"serve-config":            (*Handler).serveServeConfig,
 	"set-dns":                 (*Handler).serveSetDNS,
 	"set-expiry-sooner":       (*Handler).serveSetExpirySooner,
+	"start":                   (*Handler).serveStart,
 	"status":                  (*Handler).serveStatus,
 	"tka/init":                (*Handler).serveTKAInit,
 	"tka/log":                 (*Handler).serveTKALog,
@@ -88,6 +89,7 @@ var handler = map[string]localAPIHandler{
 	"tka/status":              (*Handler).serveTKAStatus,
 	"tka/disable":             (*Handler).serveTKADisable,
 	"upload-client-metrics":   (*Handler).serveUploadClientMetrics,
+	"watch-ipn-bus":           (*Handler).serveWatchIPNBus,
 	"whois":                   (*Handler).serveWhoIs,
 }
 
@@ -572,6 +574,34 @@ func (h *Handler) serveStatus(w http.ResponseWriter, r *http.Request) {
 	e.Encode(st)
 }
 
+func (h *Handler) serveWatchIPNBus(w http.ResponseWriter, r *http.Request) {
+	if !h.PermitWrite {
+		http.Error(w, "denied", http.StatusForbidden)
+		return
+	}
+	f, ok := w.(http.Flusher)
+	if !ok {
+		http.Error(w, "not a flusher", http.StatusInternalServerError)
+		return
+	}
+	w.Header().Set("Content-Type", "application/json")
+	f.Flush()
+
+	ctx := r.Context()
+	h.b.WatchNotifications(ctx, func(roNotify *ipn.Notify) (keepGoing bool) {
+		js, err := json.Marshal(roNotify)
+		if err != nil {
+			h.logf("json.Marshal: %v", err)
+			return false
+		}
+		if _, err := fmt.Fprintf(w, "%s\n", js); err != nil {
+			return false
+		}
+		f.Flush()
+		return true
+	})
+}
+
 func (h *Handler) serveLoginInteractive(w http.ResponseWriter, r *http.Request) {
 	if !h.PermitWrite {
 		http.Error(w, "login access denied", http.StatusForbidden)
@@ -586,6 +616,29 @@ func (h *Handler) serveLoginInteractive(w http.ResponseWriter, r *http.Request)
 	return
 }
 
+func (h *Handler) serveStart(w http.ResponseWriter, r *http.Request) {
+	if !h.PermitWrite {
+		http.Error(w, "access denied", http.StatusForbidden)
+		return
+	}
+	if r.Method != "POST" {
+		http.Error(w, "want POST", 400)
+		return
+	}
+	var o ipn.Options
+	if err := json.NewDecoder(r.Body).Decode(&o); err != nil {
+		http.Error(w, err.Error(), http.StatusBadRequest)
+		return
+	}
+	err := h.b.Start(o)
+	if err != nil {
+		// TODO(bradfitz): map error to a good HTTP error
+		http.Error(w, err.Error(), http.StatusInternalServerError)
+		return
+	}
+	w.WriteHeader(http.StatusNoContent)
+}
+
 func (h *Handler) serveLogout(w http.ResponseWriter, r *http.Request) {
 	if !h.PermitWrite {
 		http.Error(w, "logout access denied", http.StatusForbidden)