diff --git a/agent/rpc/auth_interceptor.go b/agent/rpc/auth_interceptor.go index 17aa2a587..4b4ec2a70 100644 --- a/agent/rpc/auth_interceptor.go +++ b/agent/rpc/auth_interceptor.go @@ -16,16 +16,20 @@ package rpc import ( "context" + "sync" "time" "github.com/rs/zerolog/log" "google.golang.org/grpc" + "google.golang.org/grpc/codes" "google.golang.org/grpc/metadata" + "google.golang.org/grpc/status" ) // AuthInterceptor is a client interceptor for authentication. type AuthInterceptor struct { authClient *AuthClient + mu sync.RWMutex accessToken string } @@ -43,7 +47,9 @@ func NewAuthInterceptor(ctx context.Context, authClient *AuthClient, refreshDura return interceptor, nil } -// Unary returns a client interceptor to authenticate unary RPC. +// Unary returns a client interceptor to authenticate unary RPC. If the server +// rejects the attached token, it refreshes the token before returning the +// error so the caller's retry policy can repeat the call. func (interceptor *AuthInterceptor) Unary() grpc.UnaryClientInterceptor { return func( ctx context.Context, @@ -53,7 +59,15 @@ func (interceptor *AuthInterceptor) Unary() grpc.UnaryClientInterceptor { invoker grpc.UnaryInvoker, opts ...grpc.CallOption, ) error { - return invoker(interceptor.attachToken(ctx), method, req, reply, cc, opts...) + callCtx, rejectedToken := interceptor.attachToken(ctx) + err := invoker(callCtx, method, req, reply, cc, opts...) + if status.Code(err) == codes.Unauthenticated { + refreshErr := interceptor.refreshTokenAfterUnauthenticated(ctx, rejectedToken) + if refreshErr != nil { + log.Warn().Err(refreshErr).Msg("could not reauthenticate after the server rejected the gRPC token") + } + } + return err } } @@ -67,12 +81,17 @@ func (interceptor *AuthInterceptor) Stream() grpc.StreamClientInterceptor { streamer grpc.Streamer, opts ...grpc.CallOption, ) (grpc.ClientStream, error) { - return streamer(interceptor.attachToken(ctx), desc, cc, method, opts...) + callCtx, _ := interceptor.attachToken(ctx) + return streamer(callCtx, desc, cc, method, opts...) } } -func (interceptor *AuthInterceptor) attachToken(ctx context.Context) context.Context { - return metadata.AppendToOutgoingContext(ctx, "token", interceptor.accessToken) +func (interceptor *AuthInterceptor) attachToken(ctx context.Context) (context.Context, string) { + interceptor.mu.RLock() + accessToken := interceptor.accessToken + interceptor.mu.RUnlock() + + return metadata.AppendToOutgoingContext(ctx, "token", accessToken), accessToken } func (interceptor *AuthInterceptor) scheduleRefreshToken(ctx context.Context, refreshInterval time.Duration) error { @@ -103,6 +122,26 @@ func (interceptor *AuthInterceptor) scheduleRefreshToken(ctx context.Context, re } func (interceptor *AuthInterceptor) refreshToken(ctx context.Context) error { + interceptor.mu.Lock() + defer interceptor.mu.Unlock() + + return interceptor.refreshTokenLocked(ctx) +} + +func (interceptor *AuthInterceptor) refreshTokenAfterUnauthenticated(ctx context.Context, rejectedToken string) error { + interceptor.mu.Lock() + defer interceptor.mu.Unlock() + + // Another rejected RPC or the refresh timer may already have replaced the + // token while this call was in flight. Reuse that refresh when it did. + if interceptor.accessToken != rejectedToken { + return nil + } + + return interceptor.refreshTokenLocked(ctx) +} + +func (interceptor *AuthInterceptor) refreshTokenLocked(ctx context.Context) error { accessToken, _, err := interceptor.authClient.Auth(ctx) if err != nil { return err diff --git a/agent/rpc/auth_interceptor_test.go b/agent/rpc/auth_interceptor_test.go index 232b9016c..f3437da4d 100644 --- a/agent/rpc/auth_interceptor_test.go +++ b/agent/rpc/auth_interceptor_test.go @@ -16,39 +16,150 @@ package rpc import ( "context" + "errors" + "sync" + "sync/atomic" "testing" + "time" "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "google.golang.org/grpc" + "google.golang.org/grpc/codes" "google.golang.org/grpc/metadata" + "google.golang.org/grpc/status" + + "go.woodpecker-ci.org/woodpecker/v3/rpc/proto" ) +type authClientFunc func(context.Context, *proto.AuthRequest, ...grpc.CallOption) (*proto.AuthResponse, error) + +func (f authClientFunc) Auth(ctx context.Context, req *proto.AuthRequest, opts ...grpc.CallOption) (*proto.AuthResponse, error) { + return f(ctx, req, opts...) +} + +func testAuthInterceptor(token string, auth authClientFunc) *AuthInterceptor { + return &AuthInterceptor{ + authClient: &AuthClient{client: auth}, + accessToken: token, + } +} + func TestAuthInterceptorAttachToken(t *testing.T) { - tc := []struct { - name string - token string - }{ - {"populated token", "secret-token"}, - {"empty token", ""}, + t.Parallel() + + interceptor := &AuthInterceptor{accessToken: "token"} + base := metadata.AppendToOutgoingContext(context.Background(), "extra", "value") + ctx, token := interceptor.attachToken(base) + + md, ok := metadata.FromOutgoingContext(ctx) + require.True(t, ok) + assert.Equal(t, []string{"token"}, md.Get("token")) + assert.Equal(t, []string{"value"}, md.Get("extra")) + assert.Equal(t, "token", token) +} + +func TestAuthInterceptorRefreshesRejectedTokenForCallerRetry(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + wantDeadline, _ := ctx.Deadline() + + var authCalls int + interceptor := testAuthInterceptor("old-token", func(ctx context.Context, _ *proto.AuthRequest, _ ...grpc.CallOption) (*proto.AuthResponse, error) { + authCalls++ + deadline, ok := ctx.Deadline() + require.True(t, ok) + assert.Equal(t, wantDeadline, deadline) + return &proto.AuthResponse{AccessToken: "new-token"}, nil + }) + unauthenticatedErr := status.Error(codes.Unauthenticated, "expired token") + + var tokens []string + invoker := func(ctx context.Context, _ string, _, _ any, _ *grpc.ClientConn, _ ...grpc.CallOption) error { + md, ok := metadata.FromOutgoingContext(ctx) + require.True(t, ok) + tokens = append(tokens, md.Get("token")[0]) + if len(tokens) == 1 { + return unauthenticatedErr + } + return nil } - for _, c := range tc { - t.Run(c.name, func(t *testing.T) { - interceptor := &AuthInterceptor{accessToken: c.token} - ctx := interceptor.attachToken(context.Background()) + _, err := retryRPC(ctx, &client{connectionRetryTimeout: time.Second}, "test", func() (struct{}, error) { + err := interceptor.Unary()(ctx, "/proto.Woodpecker/Next", nil, nil, nil, invoker) + return struct{}{}, classifyRPCErr(ctx, err) + }) - md, ok := metadata.FromOutgoingContext(ctx) - assert.True(t, ok) - assert.Equal(t, []string{c.token}, md.Get("token")) + require.NoError(t, err) + assert.Equal(t, 1, authCalls) + assert.Equal(t, []string{"old-token", "new-token"}, tokens) +} + +func TestAuthInterceptorPreservesErrors(t *testing.T) { + t.Run("non-authentication error", func(t *testing.T) { + var authCalls int + interceptor := testAuthInterceptor("token", func(context.Context, *proto.AuthRequest, ...grpc.CallOption) (*proto.AuthResponse, error) { + authCalls++ + return &proto.AuthResponse{AccessToken: "new-token"}, nil }) - } + permissionErr := status.Error(codes.PermissionDenied, "denied") - t.Run("preserves existing metadata", func(t *testing.T) { - base := metadata.AppendToOutgoingContext(context.Background(), "extra", "v") - interceptor := &AuthInterceptor{accessToken: "tok"} - ctx := interceptor.attachToken(base) + err := interceptor.Unary()( + context.Background(), "/proto.Woodpecker/Next", nil, nil, nil, + func(context.Context, string, any, any, *grpc.ClientConn, ...grpc.CallOption) error { + return permissionErr + }, + ) - md, _ := metadata.FromOutgoingContext(ctx) - assert.Equal(t, []string{"tok"}, md.Get("token")) - assert.Equal(t, []string{"v"}, md.Get("extra")) + assert.Equal(t, permissionErr, err) + assert.Zero(t, authCalls) + }) + + t.Run("failed reauthentication", func(t *testing.T) { + authErr := errors.New("authentication unavailable") + interceptor := testAuthInterceptor("old-token", func(context.Context, *proto.AuthRequest, ...grpc.CallOption) (*proto.AuthResponse, error) { + return nil, authErr + }) + unauthenticatedErr := status.Error(codes.Unauthenticated, "expired token") + + err := interceptor.Unary()( + context.Background(), "/proto.Woodpecker/Next", nil, nil, nil, + func(context.Context, string, any, any, *grpc.ClientConn, ...grpc.CallOption) error { + return unauthenticatedErr + }, + ) + + assert.Equal(t, unauthenticatedErr, err) }) } + +func TestAuthInterceptorCoalescesConcurrentRefresh(t *testing.T) { + var authCalls atomic.Int32 + interceptor := testAuthInterceptor("old-token", func(context.Context, *proto.AuthRequest, ...grpc.CallOption) (*proto.AuthResponse, error) { + authCalls.Add(1) + return &proto.AuthResponse{AccessToken: "new-token"}, nil + }) + + const callers = 2 + start := make(chan struct{}) + errs := make(chan error, callers) + var wg sync.WaitGroup + for range callers { + wg.Add(1) + go func() { + defer wg.Done() + <-start + errs <- interceptor.refreshTokenAfterUnauthenticated(context.Background(), "old-token") + }() + } + close(start) + wg.Wait() + close(errs) + + for err := range errs { + require.NoError(t, err) + } + _, token := interceptor.attachToken(context.Background()) + assert.Equal(t, "new-token", token) + assert.Equal(t, int32(1), authCalls.Load()) +} diff --git a/agent/rpc/client_grpc.go b/agent/rpc/client_grpc.go index 9ebf61958..99bfaa605 100644 --- a/agent/rpc/client_grpc.go +++ b/agent/rpc/client_grpc.go @@ -202,6 +202,7 @@ func classifyRPCErr(ctx context.Context, err error) error { case codes.Aborted, codes.DataLoss, codes.DeadlineExceeded, + codes.Unauthenticated, codes.Unavailable: return err default: diff --git a/agent/rpc/client_grpc_test.go b/agent/rpc/client_grpc_test.go index 1136cefdd..00982aeda 100644 --- a/agent/rpc/client_grpc_test.go +++ b/agent/rpc/client_grpc_test.go @@ -15,10 +15,16 @@ package rpc import ( + "context" + "errors" "testing" "time" + "github.com/cenkalti/backoff/v7" "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" ) func TestSetConnectionRetryTimeout(t *testing.T) { @@ -52,3 +58,50 @@ func TestIsConnected(t *testing.T) { assert.False(t, cl.IsConnected()) }) } + +func TestClassifyRPCErrUnauthenticatedIsRetryable(t *testing.T) { + t.Parallel() + + err := status.Error(codes.Unauthenticated, "expired token") + classified := classifyRPCErr(context.Background(), err) + + assert.Equal(t, codes.Unauthenticated, status.Code(classified)) + assert.False(t, errors.Is(classified, backoff.ErrPermanent)) +} + +func TestRetryRPCUnauthenticatedHonorsFiniteTimeout(t *testing.T) { + t.Parallel() + + ctx := context.Background() + c := &client{connectionRetryTimeout: time.Nanosecond} + var attempts int + + _, err := retryRPC(ctx, c, "test", func() (struct{}, error) { + attempts++ + return struct{}{}, classifyRPCErr(ctx, status.Error(codes.Unauthenticated, "expired token")) + }) + + require.Error(t, err) + assert.ErrorIs(t, err, backoff.ErrMaxElapsedTime) + assert.Equal(t, 1, attempts) +} + +func TestRetryRPCUnauthenticatedRetriesUntilContextCancellation(t *testing.T) { + t.Parallel() + + ctx, cancel := context.WithCancelCause(context.Background()) + defer cancel(nil) + c := &client{connectionRetryTimeout: 0} + var attempts int + + _, err := retryRPC(ctx, c, "test", func() (struct{}, error) { + attempts++ + if attempts == 3 { + cancel(nil) + } + return struct{}{}, classifyRPCErr(ctx, status.Error(codes.Unauthenticated, "expired token")) + }) + + require.NoError(t, err) + assert.Equal(t, 3, attempts) +} diff --git a/cmd/server/flags.go b/cmd/server/flags.go index bb1bbffdb..76e56d117 100644 --- a/cmd/server/flags.go +++ b/cmd/server/flags.go @@ -129,7 +129,7 @@ var flags = append([]cli.Flag{ ), Name: "grpc-secret", Usage: "grpc jwt secret", - Value: "secret", + Value: "", Config: cli.StringConfig{ TrimSpace: true, }, diff --git a/cmd/server/grpc_server.go b/cmd/server/grpc_server.go index 017d27d82..b408e513b 100644 --- a/cmd/server/grpc_server.go +++ b/cmd/server/grpc_server.go @@ -23,6 +23,7 @@ import ( "strings" "github.com/prometheus/client_golang/prometheus" + "github.com/rs/zerolog/log" "github.com/urfave/cli/v3" "go.woodpecker-ci.org/woodpecker/v3/server" @@ -33,6 +34,13 @@ import ( func runGrpcServer(ctx context.Context, c *cli.Command, _store store.Store) error { network := "tcp" addr := c.String("grpc-addr") + jwtSecret, generated := setupGrpcSecret(c.String("grpc-secret")) + if generated { + log.Warn().Msg( + "WOODPECKER_GRPC_SECRET is not set; generated a temporary random secret. " + + "Set and persist WOODPECKER_GRPC_SECRET to keep the same secret across restarts", + ) + } if strings.HasPrefix(addr, "unix://") { network = "unix" @@ -52,7 +60,7 @@ func runGrpcServer(ctx context.Context, c *cli.Command, _store store.Store) erro Store: _store, Scheduler: server.Config.Services.Scheduler, Logger: server.Config.Services.Logs, - JWTSecret: c.String("grpc-secret"), + JWTSecret: jwtSecret, AgentToken: server.Config.Server.AgentToken, KeepaliveMinTime: c.Duration("keepalive-min-time"), Registerer: prometheus.DefaultRegisterer, diff --git a/cmd/server/setup.go b/cmd/server/setup.go index a1560d7d3..c1909e02b 100644 --- a/cmd/server/setup.go +++ b/cmd/server/setup.go @@ -130,6 +130,14 @@ func setupLogStore(c *cli.Command, s store.Store) (service_log.Service, error) { const jwtSecretID = "jwt-secret" +func setupGrpcSecret(secret string) (string, bool) { + if secret != "" { + return secret, false + } + + return base32.StdEncoding.EncodeToString(random.GetRandomBytes(32)), true +} + func setupJWTSecret(_store store.Store) (string, error) { jwtSecret, err := _store.ServerConfigGet(jwtSecretID) if errors.Is(err, types.ErrRecordNotExist) { diff --git a/cmd/server/setup_test.go b/cmd/server/setup_test.go index 898372b7d..c95fabb99 100644 --- a/cmd/server/setup_test.go +++ b/cmd/server/setup_test.go @@ -15,6 +15,7 @@ package main import ( + "encoding/base32" "errors" "os" "path/filepath" @@ -106,3 +107,32 @@ func TestSetupJWTSecret(t *testing.T) { assert.Empty(t, secret) }) } + +func TestSetupGrpcSecret(t *testing.T) { + t.Parallel() + + t.Run("returns configured secret", func(t *testing.T) { + t.Parallel() + + secret, generated := setupGrpcSecret("configured-secret") + + assert.Equal(t, "configured-secret", secret) + assert.False(t, generated) + }) + + t.Run("generates random secret when not configured", func(t *testing.T) { + t.Parallel() + + secretA, generatedA := setupGrpcSecret("") + secretB, generatedB := setupGrpcSecret("") + + assert.True(t, generatedA) + assert.True(t, generatedB) + assert.NotEmpty(t, secretA) + assert.NotEqual(t, secretA, secretB) + + decoded, err := base32.StdEncoding.DecodeString(secretA) + require.NoError(t, err) + assert.Len(t, decoded, 32) + }) +} diff --git a/docs/docs/30-administration/10-configuration/10-server.md b/docs/docs/30-administration/10-configuration/10-server.md index d6bb942ba..1f95760df 100644 --- a/docs/docs/30-administration/10-configuration/10-server.md +++ b/docs/docs/30-administration/10-configuration/10-server.md @@ -651,9 +651,19 @@ If you want an unix socket use `unix://` prefix, for example `unix:///run/woodpe ### GRPC_SECRET - Name: `WOODPECKER_GRPC_SECRET` -- Default: `secret` +- Default: none -Configures the gRPC JWT secret. +Configures the secret used to sign JWTs for gRPC connections. + +If this setting is empty, the server generates a secure temporary secret and logs a warning. The generated secret is not persisted and changes each time the server starts. Configure and persist a secret to keep it stable across restarts. Setting this explicitly is important for high availability (HA) setups with multiple server replicas: each replica would otherwise generate its own secret and reject the gRPC tokens issued by the other replicas. Generate a secure secret with: + +```shell +openssl rand -hex 32 +``` + +Store the generated value securely and provide it through `WOODPECKER_GRPC_SECRET` or `WOODPECKER_GRPC_SECRET_FILE`. + +After this secret is rotated, connected agents reauthenticate when the server rejects their old token. Agents do not need to be restarted as long as `WOODPECKER_AGENT_SECRET` remains unchanged. --- @@ -662,7 +672,7 @@ Configures the gRPC JWT secret. - Name: `WOODPECKER_GRPC_SECRET_FILE` - Default: none -Read the value for `WOODPECKER_GRPC_SECRET` from the specified filepath. +Read the value for `WOODPECKER_GRPC_SECRET` from the specified filepath. The file should be stored persistently and only be readable by the Woodpecker server. ---