Merge pull request 'Re-mint watchpr's Gitea token when it expires' (#10) from benvin/watchpr-auth-expiry into main

Reviewed-on: #10

AGENTS.md

@@ -174,8 +174,13 @@ wrapped per stage (login / read denied / write denied) via `ErrVaultDenied`.
 ## Gotchas
 
 - `watchpr` exits 0 with no output changes on `--once` (just prints state).
-- The token cache is process-wide (`sync.Once`); tests call the unexported
+- Gitea tokens expire in ~1h, shorter than a watch: the client re-mints once on a
-  `fetchGiteaToken` to avoid it.
+  401/403 and replays the request. If the fresh token is rejected too, `watchpr`
+  exits non-zero rather than polling blind.
+- `watchpr` polls anonymously when no token can be minted (public repos work
+  fine); only a real 401/403 reaches for Vault.
+- The token cache is process-wide (mutex-guarded); `RefreshGiteaToken` replaces
+  it. Tests call the unexported `fetchGiteaToken` to avoid the cache.
 - `agentvault` never puts a secret in an error string: Vault decode failures and
   Authentik `view_key` responses are reported without their bodies, and
   `seed-oauth` reports key names only.

cmd/watchpr/main.go

@@ -58,10 +58,7 @@ func newRootCmd() *cobra.Command {
 				}
 				refs = append(refs, ref)
 			}
-			c, err := clientFor()
+			c := clientFor()
-			if err != nil {
-				return err
-			}
 			if once {
 				return runOnce(c, refs, jsonMode)
 			}
@@ -84,12 +81,16 @@ func newRootCmd() *cobra.Command {
 	return root
 }
 
-func clientFor() (*agent.GiteaClient, error) {
+// clientFor builds the Gitea client. Watching public repos works anonymously,
+// so an unavailable token is a warning, not a failure; a poll that is actually
+// rejected re-mints then.
+func clientFor() *agent.GiteaClient {
 	token, err := agent.GiteaToken()
 	if err != nil {
-		return nil, err
+		fmt.Fprintf(os.Stderr, "warning: no Gitea token (%v); polling anonymously\n", err)
+		token = ""
 	}
-	return agent.NewGiteaClient(token), nil
+	return agent.NewGiteaClient(token)
 }
 
 // runOnce fetches and prints the current state of each PR, then exits 0.
@@ -131,6 +132,9 @@ func runWatch(c *agent.GiteaClient, refs []agent.PRRef, interval time.Duration,
 
 	res, err := agent.Watch(c, refs, login, ticker.C, onBaseline, onError)
 	if err != nil {
+		if agent.IsAuthError(err) {
+			return fmt.Errorf("gitea authentication failed after re-minting the token, watch aborted: %w", err)
+		}
 		return err
 	}
 	report(res.Ref.String(), res.Reason, res.State, jsonMode)

cmd/watchpr/main_test.go

@@ -2,6 +2,8 @@ package main
 
 import (
 	"io"
+	"net/http"
+	"net/http/httptest"
 	"testing"
 )
 
@@ -28,3 +30,43 @@ func TestExecuteNoArgsErrors(t *testing.T) {
 		t.Fatal("Execute() = nil, want error when no PR references are given")
 	}
 }
+
+// Watching a public repo with no credentials available must work: the failed
+// mint is a warning, the poll goes out unauthenticated, and the command exits 0.
+func TestOnceRunsAnonymouslyWhenNoTokenIsAvailable(t *testing.T) {
+	vault := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(http.StatusForbidden)
+	}))
+	defer vault.Close()
+
+	authHeaders := 0
+	mux := http.NewServeMux()
+	mux.HandleFunc("/api/v1/repos/unkin/repo/pulls/7", func(w http.ResponseWriter, r *http.Request) {
+		if r.Header.Get("Authorization") != "" {
+			authHeaders++
+		}
+		_, _ = io.WriteString(w, `{"number":7,"state":"open","mergeable":true,"head":{"sha":"cafebabe"}}`)
+	})
+	mux.HandleFunc("/api/v1/repos/unkin/repo/commits/cafebabe/status", func(w http.ResponseWriter, r *http.Request) {
+		_, _ = io.WriteString(w, `{"state":"success"}`)
+	})
+	mux.HandleFunc("/api/v1/repos/unkin/repo/issues/7/comments", func(w http.ResponseWriter, r *http.Request) {
+		_, _ = io.WriteString(w, `[]`)
+	})
+	gitea := httptest.NewServer(mux)
+	defer gitea.Close()
+
+	t.Setenv("VAULT_ADDR", vault.URL)
+	t.Setenv("GITEA_URL", gitea.URL)
+
+	cmd := newRootCmd()
+	cmd.SetArgs([]string{"--once", "unkin/repo#7"})
+	cmd.SetOut(io.Discard)
+	cmd.SetErr(io.Discard)
+	if err := cmd.Execute(); err != nil {
+		t.Fatalf("anonymous --once should succeed without a token: %v", err)
+	}
+	if authHeaders != 0 {
+		t.Errorf("sent %d Authorization headers, want none", authHeaders)
+	}
+}

