Merging to release-4-lts: [TT-10189/TT-10467] Add OAuthPurgeLapsedTok…

…ens (#5766) (#5901)

[TT-10189/TT-10467] Add OAuthPurgeLapsedTokens (#5766)

<!-- Provide a general summary of your changes in the Title above -->

## Description

This PR adds an event `OAuthPurgeLapsedTokens`, which upon receiving
would delete all lapsed OAuth tokens.
It also adds endpoint `DELETE /tyk/oauth/tokens?scope=lapsed` to purge
lapsed OAuth tokens synchronously.
What are lapsed OAuth tokens? Lapsed OAuth tokens that are expired and
past `oauth_token_expired_retain_period` configured in gateway config.

## Related Issue
Sub Task: https://tyktech.atlassian.net/browse/TT-10467
Parent Ticket: https://tyktech.atlassian.net/browse/TT-10189

## Motivation and Context

<!-- Why is this change required? What problem does it solve? -->

## How This Has Been Tested

<!-- Please describe in detail how you tested your changes -->
<!-- Include details of your testing environment, and the tests -->
<!-- you ran to see how your change affects other areas of the code,
etc. -->
<!-- This information is helpful for reviewers and QA. -->

## Screenshots (if appropriate)

## Types of changes

<!-- What types of changes does your code introduce? Put an `x` in all
the boxes that apply: -->

- [ ] Bug fix (non-breaking change which fixes an issue)
- [ ] New feature (non-breaking change which adds functionality)
- [ ] Breaking change (fix or feature that would cause existing
functionality to change)
- [ ] Refactoring or add test (improvements in base code or adds test
coverage to functionality)

## Checklist

<!-- Go over all the following points, and put an `x` in all the boxes
that apply -->
<!-- If there are no documentation updates required, mark the item as
checked. -->
<!-- Raise up any additional concerns not covered by the checklist. -->

- [ ] I ensured that the documentation is up to date
- [ ] I explained why this PR updates go.mod in detail with reasoning
why it's required
- [ ] I would like a code coverage CI quality gate exception and have
explained why

---------

Co-authored-by: dcs3spp <dcs3spp@users.noreply.github.com>

---------

Co-authored-by: Jeffy Mathew <jeffy.mathew100@gmail.com>

gateway/api.go

@@ -34,7 +34,6 @@ import (
 	"fmt"
 	"io/ioutil"
 	"net/http"
-	"net/url"
 	"os"
 	"path/filepath"
 	"strconv"
@@ -51,6 +50,7 @@ import (
 	"github.com/TykTechnologies/tyk/apidef"
 	"github.com/TykTechnologies/tyk/ctx"
 	"github.com/TykTechnologies/tyk/headers"
+	"github.com/TykTechnologies/tyk/internal/url"
 	"github.com/TykTechnologies/tyk/storage"
 	"github.com/TykTechnologies/tyk/user"
 
@@ -59,6 +59,14 @@ import (
 	"github.com/TykTechnologies/tyk/internal/uuid"
 )
 
+const (
+	oAuthClientTokensKeyPattern = "oauth-data.*oauth-client-tokens.*"
+)
+
+var (
+	ErrRequestMalformed = errors.New("request malformed")
+)
+
 // apiModifyKeySuccess represents when a Key modification was successful
 //
 // swagger:model apiModifyKeySuccess
@@ -2044,6 +2052,26 @@ func (gw *Gateway) getOauthClientDetails(keyName, apiID string) (interface{}, in
 	return reportableClientData, http.StatusOK
 }
 
+func (gw *Gateway) oAuthTokensHandler(w http.ResponseWriter, r *http.Request) {
+	if !url.QueryHas(r.URL.Query(), "scope") {
+		doJSONWrite(w, http.StatusUnprocessableEntity, apiError("scope parameter is required"))
+		return
+	}
+
+	if r.URL.Query().Get("scope") != "lapsed" {
+		doJSONWrite(w, http.StatusBadRequest, apiError("unknown scope"))
+		return
+	}
+
+	err := gw.purgeLapsedOAuthTokens()
+	if err != nil {
+		doJSONWrite(w, http.StatusInternalServerError, apiError("error purging lapsed tokens"))
+		return
+	}
+
+	doJSONWrite(w, http.StatusOK, apiOk("lapsed tokens purged"))
+}
+
 // Delete Client
 func (gw *Gateway) handleDeleteOAuthClient(keyName, apiID string) (interface{}, int) {
 	storageID := oauthClientStorageID(keyName)

gateway/api_test.go

@@ -2042,3 +2042,71 @@ func TestOrgKeyHandler_LastUpdated(t *testing.T) {
 		}},
 	}...)
 }
+
+func TestPurgeOAuthClientTokens(t *testing.T) {
+	conf := func(globalConf *config.Config) {
+		// set tokens to be expired after 1 second
+		globalConf.OauthTokenExpire = 1
+		// cleanup tokens older than 2 seconds
+		globalConf.OauthTokenExpiredRetainPeriod = 2
+	}
+
+	ts := StartTest(conf)
+	defer ts.Close()
+
+	t.Run("scope validation", func(t *testing.T) {
+		ts.Run(t, []test.TestCase{
+			{
+				AdminAuth: true,
+				Path:      "/tyk/oauth/tokens/",
+				Method:    http.MethodDelete,
+				Code:      http.StatusUnprocessableEntity,
+			},
+			{
+				AdminAuth:   true,
+				Path:        "/tyk/oauth/tokens/",
+				QueryParams: map[string]string{"scope": "expired"},
+				Method:      http.MethodDelete,
+				Code:        http.StatusBadRequest,
+			},
+		}...)
+	})
+
+	assertTokensLen := func(t *testing.T, storageManager storage.Handler, storageKey string, expectedTokensLen int) {
+		nowTs := time.Now().Unix()
+		startScore := strconv.FormatInt(nowTs, 10)
+		tokens, _, err := storageManager.GetSortedSetRange(storageKey, startScore, "+inf")
+		assert.NoError(t, err)
+		assert.Equal(t, expectedTokensLen, len(tokens))
+	}
+
+	t.Run("scope=lapsed", func(t *testing.T) {
+		spec := ts.LoadTestOAuthSpec()
+
+		clientID1, clientID2 := uuid.New(), uuid.New()
+
+		ts.createOAuthClientIDAndTokens(t, spec, clientID1)
+		ts.createOAuthClientIDAndTokens(t, spec, clientID2)
+		storageKey1, storageKey2 := fmt.Sprintf("%s%s", prefixClientTokens, clientID1),
+			fmt.Sprintf("%s%s", prefixClientTokens, clientID2)
+
+		storageManager := ts.Gw.getGlobalStorageHandler(generateOAuthPrefix(spec.APIID), false)
+		storageManager.Connect()
+
+		assertTokensLen(t, storageManager, storageKey1, 3)
+		assertTokensLen(t, storageManager, storageKey2, 3)
+
+		time.Sleep(time.Second * 3)
+		ts.Run(t, test.TestCase{
+			ControlRequest: true,
+			AdminAuth:      true,
+			Path:           "/tyk/oauth/tokens",
+			QueryParams:    map[string]string{"scope": "lapsed"},
+			Method:         http.MethodDelete,
+			Code:           http.StatusOK,
+		})
+
+		assertTokensLen(t, storageManager, storageKey1, 0)
+		assertTokensLen(t, storageManager, storageKey2, 0)
+	})
+}

gateway/oauth_manager.go

@@ -10,19 +10,21 @@ import (
 	"net/http"
 	"net/url"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/TykTechnologies/tyk/request"
-	"github.com/sirupsen/logrus"
-
+	"github.com/hashicorp/go-multierror"
 	"github.com/lonelycode/osin"
+	"github.com/sirupsen/logrus"
 	"golang.org/x/crypto/bcrypt"
 
-	"github.com/TykTechnologies/tyk/internal/uuid"
-
 	"strconv"
 
+	"github.com/TykTechnologies/tyk/internal/uuid"
+
 	"github.com/TykTechnologies/tyk/headers"
+	tykerrors "github.com/TykTechnologies/tyk/internal/errors"
 	"github.com/TykTechnologies/tyk/storage"
 	"github.com/TykTechnologies/tyk/user"
 )
@@ -1186,3 +1188,48 @@ func (r *RedisOsinStorageInterface) SetUser(username string, session *user.Sessi
 	return nil
 
 }
+
+func (gw *Gateway) purgeLapsedOAuthTokens() error {
+	if gw.GetConfig().OauthTokenExpiredRetainPeriod <= 0 {
+		return nil
+	}
+
+	redisCluster := &storage.RedisCluster{KeyPrefix: "", HashKeys: false, RedisController: gw.RedisController}
+	keys, err := redisCluster.ScanKeys(oAuthClientTokensKeyPattern)
+
+	if err != nil {
+		log.WithError(err).Debug("error while scanning for tokens")
+		return err
+	}
+
+	nowTs := time.Now().Unix()
+	// clean up expired tokens in sorted set (remove all tokens with score up to current timestamp minus retention)
+	cleanupStartScore := strconv.FormatInt(nowTs-int64(gw.GetConfig().OauthTokenExpiredRetainPeriod), 10)
+
+	var wg sync.WaitGroup
+
+	errs := make(chan error, len(keys))
+	for _, key := range keys {
+		wg.Add(1)
+		go func(k string) {
+			defer wg.Done()
+			if err := redisCluster.RemoveSortedSetRange(k, "-inf", cleanupStartScore); err != nil {
+				errs <- err
+			}
+		}(key)
+	}
+
+	// Wait for all goroutines to finish
+	wg.Wait()
+	close(errs)
+
+	combinedErr := &multierror.Error{
+		ErrorFormat: tykerrors.Formatter,
+	}
+
+	for err := range errs {
+		combinedErr = multierror.Append(combinedErr, err)
+	}
+
+	return combinedErr.ErrorOrNil()
+}

gateway/oauth_manager_test.go

@@ -8,7 +8,9 @@ import (
 	"bytes"
 	"encoding/json"
 	"net/url"
+	"path"
 	"reflect"
+	"strconv"
 	"strings"
 	"testing"
 
@@ -150,6 +152,41 @@ func (ts *Test) createTestOAuthClient(spec *APISpec, clientID string) OAuthClien
 	return testClient
 }
 
+func (ts *Test) createOAuthClientIDAndTokens(t *testing.T, spec *APISpec, clientID string) {
+	t.Helper()
+	ts.createTestOAuthClient(spec, clientID)
+
+	param := make(url.Values)
+	param.Set("response_type", "token")
+	param.Set("redirect_uri", authRedirectUri)
+	param.Set("client_id", clientID)
+	param.Set("client_secret", authClientSecret)
+	param.Set("key_rules", keyRules)
+
+	headers := map[string]string{
+		"Content-Type": "application/x-www-form-urlencoded",
+	}
+
+	for i := 0; i < 3; i++ {
+		resp, err := ts.Run(t, test.TestCase{
+			Path:      path.Join(spec.Proxy.ListenPath, "/tyk/oauth/authorize-client/"),
+			Data:      param.Encode(),
+			AdminAuth: true,
+			Headers:   headers,
+			Method:    http.MethodPost,
+			Code:      http.StatusOK,
+		})
+		if err != nil {
+			t.Error(err)
+		}
+
+		response := map[string]interface{}{}
+		if err := json.NewDecoder(resp.Body).Decode(&response); err != nil {
+			t.Fatal(err)
+		}
+	}
+}
+
 func TestOauthMultipleAPIs(t *testing.T) {
 	ts := StartTest(nil)
 	defer ts.Close()
@@ -1269,3 +1306,52 @@ func TestJSONToFormValues(t *testing.T) {
 		}
 	})
 }
