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/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 } 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. 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.