写点什么

大数据培训 | Flink 如何监控恶意登录

作者:@零度
  • 2022 年 6 月 23 日
  • 本文字数:3375 字

    阅读完需:约 11 分钟

​模块创建和数据准备


继续在 UserBehaviorAnalysis 下新建一个 maven module 作为子项目,命名为 LoginFailDetect。在这个子模块中,我们将会用到 flink 的 CEP 库来实现事件流的模式匹配,所以需要在 pom 文件中引入 CEP 的相关依赖:


<dependency>


<groupId>org.apache.flink</groupId>


<artifactId>flink-cep-scala_${scala.binary.version}</artifactId>


<version>${flink.version}</version>


</dependency>


同样,在 src/main/目录下,将默认源文件目录 java 改名为 scala。


代码实现


对于网站而言,用户登录并不是频繁的业务操作。如果一个用户短时间内频繁登录失败,就有可能是出现了程序的恶意攻击,比如密码暴力破解。因此我们考虑,应该对用户的登录失败动作进行统计,具体来说,如果同一用户(可以是不同 IP)在 2 秒之内连续两次登录失败,就认为存在恶意登录的风险,输出相关的信息进行报警提示。这是电商网站、也是几乎所有网站风控的基本一环。


状态编程


由于同样引入了时间,我们可以想到,最简单的方法其实与之前的热门统计类似,只需要按照用户 ID 分流,然后遇到登录失败的事件时将其保存在 ListState 中,然后设置一个定时器,2 秒后触发。定时器触发时检查状态中的登录失败事件个数,如果大于等于 2,那么就输出报警信息。


更多 Java –大数据 – 前端 – UI/UE - Android - 人工智能资料下载,可访问百度:尚硅谷官网(www.atguigu.com)


在 src/main/scala 下创建 LoginFail.scala 文件,新建一个单例对象。定义样例类 LoginEvent,这是输入的登录事件流。登录数据本应该从 UserBehavior 日志里提取,由于 UserBehavior.csv 中没有做相关埋点,我们从另一个文件 LoginLog.csv 中读取登录数据_大数据培训


代码如下:


LoginFailDetect/src/main/scala/LoginFail.scala


case class LoginEvent(userId: Long, ip: String, eventType: String, eventTime: Long)


object LoginFail {


def main(args: Array[String]): Unit = {


val env = StreamExecutionEnvironment.getExecutionEnvironment


env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)


env.setParallelism(1)


val loginEventStream = env.readTextFile("YOUR_PATH\resources\LoginLog.csv")


.map( data => {


val dataArray = data.split(",")


LoginEvent(dataArray(0).toLong, dataArray(1), dataArray(2),


dataArray(3).toLong)


})


.assignTimestampsAndWatermarks(new


BoundedOutOfOrdernessTimestampExtractor[ApacheLogEvent]


(Time.milliseconds(3000)) {


override def extractTimestamp(element: ApacheLogEvent): Long ={


element.eventTime * 1000L


}


})


.keyBy(_.userId)


.process(new MatchFunction())


.print()


env.execute("Login Fail Detect Job")


}


class MatchFunction extends KeyedProcessFunction[Long, LoginEvent, LoginEvent] {


// 定义状态变量


lazy val loginState: ListState[LoginEvent] = getRuntimeContext.getListState(


new ListStateDescriptor[LoginEvent]("saved login", classOf[LoginEvent]))


override def processElement(login: LoginEvent,


context: KeyedProcessFunction[Long, LoginEvent,


LoginEvent]#Context, out: Collector[LoginEvent]): Unit = {


if (login.eventType == "fail") {


loginState.add(login)


}


// 注册定时器,触发事件设定为 2 秒后


context.timerService.registerEventTimeTimer(login.eventTime * 1000 + 2 * 1000)


}


override def onTimer(timestamp: Long,


ctx: KeyedProcessFunction[Long, LoginEvent,


LoginEvent]#OnTimerContext, out: Collector[LoginEvent]): Unit = {


val allLogins: ListBuffer[LoginEvent] = ListBuffer()


import scala.collection.JavaConversions._


for (login <- loginState.get) {


allLogins += login


}


loginState.clear()


if (allLogins.length > 1) {


out.collect(allLogins.head)


}


}


}


}


状态编程的改进


上一节的代码实现中我们可以看到,直接把每次登录失败的数据存起来、设置定时器一段时间后再读取,这种做法尽管简单,但和我们开始的需求还是略有差异的。这种做法只能隔 2 秒之后去判断一下这期间是否有多次失败登录,而不是在一次登录失败之后、再一次登录失败时就立刻报警。这个需求如果严格实现起来,相当于要判断任意紧邻的事件,是否符合某种模式。


