/* Enter a unique ExecutionPlan */
@Plan:name('APIMAnalytics-AbnormalAccessTokenRefresh')

/* Enter a unique description for ExecutionPlan */
@Plan:description('Detects abnormal access token refreshes and alerts the user')

/* define streams/tables and write queries here ... */

@Import('org.wso2.carbon.identity.oauth.token.issuance:1.0.0')
define stream OAuthStream (user string, tenantDomain string, userstoreDomain string, clientId string, grantType string, tokenId string, authzScopes string, unAuthzScopes string, isSuccess bool, errorCode string, errorMsg string, accessTokenValidityMillis long, refreshTokenValidityMillis long, issuedTime long);

@Export('org.wso2.analytics.apim.alert.abnormalTokenRefresh:1.0.0')
define stream alert_abnormalTokenRefreshStream (userId string, tenantDomain string, scope string, timestamp long, alertTimestamp long, msg string, severity int, consumerKey string);

@Export('org.wso2.analytics.apim.allApimAlertsStream:1.0.0')
define stream alert_allAPIMAlertsStream (type string, tenantDomain string, msg string, severity int, alertTimestamp long);

@from(eventtable = 'analytics.table', table.name = 'ORG_WSO2_ANALYTICS_APIM_LASTACCESSTOKENREFRESHEVENTTABLE', primary.keys = 'userId, clientId, scopes', indices = 'userId, clientId, scopes, timestamp', wait.for.indexing = 'false')
define table LastAccessTokenRefreshTable (userId string, clientId string, scopes string, timestamp long);

@from(eventtable = 'analytics.table', table.name = 'ORG_WSO2_ANALYTICS_APIM_ACCESSTOKENREFRESHTIMEDIFFERENCE', indices = 'userId, clientId, scopes, timeDifference, timestamp', wait.for.indexing = 'false')
define table AccessTokenRefreshTable (userId string, clientId string, scopes string, timeDifference long, timestamp long);

@from(eventtable = 'analytics.table', table.name = 'ORG_WSO2_ANALYTICS_APIM_ACCESSTOKENREFRESHSUMMARYTABLE', indices = 'userId, clientId, scopes, minTimeDifference, maxTimeDifference', wait.for.indexing = 'false')
define table AccessTokenRefreshSummaryTable (userId string, clientId string, scopes string, minTimeDifference double, maxTimeDifference double);


--- Filter out the successful token refresh events
@info(name = 'query1')
from OAuthStream[grantType == 'refreshToken' and isSuccess == true]
select str:concat(tenantDomain, '-', userstoreDomain , '-' , user) as userId, tenantDomain, clientId, authzScopes as scopes, issuedTime as timestamp
insert into RefreshTokenStream;

--- Calculate the time difference between consequent access token refreshes
@info(name = 'query2')
from RefreshTokenStream as rts join LastAccessTokenRefreshTable as lrt
on rts.userId == lrt.userId and rts.clientId == lrt.clientId and rts.scopes == lrt.scopes
select rts.userId as userId, rts.tenantDomain, rts.clientId as clientId, rts.scopes as scopes, (rts.timestamp - lrt.timestamp) as timeDifference, rts.timestamp as timestamp
insert into AccessTokenRefreshTimeStream;

--- Test the events from abnormal refreshes
@info(name = 'query3')
from AccessTokenRefreshTimeStream as trs join AccessTokenRefreshSummaryTable as atst
on trs.userId == atst.userId and trs.clientId == atst.clientId and trs.scopes == atst.scopes
select trs.userId as userId, trs.tenantDomain, trs.clientId as consumerKey, trs.scopes as scopes, atst.minTimeDifference as minTimeDifference, atst.maxTimeDifference as maxTimeDifference, trs.timeDifference as timeDifference, trs.timestamp as timestamp
insert into AbnormalityTesterStream;

@info(name = 'query4')
from AbnormalityTesterStream[ timeDifference < minTimeDifference or timeDifference > maxTimeDifference]
select userId, tenantDomain, scopes as scope, timestamp,(time:timestampInMilliseconds()) as alertTimestamp,str:concat('Abnormal access token refresh detected at: ', time:dateFormat(timestamp,'yyyy-MM-dd HH:mm:ss'), '.') as msg, 1 as severity, consumerKey
insert into alert_abnormalTokenRefreshStream;

@info(name = 'query5')
from alert_abnormalTokenRefreshStream
select "AbnormalRefreshAlert" as type, tenantDomain, str:concat('Abnormal access token refresh detected by user:',userId,' using consumer key: ', consumerKey, ' at: ', time:dateFormat(timestamp,'yyyy-MM-dd HH:mm:ss'), '.') as msg, severity, (time:timestampInMilliseconds()) as alertTimestamp
insert into alert_allAPIMAlertsStream;

-- Persist the time difference between consequent access token refreshes
@info(name = 'query6')
from AccessTokenRefreshTimeStream
select userId, clientId, scopes, timeDifference, timestamp
insert into AccessTokenRefreshTable;

-- Persist the latest refresh token
@info(name = 'query7')
from RefreshTokenStream
select userId, clientId, scopes, timestamp
insert into LastAccessTokenRefreshTable;
