From 7fa5f93a0ef7f911cb7feb935a10db8ea7b368cc Mon Sep 17 00:00:00 2001 From: Rob Zolkos Date: Fri, 2 Oct 2026 13:15:53 -0400 Subject: [PATCH 1/3] Report when the connector's backlog warns, pauses and recovers The queue between intake and admission decides four edges: the backlog crossing its warning depth, an offer waiting for room (which stops the feed read), that wait ending, and the backlog falling back below the warning depth. It handed them only to optional callbacks, and runConnect sets none, so in production every edge went nowhere. The package says the pause is reported rather than hidden; it was hidden. The queue now reports each edge on its logger before running any callback, with the depth and the threshold crossed, and says plainly when reading the feed is paused and why. Intake already adopts the connector's logger for the queue, so every caller that builds intake gets the report and none can forget to wire it. A callback that panics is contained after the report, so it cannot silence it. --- internal/connector/intake.go | 6 +- internal/connector/queue.go | 66 +++++++++++++++---- internal/connector/queue_test.go | 16 +++++ internal/connector/seam_contract_test.go | 81 ++++++++++++++++++++++++ 4 files changed, 155 insertions(+), 14 deletions(-) diff --git a/internal/connector/intake.go b/internal/connector/intake.go index d881c63fe..f89cbe27f 100644 --- a/internal/connector/intake.go +++ b/internal/connector/intake.go @@ -285,8 +285,10 @@ func New(opts Options) (*Intake, error) { if opts.Logger == nil { opts.Logger = slog.New(slog.DiscardHandler) } - // The queue reports a callback that panicked, and a queue built by a - // caller who did not think about that would report it nowhere. It is + // The queue reports its backlog edges - the warning, the pause of the + // feed read, the resume and the recovery - and a callback that panicked, + // and a queue built by a caller who did not think about that would report + // them nowhere. It is // adopted under the queue's own lock, because the queue may already be // in use — admission takes from it — and a logger the caller did set is // kept. diff --git a/internal/connector/queue.go b/internal/connector/queue.go index 37132d094..0b7d0056c 100644 --- a/internal/connector/queue.go +++ b/internal/connector/queue.go @@ -61,21 +61,24 @@ type Queue struct { paused bool // OnWarn fires when the depth first crosses the warning threshold, and - // OnRecover when it falls back below. Both are optional. + // OnRecover when it falls back below. Both are optional, and neither is + // needed for the operator to hear of it: every edge is reported on the + // logger first. OnWarn func(depth int) OnRecover func(depth int) // OnPause fires when an offer begins waiting for room, and OnResume when // it stops. The feed is not being consumed in between. OnPause func(depth int) OnResume func(depth int) - // log reports a callback that panicked. Held under edges: a queue can be - // adopted by intake while another component is already using it, and a - // field written by one and read by the other is a race. + // log reports every edge and any callback that panicked. Held under + // edges: a queue can be adopted by intake while another component is + // already using it, and a field written by one and read by the other is a + // race. log *slog.Logger } -// SetLogger sets where a callback that panicked is reported. Safe to call -// while the queue is in use. +// SetLogger sets where the backlog's edges, and a callback that panicked, are +// reported. Safe to call while the queue is in use. func (q *Queue) SetLogger(log *slog.Logger) { q.edges.Lock() defer q.edges.Unlock() @@ -227,10 +230,10 @@ func (q *Queue) stage(delta int) { switch { case depth >= q.warnAt && !q.warned: q.warned = true - q.pending = append(q.pending, queueEdge{fire: q.OnWarn, depth: depth}) + q.pending = append(q.pending, queueEdge{kind: edgeWarn, fire: q.OnWarn, depth: depth}) case depth < q.warnAt && q.warned: q.warned = false - q.pending = append(q.pending, queueEdge{fire: q.OnRecover, depth: depth}) + q.pending = append(q.pending, queueEdge{kind: edgeRecover, fire: q.OnRecover, depth: depth}) } } @@ -250,10 +253,10 @@ func (q *Queue) stageWait(delta int) { switch { case q.waiting > 0 && !q.paused: q.paused = true - q.pending = append(q.pending, queueEdge{fire: q.OnPause, depth: q.depth}) + q.pending = append(q.pending, queueEdge{kind: edgePause, fire: q.OnPause, depth: q.depth}) case q.waiting == 0 && q.paused: q.paused = false - q.pending = append(q.pending, queueEdge{fire: q.OnResume, depth: q.depth}) + q.pending = append(q.pending, queueEdge{kind: edgeResume, fire: q.OnResume, depth: q.depth}) } } @@ -290,7 +293,12 @@ func (q *Queue) deliver() { } } -// fire runs one callback with the lock released and a panic contained. +// fire reports one edge and runs its callback, with the lock released and a +// panic contained. +// +// The report comes first and does not depend on the callback: a caller that +// set none, as production does, still has the backlog reported rather than +// hidden, and one that panics cannot take the report down with it. // // The callbacks belong to whoever built the queue, and the goroutine they run // on is the feed's delivery path or a worker taking work off it. A callback @@ -299,9 +307,11 @@ func (q *Queue) deliver() { // callback is told about was committed before it ran, so what it does cannot // change it. func (q *Queue) fire(edge queueEdge) { + log := q.logger() + q.report(log, edge) defer func() { if p := recover(); p != nil { - q.logger().Error("a backlog callback panicked; the queue carried on without it", + log.Error("a backlog callback panicked; the queue carried on without it", "panic", richtext.SanitizeSingleLine(fmt.Sprint(p))) } }() @@ -310,6 +320,27 @@ func (q *Queue) fire(edge queueEdge) { } } +// report tells the operator about one edge, with the depth it was decided at +// and the threshold it crossed. The depth of a pause or a resume counts the id +// the waiting offer is holding: it is backlog either way. +func (q *Queue) report(log *slog.Logger, edge queueEdge) { + pauseAt := cap(q.ids) + switch edge.kind { + case edgeWarn: + log.Warn("the backlog reached its warning depth; admission is falling behind the feed", + "depth", edge.depth, "warn_at", q.warnAt, "pause_at", pauseAt) + case edgeRecover: + log.Info("the backlog fell back below its warning depth", + "depth", edge.depth, "warn_at", q.warnAt) + case edgePause: + log.Warn("the backlog reached its pause depth; reading the feed is paused until admission makes room, so the backlog cannot outgrow memory", + "depth", edge.depth, "pause_at", pauseAt) + case edgeResume: + log.Info("the backlog has room again; reading the feed resumed", + "depth", edge.depth, "pause_at", pauseAt) + } +} + // logger is read under edges, so it must never be called holding them. Its one // caller, fire, runs with the lock released. func (q *Queue) logger() *slog.Logger { @@ -322,6 +353,17 @@ func (q *Queue) logger() *slog.Logger { } type queueEdge struct { + kind edgeKind fire func(int) depth int } + +// edgeKind is which of the backlog's transitions an edge is. +type edgeKind int + +const ( + edgeWarn edgeKind = iota + edgeRecover + edgePause + edgeResume +) diff --git a/internal/connector/queue_test.go b/internal/connector/queue_test.go index 8a764b063..d570b474c 100644 --- a/internal/connector/queue_test.go +++ b/internal/connector/queue_test.go @@ -299,3 +299,19 @@ func TestAPanickingPauseCallbackLeavesNoPhantomBacklog(t *testing.T) { assert.Equal(t, int32(2), resumes.Load()) assert.Equal(t, 1, queue.Depth()) } + +// The report is the queue's, not the callback's: a callback that panics is +// contained after the edge has already been reported. +func TestAPanickingCallbackDoesNotSilenceTheReport(t *testing.T) { + queue, err := NewQueue(1, 4) + require.NoError(t, err) + var logs bytes.Buffer + queue.SetLogger(slog.New(slog.NewTextHandler(&logs, nil))) + queue.OnWarn = func(int) { panic("a warning callback panics") } + + require.NoError(t, queue.Offer(context.Background(), 1)) + + assert.Contains(t, logs.String(), "the backlog reached its warning depth") + assert.Contains(t, logs.String(), "depth=1 warn_at=1 pause_at=4") + assert.Contains(t, logs.String(), "a backlog callback panicked") +} diff --git a/internal/connector/seam_contract_test.go b/internal/connector/seam_contract_test.go index a41d9ebf0..127e46c1c 100644 --- a/internal/connector/seam_contract_test.go +++ b/internal/connector/seam_contract_test.go @@ -3,6 +3,7 @@ package connector import ( "bytes" "context" + "encoding/json" "errors" "log/slog" "net/http" @@ -11,6 +12,7 @@ import ( "sync" "sync/atomic" "testing" + "time" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -202,6 +204,85 @@ func TestIntakeGivesTheQueueItsLogger(t *testing.T) { assert.Contains(t, logs.String(), "a backlog callback panicked") } +// Production builds its queue with no callbacks and hands it to intake with +// the connector's logger, so the logger is the only place an operator can +// learn that the backlog is growing or that the feed read has stopped. Each +// edge is reported there, with the depth and the threshold it crossed. +func TestIntakeReportsTheBacklogsEdgesOnItsLogger(t *testing.T) { + var logs safeBuffer + queue, err := NewQueue(2, 3) + require.NoError(t, err) + _, err = New(Options{ + Origin: "https://3.basecampapi.com", + AccountID: "2914079", + ConsumerNamespace: "connector-test", + Ledger: newTestLedger(t), + Queue: queue, + Minter: stubMinter{}, + Polls: &scriptedPolls{}, + Logger: slog.New(slog.NewJSONHandler(&logs, nil)), + }) + require.NoError(t, err) + + ctx := context.Background() + require.NoError(t, queue.Offer(ctx, 1)) + require.NoError(t, queue.Offer(ctx, 2)) + require.NoError(t, queue.Offer(ctx, 3)) + waiting := make(chan error, 1) + go func() { waiting <- queue.Offer(ctx, 4) }() + require.Eventually(t, func() bool { return len(backlogLines(t, logs.String())) == 2 }, time.Second, time.Millisecond) + // One take makes room, and the waiting offer is let finish before the + // rest, so the resume is decided before the recovery. + _, err = queue.Take(ctx) + require.NoError(t, err) + require.NoError(t, <-waiting) + for range 2 { + _, err := queue.Take(ctx) + require.NoError(t, err) + } + + lines := backlogLines(t, logs.String()) + require.Len(t, lines, 4) + + assert.Equal(t, "WARN", lines[0]["level"]) + assert.Contains(t, lines[0]["msg"], "warning depth") + assert.EqualValues(t, 2, lines[0]["depth"]) + assert.EqualValues(t, 2, lines[0]["warn_at"]) + assert.EqualValues(t, 3, lines[0]["pause_at"]) + + assert.Equal(t, "WARN", lines[1]["level"]) + assert.Contains(t, lines[1]["msg"], "reading the feed is paused") + assert.EqualValues(t, 3, lines[1]["pause_at"]) + assert.Contains(t, lines[1], "depth") + + assert.Equal(t, "INFO", lines[2]["level"]) + assert.Contains(t, lines[2]["msg"], "reading the feed resumed") + assert.EqualValues(t, 3, lines[2]["pause_at"]) + assert.Contains(t, lines[2], "depth") + + assert.Equal(t, "INFO", lines[3]["level"]) + assert.Contains(t, lines[3]["msg"], "below its warning depth") + assert.EqualValues(t, 1, lines[3]["depth"]) + assert.EqualValues(t, 2, lines[3]["warn_at"]) +} + +// backlogLines decodes the JSON log lines that report a backlog edge. +func backlogLines(t *testing.T, logs string) []map[string]any { + t.Helper() + var lines []map[string]any + for line := range strings.SplitSeq(strings.TrimSpace(logs), "\n") { + if line == "" { + continue + } + var record map[string]any + require.NoError(t, json.Unmarshal([]byte(line), &record)) + if msg, _ := record["msg"].(string); strings.HasPrefix(msg, "the backlog ") { + lines = append(lines, record) + } + } + return lines +} + // A queue can already be in use when intake adopts it — admission takes from // the same queue — so adopting its logger must not race the callback path that // reads it. From 47d89857928c0fd1c00d8a6efeacf05d7f8111d1 Mon Sep 17 00:00:00 2001 From: Rob Zolkos Date: Fri, 2 Oct 2026 14:29:12 -0400 Subject: [PATCH 2/3] Stop the connector when Basecamp no longer takes its agent's credential A running connector learned that its agent was disconnected in Basecamp, or connected on another computer, only from the event feed. The feed asks Basecamp for something only when it connects, and while its live connection stays up it asks for nothing. Admission and intake ask all the time, but to them a refusal was one failed read among others: admission blocked the mention it was deciding as read_failed, intake kept the projects it last read, and the connector ran on. Every mention addressed to the agent was blocked, and nothing said why, for as long as the socket held. Reproduced end to end against the fake Basecamp: disconnect the agent, or connect it on another computer and let its token come due, while the connector streams; the operator's next mention is blocked read_failed and the connector does not exit. A credential watch now sees the reads admission and intake make. A token renewal the token endpoint refuses is Basecamp's answer already, and stops the run at once. A read answered 401 is only a reason to ask, since something other than Basecamp can answer 401: the watch asks Basecamp itself, as the feed's own stop does, and stops the run only when it confirms. Either way the connector exits 3 with the words the feed's stop uses, and status records it as disconnected. --- internal/commands/connect_credential.go | 114 ++++++++++++++++ internal/commands/connect_credential_test.go | 134 +++++++++++++++++++ internal/commands/connect_run.go | 24 +++- 3 files changed, 268 insertions(+), 4 deletions(-) create mode 100644 internal/commands/connect_credential.go create mode 100644 internal/commands/connect_credential_test.go diff --git a/internal/commands/connect_credential.go b/internal/commands/connect_credential.go new file mode 100644 index 000000000..8a31ce44c --- /dev/null +++ b/internal/commands/connect_credential.go @@ -0,0 +1,114 @@ +package commands + +import ( + "context" + "errors" + "net/http" + + "github.com/basecamp/basecamp-sdk/go/pkg/basecamp" + + "github.com/basecamp/basecamp-cli/internal/auth" +) + +// errAgentCredentialNotTaken is the credential watch's stop when Basecamp, +// asked, confirms that it no longer takes the agent's credential. +var errAgentCredentialNotTaken = errors.New("the agent's credential is no longer taken by Basecamp") + +// credentialWatch stops a running connector once Basecamp no longer takes +// its agent's credential, whichever of its reads meets that first. +// +// The feed stops on a refused credential itself, but only when it asks +// Basecamp for something, and while its live connection stays up it asks +// for nothing. Admission and intake ask all the time, and to them a refusal +// is one failed read among others: admission blocks the record it was +// deciding, intake keeps the projects it last read, and both carry on. A +// connector whose agent was disconnected in Basecamp, or connected on +// another computer, would then run on for as long as its socket held, +// blocking every mention addressed to it. The watch sees what their reads +// meet, and ends the run instead, with the disconnect said in the words +// the feed's own stop uses. +// +// It hears two things. A token renewal the token endpoint refuses is +// Basecamp's answer already, and stops the run at once. A read answered 401 +// is only a reason to ask: something other than Basecamp can answer 401, so +// the watch asks Basecamp itself, and stops the run only when it confirms. +type credentialWatch struct { + // refused carries the first refused renewal. + refused chan error + // suspect is signaled by a read answered 401; signals that arrive + // while one is pending coalesce into it. + suspect chan struct{} +} + +func newCredentialWatch() *credentialWatch { + return &credentialWatch{refused: make(chan error, 1), suspect: make(chan struct{}, 1)} +} + +// Tokens is p, watched for a renewal the token endpoint refuses. +func (w *credentialWatch) Tokens(p basecamp.TokenProvider) basecamp.TokenProvider { + return watchedTokens{provider: p, watch: w} +} + +// WrapTransport implements basecamp.TransportWrapper: inner, watched for a +// response answered 401. +func (w *credentialWatch) WrapTransport(inner http.RoundTripper) http.RoundTripper { + return watchedTransport{inner: inner, watch: w} +} + +// Run waits until the credential is refused, and returns why, or until ctx +// ends, and returns nil. confirm asks Basecamp whether it still takes the +// credential, and reports true when it does not. +func (w *credentialWatch) Run(ctx context.Context, confirm func(context.Context) bool) error { + for { + select { + case <-ctx.Done(): + return nil + case err := <-w.refused: + return err + case <-w.suspect: + if confirm(ctx) { + return errAgentCredentialNotTaken + } + } + } +} + +func (w *credentialWatch) refuse(err error) { + select { + case w.refused <- err: + default: + } +} + +func (w *credentialWatch) unauthorized() { + select { + case w.suspect <- struct{}{}: + default: + } +} + +type watchedTokens struct { + provider basecamp.TokenProvider + watch *credentialWatch +} + +func (t watchedTokens) AccessToken(ctx context.Context) (string, error) { + token, err := t.provider.AccessToken(ctx) + if errors.Is(err, auth.ErrAgentCredentialRefused) { + t.watch.refuse(err) + } + return token, err +} + +type watchedTransport struct { + inner http.RoundTripper + watch *credentialWatch +} + +func (t watchedTransport) RoundTrip(req *http.Request) (*http.Response, error) { + resp, err := t.inner.RoundTrip(req) + if err == nil && resp.StatusCode == http.StatusUnauthorized { + t.watch.unauthorized() + } + return resp, err +} diff --git a/internal/commands/connect_credential_test.go b/internal/commands/connect_credential_test.go new file mode 100644 index 000000000..89b46628e --- /dev/null +++ b/internal/commands/connect_credential_test.go @@ -0,0 +1,134 @@ +package commands + +import ( + "context" + "net/http" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/basecamp/basecamp-cli/internal/auth" + "github.com/basecamp/basecamp-cli/internal/connector" + "github.com/basecamp/basecamp-cli/internal/output" +) + +// A token renewal Basecamp refuses, met by any read the watch sees, stops +// the run at once, and the run says it as the disconnect it is without +// asking Basecamp again. +func TestTheCredentialWatchStopsOnARefusedRenewal(t *testing.T) { + w := newCredentialWatch() + refused := output.ErrAuth("Minting an agent token was refused (invalid_client)") + refused.Cause = auth.ErrAgentCredentialRefused + tokens := w.Tokens(tokenFunc(func() (string, error) { return "", refused })) + + _, err := tokens.AccessToken(t.Context()) + require.ErrorIs(t, err, auth.ErrAgentCredentialRefused, "the read still sees its own failure") + stop := runWatch(t, w, mustNotConfirm(t)) + require.ErrorIs(t, stop, auth.ErrAgentCredentialRefused) + + state, _, exit := connectorStoppedBy(stop, "Ryan Singer (agent)", "agent", mustNotAsk(t)) + var e *output.Error + require.ErrorAs(t, exit, &e) + assert.Equal(t, "Ryan Singer (agent) was disconnected in Basecamp, or connected on another computer", e.Message) + assert.Equal(t, connector.ConnectionDisconnected, state) +} + +// A read answered 401 asks Basecamp, and stops the run only when Basecamp +// confirms it no longer takes the credential. +func TestTheCredentialWatchAsksAboutAnUnauthorizedRead(t *testing.T) { + w := newCredentialWatch() + transport := w.WrapTransport(statusTransport(http.StatusUnauthorized)) + roundTrip(t, transport) + + asked := 0 + stop := runWatch(t, w, func(context.Context) bool { asked++; return true }) + require.ErrorIs(t, stop, errAgentCredentialNotTaken) + assert.Equal(t, 1, asked) + + state, _, exit := connectorStoppedBy(stop, "Ryan Singer (agent)", "agent", mustNotAsk(t)) + var e *output.Error + require.ErrorAs(t, exit, &e) + assert.Equal(t, output.CodeAuth, e.Code) + assert.Equal(t, connector.ConnectionDisconnected, state) +} + +// A 401 Basecamp does not confirm, from a proxy say, leaves the run going, +// and the next one asks again. +func TestTheCredentialWatchRunsOnWhenBasecampStillTakesTheCredential(t *testing.T) { + w := newCredentialWatch() + transport := w.WrapTransport(statusTransport(http.StatusUnauthorized)) + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + asks := make(chan struct{}) + done := make(chan error, 1) + go func() { + done <- w.Run(ctx, func(context.Context) bool { asks <- struct{}{}; return false }) + }() + + for range 2 { + roundTrip(t, transport) + select { + case <-asks: + case <-time.After(10 * time.Second): + t.Fatal("the watch did not ask about a 401") + } + } + cancel() + require.NoError(t, <-done, "ended by its context, the watch stops nothing") +} + +// Every other failure is the reads' own: neither a renewal that failed for +// another reason nor another status asks Basecamp or stops the run. +func TestTheCredentialWatchIgnoresOtherFailures(t *testing.T) { + w := newCredentialWatch() + tokens := w.Tokens(tokenFunc(func() (string, error) { return "", output.ErrRateLimit(1) })) + _, err := tokens.AccessToken(t.Context()) + require.Error(t, err) + for _, status := range []int{http.StatusForbidden, http.StatusNotFound, http.StatusTooManyRequests, http.StatusInternalServerError} { + roundTrip(t, w.WrapTransport(statusTransport(status))) + } + + ctx, cancel := context.WithCancel(t.Context()) + cancel() + assert.NoError(t, w.Run(ctx, mustNotConfirm(t))) + assert.Empty(t, w.refused) + assert.Empty(t, w.suspect) +} + +type tokenFunc func() (string, error) + +func (f tokenFunc) AccessToken(context.Context) (string, error) { return f() } + +type statusTransport int + +func (s statusTransport) RoundTrip(req *http.Request) (*http.Response, error) { + return &http.Response{StatusCode: int(s), Body: http.NoBody, Request: req}, nil +} + +func roundTrip(t *testing.T, rt http.RoundTripper) { + t.Helper() + req, err := http.NewRequestWithContext(t.Context(), http.MethodGet, "http://basecamp.test/999/my/profile.json", nil) + require.NoError(t, err) + resp, err := rt.RoundTrip(req) + require.NoError(t, err) + require.NoError(t, resp.Body.Close()) +} + +// runWatch runs w until it stops on its own, failing the test if it does +// not within a deadline that only bounds a failure. +func runWatch(t *testing.T, w *credentialWatch, confirm func(context.Context) bool) error { + t.Helper() + ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second) + defer cancel() + err := w.Run(ctx, confirm) + if ctx.Err() != nil { + t.Fatal("the watch did not stop") + } + return err +} + +func mustNotConfirm(t *testing.T) func(context.Context) bool { + return func(context.Context) bool { t.Error("asked Basecamp to confirm what it had already said"); return true } +} diff --git a/internal/commands/connect_run.go b/internal/commands/connect_run.go index 5c59e7034..b0bd3b4d3 100644 --- a/internal/commands/connect_run.go +++ b/internal/commands/connect_run.go @@ -249,6 +249,14 @@ func runConnect(cmd *cobra.Command, f *connectRunFlags) error { } lines := ndjson.NewWriter(cmd.OutOrStdout()) + // The reads admission and intake make are watched for Basecamp refusing + // the agent's credential, which the feed alone would not see while its + // live connection stays up. The start's own reads are not: a refusal + // there ends the start as it is. + watch := newCredentialWatch() + watchedOpts := append(connectSDKOptions(), basecamp.WithTransportWrapper(watch)) + watchedTokens := watch.Tokens(tokens) + queue, err := connector.NewQueue(connector.DefaultBacklogWarn, connector.DefaultBacklogPause) if err != nil { return err @@ -266,7 +274,9 @@ func runConnect(cmd *cobra.Command, f *connectRunFlags) error { intakeOpts.Queue = queue intakeOpts.Lines = lines intakeOpts.Logger = logger - intakeOpts.Membership = connector.SDKMembership{Client: accountClient} + intakeOpts.Membership = connector.SDKMembership{ + Client: basecamp.NewClient(&basecamp.Config{BaseURL: app.Config.BaseURL}, watchedTokens, watchedOpts...).ForAccount(account), + } intake, err := connector.New(intakeOpts) if err != nil { return err @@ -278,7 +288,7 @@ func runConnect(cmd *cobra.Command, f *connectRunFlags) error { // is at most connectServedTTL old — see TestConnectServedTTLStaysSmall. served := newConnectServed(path, file, logger) - reads := admission.NewSDKReads(&basecamp.Config{BaseURL: app.Config.BaseURL}, tokens, account, connectSDKOptions()...) + reads := admission.NewSDKReads(&basecamp.Config{BaseURL: app.Config.BaseURL}, watchedTokens, account, watchedOpts...) admitter, err := admission.NewAdmitter(policy, reads, admission.WithServed(served.Current)) if err != nil { return output.ErrUsage(err.Error()) @@ -347,6 +357,9 @@ func runConnect(cmd *cobra.Command, f *connectRunFlags) error { if err := ledger.NoteConnection(ctx, connector.ConnectionRunning, ""); err != nil { logger.Warn("connector: could not record that it runs, for status", "error", err) } + runPart("credential", func(ctx context.Context) error { + return watch.Run(ctx, func(ctx context.Context) bool { return confirmAgentRefused(ctx, app, kind, name) }) + }) runPart("intake", intake.Run) runPart("admission", func(ctx context.Context) error { return connector.RunAdmission(ctx, connector.AdmissionOptions{Ledger: ledger, Queue: queue, Admitter: admitter, Lines: lines, Logger: logger}) @@ -390,9 +403,12 @@ func runConnect(cmd *cobra.Command, f *connectRunFlags) error { // another computer — is said in their words; anything else is returned as it // is. The feed's authorization_failed counts forbidden answers too, so it is // called a disconnect only when refused, asking Basecamp, confirms it; a -// refused token renewal is already Basecamp's answer. +// refused token renewal is already Basecamp's answer, and so is a refusal +// the credential watch has confirmed. func connectorStoppedBy(err error, agent, profile string, refused func() bool) (state, detail string, exit error) { - disconnected := errors.Is(err, auth.ErrAgentCredentialRefused) || (feedAuthorizationFailed(err) && refused()) + disconnected := errors.Is(err, auth.ErrAgentCredentialRefused) || + errors.Is(err, errAgentCredentialNotTaken) || + (feedAuthorizationFailed(err) && refused()) if !disconnected { return connector.ConnectionStopped, "", err } From 93b287ac0b9bb3c98664ee4adc655f0ec8eca7e4 Mon Sep 17 00:00:00 2001 From: Rob Zolkos Date: Fri, 2 Oct 2026 14:29:12 -0400 Subject: [PATCH 3/3] Test how basecamp connect fails and recovers, end to end Every fault is one Basecamp can produce, met the way the connector would meet it, with nothing in the CLI told a test is running: - SIGINT on a running connector logs that it is shutting down and exits 130. - A second connector for the same agent, from another directory sharing the state directory, exits 5 at once, names the process holding the agent, and asks Basecamp for nothing. - Two agents connected on one computer run side by side, and each hands off only its own mention. - An agent disconnected while its connector runs stops it with exit 3, whether the next read or the feed's reconnect meets it first. An agent connected on another computer stops it at its next token renewal. - A token endpoint rate-limited at start is waited out, said, and the connector comes up and hands off a mention. - An event committed while the cable is down arrives by the reconnect's catch-up, once. - A cable that stops pinging is called stale and replaced. - A recording read that fails twice then serves is admitted in one decision. - A mention admitted under --hold by a connector then killed is handed off once after `connect release` and a restart. - A backlog past its warning depth, behind held reads, is reported, and its recovery too. The fake gains Disconnect (the agent's tokens revoked, its secret cleared) and AgentClient.TokenLifetime, so a renewal comes due within a test; the harness gains a profile parameter for a second agent and AnotherComputer for a second home against the same Basecamp. --- e2e/connect/connector_test.go | 12 +- e2e/connect/failures_test.go | 404 ++++++++++++++++++ e2e/connect/harness_test.go | 39 +- .../connector/fakebasecamp/connect_test.go | 54 +++ internal/connector/fakebasecamp/doc.go | 8 +- internal/connector/fakebasecamp/oauth.go | 42 +- internal/connector/fakebasecamp/world.go | 8 + 7 files changed, 553 insertions(+), 14 deletions(-) create mode 100644 e2e/connect/failures_test.go diff --git a/e2e/connect/connector_test.go b/e2e/connect/connector_test.go index 085385fa0..0ac9246c3 100644 --- a/e2e/connect/connector_test.go +++ b/e2e/connect/connector_test.go @@ -65,7 +65,14 @@ type Output struct { // harness's home. func (h *Harness) Connect(t *testing.T, extra ...string) *Connector { t.Helper() - return h.Start(t, h.Home, append([]string{"connect", "-P", Agent}, extra...)...) + return h.ConnectProfile(t, DefaultAgent, extra...) +} + +// ConnectProfile starts `basecamp connect` for p's profile with extra +// flags, in the harness's home. +func (h *Harness) ConnectProfile(t *testing.T, p Profile, extra ...string) *Connector { + t.Helper() + return h.Start(t, h.Home, append([]string{"connect", "-P", p.Name}, extra...)...) } // Start starts a CLI command in the background, in dir, with the harness's @@ -209,6 +216,9 @@ func (c *Connector) WaitStreaming(t *testing.T, nth int) { c.WaitLogged(t, MsgStreaming, nth) } +// Pid is the process's id. +func (c *Connector) Pid() int { return c.cmd.Process.Pid } + // Signal sends sig to the process as it stands. Interrupt is the safe way // to send SIGINT. func (c *Connector) Signal(t *testing.T, sig os.Signal) { diff --git a/e2e/connect/failures_test.go b/e2e/connect/failures_test.go new file mode 100644 index 000000000..d5678f23c --- /dev/null +++ b/e2e/connect/failures_test.go @@ -0,0 +1,404 @@ +//go:build linux || darwin + +package connect + +import ( + "context" + "encoding/json" + "net/http" + "strconv" + "strings" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/basecamp/basecamp-cli/internal/connector" + "github.com/basecamp/basecamp-cli/internal/connector/admission" + "github.com/basecamp/basecamp-cli/internal/connector/fakebasecamp" +) + +// How the connector fails, and how it recovers, end to end. Every fault is +// one Basecamp can really produce, put to the connector the way it would +// meet it: nothing in the CLI is told a test is running. + +// Exit codes the connector ends with, as output.ExitCodeFor maps its errors. +const ( + exitAuth = 3 + exitLock = 5 +) + +// MsgShuttingDown is logged when a signal starts the connector's shutdown. +const MsgShuttingDown = "connector: shutting down" + +// Ctrl-C on a running connector: it says it is shutting down, and exits +// 130, the shell's code for an interrupt, having written nothing it should +// not have. +func TestSIGINTShutsTheConnectorDownCleanly(t *testing.T) { + t.Parallel() + _, c := streaming(t) + out := c.Stop(t) + + shutdown := out.Logged(MsgShuttingDown) + require.Len(t, shutdown, 1, out.Transcript()) + assert.Equal(t, "interrupt", shutdown[0].Attrs["signal"]) + for _, l := range out.Logs { + assert.NotEqual(t, "ERROR", l.Level, "%q", l.Raw) + } +} + +// A second connector for an agent already running, started from another +// repository on the same computer, refuses at once: it says which process +// holds the agent, and asks Basecamp for nothing, not even a token. +func TestASecondConnectorForTheSameAgentRefusesAtOnce(t *testing.T) { + t.Parallel() + h, first := streaming(t) + before := len(h.Fake.Requests()) + + second := h.Start(t, t.TempDir(), "connect", "-P", Agent).Wait(t) + require.Equal(t, exitLock, second.ExitCode, second.Transcript()) + assert.Empty(t, second.Lines, "nothing on stdout") + refusal := strings.Join(rawLogs(second), "\n") + assert.Contains(t, refusal, strconv.Itoa(first.Pid()), "the refusal names the process that holds the agent") + + for _, r := range h.Fake.Requests()[before:] { + t.Errorf("the refused connector reached the fake: %s %s", r.Method, r.Path) + } + first.Stop(t) +} + +// SecondAgentID is a second agent on the default project, connected through +// a client of its own. +const ( + SecondAgentID int64 = 52007413 + SecondAgentName = "Second Chef" + SecondAgentClientID = "second-agent-client" +) + +// SecondAgent is the second agent, as a profile of its own. +var SecondAgent = Profile{Name: "second", ClientID: SecondAgentClientID} + +// Two agents connected on one computer are two connectors, and each hands +// off only what was asked of it. +func TestTwoAgentsRunSideBySideAndEachGetsItsOwnMentions(t *testing.T) { + t.Parallel() + w := world() + w.People[SecondAgentID] = &fakebasecamp.Person{ID: SecondAgentID, Name: SecondAgentName, Agent: true} + w.Agents[SecondAgentClientID] = &fakebasecamp.AgentClient{ + ID: SecondAgentClientID, Secret: "second-secret", PersonID: SecondAgentID, Scope: fakebasecamp.ScopeFull, + } + p := w.Projects[fakebasecamp.ProjectID] + p.Members = append(p.Members, SecondAgentID) + h := NewHarness(t, w) + h.Setup(t) + h.SetupProfile(t, SecondAgent) + + first := h.Connect(t) + second := h.ConnectProfile(t, SecondAgent) + first.WaitStreaming(t, 1) + second.WaitStreaming(t, 1) + + toFirst := comment(h, fakebasecamp.OperatorID, mentioning(fakebasecamp.AgentID), fakebasecamp.Live) + toSecond := comment(h, fakebasecamp.OperatorID, mentioning(SecondAgentID), fakebasecamp.Live) + // Each decides both, and hands off its own: a request line comes after + // its verdict, so waiting for it waits for both. + first.WaitFor(t, "its request and the other's verdict", func(o Output) bool { + return len(o.RequestsFor(toFirst.ID)) > 0 && len(o.EventsFor(toSecond.ID)) > 0 + }) + second.WaitFor(t, "its request and the other's verdict", func(o Output) bool { + return len(o.RequestsFor(toSecond.ID)) > 0 && len(o.EventsFor(toFirst.ID)) > 0 + }) + firstOut, secondOut := first.Stop(t), second.Stop(t) + + assert.Len(t, firstOut.RequestsFor(toFirst.ID), 1) + assert.Equal(t, "not_addressed", firstOut.EventsFor(toSecond.ID)[0].Reason) + assert.Empty(t, firstOut.RequestsFor(toSecond.ID)) + assert.Len(t, secondOut.RequestsFor(toSecond.ID), 1) + assert.Equal(t, "not_addressed", secondOut.EventsFor(toFirst.ID)[0].Reason) + assert.Empty(t, secondOut.RequestsFor(toFirst.ID)) + assert.Equal(t, strconv.FormatInt(fakebasecamp.AgentID, 10), firstOut.Logged(MsgRunning)[0].Attrs["agent_person_id"]) + assert.Equal(t, strconv.FormatInt(SecondAgentID, 10), secondOut.Logged(MsgRunning)[0].Attrs["agent_person_id"]) +} + +// disconnected is the connector's refusal, in the words guided setup uses +// for the same thing. +const disconnected = "was disconnected in Basecamp, or connected on another computer" + +// The operator disconnects the agent in Basecamp while its connector runs. +// Basecamp revokes the agent's tokens and clears its secret, and leaves the +// live connection up, so the connector learns of it from the next read it +// makes: the operator's next mention. It exits 3, and says what happened and +// how to reconnect. +func TestAnAgentDisconnectedWhileRunningStopsTheConnector(t *testing.T) { + t.Parallel() + h, c := streaming(t) + h.Fake.Disconnect(fakebasecamp.AgentClientID) + ev := comment(h, fakebasecamp.OperatorID, askAgent(), fakebasecamp.Live) + out := c.Wait(t) + + requireDisconnected(t, out) + assert.Empty(t, out.RequestsFor(ev.ID), "nothing is handed off as an agent Basecamp no longer takes") +} + +// The same disconnect, met by the feed instead: the live connection drops, +// and the reconnect's stream ticket is refused. +func TestAnAgentDisconnectedWhileRunningStopsTheConnectorOnReconnect(t *testing.T) { + t.Parallel() + h, c := streaming(t) + h.Fake.Disconnect(fakebasecamp.AgentClientID) + require.Equal(t, 1, h.Fake.DropCable()) + requireDisconnected(t, c.Wait(t)) +} + +// The agent is connected on another computer while its connector runs here. +// Basecamp rotates the agent's secret, and leaves the token this computer +// holds alive, so the connector runs on until that token comes due. The +// renewal is refused invalid_client, and the connector exits 3, saying so. +func TestAnAgentConnectedElsewhereStopsTheConnectorAtItsNextRenewal(t *testing.T) { + t.Parallel() + h := NewHarness(t, shortTokens(world())) + h.Setup(t) + c := h.Connect(t) + c.WaitStreaming(t, 1) + + h.AnotherComputer(t).MustRun(t, "auth", "agent", "connect", "-P", Agent, "--no-browser") + + h.waitTokenDue(t, DefaultAgent) + ev := comment(h, fakebasecamp.OperatorID, askAgent(), fakebasecamp.Live) + out := c.Wait(t) + + requireDisconnected(t, out) + assert.Empty(t, out.RequestsFor(ev.ID)) + invalid := false + for _, r := range h.Fake.Requests() { + invalid = invalid || (r.Route == fakebasecamp.RouteToken && r.Status == http.StatusUnauthorized) + } + assert.True(t, invalid, "the renewal was refused") +} + +// The token endpoint is rate-limited when the connector starts and needs a +// new token: it says it is waiting, waits as long as Basecamp asked, and +// comes up and hands off a mention. +func TestARateLimitedTokenAtStartIsWaitedOut(t *testing.T) { + t.Parallel() + h := NewHarness(t, shortTokens(world())) + h.Setup(t) + h.waitTokenDue(t, DefaultAgent) + limited := fakebasecamp.TokenRateLimited(1) + limited.Times = 1 + fault := h.Fake.Inject(limited) + + c := h.Connect(t) + c.WaitStreaming(t, 1) + ev := comment(h, fakebasecamp.OperatorID, askAgent(), fakebasecamp.Live) + c.WaitFor(t, "the request line", func(o Output) bool { return len(o.RequestsFor(ev.ID)) > 0 }) + out := c.Stop(t) + + waited := false + for _, l := range out.Logs { + waited = waited || (l.Level == "WARN" && strings.HasPrefix(l.Msg, "connector: could not get the agent's token yet")) + } + assert.True(t, waited, "the wait is said\n%s", out.Transcript()) + assert.Equal(t, 1, fault.Hits(), "the token was rate-limited once") + assert.Len(t, out.RequestsFor(ev.ID), 1) +} + +// The live connection drops, and an event Basecamp commits while it is down +// reaches only the poll lane. The reconnect's catch-up walk delivers it, +// and it is handed off once. +func TestAnEventCommittedWhileTheCableIsDownArrivesByCatchUp(t *testing.T) { + t.Parallel() + h, c := streaming(t) + // Hold the reconnect at its stream ticket, so the event is committed + // while the connector is certainly between connections. + reconnect := h.Fake.Gate(fakebasecamp.RouteStreamTicket) + require.Equal(t, 1, h.Fake.DropCable()) + h.await(t, "the reconnect", func() bool { return reconnect.Waiting() > 0 }) + ev := comment(h, fakebasecamp.OperatorID, askAgent(), fakebasecamp.Poll) + reconnect.Release() + + c.WaitFor(t, "the request line", func(o Output) bool { return len(o.RequestsFor(ev.ID)) > 0 }) + c.WaitStreaming(t, 2) + out := c.Stop(t) + + pointers := out.PointersFor(ev.ID) + require.Len(t, pointers, 1) + assert.Equal(t, "poll", pointers[0].Lane) + assert.Len(t, out.RequestsFor(ev.ID), 1) +} + +// The live connection stays open and goes quiet: no pings. The connector +// calls it stale, reconnects, and goes on handing off what is published +// live. Staleness is seven and a half seconds without a ping, which no +// option shortens, so this runs once. +func TestAStaleCableIsReplaced(t *testing.T) { + t.Parallel() + h, c := streaming(t) + h.Fake.StopPings() + c.WaitStreaming(t, 2) + ev := comment(h, fakebasecamp.OperatorID, askAgent(), fakebasecamp.Live) + c.WaitFor(t, "the request line", func(o Output) bool { return len(o.RequestsFor(ev.ID)) > 0 }) + out := c.Stop(t) + assert.Len(t, out.RequestsFor(ev.ID), 1) +} + +// Basecamp fails the recording read twice, then serves it. Admission +// retries within the one decision, so the mention is admitted in a single +// verdict and handed off once. +func TestAReadThatFailsTwiceThenServesIsAdmittedInOneDecision(t *testing.T) { + t.Parallel() + h, c := streaming(t) + failing := h.Fake.Inject(fakebasecamp.ServerError(2, fakebasecamp.RouteComment)) + ev := comment(h, fakebasecamp.OperatorID, askAgent(), fakebasecamp.Live) + c.WaitFor(t, "the request line", func(o Output) bool { return len(o.RequestsFor(ev.ID)) > 0 }) + out := c.Stop(t) + + assert.Equal(t, 2, failing.Hits()) + reads := 0 + for _, r := range h.Fake.Requests() { + if r.Route == fakebasecamp.RouteComment && r.Path == "/"+strconv.FormatInt(fakebasecamp.AccountID, 10)+"/comments/"+strconv.FormatInt(ev.RecordingID, 10) { + reads++ + } + } + assert.Equal(t, 3, reads, "two failures and the read that served") + events := out.EventsFor(ev.ID) + require.Len(t, events, 1, "one decision") + assert.Equal(t, "admitted", events[0].State) + assert.Len(t, out.RequestsFor(ev.ID), 1) +} + +// A connector held by its operator admits a mention and hands nothing off. +// It is killed outright, the hold released, and the connector started +// again within the grace period: the request it admitted and never handed +// off is handed off now, once. +func TestARequestAdmittedBeforeACrashIsHandedOffOnRestart(t *testing.T) { + t.Parallel() + h := NewHarness(t, world()) + h.Setup(t) + held := h.Connect(t, "--hold") + held.WaitStreaming(t, 1) + ev := comment(h, fakebasecamp.OperatorID, askAgent(), fakebasecamp.Live) + before := held.WaitFor(t, "the event line", func(o Output) bool { return len(o.EventsFor(ev.ID)) > 0 }) + require.Equal(t, "admitted", before.EventsFor(ev.ID)[0].State) + require.Empty(t, before.RequestsFor(ev.ID), "held: nothing is handed off") + held.Kill(t) + require.Equal(t, -1, held.Wait(t).ExitCode, "killed") + + h.MustRun(t, "connect", "release", "-P", Agent) + c := h.Connect(t) + c.WaitFor(t, "the request line", func(o Output) bool { return len(o.RequestsFor(ev.ID)) > 0 }) + out := c.Stop(t) + assert.Len(t, out.RequestsFor(ev.ID), 1) +} + +// Admission falls behind the feed: Basecamp holds every recording read +// open while events arrive live. The connector warns that its backlog +// reached the warning depth, and says so again when the reads come back and +// the backlog falls below it. +func TestABacklogPastItsWarningDepthIsReportedAndRecovers(t *testing.T) { + t.Parallel() + h, c := streaming(t) + reads := h.Fake.Gate(fakebasecamp.RecordingReads...) + // Each admission worker takes one event off the backlog and is held at + // its read, so the backlog reaches the warning depth only once that + // many more have arrived. + const backlog = connector.DefaultBacklogWarn + admission.DefaultWorkers + recordings := make([]int64, backlog) + h.Fake.Update(func(w *fakebasecamp.World) { + w.Recordings[kickoffID] = &fakebasecamp.Recording{ + ID: kickoffID, Type: "Message", BucketID: fakebasecamp.ProjectID, CreatorID: fakebasecamp.OperatorID, Title: "Kickoff", + } + for i := range recordings { + id := kickoffID + 1 + int64(i) + recordings[i] = id + w.Recordings[id] = &fakebasecamp.Recording{ + ID: id, Type: "Comment", BucketID: fakebasecamp.ProjectID, ParentID: kickoffID, CreatorID: fakebasecamp.OperatorID, Content: "

Noted

", + } + } + }) + for _, id := range recordings { + h.Fake.Emit(fakebasecamp.Event{ + EventType: "comment.created", BucketID: fakebasecamp.ProjectID, RecordingID: id, CreatorID: fakebasecamp.OperatorID, + }, fakebasecamp.Live) + } + warned := c.WaitLogged(t, MsgBacklogWarn, 1).Logged(MsgBacklogWarn)[0] + reads.Release() + recovered := c.WaitLogged(t, MsgBacklogRecover, 1).Logged(MsgBacklogRecover)[0] + c.Stop(t) + + assert.Equal(t, "WARN", warned.Level) + assert.Equal(t, strconv.Itoa(connector.DefaultBacklogWarn), warned.Attrs["warn_at"]) + assert.Equal(t, "INFO", recovered.Level) +} + +// Messages the connector's queue logs at the edges of its backlog. +const ( + MsgBacklogWarn = "the backlog reached its warning depth; admission is falling behind the feed" + MsgBacklogRecover = "the backlog fell back below its warning depth" +) + +// requireDisconnected requires that the connector stopped because Basecamp +// no longer takes its agent's credential, and said so in the person's words. +func requireDisconnected(t *testing.T, out Output) { + t.Helper() + require.Equal(t, exitAuth, out.ExitCode, out.Transcript()) + assert.Contains(t, strings.Join(rawLogs(out), "\n"), disconnected, out.Transcript()) + assert.Contains(t, strings.Join(rawLogs(out), "\n"), "basecamp connect setup -P "+Agent) +} + +// mentioning is a comment that asks one agent for something. +func mentioning(agentID int64) string { + return "

Could you take this, " + fakebasecamp.Mention(agentID) + "?

" +} + +// shortTokens is w with every agent's tokens lasting a second, so a test +// can see one come due without waiting an hour. +func shortTokens(w *fakebasecamp.World) *fakebasecamp.World { + for _, a := range w.Agents { + a.TokenLifetime = time.Second + } + return w +} + +// waitTokenDue waits until the token p's profile holds is due for renewal: +// past the expiry `basecamp auth status` reports for it, which is the +// latest the CLI renews it. This is the one wait here on the clock, and it +// is the scenario's premise, not a guess at how long something takes: a +// token comes due only with time. +func (h *Harness) waitTokenDue(t *testing.T, p Profile) { + t.Helper() + res := h.MustRun(t, "auth", "status", "-P", p.Name, "--json") + var status struct { + Data struct { + ExpiresAt time.Time `json:"expires_at"` + } `json:"data"` + } + require.NoError(t, json.Unmarshal([]byte(res.Stdout), &status), res.Stdout) + require.False(t, status.Data.ExpiresAt.IsZero(), res.Stdout) + // expires_at is whole seconds, and the CLI compares whole seconds. + due := time.Until(status.Data.ExpiresAt.Add(time.Second)) + require.Less(t, due, waitFor, "the token is not short-lived") + select { + case <-time.After(due): + case <-t.Context().Done(): + t.Fatal(t.Context().Err()) + } +} + +// await waits for the fake to come to a state, failing the test when it +// does not within waitFor. +func (h *Harness) await(t *testing.T, what string, cond func() bool) { + t.Helper() + ctx, cancel := context.WithTimeout(t.Context(), waitFor) + defer cancel() + require.NoError(t, h.Fake.Await(ctx, cond), "waiting for %s", what) +} + +func rawLogs(o Output) []string { + out := make([]string, len(o.Logs)) + for i, l := range o.Logs { + out[i] = l.Raw + } + return out +} diff --git a/e2e/connect/harness_test.go b/e2e/connect/harness_test.go index 2d20e5400..df70e0e5e 100644 --- a/e2e/connect/harness_test.go +++ b/e2e/connect/harness_test.go @@ -28,6 +28,18 @@ const waitFor = 30 * time.Second // Agent is the profile every test connects the default world's agent as. const Agent = "agent" +// Profile is an agent a test connects, and the profile it connects it as. +type Profile struct { + // Name is the CLI profile, as -P takes it. + Name string + // ClientID is the agent's OAuth client in the fake's world: the one the + // operator approves in the profile's connection ceremony. + ClientID string +} + +// DefaultAgent is the default world's agent, as the Agent profile. +var DefaultAgent = Profile{Name: Agent, ClientID: fakebasecamp.AgentClientID} + // Harness is one test's Basecamp, and a home of its own to run the CLI in: // its config, its credentials and the connector's state are the test's // alone. @@ -119,18 +131,35 @@ func (h *Harness) MustRun(t *testing.T, args ...string) Result { } // Setup connects the default world's agent as the Agent profile and sets -// it up to serve the default project for the operator, the way a person -// does: the connection ceremony with no browser, which the fake approves as -// soon as it is asked, then `connect setup`. +// it up to serve the default project for the operator. func (h *Harness) Setup(t *testing.T) { t.Helper() - h.MustRun(t, "auth", "agent", "connect", "-P", Agent, "--no-browser") - h.MustRun(t, "connect", "setup", "-P", Agent, + h.SetupProfile(t, DefaultAgent) +} + +// SetupProfile connects p's agent as p's profile and sets it up to serve +// the default project for the operator, the way a person does: the +// connection ceremony with no browser, in which the fake's operator +// approves p's agent as soon as it is asked, then `connect setup`. +func (h *Harness) SetupProfile(t *testing.T, p Profile) { + t.Helper() + h.Fake.Update(func(w *fakebasecamp.World) { w.Connection.ClientID = p.ClientID }) + h.MustRun(t, "auth", "agent", "connect", "-P", p.Name, "--no-browser") + h.MustRun(t, "connect", "setup", "-P", p.Name, "--operator", strconv.FormatInt(fakebasecamp.OperatorID, 10), "--serve", strconv.FormatInt(fakebasecamp.ProjectID, 10), "--json") } +// AnotherComputer is a home of its own against the same Basecamp: the +// operator's other computer, with nothing of this one's config, +// credentials or connector state. +func (h *Harness) AnotherComputer(t *testing.T) *Harness { + t.Helper() + home := t.TempDir() + return &Harness{t: t, Fake: h.Fake, Home: home, env: cliEnv(home, h.Fake.URL())} +} + // ConnectJSON is where `connect setup` writes the Agent profile's // connect.json in this home. func (h *Harness) ConnectJSON() string { diff --git a/internal/connector/fakebasecamp/connect_test.go b/internal/connector/fakebasecamp/connect_test.go index a3653dfa4..0a4e45591 100644 --- a/internal/connector/fakebasecamp/connect_test.go +++ b/internal/connector/fakebasecamp/connect_test.go @@ -1,10 +1,13 @@ package fakebasecamp_test import ( + "encoding/json" "net/http" "net/url" "testing" + "time" + "github.com/basecamp/basecamp-sdk/go/pkg/basecamp" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -98,3 +101,54 @@ func TestAgentConnectionHandsOverTheApprovedScope(t *testing.T) { assert.Equal(t, fakebasecamp.ScopeRead, w.Agents[fakebasecamp.AgentClientID].Scope) }) } + +// mint asks the fake for a client_credentials token for the default agent's +// client with secret, and returns the status and the decoded body. +func mint(t *testing.T, s *fakebasecamp.Server, secret string) (int, map[string]any) { + t.Helper() + 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()) + var decoded map[string]any + require.NoError(t, json.Unmarshal([]byte(body), &decoded), body) + return status, decoded +} + +// A token says it lasts an hour, as Basecamp's do, unless the client is +// set to mint shorter ones. +func TestTokensSayHowLongTheyLast(t *testing.T) { + s := fakebasecamp.Start(t, fakebasecamp.DefaultWorld()) + status, body := mint(t, s, fakebasecamp.AgentSecret) + require.Equal(t, http.StatusOK, status) + assert.InDelta(t, 3600, body["expires_in"], 0) + + s.Update(func(w *fakebasecamp.World) { w.Agents[fakebasecamp.AgentClientID].TokenLifetime = 2 * time.Second }) + _, body = mint(t, s, fakebasecamp.AgentSecret) + assert.InDelta(t, 2, body["expires_in"], 0) + + s.Update(func(w *fakebasecamp.World) { w.Agents[fakebasecamp.AgentClientID].TokenLifetime = time.Millisecond }) + _, body = mint(t, s, fakebasecamp.AgentSecret) + assert.InDelta(t, 1, body["expires_in"], 0, "never less than a second") +} + +// Disconnecting an agent revokes every token its client minted and clears +// its secret, and leaves everyone else's tokens alone. +func TestDisconnectRevokesTheAgentsTokensAndSecret(t *testing.T) { + s := fakebasecamp.Start(t, fakebasecamp.DefaultWorld()) + status, body := mint(t, s, fakebasecamp.AgentSecret) + require.Equal(t, http.StatusOK, status) + token, _ := body["access_token"].(string) + _, err := accountClient(s, token).People().Me(bounded(t)) + require.NoError(t, err) + + s.Disconnect(fakebasecamp.AgentClientID) + + _, err = accountClient(s, token).People().Me(bounded(t)) + var apiErr *basecamp.Error + require.ErrorAs(t, err, &apiErr) + assert.Equal(t, http.StatusUnauthorized, apiErr.HTTPStatus) + status, body = mint(t, s, fakebasecamp.AgentSecret) + assert.Equal(t, http.StatusUnauthorized, status) + assert.Equal(t, "invalid_client", body["error"]) + _, err = accountClient(s, fakebasecamp.OperatorToken).People().Me(bounded(t)) + assert.NoError(t, err, "the operator's own login stands") +} diff --git a/internal/connector/fakebasecamp/doc.go b/internal/connector/fakebasecamp/doc.go index 2a52b3c97..362479a80 100644 --- a/internal/connector/fakebasecamp/doc.go +++ b/internal/connector/fakebasecamp/doc.go @@ -52,6 +52,10 @@ // 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. +// StopPings silences them. Disconnect does what disconnecting an agent in +// Basecamp does: its tokens stop working and its secret stops minting. An +// agent's client can mint tokens that say they last as little as a second +// (AgentClient.TokenLifetime), so a renewal comes due within a test. +// 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/oauth.go b/internal/connector/fakebasecamp/oauth.go index 4e1f640f8..527cb8e94 100644 --- a/internal/connector/fakebasecamp/oauth.go +++ b/internal/connector/fakebasecamp/oauth.go @@ -2,7 +2,9 @@ package fakebasecamp import ( "fmt" + "maps" "net/http" + "time" ) // The authorization server: discovery, the agent connection ceremony, and @@ -80,9 +82,10 @@ func (c *call) agentConnectionTokens() answer { } // 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 +// the client does not hold now, rotated away, cleared by a disconnect 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, and says it lasts +// the client's TokenLifetime; 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 { @@ -91,7 +94,7 @@ func (c *call) token() answer { 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") { + if !ok || client.Secret == "" || client.Secret != form.Get("client_secret") { return oauthError(http.StatusUnauthorized, "invalid_client") } scope := client.Scope @@ -102,11 +105,15 @@ func (c *call) token() answer { scope = asked } token := fmt.Sprintf("fake-token-%d", c.s.nextSerialLocked()) - c.s.world.Tokens[token] = &Token{PersonID: client.PersonID, Scope: scope} + c.s.world.Tokens[token] = &Token{PersonID: client.PersonID, Scope: scope, ClientID: client.ID} + lifetime := int64(time.Hour / time.Second) + if client.TokenLifetime > 0 { + lifetime = max(int64(client.TokenLifetime/time.Second), 1) + } return c.jsonAnswer(http.StatusOK, map[string]any{ "access_token": token, "token_type": "bearer", - "expires_in": 3600, + "expires_in": lifetime, "resource": fmt.Sprintf("urn:bc:agent:%d", client.PersonID), "scope": scope, }) @@ -124,3 +131,26 @@ func within(asked, approved string) bool { return false } } + +// Disconnect disconnects an agent's client, as Basecamp's disconnect does: +// every token it minted is revoked, so a request carrying one is answered +// 401, and its secret is cleared, so a mint is refused invalid_client. The +// live connections stay up, as they do on Basecamp: a stream ticket is not +// a token. A later connection ceremony hands the client over again. +func (s *Server) Disconnect(clientID string) { + s.mu.Lock() + defer s.mu.Unlock() + client, ok := s.world.Agents[clientID] + if !ok { + s.r.Errorf("fakebasecamp: disconnecting client %q, which the world does not hold", clientID) + return + } + s.revokeLocked(clientID) + client.Secret = "" + s.notifyLocked() +} + +// revokeLocked forgets every token a client minted. +func (s *Server) revokeLocked(clientID string) { + maps.DeleteFunc(s.world.Tokens, func(_ string, t *Token) bool { return t.ClientID == clientID }) +} diff --git a/internal/connector/fakebasecamp/world.go b/internal/connector/fakebasecamp/world.go index b59c4752e..fad264258 100644 --- a/internal/connector/fakebasecamp/world.go +++ b/internal/connector/fakebasecamp/world.go @@ -141,12 +141,20 @@ type AgentClient struct { PersonID int64 // Scope is what the client was approved for. Scope string + // TokenLifetime is how long the tokens it mints say they last, in + // whole seconds of at least one; zero is Basecamp's hour. The fake + // does not retire a token when it runs out: the client renews by what + // it was told, and a test can wait for that to come due. + TokenLifetime time.Duration } // Token is what a bearer token authenticates as. type Token struct { PersonID int64 Scope string + // ClientID is the agent client that minted it; empty for a person's + // own login. + ClientID string } // Connection is the operator's approval on the agent connection page.