修改判定逻辑，增加基线敏感阈值作为判定条件。
This commit is contained in:
wanglihui
@@ -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");
|
||||
|
||||
@@ -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) {
|
||||
|
||||
@@ -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()
|
||||
);
|
||||
}
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user