DetectAbnormalLoginBehaviour.siddhi

@App:name("AbnormalLoginBehaviour")

-- Please refer to https://docs.wso2.com/display/SP400/Quick+Start+Guide on getting started with SP editor. 

@sink(type = 'log', @map(type = 'text'))
@Source(type = 'http-request', source.id='testsource', 
receiver.url="https://localhost:8280/Analytics/InputStream", 
@map(type='json', @attributes(messageId='trp:messageId', username='$.event.username', loginTime='$.event.loginTime', loginIp = '$.event.loginIp')))
define stream InputStream (
    messageId string,
    username string,
    loginIp string,
    loginTime long);
    
-- Output stream with risk value of the current login attempt.
@sink(type = 'log', @map(type = 'text'))
@sink(type='http-response', source.id='testsource', message.id='{{messageId}}', @map(type='json'))
define stream OutputStream (
    messageId string,
    username string,
    risk double);
       
-- OverallAuthTableTable for all the authentication data.
@store(type='rdbms', datasource='IS_ANALYTICS_DB')
@primaryKey('meta_tenantId', 'eventId', 'eventType')
define table OverallAuthTable (
    meta_tenantId int,
    contextId string,
    eventId string,
    eventType string,
    username string,
    localUsername string,
    userStoreDomain string,
    tenantDomain string,
    remoteIp string,
    region string,
    inboundAuthType string,
    serviceProvider string,
    rememberMeEnabled bool,
    forceAuthEnabled bool,
    passiveAuthEnabled bool,
    rolesCommaSeparated string,
    authenticationStep string,
    identityProvider string,
    authenticationSuccess bool,
    authStepSuccess bool,
    stepAuthenticator string,
    isFirstLogin bool,
    identityProviderType string,
    utcTime string,
    timestamp long
);

@sink(type='log')
define stream dayFilterStream(username string, loginIp string, loginHour long, minLoginTime long, maxLoginTime long, messageId string);

from InputStream#rdbms:query('IS_ANALYTICS_DB', 'select hour(FROM_UNIXTIME(?/1000)) as loginHour, hour(FROM_UNIXTIME(min(timestamp)/1000)) as minLoginTime, hour(FROM_UNIXTIME(max(timestamp)/1000)) as maxLoginTime from OverallAuthTable where DAYOFWEEK(FROM_UNIXTIME(?/1000)) = DAYOFWEEK(FROM_UNIXTIME(timestamp/1000))  and (timestamp > UNIX_TIMESTAMP(subdate(current_date, 28))) and authStepSuccess = 1 and eventType=?;', loginTime, loginTime, 'overall', 'loginHour long, minLoginTime long, maxLoginTime long' )
select username, loginIp, loginHour, minLoginTime, maxLoginTime, messageId
insert into dayFilterStream;

from dayFilterStream
select messageId, username, ifThenElse(minLoginTime<=loginHour and loginHour<=maxLoginTime, 0.0, ifThenElse(minLoginTime-2<loginHour and loginHour<maxLoginTime+2, 0.5, 1.0)) as risk
insert into OutputStream;