修改判定逻辑，增加基线敏感阈值作为判定条件。

4 files changed, 30 insertions, 19 deletions

diff --git a/src/main/java/com/zdjizhi/common/CommonConfig.java b/src/main/java/com/zdjizhi/common/CommonConfig.java
index 361fe9e..7ca131b 100644
--- a/src/main/java/com/zdjizhi/common/CommonConfig.java
+++ b/src/main/java/com/zdjizhi/common/CommonConfig.java
@@ -39,7 +39,8 @@ public class CommonConfig {
 
     public static final String IP_MMDB_PATH = CommonConfigurations.getStringProperty("ip.mmdb.path");
 
-    public static final int SENSITIVITY_THRESHOLD = CommonConfigurations.getIntProperty("sensitivity.threshold");
+    public static final int STATIC_SENSITIVITY_THRESHOLD = CommonConfigurations.getIntProperty("static.sensitivity.threshold");
+    public static final double BASELINE_SENSITIVITY_THRESHOLD = CommonConfigurations.getDoubleProperty("baseline.sensitivity.threshold");
 
     public static final double BASELINE_SESSIONS_MINOR_THRESHOLD = CommonConfigurations.getDoubleProperty("baseline.sessions.minor.threshold");
     public static final double BASELINE_SESSIONS_WARNING_THRESHOLD = CommonConfigurations.getDoubleProperty("baseline.sessions.warning.threshold");
diff --git a/src/main/java/com/zdjizhi/etl/DosDetection.java b/src/main/java/com/zdjizhi/etl/DosDetection.java
index 42bbbd7..815cc05 100644
--- a/src/main/java/com/zdjizhi/etl/DosDetection.java
+++ b/src/main/java/com/zdjizhi/etl/DosDetection.java
@@ -85,8 +85,8 @@ public class DosDetection extends RichMapFunction<DosSketchLog, DosEventLog> {
     private DosEventLog getDosEventLogBySensitivityThreshold(DosSketchLog value){
         DosEventLog result = null;
         long sketchSessions = value.getSketch_sessions();
-        if (sketchSessions > CommonConfig.SENSITIVITY_THRESHOLD){
-            result = getDosEventLog(value, CommonConfig.SENSITIVITY_THRESHOLD, sketchSessions - CommonConfig.SENSITIVITY_THRESHOLD, "sensitivity");
+        if (sketchSessions > CommonConfig.STATIC_SENSITIVITY_THRESHOLD){
+            result = getDosEventLog(value, CommonConfig.STATIC_SENSITIVITY_THRESHOLD, sketchSessions - CommonConfig.STATIC_SENSITIVITY_THRESHOLD, "sensitivity");
             result.setSeverity(Severity.MAJOR.severity);
         }
         return result;
@@ -95,7 +95,7 @@ public class DosDetection extends RichMapFunction<DosSketchLog, DosEventLog> {
     private DosEventLog getDosEventLogByBaseline(DosSketchLog value, String destinationIp, String attackType) {
         DosEventLog result = null;
         long sketchSessions = value.getSketch_sessions();
-        if (sketchSessions > CommonConfig.SENSITIVITY_THRESHOLD){
+        if (sketchSessions > CommonConfig.STATIC_SENSITIVITY_THRESHOLD){
             Tuple2<ArrayList<Integer>, Integer> floodTypeTup = baselineMap.get(destinationIp).get(attackType);
             Integer base = getBaseValue(floodTypeTup, value);
             result = getDosEventLog(value, base, sketchSessions - base, "baseline");
@@ -123,23 +123,27 @@ public class DosDetection extends RichMapFunction<DosSketchLog, DosEventLog> {
             double percent = getDiffPercent(diff, base);
             Severity severity = judgeSeverity(percent);
             if (severity != Severity.NORMAL) {
-                result = getResult(value, severity, percent, tag);
-                logger.info("检测到当前server IP {} 存在 {} 异常,超出基线{} {}倍,日志详情\n {}", destinationIp,attackType,base,percent,result);
+                if ("baseline".equals(tag) && percent < CommonConfig.BASELINE_SENSITIVITY_THRESHOLD){
+                    logger.debug("当前server IP:{},类型:{},基线值{}百分比{}未超过基线敏感阈值,日志详情\n{}",destinationIp,attackType,base,percent,value);
+                }else {
+                    result = getResult(value,base, severity, percent, tag);
+                    logger.info("检测到当前server IP {} 存在 {} 异常,超出基线{} {}倍,日志详情\n {}", destinationIp,attackType,base,percent,result);
+                }
             } else {
-                logger.debug("当前server IP：{} 未出现 {} 异常,日志详情 {}", destinationIp, attackType, value.toString());
+                logger.debug("当前server IP:{} 未出现 {} 异常,日志详情 {}", destinationIp, attackType, value);
             }
         }
         return result;
     }
 
-    private DosEventLog getResult(DosSketchLog value, Severity severity, double percent, String tag) {
+    private DosEventLog getResult(DosSketchLog value,long base, Severity severity, double percent, String tag) {
         DosEventLog dosEventLog = new DosEventLog();
         dosEventLog.setLog_id(SnowflakeId.generateId());
         dosEventLog.setStart_time(value.getSketch_start_time());
         dosEventLog.setEnd_time(value.getSketch_start_time() + CommonConfig.FLINK_WINDOW_MAX_TIME);
         dosEventLog.setAttack_type(value.getAttack_type());
         dosEventLog.setSeverity(severity.severity);
-        dosEventLog.setConditions(getConditions(PERCENT_INSTANCE.format(percent), value.getSketch_sessions(), tag));
+        dosEventLog.setConditions(getConditions(PERCENT_INSTANCE.format(percent),base, value.getSketch_sessions(), tag));
         dosEventLog.setDestination_ip(value.getDestination_ip());
         dosEventLog.setDestination_country(IpUtils.ipLookup.countryLookup(value.getDestination_ip()));
         String ipList = value.getSource_ip();
@@ -172,12 +176,12 @@ public class DosDetection extends RichMapFunction<DosSketchLog, DosEventLog> {
         return base;
     }
 
-    private String getConditions(String percent, long sessions, String tag) {
+    private String getConditions(String percent,long base, long sessions, String tag) {
         switch (tag) {
             case "baseline":
                 return "sessions > " + percent + " of baseline";
             case "static":
-                return "sessions > " + sessions + " sessions/s";
+                return "sessions > " + base + " sessions/s";
             case "sensitivity":
                 return sessions+" sessions/s Unusually high Sessions";
             default:
@@ -206,6 +210,7 @@ public class DosDetection extends RichMapFunction<DosSketchLog, DosEventLog> {
         Date p1D = DateUtils.getTimeFloor(date, "P1D");
         System.out.println(p1D+" "+p1D.getTime()/1000);
         System.out.println(new DosDetection().getCurrentTimeIndex(1631548860));
+        System.out.println(10+10*0.2);
     }
 
     private Double getDiffPercent(long diff, long base) {
diff --git a/src/main/java/com/zdjizhi/utils/KafkaUtils.java b/src/main/java/com/zdjizhi/utils/KafkaUtils.java
index 6042b37..6e6167a 100644
--- a/src/main/java/com/zdjizhi/utils/KafkaUtils.java
+++ b/src/main/java/com/zdjizhi/utils/KafkaUtils.java
@@ -4,6 +4,7 @@ import com.zdjizhi.common.CommonConfig;
 import org.apache.flink.api.common.serialization.SimpleStringSchema;
 import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
 
+import java.util.Optional;
 import java.util.Properties;
 
 public class KafkaUtils {
@@ -24,7 +25,8 @@ public class KafkaUtils {
         return new FlinkKafkaProducer<String>(
                 topic,
                 new SimpleStringSchema(),
-                getKafkaSinkProperty()
+                getKafkaSinkProperty(),
+                Optional.empty()
         );
     }
 
diff --git a/src/main/resources/common.properties b/src/main/resources/common.properties
index b700e2c..75f5015 100644
--- a/src/main/resources/common.properties
+++ b/src/main/resources/common.properties
@@ -15,7 +15,7 @@ kafka.input.topic.name=DOS-SKETCH-RECORD
 kafka.input.bootstrap.servers=192.168.44.11:9094,192.168.44.14:9094,192.168.44.15:9094
 
 #读取kafka group id
-kafka.input.group.id=2108231709
+kafka.input.group.id=2109160928
 #kafka.input.group.id=dos-detection-job-210813-1
 
 #发送kafka metrics并行度大小
@@ -57,10 +57,10 @@ flink.first.agg.parallelism=1
 flink.detection.map.parallelism=1
 
 #watermark延迟
-flink.watermark.max.orderness=1
+flink.watermark.max.orderness=10
 
 #计算窗口大小，默认600s
-flink.window.max.time=10
+flink.window.max.time=600
 
 #dos event结果中distinct source IP限制
 source.ip.list.limit=10000
@@ -75,18 +75,21 @@ ip.mmdb.path=D:\\data\\dat\\
 #ip.mmdb.path=/home/bigdata/topology/dat/
 #ip.mmdb.path=/home/bigdata/wlh/topology/dos-detection/dat/
 
-#敏感阈值，速率小于此值不报警
-sensitivity.threshold=1
+#静态敏感阈值，速率小于此值不报警
+static.sensitivity.threshold=100
+
+#基线敏感阈值
+baseline.sensitivity.threshold=0.2
 
 #基于baseline判定dos攻击的上下限
-baseline.sessions.minor.threshold=0.1
+baseline.sessions.minor.threshold=0.2
 baseline.sessions.warning.threshold=0.5
 baseline.sessions.major.threshold=1
 baseline.sessions.severe.threshold=3
 baseline.sessions.critical.threshold=8
 
 #bifang服务访问地址
-bifang.server.uri=http://192.168.44.3:80
+bifang.server.uri=http://192.168.44.72:80
 
 #访问bifang只读权限token，bifang内置，无需修改
 bifang.server.token=ed04b942-7df4-4e3d-b9a9-a881ca98a867