internal/agent/client_test.go

@@ -2,9 +2,11 @@ package agent
 
 import (
 	"encoding/json"
+	"errors"
 	"io"
 	"net/http"
 	"net/http/httptest"
+	"strings"
 	"testing"
 )
 
@@ -231,3 +233,219 @@ func TestGiteaAPIError(t *testing.T) {
 		t.Fatal("expected error on 422")
 	}
 }
+
+// expiringGitea serves the PR endpoint, rejecting every token other than
+// wantToken with a 401 exactly as Gitea does once a Vault-minted token expires.
+// It records the tokens it saw, newest last.
+func expiringGitea(t *testing.T, wantToken string, seen *[]string) *httptest.Server {
+	t.Helper()
+	mux := http.NewServeMux()
+	mux.HandleFunc("/api/v1/repos/unkin/repo/pulls/7", func(w http.ResponseWriter, r *http.Request) {
+		tok := strings.TrimPrefix(r.Header.Get("Authorization"), "token ")
+		*seen = append(*seen, tok)
+		if tok != wantToken {
+			w.WriteHeader(http.StatusUnauthorized)
+			_, _ = io.WriteString(w, `{"message":"invalid username, password or token"}`)
+			return
+		}
+		_, _ = io.WriteString(w, `{"number":7,"state":"open","mergeable":true,"head":{"sha":"cafebabe"}}`)
+	})
+	return httptest.NewServer(mux)
+}
+
+// The production failure: the token expired mid-run. The client must re-mint
+// once and replay the request with the fresh token.
+func TestExpiredTokenIsRemintedAndRetried(t *testing.T) {
+	var seen []string
+	srv := expiringGitea(t, "fresh", &seen)
+	defer srv.Close()
+
+	refreshes := 0
+	c := &GiteaClient{BaseURL: srv.URL, Token: "stale", HTTP: srv.Client(),
+		Refresh: func() (string, error) { refreshes++; return "fresh", nil }}
+
+	pr, err := c.GetPR("unkin/repo", 7)
+	if err != nil {
+		t.Fatalf("GetPR after re-mint: %v", err)
+	}
+	if pr.Number != 7 {
+		t.Errorf("PR number = %d, want 7", pr.Number)
+	}
+	if refreshes != 1 {
+		t.Errorf("refreshes = %d, want 1", refreshes)
+	}
+	if len(seen) != 2 || seen[0] != "stale" || seen[1] != "fresh" {
+		t.Errorf("tokens seen = %v, want [stale fresh]", seen)
+	}
+	if c.Token != "fresh" {
+		t.Errorf("client token = %q, want the refreshed token", c.Token)
+	}
+}
+
+// A fresh token that is also rejected is a real auth failure: report it as one
+// rather than re-minting forever.
+func TestAuthFailureSurvivesRemint(t *testing.T) {
+	var seen []string
+	srv := expiringGitea(t, "never-issued", &seen)
+	defer srv.Close()
+
+	refreshes := 0
+	c := &GiteaClient{BaseURL: srv.URL, Token: "stale", HTTP: srv.Client(),
+		Refresh: func() (string, error) { refreshes++; return "still-bad", nil }}
+
+	_, err := c.GetPR("unkin/repo", 7)
+	if err == nil {
+		t.Fatal("GetPR should fail when the fresh token is rejected too")
+	}
+	if !IsAuthError(err) {
+		t.Errorf("IsAuthError(%v) = false, want true", err)
+	}
+	if refreshes != 1 {
+		t.Errorf("refreshes = %d, want 1 (re-mint exactly once)", refreshes)
+	}
+	if len(seen) != 2 {
+		t.Errorf("requests = %d, want 2", len(seen))
+	}
+}
+
+// A refresh that itself fails must surface as an auth error, not as a silent
+// success or a bare Vault error.
+func TestRemintErrorIsReportedAsAuthFailure(t *testing.T) {
+	var seen []string
+	srv := expiringGitea(t, "fresh", &seen)
+	defer srv.Close()
+
+	c := &GiteaClient{BaseURL: srv.URL, Token: "stale", HTTP: srv.Client(),
+		Refresh: func() (string, error) { return "", errors.New("vault approle login: HTTP 503") }}
+
+	_, err := c.GetPR("unkin/repo", 7)
+	if err == nil || !IsAuthError(err) {
+		t.Fatalf("GetPR error = %v, want an auth error", err)
+	}
+	if !strings.Contains(err.Error(), "vault approle login") {
+		t.Errorf("error %q should name the re-mint failure", err)
+	}
+	if len(seen) != 1 {
+		t.Errorf("requests = %d, want 1 (no replay without a token)", len(seen))
+	}
+}
+
+// A 5xx is transient, not an auth problem: no re-mint, no retry, and the caller
+// keeps its existing retry behaviour.
+func TestServerErrorDoesNotRemint(t *testing.T) {
+	requests := 0
+	mux := http.NewServeMux()
+	mux.HandleFunc("/api/v1/repos/unkin/repo/pulls/7", func(w http.ResponseWriter, r *http.Request) {
+		requests++
+		w.WriteHeader(http.StatusBadGateway)
+	})
+	srv := httptest.NewServer(mux)
+	defer srv.Close()
+
+	refreshes := 0
+	c := &GiteaClient{BaseURL: srv.URL, Token: "t", HTTP: srv.Client(),
+		Refresh: func() (string, error) { refreshes++; return "fresh", nil }}
+
+	_, err := c.GetPR("unkin/repo", 7)
+	if err == nil {
+		t.Fatal("expected error on 502")
+	}
+	if IsAuthError(err) {
+		t.Errorf("502 must not be an auth error")
+	}
+	if refreshes != 0 || requests != 1 {
+		t.Errorf("refreshes = %d, requests = %d, want 0 and 1", refreshes, requests)
+	}
+}
+
+// The replayed request must carry the original body, not an empty one.
+func TestRemintReplaysRequestBody(t *testing.T) {
+	var bodies []CreatePROptions
+	mux := http.NewServeMux()
+	mux.HandleFunc("/api/v1/repos/unkin/repo/pulls", func(w http.ResponseWriter, r *http.Request) {
+		var body CreatePROptions
+		_ = json.NewDecoder(r.Body).Decode(&body)
+		bodies = append(bodies, body)
+		if strings.TrimPrefix(r.Header.Get("Authorization"), "token ") != "fresh" {
+			w.WriteHeader(http.StatusUnauthorized)
+			return
+		}
+		_, _ = io.WriteString(w, `{"number":7}`)
+	})
+	srv := httptest.NewServer(mux)
+	defer srv.Close()
+
+	c := &GiteaClient{BaseURL: srv.URL, Token: "stale", HTTP: srv.Client(),
+		Refresh: func() (string, error) { return "fresh", nil }}
+
+	if _, err := c.CreatePR("unkin/repo", CreatePROptions{Base: "main", Head: "feature", Title: "T", Body: "B"}); err != nil {
+		t.Fatalf("CreatePR: %v", err)
+	}
+	if len(bodies) != 2 {
+		t.Fatalf("requests = %d, want 2", len(bodies))
+	}
+	if bodies[1] != bodies[0] {
+		t.Errorf("replayed body = %+v, want %+v", bodies[1], bodies[0])
+	}
+}
+
+func TestIsAuthError(t *testing.T) {
+	tests := []struct {
+		status int
+		want   bool
+	}{
+		{http.StatusUnauthorized, true},
+		{http.StatusForbidden, true},
+		{http.StatusNotFound, false},
+		{http.StatusUnprocessableEntity, false},
+		{http.StatusBadGateway, false},
+	}
+	for _, tt := range tests {
+		err := error(&APIError{Method: "GET", Path: "/p", StatusCode: tt.status})
+		if got := IsAuthError(err); got != tt.want {
+			t.Errorf("IsAuthError(HTTP %d) = %v, want %v", tt.status, got, tt.want)
+		}
+	}
+	if IsAuthError(errors.New("dial tcp: timeout")) {
+		t.Errorf("a network error is not an auth error")
+	}
+}
+
+// Anonymous polling of a public repo is a supported mode: with no token the
+// client must send no Authorization header, and must never reach for Vault.
+func TestAnonymousPollingNeverMints(t *testing.T) {
+	authHeaders := 0
+	mux := http.NewServeMux()
+	mux.HandleFunc("/api/v1/repos/unkin/repo/pulls/7", func(w http.ResponseWriter, r *http.Request) {
+		if r.Header.Get("Authorization") != "" {
+			authHeaders++
+		}
+		_, _ = io.WriteString(w, `{"number":7,"state":"open","mergeable":true,"head":{"sha":"cafebabe"}}`)
+	})
+	mux.HandleFunc("/api/v1/repos/unkin/repo/commits/cafebabe/status", func(w http.ResponseWriter, r *http.Request) {
+		_, _ = io.WriteString(w, `{"state":"success"}`)
+	})
+	mux.HandleFunc("/api/v1/repos/unkin/repo/issues/7/comments", func(w http.ResponseWriter, r *http.Request) {
+		_, _ = io.WriteString(w, `[]`)
+	})
+	srv := httptest.NewServer(mux)
+	defer srv.Close()
+
+	refreshes := 0
+	c := &GiteaClient{BaseURL: srv.URL, HTTP: srv.Client(),
+		Refresh: func() (string, error) { refreshes++; return "", errors.New("vault unreachable") }}
+
+	st, err := FetchState(c, PRRef{Owner: "unkin", Repo: "repo", Number: 7}, "unkin-agent")
+	if err != nil {
+		t.Fatalf("anonymous FetchState: %v", err)
+	}
+	if st.State != "open" || st.CIStatus != "success" || st.HeadSHA != "cafebabe" {
+		t.Errorf("state = %+v", st)
+	}
+	if refreshes != 0 {
+		t.Errorf("refreshes = %d, want 0 (a 200 must never trigger a mint)", refreshes)
+	}
+	if authHeaders != 0 {
+		t.Errorf("sent %d Authorization headers, want none", authHeaders)
+	}
+}

