fix: access_denied error handling for OAuth destinations

router/handle.go

@@ -616,8 +616,8 @@ func (rt *Handle) handleOAuthDestResponse(params *HandleDestOAuthRespParams) (in
 			return trRespStatusCode, trRespBody
 		}
 		switch destErrOutput.AuthErrorCategory {
-		case oauth.DISABLE_DEST:
-			return rt.execDisableDestination(&destinationJob.Destination, workspaceID, trRespBody, rudderAccountID)
+		case oauth.AUTH_STATUS_INACTIVE:
+			return rt.updateAuthStatusToInactive(&destinationJob.Destination, workspaceID, trRespBody, rudderAccountID)
 		case oauth.REFRESH_TOKEN:
 			var refSecret *oauth.AuthResponse
 			refTokenParams := &oauth.RefreshTokenParams{
@@ -630,20 +630,20 @@ func (rt *Handle) handleOAuthDestResponse(params *HandleDestOAuthRespParams) (in
 			}
 			errCatStatusCode, refSecret = rt.oauth.RefreshToken(refTokenParams)
 			refSec := *refSecret
-			if routerutils.IsNotEmptyString(refSec.Err) && refSec.Err == oauth.INVALID_REFRESH_TOKEN_GRANT {
+			if routerutils.IsNotEmptyString(refSec.Err) && refSec.Err == oauth.REF_TOKEN_INVALID_GRANT {
 				// In-case the refresh token has been revoked, this error comes in
 				// Even trying to refresh the token also doesn't work here. Hence, this would be more ideal to Abort Events
 				// As well as to disable destination as well.
 				// Alert the user in this error as well, to check if the refresh token also has been revoked & fix it
-				disableStCd, _ := rt.execDisableDestination(&destinationJob.Destination, workspaceID, trRespBody, rudderAccountID)
-				stats.Default.NewTaggedStat(oauth.INVALID_REFRESH_TOKEN_GRANT, stats.CountType, stats.Tags{
+				disableStCd, _ := rt.updateAuthStatusToInactive(&destinationJob.Destination, workspaceID, trRespBody, rudderAccountID)
+				stats.Default.NewTaggedStat(oauth.REF_TOKEN_INVALID_GRANT, stats.CountType, stats.Tags{
 					"destinationId": destinationJob.Destination.ID,
 					"workspaceId":   refTokenParams.WorkspaceId,
 					"accountId":     refTokenParams.AccountId,
 					"destType":      refTokenParams.DestDefName,
 					"flowType":      string(oauth.RudderFlow_Delivery),
 				}).Increment()
-				rt.logger.Errorf(`[OAuth request] Aborting the event as %v`, oauth.INVALID_REFRESH_TOKEN_GRANT)
+				rt.logger.Errorf(`[OAuth request] Aborting the event as %v`, oauth.REF_TOKEN_INVALID_GRANT)
 				return disableStCd, refSec.Err
 			}
 			// Error while refreshing the token or Has an error while refreshing or sending empty access token
@@ -658,24 +658,24 @@ func (rt *Handle) handleOAuthDestResponse(params *HandleDestOAuthRespParams) (in
 	return trRespStatusCode, trRespBody
 }
 
-func (rt *Handle) execDisableDestination(destination *backendconfig.DestinationT, workspaceID, destResBody, rudderAccountId string) (int, string) {
-	disableDestStatTags := stats.Tags{
+func (rt *Handle) updateAuthStatusToInactive(destination *backendconfig.DestinationT, workspaceID, destResBody, rudderAccountId string) (int, string) {
+	inactiveAuthStatusStatTags := stats.Tags{
 		"id":          destination.ID,
 		"destType":    destination.DestinationDefinition.Name,
 		"workspaceId": workspaceID,
 		"success":     "true",
 		"flowType":    string(oauth.RudderFlow_Delivery),
 	}
-	errCatStatusCode, errCatResponse := rt.oauth.DisableDestination(destination, workspaceID, rudderAccountId)
+	errCatStatusCode, errCatResponse := rt.oauth.UpdateAuthStatusToInactive(destination, workspaceID, rudderAccountId)
 	if errCatStatusCode != http.StatusOK {
 		// Error while disabling a destination
 		// High-Priority notification to rudderstack needs to be sent
-		disableDestStatTags["success"] = "false"
-		stats.Default.NewTaggedStat("disable_destination_category_count", stats.CountType, disableDestStatTags).Increment()
+		inactiveAuthStatusStatTags["success"] = "false"
+		stats.Default.NewTaggedStat("auth_status_inactive_category_count", stats.CountType, inactiveAuthStatusStatTags).Increment()
 		return http.StatusBadRequest, errCatResponse
 	}
 	// High-Priority notification to workspace(& rudderstack) needs to be sent
-	stats.Default.NewTaggedStat("disable_destination_category_count", stats.CountType, disableDestStatTags).Increment()
+	stats.Default.NewTaggedStat("auth_status_inactive_category_count", stats.CountType, inactiveAuthStatusStatTags).Increment()
 	// Abort the jobs as the destination is disabled
 	return http.StatusBadRequest, destResBody
 }

services/oauth/oauth.go

@@ -67,6 +67,12 @@ type DisableDestinationResponse struct {
 	DestinationId string `json:"id"`
 }
 
+type AuthStatusInactiveResponse struct {
+	Success    string `json:"success"`
+	Error      string `json:"error"`
+	StatusCode string `json:"statusCode"`
+}
+
 type RefreshTokenParams struct {
 	AccountId       string
 	WorkspaceId     string
@@ -78,21 +84,21 @@ type RefreshTokenParams struct {
 
 // OAuthErrResHandler is the handle for this class
 type OAuthErrResHandler struct {
-	tr                   *http.Transport
-	client               *http.Client
-	logger               logger.Logger
-	destLockMap          map[string]*sync.RWMutex // This mutex map is used for disable destination locking
-	accountLockMap       map[string]*sync.RWMutex // This mutex map is used for refresh token locking
-	lockMapWMutex        *sync.RWMutex            // This mutex is used to prevent concurrent writes in lockMap(s) mentioned in the struct
-	destAuthInfoMap      map[string]*AuthResponse
-	refreshActiveMap     map[string]bool // Used to check if a refresh request for an account is already InProgress
-	disableDestActiveMap map[string]bool // Used to check if a disable destination request for a destination is already InProgress
-	tokenProvider        tokenProvider
-	rudderFlowType       RudderFlow
+	tr                        *http.Transport
+	client                    *http.Client
+	logger                    logger.Logger
+	destLockMap               map[string]*sync.RWMutex // This mutex map is used for disable destination locking
+	accountLockMap            map[string]*sync.RWMutex // This mutex map is used for refresh token locking
+	lockMapWMutex             *sync.RWMutex            // This mutex is used to prevent concurrent writes in lockMap(s) mentioned in the struct
+	destAuthInfoMap           map[string]*AuthResponse
+	refreshActiveMap          map[string]bool // Used to check if a refresh request for an account is already InProgress
+	authStatusUpdateActiveMap map[string]bool // Used to check if a authStatusInactive request for a destination is already InProgress
+	tokenProvider             tokenProvider
+	rudderFlowType            RudderFlow
 }
 
 type Authorizer interface {
-	DisableDestination(destination *backendconfig.DestinationT, workspaceId, rudderAccountId string) (statusCode int, resBody string)
+	UpdateAuthStatusToInactive(destination *backendconfig.DestinationT, workspaceId, rudderAccountId string) (statusCode int, resBody string)
 	RefreshToken(refTokenParams *RefreshTokenParams) (int, *AuthResponse)
 	FetchToken(fetchTokenParams *RefreshTokenParams) (int, *AuthResponse)
 }
@@ -113,9 +119,12 @@ var (
 )
 
 const (
-	DISABLE_DEST                = "DISABLE_DESTINATION"
-	REFRESH_TOKEN               = "REFRESH_TOKEN"
-	INVALID_REFRESH_TOKEN_GRANT = "refresh_token_invalid_grant"
+	REFRESH_TOKEN = "REFRESH_TOKEN"
+	// Identifier to be sent from destination(during transformation/delivery)
+	AUTH_STATUS_INACTIVE = "AUTH_STATUS_INACTIVE"
+
+	// Identifier for invalid_grant or access_denied errors(during refreshing the token)
+	REF_TOKEN_INVALID_GRANT = "ref_token_invalid_grant"
 )
 
 // This struct only exists for marshalling and sending payload to control-plane
@@ -155,13 +164,13 @@ func NewOAuthErrorHandler(provider tokenProvider, options ...func(*OAuthErrResHa
 		tr:            &http.Transport{},
 		client:        &http.Client{Timeout: config.GetDuration("HttpClient.oauth.timeout", 30, time.Second)},
 		// This timeout is kind of modifiable & it seemed like 10 mins for this is too much!
-		destLockMap:          make(map[string]*sync.RWMutex),
-		accountLockMap:       make(map[string]*sync.RWMutex),
-		lockMapWMutex:        &sync.RWMutex{},
-		destAuthInfoMap:      make(map[string]*AuthResponse),
-		refreshActiveMap:     make(map[string]bool),
-		disableDestActiveMap: make(map[string]bool),
-		rudderFlowType:       RudderFlow_Delivery,
+		destLockMap:               make(map[string]*sync.RWMutex),
+		accountLockMap:            make(map[string]*sync.RWMutex),
+		lockMapWMutex:             &sync.RWMutex{},
+		destAuthInfoMap:           make(map[string]*AuthResponse),
+		refreshActiveMap:          make(map[string]bool),
+		authStatusUpdateActiveMap: make(map[string]bool),
+		rudderFlowType:            RudderFlow_Delivery,
 	}
 	for _, opt := range options {
 		opt(oAuthErrResHandler)
@@ -350,7 +359,7 @@ func (authErrHandler *OAuthErrResHandler) fetchAccountInfoFromCp(refTokenParams
 		authStats.statName = fmt.Sprintf("%s_failure", refTokenParams.EventNamePrefix)
 		authStats.errorMessage = refErrMsg
 		authStats.SendCountStat()
-		if refErrMsg == INVALID_REFRESH_TOKEN_GRANT {
+		if refErrMsg == REF_TOKEN_INVALID_GRANT {
 			// Should abort the event as refresh is not going to work
 			// until we have new refresh token for the account
 			return http.StatusBadRequest
@@ -372,9 +381,9 @@ func getRefreshTokenErrResp(response string, accountSecret *AccountSecret) (mess
 	if err := json.Unmarshal([]byte(response), &accountSecret); err != nil {
 		// Some problem with AccountSecret unmarshalling
 		message = fmt.Sprintf("Unmarshal of response unsuccessful: %v", response)
-	} else if gjson.Get(response, "body.code").String() == INVALID_REFRESH_TOKEN_GRANT {
+	} else if gjson.Get(response, "body.code").String() == REF_TOKEN_INVALID_GRANT {
 		// User (or) AccessToken (or) RefreshToken has been revoked
-		message = INVALID_REFRESH_TOKEN_GRANT
+		message = REF_TOKEN_INVALID_GRANT
 	}
 	return message
 }
@@ -406,83 +415,89 @@ func (refStats *OAuthStats) SendCountStat() {
 	}).Increment()
 }
 
-func (authErrHandler *OAuthErrResHandler) DisableDestination(destination *backendconfig.DestinationT, workspaceId, rudderAccountId string) (statusCode int, respBody string) {
+func (authErrHandler *OAuthErrResHandler) UpdateAuthStatusToInactive(destination *backendconfig.DestinationT, workspaceId, rudderAccountId string) (statusCode int, respBody string) {
 	authErrHandlerTimeStart := time.Now()
 	destinationId := destination.ID
-	disableDestMutex := authErrHandler.getKeyMutex(authErrHandler.destLockMap, destinationId)
+	authStatusInactiveMutex := authErrHandler.getKeyMutex(authErrHandler.destLockMap, destinationId)
 
-	disableDestStats := &OAuthStats{
+	getStatName := func(statName string) string {
+		return fmt.Sprintf("auth_status_inactive_%v", statName)
+	}
+
+	authStatusInactiveStats := &OAuthStats{
 		id:              destinationId,
 		workspaceId:     workspaceId,
 		rudderCategory:  "destination",
 		statName:        "",
 		isCallToCpApi:   false,
-		authErrCategory: DISABLE_DEST,
+		authErrCategory: AUTH_STATUS_INACTIVE,
 		errorMessage:    "",
 		destDefName:     destination.DestinationDefinition.Name,
 		flowType:        authErrHandler.rudderFlowType,
 	}
 	defer func() {
-		disableDestStats.statName = "disable_destination_total_req_latency"
-		disableDestStats.isCallToCpApi = false
-		disableDestStats.SendTimerStats(authErrHandlerTimeStart)
+		authStatusInactiveStats.statName = getStatName("total_req_latency")
+		authStatusInactiveStats.isCallToCpApi = false
+		authStatusInactiveStats.SendTimerStats(authErrHandlerTimeStart)
 	}()
 
-	disableDestMutex.Lock()
-	isDisableDestActive, isDisableDestReqPresent := authErrHandler.disableDestActiveMap[destinationId]
-	disableActiveReq := strconv.FormatBool(isDisableDestReqPresent && isDisableDestActive)
-	if isDisableDestReqPresent && isDisableDestActive {
-		disableDestMutex.Unlock()
-		authErrHandler.logger.Debugf("[%s request] :: Disable Destination Active : %s\n", loggerNm, disableActiveReq)
-		return http.StatusOK, fmt.Sprintf(`{response: {isDisabled: %v, activeRequest: %v}`, false, disableActiveReq)
+	authStatusInactiveMutex.Lock()
+	isAuthStatusUpdateActive, isAuthStatusUpdateReqPresent := authErrHandler.authStatusUpdateActiveMap[destinationId]
+	authStatusUpdateActiveReq := strconv.FormatBool(isAuthStatusUpdateReqPresent && isAuthStatusUpdateActive)
+	if isAuthStatusUpdateReqPresent && isAuthStatusUpdateActive {
+		authStatusInactiveMutex.Unlock()
+		authErrHandler.logger.Debugf("[%s request] :: AuthStatusInactive request Active : %s\n", loggerNm, authStatusUpdateActiveReq)
+		return http.StatusOK, fmt.Sprintf(`{response: {authStatusInactive: %v, activeRequest: %v}`, false, authStatusUpdateActiveReq)
 	}
 
-	authErrHandler.disableDestActiveMap[destinationId] = true
-	disableDestMutex.Unlock()
+	authErrHandler.authStatusUpdateActiveMap[destinationId] = true
+	authStatusInactiveMutex.Unlock()
 
 	defer func() {
-		disableDestMutex.Lock()
-		authErrHandler.disableDestActiveMap[destinationId] = false
-		authErrHandler.logger.Debugf("[%s request] :: Disable request is inactive!", loggerNm)
-		disableDestMutex.Unlock()
+		authStatusInactiveMutex.Lock()
+		authErrHandler.authStatusUpdateActiveMap[destinationId] = false
+		authErrHandler.logger.Debugf("[%s request] :: AuthStatusInactive request is inactive!", loggerNm)
+		authStatusInactiveMutex.Unlock()
 	}()
 
-	disableURL := fmt.Sprintf("%s/workspaces/%s/destinations/%s/disable", configBEURL, workspaceId, destinationId)
-	disableCpReq := &ControlPlaneRequestT{
-		Url:         disableURL,
-		Method:      http.MethodDelete,
+	authStatusInactiveUrl := fmt.Sprintf("%s/workspaces/%s/destinations/%s/authStatus/toggle", configBEURL, workspaceId, destinationId)
+	authStatusInactiveCpReq := &ControlPlaneRequestT{
+		Url:         authStatusInactiveUrl,
+		Method:      http.MethodPost,
+		Body:        `{"authStatus": "inactive"}`,
+		ContentType: "application/json",
 		destName:    destination.DestinationDefinition.Name,
-		RequestType: "Disable destination",
+		RequestType: "Auth Status inactive",
 	}
 
-	disableDestStats.statName = "disable_destination_request_sent"
-	disableDestStats.isCallToCpApi = true
-	disableDestStats.SendCountStat()
+	authStatusInactiveStats.statName = getStatName("request_sent")
+	authStatusInactiveStats.isCallToCpApi = true
+	authStatusInactiveStats.SendCountStat()
 
 	cpiCallStartTime := time.Now()
-	statusCode, respBody = authErrHandler.cpApiCall(disableCpReq)
-	disableDestStats.statName = `disable_destination_request_latency`
-	defer disableDestStats.SendTimerStats(cpiCallStartTime)
-	authErrHandler.logger.Debugf(`Response from CP(stCd: %v) for disable dest req: %v`, statusCode, respBody)
+	statusCode, respBody = authErrHandler.cpApiCall(authStatusInactiveCpReq)
+	authStatusInactiveStats.statName = getStatName("request_latency")
+	defer authStatusInactiveStats.SendTimerStats(cpiCallStartTime)
+	authErrHandler.logger.Errorf(`Response from CP(stCd: %v) for auth status inactive req: %v`, statusCode, respBody)
 
-	var disableDestRes *DisableDestinationResponse
-	if disableErr := json.Unmarshal([]byte(respBody), &disableDestRes); disableErr != nil || !router_utils.IsNotEmptyString(disableDestRes.DestinationId) {
+	var authStatusInactiveRes *AuthStatusInactiveResponse
+	if unmarshalErr := json.Unmarshal([]byte(respBody), &authStatusInactiveRes); unmarshalErr != nil || !router_utils.IsNotEmptyString(authStatusInactiveRes.Error) {
 		var msg string
-		if disableErr != nil {
-			msg = disableErr.Error()
+		if unmarshalErr != nil {
+			msg = unmarshalErr.Error()
 		} else {
-			msg = "Could not disable the destination"
+			msg = "Could not update authStatus to inactive for destination the destination"
 		}
-		disableDestStats.statName = "disable_destination_failure"
-		disableDestStats.errorMessage = msg
-		disableDestStats.SendCountStat()
+		authStatusInactiveStats.statName = getStatName("failure")
+		authStatusInactiveStats.errorMessage = msg
+		authStatusInactiveStats.SendCountStat()
 		return http.StatusBadRequest, msg
 	}
 
-	authErrHandler.logger.Debugf("[%s request] :: (Write) Disable Response received : %s\n", loggerNm, respBody)
-	disableDestStats.statName = "disable_destination_success"
-	disableDestStats.errorMessage = ""
-	disableDestStats.SendCountStat()
+	authErrHandler.logger.Errorf("[%s request] :: (Write) auth status inactive Response received : %s\n", loggerNm, respBody)
+	authStatusInactiveStats.statName = getStatName("success")
+	authStatusInactiveStats.errorMessage = ""
+	authStatusInactiveStats.SendCountStat()
 
 	// After a successfully disabling the destination, need to remove existing accessToken(from in-memory cache)
 	// This is being done to obtain new token after re-enabling disabled destination
@@ -491,7 +506,7 @@ func (authErrHandler *OAuthErrResHandler) DisableDestination(destination *backen
 	defer accountMutex.Unlock()
 	delete(authErrHandler.destAuthInfoMap, rudderAccountId)
 
-	return statusCode, fmt.Sprintf(`{response: {isDisabled: %v, activeRequest: %v}`, !disableDestRes.Enabled, false)
+	return statusCode, fmt.Sprintf(`{response: {"authStatus": "inactive", "activeRequest": %v}`, false)
 }
 
 func processResponse(resp *http.Response) (statusCode int, respBody string) {