+
+func TestPurgeOAuthClientTokensEvent(t *testing.T) {
+	conf := func(globalConf *config.Config) {
+		// set tokens to be expired after 1 second
+		globalConf.OauthTokenExpire = 1
+		// cleanup tokens older than 2 seconds
+		globalConf.OauthTokenExpiredRetainPeriod = 2
+	}
+
+	ts := StartTest(conf)
+	defer ts.Close()
+
+	assertTokensLen := func(t *testing.T, storageManager storage.Handler, storageKey string, expectedTokensLen int) {
+		nowTs := time.Now().Unix()
+		startScore := strconv.FormatInt(nowTs, 10)
+		tokens, _, err := storageManager.GetSortedSetRange(storageKey, startScore, "+inf")
+		assert.NoError(t, err)
+		assert.Equal(t, expectedTokensLen, len(tokens))
+	}
+
+	spec := ts.LoadTestOAuthSpec()
+
+	clientID1, clientID2 := uuid.New(), uuid.New()
+
+	ts.createOAuthClientIDAndTokens(t, spec, clientID1)
+	ts.createOAuthClientIDAndTokens(t, spec, clientID2)
+	storageKey1, storageKey2 := fmt.Sprintf("%s%s", prefixClientTokens, clientID1),
+		fmt.Sprintf("%s%s", prefixClientTokens, clientID2)
+
+	storageManager := ts.Gw.getGlobalStorageHandler(generateOAuthPrefix(spec.APIID), false)
+	storageManager.Connect()
+
+	assertTokensLen(t, storageManager, storageKey1, 3)
+	assertTokensLen(t, storageManager, storageKey2, 3)
+
+	time.Sleep(time.Second * 3)
+
+	// emit event
+
+	n := Notification{
+		Command: OAuthPurgeLapsedTokens,
+		Gw:      ts.Gw,
+	}
+	ts.Gw.MainNotifier.Notify(n)
+
+	assertTokensLen(t, storageManager, storageKey1, 0)
+	assertTokensLen(t, storageManager, storageKey2, 0)
+
+}