internal/agent/gitea.go

@@ -30,34 +30,67 @@ func isNotFound(err error) bool {
 	return errors.As(err, &apiErr) && apiErr.StatusCode == http.StatusNotFound
 }
 
+// IsAuthError reports whether err is a Gitea 401/403: the token is expired or
+// unauthorised, which retrying the same request cannot fix.
+func IsAuthError(err error) bool {
+	var apiErr *APIError
+	return errors.As(err, &apiErr) &&
+		(apiErr.StatusCode == http.StatusUnauthorized || apiErr.StatusCode == http.StatusForbidden)
+}
+
 // GiteaClient talks to the Gitea REST API as the agent user.
 type GiteaClient struct {
 	BaseURL string
 	Token   string
 	HTTP    *http.Client
+	// Refresh mints a replacement token when the current one is rejected; Vault's
+	// Gitea tokens expire in ~1h, far short of a watchpr run.
+	Refresh func() (string, error)
 }
 
 // NewGiteaClient builds a client from the configured base URL and a Vault-minted
-// token.
+// token, re-minting from Vault when that token expires.
 func NewGiteaClient(token string) *GiteaClient {
-	return &GiteaClient{BaseURL: GiteaURL(), Token: token, HTTP: httpClient}
+	return &GiteaClient{BaseURL: GiteaURL(), Token: token, HTTP: httpClient, Refresh: RefreshGiteaToken}
 }
 