于是我们可以想到,这个需求其实可以不用定时器触发,直接在状态中存取上一次登录失败的事件,每次都做判断和比对,就可以实现最初的需求_大数据视频


上节的代码 MatchFunction 中删掉 onTimer,processElement 改为:


override def processElement(value: LoginEvent, ctx: KeyedProcessFunction[Long,


LoginEvent, Warning]#Context, out: Collector[Warning]): Unit = {


// 首先按照 type 做筛选,如果 success 直接清空,如果 fail 再做处理


if ( value.eventType == "fail" ){


// 如果已经有登录失败的数据,那么就判断是否在两秒内


val iter = loginState.get().iterator()


if ( iter.hasNext ){


val firstFail = iter.next()


// 如果两次登录失败时间间隔小于 2 秒,输出报警


if ( value.eventTime < firstFail.eventTime + 2 ){


out.collect( Warning( value.userId, firstFail.eventTime, value.eventTime,


"login fail in 2 seconds." ) )


}


// 把最近一次的登录失败数据,更新写入 state 中


val failList = new util.ArrayListLoginEvent


failList.add(value)


loginState.update( failList )


} else {


// 如果 state 中没有登录失败的数据,那就直接添加进去


loginState.add(value)


}


} else


loginState.clear()


}


更多 Java –大数据 – 前端 – UI/UE - Android - 人工智能资料下载,可访问百度:尚硅谷官网(www.atguigu.com)


CEP 编程


上一节我们通过对状态编程的改进,去掉了定时器,在 process function 中做了更多的逻辑处理,实现了最初的需求。不过这种方法里有很多的条件判断,而我们目前仅仅实现的是检测“连续 2 次登录失败”,这是最简单的情形。如果需要检测更多次,内部逻辑显然会变得非常复杂。那有什么方式可以方便地实现呢?


很幸运,flink 为我们提供了 CEP(Complex Event Processing,复杂事件处理)库,用于在流中筛选符合某种复杂模式的事件。接下来我们就基于 CEP 来完成这个模块的实现。


在 src/main/scala 下继续创建 LoginFailWithCep.scala 文件,新建一个单例对象。样例类 LoginEvent 由于在 LoginFail.scala 已经定义,我们在同一个模块中就不需要再定义了。


代码如下:


LoginFailDetect/src/main/scala/LoginFailWithCep.scala


object LoginFailWithCep {


def main(args: Array[String]): Unit = {


val env = StreamExecutionEnvironment.getExecutionEnvironment


env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)


env.setParallelism(1)


val loginEventStream = env.readTextFile("YOUR_PATH\resources\LoginLog.csv")


.map( data => {


val dataArray = data.split(",")


LoginEvent(dataArray(0).toLong, dataArray(1), dataArray(2),


dataArray(3).toLong)


})


.assignTimestampsAndWatermarks(new


BoundedOutOfOrdernessTimestampExtractor[ApacheLogEvent]


(Time.milliseconds(3000)) {


override def extractTimestamp(element: ApacheLogEvent): Long ={


element.eventTime * 1000L


}


})


更多 Java –大数据 – 前端 – UI/UE - Android - 人工智能资料下载,可访问百度:尚硅谷官网(www.atguigu.com)


// 定义匹配模式


val loginFailPattern = Pattern.beginLoginEvent


.where(_.eventType == "fail")


.next("next")


.where(_.eventType == "fail")


.within(Time.seconds(2))


// 在数据流中匹配出定义好的模式


val patternStream = CEP.pattern(loginEventStream.keyBy(_.userId), loginFailPattern)


// .select 方法传入一个 pattern select function,当检测到定义好的模式序列时就会调用


val loginFailDataStream = patternStream


.select((pattern: Map[String, Iterable[LoginEvent]]) => {


val first = pattern.getOrElse("begin", null).iterator.next()


val second = pattern.getOrElse("next", null).iterator.next()


(second.userId, second.ip, second.eventType)


})


// 将匹配到的符合条件的事件打印出来


loginFailDataStream.print()


env.execute("Login Fail Detect Job")


}


}


用户头像

@零度

关注

关注尚硅谷,轻松学IT 2021.11.23 加入

IT培训 www.atguigu.com

评论

发布
暂无评论
大数据培训 | Flink如何监控恶意登录_大数据_@零度_InfoQ写作社区