Could you take a look, " + fakebasecamp.Mention(fakebasecamp.AgentID) + "?
", + } + }) + return s.Emit(fakebasecamp.Event{ + EventType: "comment.created", BucketID: fakebasecamp.ProjectID, RecordingID: id, CreatorID: fakebasecamp.OperatorID, + }, fakebasecamp.Both) +} + +// logReporter reports to the log, and runs the fake's cleanup at exit. +type logReporter struct { + mu sync.Mutex + cleanups []func() +} + +func (*logReporter) Helper() {} + +func (*logReporter) Errorf(format string, args ...any) { log.Printf(format, args...) } + +func (r *logReporter) Cleanup(fn func()) { + r.mu.Lock() + defer r.mu.Unlock() + r.cleanups = append(r.cleanups, fn) +} + +func (r *logReporter) cleanup() { + r.mu.Lock() + defer r.mu.Unlock() + for i := len(r.cleanups) - 1; i >= 0; i-- { + r.cleanups[i]() + } +} diff --git a/go.mod b/go.mod index c82feb328..7f8cc6194 100644 --- a/go.mod +++ b/go.mod @@ -16,6 +16,7 @@ require ( github.com/charmbracelet/huh v1.0.0 github.com/charmbracelet/x/ansi v0.11.8 github.com/charmbracelet/x/term v0.2.2 + github.com/coder/websocket v1.8.15 github.com/fsnotify/fsnotify v1.10.1 github.com/gofrs/flock v0.13.1 github.com/itchyny/gojq v0.12.19 diff --git a/internal/connector/fakebasecamp/account.go b/internal/connector/fakebasecamp/account.go new file mode 100644 index 000000000..496c0a524 --- /dev/null +++ b/internal/connector/fakebasecamp/account.go @@ -0,0 +1,239 @@ +package fakebasecamp + +import ( + "cmp" + "net/http" + "slices" + "strings" + "time" +) + +// The account reads: who a token is, who someone else is, and the projects. + +// epoch is when everything in the fake was created and last updated. +var epoch = time.Date(2026, time.January, 5, 9, 0, 0, 0, time.UTC) + +// personJSON is a person as Basecamp renders one. +type personJSON struct { + ID int64 `json:"id"` + AttachableSGID string `json:"attachable_sgid"` + Name string `json:"name"` + EmailAddress string `json:"email_address,omitempty"` + PersonableType string `json:"personable_type"` + Title string `json:"title"` + CreatedAt time.Time `json:"created_at"` + UpdatedAt time.Time `json:"updated_at"` + Admin bool `json:"admin"` + Owner bool `json:"owner"` + Client bool `json:"client"` + Employee bool `json:"employee"` + TimeZone string `json:"time_zone"` + AvatarURL string `json:"avatar_url"` + // Boss is in the person's own profile only. + Boss *named `json:"boss,omitempty"` + Company *named `json:"company,omitempty"` + CanPing bool `json:"can_ping"` +} + +type named struct { + ID int64 `json:"id"` + Name string `json:"name"` +} + +func (c *call) renderPerson(p *Person) personJSON { + kind := "User" + if p.Agent { + kind = "Agent" + } + out := personJSON{ + ID: p.ID, + AttachableSGID: attachableSGID(p.ID), + Name: p.Name, + PersonableType: kind, + CreatedAt: epoch, + UpdatedAt: epoch, + Client: p.Client, + Employee: !p.Client && !p.Agent, + TimeZone: "Etc/UTC", + AvatarURL: c.base() + "/avatars/" + slug(p.Name), + Company: &named{ID: 1, Name: c.s.world.Account.Name}, + CanPing: !p.Client, + } + if !p.Agent { + out.EmailAddress = slug(p.Name) + "@example.com" + } + return out +} + +func slug(name string) string { + return strings.ToLower(strings.ReplaceAll(name, " ", ".")) +} + +// authorization is Launchpad's authorization.json: the identity behind a +// person's token, and the accounts it reaches. +func (c *call) authorization() answer { + caller, ok := c.s.callerLocked(c.req) + p := c.s.world.People[caller] + if !ok || p == nil || p.IdentityID == 0 { + return answer{status: http.StatusUnauthorized} + } + first, last, _ := strings.Cut(p.Name, " ") + a := c.s.world.Account + return c.jsonAnswer(http.StatusOK, map[string]any{ + "expires_at": epoch.Add(24 * time.Hour * 365), + "identity": map[string]any{ + "id": p.IdentityID, "first_name": first, "last_name": last, + "email_address": slug(p.Name) + "@example.com", + }, + "accounts": []map[string]any{{ + "product": "bc3", "id": a.ID, "name": a.Name, + "href": c.accountURL(""), "app_href": c.accountURL(""), + }}, + }) +} + +// profile is the caller's own profile, the one read that names their boss. +func (c *call) profile() answer { + p, ok := c.s.world.People[c.caller] + if !ok { + return answer{status: http.StatusUnauthorized} + } + out := c.renderPerson(p) + if boss, ok := c.s.world.People[p.BossID]; ok { + out.Boss = &named{ID: boss.ID, Name: boss.Name} + } + return c.jsonAnswer(http.StatusOK, out) +} + +// person is anyone in the account, unless they are readable only by others. +func (c *call) person() answer { + id, ok := c.pathID("id") + if !ok { + return answer{status: http.StatusNotFound} + } + p, ok := c.s.world.People[id] + if !ok { + return answer{status: http.StatusNotFound} + } + if p.ReadableBy != nil && !slices.Contains(p.ReadableBy, c.caller) { + return answer{status: http.StatusForbidden} + } + return c.jsonAnswer(http.StatusOK, c.renderPerson(p)) +} + +// projectJSON is a project as Basecamp renders one, with its dock. +type projectJSON struct { + ID int64 `json:"id"` + Status string `json:"status"` + CreatedAt time.Time `json:"created_at"` + UpdatedAt time.Time `json:"updated_at"` + Name string `json:"name"` + Description string `json:"description"` + Purpose string `json:"purpose"` + ClientsEnabled bool `json:"clients_enabled"` + BookmarkURL string `json:"bookmark_url"` + URL string `json:"url"` + AppURL string `json:"app_url"` + Dock []dockJSON `json:"dock"` + Bookmarked bool `json:"bookmarked"` +} + +type dockJSON struct { + ID int64 `json:"id"` + Title string `json:"title"` + Name string `json:"name"` + Enabled bool `json:"enabled"` + Position *int `json:"position"` + URL string `json:"url"` + AppURL string `json:"app_url"` +} + +func (c *call) renderProject(p *Project) projectJSON { + out := projectJSON{ + ID: p.ID, + Status: "active", + CreatedAt: epoch, + UpdatedAt: epoch, + Name: p.Name, + Purpose: "topic", + BookmarkURL: c.accountURL("my/bookmarks/%d.json", p.ID), + URL: c.accountURL("projects/%d.json", p.ID), + AppURL: c.accountURL("projects/%d", p.ID), + Dock: []dockJSON{}, + } + // A Campfire is docked as "chat": that is where a line's Campfire is + // found first. + for _, r := range c.s.sortedRecordings() { + if r.BucketID == p.ID && r.Type == campfireType { + position := len(out.Dock) + 1 + out.Dock = append(out.Dock, dockJSON{ + ID: r.ID, Title: r.Title, Name: "chat", Enabled: true, Position: &position, + URL: c.accountURL("chats/%d.json", r.ID), AppURL: c.accountURL("chats/%d", r.ID), + }) + } + } + return out +} + +// projects is every project the caller is on. +func (c *call) projects() answer { + out := []projectJSON{} + for _, p := range c.s.sortedProjects() { + if c.s.world.member(p.ID, c.caller) { + out = append(out, c.renderProject(p)) + } + } + return c.jsonAnswer(http.StatusOK, out) +} + +// visibleProject is a project the caller is on; a project they are not on +// is not found, as Basecamp answers it. +func (c *call) visibleProject() (*Project, bool) { + id, ok := c.pathID("id") + if !ok || !c.s.world.member(id, c.caller) { + return nil, false + } + return c.s.world.Projects[id], true +} + +func (c *call) project() answer { + p, ok := c.visibleProject() + if !ok { + return answer{status: http.StatusNotFound} + } + return c.jsonAnswer(http.StatusOK, c.renderProject(p)) +} + +func (c *call) projectPeople() answer { + p, ok := c.visibleProject() + if !ok { + return answer{status: http.StatusNotFound} + } + out := []personJSON{} + for _, id := range p.Members { + if person, ok := c.s.world.People[id]; ok { + out = append(out, c.renderPerson(person)) + } + } + return c.jsonAnswer(http.StatusOK, out) +} + +// sortedProjects is the projects in id order, so every listing is stable. +func (s *Server) sortedProjects() []*Project { + out := make([]*Project, 0, len(s.world.Projects)) + for _, p := range s.world.Projects { + out = append(out, p) + } + slices.SortFunc(out, func(a, b *Project) int { return cmp.Compare(a.ID, b.ID) }) + return out +} + +// sortedRecordings is the recordings in id order. +func (s *Server) sortedRecordings() []*Recording { + out := make([]*Recording, 0, len(s.world.Recordings)) + for _, r := range s.world.Recordings { + out = append(out, r) + } + slices.SortFunc(out, func(a, b *Recording) int { return cmp.Compare(a.ID, b.ID) }) + return out +} diff --git a/internal/connector/fakebasecamp/cable.go b/internal/connector/fakebasecamp/cable.go new file mode 100644 index 000000000..30cbba51b --- /dev/null +++ b/internal/connector/fakebasecamp/cable.go @@ -0,0 +1,398 @@ +package fakebasecamp + +import ( + "context" + "encoding/json" + "fmt" + "maps" + "net/http" + "slices" + "strings" + "time" + + "github.com/coder/websocket" +) + +// The live lane: stream tickets and the Action Cable connection they open, +// as eventfeed speaks it (its cable.go, and the loopback server in its +// scenario_harness_test.go). + +const ( + cableSubprotocol = "actioncable-v1-json" + cableChannel = "EventsChannel" + // cableWriteTimeout bounds one frame's write: a client that stops + // reading for longer is dropped, as a server drops it. + cableWriteTimeout = 10 * time.Second + // streamTicketLifetime is how long a stream ticket opens the cable: + // about 120 seconds, the SDK's StreamTicket says. + streamTicketLifetime = 120 * time.Second +) + +// streamTicket is who a ticket was minted for, and until when it opens the +// cable. A ticket is not spent by the connection it opens: the SDK's +// StreamTicket is "an opaque replayable bearer for its window", minted +// statelessly, so it opens any number of connections until it expires. +type streamTicket struct { + caller int64 + expires time.Time +} + +// streamTicket mints a ticket for the caller, and the cable URL that +// carries it: a ws:// URL on the fake's own loopback port. Minting forgets +// the tickets that have expired. +func (c *call) streamTicket() answer { + now := c.s.opts.now() + maps.DeleteFunc(c.s.tickets, func(_ string, t streamTicket) bool { return !now.Before(t.expires) }) + ticket := fmt.Sprintf("fake-ticket-%d", c.s.nextSerialLocked()) + c.s.tickets[ticket] = streamTicket{caller: c.caller, expires: now.Add(streamTicketLifetime)} + return c.jsonAnswer(http.StatusOK, map[string]any{ + "ticket": ticket, + "expires_in": int(streamTicketLifetime / time.Second), + "url": "ws" + strings.TrimPrefix(c.base(), "http") + "/cable?ticket=" + ticket, + }) +} + +// ticketLocked is who the request's stream ticket was minted for, and +// whether it carried one the fake minted that has not expired. +func (s *Server) ticketLocked(req *http.Request) (int64, bool) { + t, ok := s.tickets[req.URL.Query().Get("ticket")] + if !ok || !s.opts.now().Before(t.expires) { + return 0, false + } + return t.caller, true +} + +// cableConn is one live connection. +type cableConn struct { + s *Server + ws *websocket.Conn + caller int64 + // cancel ends the connection's context when it is detached, which + // stops its reader, its writer and its pings. wake has a frame to + // write. + cancel context.CancelFunc + wake chan struct{} + + // Guarded by the Server's lock. subscriptions are the confirmed ones, + // by identifier. + subscriptions map[string]filters + queue [][]byte + silent bool + detached bool +} + +// statusRecorder logs the status an upgrade answers before it is written, +// and hands the connection over to the WebSocket. +type statusRecorder struct { + http.ResponseWriter + logStatus func(int) +} + +func (r *statusRecorder) WriteHeader(code int) { + r.logStatus(code) + r.ResponseWriter.WriteHeader(code) +} + +func (r *statusRecorder) Unwrap() http.ResponseWriter { return r.ResponseWriter } + +// serveCable upgrades a ticketed request and serves it until either side +// ends it. +func (s *Server) serveCable(w http.ResponseWriter, req *http.Request) { + c, ok := s.begin(RouteCable, w, req) + if !ok { + return + } + s.mu.Lock() + caller, known := s.ticketLocked(req) + s.mu.Unlock() + if !known { + s.finish(c, w, answer{status: http.StatusUnauthorized}) + return + } + rec := &statusRecorder{ResponseWriter: w, logStatus: func(status int) { s.setStatus(c.index, status) }} + ws, err := websocket.Accept(rec, req, &websocket.AcceptOptions{Subprotocols: []string{cableSubprotocol}}) + if err != nil { + return + } + if ws.Subprotocol() != cableSubprotocol { + s.r.Errorf("fakebasecamp: a cable client did not negotiate %s", cableSubprotocol) + _ = ws.CloseNow() + return + } + + // The connection outlives the request that opened it: it keeps the + // request's values, and ends when either side closes it. + ctx, cancel := context.WithCancel(context.WithoutCancel(req.Context())) + cc := &cableConn{ + s: s, ws: ws, caller: caller, cancel: cancel, wake: make(chan struct{}, 1), + subscriptions: map[string]filters{}, + } + s.mu.Lock() + if s.closed { + s.mu.Unlock() + cancel() + _ = ws.CloseNow() + return + } + // httptest does not wait for a hijacked connection, so Close waits for + // these three: this reader, the writer and the pings. + s.cables.Add(3) + s.conns[cc] = struct{}{} + cc.enqueueLocked([]byte(`{"type":"welcome"}`)) + s.notifyLocked() + s.mu.Unlock() + + go cc.write(ctx) + go cc.ping(ctx) + defer s.cables.Done() + cc.read(ctx) +} + +// read handles the client's commands until the connection ends. +func (cc *cableConn) read(ctx context.Context) { + defer cc.drop() + for { + typ, data, err := cc.ws.Read(ctx) + if err != nil { + return + } + if typ != websocket.MessageText { + cc.s.r.Errorf("fakebasecamp: a cable client sent a binary frame") + continue + } + var cmd struct { + Command string `json:"command"` + Identifier string `json:"identifier"` + } + if err := json.Unmarshal(data, &cmd); err != nil { + cc.s.r.Errorf("fakebasecamp: a cable client sent a frame that is not a command: %q", data) + continue + } + switch cmd.Command { + case "subscribe": + cc.subscribe(cmd.Identifier) + case "unsubscribe": + cc.s.mu.Lock() + if _, ok := cc.subscriptions[cmd.Identifier]; ok { + delete(cc.subscriptions, cmd.Identifier) + cc.s.notifyLocked() + } + cc.s.mu.Unlock() + default: + cc.s.r.Errorf("fakebasecamp: a cable client sent the command %q", cmd.Command) + } + } +} + +// subscribe confirms a subscription to the account's events, or rejects +// one the fake does not serve. As in Action Cable, a connection holds any +// number of subscriptions, one per identifier, each confirmed and pushed to +// on its own; subscribing again to an identifier already held answers +// nothing at all (Action Cable raises AlreadySubscribedError on the server, +// and sends the client no frame). +func (cc *cableConn) subscribe(identifier string) { + var params map[string]any + f, err := filters{}, json.Unmarshal([]byte(identifier), ¶ms) + if err == nil { + f, err = subscriptionFilters(params) + } + cc.s.mu.Lock() + defer cc.s.mu.Unlock() + _, held := cc.subscriptions[identifier] + switch { + case err != nil: + cc.s.r.Errorf("fakebasecamp: rejecting the cable subscription %s: %v", identifier, err) + cc.enqueueLocked(cableFrame("reject_subscription", identifier)) + case held: + default: + cc.subscriptions[identifier] = f + cc.enqueueLocked(cableFrame("confirm_subscription", identifier)) + cc.s.notifyLocked() + } +} + +// subscriptionFilters reads an EventsChannel identifier's filters. The +// fake serves the account lane only. +func subscriptionFilters(params map[string]any) (filters, error) { + if params["channel"] != cableChannel { + return filters{}, fmt.Errorf("the channel is not %s", cableChannel) + } + if _, inbox := params["inbox"]; inbox { + return filters{}, fmt.Errorf("the fake serves the account lane, not the inbox") + } + var bad error + f, err := parseFilters(func(key string) string { + v, ok := params[key] + if !ok { + return "" + } + s, isString := v.(string) + if !isString { + bad = fmt.Errorf("%s is not a string", key) + } + return s + }) + if bad != nil { + return filters{}, bad + } + return f, err +} + +func cableFrame(typ, identifier string) []byte { + frame, _ := json.Marshal(map[string]string{"type": typ, "identifier": identifier}) + return frame +} + +// pushLocked queues ev to every subscription that sees it. +func (s *Server) pushLocked(ev Event) { + payload := pushPayload{pollRow: rowOf(ev), ActorType: ev.ActorType, VisibleToClients: ev.VisibleToClients} + for cc := range s.conns { + // In identifier order, so the frames' order does not hang on map + // iteration. + for _, identifier := range slices.Sorted(maps.Keys(cc.subscriptions)) { + if !s.visibleLocked(ev, cc.subscriptions[identifier], cc.caller) { + continue + } + frame, err := json.Marshal(struct { + Identifier string `json:"identifier"` + Message pushPayload `json:"message"` + }{identifier, payload}) + if err != nil { + s.r.Errorf("fakebasecamp: rendering event %d: %v", ev.ID, err) + return + } + cc.enqueueLocked(frame) + } + } +} + +// enqueueLocked queues a frame for the writer. +func (cc *cableConn) enqueueLocked(frame []byte) { + if cc.detached { + return + } + cc.queue = append(cc.queue, frame) + select { + case cc.wake <- struct{}{}: + default: + } +} + +// write sends queued frames, in order, until the connection is detached. +func (cc *cableConn) write(ctx context.Context) { + defer cc.s.cables.Done() + for { + select { + case <-cc.wake: + case <-ctx.Done(): + return + } + cc.s.mu.Lock() + frames := cc.queue + cc.queue = nil + cc.s.mu.Unlock() + for _, frame := range frames { + writeCtx, cancel := context.WithTimeout(ctx, cableWriteTimeout) + err := cc.ws.Write(writeCtx, websocket.MessageText, frame) + cancel() + if err != nil { + cc.drop() + return + } + } + } +} + +// ping queues a ping every interval, unless the connection was silenced. +func (cc *cableConn) ping(ctx context.Context) { + defer cc.s.cables.Done() + ticker := time.NewTicker(cc.s.opts.pingInterval) + defer ticker.Stop() + for { + select { + case now := <-ticker.C: + cc.s.mu.Lock() + if !cc.silent { + cc.enqueueLocked([]byte(fmt.Sprintf(`{"type":"ping","message":%d}`, now.Unix()))) + } + cc.s.mu.Unlock() + case <-ctx.Done(): + return + } + } +} + +// drop detaches the connection and closes its socket. +func (cc *cableConn) drop() { + cc.s.mu.Lock() + cc.detachLocked() + cc.s.notifyLocked() + cc.s.mu.Unlock() + cc.sever() +} + +// detachLocked takes the connection out of the fake: nothing more is queued +// to it, and its writer and pings stop. +func (cc *cableConn) detachLocked() { + if cc.detached { + return + } + cc.detached = true + delete(cc.s.conns, cc) + cc.cancel() +} + +// sever closes the socket without a closing handshake, as a dropped +// connection ends. +func (cc *cableConn) sever() { _ = cc.ws.CloseNow() } + +// detachConnsLocked detaches every connection, and returns them to sever. +func (s *Server) detachConnsLocked() []*cableConn { + conns := make([]*cableConn, 0, len(s.conns)) + for cc := range s.conns { + conns = append(conns, cc) + } + for _, cc := range conns { + cc.detachLocked() + } + return conns +} + +// DropCable severs every live connection, and reports how many there were. +// A client reconnects with a fresh ticket, and catches up from the poll +// lane before it streams again. +func (s *Server) DropCable() int { + s.mu.Lock() + conns := s.detachConnsLocked() + s.notifyLocked() + s.mu.Unlock() + for _, cc := range conns { + cc.sever() + } + return len(conns) +} + +// StopPings silences the live connections open now: they stay up, and stop +// pinging, until the client gives up on them. A connection opened later +// pings as usual. +func (s *Server) StopPings() { + s.mu.Lock() + defer s.mu.Unlock() + for cc := range s.conns { + cc.silent = true + } +} + +// Subscribers is how many live connections hold a confirmed subscription +// now. +func (s *Server) Subscribers() int { + s.mu.Lock() + defer s.mu.Unlock() + n := 0 + for cc := range s.conns { + if len(cc.subscriptions) > 0 { + n++ + } + } + return n +} diff --git a/internal/connector/fakebasecamp/cable_test.go b/internal/connector/fakebasecamp/cable_test.go new file mode 100644 index 000000000..711e1b336 --- /dev/null +++ b/internal/connector/fakebasecamp/cable_test.go @@ -0,0 +1,280 @@ +package fakebasecamp_test + +import ( + "context" + "encoding/json" + "errors" + "net/http" + "sync" + "testing" + "time" + + "github.com/coder/websocket" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/basecamp/basecamp-cli/internal/connector/fakebasecamp" +) + +// cableClient speaks Action Cable to the fake by hand, to hold it to the +// frames themselves. +type cableClient struct { + t *testing.T + ws *websocket.Conn +} + +func dialCable(t *testing.T, s *fakebasecamp.Server, bearer string) *cableClient { + t.Helper() + status, _, body := do(t, s, http.MethodPost, "/999/events/stream_ticket.json", bearer, "") + require.Equal(t, http.StatusOK, status, body) + var ticket struct { + Ticket string `json:"ticket"` + ExpiresIn int `json:"expires_in"` + URL string `json:"url"` + } + require.NoError(t, json.Unmarshal([]byte(body), &ticket)) + assert.Positive(t, ticket.ExpiresIn) + assert.Equal(t, "ws"+s.URL()[len("http"):]+"/cable?ticket="+ticket.Ticket, ticket.URL) + + ws, resp, err := websocket.Dial(bounded(t), ticket.URL, &websocket.DialOptions{Subprotocols: []string{"actioncable-v1-json"}}) + require.NoError(t, err) + if resp.Body != nil { + _ = resp.Body.Close() + } + assert.Equal(t, "actioncable-v1-json", ws.Subprotocol()) + t.Cleanup(func() { _ = ws.CloseNow() }) + return &cableClient{t: t, ws: ws} +} + +// frame reads the next frame, within d. +func (c *cableClient) frame(d time.Duration) (map[string]any, error) { + ctx, cancel := context.WithTimeout(c.t.Context(), d) + defer cancel() + _, data, err := c.ws.Read(ctx) + if err != nil { + return nil, err + } + var frame map[string]any + require.NoError(c.t, json.Unmarshal(data, &frame), "%s", data) + return frame, nil +} + +// next is the next frame that is not a ping. +func (c *cableClient) next() map[string]any { + c.t.Helper() + for { + frame, err := c.frame(waitFor) + require.NoError(c.t, err) + if frame["type"] != "ping" { + return frame + } + } +} + +func (c *cableClient) send(v any) { + c.t.Helper() + data, err := json.Marshal(v) + require.NoError(c.t, err) + require.NoError(c.t, c.ws.Write(bounded(c.t), websocket.MessageText, data)) +} + +// The cable welcomes, confirms the subscription it is sent, and pushes the +// events the subscription's filters and the caller's projects let through, +// in the push shape: the poll row plus actor_type and visible_to_clients. +func TestCableSpeaksActionCable(t *testing.T) { + s, bearer := start(t) + c := dialCable(t, s, bearer) + assert.Equal(t, map[string]any{"type": "welcome"}, c.next()) + + identifier := `{"channel":"EventsChannel","buckets":"48699913","exclude_performers":"52007412"}` + c.send(map[string]string{"command": "subscribe", "identifier": identifier}) + assert.Equal(t, map[string]any{"type": "confirm_subscription", "identifier": identifier}, c.next()) + assert.Equal(t, 1, s.Subscribers()) + + s.Update(func(w *fakebasecamp.World) { + w.Projects[777] = &fakebasecamp.Project{ID: 777, Name: "Elsewhere", Members: []int64{fakebasecamp.OperatorID}} + }) + s.Emit(commentBy(fakebasecamp.OperatorID, otherProject), fakebasecamp.Live) // another bucket + s.Emit(commentBy(fakebasecamp.AgentID, fakebasecamp.ProjectID), fakebasecamp.Live) // excluded performer + s.Emit(commentBy(fakebasecamp.OperatorID, 777), fakebasecamp.Live) // a project the agent is not on + s.Emit(commentBy(fakebasecamp.OperatorID, fakebasecamp.ProjectID), fakebasecamp.Poll) // the poll lane only + want := s.Emit(fakebasecamp.Event{ + EventType: "todo.assignment_changed", BucketID: fakebasecamp.ProjectID, RecordingID: 9002, + CreatorID: fakebasecamp.OperatorID, Details: json.RawMessage(`{"column_id":1}`), + }, fakebasecamp.Live) + + frame := c.next() + assert.Equal(t, identifier, frame["identifier"]) + message, ok := frame["message"].(map[string]any) + require.True(t, ok, "%v", frame) + assert.Equal(t, map[string]any{ + "id": float64(want.ID), + "kind": "todo_assignment_changed", + "action": "assignment_changed", + "event_type": "todo.assignment_changed", + "bucket_id": float64(fakebasecamp.ProjectID), + "creator_id": float64(fakebasecamp.OperatorID), + "performed_by_id": nil, + "recording_id": float64(9002), + "created_at": want.CreatedAt.Format(time.RFC3339Nano), + "details": map[string]any{"column_id": float64(1)}, + "actor_type": "person", + "visible_to_clients": false, + }, message) + + // As in Action Cable, the same subscription again answers nothing, and + // a different one is a second subscription, confirmed and pushed to on + // its own. + c.send(map[string]string{"command": "subscribe", "identifier": identifier}) + everything := `{"channel":"EventsChannel"}` + c.send(map[string]string{"command": "subscribe", "identifier": everything}) + assert.Equal(t, map[string]any{"type": "confirm_subscription", "identifier": everything}, c.next(), + "the repeated subscribe sent no frame ahead of this one") + assert.Equal(t, 1, s.Subscribers(), "one connection, however many subscriptions") + both := s.Emit(commentBy(fakebasecamp.OperatorID, fakebasecamp.ProjectID), fakebasecamp.Live) + pushed := map[string]float64{} + for range 2 { + frame := c.next() + pushed[frame["identifier"].(string)] = frame["message"].(map[string]any)["id"].(float64) + } + assert.Equal(t, map[string]float64{identifier: float64(both.ID), everything: float64(both.ID)}, pushed) + + c.send(map[string]string{"command": "unsubscribe", "identifier": identifier}) + c.send(map[string]string{"command": "unsubscribe", "identifier": everything}) + await(t, s, "the unsubscribes", func() bool { return s.Subscribers() == 0 }) +} + +// The cable pings on its interval until it is silenced; then nothing more +// arrives but what is published. +func TestCablePingsUntilSilenced(t *testing.T) { + s, bearer := start(t, fakebasecamp.WithPingInterval(20*time.Millisecond)) + c := dialCable(t, s, bearer) + assert.Equal(t, "welcome", c.next()["type"]) + frame, err := c.frame(waitFor) + require.NoError(t, err) + assert.Equal(t, "ping", frame["type"]) + assert.IsType(t, float64(0), frame["message"], "the ping carries the server's epoch") + + identifier := `{"channel":"EventsChannel"}` + c.send(map[string]string{"command": "subscribe", "identifier": identifier}) + assert.Equal(t, "confirm_subscription", c.next()["type"]) + s.StopPings() + // Every frame after this one was queued after the silence. + marker := s.Emit(commentBy(fakebasecamp.OperatorID, fakebasecamp.ProjectID), fakebasecamp.Live) + for { + frame, err := c.frame(waitFor) + require.NoError(t, err) + if frame["type"] == "ping" { + continue + } + require.Equal(t, float64(marker.ID), frame["message"].(map[string]any)["id"]) + break + } + assert.Equal(t, 1, s.Subscribers(), "silenced, not dropped") + // Ten intervals of nothing. The read gives up by closing the socket, so + // it comes last. + _, err = c.frame(10 * 20 * time.Millisecond) + assert.True(t, errors.Is(err, context.DeadlineExceeded), "a silenced connection pinged: %v", err) +} + +// A cable URL whose ticket the fake never minted is refused before the +// upgrade, and a dropped connection is gone at once. +func TestCableRefusesAnUnknownTicketAndDrops(t *testing.T) { + s, bearer := start(t) + _, resp, err := websocket.Dial(bounded(t), "ws"+s.URL()[len("http"):]+"/cable?ticket=forged", + &websocket.DialOptions{Subprotocols: []string{"actioncable-v1-json"}}) + require.Error(t, err) + require.NotNil(t, resp) + assert.Equal(t, http.StatusUnauthorized, resp.StatusCode) + + c := dialCable(t, s, bearer) + c.next() + c.send(map[string]string{"command": "subscribe", "identifier": `{"channel":"EventsChannel"}`}) + c.next() + assert.Equal(t, 1, s.DropCable()) + assert.Zero(t, s.Subscribers()) + _, err = c.frame(waitFor) + assert.Error(t, err) + assert.Equal(t, 2, s.Count(fakebasecamp.RouteCable)) +} + +// A stream ticket opens the cable as the person it was minted for, and +// again, until it expires by the fake's clock; then it is refused before +// the upgrade. +func TestStreamTicketsOpenTheCableUntilTheyExpire(t *testing.T) { + var mu sync.Mutex + now := time.Now() + clock := func() time.Time { + mu.Lock() + defer mu.Unlock() + return now + } + s, bearer := start(t, fakebasecamp.WithClock(clock)) + status, _, body := do(t, s, http.MethodPost, "/999/events/stream_ticket.json", bearer, "") + require.Equal(t, http.StatusOK, status, body) + var ticket struct { + ExpiresIn int `json:"expires_in"` + URL string `json:"url"` + } + require.NoError(t, json.Unmarshal([]byte(body), &ticket)) + assert.Equal(t, 120, ticket.ExpiresIn) + + dial := func() (int, error) { + t.Helper() + ws, resp, err := websocket.Dial(bounded(t), ticket.URL, &websocket.DialOptions{Subprotocols: []string{"actioncable-v1-json"}}) + if ws != nil { + _ = ws.CloseNow() + } + require.NotNil(t, resp) + return resp.StatusCode, err + } + for range 2 { + status, err := dial() + require.NoError(t, err, "a ticket is replayable within its window") + assert.Equal(t, http.StatusSwitchingProtocols, status) + } + mu.Lock() + now = now.Add(time.Duration(ticket.ExpiresIn) * time.Second) + mu.Unlock() + status, err := dial() + require.Error(t, err) + assert.Equal(t, http.StatusUnauthorized, status) + + var cable []fakebasecamp.Request + for _, r := range s.Requests() { + if r.Route == fakebasecamp.RouteCable { + cable = append(cable, r) + } + } + require.Len(t, cable, 3) + for _, r := range cable[:2] { + assert.Equal(t, fakebasecamp.AgentID, r.Caller, "the caller is the ticket's") + assert.True(t, r.Agent) + assert.Equal(t, http.StatusSwitchingProtocols, r.Status) + } + assert.Zero(t, cable[2].Caller, "an expired ticket names nobody") + assert.Equal(t, http.StatusUnauthorized, cable[2].Status) + ev := s.Emit(commentBy(fakebasecamp.OperatorID, fakebasecamp.ProjectID), fakebasecamp.Poll) + assert.True(t, clock().Equal(ev.CreatedAt), "an event is stamped by the same clock") +} + +// A fault narrowed to Agents reaches the cable, whose caller comes from the +// ticket rather than a bearer token. +func TestCableFaultsNarrowByTicketCaller(t *testing.T) { + s, bearer := start(t) + refused := fakebasecamp.Refused(http.StatusForbidden, fakebasecamp.RouteCable) + refused.When = fakebasecamp.ByAgents + s.Inject(refused) + status, _, body := do(t, s, http.MethodPost, "/999/events/stream_ticket.json", bearer, "") + require.Equal(t, http.StatusOK, status, body) + var ticket struct { + URL string `json:"url"` + } + require.NoError(t, json.Unmarshal([]byte(body), &ticket)) + _, resp, err := websocket.Dial(bounded(t), ticket.URL, &websocket.DialOptions{Subprotocols: []string{"actioncable-v1-json"}}) + require.Error(t, err) + require.NotNil(t, resp) + assert.Equal(t, http.StatusForbidden, resp.StatusCode) + + dialCable(t, s, fakebasecamp.OperatorToken) +} diff --git a/internal/connector/fakebasecamp/connect_test.go b/internal/connector/fakebasecamp/connect_test.go new file mode 100644 index 000000000..a3653dfa4 --- /dev/null +++ b/internal/connector/fakebasecamp/connect_test.go @@ -0,0 +1,100 @@ +package fakebasecamp_test + +import ( + "net/http" + "net/url" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/basecamp/basecamp-cli/internal/auth" + "github.com/basecamp/basecamp-cli/internal/config" + "github.com/basecamp/basecamp-cli/internal/connector/fakebasecamp" +) + +// The connection ceremony, run by the CLI's own auth package against the +// fake: the intake, the poll, the handover of a freshly rotated secret, and +// the client_credentials mint that proves it. +func TestAgentConnectionCeremony(t *testing.T) { + s := fakebasecamp.Start(t, fakebasecamp.DefaultWorld()) + t.Setenv("BASECAMP_NO_KEYRING", "1") + t.Setenv("BASECAMP_TOKEN", "") + t.Setenv("BASECAMP_OAUTH_ISSUER", s.URL()) + t.Setenv("XDG_CONFIG_HOME", t.TempDir()) + + connect := func() (*auth.Manager, *auth.AgentConnectResult) { + t.Helper() + mgr := auth.NewManager(&config.Config{BaseURL: s.URL(), ActiveProfile: "agent", Sources: map[string]string{}}, nil) + mgr.SetStore(auth.NewStore(t.TempDir())) + result, err := mgr.ConnectAgent(bounded(t), auth.AgentConnectOptions{DeviceName: "build-box", NoBrowser: true}) + require.NoError(t, err) + return mgr, result + } + mgr, result := connect() + assert.Equal(t, fakebasecamp.AgentClientID, result.ClientID) + assert.Equal(t, "999", result.AccountID) + assert.Equal(t, fakebasecamp.ScopeFull, result.Scope) + assert.Equal(t, 1, s.Count(fakebasecamp.RouteAgentConnections)) + assert.Equal(t, 1, s.Count(fakebasecamp.RouteToken), "one mint proves the handover") + + var secret string + s.View(func(w *fakebasecamp.World) { secret = w.Agents[fakebasecamp.AgentClientID].Secret }) + assert.NotEqual(t, fakebasecamp.AgentSecret, secret, "the handover rotates the secret") + creds, err := mgr.GetStore().Load("profile:agent") + require.NoError(t, err) + assert.Equal(t, secret, creds.ClientSecret) + + // The stored token is the agent's. + token, err := mgr.AccessToken(bounded(t)) + require.NoError(t, err) + me, err := accountClient(s, token).People().Me(bounded(t)) + require.NoError(t, err) + assert.Equal(t, fakebasecamp.AgentID, me.ID) + assert.Equal(t, "Agent", me.PersonableType) + + // Connecting again elsewhere rotates the secret once more, and the one + // this computer holds stops minting, as Basecamp refuses it. + connect() + mint := func(secret string) (int, string) { + form := url.Values{"grant_type": {"client_credentials"}, "client_id": {fakebasecamp.AgentClientID}, "client_secret": {secret}} + status, _, body := do(t, s, http.MethodPost, "/oauth/tokens", "", form.Encode()) + return status, body + } + status, body := mint(secret) + assert.Equal(t, http.StatusUnauthorized, status) + assert.JSONEq(t, `{"error":"invalid_client"}`, body) + + // A spent code is not handed over twice. + poll := s.Requests() + var code string + for _, r := range poll { + if r.Route == fakebasecamp.RouteAgentConnectionTokens { + code = r.Form.Get("device_code") + } + } + require.NotEmpty(t, code) + status, _, body = do(t, s, http.MethodPost, "/oauth/agent_connection_tokens", "", url.Values{"device_code": {code}}.Encode()) + assert.Equal(t, http.StatusBadRequest, status) + assert.JSONEq(t, `{"error":"invalid_grant"}`, body) +} + +// The approval is the world's to say: a read-only approval provisions the +// client read-only, and its tokens say so. +func TestAgentConnectionHandsOverTheApprovedScope(t *testing.T) { + s := fakebasecamp.Start(t, fakebasecamp.DefaultWorld()) + s.Update(func(w *fakebasecamp.World) { w.Connection.Scope = fakebasecamp.ScopeRead }) + t.Setenv("BASECAMP_NO_KEYRING", "1") + t.Setenv("BASECAMP_TOKEN", "") + t.Setenv("BASECAMP_OAUTH_ISSUER", s.URL()) + t.Setenv("XDG_CONFIG_HOME", t.TempDir()) + + mgr := auth.NewManager(&config.Config{BaseURL: s.URL(), ActiveProfile: "agent", Sources: map[string]string{}}, nil) + mgr.SetStore(auth.NewStore(t.TempDir())) + result, err := mgr.ConnectAgent(bounded(t), auth.AgentConnectOptions{DeviceName: "build-box", NoBrowser: true}) + require.NoError(t, err) + assert.Equal(t, fakebasecamp.ScopeRead, result.Scope) + s.View(func(w *fakebasecamp.World) { + assert.Equal(t, fakebasecamp.ScopeRead, w.Agents[fakebasecamp.AgentClientID].Scope) + }) +} diff --git a/internal/connector/fakebasecamp/doc.go b/internal/connector/fakebasecamp/doc.go new file mode 100644 index 000000000..2a52b3c97 --- /dev/null +++ b/internal/connector/fakebasecamp/doc.go @@ -0,0 +1,57 @@ +// Package fakebasecamp is a Basecamp on a loopback port, for testing +// `basecamp connect` end to end: the agent connection ceremony and the +// client_credentials tokens it leads to, the account reads setup and +// admission make, the event feed's two lanes, and the faults a connector has +// to live through. +// +// It is a test double, but not a test file: the connector's own tests, the +// command tests and a real `basecamp connect` subprocess all run against the +// same one, and `go run ./e2e/fakebasecamp` serves it by hand. It takes a +// Reporter rather than a *testing.T, so nothing here imports a test library. +// +// # One world, one lock +// +// Everything the fake knows is a World: the account, its people and +// projects, the recordings in them, the agents' OAuth clients, and the bearer +// tokens it has issued. Every request is answered from it under one mutex, +// and a test changes it through Update, under the same mutex. Nothing is +// answered from a copy, so a change is seen by the next request, whichever +// lane it arrives on. +// +// Requests are authorized as Basecamp authorizes them: a bearer token names a +// person, and a person reads what their projects hold. A project the caller +// is not a member of, and anything in it, answers 404, on the reads and on +// both lanes of the feed. +// +// # The two lanes are published explicitly +// +// Basecamp's poll lane runs about thirty seconds behind its live lane. The +// fake has no lag, and nothing in it runs on its clock: Emit puts an event +// on the lanes it is told to (Live, Poll, or Both), and Publish adds an +// emitted event to another lane later. The clock (WithClock) only stamps +// an event's created_at and ages stream tickets. A test therefore says exactly what each lane serves, which is what +// makes a catch-up, a gap or a duplicate a deterministic scenario rather +// than a race. +// +// A live event is queued to every matching cable subscription under the +// world's lock, so it is ordered against everything else that lock orders: a +// subscription confirmed before Emit receives the event, and one confirmed +// after it does not. The socket writes happen outside the lock, one writer +// per connection, so a client that stops reading stalls only itself. +// +// The poll lane follows the contract eventfeed checks (events always +// present and in ascending id order, an opaque position, and a `next` URL +// that repeats the request's own filters exactly), and the cable follows +// Action Cable as eventfeed speaks it: the actioncable-v1-json subprotocol, +// a welcome, a confirmed subscription, and a ping every three seconds. +// +// # Faults +// +// Inject answers a route with something else (a 429, a 5xx, an +// invalid_client, a different body) for a number of requests or until it is +// removed. Gate holds a route's requests open until it is released. Before +// runs a function as a route is reached, for a change that has to land in +// the middle of a command. DropCable severs the live connections and +// StopPings silences them. Every request is logged, and Await waits, with a +// deadline and never a sleep, for whatever a test needs to have happened. +package fakebasecamp diff --git a/internal/connector/fakebasecamp/faults_test.go b/internal/connector/fakebasecamp/faults_test.go new file mode 100644 index 000000000..efb53cba3 --- /dev/null +++ b/internal/connector/fakebasecamp/faults_test.go @@ -0,0 +1,274 @@ +package fakebasecamp_test + +import ( + "encoding/json" + "net/http" + "net/url" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/basecamp/basecamp-cli/internal/connector/fakebasecamp" +) + +func mintForm(secret string) string { + return url.Values{ + "grant_type": {"client_credentials"}, "client_id": {fakebasecamp.AgentClientID}, "client_secret": {secret}, + }.Encode() +} + +// The token endpoint's faults: a rate limit asking for a second by +// default, an outage, and a refused client; each lapses when removed. +func TestTokenFaults(t *testing.T) { + s, _ := start(t) + + limited := s.Inject(fakebasecamp.TokenRateLimited(0)) + status, header, _ := do(t, s, http.MethodPost, "/oauth/tokens", "", mintForm(fakebasecamp.AgentSecret)) + assert.Equal(t, http.StatusTooManyRequests, status) + assert.Equal(t, "1", header.Get("Retry-After")) + limited.Remove() + + down := s.Inject(fakebasecamp.TokenUnavailable()) + status, _, _ = do(t, s, http.MethodPost, "/oauth/tokens", "", mintForm(fakebasecamp.AgentSecret)) + assert.Equal(t, http.StatusServiceUnavailable, status) + refused := s.Inject(fakebasecamp.TokenInvalidClient()) + status, _, body := do(t, s, http.MethodPost, "/oauth/tokens", "", mintForm(fakebasecamp.AgentSecret)) + assert.Equal(t, http.StatusUnauthorized, status, "the fault injected last answers") + assert.JSONEq(t, `{"error":"invalid_client"}`, body) + refused.Remove() + down.Remove() + + status, _, body = do(t, s, http.MethodPost, "/oauth/tokens", "", mintForm(fakebasecamp.AgentSecret)) + require.Equal(t, http.StatusOK, status, body) + var token struct { + AccessToken string `json:"access_token"` + Scope string `json:"scope"` + } + require.NoError(t, json.Unmarshal([]byte(body), &token)) + assert.Equal(t, fakebasecamp.ScopeFull, token.Scope) + status, _, _ = do(t, s, http.MethodGet, "/999/my/profile.json", token.AccessToken, "") + assert.Equal(t, http.StatusOK, status, "the minted token reads as the agent") + assert.Equal(t, 4, s.Count(fakebasecamp.RouteToken), "a faulted mint is still a mint") +} + +// A mint carries the scope it asks for, within what the client was +// approved for: read from a full client, never full from a read one. +func TestTokenMintHonorsTheAskedScope(t *testing.T) { + s, _ := start(t) + mint := func(scope string) (int, string) { + t.Helper() + form := url.Values{ + "grant_type": {"client_credentials"}, "client_id": {fakebasecamp.AgentClientID}, + "client_secret": {fakebasecamp.AgentSecret}, "scope": {scope}, + } + status, _, body := do(t, s, http.MethodPost, "/oauth/tokens", "", form.Encode()) + return status, body + } + scopeOf := func(body string) string { + t.Helper() + var token struct { + Scope string `json:"scope"` + } + require.NoError(t, json.Unmarshal([]byte(body), &token), body) + return token.Scope + } + + status, body := mint(fakebasecamp.ScopeRead) + require.Equal(t, http.StatusOK, status, body) + assert.Equal(t, fakebasecamp.ScopeRead, scopeOf(body), "narrowed to what was asked") + status, body = mint("admin") + assert.Equal(t, http.StatusBadRequest, status) + assert.JSONEq(t, `{"error":"invalid_scope"}`, body) + + s.Update(func(w *fakebasecamp.World) { w.Agents[fakebasecamp.AgentClientID].Scope = fakebasecamp.ScopeRead }) + status, body = mint(fakebasecamp.ScopeFull) + assert.Equal(t, http.StatusBadRequest, status, "a read client is not widened") + assert.JSONEq(t, `{"error":"invalid_scope"}`, body) + status, body = mint("") + require.Equal(t, http.StatusOK, status, body) + assert.Equal(t, fakebasecamp.ScopeRead, scopeOf(body), "no scope asked is the client's own") +} + +// A fault can be limited to a number of requests, and to the requests a +// predicate picks, such as an Agent's. +func TestFaultsNarrowAndLapse(t *testing.T) { + s, bearer := start(t) + s.Update(func(w *fakebasecamp.World) { + w.Recordings[101] = &fakebasecamp.Recording{ID: 101, Type: "Comment", BucketID: fakebasecamp.ProjectID, CreatorID: fakebasecamp.OperatorID} + }) + + failing := s.Inject(fakebasecamp.ServerError(2, fakebasecamp.RecordingReads...)) + for range 2 { + status, _, _ := do(t, s, http.MethodGet, "/999/comments/101", bearer, "") + assert.Equal(t, http.StatusInternalServerError, status) + } + status, _, _ := do(t, s, http.MethodGet, "/999/comments/101", bearer, "") + assert.Equal(t, http.StatusOK, status, "the fault lapsed after two") + assert.Equal(t, 2, failing.Hits()) + + agentsOnly := fakebasecamp.Refused(http.StatusForbidden, fakebasecamp.RouteProject, fakebasecamp.RouteProjectPeople) + agentsOnly.When = fakebasecamp.ByAgents + s.Inject(agentsOnly) + status, _, _ = do(t, s, http.MethodGet, "/999/projects/48699913", bearer, "") + assert.Equal(t, http.StatusForbidden, status) + status, _, _ = do(t, s, http.MethodGet, "/999/projects/48699913", fakebasecamp.OperatorToken, "") + assert.Equal(t, http.StatusOK, status) + + // A fault's answer can be a success of its own. + s.Inject(fakebasecamp.Fault{Routes: []fakebasecamp.Route{fakebasecamp.RouteProfile}, Status: http.StatusOK, Body: `{"id":0}`}) + _, _, body := do(t, s, http.MethodGet, "/999/my/profile.json", bearer, "") + assert.JSONEq(t, `{"id":0}`, body) +} + +// A gate holds a route's requests unanswered until it is released, and a +// hook runs before a request is answered, in time to change what it is +// answered with. +func TestGatesAndHooks(t *testing.T) { + s, bearer := start(t) + gate := s.Gate(fakebasecamp.RouteProjects) + answered := make(chan int, 1) + go func() { + status, _, _ := do(t, s, http.MethodGet, "/999/projects.json", bearer, "") + answered <- status + }() + await(t, s, "the held request", func() bool { return gate.Waiting() == 1 }) + requests := s.Requests() + require.Len(t, requests, 1) + assert.Zero(t, requests[0].Status, "held, not answered") + assert.Equal(t, fakebasecamp.AgentID, requests[0].Caller) + assert.True(t, requests[0].Agent) + gate.Release() + assert.Equal(t, http.StatusOK, <-answered) + assert.Equal(t, http.StatusOK, s.Requests()[0].Status, "logged before the client had it") + + remove := s.Before(fakebasecamp.RouteStreamTicket, func(fakebasecamp.Request) { + s.Update(func(w *fakebasecamp.World) { delete(w.Tokens, bearer) }) + }) + status, _, _ := do(t, s, http.MethodPost, "/999/events/stream_ticket.json", bearer, "") + assert.Equal(t, http.StatusUnauthorized, status, "the token the hook removed is refused") + remove() +} + +// Overlapping gates hold a request until every one of them is released, in +// either order. +func TestOverlappingGates(t *testing.T) { + s, bearer := start(t) + first := s.Gate(fakebasecamp.RouteProjects) + second := s.Gate(fakebasecamp.RouteProjects, fakebasecamp.RouteProfile) + answered := make(chan int, 1) + go func() { + status, _, _ := do(t, s, http.MethodGet, "/999/projects.json", bearer, "") + answered <- status + }() + await(t, s, "the request at the first gate", func() bool { return first.Waiting() == 1 }) + second.Release() + first.Release() + await(t, s, "the answer", func() bool { return s.Requests()[0].Status != 0 }) + assert.Equal(t, http.StatusOK, <-answered) + + first = s.Gate(fakebasecamp.RouteProjects) + second = s.Gate(fakebasecamp.RouteProjects) + go func() { + status, _, _ := do(t, s, http.MethodGet, "/999/projects.json", bearer, "") + answered <- status + }() + await(t, s, "the request at the first gate", func() bool { return first.Waiting() == 1 }) + first.Release() + await(t, s, "the request at the second gate", func() bool { return second.Waiting() == 1 }) + assert.Zero(t, s.Requests()[1].Status, "still held by the second gate") + second.Release() + assert.Equal(t, http.StatusOK, <-answered) +} + +// Await gives up once the fake is closed, rather than waiting out its +// context for a change that can no longer come. +func TestAwaitEndsWhenClosed(t *testing.T) { + s, _ := start(t) + done := make(chan error, 1) + go func() { done <- s.Await(t.Context(), func() bool { return false }) }() + s.Close() + assert.ErrorIs(t, <-done, fakebasecamp.ErrClosed) + assert.ErrorIs(t, s.Await(t.Context(), func() bool { return false }), fakebasecamp.ErrClosed) + assert.NoError(t, s.Await(t.Context(), func() bool { return true }), "a condition that holds still returns") +} + +// Requests hands out copies: changing one changes nothing the fake logged. +func TestRequestsAreCopies(t *testing.T) { + s, bearer := start(t) + do(t, s, http.MethodGet, "/999/events.json?since=now", bearer, "") + do(t, s, http.MethodPost, "/oauth/tokens", "", mintForm(fakebasecamp.AgentSecret)) + requests := s.Requests() + requests[0].Query.Set("since", "0") + requests[1].Form.Set("client_secret", "changed") + assert.Equal(t, "now", s.Requests()[0].Query.Get("since")) + assert.Equal(t, fakebasecamp.AgentSecret, s.Requests()[1].Form.Get("client_secret")) +} + +// The poll lane's own contract: the present is an empty page at the head, +// a malformed position is the 400 eventfeed recovers from, and a page that +// stops short says where to continue with the request's filters intact. +func TestPollLaneContract(t *testing.T) { + s, bearer := start(t, fakebasecamp.WithPageSize(1)) + first := s.Emit(commentBy(fakebasecamp.OperatorID, fakebasecamp.ProjectID), fakebasecamp.Both) + s.Emit(commentBy(fakebasecamp.OperatorID, fakebasecamp.ProjectID), fakebasecamp.Live) + last := s.Emit(commentBy(fakebasecamp.OperatorID, fakebasecamp.ProjectID), fakebasecamp.Poll) + + _, _, body := do(t, s, http.MethodGet, "/999/events.json?since=now", bearer, "") + assert.JSONEq(t, `{"events":[],"position":"fake-3"}`, body) + + status, _, body := do(t, s, http.MethodGet, "/999/events.json?position=garbage", bearer, "") + assert.Equal(t, http.StatusBadRequest, status) + assert.JSONEq(t, `{"error":"Unrecognized position","reason":"invalid_position"}`, body) + + status, _, _ = do(t, s, http.MethodGet, "/999/events.json?since=0&buckets=x", bearer, "") + assert.Equal(t, http.StatusBadRequest, status) + + type pollPage struct { + Events []struct { + ID int64 `json:"id"` + } `json:"events"` + Position string `json:"position"` + Next string `json:"next"` + } + poll := func(query string) (pollPage, string) { + t.Helper() + _, _, body := do(t, s, http.MethodGet, "/999/events.json?"+query, bearer, "") + var page pollPage + require.NoError(t, json.Unmarshal([]byte(body), &page), body) + return page, body + } + page, body := poll("since=0&exclude_performers=self") + require.Len(t, page.Events, 1) + assert.Equal(t, first.ID, page.Events[0].ID) + assert.Contains(t, body, `"performed_by_id":null`) + assert.Equal(t, s.URL()+"/999/events.json?exclude_performers=self&position=fake-1", page.Next) + + page, _ = poll("exclude_performers=self&position=fake-1") + require.Len(t, page.Events, 1, "the live-only event is not on the poll lane") + assert.Equal(t, last.ID, page.Events[0].ID) + assert.Empty(t, page.Next) + + // Published to the poll lane later, an event is served to a walk that + // has not passed it. + s.Publish(2, fakebasecamp.Poll) + page, _ = poll("position=fake-1") + require.Len(t, page.Events, 1) + assert.Equal(t, int64(2), page.Events[0].ID) +} + +// A request no route serves is answered 404 and reported, so a client that +// starts calling something the fake does not know fails where it is seen. +func TestUnroutedRequestsAreReported(t *testing.T) { + r := &reporter{T: t} + s := fakebasecamp.Start(r, fakebasecamp.DefaultWorld()) + status, _, _ := do(t, s, http.MethodGet, "/999/vaults/1", fakebasecamp.OperatorToken, "") + assert.Equal(t, http.StatusNotFound, status) + status, _, _ = do(t, s, http.MethodDelete, "/999/projects.json", fakebasecamp.OperatorToken, "") + assert.Equal(t, http.StatusNotFound, status) + assert.Equal(t, []string{ + "fakebasecamp: no route serves GET /999/vaults/1", + "fakebasecamp: no route serves DELETE /999/projects.json", + }, r.reported()) + assert.Equal(t, []fakebasecamp.Route{"", ""}, []fakebasecamp.Route{s.Requests()[0].Route, s.Requests()[1].Route}) +} diff --git a/internal/connector/fakebasecamp/feed.go b/internal/connector/fakebasecamp/feed.go new file mode 100644 index 000000000..2fe884356 --- /dev/null +++ b/internal/connector/fakebasecamp/feed.go @@ -0,0 +1,360 @@ +package fakebasecamp + +import ( + "cmp" + "encoding/json" + "fmt" + "net/http" + "net/url" + "slices" + "strconv" + "strings" + "time" +) + +// The account event feed's poll lane, and the events both lanes carry. + +// Lane is where an event is published: the live cable, the poll walk, or +// both. +type Lane int + +// The lanes. +const ( + Live Lane = 1 << iota + Poll + Both = Live | Poll +) + +// Event is one account feed event. Emit fills in what is left zero. +type Event struct { + // ID is the feed-global event id. Zero takes the next one. + ID int64 + // EventType is the cataloged type, as "comment.created". Required. + EventType string + // Kind and Action default from EventType: comment_created, created. + Kind string + Action string + + BucketID int64 + RecordingID int64 + CreatorID int64 + // PerformedByID is the agent that carried out a delegated action; zero + // for an action performed directly. + PerformedByID int64 + // ActorType is "person" or "agent"; it defaults to what the effective + // performer is. + ActorType string + VisibleToClients bool + // CreatedAt defaults to the moment it is emitted, by the fake's clock + // (WithClock). It is a real time and not the fake's fixed epoch: the + // connector hands off only a request created at most + // connector.HandoffGrace before its run started, judging by this + // created_at as the feed served it, so an event stamped in the past is + // discarded as before_this_run. + CreatedAt time.Time + // Details is the feed's own detail object, for the types that publish + // one (boost.created, card.moved). + Details json.RawMessage + // AddedPersonIDs is an assignment change's delta. It is not on the + // feed: it goes into the recording's own history, where admission + // reads it. + AddedPersonIDs []int64 +} + +// performer is who the feed's performer filters match: the delegate when +// there is one, the creator otherwise. +func (ev Event) performer() int64 { + if ev.PerformedByID != 0 { + return ev.PerformedByID + } + return ev.CreatorID +} + +// fedEvent is an emitted event, and the lanes it is on. +type fedEvent struct { + ev Event + lanes Lane +} + +// Emit publishes a new event on lanes, and returns it as published. An +// event about a recording the world holds is added to that recording's +// history too, under the same id, as Basecamp records it. +func (s *Server) Emit(ev Event, lanes Lane) Event { + s.mu.Lock() + defer s.mu.Unlock() + last := int64(0) + if n := len(s.events); n > 0 { + last = s.events[n-1].ev.ID + } + if ev.ID == 0 { + ev.ID = last + 1 + } + ev.Details = slices.Clone(ev.Details) + ev.AddedPersonIDs = slices.Clone(ev.AddedPersonIDs) + if err := s.completeLocked(&ev, last); err != nil { + s.r.Errorf("fakebasecamp: emitting %s: %v", ev.EventType, err) + return ev + } + s.events = append(s.events, &fedEvent{ev: ev, lanes: lanes & Both}) + if r, ok := s.world.Recordings[ev.RecordingID]; ok { + r.Events = append(r.Events, RecordingEvent{ + ID: ev.ID, Action: ev.Action, CreatorID: ev.CreatorID, PerformedByID: ev.PerformedByID, + CreatedAt: ev.CreatedAt, AddedPersonIDs: ev.AddedPersonIDs, + }) + } + if lanes&Live != 0 { + s.pushLocked(ev) + } + s.notifyLocked() + return ev +} + +// Publish puts an emitted event on more lanes. On Live it is pushed again +// even when it was pushed before: a duplicate, as the live lane may send. +// On Poll it is served from then on, to every walk that has not yet passed +// its id; a walk already past it never sees it, as on Basecamp. +func (s *Server) Publish(id int64, lanes Lane) { + s.mu.Lock() + defer s.mu.Unlock() + i, found := slices.BinarySearchFunc(s.events, id, func(e *fedEvent, id int64) int { return cmp.Compare(e.ev.ID, id) }) + if !found { + s.r.Errorf("fakebasecamp: publishing event %d, which was never emitted", id) + return + } + fe := s.events[i] + fe.lanes |= lanes & Both + if lanes&Live != 0 { + s.pushLocked(fe.ev) + } + s.notifyLocked() +} + +// completeLocked fills an event's defaults and refuses one eventfeed would. +func (s *Server) completeLocked(ev *Event, last int64) error { + if ev.ID <= last { + return fmt.Errorf("event %d does not follow event %d: the feed's ids only ascend", ev.ID, last) + } + i := strings.LastIndex(ev.EventType, ".") + if i <= 0 || i == len(ev.EventType)-1 { + return fmt.Errorf("%q is not a cataloged event type", ev.EventType) + } + if ev.BucketID <= 0 || ev.RecordingID <= 0 || ev.CreatorID <= 0 || ev.PerformedByID < 0 { + return fmt.Errorf("event %d needs a bucket, a recording and a creator", ev.ID) + } + if ev.Kind == "" { + ev.Kind = strings.ReplaceAll(ev.EventType, ".", "_") + } + if ev.Action == "" { + ev.Action = ev.EventType[i+1:] + } + if ev.ActorType == "" { + ev.ActorType = "person" + if s.world.isAgent(ev.performer()) { + ev.ActorType = "agent" + } + } + if ev.CreatedAt.IsZero() { + ev.CreatedAt = s.opts.now().UTC() + } + return nil +} + +// pollRow is an event as the poll lane serves it. +type pollRow struct { + ID int64 `json:"id"` + Kind string `json:"kind"` + Action string `json:"action"` + EventType string `json:"event_type"` + BucketID int64 `json:"bucket_id"` + CreatorID int64 `json:"creator_id"` + PerformedByID *int64 `json:"performed_by_id"` + RecordingID int64 `json:"recording_id"` + CreatedAt time.Time `json:"created_at"` + Details json.RawMessage `json:"details,omitempty"` +} + +// pushPayload is an event as the cable pushes it: the poll row and the two +// transport members only the push carries. +type pushPayload struct { + pollRow + ActorType string `json:"actor_type"` + VisibleToClients bool `json:"visible_to_clients"` +} + +func rowOf(ev Event) pollRow { + row := pollRow{ + ID: ev.ID, Kind: ev.Kind, Action: ev.Action, EventType: ev.EventType, + BucketID: ev.BucketID, CreatorID: ev.CreatorID, RecordingID: ev.RecordingID, + CreatedAt: ev.CreatedAt, Details: ev.Details, + } + if ev.PerformedByID != 0 { + performer := ev.PerformedByID + row.PerformedByID = &performer + } + return row +} + +// filters are the feed's filter dimensions, on either lane. +type filters struct { + // present is each dimension the request named, with its value as it + // was presented: what a page's next URL has to repeat. + present url.Values + types []string + buckets []int64 + creators []int64 + // performers and excluded hold decimal ids or "self". + performers []string + excluded []string + actorTypes []string +} + +// filterKeys are the dimensions, in the order the SDK writes them. +var filterKeys = []string{"types", "buckets", "creators", "performers", "exclude_performers", "actor_types"} + +// parseFilters reads the dimensions through get, which answers a key's +// comma-joined value. +func parseFilters(get func(key string) string) (filters, error) { + f := filters{present: url.Values{}} + for _, key := range filterKeys { + raw := get(key) + if raw == "" { + continue + } + f.present.Set(key, raw) + values := strings.Split(raw, ",") + switch key { + case "types": + f.types = values + case "buckets", "creators": + ids := make([]int64, 0, len(values)) + for _, v := range values { + id, err := strconv.ParseInt(v, 10, 64) + if err != nil || id <= 0 { + return filters{}, fmt.Errorf("%s holds %q, which is not an id", key, v) + } + ids = append(ids, id) + } + if key == "buckets" { + f.buckets = ids + } else { + f.creators = ids + } + case "performers", "exclude_performers": + for _, v := range values { + if id, err := strconv.ParseInt(v, 10, 64); v != "self" && (err != nil || id <= 0) { + return filters{}, fmt.Errorf("%s holds %q, which is neither an id nor self", key, v) + } + } + if key == "performers" { + f.performers = values + } else { + f.excluded = values + } + case "actor_types": + for _, v := range values { + if v != "person" && v != "agent" { + return filters{}, fmt.Errorf("actor_types holds %q", v) + } + } + f.actorTypes = values + } + } + return f, nil +} + +// visibleLocked reports whether caller sees ev through these filters. An +// event in a project the caller is not on is never seen. +func (s *Server) visibleLocked(ev Event, f filters, caller int64) bool { + if !s.world.member(ev.BucketID, caller) { + return false + } + names := func(list []string, person int64) bool { + for _, v := range list { + if v == "self" && person == caller || v == strconv.FormatInt(person, 10) { + return true + } + } + return false + } + switch { + case f.types != nil && !slices.Contains(f.types, ev.EventType), + f.buckets != nil && !slices.Contains(f.buckets, ev.BucketID), + f.creators != nil && !slices.Contains(f.creators, ev.CreatorID), + f.performers != nil && !names(f.performers, ev.performer()), + names(f.excluded, ev.performer()), + f.actorTypes != nil && !slices.Contains(f.actorTypes, ev.ActorType): + return false + } + return true +} + +// position is the opaque token for a walk that has passed id. +func position(id int64) string { return "fake-" + strconv.FormatInt(id, 10) } + +// events is the poll lane: a page of the events past the cursor that the +// filters let through. A page walks at most the page size of poll-lane +// rows, so a page can be empty while the walk has more to serve, and it +// says so with next. +func (c *call) events() answer { + q := c.req.URL.Query() + f, err := parseFilters(q.Get) + if err != nil { + return c.jsonAnswer(http.StatusBadRequest, map[string]string{"error": err.Error(), "reason": "invalid_filter"}) + } + poll := make([]Event, 0, len(c.s.events)) + for _, fe := range c.s.events { + if fe.lanes&Poll != 0 { + poll = append(poll, fe.ev) + } + } + head := int64(0) + if len(poll) > 0 { + head = poll[len(poll)-1].ID + } + + since, pos := q.Get("since"), q.Get("position") + var cursor int64 + switch { + case since != "" && pos != "": + return c.jsonAnswer(http.StatusBadRequest, map[string]string{"error": "Pass a since or a position, not both"}) + case pos != "": + n, err := strconv.ParseInt(strings.TrimPrefix(pos, "fake-"), 10, 64) + if !strings.HasPrefix(pos, "fake-") || err != nil || n < 0 { + return c.jsonAnswer(http.StatusBadRequest, map[string]string{"error": "Unrecognized position", "reason": "invalid_position"}) + } + cursor = n + case since == "" || since == "now": + // Neither is the bare present entry, which Basecamp reads as + // since=now (the SDK's PollEventsOptions: "leave both empty to + // enter at the present"). eventfeed never sends one, but `basecamp + // events poll` does on purpose, so it is on the contract and not + // reported. + return c.jsonAnswer(http.StatusOK, map[string]any{"events": []pollRow{}, "position": position(head)}) + default: + n, err := strconv.ParseInt(since, 10, 64) + if err != nil || n < 0 { + return c.jsonAnswer(http.StatusBadRequest, map[string]string{"error": "since is neither now nor an event id"}) + } + cursor = n + } + + start, _ := slices.BinarySearchFunc(poll, cursor+1, func(ev Event, id int64) int { return cmp.Compare(ev.ID, id) }) + end := min(start+c.s.opts.pageSize, len(poll)) + rows := []pollRow{} + at := cursor + for _, ev := range poll[start:end] { + at = ev.ID + if c.s.visibleLocked(ev, f, c.caller) { + rows = append(rows, rowOf(ev)) + } + } + page := map[string]any{"events": rows, "position": position(at)} + if end < len(poll) { + next := url.Values{"position": {position(at)}} + for key, values := range f.present { + next[key] = values + } + page["next"] = c.accountURL("events.json") + "?" + next.Encode() + } + return c.jsonAnswer(http.StatusOK, page) +} diff --git a/internal/connector/fakebasecamp/feed_test.go b/internal/connector/fakebasecamp/feed_test.go new file mode 100644 index 000000000..5367a0020 --- /dev/null +++ b/internal/connector/fakebasecamp/feed_test.go @@ -0,0 +1,169 @@ +package fakebasecamp_test + +import ( + "context" + "sync" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/basecamp/basecamp-sdk/go/pkg/basecamp" + "github.com/basecamp/basecamp-sdk/go/pkg/basecamp/eventfeed" + + "github.com/basecamp/basecamp-cli/internal/connector/fakebasecamp" +) + +// The fake is held to eventfeed itself: the SDK's live connector, over its +// real seams and its real WebSocket transport, run in process against it. + +// connectorFilters are the filters `basecamp connect` runs with. +var connectorFilters = eventfeed.Filters{ + Buckets: []int64{fakebasecamp.ProjectID}, + ExcludePerformers: []int64{fakebasecamp.AgentID}, + ActorTypes: []string{"person"}, +} + +// liveFeed is an eventfeed connector reading the fake, and what it delivers. +type liveFeed struct { + delivered chan eventfeed.Event + failed chan error + + mu sync.Mutex + dropped int +} + +func runLiveFeed(t *testing.T, s *fakebasecamp.Server, bearer string, startAt eventfeed.Start) *liveFeed { + t.Helper() + live, err := eventfeed.NewLive(&basecamp.Config{BaseURL: s.URL()}, &basecamp.StaticTokenProvider{Token: bearer}, + "999", eventfeed.AccountLane, basecamp.WithMaxRetries(0)) + require.NoError(t, err) + f := &liveFeed{delivered: make(chan eventfeed.Event, 100), failed: make(chan error, 1)} + conn, err := live.Connect( + eventfeed.WithStart(startAt), + eventfeed.WithFilters(connectorFilters), + eventfeed.WithObserver(eventfeed.Observer{Disconnected: func(string, error) { + f.mu.Lock() + defer f.mu.Unlock() + f.dropped++ + }}), + ) + require.NoError(t, err) + + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + go func() { + defer close(done) + for ev, err := range conn.Events(ctx) { + if err != nil { + f.failed <- err + return + } + f.delivered <- ev + } + }() + t.Cleanup(func() { + cancel() + _ = conn.Close() + conn.Wait() + <-done + }) + return f +} + +// next is the next event the feed delivers. +func (f *liveFeed) next(t *testing.T) eventfeed.Event { + t.Helper() + select { + case ev := <-f.delivered: + return ev + case err := <-f.failed: + require.FailNow(t, "the feed ended", "%v", err) + case <-bounded(t).Done(): + require.FailNow(t, "no event was delivered") + } + return eventfeed.Event{} +} + +// expect asserts the feed delivers exactly these events next, in order. +func (f *liveFeed) expect(t *testing.T, events ...fakebasecamp.Event) { + t.Helper() + for _, want := range events { + got := f.next(t) + assert.Equal(t, want.ID, got.ID) + assert.Equal(t, want.EventType, got.EventType) + assert.Equal(t, want.BucketID, got.BucketID) + assert.Equal(t, want.RecordingID, got.RecordingID) + assert.Equal(t, want.CreatorID, got.CreatorID) + } +} + +func commentBy(creator, bucket int64) fakebasecamp.Event { + return fakebasecamp.Event{EventType: "comment.created", BucketID: bucket, RecordingID: 9001, CreatorID: creator} +} + +// The catch-up walks the poll lane page by page, through a page the +// filters empty, and an event both lanes carry is delivered once. +func TestLiveFeedWalksThePollLaneAndDedupesAcrossLanes(t *testing.T) { + s, bearer := start(t, fakebasecamp.WithPageSize(2)) + op, agent := fakebasecamp.OperatorID, fakebasecamp.AgentID + served, other := fakebasecamp.ProjectID, otherProject + + // History, on the poll lane only. Pages of two: the second page holds + // only the agent's own event and one in another project, and the + // filters leave it empty. + e1 := s.Emit(commentBy(op, served), fakebasecamp.Poll) + s.Emit(commentBy(op, other), fakebasecamp.Poll) + s.Emit(commentBy(agent, served), fakebasecamp.Poll) + s.Emit(commentBy(op, other), fakebasecamp.Poll) + e5 := s.Emit(commentBy(op, served), fakebasecamp.Poll) + + // Hold the catch-up walk until the subscription is confirmed, so an + // event emitted now is both buffered from the cable and served by the + // walk. + walk := s.Gate(fakebasecamp.RouteEvents) + feed := runLiveFeed(t, s, bearer, eventfeed.StartBeginning()) + await(t, s, "a confirmed subscription and a held walk", func() bool { return s.Subscribers() == 1 && walk.Waiting() == 1 }) + e6 := s.Emit(commentBy(op, served), fakebasecamp.Both) + walk.Release() + + feed.expect(t, e1, e5, e6) + e7 := s.Emit(commentBy(op, served), fakebasecamp.Live) + feed.expect(t, e7) // not e6 again + + var polls []string + for _, r := range s.Requests() { + if r.Route == fakebasecamp.RouteEvents { + polls = append(polls, r.Query.Encode()) + } + } + filters := "actor_types=person&buckets=48699913&exclude_performers=52007412" + assert.Equal(t, []string{ + filters + "&since=0", + filters + "&position=fake-2", + filters + "&position=fake-4", + }, polls) +} + +// A cable dropped while events are published is caught up from the poll +// lane on the way back, before the feed streams again. +func TestLiveFeedCatchesUpAfterADroppedCable(t *testing.T) { + s, bearer := start(t) + feed := runLiveFeed(t, s, bearer, eventfeed.StartPresent()) + await(t, s, "a confirmed subscription", func() bool { return s.Subscribers() == 1 }) + feed.expect(t, s.Emit(commentBy(fakebasecamp.OperatorID, fakebasecamp.ProjectID), fakebasecamp.Live)) + + // Hold the reconnect at its mint: the gap stays open while the event is + // published, so only the catch-up can deliver it. + mint := s.Gate(fakebasecamp.RouteStreamTicket) + require.Equal(t, 1, s.DropCable()) + await(t, s, "the reconnect", func() bool { return mint.Waiting() == 1 }) + missed := s.Emit(commentBy(fakebasecamp.OperatorID, fakebasecamp.ProjectID), fakebasecamp.Both) + mint.Release() + + feed.expect(t, missed) + feed.mu.Lock() + assert.Equal(t, 1, feed.dropped) + feed.mu.Unlock() + feed.expect(t, s.Emit(commentBy(fakebasecamp.OperatorID, fakebasecamp.ProjectID), fakebasecamp.Live)) +} diff --git a/internal/connector/fakebasecamp/helpers_test.go b/internal/connector/fakebasecamp/helpers_test.go new file mode 100644 index 000000000..30360b605 --- /dev/null +++ b/internal/connector/fakebasecamp/helpers_test.go @@ -0,0 +1,103 @@ +package fakebasecamp_test + +import ( + "context" + "fmt" + "io" + "net/http" + "strings" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/require" + + "github.com/basecamp/basecamp-sdk/go/pkg/basecamp" + + "github.com/basecamp/basecamp-cli/internal/connector/fakebasecamp" +) + +// otherProject is a second project the agent is on. +const otherProject int64 = 48699914 + +// waitFor is how long anything here waits for the fake or a client to get +// somewhere. Nothing waits on a timer: this is only the bound. +const waitFor = 15 * time.Second + +// start serves the default world, with a second project the agent is on and +// a bearer token for the agent, which it returns. +func start(t *testing.T, opts ...fakebasecamp.Option) (*fakebasecamp.Server, string) { + t.Helper() + s := fakebasecamp.Start(t, fakebasecamp.DefaultWorld(), opts...) + const token = "agent-token" + s.Update(func(w *fakebasecamp.World) { + w.Projects[otherProject] = &fakebasecamp.Project{ + ID: otherProject, Name: "Launch", Members: []int64{fakebasecamp.OperatorID, fakebasecamp.AgentID}, + } + w.Tokens[token] = &fakebasecamp.Token{PersonID: fakebasecamp.AgentID, Scope: fakebasecamp.ScopeFull} + }) + return s, token +} + +// bounded is the test's context, ending after waitFor. +func bounded(t *testing.T) context.Context { + t.Helper() + ctx, cancel := context.WithTimeout(t.Context(), waitFor) + t.Cleanup(cancel) + return ctx +} + +// await waits for cond, failing the test when it does not come to hold. +func await(t *testing.T, s *fakebasecamp.Server, what string, cond func() bool) { + t.Helper() + require.NoError(t, s.Await(bounded(t), cond), "waiting for %s", what) +} + +// accountClient is the SDK, in the fake's account, as the bearer. +func accountClient(s *fakebasecamp.Server, bearer string) *basecamp.AccountClient { + return basecamp.NewClient(&basecamp.Config{BaseURL: s.URL()}, &basecamp.StaticTokenProvider{Token: bearer}, + basecamp.WithMaxRetries(1)).ForAccount("999") +} + +// do makes one request to the fake and returns the status and the body. +func do(t *testing.T, s *fakebasecamp.Server, method, path, bearer, form string) (int, http.Header, string) { + t.Helper() + var body io.Reader + if form != "" { + body = strings.NewReader(form) + } + req, err := http.NewRequestWithContext(bounded(t), method, s.URL()+path, body) + require.NoError(t, err) + if form != "" { + req.Header.Set("Content-Type", "application/x-www-form-urlencoded") + } + if bearer != "" { + req.Header.Set("Authorization", "Bearer "+bearer) + } + resp, err := http.DefaultClient.Do(req) + require.NoError(t, err) + defer resp.Body.Close() + data, err := io.ReadAll(resp.Body) + require.NoError(t, err) + return resp.StatusCode, resp.Header, string(data) +} + +// reporter is a Reporter that keeps what the fake reports, for a test of +// the reporting itself. +type reporter struct { + *testing.T + mu sync.Mutex + errors []string +} + +func (r *reporter) Errorf(format string, args ...any) { + r.mu.Lock() + defer r.mu.Unlock() + r.errors = append(r.errors, fmt.Sprintf(format, args...)) +} + +func (r *reporter) reported() []string { + r.mu.Lock() + defer r.mu.Unlock() + return append([]string(nil), r.errors...) +} diff --git a/internal/connector/fakebasecamp/oauth.go b/internal/connector/fakebasecamp/oauth.go new file mode 100644 index 000000000..4e1f640f8 --- /dev/null +++ b/internal/connector/fakebasecamp/oauth.go @@ -0,0 +1,126 @@ +package fakebasecamp + +import ( + "fmt" + "net/http" +) + +// The authorization server: discovery, the agent connection ceremony, and +// the client_credentials mint. Basecamp's contract for the ceremony is +// bc3's doc/oauth/agent_connections.md; internal/auth is built to it, and +// this answers it. + +// asMetadata is RFC 8414 metadata naming the fake as its own issuer. +func (c *call) asMetadata() answer { + return c.jsonAnswer(http.StatusOK, map[string]any{ + "issuer": c.base(), + "token_endpoint": c.base() + "/oauth/tokens", + "device_authorization_endpoint": c.base() + "/oauth/device_authorizations", + "grant_types_supported": []string{ + "urn:ietf:params:oauth:grant-type:device_code", "client_credentials", "refresh_token", + }, + }) +} + +// resourceMetadata is RFC 9728 metadata pointing the resource at the fake's +// authorization server. +func (c *call) resourceMetadata() answer { + return c.jsonAnswer(http.StatusOK, map[string]any{ + "resource": c.base(), + "authorization_servers": []string{c.base()}, + }) +} + +// agentConnections is the anonymous intake: a code for the operator to +// approve, and where to poll for the answer. +func (c *call) agentConnections() answer { + if c.req.PostForm.Get("device_name") == "" { + return oauthError(http.StatusBadRequest, "invalid_request") + } + code := fmt.Sprintf("fake-device-%d", c.s.nextSerialLocked()) + c.s.devices[code] = true + return c.jsonAnswer(http.StatusOK, map[string]any{ + "device_code": code, + "user_code": "WDJB-MJHT", + "verification_uri": c.base() + "/connect", + "verification_uri_complete": c.base() + "/connect?user_code=WDJB-MJHT", + "token_uri": c.base() + "/oauth/agent_connection_tokens", + "expires_in": 600, + "interval": 1, + }) +} + +// agentConnectionTokens is the poll. The operator has always approved by +// the time it arrives: it hands over the client World.Connection names, +// provisioned with the scope approved, under a newly minted secret. The +// code is spent by the handover, as Basecamp spends it. +func (c *call) agentConnectionTokens() answer { + code := c.req.PostForm.Get("device_code") + if !c.s.devices[code] { + return oauthError(http.StatusBadRequest, "invalid_grant") + } + conn := c.s.world.Connection + client, ok := c.s.world.Agents[conn.ClientID] + if !ok { + c.s.r.Errorf("fakebasecamp: the connection hands over client %q, which the world does not hold", conn.ClientID) + return oauthError(http.StatusBadRequest, "invalid_grant") + } + delete(c.s.devices, code) + client.Scope = conn.Scope + if client.Scope == "" { + client.Scope = ScopeFull + } + client.Secret = fmt.Sprintf("%s-secret-%d", client.ID, c.s.nextSerialLocked()) + return c.jsonAnswer(http.StatusOK, map[string]any{ + "client_id": client.ID, + "client_secret": client.Secret, + "account_id": c.s.world.Account.ID, + "scope": client.Scope, + }) +} + +// token mints a client_credentials token for an agent's client. A secret +// the client does not hold now, rotated away or never issued, is +// invalid_client. The token carries the scope the request asks for, or the +// client's own when it asks for none; asking for more than the client was +// approved for, or for a scope there is no such thing as, is invalid_scope +// (RFC 6749 section 5.2), as Basecamp refuses to widen a credential. +func (c *call) token() answer { + form := c.req.PostForm + if form.Get("grant_type") != "client_credentials" { + return oauthError(http.StatusBadRequest, "unsupported_grant_type") + } + client, ok := c.s.world.Agents[form.Get("client_id")] + if !ok || client.Secret != form.Get("client_secret") { + return oauthError(http.StatusUnauthorized, "invalid_client") + } + scope := client.Scope + if asked := form.Get("scope"); asked != "" { + if !within(asked, client.Scope) { + return oauthError(http.StatusBadRequest, "invalid_scope") + } + scope = asked + } + token := fmt.Sprintf("fake-token-%d", c.s.nextSerialLocked()) + c.s.world.Tokens[token] = &Token{PersonID: client.PersonID, Scope: scope} + return c.jsonAnswer(http.StatusOK, map[string]any{ + "access_token": token, + "token_type": "bearer", + "expires_in": 3600, + "resource": fmt.Sprintf("urn:bc:agent:%d", client.PersonID), + "scope": scope, + }) +} + +// within reports whether asked is a scope no wider than approved: read is +// within either, full only within full. +func within(asked, approved string) bool { + switch asked { + case ScopeRead: + return approved == ScopeRead || approved == ScopeFull + case ScopeFull: + return approved == ScopeFull + default: + return false + } +} diff --git a/internal/connector/fakebasecamp/reads_test.go b/internal/connector/fakebasecamp/reads_test.go new file mode 100644 index 000000000..11f1d50dd --- /dev/null +++ b/internal/connector/fakebasecamp/reads_test.go @@ -0,0 +1,160 @@ +package fakebasecamp_test + +import ( + "net/http" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/basecamp/basecamp-sdk/go/pkg/basecamp" + + "github.com/basecamp/basecamp-cli/internal/connector/fakebasecamp" +) + +// The reads setup and admission make decode through the SDK the way the +// real ones do. + +func TestAccountReads(t *testing.T) { + s, bearer := start(t) + s.Update(func(w *fakebasecamp.World) { + w.People[fakebasecamp.AgentID].BossID = fakebasecamp.OperatorID + w.People[1003] = &fakebasecamp.Person{ID: 1003, Name: "Client Person", Client: true} + w.People[1004] = &fakebasecamp.Person{ID: 1004, Name: "Colleague", ReadableBy: []int64{fakebasecamp.OperatorID}} + }) + agent := accountClient(s, bearer) + operator := accountClient(s, fakebasecamp.OperatorToken) + + me, err := agent.People().Me(bounded(t)) + require.NoError(t, err) + assert.Equal(t, fakebasecamp.AgentID, me.ID) + assert.Equal(t, "Agent", me.PersonableType) + id, ok := basecamp.PersonIDFromSGID(me.AttachableSGID) + assert.True(t, ok) + assert.Equal(t, fakebasecamp.AgentID, id, "the sgid names the person, as a mention's does") + _, _, body := do(t, s, http.MethodGet, "/999/my/profile.json", bearer, "") + assert.Contains(t, body, `"boss":{"id":26909558,"name":"Operator"}`) + + client, err := agent.People().Get(bounded(t), 1003) + require.NoError(t, err) + assert.True(t, client.Client) + _, err = agent.People().Get(bounded(t), 1004) + assertStatus(t, http.StatusForbidden, err) + colleague, err := operator.People().Get(bounded(t), 1004) + require.NoError(t, err) + assert.Equal(t, "Colleague", colleague.Name) + _, err = agent.People().Get(bounded(t), 424242) + assertStatus(t, http.StatusNotFound, err) + + projects, err := agent.Projects().List(bounded(t), nil) + require.NoError(t, err) + require.Len(t, projects.Projects, 2) + assert.Equal(t, fakebasecamp.ProjectID, projects.Projects[0].ID) + project, err := agent.Projects().Get(bounded(t), fakebasecamp.ProjectID) + require.NoError(t, err) + assert.Equal(t, fakebasecamp.ProjectName, project.Name) + people, err := agent.People().ListProjectPeople(bounded(t), fakebasecamp.ProjectID, nil) + require.NoError(t, err) + assert.Len(t, people.People, 2) + + // A project the caller is not on is not found. + s.Update(func(w *fakebasecamp.World) { + w.Projects[777] = &fakebasecamp.Project{ID: 777, Name: "Elsewhere", Members: []int64{fakebasecamp.OperatorID}} + }) + _, err = agent.Projects().Get(bounded(t), 777) + assertStatus(t, http.StatusNotFound, err) + + // authorization.json answers a person with a Launchpad identity, and + // nobody else. + status, _, _ := do(t, s, http.MethodGet, "/authorization.json", bearer, "") + assert.Equal(t, http.StatusUnauthorized, status) + s.Update(func(w *fakebasecamp.World) { w.People[fakebasecamp.OperatorID].IdentityID = 4242 }) + status, _, body = do(t, s, http.MethodGet, "/authorization.json", fakebasecamp.OperatorToken, "") + assert.Equal(t, http.StatusOK, status) + assert.Contains(t, body, `"id":4242`) + + // Nothing under the account answers without a token the fake issued. + status, _, _ = do(t, s, http.MethodGet, "/999/projects.json", "forged", "") + assert.Equal(t, http.StatusUnauthorized, status) +} + +// Every recording the connector admits summarizes through the SDK's own +// router, with the agent's mention found where that type keeps its rich +// text; and the subscription and history reads answer what admission asks. +func TestRecordingReads(t *testing.T) { + s, bearer := start(t) + mention := "Over to you " + fakebasecamp.Mention(fakebasecamp.AgentID) + "
" + project := fakebasecamp.ProjectID + s.Update(func(w *fakebasecamp.World) { + for _, r := range []*fakebasecamp.Recording{ + {ID: 101, Type: "Message", Title: "Kickoff", Content: mention}, + {ID: 102, Type: "Comment", ParentID: 101, Content: mention, SubscriberIDs: []int64{fakebasecamp.AgentID}}, + {ID: 103, Type: "Todo", Title: "Ship it", Content: mention, AssigneeIDs: []int64{fakebasecamp.AgentID}}, + {ID: 104, Type: "Kanban::Card", Title: "Card", Content: mention}, + {ID: 105, Type: "Chat::Transcript", Title: "Campfire"}, + {ID: 106, Type: "Chat::Lines::RichText", ParentID: 105, Content: mention}, + } { + r.BucketID, r.CreatorID = project, fakebasecamp.OperatorID + w.Recordings[r.ID] = r + } + }) + agent := accountClient(s, bearer) + + for eventType, id := range map[string]int64{ + "message.created": 101, "comment.created": 102, "todo.created": 103, "card.created": 104, "chat.line.created": 106, + } { + summary, err := agent.Recordings().Summarize(bounded(t), basecamp.RecordingRef{BucketID: project, RecordingID: id, EventType: eventType}) + require.NoError(t, err, eventType) + assert.Equal(t, id, summary.ID, eventType) + assert.Equal(t, []int64{fakebasecamp.AgentID}, summary.MentionedPersonIDs, eventType) + assert.Equal(t, fakebasecamp.OperatorID, summary.Creator.ID, eventType) + } + todo, err := agent.Todos().Get(bounded(t), 103) + require.NoError(t, err) + require.Len(t, todo.Assignees, 1) + assert.Equal(t, fakebasecamp.AgentID, todo.Assignees[0].ID) + + // A typed read of another type's recording is not found. + _, err = agent.Comments().Get(bounded(t), 101) + assertStatus(t, http.StatusNotFound, err) + + sub, err := agent.Subscriptions().Get(bounded(t), 102) + require.NoError(t, err) + assert.True(t, sub.Subscribed) + sub, err = agent.Subscriptions().Get(bounded(t), 101) + require.NoError(t, err) + assert.False(t, sub.Subscribed) + + // An emitted event about a recording lands in its history, under the + // feed's id, with the delta admission reads. + ev := s.Emit(fakebasecamp.Event{ + EventType: "todo.assignment_changed", BucketID: project, RecordingID: 103, + CreatorID: fakebasecamp.OperatorID, AddedPersonIDs: []int64{fakebasecamp.AgentID}, + }, fakebasecamp.Both) + history, err := agent.Events().List(bounded(t), 103, &basecamp.EventListOptions{Page: 1}) + require.NoError(t, err) + require.Len(t, history.Events, 1) + assert.Equal(t, ev.ID, history.Events[0].ID) + assert.Equal(t, "assignment_changed", history.Events[0].Action) + require.NotNil(t, history.Events[0].Details) + assert.Equal(t, []int64{fakebasecamp.AgentID}, history.Events[0].Details.AddedPersonIDs) + empty, err := agent.Events().List(bounded(t), 103, &basecamp.EventListOptions{Page: 2}) + require.NoError(t, err) + assert.Empty(t, empty.Events) + + // Nothing in a project the caller is not on can be read. + s.Update(func(w *fakebasecamp.World) { + w.Projects[project].Members = []int64{fakebasecamp.OperatorID} + }) + _, err = agent.Recordings().Summarize(bounded(t), basecamp.RecordingRef{BucketID: project, RecordingID: 102, EventType: "comment.created"}) + assertStatus(t, http.StatusNotFound, err) + _, err = agent.Subscriptions().Get(bounded(t), 102) + assertStatus(t, http.StatusNotFound, err) +} + +func assertStatus(t *testing.T, status int, err error) { + t.Helper() + var apiErr *basecamp.Error + require.ErrorAs(t, err, &apiErr) + assert.Equal(t, status, apiErr.HTTPStatus) +} diff --git a/internal/connector/fakebasecamp/recordings.go b/internal/connector/fakebasecamp/recordings.go new file mode 100644 index 000000000..fd23d6b3b --- /dev/null +++ b/internal/connector/fakebasecamp/recordings.go @@ -0,0 +1,221 @@ +package fakebasecamp + +import ( + "net/http" + "slices" + "strconv" + "strings" +) + +// The recording reads admission makes: the typed read a recording summary +// is made from, the agent's subscription, and a recording's own history. +// Each is served only to a member of the recording's project; anyone else +// is answered 404, as Basecamp answers a recording they cannot see. + +const ( + campfireType = "Chat::Transcript" + chatLinePrefix = "Chat::Lines::" + // recordingEventsPage is how many history entries a page carries: + // fifteen, the first page Basecamp's geared pagination serves. The + // connector walks pages until one comes back empty and never counts on + // the size, so the fake keeps one size for every page; a small one + // makes a walk past the first page cheap to set up. + recordingEventsPage = 15 +) + +// recordingRead serves the typed read for one recording type. The read +// answers 404 for a recording of another type, as Basecamp's does. +func recordingRead(typ string) func(c *call) answer { + return func(c *call) answer { + r, ok := c.visibleRecording("id") + if !ok || r.Type != typ { + return answer{status: http.StatusNotFound} + } + return c.jsonAnswer(http.StatusOK, c.renderRecording(r)) + } +} + +// visibleRecording is the recording a path value names, when the caller is +// on its project. +func (c *call) visibleRecording(name string) (*Recording, bool) { + id, ok := c.pathID(name) + if !ok { + return nil, false + } + r, ok := c.s.world.Recordings[id] + if !ok || !c.s.world.member(r.BucketID, c.caller) { + return nil, false + } + return r, true +} + +// renderRecording is a recording as its typed read renders it. The members +// each type keeps its rich text in differ, and are the ones that matter: a +// to-do's title is plain and its description rich, a card's content is. +func (c *call) renderRecording(r *Recording) map[string]any { + status := r.Status + if status == "" { + status = "active" + } + out := map[string]any{ + "id": r.ID, + "status": status, + "visible_to_clients": false, + "created_at": epoch, + "updated_at": epoch, + "title": r.Title, + "inherits_status": true, + "type": r.Type, + "url": c.recordingURL(r) + ".json", + "app_url": c.recordingURL(r), + "bookmark_url": c.accountURL("my/bookmarks/%d.json", r.ID), + "bucket": c.renderBucket(r.BucketID), + } + if creator, ok := c.s.world.People[r.CreatorID]; ok { + out["creator"] = c.renderPerson(creator) + } + if parent, ok := c.s.world.Recordings[r.ParentID]; ok { + out["parent"] = map[string]any{ + "id": parent.ID, "title": parent.Title, "type": parent.Type, + "url": c.recordingURL(parent) + ".json", "app_url": c.recordingURL(parent), + } + } + switch { + case r.Type == "Todo": + out["content"] = r.Title + out["description"] = r.Content + out["completed"] = r.Completed + out["assignees"] = c.renderPeople(r.AssigneeIDs) + out["completion_subscribers"] = []personJSON{} + case r.Type == "Kanban::Card": + out["content"] = r.Content + out["completed"] = r.Completed + out["assignees"] = c.renderPeople(r.AssigneeIDs) + case r.Type == "Message": + out["subject"] = r.Title + out["content"] = r.Content + case r.Type == "Comment", strings.HasPrefix(r.Type, chatLinePrefix): + out["content"] = r.Content + } + return out +} + +func (c *call) renderBucket(id int64) map[string]any { + name := "" + if p, ok := c.s.world.Projects[id]; ok { + name = p.Name + } + return map[string]any{"id": id, "name": name, "type": "Project"} +} + +func (c *call) renderPeople(ids []int64) []personJSON { + out := []personJSON{} + for _, id := range ids { + if p, ok := c.s.world.People[id]; ok { + out = append(out, c.renderPerson(p)) + } + } + return out +} + +// recordingURL is where a recording lives, under its type's route. +func (c *call) recordingURL(r *Recording) string { + switch { + case r.Type == "Comment": + return c.accountURL("comments/%d", r.ID) + case r.Type == "Message": + return c.accountURL("messages/%d", r.ID) + case r.Type == "Todo": + return c.accountURL("todos/%d", r.ID) + case r.Type == "Kanban::Card": + return c.accountURL("card_tables/cards/%d", r.ID) + case r.Type == campfireType: + return c.accountURL("chats/%d", r.ID) + case strings.HasPrefix(r.Type, chatLinePrefix): + return c.accountURL("chats/%d/lines/%d", r.ParentID, r.ID) + default: + return c.accountURL("recordings/%d", r.ID) + } +} + +// campfires is every Campfire the caller can see: where a line is looked +// for when its project's dock does not hold it. +func (c *call) campfires() answer { + out := []map[string]any{} + for _, r := range c.s.sortedRecordings() { + if r.Type == campfireType && c.s.world.member(r.BucketID, c.caller) { + out = append(out, c.renderRecording(r)) + } + } + return c.jsonAnswer(http.StatusOK, out) +} + +// chatLine is a line, read under the Campfire it belongs to. +func (c *call) chatLine() answer { + campfire, ok := c.visibleRecording("campfire") + if !ok || campfire.Type != campfireType { + return answer{status: http.StatusNotFound} + } + line, ok := c.visibleRecording("id") + if !ok || !strings.HasPrefix(line.Type, chatLinePrefix) || line.ParentID != campfire.ID { + return answer{status: http.StatusNotFound} + } + return c.jsonAnswer(http.StatusOK, c.renderRecording(line)) +} + +// subscription is whether the caller is subscribed to a recording. +func (c *call) subscription() answer { + r, ok := c.visibleRecording("id") + if !ok { + return answer{status: http.StatusNotFound} + } + return c.jsonAnswer(http.StatusOK, map[string]any{ + "subscribed": slices.Contains(r.SubscriberIDs, c.caller), + "count": len(r.SubscriberIDs), + "url": c.accountURL("recordings/%d/subscription.json", r.ID), + "subscribers": c.renderPeople(r.SubscriberIDs), + }) +} + +// recordingEvents is a recording's history, newest first, a page at a time. +func (c *call) recordingEvents() answer { + r, ok := c.visibleRecording("id") + if !ok { + return answer{status: http.StatusNotFound} + } + page := 1 + if raw := c.req.URL.Query().Get("page"); raw != "" { + n, err := strconv.Atoi(raw) + if err != nil || n < 1 { + return answer{status: http.StatusBadRequest} + } + page = n + } + newest := slices.Clone(r.Events) + slices.Reverse(newest) + out := []map[string]any{} + for i := (page - 1) * recordingEventsPage; i < len(newest) && i < page*recordingEventsPage; i++ { + ev := newest[i] + createdAt := ev.CreatedAt + if createdAt.IsZero() { + createdAt = epoch + } + row := map[string]any{ + "id": ev.ID, + "recording_id": r.ID, + "action": ev.Action, + "created_at": createdAt, + } + if ev.AddedPersonIDs != nil { + row["details"] = map[string]any{"added_person_ids": ev.AddedPersonIDs} + } + if creator, ok := c.s.world.People[ev.CreatorID]; ok { + row["creator"] = c.renderPerson(creator) + } + if performer, ok := c.s.world.People[ev.PerformedByID]; ok { + row["performed_by"] = c.renderPerson(performer) + } + out = append(out, row) + } + return c.jsonAnswer(http.StatusOK, out) +} diff --git a/internal/connector/fakebasecamp/server.go b/internal/connector/fakebasecamp/server.go new file mode 100644 index 000000000..d81ed2abc --- /dev/null +++ b/internal/connector/fakebasecamp/server.go @@ -0,0 +1,707 @@ +package fakebasecamp + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "net" + "net/http" + "net/http/httptest" + "net/url" + "slices" + "strconv" + "strings" + "sync" + "time" +) + +// Reporter is the part of testing.TB the fake needs: where to report a +// request it could not make sense of, and where to register its own Close. +type Reporter interface { + Helper() + Errorf(format string, args ...any) + Cleanup(func()) +} + +// Route names one endpoint the fake serves, as the request pattern it is +// mounted on. +type Route string + +// The routes the fake serves. +const ( + RouteASMetadata Route = "GET /.well-known/oauth-authorization-server" + RouteResourceMetadata Route = "GET /.well-known/oauth-protected-resource" + RouteAgentConnections Route = "POST /oauth/agent_connections" + RouteAgentConnectionTokens Route = "POST /oauth/agent_connection_tokens" //nolint:gosec // G101: a route, not a credential + RouteToken Route = "POST /oauth/tokens" //nolint:gosec // G101: a route, not a credential + RouteAuthorization Route = "GET /authorization.json" + + RouteProfile Route = "GET /{account}/my/profile.json" + RoutePerson Route = "GET /{account}/people/{id}" + RouteProjects Route = "GET /{account}/projects.json" + RouteProject Route = "GET /{account}/projects/{id}" + RouteProjectPeople Route = "GET /{account}/projects/{id}/people.json" + + RouteStreamTicket Route = "POST /{account}/events/stream_ticket.json" + RouteEvents Route = "GET /{account}/events.json" + RouteCable Route = "GET /cable" + + RouteComment Route = "GET /{account}/comments/{id}" + RouteMessage Route = "GET /{account}/messages/{id}" + RouteTodo Route = "GET /{account}/todos/{id}" + RouteCard Route = "GET /{account}/card_tables/cards/{id}" + RouteCampfires Route = "GET /{account}/chats.json" + RouteChatLine Route = "GET /{account}/chats/{campfire}/lines/{id}" + RouteSubscription Route = "GET /{account}/recordings/{id}/subscription.json" + RouteRecordingEvents Route = "GET /{account}/recordings/{id}/events.json" +) + +// RecordingReads are the typed reads a recording summary is made from, one +// per event type the connector admits. +var RecordingReads = []Route{RouteComment, RouteMessage, RouteTodo, RouteCard, RouteChatLine} + +// Request is one request the fake received. +type Request struct { + Method string + Path string + Query url.Values + // Form is a form-encoded body, parsed; nil for any other. + Form url.Values + // Route is the route it reached; empty for a request no route serves. + Route Route + // Caller is the person its credential authenticates as, zero for none: + // the bearer token's person, or on RouteCable the person the stream + // ticket was minted for. Agent says the caller is an Agent principal. + Caller int64 + Agent bool + // Status is what it was answered, zero while it is unanswered: held at + // a gate, or not yet answered. It is set before the response is + // written, so a client that has its answer finds it here. + Status int +} + +// clone is the request with maps of its own, so what one reader is handed +// cannot change what another reads. +func (r Request) clone() Request { + r.Query = cloneValues(r.Query) + r.Form = cloneValues(r.Form) + return r +} + +func cloneValues(v url.Values) url.Values { + if v == nil { + return nil + } + out := make(url.Values, len(v)) + for key, values := range v { + out[key] = slices.Clone(values) + } + return out +} + +// ErrClosed is what Await returns once the fake is closed: nothing will +// change any more, so a condition that does not hold then never will. +var ErrClosed = errors.New("fakebasecamp: the server is closed") + +// Option configures a Server. +type Option func(*options) + +// WithPageSize sets how many poll-lane rows a page walks. The default is +// 100. +func WithPageSize(n int) Option { return func(o *options) { o.pageSize = n } } + +// WithPingInterval sets how often the cable pings. The default is Action +// Cable's three seconds, which is what eventfeed's staleness rule is built +// around. +func WithPingInterval(d time.Duration) Option { return func(o *options) { o.pingInterval = d } } + +// WithListener serves on l instead of a fresh loopback port. +func WithListener(l net.Listener) Option { return func(o *options) { o.listener = l } } + +// WithClock sets what the fake reads the time from: the default created_at +// of an emitted event, and when a stream ticket expires. The default is +// time.Now. A clock that does not keep up with real time has to stay close +// to it: the connector discards a request created more than +// connector.HandoffGrace before its run started, judging by the event's +// created_at. +func WithClock(now func() time.Time) Option { return func(o *options) { o.now = now } } + +// Server is the fake Basecamp. +type Server struct { + r Reporter + srv *httptest.Server + opts options + + // done is closed by Close: it releases every held request and stops + // every cable goroutine. + done chan struct{} + // cables counts the goroutines serving cable connections, which + // httptest does not wait for once a connection is hijacked. + cables sync.WaitGroup + + mu sync.Mutex + // changed is closed, and replaced, on every change Await can observe. + changed chan struct{} + closed bool + world *World + log []Request + faults []*Injected + gates []*Gate + hooks []*hook + // events is every event emitted, in id order, on either lane. + events []*fedEvent + conns map[*cableConn]struct{} + tickets map[string]streamTicket + // devices are the agent connection codes issued and not yet spent. + devices map[string]bool + // serial numbers every token, ticket and code the fake issues. + serial int +} + +// Start serves world on a loopback port until the Reporter's cleanup. +func Start(r Reporter, world *World, opts ...Option) *Server { + r.Helper() + o := options{pageSize: 100, pingInterval: 3 * time.Second, now: time.Now} + for _, opt := range opts { + opt(&o) + } + if err := world.check(); err != nil { + r.Errorf("fakebasecamp: the world is inconsistent: %v", err) + } + s := &Server{ + r: r, + opts: o, + done: make(chan struct{}), + changed: make(chan struct{}), + world: world, + conns: map[*cableConn]struct{}{}, + tickets: map[string]streamTicket{}, + devices: map[string]bool{}, + } + s.srv = httptest.NewUnstartedServer(s.routes()) + if o.listener != nil { + _ = s.srv.Listener.Close() + s.srv.Listener = o.listener + } + s.srv.Start() + r.Cleanup(s.Close) + return s +} + +// URL is the fake's origin: the base URL, the OAuth issuer, and the cable's +// host. +func (s *Server) URL() string { return s.srv.URL } + +// Close stops the fake: every cable connection is severed, every held +// request answered, and every goroutine the fake started has returned when +// it does. It is safe to call more than once. +func (s *Server) Close() { + s.mu.Lock() + if s.closed { + s.mu.Unlock() + return + } + s.closed = true + close(s.done) + conns := s.detachConnsLocked() + s.notifyLocked() + s.mu.Unlock() + for _, c := range conns { + c.sever() + } + s.srv.Close() + s.cables.Wait() +} + +// Update changes the world under the fake's lock. fn must not call the +// Server. +func (s *Server) Update(fn func(w *World)) { + s.mu.Lock() + defer s.mu.Unlock() + fn(s.world) + if err := s.world.check(); err != nil { + s.r.Errorf("fakebasecamp: the world is inconsistent after an update: %v", err) + } + s.notifyLocked() +} + +// View reads the world under the fake's lock. fn must not keep what it is +// given, nor call the Server. +func (s *Server) View(fn func(w *World)) { + s.mu.Lock() + defer s.mu.Unlock() + fn(s.world) +} + +// Requests is every request received so far, in arrival order. The +// requests are copies: changing one changes nothing the fake holds. +func (s *Server) Requests() []Request { + s.mu.Lock() + defer s.mu.Unlock() + out := make([]Request, len(s.log)) + for i, r := range s.log { + out[i] = r.clone() + } + return out +} + +// Count is how many requests reached any of routes, answered or not. +func (s *Server) Count(routes ...Route) int { + s.mu.Lock() + defer s.mu.Unlock() + n := 0 + for _, r := range s.log { + if slices.Contains(routes, r.Route) { + n++ + } + } + return n +} + +// Await returns once cond holds, with ctx's error when ctx ends first, or +// with ErrClosed when the fake is closed first. cond is called without the +// fake's lock, so it may call the Server; it is called again after every +// change the fake makes, never on a timer. +func (s *Server) Await(ctx context.Context, cond func() bool) error { + for { + s.mu.Lock() + changed, closed := s.changed, s.closed + s.mu.Unlock() + if cond() { + return nil + } + if closed { + return ErrClosed + } + select { + case <-changed: + case <-ctx.Done(): + return ctx.Err() + } + } +} + +// Fault answers matching requests in place of their route. Its answer can +// be any status: a 200 with a body of its own is a fault too. +type Fault struct { + // Routes are the routes it answers. + Routes []Route + // When, when set, narrows it to the requests it accepts. It is called + // under the fake's lock and must not call the Server. + When func(Request) bool + // Times is how many requests it answers before it lapses; zero answers + // every one until it is removed. + Times int + + Status int + Header http.Header + Body string +} + +// Injected is a fault in force. +type Injected struct { + s *Server + fault Fault + hits int +} + +// Inject puts a fault in force. The fault injected last is the one that +// answers a request more than one would match. +func (s *Server) Inject(f Fault) *Injected { + s.mu.Lock() + defer s.mu.Unlock() + in := &Injected{s: s, fault: f} + s.faults = append(s.faults, in) + return in +} + +// Remove takes the fault out of force. +func (i *Injected) Remove() { + i.s.mu.Lock() + defer i.s.mu.Unlock() + i.s.faults = slices.DeleteFunc(i.s.faults, func(f *Injected) bool { return f == i }) +} + +// Hits is how many requests the fault has answered. +func (i *Injected) Hits() int { + i.s.mu.Lock() + defer i.s.mu.Unlock() + return i.hits +} + +// TokenRateLimited is the token endpoint answering 429, asking for +// retryAfter seconds; anything under one asks for one. +func TokenRateLimited(retryAfter int) Fault { + retryAfter = max(retryAfter, 1) + return Fault{ + Routes: []Route{RouteToken}, + Status: http.StatusTooManyRequests, + Header: http.Header{"Retry-After": {strconv.Itoa(retryAfter)}}, + } +} + +// TokenUnavailable is the token endpoint answering 503. +func TokenUnavailable() Fault { + return Fault{Routes: []Route{RouteToken}, Status: http.StatusServiceUnavailable} +} + +// TokenInvalidClient is the token endpoint refusing the client, as Basecamp +// does for an agent disconnected, or connected on another computer since. +func TokenInvalidClient() Fault { + return Fault{ + Routes: []Route{RouteToken}, + Status: http.StatusUnauthorized, + Header: http.Header{"Content-Type": {"application/json"}}, + Body: `{"error":"invalid_client"}`, + } +} + +// ServerError is routes answering 500, times times. +func ServerError(times int, routes ...Route) Fault { + return Fault{Routes: routes, Times: times, Status: http.StatusInternalServerError} +} + +// Refused is routes answering status, with no body, until removed. +func Refused(status int, routes ...Route) Fault { + return Fault{Routes: routes, Status: status} +} + +// ByAgents narrows a fault to requests an Agent makes. +func ByAgents(r Request) bool { return r.Agent } + +// Gate holds requests open. +type Gate struct { + s *Server + routes []Route + release chan struct{} + // waiting and released are guarded by the Server's lock. + waiting int + released bool +} + +// Gate holds every request to routes open, unanswered, until Release. Gates +// may overlap: a request is held until every gate on its route is +// released, whichever order they are released in. +func (s *Server) Gate(routes ...Route) *Gate { + s.mu.Lock() + defer s.mu.Unlock() + g := &Gate{s: s, routes: slices.Clone(routes), release: make(chan struct{})} + s.gates = append(s.gates, g) + return g +} + +// Waiting is how many requests the gate is holding now. +func (g *Gate) Waiting() int { + g.s.mu.Lock() + defer g.s.mu.Unlock() + return g.waiting +} + +// Release lets every held request through, and lets later ones pass. +func (g *Gate) Release() { + g.s.mu.Lock() + defer g.s.mu.Unlock() + if g.released { + return + } + g.released = true + close(g.release) + g.s.gates = slices.DeleteFunc(g.s.gates, func(o *Gate) bool { return o == g }) + g.s.notifyLocked() +} + +// Before runs fn as each request reaches route, before it is answered and +// outside the fake's lock, so fn may call the Server. The returned function +// stops it. +func (s *Server) Before(route Route, fn func(Request)) (remove func()) { + s.mu.Lock() + defer s.mu.Unlock() + h := &hook{route: route, fn: fn} + s.hooks = append(s.hooks, h) + return func() { + s.mu.Lock() + defer s.mu.Unlock() + s.hooks = slices.DeleteFunc(s.hooks, func(o *hook) bool { return o == h }) + } +} + +type options struct { + pageSize int + pingInterval time.Duration + listener net.Listener + now func() time.Time +} + +// notifyLocked wakes every Await. +func (s *Server) notifyLocked() { + close(s.changed) + s.changed = make(chan struct{}) +} + +// nextSerialLocked numbers the next thing the fake issues. +func (s *Server) nextSerialLocked() int { + s.serial++ + return s.serial +} + +// faultLocked claims the fault that answers r, if one does. +func (s *Server) faultLocked(r Request) (answer, bool) { + for idx := len(s.faults) - 1; idx >= 0; idx-- { + in := s.faults[idx] + f := in.fault + if !slices.Contains(f.Routes, r.Route) || (f.When != nil && !f.When(r)) { + continue + } + in.hits++ + if f.Times > 0 && in.hits >= f.Times { + s.faults = slices.Delete(s.faults, idx, idx+1) + } + return answer{status: f.Status, header: f.Header.Clone(), body: []byte(f.Body)}, true + } + return answer{}, false +} + +type hook struct { + route Route + fn func(Request) +} + +// answer is a response worked out under the lock and written outside it. +type answer struct { + status int + header http.Header + body []byte +} + +// call is one request on its way through a route. +type call struct { + s *Server + req *http.Request + route Route + index int + // caller is who the bearer authenticates as when the handler runs. + caller int64 +} + +func (s *Server) routes() http.Handler { + mux := http.NewServeMux() + handle := func(route Route, h func(c *call) answer) { + mux.HandleFunc(string(route), func(w http.ResponseWriter, req *http.Request) { + c, ok := s.begin(route, w, req) + if !ok { + return + } + s.mu.Lock() + c.caller, _ = s.callerLocked(req) + a := h(c) + s.mu.Unlock() + s.finish(c, w, a) + }) + } + handle(RouteASMetadata, (*call).asMetadata) + handle(RouteResourceMetadata, (*call).resourceMetadata) + handle(RouteAgentConnections, (*call).agentConnections) + handle(RouteAgentConnectionTokens, (*call).agentConnectionTokens) + handle(RouteToken, (*call).token) + handle(RouteAuthorization, (*call).authorization) + handle(RouteProfile, account((*call).profile)) + handle(RoutePerson, account((*call).person)) + handle(RouteProjects, account((*call).projects)) + handle(RouteProject, account((*call).project)) + handle(RouteProjectPeople, account((*call).projectPeople)) + handle(RouteStreamTicket, account((*call).streamTicket)) + handle(RouteEvents, account((*call).events)) + handle(RouteComment, account(recordingRead("Comment"))) + handle(RouteMessage, account(recordingRead("Message"))) + handle(RouteTodo, account(recordingRead("Todo"))) + handle(RouteCard, account(recordingRead("Kanban::Card"))) + handle(RouteCampfires, account((*call).campfires)) + handle(RouteChatLine, account((*call).chatLine)) + handle(RouteSubscription, account((*call).subscription)) + handle(RouteRecordingEvents, account((*call).recordingEvents)) + mux.HandleFunc(string(RouteCable), s.serveCable) + mux.HandleFunc("/", func(w http.ResponseWriter, req *http.Request) { + c, ok := s.begin("", w, req) + if !ok { + return + } + s.r.Errorf("fakebasecamp: no route serves %s %s", req.Method, req.URL.Path) + s.finish(c, w, answer{status: http.StatusNotFound}) + }) + return mux +} + +// begin logs a request and takes it through the hooks, the gates and the +// faults. It reports false when one of them answered it. +func (s *Server) begin(route Route, w http.ResponseWriter, req *http.Request) (*call, bool) { + var form url.Values + if req.Method == http.MethodPost && strings.HasPrefix(req.Header.Get("Content-Type"), "application/x-www-form-urlencoded") { + if err := req.ParseForm(); err == nil { + form = req.PostForm + } + } + s.mu.Lock() + caller, _ := s.callerLocked(req) + if route == RouteCable { + caller, _ = s.ticketLocked(req) + } + c := &call{s: s, req: req, route: route, index: len(s.log)} + r := Request{ + Method: req.Method, + Path: req.URL.Path, + Query: req.URL.Query(), + Form: form, + Route: route, + Caller: caller, + Agent: s.world.isAgent(caller), + } + s.log = append(s.log, r) + var hooks []func(Request) + for _, h := range s.hooks { + if h.route == route { + hooks = append(hooks, h.fn) + } + } + s.notifyLocked() + s.mu.Unlock() + + for _, fn := range hooks { + fn(r.clone()) + } + if !s.hold(req.Context(), c) { + s.finish(c, w, answer{status: http.StatusServiceUnavailable}) + return nil, false + } + s.mu.Lock() + a, faulted := s.faultLocked(r) + s.mu.Unlock() + if faulted { + s.finish(c, w, a) + return nil, false + } + return c, true +} + +// hold waits until no gate holds the request's route: at one gate after +// another, since a released gate leaves the list and the next one on the +// route is then the first. It reports false when the fake closed, or the +// client left, before every gate opened. +func (s *Server) hold(ctx context.Context, c *call) bool { + for { + s.mu.Lock() + var gate *Gate + for _, g := range s.gates { + if slices.Contains(g.routes, c.route) { + gate = g + break + } + } + if gate == nil { + s.mu.Unlock() + return true + } + gate.waiting++ + s.notifyLocked() + s.mu.Unlock() + + released := false + select { + case <-gate.release: + released = true + case <-s.done: + case <-ctx.Done(): + } + s.mu.Lock() + gate.waiting-- + s.notifyLocked() + s.mu.Unlock() + if !released { + return false + } + } +} + +// finish logs an answer's status and writes it. The status is logged first, +// so a client that has read its answer finds it logged. +func (s *Server) finish(c *call, w http.ResponseWriter, a answer) { + for name, values := range a.header { + for _, v := range values { + w.Header().Add(name, v) + } + } + if len(a.body) > 0 && w.Header().Get("Content-Type") == "" { + w.Header().Set("Content-Type", "application/json") + } + status := a.status + if status == 0 { + status = http.StatusOK + } + s.setStatus(c.index, status) + w.WriteHeader(status) + _, _ = w.Write(a.body) +} + +func (s *Server) setStatus(index, status int) { + s.mu.Lock() + defer s.mu.Unlock() + s.log[index].Status = status + s.notifyLocked() +} + +// callerLocked is who the request's bearer token authenticates as, and +// whether it carried one the fake issued. +func (s *Server) callerLocked(req *http.Request) (int64, bool) { + bearer, ok := strings.CutPrefix(req.Header.Get("Authorization"), "Bearer ") + if !ok { + return 0, false + } + t, ok := s.world.Tokens[bearer] + if !ok { + return 0, false + } + return t.PersonID, true +} + +// account guards a route mounted under an account: the account has to be +// the fake's, and the caller someone the fake issued a token to. +func account(h func(c *call) answer) func(c *call) answer { + return func(c *call) answer { + if c.req.PathValue("account") != strconv.FormatInt(c.s.world.Account.ID, 10) { + return answer{status: http.StatusNotFound} + } + if _, ok := c.s.callerLocked(c.req); !ok { + return answer{status: http.StatusUnauthorized} + } + return h(c) + } +} + +// pathID is a numeric path value; ok is false when it is not one. +func (c *call) pathID(name string) (int64, bool) { + id, err := strconv.ParseInt(strings.TrimSuffix(c.req.PathValue(name), ".json"), 10, 64) + return id, err == nil && id > 0 +} + +// jsonAnswer renders v as a JSON answer. +func (c *call) jsonAnswer(status int, v any) answer { + body, err := json.Marshal(v) + if err != nil { + c.s.r.Errorf("fakebasecamp: rendering %s: %v", c.route, err) + return answer{status: http.StatusInternalServerError} + } + return answer{status: status, body: body} +} + +// oauthError is an RFC 6749 error answer. +func oauthError(status int, code string) answer { + return answer{status: status, body: []byte(fmt.Sprintf(`{"error":%q}`, code))} +} + +// base is the fake's origin, as the client reached it. +func (c *call) base() string { return c.s.srv.URL } + +// accountURL is a URL under the account. +func (c *call) accountURL(format string, args ...any) string { + return fmt.Sprintf("%s/%d/", c.base(), c.s.world.Account.ID) + fmt.Sprintf(format, args...) +} diff --git a/internal/connector/fakebasecamp/world.go b/internal/connector/fakebasecamp/world.go new file mode 100644 index 000000000..b59c4752e --- /dev/null +++ b/internal/connector/fakebasecamp/world.go @@ -0,0 +1,246 @@ +package fakebasecamp + +import ( + "encoding/base64" + "fmt" + "strconv" + "time" +) + +// The default world's facts. The ids are the shapes Basecamp uses; none of +// them, and none of the secrets, names anything real. +const ( + AccountID int64 = 999 + AccountName = "Acme" + + OperatorID int64 = 26909558 + OperatorName = "Operator" + AgentID int64 = 52007412 + AgentName = "Marie Chef" + ProjectID int64 = 48699913 + ProjectName = "Connector" + + // AgentClientID and AgentSecret are the agent's OAuth client as the + // default world starts. The connection ceremony rotates the secret, as + // Basecamp does, so read the current one from the World after it. + AgentClientID = "agent-client" + AgentSecret = "not-a-real-secret" //nolint:gosec // G101: the fake's placeholder, not a credential + + // OperatorToken is a bearer token for the operator, as a completed + // person login would have stored. + OperatorToken = "operator-token" +) + +// Scopes a credential can carry. +const ( + ScopeFull = "full" + ScopeRead = "read" +) + +// World is everything the fake knows. The maps are keyed by the id of the +// value they hold. A World passed to Start, and every pointer in it, belongs +// to the Server from then on: change it only inside Update. +type World struct { + Account Account + + // People is everyone in the account, agents included. + People map[int64]*Person + // Projects are the account's projects. + Projects map[int64]*Project + // Recordings are the recordings in the projects: comments, messages, + // to-dos, cards, Campfires and their lines. + Recordings map[int64]*Recording + + // Agents are the OAuth clients that mint client_credentials tokens, + // keyed by client id. + Agents map[string]*AgentClient + // Tokens are the bearer tokens the fake accepts, keyed by the token. + // Minting adds to it; a test adds a person's own login here. + Tokens map[string]*Token + + // Connection is what the operator approves on the agent connection + // page, for every ceremony that reaches it. + Connection Connection +} + +// Account is the one Basecamp account the fake serves. +type Account struct { + ID int64 + Name string +} + +// Person is someone in the account. +type Person struct { + ID int64 + Name string + // Agent makes the person an Agent principal; otherwise a User. + Agent bool + // Client marks a client of the account. + Client bool + // BossID is who the person works for, as their own profile names them: + // a personal agent's owner. Zero names nobody. + BossID int64 + // IdentityID is the Launchpad identity authorization.json answers for + // the person's token. Zero answers 401, as it does for an agent. + IdentityID int64 + // ReadableBy, when not nil, is everyone whose read of this person + // succeeds; anyone else is refused with 403. + ReadableBy []int64 +} + +// Project is a project, and who is on it. +type Project struct { + ID int64 + Name string + Members []int64 +} + +// Recording is one recording in a project. +type Recording struct { + ID int64 + // Type is Basecamp's recording type: Comment, Message, Todo, + // Kanban::Card, Chat::Transcript (a Campfire), or a Chat::Lines:: line. + Type string + BucketID int64 + // ParentID is the recording this one belongs to: a comment's parent, + // a line's Campfire. + ParentID int64 + CreatorID int64 + Title string + // Content is the rich text mentions live in: a to-do's description, a + // card's content, a comment's or a line's body. + Content string + AssigneeIDs []int64 + SubscriberIDs []int64 + Completed bool + // Status is "active" when empty. + Status string + // Events is the recording's own history, oldest first. Emit appends an + // event about the recording to it. + Events []RecordingEvent +} + +// RecordingEvent is one entry in a recording's history. +type RecordingEvent struct { + ID int64 + Action string + CreatorID int64 + PerformedByID int64 + // CreatedAt is the fake's epoch when zero. + CreatedAt time.Time + // AddedPersonIDs is an assignment change's delta; nil carries no + // details at all. + AddedPersonIDs []int64 +} + +// AgentClient is an agent's confidential OAuth client. +type AgentClient struct { + ID string + Secret string + // PersonID is the Agent its tokens act as. + PersonID int64 + // Scope is what the client was approved for. + Scope string +} + +// Token is what a bearer token authenticates as. +type Token struct { + PersonID int64 + Scope string +} + +// Connection is the operator's approval on the agent connection page. +type Connection struct { + // ClientID is the agent the ceremony hands over. + ClientID string + // Scope is the access the operator approves; empty approves full. + Scope string +} + +// DefaultWorld is an account with an operator, an agent connected to its +// OAuth client, and one project they are both on. +func DefaultWorld() *World { + return &World{ + Account: Account{ID: AccountID, Name: AccountName}, + People: map[int64]*Person{ + OperatorID: {ID: OperatorID, Name: OperatorName}, + AgentID: {ID: AgentID, Name: AgentName, Agent: true}, + }, + Projects: map[int64]*Project{ + ProjectID: {ID: ProjectID, Name: ProjectName, Members: []int64{OperatorID, AgentID}}, + }, + Recordings: map[int64]*Recording{}, + Agents: map[string]*AgentClient{ + AgentClientID: {ID: AgentClientID, Secret: AgentSecret, PersonID: AgentID, Scope: ScopeFull}, + }, + Tokens: map[string]*Token{ + OperatorToken: {PersonID: OperatorID, Scope: ScopeFull}, + }, + Connection: Connection{ClientID: AgentClientID}, + } +} + +// Mention is the rich-text attachment that mentions a person, as Basecamp +// renders one: the person is named only inside the attachment's sgid. +func Mention(personID int64) string { + return `