+// do sends the request and, if the token was rejected, re-mints it once and
+// replays the request with the fresh token.
 func (c *GiteaClient) do(method, path string, body any, out any) error {
-	var reader io.Reader
+	var payload []byte
 	if body != nil {
 		b, err := json.Marshal(body)
 		if err != nil {
 			return err
 		}
-		reader = bytes.NewReader(b)
+		payload = b
+	}
+	err := c.attempt(method, path, payload, out)
+	if !IsAuthError(err) || c.Refresh == nil {
+		return err
+	}
+	token, refreshErr := c.Refresh()
+	if refreshErr != nil {
+		return fmt.Errorf("%w; re-minting token: %v", err, refreshErr)
+	}
+	c.Token = token
+	return c.attempt(method, path, payload, out)
+}
+
+func (c *GiteaClient) attempt(method, path string, body []byte, out any) error {
+	var reader io.Reader
+	if body != nil {
+		reader = bytes.NewReader(body)
 	}
 	url := strings.TrimRight(c.BaseURL, "/") + path
 	req, err := http.NewRequest(method, url, reader)
 	if err != nil {
 		return err
 	}
-	req.Header.Set("Authorization", "token "+c.Token)
+	// An empty token means anonymous access, which public repos serve fine.
+	if c.Token != "" {
+		req.Header.Set("Authorization", "token "+c.Token)
+	}
 	req.Header.Set("Accept", "application/json")
 	if body != nil {
 		req.Header.Set("Content-Type", "application/json")

internal/agent/token.go

@@ -86,17 +86,31 @@ func AuthentikURL() string {
 }
 
 var (
-	tokenOnce  sync.Once
+	tokenMu     sync.Mutex
-	tokenValue string
+	tokenMinted bool
-	tokenErr   error
+	tokenValue  string
+	tokenErr    error
 )
 
 // GiteaToken returns a Gitea token, minting it via Vault AppRole on first call
 // and caching it in-process for the lifetime of the command.
 func GiteaToken() (string, error) {
-	tokenOnce.Do(func() {
+	tokenMu.Lock()
+	defer tokenMu.Unlock()
+	if !tokenMinted {
 		tokenValue, tokenErr = fetchGiteaToken(VaultAddr(), RoleID(), GiteaCredsPath())
-	})
+		tokenMinted = true
+	}
+	return tokenValue, tokenErr
+}
+
+// RefreshGiteaToken mints a fresh Gitea token and replaces the cached one, for
+// callers that outlive the ~1h token TTL.
+func RefreshGiteaToken() (string, error) {
+	tokenMu.Lock()
+	defer tokenMu.Unlock()
+	tokenValue, tokenErr = fetchGiteaToken(VaultAddr(), RoleID(), GiteaCredsPath())
+	tokenMinted = true
 	return tokenValue, tokenErr
 }
 

internal/agent/watch.go

@@ -1,6 +1,9 @@
 package agent
 
-import "time"
+import (
+	"fmt"
+	"time"
+)
 
 // PRState is a point-in-time snapshot of the PR attributes watchpr tracks.
 type PRState struct {
@@ -82,8 +85,10 @@ func terminalState(st PRState) (bool, string) {
 // tracked PR changes meaningfully, returning the first such change. A PR that is
 // already terminal (merged/closed) at baseline is reported immediately rather
 // than polled forever. Poll errors are handed to onError and never stop the
-// loop; only a baseline fetch error aborts. onBaseline, if set, fires once after
+// loop; a baseline fetch error and an authentication failure (the token was
-// all baselines are captured and before the first tick.
+// rejected and re-minting it did not help) abort instead, because a watcher that
+// cannot authenticate sees nothing. onBaseline, if set, fires once after all
+// baselines are captured and before the first tick.
 func Watch(f StateFetcher, refs []PRRef, agentLogin string, ticks <-chan time.Time, onBaseline func(), onError func(PRRef, error)) (WatchResult, error) {
 	prev := make(map[string]PRState, len(refs))
 	for _, ref := range refs {
@@ -104,6 +109,9 @@ func Watch(f StateFetcher, refs []PRRef, agentLogin string, ticks <-chan time.Ti
 			key := ref.String()
 			cur, err := f.FetchState(ref, agentLogin)
 			if err != nil {
+				if IsAuthError(err) {
+					return WatchResult{}, fmt.Errorf("polling %s: %w", key, err)
+				}
 				if onError != nil {
 					onError(ref, err)
 				}

internal/agent/watch_test.go

@@ -360,3 +360,86 @@ func TestCountNonAgentComments(t *testing.T) {
 		t.Errorf("countNonAgentComments = %d, want 2", n)
 	}
 }
+
+// The production failure: the Vault-minted token expired mid-watch and every
+// poll 401'd, which the loop logged as a warning and polled past forever. An
+// auth error that survived the client's re-mint must end the watch with an
+// error so watchpr exits non-zero instead of watching blind.
+func TestWatchAbortsOnAuthError(t *testing.T) {
+	open := base()
+	merged := base()
+	merged.State = "closed"
+	merged.Merged = true
+	f := &fakeFetcher{
+		states: []PRState{open, open, merged},
+		errs:   []error{nil, &APIError{Method: "GET", Path: "/p", StatusCode: 401, Body: "invalid token"}, nil},
+	}
+
+	warned := 0
+	ticks := make(chan time.Time, 2)
+	ticks <- time.Now()
+	ticks <- time.Now()
+	_, err := Watch(f, []PRRef{open.Ref}, "unkin-agent", ticks, nil, func(PRRef, error) { warned++ })
+	if err == nil {
+		t.Fatal("Watch should return the auth failure, not keep polling")
+	}
+	if !IsAuthError(err) {
+		t.Errorf("Watch error = %v, want an auth error", err)
+	}
+	if warned != 0 {
+		t.Errorf("auth failure was logged as a warning %d time(s); it must abort", warned)
+	}
+	if f.calls != 2 {
+		t.Errorf("fetch calls = %d, want 2 (baseline + the failing poll)", f.calls)
+	}
+}
+
+// A 5xx keeps its retry behaviour: warn and poll on.
+func TestWatchContinuesPastServerError(t *testing.T) {
+	open := base()
+	merged := base()
+	merged.State = "closed"
+	merged.Merged = true
+	f := &fakeFetcher{
+		states: []PRState{open, open, merged},
+		errs:   []error{nil, &APIError{Method: "GET", Path: "/p", StatusCode: 502, Body: "bad gateway"}, nil},
+	}
+
+	warned := 0
+	ticks := make(chan time.Time, 2)
+	ticks <- time.Now()
+	ticks <- time.Now()
+	res, err := Watch(f, []PRRef{open.Ref}, "unkin-agent", ticks, nil, func(PRRef, error) { warned++ })
+	if err != nil {
+		t.Fatalf("Watch: %v", err)
+	}
+	if warned != 1 {
+		t.Errorf("warnings = %d, want 1", warned)
+	}
+	if res.Reason != "PR merged" {
+		t.Errorf("reason = %q, want %q", res.Reason, "PR merged")
+	}
+}
+
+// Anonymous watching of a public repo must poll on without a credential in
+// sight: no token, no mint, no exit until something actually changes.
+func TestWatchAnonymousKeepsPolling(t *testing.T) {
+	open := base()
+	f := &fakeFetcher{states: []PRState{open}}
+
+	ticks := make(chan time.Time, 2)
+	ticks <- time.Now()
+	ticks <- time.Now()
+	close(ticks)
+	res, err := Watch(f, []PRRef{open.Ref}, "unkin-agent", ticks, nil,
+		func(_ PRRef, e error) { t.Errorf("unexpected poll error: %v", e) })
+	if err != nil {
+		t.Fatalf("Watch: %v", err)
+	}
+	if res.Reason != "" {
+		t.Errorf("reason = %q, want no change reported", res.Reason)
+	}
+	if f.calls != 3 {
+		t.Errorf("fetch calls = %d, want 3 (baseline + two polls)", f.calls)
+	}
+}