gateway/redis_signals.go

@@ -34,6 +34,7 @@ const (
 	NoticeGatewayDRLNotification NotificationCommand = "NoticeGatewayDRLNotification"
 	NoticeGatewayLENotification  NotificationCommand = "NoticeGatewayLENotification"
 	KeySpaceUpdateNotification   NotificationCommand = "KeySpaceUpdateNotification"
+	OAuthPurgeLapsedTokens       NotificationCommand = "OAuthPurgeLapsedTokens"
 )
 
 // Notification is a type that encodes a message published to a pub sub channel (shared between implementations)
@@ -119,6 +120,10 @@ func (gw *Gateway) handleRedisEvent(v interface{}, handled func(NotificationComm
 		gw.reloadURLStructure(reloaded)
 	case KeySpaceUpdateNotification:
 		gw.handleKeySpaceEventCacheFlush(notif.Payload)
+	case OAuthPurgeLapsedTokens:
+		if err := gw.purgeLapsedOAuthTokens(); err != nil {
+			log.WithError(err).Errorf("error while purging tokens for event %s", OAuthPurgeLapsedTokens)
+		}
 	default:
 		pubSubLog.Warnf("Unknown notification command: %q", notif.Command)
 		return

gateway/server.go

@@ -625,6 +625,7 @@ func (gw *Gateway) loadControlAPIEndpoints(muxer *mux.Router) {
 	r.HandleFunc("/oauth/clients/{apiID}", gw.oAuthClientHandler).Methods("GET", "DELETE")
 	r.HandleFunc("/oauth/clients/{apiID}/{keyName:[^/]*}", gw.oAuthClientHandler).Methods("GET", "DELETE")
 	r.HandleFunc("/oauth/clients/{apiID}/{keyName}/tokens", gw.oAuthClientTokensHandler).Methods("GET")
+	r.HandleFunc("/oauth/tokens", gw.oAuthTokensHandler).Methods(http.MethodDelete)
 
 	mainLog.Debug("Loaded API Endpoints")
 }

go.mod

@@ -47,7 +47,9 @@ require (
 	github.com/hashicorp/go-hclog v0.14.1 // indirect
 	github.com/hashicorp/go-immutable-radix v1.3.0 // indirect
 	github.com/hashicorp/go-msgpack v0.5.5 // indirect
+	github.com/hashicorp/go-multierror v1.1.0
 	github.com/hashicorp/go-retryablehttp v0.6.7 // indirect
+	github.com/hashicorp/go-version v1.1.0
 	github.com/hashicorp/memberlist v0.1.6 // indirect
 	github.com/hashicorp/serf v0.8.6 // indirect
 	github.com/hashicorp/vault/api v1.0.5-0.20200717191844-f687267c8086
@@ -108,3 +110,6 @@ require (
 )
 
 //replace github.com/jensneuse/graphql-go-tools => ../graphql-go-tools
+replace sourcegraph.com/sourcegraph/appdash => github.com/sourcegraph/appdash v0.0.0-20211028080628-e2786a622600
+
+replace sourcegraph.com/sourcegraph/appdash-data => github.com/sourcegraph/appdash-data v0.0.0-20151005221446-73f23eafcf67

go.sum

@@ -271,6 +271,7 @@ github.com/hashicorp/go-uuid v1.0.0/go.mod h1:6SBZvOh/SIDV7/2o3Jml5SYk/TvGqwFJ/b
 github.com/hashicorp/go-uuid v1.0.1/go.mod h1:6SBZvOh/SIDV7/2o3Jml5SYk/TvGqwFJ/bN7x4byOro=
 github.com/hashicorp/go-uuid v1.0.2 h1:cfejS+Tpcp13yd5nYHWDI6qVCny6wyX2Mt5SGur2IGE=
 github.com/hashicorp/go-uuid v1.0.2/go.mod h1:6SBZvOh/SIDV7/2o3Jml5SYk/TvGqwFJ/bN7x4byOro=
+github.com/hashicorp/go-version v1.1.0 h1:bPIoEKD27tNdebFGGxxYwcL4nepeY4j1QP23PFRGzg0=
 github.com/hashicorp/go-version v1.1.0/go.mod h1:fltr4n8CU8Ke44wwGCBoEymUuxUHl09ZGVZPK5anwXA=
 github.com/hashicorp/go.net v0.0.1/go.mod h1:hjKkEWcCURg++eb33jQU7oqQcI9XDCnUzHA0oac0k90=
 github.com/hashicorp/golang-lru v0.5.0/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ4Ao+sR/qLZy8=
@@ -519,6 +520,8 @@ github.com/shurcooL/vfsgen v0.0.0-20180121065927-ffb13db8def0/go.mod h1:TrYk7fJV
 github.com/sirupsen/logrus v1.2.0/go.mod h1:LxeOpSwHxABJmUn/MG1IvRgCAasNZTLOkJPxbbu5VWo=
 github.com/sirupsen/logrus v1.4.2 h1:SPIRibHv4MatM3XXNO2BJeFLZwZ2LvZgfQ5+UNI2im4=
 github.com/sirupsen/logrus v1.4.2/go.mod h1:tLMulIdttU9McNUspp0xgXVQah82FyeX6MwdIuYE2rE=
+github.com/sourcegraph/appdash v0.0.0-20211028080628-e2786a622600/go.mod h1:V952P4GGl1v/MMynLwxVdWEbSZJx+n0oOO3ljnez+WU=
+github.com/sourcegraph/appdash-data v0.0.0-20151005221446-73f23eafcf67/go.mod h1:tNZjgbYncKL5HxvDULAr/mWDmFz4B7H8yrXEDlnoIiw=
 github.com/spf13/afero v1.1.2/go.mod h1:j4pytiNVoe2o6bmDsKpLACNPDBIoEAkihy7loJ1B0CQ=
 github.com/spf13/afero v1.6.0 h1:xoax2sJ2DT8S8xA2paPFjDCScCNeWsg75VG0DLRreiY=
 github.com/spf13/afero v1.6.0/go.mod h1:Ai8FlHk4v/PARR026UzYexafAt9roJ7LcLMAmO6Z93I=

internal/errors/errors.go

@@ -0,0 +1,15 @@
+package errors
+
+import "strings"
+
+func Formatter(errs []error) string {
+	var result strings.Builder
+	for i, err := range errs {
+		result.WriteString(err.Error())
+		if i < len(errs)-1 {
+			result.WriteString("\n")
+		}
+	}
+
+	return result.String()
+}