gfwleak/galaxy-tsg-olap-dos-detection-job
Archived
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

修复读取配置IP冲突问题

Browse Source
This commit is contained in:
wanglihui
2021-09-23 18:36:27 +08:00
parent e930fa23ed
commit 77bc6a844e
5 changed files with 47 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
pom.xml
View File

@@ -149,6 +149,7 @@
<groupId>org.apache.hbase</groupId>
<artifactId>hbase-client</artifactId>
<version>2.2.3</version>
<!--<scope>provided</scope>-->
<exclusions>
<exclusion>
<artifactId>slf4j-log4j12</artifactId>

22
src/main/java/com/zdjizhi/etl/DosDetection.java
View File

@@ -42,15 +42,11 @@ public class DosDetection extends RichMapFunction<DosSketchLog, DosEventLog> {
ScheduledExecutorService executorService = new ScheduledThreadPoolExecutor(2,
new BasicThreadFactory.Builder().namingPattern("Dos-Detection-%d").daemon(true).build());
try {
executorService.scheduleAtFixedRate(() -> {
//do something
thresholdRangeMap = ParseStaticThreshold.createStaticThreshold();
}, 0, CommonConfig.STATIC_THRESHOLD_SCHEDULE_MINUTES, TimeUnit.MINUTES);
executorService.scheduleAtFixedRate(() -> thresholdRangeMap = ParseStaticThreshold.createStaticThreshold(), 0,
CommonConfig.STATIC_THRESHOLD_SCHEDULE_MINUTES, TimeUnit.MINUTES);
executorService.scheduleAtFixedRate(() -> {
//do something
baselineMap = HbaseUtils.readFromHbase();
}, 0, CommonConfig.BASELINE_THRESHOLD_SCHEDULE_DAYS, TimeUnit.DAYS);
executorService.scheduleAtFixedRate(() -> baselineMap = HbaseUtils.readFromHbase(), 0,
CommonConfig.BASELINE_THRESHOLD_SCHEDULE_DAYS, TimeUnit.DAYS);
} catch (Exception e) {
logger.error("定时器任务执行失败", e);
}
@@ -67,10 +63,10 @@ public class DosDetection extends RichMapFunction<DosSketchLog, DosEventLog> {
Map<String, DosDetectionThreshold> thresholdMap = thresholdRangeMap.get(destinationIpAddress);
logger.debug("当前判断IP：{}, 类型: {}", destinationIp, attackType);
if (thresholdMap == null && baselineMap.containsKey(destinationIp)) {
finalResult = getDosEventLogByBaseline(value, destinationIp, attackType);
finalResult = getDosEventLogByBaseline(value);
}else if (thresholdMap == null && !baselineMap.containsKey(destinationIp)){
finalResult = getDosEventLogBySensitivityThreshold(value);
} else if (thresholdMap != null){
}else if (thresholdMap != null){
finalResult = getDosEventLogByStaticThreshold(value, thresholdMap);
}else {
logger.debug("未获取到当前server IP：{} 类型 {} 静态阈值 和 baseline", destinationIp, attackType);
@@ -92,8 +88,10 @@ public class DosDetection extends RichMapFunction<DosSketchLog, DosEventLog> {
return result;
}
private DosEventLog getDosEventLogByBaseline(DosSketchLog value, String destinationIp, String attackType) {
private DosEventLog getDosEventLogByBaseline(DosSketchLog value) {
DosEventLog result = null;
String destinationIp = value.getDestination_ip();
String attackType = value.getAttack_type();
long sketchSessions = value.getSketch_sessions();
if (sketchSessions > CommonConfig.STATIC_SENSITIVITY_THRESHOLD){
Tuple2<ArrayList<Integer>, Integer> floodTypeTup = baselineMap.get(destinationIp).get(attackType);
@@ -127,7 +125,7 @@ public class DosDetection extends RichMapFunction<DosSketchLog, DosEventLog> {
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);
logger.info("检测到当前server IP {} 存在 {} 异常,超出基线{} {}倍,基于{}检测,日志详情\n {}", destinationIp,attackType,base,percent,tag,result);
}
} else {
logger.debug("当前server IP:{} 未出现 {} 异常,日志详情 {}", destinationIp, attackType, value);

42
src/main/java/com/zdjizhi/etl/ParseStaticThreshold.java
View File

@@ -16,6 +16,7 @@ import org.slf4j.LoggerFactory;
import java.net.URISyntaxException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
@@ -109,11 +110,14 @@ public class ParseStaticThreshold {
URIBuilder uriBuilder = new URIBuilder(CommonConfig.BIFANG_SERVER_URI);
HashMap<String, Object> parms = new HashMap<>();
parms.put("pageSize",-1);
parms.put("orderBy","profileId asc");
parms.put("isValid",1);
HttpClientUtils.setUrlWithParams(uriBuilder, CommonConfig.BIFANG_SERVER_POLICY_THRESHOLD_PATH, parms);
String token = CommonConfig.BIFANG_SERVER_TOKEN;
if (!HttpClientUtils.ERROR_MESSAGE.equals(token)) {
BasicHeader authorization = new BasicHeader("Authorization", token);
String resposeJsonStr = HttpClientUtils.httpGet(uriBuilder.build(), authorization);
BasicHeader authorization1 = new BasicHeader("Content-Type", "application/x-www-form-urlencoded");
String resposeJsonStr = HttpClientUtils.httpGet(uriBuilder.build(), authorization,authorization1);
if (!HttpClientUtils.ERROR_MESSAGE.equals(resposeJsonStr)) {
HashMap<String, Object> resposeMap = jsonMapperInstance.fromJson(resposeJsonStr, hashmapJsonType);
boolean success = (boolean) resposeMap.get("success");
@@ -149,17 +153,31 @@ public class ParseStaticThreshold {
IPAddressString ipAddressString = new IPAddressString(sip);
if (ipAddressString.isIPAddress()) {
IPAddress address = ipAddressString.getAddress();
Map<String, DosDetectionThreshold> floodTypeThresholdMap = thresholdRangeMap.get(address);
if (floodTypeThresholdMap == null) {
floodTypeThresholdMap = new HashMap<>();
}
Map<String, DosDetectionThreshold> floodTypeThresholdMap = new HashMap<>();
floodTypeThresholdMap.put(threshold.getAttackType(), threshold);
if (address.isPrefixed()){
if (address.isMultiple()){
thresholdRangeMap.put(Range.closed(address.getLower(), address.getUpper()), floodTypeThresholdMap);
}else {
thresholdRangeMap.put(Range.closed(address.adjustPrefixLength(address.getBitCount()),
address.toMaxHost().withoutPrefixLength()), floodTypeThresholdMap);
IPAddress lower = address.getLower();
IPAddress upper = address.getUpper();
if (!address.isMultiple()){
lower = address.adjustPrefixLength(address.getBitCount());
upper = address.toMaxHost().withoutPrefixLength();
}
Map.Entry<Range<IPAddress>, Map<String, DosDetectionThreshold>> lowerEntry = thresholdRangeMap.getEntry(lower);
Map.Entry<Range<IPAddress>, Map<String, DosDetectionThreshold>> upperEntry = thresholdRangeMap.getEntry(upper);
if (lowerEntry == null && upperEntry == null){
thresholdRangeMap.put(Range.closed(lower, upper), floodTypeThresholdMap);
}else if (lowerEntry != null && upperEntry == null){
Range<IPAddress> lowerEntryKey = lowerEntry.getKey();
Map<String, DosDetectionThreshold> lowerEntryValue = lowerEntry.getValue();
lowerEntryValue.put(threshold.getAttackType(), threshold);
thresholdRangeMap.put(Range.closedOpen(lowerEntryKey.lowerEndpoint(), lower), lowerEntryValue);
thresholdRangeMap.put(Range.closed(lower, upper), floodTypeThresholdMap);
}else if (lowerEntry == null){
Range<IPAddress> upperEntryKey = upperEntry.getKey();
Map<String, DosDetectionThreshold> upperEntryValue = upperEntry.getValue();
upperEntryValue.put(threshold.getAttackType(), threshold);
thresholdRangeMap.put(Range.openClosed(upper, upperEntryKey.upperEndpoint()), upperEntryValue);
thresholdRangeMap.put(Range.closed(lower, upper), floodTypeThresholdMap);
}
}else {
thresholdRangeMap.put(Range.closed(address, address), floodTypeThresholdMap);
@@ -183,7 +201,7 @@ public class ParseStaticThreshold {
System.out.println("------------------------");
TreeRangeMap<IPAddress, Map<String, DosDetectionThreshold>> staticThreshold = createStaticThreshold();
/*
System.out.println("------------------------");
Map<Range<IPAddress>, Map<String, DosDetectionThreshold>> rangeMapMap = staticThreshold.asMapOfRanges();
for (Range<IPAddress> range : rangeMapMap.keySet()) {
Map<String, DosDetectionThreshold> thresholdMap = rangeMapMap.get(range);
@@ -192,7 +210,7 @@ public class ParseStaticThreshold {
System.out.println(range + "---" + type + "---" + threshold);
}
}
*/
}

4
src/main/resources/common.properties
View File

@@ -33,7 +33,7 @@ kafka.output.event.parallelism=1
kafka.output.event.topic.name=storm-dos-test
#kafka输出地址
kafka.output.bootstrap.servers=192.168.44.12:9092
kafka.output.bootstrap.servers=192.168.44.12:9094
#kafka.output.bootstrap.servers=192.168.44.11:9092,192.168.44.14:9092,192.168.44.15:9092
#zookeeper地址
@@ -60,7 +60,7 @@ flink.detection.map.parallelism=1
flink.watermark.max.orderness=10
#计算窗口大小，默认600s
flink.window.max.time=600
flink.window.max.time=10
#dos event结果中distinct source IP限制
source.ip.list.limit=10000

6
src/test/java/com/zdjizhi/common/IpTest.java
View File

@@ -41,8 +41,8 @@ public class IpTest {
IPAddress pv43 = new IPAddressString("fc00::").getAddress();
IPAddress pv44 = new IPAddressString("fc00::10:1").getAddress();
IPAddress pv45 = new IPAddressString("12.56.4.3/24").getAddress();
IPAddress pv46 = new IPAddressString("12.56.4.0/24").getAddress();
IPAddress pv45 = new IPAddressString("192.168.42.1/32").getAddress();
IPAddress pv46 = new IPAddressString("192.168.42.1/32").getAddress();
IPAddress pv47 = new IPAddressString("12.56.4.0").getAddress();
System.out.println(pv45.isMultiple());
System.out.println(pv46.isMultiple());
@@ -50,6 +50,8 @@ public class IpTest {
System.out.println(pv47.isPrefixed());
System.out.println(pv45+"---"+pv45.toMaxHost().withoutPrefixLength()+"---"+pv45.adjustPrefixLength(pv45.getBitCount()));
System.out.println(pv45.adjustPrefixLength(pv45.getBitCount())+"---"+pv45.toMaxHost().withoutPrefixLength());
/*
System.out.println(str5.getUpper()+"---"+str5.getLower());