如何使用Flink CEP实施模式以匹配暴力登录和端口扫描攻击

如何解决如何使用Flink CEP实施模式以匹配暴力登录和端口扫描攻击

我有一个用例，其中大量日志将被消耗到apache flink CEP。我的用例是找到蛮力攻击和端口扫描攻击。这里的挑战是，在普通CEP中，我们将值与常量进行比较，例如“ event” = login。在这种情况下，条件与在蛮力攻击中不同，我们具有以下条件。

用户名是常量，并且event =“ login failure”（分隔符，该事件在5分钟内发生5次）。 这意味着在5分钟内5次收到相同用户名的带有登录失败事件的日志

对于端口扫描，我们具有以下criteira。

ip地址是恒定的，dest端口是可变的（定界符是事件在1分钟内发生10次）。这意味着在1分钟内会收到10个不同端口的IP地址不变的日志。

解决方法

使用Flink，当您想单独处理一个用户名或一个IP地址之类的事件时，执行此操作的方法是使用keyBy()通过密钥对流进行分区。 Flink文档中的培训材料带有section on Keyed Streams，可更详细地说明DataStream API的这一部分。如果有帮助，keyBy()与SQL中的GROUP BY大致相同。

使用CEP，如果您首先为流设置密钥，则将针对密钥的每个不同值分别匹配模式。

但是，对于此用例，我建议使用Flink SQL（而不是CEP），或者与MATCH_RECOGNIZE组合使用。 MATCH_RECOGNIZE是基于CEP构建的更高级别的API，使用起来更容易。与SQL结合使用，结果非常强大。

您将找到一些Flink SQL培训材料和示例（包括使用MATCH_RECOGNIZE的示例）in Ververica's github account。

更新

需要明确的是，我不会将MATCH_RECOGNIZE用于这些特定规则；此用例既不需要它，也不需要CEP。我提到它是为了防止其他规则对您有所帮助。 （在这种情况下，我不建议使用CEP的原因是，实施不同的约束可能很麻烦。）

例如，对于端口扫描，您可以执行以下操作：

SELECT e1.ip,COUNT(DISTINCT e2.port) 
FROM events e1,events e2 
WHERE e1.ip = e2.ip AND timestampDiff(MINUTE,e1.ts,e2.ts) < 1 
GROUP BY e1.ip HAVING COUNT(DISTINCT e2.port) >= 10;

登录情况类似，但更简单。

请注意，在使用流式SQL时，应考虑state retention。

进一步更新

此查询可能会多次返回给定的IP地址，但不希望生成多个警报。

这可以通过将匹配的IP地址插入“警报”表中来解决，并且仅针对尚未存在的IP生成警报。

或者SQL查询的输出可以由使用DataStream API（类似于example in the Flink docs）实现的重复数据删除器处理。如果只想在一段时间内禁止重复警报，请使用KeyedProcessFunction而不是RichFlatMapFunction，并在需要为给定IP重新启用警报时使用计时器清除状态

又一次更新（关于CEP和清晰度）

应该可以使用CEP来实现。您需要按IP地址键入流，并具有必须在一分钟内匹配的模式。

模式大致如下：

Pattern<Event,?> pattern = Pattern
  .<Event>begin("distinctPorts")
  .where(iterative condition 1)
  .oneOrMore()
  .followedBy("end")
  .where(iterative condition 2)
  .within(1 minute)

如果要添加到模式的事件与所有先前匹配的事件具有不同的端口，则第一个迭代条件返回true。类似于示例here,in the docs。

如果size("distinctPorts") >= 9，则第二个迭代条件返回true，并且此事件还具有另一个不同的端口。

请参见this Flink Forward talk（youtube video），以在演讲结束时获得一个类似的示例。

如果您尝试这样做并被卡住，请提出一个新问题，向我们展示您尝试过的内容以及被卡住的地方。