feat(github): add support for refresh tokens and token management

backend/helpers/pluginhelper/api/api_client.go

@@ -299,6 +299,11 @@ func (apiClient *ApiClient) SetLogger(logger log.Logger) {
 	apiClient.logger = logger
 }
 
+// GetClient returns the underlying http.Client
+func (apiClient *ApiClient) GetClient() *http.Client {
+	return apiClient.client
+}
+
 func (apiClient *ApiClient) logDebug(format string, a ...interface{}) {
 	if apiClient.logger != nil {
 		apiClient.logger.Debug(format, a...)

backend/plugins/github/models/connection.go

@@ -56,6 +56,21 @@ type GithubConn struct {
 	helper.MultiAuth      `mapstructure:",squash"`
 	GithubAccessToken     `mapstructure:",squash" authMethod:"AccessToken"`
 	GithubAppKey          `mapstructure:",squash" authMethod:"AppKey"`
+	RefreshToken          string    `mapstructure:"refreshToken" json:"refreshToken" gorm:"type:text;serializer:encdec"`
+	TokenExpiresAt        time.Time `mapstructure:"tokenExpiresAt" json:"tokenExpiresAt"`
+	RefreshTokenExpiresAt time.Time `mapstructure:"refreshTokenExpiresAt" json:"refreshTokenExpiresAt"`
+}
+
+// UpdateToken updates the token and refresh token information
+func (conn *GithubConn) UpdateToken(newToken, newRefreshToken string, expiry, refreshExpiry time.Time) {
+	conn.Token = newToken
+	conn.RefreshToken = newRefreshToken
+	conn.TokenExpiresAt = expiry
+	conn.RefreshTokenExpiresAt = refreshExpiry
+
+	// Update the internal tokens slice used by SetupAuthentication
+	conn.tokens = []string{newToken}
+	conn.tokenIndex = 0
 }
 
 // PrepareApiClient splits Token to tokens for SetupAuthentication to utilize
@@ -249,7 +264,7 @@ func (conn *GithubConn) typeIs(token string) string {
 	// total len is 40, {prefix}{showPrefix}{secret}{showSuffix}
 	// fine-grained tokens
 	// github_pat_{82_characters}
-	classicalTokenClassicalPrefixes := []string{"ghp_", "gho_", "ghs_", "ghr_"}
+	classicalTokenClassicalPrefixes := []string{"ghp_", "gho_", "ghs_", "ghr_", "ghu_"}
 	classicalTokenFindGrainedPrefixes := []string{"github_pat_"}
 	for _, prefix := range classicalTokenClassicalPrefixes {
 		if strings.HasPrefix(token, prefix) {

backend/plugins/github/models/connection_test.go

@@ -227,3 +227,14 @@ func TestGithubConnection_Sanitize(t *testing.T) {
 		})
 	}
 }
+
+func TestIsUserToServerToken(t *testing.T) {
+	conn := &GithubConn{}
+	assert.Equal(t, GithubTokenTypeClassical, conn.typeIs("ghp_123"))
+	assert.Equal(t, GithubTokenTypeClassical, conn.typeIs("gho_123"))
+	assert.Equal(t, GithubTokenTypeClassical, conn.typeIs("ghu_123"))
+	assert.Equal(t, GithubTokenTypeClassical, conn.typeIs("ghs_123"))
+	assert.Equal(t, GithubTokenTypeClassical, conn.typeIs("ghr_123"))
+	assert.Equal(t, GithubTokenTypeFineGrained, conn.typeIs("github_pat_123"))
+	assert.Equal(t, GithubTokenTypeUnknown, conn.typeIs("some_other_token"))
+}

backend/plugins/github/models/migrationscripts/20241120_add_refresh_token_fields.go

@@ -0,0 +1,42 @@
+/*
+Licensed to the Apache Software Foundation (ASF) under one or more
+contributor license agreements.  See the NOTICE file distributed with
+this work for additional information regarding copyright ownership.
+The ASF licenses this file to You under the Apache License, Version 2.0
+(the "License"); you may not use this file except in compliance with
+the License.  You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package migrationscripts
+
+import (
+	"github.com/apache/incubator-devlake/core/context"
+	"github.com/apache/incubator-devlake/core/errors"
+	"github.com/apache/incubator-devlake/helpers/migrationhelper"
+	"github.com/apache/incubator-devlake/plugins/github/models"
+)
+
+type addRefreshTokenFields struct{}
+
+func (*addRefreshTokenFields) Up(basicRes context.BasicRes) errors.Error {
+	return migrationhelper.AutoMigrateTables(
+		basicRes,
+		&models.GithubConnection{},
+	)
+}
+
+func (*addRefreshTokenFields) Version() uint64 {
+	return 20241120000001
+}
+
+func (*addRefreshTokenFields) Name() string {
+	return "add refresh token fields to github_connections"
+}

backend/plugins/github/models/migrationscripts/register.go

@@ -55,5 +55,6 @@ func All() []plugin.MigrationScript {
 		new(addIsDraftToPr),
 		new(changeIssueComponentType),
 		new(addIndexToGithubJobs),
+		new(addRefreshTokenFields),
 	}
 }

backend/plugins/github/tasks/api_client.go

@@ -26,6 +26,7 @@ import (
 	"github.com/apache/incubator-devlake/core/plugin"
 	"github.com/apache/incubator-devlake/helpers/pluginhelper/api"
 	"github.com/apache/incubator-devlake/plugins/github/models"
+	"github.com/apache/incubator-devlake/plugins/github/token"
 )
 
 func CreateApiClient(taskCtx plugin.TaskContext, connection *models.GithubConnection) (*api.ApiAsyncClient, errors.Error) {
@@ -34,6 +35,24 @@ func CreateApiClient(taskCtx plugin.TaskContext, connection *models.GithubConnec
 		return nil, err
 	}
 
+	// Inject TokenProvider if refresh token is present
+	if connection.RefreshToken != "" {
+		logger := taskCtx.GetLogger()
+		db := taskCtx.GetDal()
+
+		// Create TokenProvider
+		tp := token.NewTokenProvider(connection, db, apiClient.GetClient(), logger)
+
+		// Wrap the transport
+		baseTransport := apiClient.GetClient().Transport
+		if baseTransport == nil {
+			baseTransport = http.DefaultTransport
+		}
+
+		rt := token.NewRefreshRoundTripper(baseTransport, tp)
+		apiClient.GetClient().Transport = rt
+	}
+
 	// create rate limit calculator
 	rateLimiter := &api.ApiRateLimitCalculator{
 		UserRateLimitPerHour: connection.RateLimitPerHour,

backend/plugins/github/token/round_tripper.go

@@ -0,0 +1,76 @@
+/*
+Licensed to the Apache Software Foundation (ASF) under one or more
+contributor license agreements.  See the NOTICE file distributed with
+this work for additional information regarding copyright ownership.
+The ASF licenses this file to You under the Apache License, Version 2.0
+(the "License"); you may not use this file except in compliance with
+the License.  You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package token
+
+import (
+	"net/http"
+)
+
+type RefreshRoundTripper struct {
+	base          http.RoundTripper
+	tokenProvider *TokenProvider
+}
+
+func NewRefreshRoundTripper(base http.RoundTripper, tp *TokenProvider) *RefreshRoundTripper {
+	return &RefreshRoundTripper{
+		base:          base,
+		tokenProvider: tp,
+	}
+}
+
+func (rt *RefreshRoundTripper) RoundTrip(req *http.Request) (*http.Response, error) {
+	// Get token
+	token, err := rt.tokenProvider.GetToken()
+	if err != nil {
+		return nil, err
+	}
+
+	// Clone request before modifying
+	reqClone := req.Clone(req.Context())
+	reqClone.Header.Set("Authorization", "Bearer "+token)
+
+	// Execute request
+	resp, reqErr := rt.base.RoundTrip(reqClone)
+	if reqErr != nil {
+		return nil, reqErr
+	}
+
+	// Reactive refresh on 401
+	if resp.StatusCode == http.StatusUnauthorized {
+		// Close previous response body
+		resp.Body.Close()
+
+		// Force refresh
+		if err := rt.tokenProvider.ForceRefresh(token); err != nil {
+			return nil, err
+		}
+
+		// Get new token
+		newToken, err := rt.tokenProvider.GetToken()
+		if err != nil {
+			return nil, err
+		}
+
+		// Retry request with new token
+		reqRetry := req.Clone(req.Context())
+		reqRetry.Header.Set("Authorization", "Bearer "+newToken)
+		return rt.base.RoundTrip(reqRetry)
+	}
+
+	return resp, nil
+}

backend/plugins/github/token/token_provider.go

@@ -0,0 +1,171 @@
+/*
+Licensed to the Apache Software Foundation (ASF) under one or more
+contributor license agreements.  See the NOTICE file distributed with
+this work for additional information regarding copyright ownership.
+The ASF licenses this file to You under the Apache License, Version 2.0
+(the "License"); you may not use this file except in compliance with
+the License.  You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package token
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"os"
+	"strconv"
+	"sync"
+	"time"
+
+	"github.com/apache/incubator-devlake/core/dal"
+	"github.com/apache/incubator-devlake/core/errors"
+	"github.com/apache/incubator-devlake/core/log"
+	"github.com/apache/incubator-devlake/plugins/github/models"
+)
+
+const (
+	DefaultRefreshBuffer = 5 * time.Minute
+)
+
+type TokenProvider struct {
+	conn       *models.GithubConnection
+	dal        dal.Dal
+	httpClient *http.Client
+	logger     log.Logger
+	mu         sync.Mutex
+	refreshURL string
+}
+
+func NewTokenProvider(conn *models.GithubConnection, d dal.Dal, client *http.Client, logger log.Logger) *TokenProvider {
+	return &TokenProvider{
+		conn:       conn,
+		dal:        d,
+		httpClient: client,
+		logger:     logger,
+		refreshURL: "https://github.com/login/oauth/access_token",
+	}
+}
+
+func (tp *TokenProvider) GetToken() (string, errors.Error) {
+	tp.mu.Lock()
+	defer tp.mu.Unlock()
+
+	if tp.needsRefresh() {
+		if err := tp.refreshToken(); err != nil {
+			return "", err
+		}
+	}
+	return tp.conn.Token, nil
+}
+
+func (tp *TokenProvider) needsRefresh() bool {
+	if tp.conn.RefreshToken == "" {
+		return false
+	}
+
+	buffer := DefaultRefreshBuffer
+	if envBuffer := os.Getenv("GITHUB_TOKEN_REFRESH_BUFFER_MINUTES"); envBuffer != "" {
+		if val, err := strconv.Atoi(envBuffer); err == nil {
+			buffer = time.Duration(val) * time.Minute
+		}
+	}
+
+	return time.Now().Add(buffer).After(tp.conn.TokenExpiresAt)
+}
+
+func (tp *TokenProvider) refreshToken() errors.Error {
+	tp.logger.Info("Refreshing GitHub token for connection %d", tp.conn.ID)
+
+	data := map[string]string{
+		"refresh_token": tp.conn.RefreshToken,
+		"grant_type":    "refresh_token",
+		"client_id":     tp.conn.AppId,
+		"client_secret": tp.conn.SecretKey,
+	}
+	jsonData, _ := json.Marshal(data)
+
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+
+	req, err := http.NewRequestWithContext(ctx, "POST", tp.refreshURL, bytes.NewBuffer(jsonData))
+	if err != nil {
+		return errors.Convert(err)
+	}
+	req.Header.Set("Content-Type", "application/json")
+	req.Header.Set("Accept", "application/json")
+
+	resp, err := tp.httpClient.Do(req)
+	if err != nil {
+		return errors.Convert(err)
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode != http.StatusOK {
+		return errors.Default.New(fmt.Sprintf("failed to refresh token: %d", resp.StatusCode))
+	}
+
+	body, err := io.ReadAll(resp.Body)
+	if err != nil {
+		return errors.Convert(err)
+	}
+
+	var result struct {
+		AccessToken           string `json:"access_token"`
+		RefreshToken          string `json:"refresh_token"`
+		ExpiresIn             int    `json:"expires_in"`
+		RefreshTokenExpiresIn int    `json:"refresh_token_expires_in"`
+	}
+	if err := json.Unmarshal(body, &result); err != nil {
+		return errors.Convert(err)
+	}
+
+	if result.AccessToken == "" {
+		return errors.Default.New("empty access token returned")
+	}
+
+	tp.conn.UpdateToken(
+		result.AccessToken,
+		result.RefreshToken,
+		time.Now().Add(time.Duration(result.ExpiresIn)*time.Second),
+		time.Now().Add(time.Duration(result.RefreshTokenExpiresIn)*time.Second),
+	)
+
+	if tp.dal != nil {
+		err := tp.dal.UpdateColumns(tp.conn, []dal.DalSet{
+			{ColumnName: "token", Value: tp.conn.Token},
+			{ColumnName: "refresh_token", Value: tp.conn.RefreshToken},
+			{ColumnName: "token_expires_at", Value: tp.conn.TokenExpiresAt},
+			{ColumnName: "refresh_token_expires_at", Value: tp.conn.RefreshTokenExpiresAt},
+		})
+		if err != nil {
+			tp.logger.Warn(err, "failed to persist refreshed token")
+		}
+	}
+
+	return nil
+}
+
+func (tp *TokenProvider) ForceRefresh(oldToken string) errors.Error {
+	tp.mu.Lock()
+	defer tp.mu.Unlock()
+
+	// If the token has changed since the request was made, it means another thread
+	// has already refreshed it.
+	if tp.conn.Token != oldToken {
+		return nil
+	}
+
+	return tp.refreshToken()
+}