You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spot.apache.org by ev...@apache.org on 2017/01/25 18:36:47 UTC
[35/49] incubator-spot git commit: Merge branch 'spot' into
unit_test_cleanup
Merge branch 'spot' into unit_test_cleanup
# Conflicts:
# spot-ml/src/main/scala/org/apache/spot/SuspiciousConnects.scala
# spot-ml/src/main/scala/org/apache/spot/dns/DNSSuspiciousConnectsAnalysis.scala
# spot-ml/src/main/scala/org/apache/spot/netflow/FlowSuspiciousConnectsAnalysis.scala
# spot-ml/src/main/scala/org/apache/spot/netflow/FlowWordCreator.scala
# spot-ml/src/main/scala/org/apache/spot/proxy/ProxySuspiciousConnectsAnalysis.scala
# spot-ml/src/test/scala/org/apache/spot/dns/DNSSuspiciousConnectsAnalysisTest.scala
# spot-ml/src/test/scala/org/apache/spot/dns/DNSWordCreationTest.scala
# spot-ml/src/test/scala/org/apache/spot/proxy/ProxySuspiciousConnectsAnalysisTest.scala
Project: http://git-wip-us.apache.org/repos/asf/incubator-spot/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spot/commit/24b3a37b
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spot/tree/24b3a37b
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spot/diff/24b3a37b
Branch: refs/heads/master
Commit: 24b3a37bfaf8bab21015cb9da40eb2897e83f9df
Parents: 9ac5a8c
Author: nlsegerl <na...@intel.com>
Authored: Wed Jan 4 15:32:32 2017 -0800
Committer: nlsegerl <na...@intel.com>
Committed: Wed Jan 4 15:32:32 2017 -0800
----------------------------------------------------------------------
.../spot/SuspiciousConnectsScoreFunction.scala | 14 ++-
.../scala/org/apache/spot/dns/DNSSchema.scala | 5 -
.../dns/DNSSuspiciousConnectsAnalysis.scala | 21 ++--
.../org/apache/spot/dns/DNSWordCreation.scala | 16 +--
.../dns/model/DNSSuspiciousConnectsModel.scala | 70 +++++------
.../FlowSuspiciousConnectsAnalysis.scala | 19 ++-
.../apache/spot/netflow/FlowWordCreator.scala | 10 +-
.../spot/netflow/model/FlowScoreFunction.scala | 69 +++++------
.../model/FlowSuspiciousConnectsModel.scala | 64 ++++++-----
.../proxy/ProxySuspiciousConnectsModel.scala | 30 ++---
.../utilities/data/InputOutputDataHandler.scala | 34 +++---
.../dns/DNSSuspiciousConnectsAnalysisTest.scala | 49 ++++----
.../apache/spot/dns/DNSWordCreationTest.scala | 1 -
.../FlowSuspiciousConnectsAnalysisTest.scala | 115 ++++++++++++++++++-
14 files changed, 314 insertions(+), 203 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/24b3a37b/spot-ml/src/main/scala/org/apache/spot/SuspiciousConnectsScoreFunction.scala
----------------------------------------------------------------------
diff --git a/spot-ml/src/main/scala/org/apache/spot/SuspiciousConnectsScoreFunction.scala b/spot-ml/src/main/scala/org/apache/spot/SuspiciousConnectsScoreFunction.scala
index 04db60e..92cfea8 100644
--- a/spot-ml/src/main/scala/org/apache/spot/SuspiciousConnectsScoreFunction.scala
+++ b/spot-ml/src/main/scala/org/apache/spot/SuspiciousConnectsScoreFunction.scala
@@ -3,8 +3,15 @@ package org.apache.spot
import org.apache.spark.broadcast.Broadcast
import org.apache.spot.utilities.data.validation.InvalidDataHandler
-
-
+/**
+ * Base class for scoring suspicious connects models.
+ * Assumes that distribution of words is independent of the IP when conditioned on the topic
+ * and performs a simple sum over a partition of the space by topic.
+ *
+ * @param topicCount Number of topics produced by the topic modelling analysis.
+ * @param ipToTopicMixBC Broadcast of map assigning IPs to topic mixes.
+ * @param wordToPerTopicProbBC Broadcast of map assigning words to per-topic conditional probability.
+ */
class SuspiciousConnectsScoreFunction(topicCount: Int,
ipToTopicMixBC: Broadcast[Map[String, Array[Double]]],
wordToPerTopicProbBC: Broadcast[Map[String, Array[Double]]]) extends Serializable {
@@ -25,5 +32,4 @@ class SuspiciousConnectsScoreFunction(topicCount: Int,
.sum
}
}
-
-}
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/24b3a37b/spot-ml/src/main/scala/org/apache/spot/dns/DNSSchema.scala
----------------------------------------------------------------------
diff --git a/spot-ml/src/main/scala/org/apache/spot/dns/DNSSchema.scala b/spot-ml/src/main/scala/org/apache/spot/dns/DNSSchema.scala
index 020fc46..2244085 100644
--- a/spot-ml/src/main/scala/org/apache/spot/dns/DNSSchema.scala
+++ b/spot-ml/src/main/scala/org/apache/spot/dns/DNSSchema.scala
@@ -39,7 +39,6 @@ object DNSSchema {
val AnswerAddress = "dns_a"
val AnswerAddressField = StructField(AnswerAddress, StringType, nullable= true)
-
// intermediate and derived fields
val Domain = "domain"
@@ -67,8 +66,4 @@ object DNSSchema {
val Score = "score"
val ScoreField = StructField(Score, DoubleType)
-
-
- val ScoreSchema = StructType(List(ScoreField))
-
}
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/24b3a37b/spot-ml/src/main/scala/org/apache/spot/dns/DNSSuspiciousConnectsAnalysis.scala
----------------------------------------------------------------------
diff --git a/spot-ml/src/main/scala/org/apache/spot/dns/DNSSuspiciousConnectsAnalysis.scala b/spot-ml/src/main/scala/org/apache/spot/dns/DNSSuspiciousConnectsAnalysis.scala
index 929b69e..1688bed 100644
--- a/spot-ml/src/main/scala/org/apache/spot/dns/DNSSuspiciousConnectsAnalysis.scala
+++ b/spot-ml/src/main/scala/org/apache/spot/dns/DNSSuspiciousConnectsAnalysis.scala
@@ -8,14 +8,13 @@ import org.apache.spark.sql.{DataFrame, SQLContext}
import org.apache.spot.SuspiciousConnectsArgumentParser.SuspiciousConnectsConfig
import org.apache.spot.dns.DNSSchema._
import org.apache.spot.dns.model.DNSSuspiciousConnectsModel
-import org.apache.log4j.Logger
import org.apache.spot.dns.model.DNSSuspiciousConnectsModel.ModelSchema
import org.apache.spot.proxy.ProxySchema.Score
import org.apache.spot.utilities.data.validation.{InvalidDataHandler => dataValidation}
/**
* The suspicious connections analysis of DNS log data develops a probabilistic model the DNS queries
- * made by each client IP and flags
+ * made by each client IP and flags those assigned a low probability as "suspicious"
*/
object DNSSuspiciousConnectsAnalysis {
@@ -37,13 +36,9 @@ object DNSSuspiciousConnectsAnalysis {
logger.info("Starting DNS suspicious connects analysis.")
- val userDomain = config.userDomain
-
val cleanDNSRecords = filterAndSelectCleanDNSRecords(inputDNSRecords)
- logger.info("Training the model")
-
- val scoredDNSRecords = detectDNSAnomalies(cleanDNSRecords, config, sparkContext, sqlContext, logger)
+ val scoredDNSRecords = scoreDNSRecords(cleanDNSRecords, config, sparkContext, sqlContext, logger)
val filteredDNSRecords = filterScoredDNSRecords(scoredDNSRecords, config.threshold)
@@ -55,6 +50,7 @@ object DNSSuspiciousConnectsAnalysis {
logger.info("DNS suspicious connects analysis completed.")
logger.info("Saving results to : " + config.hdfsScoredConnect)
+
outputDNSRecords.map(_.mkString(config.outputDelimiter)).saveAsTextFile(config.hdfsScoredConnect)
val invalidDNSRecords = filterAndSelectInvalidDNSRecords(inputDNSRecords)
@@ -76,18 +72,17 @@ object DNSSuspiciousConnectsAnalysis {
* @return
*/
- def detectDNSAnomalies(data: DataFrame, config: SuspiciousConnectsConfig,
- sparkContext: SparkContext,
- sqlContext: SQLContext,
- logger: Logger) : DataFrame = {
+ def scoreDNSRecords(data: DataFrame, config: SuspiciousConnectsConfig,
+ sparkContext: SparkContext,
+ sqlContext: SQLContext,
+ logger: Logger) : DataFrame = {
- val userDomain = config.userDomain
logger.info("Fitting probabilistic model to data")
val model =
DNSSuspiciousConnectsModel.trainNewModel(sparkContext, sqlContext, logger, config, data, config.topicCount)
logger.info("Identifying outliers")
- model.score(sparkContext, sqlContext, data, userDomain)
+ model.score(sparkContext, sqlContext, data, config.userDomain)
}
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/24b3a37b/spot-ml/src/main/scala/org/apache/spot/dns/DNSWordCreation.scala
----------------------------------------------------------------------
diff --git a/spot-ml/src/main/scala/org/apache/spot/dns/DNSWordCreation.scala b/spot-ml/src/main/scala/org/apache/spot/dns/DNSWordCreation.scala
index 383eb2f..e080b3e 100644
--- a/spot-ml/src/main/scala/org/apache/spot/dns/DNSWordCreation.scala
+++ b/spot-ml/src/main/scala/org/apache/spot/dns/DNSWordCreation.scala
@@ -6,19 +6,19 @@ import org.apache.spot.utilities.DomainProcessor.{DomainInfo, extractDomainInfo}
import org.apache.spot.utilities.Quantiles
import org.apache.spot.utilities.data.validation.InvalidDataHandler
-import scala.util.{Failure, Success, Try}
+import scala.util.{Success, Try}
/**
* Convert DNS log entries into "words" for topic modelling analyses.
*
- * @param frameLengthCuts
- * @param timeCuts
- * @param subdomainLengthCuts
- * @param entropyCuts
- * @param numberPeriodsCuts
- * @param topDomainsBC
- * @param userDomain
+ * @param frameLengthCuts Quantile cut-offs for discretizing frame length in word construction.
+ * @param timeCuts Quantile cut-offs for discretizing the time of day in word construction.
+ * @param subdomainLengthCuts Quantile cut-offs for discretizing subdomain length in word construction.
+ * @param entropyCuts Quantile cut-offs for discretizing entropy in word construction.
+ * @param numberPeriodsCuts Quantile cut-offs for discretizing the number of periods in word construction.
+ * @param topDomainsBC List of most popular top level domain names.
+ * @param userDomain User's domain for internal network.
*/
class DNSWordCreation(frameLengthCuts: Array[Double],
timeCuts: Array[Double],
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/24b3a37b/spot-ml/src/main/scala/org/apache/spot/dns/model/DNSSuspiciousConnectsModel.scala
----------------------------------------------------------------------
diff --git a/spot-ml/src/main/scala/org/apache/spot/dns/model/DNSSuspiciousConnectsModel.scala b/spot-ml/src/main/scala/org/apache/spot/dns/model/DNSSuspiciousConnectsModel.scala
index 47c32a7..f1893e9 100644
--- a/spot-ml/src/main/scala/org/apache/spot/dns/model/DNSSuspiciousConnectsModel.scala
+++ b/spot-ml/src/main/scala/org/apache/spot/dns/model/DNSSuspiciousConnectsModel.scala
@@ -73,7 +73,6 @@ class DNSSuspiciousConnectsModel(inTopicCount: Int,
*/
def score(sc: SparkContext, sqlContext: SQLContext, inDF: DataFrame, userDomain: String): DataFrame = {
- val countryCodesBC = sc.broadcast(CountryCodes.CountryCodes)
val topDomainsBC = sc.broadcast(TopDomains.TopDomains)
val ipToTopicMixBC = sc.broadcast(ipToTopicMix)
val wordToPerTopicProbBC = sc.broadcast(wordToPerTopicProb)
@@ -139,10 +138,10 @@ object DNSSuspiciousConnectsModel {
* @param sparkContext
* @param sqlContext
* @param logger
- * @param config Analysis configuration object containing CLI parameters.
- * Contains the path to the feedback file in config.scoresFile
- * @param inputRecords Data used to train the model.
- * @param topicCount Number of topics (traffic profiles) used to build the model.
+ * @param config Analysis configuration object containing CLI parameters.
+ * Contains the path to the feedback file in config.scoresFile
+ * @param inputRecords Data used to train the model.
+ * @param topicCount Number of topics (traffic profiles) used to build the model.
* @return A new [[DNSSuspiciousConnectsModel]] instance trained on the dataframe and feedback file.
*/
def trainNewModel(sparkContext: SparkContext,
@@ -171,11 +170,12 @@ object DNSSuspiciousConnectsModel {
Quantiles.computeDeciles(totalRecords
.select(UnixTimestamp)
.rdd
- .flatMap({ case Row(unixTimeStamp: Long) => {
- Try {unixTimeStamp.toDouble} match {
- case Failure(_) => Seq()
- case Success(timestamp) => Seq(timestamp)
- }
+ .flatMap({ case Row(unixTimeStamp: Long) =>
+ Try {
+ unixTimeStamp.toDouble
+ } match {
+ case Failure(_) => Seq()
+ case Success(timestamp) => Seq(timestamp)
}
}))
@@ -183,11 +183,12 @@ object DNSSuspiciousConnectsModel {
Quantiles.computeDeciles(totalRecords
.select(FrameLength)
.rdd
- .flatMap({case Row(frameLen: Int) => {
- Try{frameLen.toDouble} match{
- case Failure(_) => Seq()
- case Success(frameLen) => Seq(frameLen)
- }
+ .flatMap({ case Row(frameLen: Int) =>
+ Try {
+ frameLen.toDouble
+ } match {
+ case Failure(_) => Seq()
+ case Success(frameLength) => Seq(frameLength)
}
}))
@@ -198,11 +199,12 @@ object DNSSuspiciousConnectsModel {
.filter(domainStatsRecords(SubdomainLength).gt(0))
.select(SubdomainLength)
.rdd
- .flatMap({ case Row(subdomainLength: Int) => {
- Try{subdomainLength.toDouble} match {
- case Failure(_) => Seq()
- case Success(subdomainLength) => Seq(subdomainLength)
- }
+ .flatMap({ case Row(subdomainLength: Int) =>
+ Try {
+ subdomainLength.toDouble
+ } match {
+ case Failure(_) => Seq()
+ case Success(subdomainLength) => Seq(subdomainLength)
}
}))
@@ -211,11 +213,12 @@ object DNSSuspiciousConnectsModel {
.filter(domainStatsRecords(SubdomainEntropy).gt(0))
.select(SubdomainEntropy)
.rdd
- .flatMap({ case Row(subdomainEntropy: Double) => {
- Try{subdomainEntropy.toDouble} match {
+ .flatMap({ case Row(subdomainEntropy: Double) =>
+ Try {
+ subdomainEntropy.toDouble
+ } match {
case Failure(_) => Seq()
case Success(subdomainEntropy) => Seq(subdomainEntropy)
- }
}
}))
@@ -224,23 +227,24 @@ object DNSSuspiciousConnectsModel {
.filter(domainStatsRecords(NumPeriods).gt(0))
.select(NumPeriods)
.rdd
- .flatMap({ case Row(numberPeriods: Int) => {
- Try {numberPeriods.toDouble} match {
+ .flatMap({ case Row(numberPeriods: Int) =>
+ Try {
+ numberPeriods.toDouble
+ } match {
case Failure(_) => Seq()
case Success(numberPeriods) => Seq(numberPeriods)
- }
}
}))
// simplify DNS log entries into "words"
val dnsWordCreator = new DNSWordCreation(frameLengthCuts,
- timeCuts,
- subdomainLengthCuts,
- entropyCuts,
- numberPeriodsCuts,
- topDomainsBC,
- userDomain)
+ timeCuts,
+ subdomainLengthCuts,
+ entropyCuts,
+ numberPeriodsCuts,
+ topDomainsBC,
+ userDomain)
val dataWithWord = totalRecords.withColumn(Word, dnsWordCreator.wordCreationUDF(modelColumns: _*))
@@ -267,7 +271,7 @@ object DNSSuspiciousConnectsModel {
// Since DNS is still broadcasting ip to topic mix, we need to convert data frame to Map[String, Array[Double]]
val ipToTopicMix = ipToTopicMixDF
.rdd
- .map({ case (ipToTopicMixRow: Row) => (ipToTopicMixRow.toSeq.toArray) })
+ .map({ case (ipToTopicMixRow: Row) => ipToTopicMixRow.toSeq.toArray })
.map({
case (ipToTopicMixSeq) => (ipToTopicMixSeq(0).asInstanceOf[String], ipToTopicMixSeq(1).asInstanceOf[Seq[Double]]
.toArray)
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/24b3a37b/spot-ml/src/main/scala/org/apache/spot/netflow/FlowSuspiciousConnectsAnalysis.scala
----------------------------------------------------------------------
diff --git a/spot-ml/src/main/scala/org/apache/spot/netflow/FlowSuspiciousConnectsAnalysis.scala b/spot-ml/src/main/scala/org/apache/spot/netflow/FlowSuspiciousConnectsAnalysis.scala
index aad3e66..bb0a951 100644
--- a/spot-ml/src/main/scala/org/apache/spot/netflow/FlowSuspiciousConnectsAnalysis.scala
+++ b/spot-ml/src/main/scala/org/apache/spot/netflow/FlowSuspiciousConnectsAnalysis.scala
@@ -33,7 +33,7 @@ object FlowSuspiciousConnectsAnalysis {
val orderedFlowRecords = filteredFlowRecords.orderBy(Score)
val mostSuspiciousFlowRecords =
- if(config.maxResults > 0 ) orderedFlowRecords.limit(config.maxResults) else orderedFlowRecords
+ if (config.maxResults > 0) orderedFlowRecords.limit(config.maxResults) else orderedFlowRecords
val outputFlowRecords = mostSuspiciousFlowRecords.select(OutSchema: _*)
@@ -60,9 +60,9 @@ object FlowSuspiciousConnectsAnalysis {
*/
def detectFlowAnomalies(data: DataFrame,
config: SuspiciousConnectsConfig,
- sparkContext: SparkContext,
- sqlContext: SQLContext,
- logger: Logger) : DataFrame = {
+ sparkContext: SparkContext,
+ sqlContext: SQLContext,
+ logger: Logger): DataFrame = {
logger.info("Fitting probabilistic model to data")
@@ -78,7 +78,7 @@ object FlowSuspiciousConnectsAnalysis {
* @param inputFlowRecords raw flow records
* @return
*/
- def filterAndSelectCleanFlowRecords(inputFlowRecords: DataFrame): DataFrame ={
+ def filterAndSelectCleanFlowRecords(inputFlowRecords: DataFrame): DataFrame = {
val cleanFlowRecordsFilter = inputFlowRecords(Hour).between(0, 23) &&
inputFlowRecords(Minute).between(0, 59) &&
@@ -104,9 +104,9 @@ object FlowSuspiciousConnectsAnalysis {
*/
def filterAndSelectInvalidFlowRecords(inputFlowRecords: DataFrame): DataFrame = {
- val invalidFlowRecordsFilter = inputFlowRecords(Hour).between(0,23) &&
- inputFlowRecords(Minute).between(0,59) &&
- inputFlowRecords(Second).between(0,59) &&
+ val invalidFlowRecordsFilter = inputFlowRecords(Hour).between(0, 23) &&
+ inputFlowRecords(Minute).between(0, 59) &&
+ inputFlowRecords(Second).between(0, 59) &&
inputFlowRecords(TimeReceived).isNull ||
inputFlowRecords(SourceIP).isNull ||
inputFlowRecords(DestinationIP).isNull ||
@@ -123,7 +123,7 @@ object FlowSuspiciousConnectsAnalysis {
/**
*
* @param scoredFlowRecords scored flow records.
- * @param threshold score tolerance.
+ * @param threshold score tolerance.
* @return
*/
def filterScoredFlowRecords(scoredFlowRecords: DataFrame, threshold: Double): DataFrame = {
@@ -150,7 +150,6 @@ object FlowSuspiciousConnectsAnalysis {
}
-
val InSchema = StructType(List(TimeReceivedField,
YearField,
MonthField,
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/24b3a37b/spot-ml/src/main/scala/org/apache/spot/netflow/FlowWordCreator.scala
----------------------------------------------------------------------
diff --git a/spot-ml/src/main/scala/org/apache/spot/netflow/FlowWordCreator.scala b/spot-ml/src/main/scala/org/apache/spot/netflow/FlowWordCreator.scala
index bd94df5..b825671 100644
--- a/spot-ml/src/main/scala/org/apache/spot/netflow/FlowWordCreator.scala
+++ b/spot-ml/src/main/scala/org/apache/spot/netflow/FlowWordCreator.scala
@@ -4,7 +4,7 @@ import org.apache.spark.sql.functions._
import org.apache.spot.utilities.Quantiles
import org.apache.spot.utilities.data.validation.InvalidDataHandler
-import scala.util.{Failure, Success, Try}
+import scala.util.{Success, Try}
/**
@@ -30,6 +30,7 @@ class FlowWordCreator(timeCuts: Array[Double],
/**
* Spark SQL UDF for calculating the word summarizing a netflow transaction at the source IP
+ *
* @return String "word" summarizing a netflow connection.
*/
def srcWordUDF = udf((hour: Int,
@@ -46,6 +47,7 @@ class FlowWordCreator(timeCuts: Array[Double],
/**
* Spark SQL UDF for calculating the word summarizing a netflow transaction at the destination IP
+ *
* @return String "word" summarizing a netflow connection.
*/
def dstWordUDF = udf((hour: Int,
@@ -96,7 +98,7 @@ class FlowWordCreator(timeCuts: Array[Double],
} else if (srcPort == 0 && dstPort > 0) {
- val baseWord = Array(dstPort.toString(), timeBin, ibytBin, ipktBin).mkString("_")
+ val baseWord = Array(dstPort.toString, timeBin, ibytBin, ipktBin).mkString("_")
FlowWords(srcWord = baseWord, dstWord = "-1_" + baseWord)
} else if (srcPort <= 1024 && dstPort <= 1024) {
@@ -106,12 +108,12 @@ class FlowWordCreator(timeCuts: Array[Double],
} else if (srcPort <= 1024 && dstPort > 1024) {
- val baseWord = Array(srcPort.toString(), timeBin, ibytBin, ipktBin).mkString("_")
+ val baseWord = Array(srcPort.toString, timeBin, ibytBin, ipktBin).mkString("_")
FlowWords(srcWord = "-1_" + baseWord, dstWord = baseWord)
} else if (srcPort > 1024 && dstPort <= 1024) {
- val baseWord = Array(dstPort.toString(), timeBin, ibytBin, ipktBin).mkString("_")
+ val baseWord = Array(dstPort.toString, timeBin, ibytBin, ipktBin).mkString("_")
FlowWords(srcWord = baseWord, dstWord = "-1_" + baseWord)
} else {
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/24b3a37b/spot-ml/src/main/scala/org/apache/spot/netflow/model/FlowScoreFunction.scala
----------------------------------------------------------------------
diff --git a/spot-ml/src/main/scala/org/apache/spot/netflow/model/FlowScoreFunction.scala b/spot-ml/src/main/scala/org/apache/spot/netflow/model/FlowScoreFunction.scala
index b82cd3f..b37f2f1 100644
--- a/spot-ml/src/main/scala/org/apache/spot/netflow/model/FlowScoreFunction.scala
+++ b/spot-ml/src/main/scala/org/apache/spot/netflow/model/FlowScoreFunction.scala
@@ -1,29 +1,27 @@
package org.apache.spot.netflow.model
import org.apache.spark.broadcast.Broadcast
-import org.apache.spot.SuspiciousConnectsScoreFunction
import org.apache.spot.netflow.{FlowWordCreator, FlowWords}
import org.apache.spot.utilities.data.validation.InvalidDataHandler
/**
* Estimate the probabilities of network events using a [[FlowSuspiciousConnectsModel]]
- * @param timeCuts
- * @param ibytCuts
- * @param ipktCuts
- * @param topicCount
- * @param wordToPerTopicProbBC
+ *
+ * @param timeCuts Quantile cut-offs for binning time-of-day values when forming words from netflow records.
+ * @param ibytCuts Quantile cut-offs for binning ibyt values when forming words from netflow records.
+ * @param ipktCuts Quantile cut-offs for binning ipkt values when forming words from netflow records.
+ * @param topicCount Number of topics used in the topic modelling analysis.
+ * @param wordToPerTopicProbBC Broadcast map assigning to each word it's per-topic probabilities.
+ * Ie. Prob [word | t ] for t = 0 to topicCount -1
*/
-
class FlowScoreFunction(timeCuts: Array[Double],
ibytCuts: Array[Double],
ipktCuts: Array[Double],
- topicCount: Int,
- wordToPerTopicProbBC: Broadcast[Map[String, Array[Double]]]) extends Serializable {
-
-
+ topicCount: Int,
+ wordToPerTopicProbBC: Broadcast[Map[String, Array[Double]]]) extends Serializable {
val flowWordCreator = new FlowWordCreator(timeCuts, ibytCuts, ipktCuts)
@@ -31,17 +29,18 @@ class FlowScoreFunction(timeCuts: Array[Double],
/**
* Estimate the probability of a netflow connection as distributed from the source IP and from the destination IP
* and assign it the least of these two values.
- * @param hour
- * @param minute
- * @param second
- * @param srcIP
- * @param dstIP
- * @param srcPort
- * @param dstPort
- * @param ipkt
- * @param ibyt
- * @param srcTopicMix
- * @param dstTopicMix
+ *
+ * @param hour Hour of flow record.
+ * @param minute Minute of flow record.
+ * @param second Second of flow record.
+ * @param srcIP Source IP of flow record.
+ * @param dstIP Destination IP of flow record.
+ * @param srcPort Source port of flow record.
+ * @param dstPort Destination port of flow record.
+ * @param ipkt ipkt entry of flow record
+ * @param ibyt ibyt entry of flow record
+ * @param srcTopicMix topic mix assigned of source IP
+ * @param dstTopicMix topic mix assigned of destination IP
* @return Minium of probability of this word from the source IP and probability of this word from the dest IP.
*/
def score(hour: Int,
@@ -60,28 +59,30 @@ class FlowScoreFunction(timeCuts: Array[Double],
val FlowWords(srcWord, dstWord) = flowWordCreator.flowWords(hour: Int, minute: Int, second: Int,
srcPort: Int, dstPort: Int, ipkt: Long, ibyt: Long)
- val zeroProb = Array.fill(topicCount) { 0.0 }
+ val zeroProb = Array.fill(topicCount) {
+ 0.0
+ }
/** WordError indicates there was a problem creating a word and should not be used for scoring.
-
- A null value for srcTopicMix or dstTopicMix indicated the ip (source or dest respectively)
+ * *
+ * A null value for srcTopicMix or dstTopicMix indicated the ip (source or dest respectively)
* were not seen in training.
*/
- if(srcWord == InvalidDataHandler.WordError || dstWord == InvalidDataHandler.WordError){
+ if (srcWord == InvalidDataHandler.WordError || dstWord == InvalidDataHandler.WordError) {
InvalidDataHandler.ScoreError
} else if (srcTopicMix == null || dstTopicMix == null) {
- 0.0
+ 0.0
} else {
- val scoreOfConnectionFromSrcIP = srcTopicMix.zip(wordToPerTopicProbBC.value.getOrElse(srcWord, zeroProb))
- .map({ case (pWordGivenTopic, pTopicGivenDoc) => pWordGivenTopic * pTopicGivenDoc })
- .sum
+ val scoreOfConnectionFromSrcIP = srcTopicMix.zip(wordToPerTopicProbBC.value.getOrElse(srcWord, zeroProb))
+ .map({ case (pWordGivenTopic, pTopicGivenDoc) => pWordGivenTopic * pTopicGivenDoc })
+ .sum
- val scoreOfConnectionsFromDstIP = dstTopicMix.zip(wordToPerTopicProbBC.value.getOrElse(dstWord, zeroProb))
- .map({ case (pWordGivenTopic, pTopicGivenDoc) => pWordGivenTopic * pTopicGivenDoc })
- .sum
+ val scoreOfConnectionsFromDstIP = dstTopicMix.zip(wordToPerTopicProbBC.value.getOrElse(dstWord, zeroProb))
+ .map({ case (pWordGivenTopic, pTopicGivenDoc) => pWordGivenTopic * pTopicGivenDoc })
+ .sum
- Math.min(scoreOfConnectionFromSrcIP, scoreOfConnectionsFromDstIP)
+ Math.min(scoreOfConnectionFromSrcIP, scoreOfConnectionsFromDstIP)
}
}
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/24b3a37b/spot-ml/src/main/scala/org/apache/spot/netflow/model/FlowSuspiciousConnectsModel.scala
----------------------------------------------------------------------
diff --git a/spot-ml/src/main/scala/org/apache/spot/netflow/model/FlowSuspiciousConnectsModel.scala b/spot-ml/src/main/scala/org/apache/spot/netflow/model/FlowSuspiciousConnectsModel.scala
index 4f07fba..2ab682e 100644
--- a/spot-ml/src/main/scala/org/apache/spot/netflow/model/FlowSuspiciousConnectsModel.scala
+++ b/spot-ml/src/main/scala/org/apache/spot/netflow/model/FlowSuspiciousConnectsModel.scala
@@ -22,23 +22,23 @@ import scala.util.{Failure, Success, Try}
*
* The model uses a topic-modelling approach that:
* 1. Simplifies netflow records into words, one word at the source IP and another (possibly different) at the
- * destination IP.
+ * destination IP.
* 2. The netflow words about each IP are treated as collections of thes words.
* 3. A topic modelling approach is used to infer a collection of "topics" that represent common profiles
- * of network traffic. These "topics" are probability distributions on words.
+ * of network traffic. These "topics" are probability distributions on words.
* 4. Each IP has a mix of topics corresponding to its behavior.
* 5. The probability of a word appearing in the traffic about an IP is estimated by simplifying its netflow record
- * into a word, and then combining the word probabilities per topic using the topic mix of the particular IP.
+ * into a word, and then combining the word probabilities per topic using the topic mix of the particular IP.
*
* Create these models using the factory in the companion object.
*
- * @param topicCount Number of topics (profiles of common traffic patterns) used in the topic modelling routine.
- * @param ipToTopicMix DataFrame assigning a distribution on topics to each document or IP.
+ * @param topicCount Number of topics (profiles of common traffic patterns) used in the topic modelling routine.
+ * @param ipToTopicMix DataFrame assigning a distribution on topics to each document or IP.
* @param wordToPerTopicProb Map assigning to each word it's per-topic probabilities.
* Ie. Prob [word | t ] for t = 0 to topicCount -1
- * @param timeCuts Quantile cut-offs for binning time-of-day values when forming words from netflow records.
- * @param ibytCuts Quantile cut-offs for binning ibyt values when forming words from netflow records.
- * @param ipktCuts Quantile cut-offs for binning ipkt values when forming words from netflow records.
+ * @param timeCuts Quantile cut-offs for binning time-of-day values when forming words from netflow records.
+ * @param ibytCuts Quantile cut-offs for binning ibyt values when forming words from netflow records.
+ * @param ipktCuts Quantile cut-offs for binning ipkt values when forming words from netflow records.
*/
class FlowSuspiciousConnectsModel(topicCount: Int,
@@ -66,15 +66,15 @@ class FlowSuspiciousConnectsModel(topicCount: Int,
val recordsWithIPTopicMixes = dataWithSrcIpProb.join(ipToTopicMix,
dataWithSrcIpProb(DestinationIP) === ipToTopicMix(DocumentName), "left_outer")
- val schema = dataWithSrcIpProb.schema.fieldNames :+ TopicProbabilityMix
- recordsWithIPTopicMixes.selectExpr(schema: _*).withColumnRenamed(TopicProbabilityMix, DstIpTopicMix)
+ val schema = dataWithSrcIpProb.schema.fieldNames :+ TopicProbabilityMix
+ recordsWithIPTopicMixes.selectExpr(schema: _*).withColumnRenamed(TopicProbabilityMix, DstIpTopicMix)
}
- val scoreFunction = new FlowScoreFunction(timeCuts,
- ibytCuts,
- ipktCuts,
- topicCount,
- wordToPerTopicProbBC)
+ val scoreFunction = new FlowScoreFunction(timeCuts,
+ ibytCuts,
+ ipktCuts,
+ topicCount,
+ wordToPerTopicProbBC)
val scoringUDF = udf((hour: Int,
@@ -149,11 +149,12 @@ object FlowSuspiciousConnectsModel {
val timeCuts = Quantiles.computeDeciles(totalRecords
.select(Hour, Minute, Second)
.rdd
- .flatMap({ case Row(hours: Int, minutes: Int, seconds: Int) => {
- Try { (3600 * hours + 60 * minutes + seconds).toDouble } match{
- case Failure(_) => Seq()
- case Success(time) => Seq(time)
- }
+ .flatMap({ case Row(hours: Int, minutes: Int, seconds: Int) =>
+ Try {
+ (3600 * hours + 60 * minutes + seconds).toDouble
+ } match {
+ case Failure(_) => Seq()
+ case Success(time) => Seq(time)
}
}))
@@ -164,11 +165,12 @@ object FlowSuspiciousConnectsModel {
val ibytCuts = Quantiles.computeDeciles(totalRecords
.select(Ibyt)
.rdd
- .flatMap({ case Row(ibyt: Long) => {
- Try { ibyt.toDouble } match{
- case Failure(_) => Seq()
- case Success(ibyt) => Seq(ibyt)
- }
+ .flatMap({ case Row(ibyt: Long) =>
+ Try {
+ ibyt.toDouble
+ } match {
+ case Failure(_) => Seq()
+ case Success(ibyt) => Seq(ibyt)
}
}))
@@ -179,11 +181,12 @@ object FlowSuspiciousConnectsModel {
val ipktCuts = Quantiles.computeQuintiles(totalRecords
.select(Ipkt)
.rdd
- .flatMap({ case Row(ipkt: Long) => {
- Try { ipkt.toDouble } match {
- case Failure(_) => Seq()
- case Success(ipkt) => Seq(ipkt)
- }
+ .flatMap({ case Row(ipkt: Long) =>
+ Try {
+ ipkt.toDouble
+ } match {
+ case Failure(_) => Seq()
+ case Success(ipkt) => Seq(ipkt)
}
}))
@@ -232,5 +235,4 @@ object FlowSuspiciousConnectsModel {
ibytCuts,
ipktCuts)
}
-
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/24b3a37b/spot-ml/src/main/scala/org/apache/spot/proxy/ProxySuspiciousConnectsModel.scala
----------------------------------------------------------------------
diff --git a/spot-ml/src/main/scala/org/apache/spot/proxy/ProxySuspiciousConnectsModel.scala b/spot-ml/src/main/scala/org/apache/spot/proxy/ProxySuspiciousConnectsModel.scala
index c38ed93..edfa341 100644
--- a/spot-ml/src/main/scala/org/apache/spot/proxy/ProxySuspiciousConnectsModel.scala
+++ b/spot-ml/src/main/scala/org/apache/spot/proxy/ProxySuspiciousConnectsModel.scala
@@ -7,11 +7,11 @@ import org.apache.spark.rdd.RDD
import org.apache.spark.sql.functions._
import org.apache.spark.sql.{DataFrame, Row, SQLContext}
import org.apache.spot.SuspiciousConnectsArgumentParser.SuspiciousConnectsConfig
-import org.apache.spot.proxy.ProxySchema._
-import org.apache.spot.utilities._
import org.apache.spot.SuspiciousConnectsScoreFunction
import org.apache.spot.lda.SpotLDAWrapper
import org.apache.spot.lda.SpotLDAWrapper.{SpotLDAInput, SpotLDAOutput}
+import org.apache.spot.proxy.ProxySchema._
+import org.apache.spot.utilities._
import org.apache.spot.utilities.data.validation.InvalidDataHandler
import scala.util.{Failure, Success, Try}
@@ -88,7 +88,7 @@ object ProxySuspiciousConnectsModel {
* @param sqlContext SQL context.
* @param logger Logge object.
* @param config SuspiciousConnetsArgumnetParser.Config object containg CLI arguments.
- * @param inputRecords Dataframe for training data, with columns Host, Time, ReqMethod, FullURI, ResponseContentType,
+ * @param inputRecords Dataframe for training data, with columns Host, Time, ReqMethod, FullURI, ResponseContentType,
* UserAgent, RespCode (as defined in ProxySchema object).
* @return ProxySuspiciousConnectsModel
*/
@@ -108,23 +108,27 @@ object ProxySuspiciousConnectsModel {
Quantiles.computeDeciles(selectedRecords
.select(Time)
.rdd
- .flatMap({ case Row(t: String) => {
- Try {TimeUtilities.getTimeAsDouble(t)} match {
- case Failure(_) => Seq()
- case Success(time) => Seq(time)
- }
+ .flatMap({ case Row(t: String) =>
+ Try {
+ TimeUtilities.getTimeAsDouble(t)
+ } match {
+ case Failure(_) => Seq()
+ case Success(time) => Seq(time)
+
}
}))
val entropyCuts = Quantiles.computeQuintiles(selectedRecords
.select(FullURI)
.rdd
- .flatMap({ case Row(uri: String) => {
- Try {Entropy.stringEntropy(uri)} match {
- case Failure(_) => Seq()
- case Success(entropy) => Seq(entropy)
- }
+ .flatMap({ case Row(uri: String) =>
+ Try {
+ Entropy.stringEntropy(uri)
+ } match {
+ case Failure(_) => Seq()
+ case Success(entropy) => Seq(entropy)
}
+
}))
val agentToCount: Map[String, Long] =
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/24b3a37b/spot-ml/src/main/scala/org/apache/spot/utilities/data/InputOutputDataHandler.scala
----------------------------------------------------------------------
diff --git a/spot-ml/src/main/scala/org/apache/spot/utilities/data/InputOutputDataHandler.scala b/spot-ml/src/main/scala/org/apache/spot/utilities/data/InputOutputDataHandler.scala
index e7934f4..fbbd085 100644
--- a/spot-ml/src/main/scala/org/apache/spot/utilities/data/InputOutputDataHandler.scala
+++ b/spot-ml/src/main/scala/org/apache/spot/utilities/data/InputOutputDataHandler.scala
@@ -1,7 +1,7 @@
package org.apache.spot.utilities.data
-import org.apache.log4j.Logger
import org.apache.hadoop.fs.{LocatedFileStatus, Path, RemoteIterator, FileUtil => fileUtil}
+import org.apache.log4j.Logger
import org.apache.spark.SparkContext
import org.apache.spark.sql.{DataFrame, SQLContext}
@@ -15,27 +15,25 @@ object InputOutputDataHandler {
/**
*
* @param sqlContext Application SqlContext.
- * @param inputPath HDFS input folder for every execution; flow, dns or proxy.
- * @param logger Application logger.
+ * @param inputPath HDFS input folder for every execution; flow, dns or proxy.
+ * @param logger Application logger.
* @return raw data frame.
*/
- def getInputDataFrame(sqlContext: SQLContext, inputPath: String, logger: Logger): Option[DataFrame] ={
+ def getInputDataFrame(sqlContext: SQLContext, inputPath: String, logger: Logger): Option[DataFrame] = {
try {
logger.info("Loading data from: " + inputPath)
Some(sqlContext.read.parquet(inputPath))
} catch {
- case _ : Throwable => {
- None
- }
+ case _: Throwable => None
}
}
/**
*
- * @param sparkContext Application SparkContext.
+ * @param sparkContext Application SparkContext.
* @param hdfsScoredConnect HDFS output folder. The location where results were saved; flow, dns or proxy.
- * @param analysis Data type to analyze.
- * @param logger Application Logger.
+ * @param analysis Data type to analyze.
+ * @param logger Application Logger.
*/
def mergeResultsFiles(sparkContext: SparkContext, hdfsScoredConnect: String, analysis: String, logger: Logger) {
val hadoopConfiguration = sparkContext.hadoopConfiguration
@@ -43,21 +41,21 @@ object InputOutputDataHandler {
val exists = fileSystem.exists(new org.apache.hadoop.fs.Path(hdfsScoredConnect))
- if(exists){
+ if (exists) {
val srcDir = new Path(hdfsScoredConnect)
- val dstFile = new Path(hdfsScoredConnect+"/"+analysis+"_results.csv")
- fileUtil.copyMerge(fileSystem,srcDir, fileSystem, dstFile, false, hadoopConfiguration, "")
+ val dstFile = new Path(hdfsScoredConnect + "/" + analysis + "_results.csv")
+ fileUtil.copyMerge(fileSystem, srcDir, fileSystem, dstFile, false, hadoopConfiguration, "")
val files: RemoteIterator[LocatedFileStatus] = fileSystem.listFiles(srcDir, false)
- while (files.hasNext){
- val filePath = files.next().getPath()
- if(filePath.toString.contains("part-")){
+ while (files.hasNext) {
+ val filePath = files.next.getPath
+ if (filePath.toString.contains("part-")) {
fileSystem.delete(filePath, false)
}
}
}
else logger.info(s"Couldn't find results in $hdfsScoredConnect." +
- s"Please check previous logs to see if there were errors.")
- }
+ s"Please check previous logs to see if there were errors.")
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/24b3a37b/spot-ml/src/test/scala/org/apache/spot/dns/DNSSuspiciousConnectsAnalysisTest.scala
----------------------------------------------------------------------
diff --git a/spot-ml/src/test/scala/org/apache/spot/dns/DNSSuspiciousConnectsAnalysisTest.scala b/spot-ml/src/test/scala/org/apache/spot/dns/DNSSuspiciousConnectsAnalysisTest.scala
index b106e6e..6e03585 100644
--- a/spot-ml/src/test/scala/org/apache/spot/dns/DNSSuspiciousConnectsAnalysisTest.scala
+++ b/spot-ml/src/test/scala/org/apache/spot/dns/DNSSuspiciousConnectsAnalysisTest.scala
@@ -2,32 +2,31 @@ package org.apache.spot.dns
import org.apache.log4j.{Level, LogManager}
-import org.apache.spot.SuspiciousConnectsArgumentParser.SuspiciousConnectsConfig
-import org.apache.spark.sql.SQLContext
+import org.apache.spark.sql.{Row, SQLContext}
import org.apache.spark.sql.types.StructType
-import org.apache.spark.sql.Row
+import org.apache.spot.SuspiciousConnectsArgumentParser.SuspiciousConnectsConfig
import org.apache.spot.dns.DNSSchema._
import org.apache.spot.testutils.TestingSparkContextFlatSpec
import org.scalatest.Matchers
-case class DNSInput(frame_time:String, unix_tstamp:Long, frame_len:Int, ip_dst: String, dns_qry_name:String, dns_qry_class:String, dns_qry_type: Int, dns_qry_rcode: Int)
+case class DNSInput(frame_time: String, unix_tstamp: Long, frame_len: Int, ip_dst: String, dns_qry_name: String, dns_qry_class: String, dns_qry_type: Int, dns_qry_rcode: Int)
-class DNSSuspiciousConnectsAnalysisTest extends TestingSparkContextFlatSpec with Matchers {
+class DNSSuspiciousConnectsAnalysisTest extends TestingSparkContextFlatSpec with Matchers {
val testConfig = SuspiciousConnectsConfig(analysis = "dns",
- inputPath = "",
- feedbackFile = "",
- duplicationFactor = 1,
- topicCount = 20,
- hdfsScoredConnect = "",
- threshold = 1.0d,
- maxResults = 1000,
- outputDelimiter = "\t",
- ldaPRGSeed = None,
- ldaMaxiterations = 20,
- ldaAlpha = 1.02,
- ldaBeta = 1.001)
+ inputPath = "",
+ feedbackFile = "",
+ duplicationFactor = 1,
+ topicCount = 20,
+ hdfsScoredConnect = "",
+ threshold = 1.0d,
+ maxResults = 1000,
+ outputDelimiter = "\t",
+ ldaPRGSeed = None,
+ ldaMaxiterations = 20,
+ ldaAlpha = 1.02,
+ ldaBeta = 1.001)
"dns supicious connects analysis" should "estimate correct probabilities in toy data with framelength anomaly" in {
@@ -35,12 +34,12 @@ class DNSSuspiciousConnectsAnalysisTest extends TestingSparkContextFlatSpec wit
val logger = LogManager.getLogger("SuspiciousConnectsAnalysis")
logger.setLevel(Level.WARN)
- val anomalousRecord = DNSInput("May 20 2016 02:10:25.970987000 PDT", 1463735425L, 1, "172.16.9.132", "turner.com.122.2o7.net", "0x00000001", 1, 0)
- val typicalRecord = DNSInput("May 20 2016 02:10:25.970987000 PDT", 1463735425L, 168, "172.16.9.132", "turner.com.122.2o7.net", "0x00000001", 1, 0)
+ val anomalousRecord = DNSInput("May 20 2016 02:10:25.970987000 PDT", 1463735425L, 1, "172.16.9.132", "turner.com.122.2o7.net", "0x00000001", 1, 0)
+ val typicalRecord = DNSInput("May 20 2016 02:10:25.970987000 PDT", 1463735425L, 168, "172.16.9.132", "turner.com.122.2o7.net", "0x00000001", 1, 0)
val data = sqlContext.createDataFrame(Seq(anomalousRecord, typicalRecord, typicalRecord, typicalRecord, typicalRecord))
- val scoredData = DNSSuspiciousConnectsAnalysis.detectDNSAnomalies(data, testConfig,
+ val scoredData = DNSSuspiciousConnectsAnalysis.scoreDNSRecords(data, testConfig,
sparkContext,
sqlContext,
logger)
@@ -49,12 +48,12 @@ class DNSSuspiciousConnectsAnalysisTest extends TestingSparkContextFlatSpec wit
val anomalyScore = scoredData.filter(scoredData(FrameLength) === 1).first().getAs[Double](Score)
val typicalScores = scoredData.filter(scoredData(FrameLength) === 168).collect().map(_.getAs[Double](Score))
- Math.abs(anomalyScore - 0.2d) should be <= 0.01d
+ Math.abs(anomalyScore - 0.2d) should be <= 0.01d
typicalScores.length shouldBe 4
- Math.abs(typicalScores(0) - 0.8d) should be <= 0.01d
- Math.abs(typicalScores(1) - 0.8d) should be <= 0.01d
- Math.abs(typicalScores(2) - 0.8d) should be <= 0.01d
- Math.abs(typicalScores(3) - 0.8d) should be <= 0.01d
+ Math.abs(typicalScores(0) - 0.8d) should be <= 0.01d
+ Math.abs(typicalScores(1) - 0.8d) should be <= 0.01d
+ Math.abs(typicalScores(2) - 0.8d) should be <= 0.01d
+ Math.abs(typicalScores(3) - 0.8d) should be <= 0.01d
}
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/24b3a37b/spot-ml/src/test/scala/org/apache/spot/dns/DNSWordCreationTest.scala
----------------------------------------------------------------------
diff --git a/spot-ml/src/test/scala/org/apache/spot/dns/DNSWordCreationTest.scala b/spot-ml/src/test/scala/org/apache/spot/dns/DNSWordCreationTest.scala
index fa233a2..1422bb6 100644
--- a/spot-ml/src/test/scala/org/apache/spot/dns/DNSWordCreationTest.scala
+++ b/spot-ml/src/test/scala/org/apache/spot/dns/DNSWordCreationTest.scala
@@ -13,5 +13,4 @@ class DNSWordCreationTest extends TestingSparkContextFlatSpec with Matchers {
result shouldBe 2.807354922057604
}
-
}
http://git-wip-us.apache.org/repos/asf/incubator-spot/blob/24b3a37b/spot-ml/src/test/scala/org/apache/spot/netflow/FlowSuspiciousConnectsAnalysisTest.scala
----------------------------------------------------------------------
diff --git a/spot-ml/src/test/scala/org/apache/spot/netflow/FlowSuspiciousConnectsAnalysisTest.scala b/spot-ml/src/test/scala/org/apache/spot/netflow/FlowSuspiciousConnectsAnalysisTest.scala
index d59635d..8f9f89f 100644
--- a/spot-ml/src/test/scala/org/apache/spot/netflow/FlowSuspiciousConnectsAnalysisTest.scala
+++ b/spot-ml/src/test/scala/org/apache/spot/netflow/FlowSuspiciousConnectsAnalysisTest.scala
@@ -1,8 +1,8 @@
package org.apache.spot.netflow
-import org.apache.log4j.{Level, LogManager, Logger}
-import org.apache.spark.SparkContext
-import org.apache.spark.sql.{DataFrame, SQLContext}
+import org.apache.log4j.{Level, LogManager}
+import org.apache.spark.sql.types.StructType
+import org.apache.spark.sql.{DataFrame, Row, SQLContext}
import org.apache.spot.SuspiciousConnectsArgumentParser.SuspiciousConnectsConfig
import org.apache.spot.netflow.FlowSchema._
import org.apache.spot.testutils.TestingSparkContextFlatSpec
@@ -56,7 +56,7 @@ class FlowSuspiciousConnectsAnalysisTest extends TestingSparkContextFlatSpec wit
typicalRecord, typicalRecord, typicalRecord, typicalRecord))
- val scoredData : DataFrame = FlowSuspiciousConnectsAnalysis.detectFlowAnomalies(data,
+ val scoredData: DataFrame = FlowSuspiciousConnectsAnalysis.detectFlowAnomalies(data,
testConfig,
sparkContext,
sqlContext,
@@ -80,7 +80,114 @@ class FlowSuspiciousConnectsAnalysisTest extends TestingSparkContextFlatSpec wit
Math.abs(typicalScores(8) - 0.9d) should be < 0.01
+ }
+ "filterAndSelectCleanFlowRecords" should "return data set without garbage" in {
+
+ val cleanedFlowRecords = FlowSuspiciousConnectsAnalysis
+ .filterAndSelectCleanFlowRecords(testFlowRecords.inputFlowRecordsDF)
+
+ cleanedFlowRecords.count should be(5)
+ cleanedFlowRecords.schema.size should be(17)
+ }
+
+ "filterAndSelectInvalidFlowRecords" should "return invalid records" in {
+
+ val invalidFlowRecords = FlowSuspiciousConnectsAnalysis
+ .filterAndSelectInvalidFlowRecords(testFlowRecords.inputFlowRecordsDF)
+
+ invalidFlowRecords.count should be(7)
+ invalidFlowRecords.schema.size should be(17)
+ }
+ "filterScoredFlowRecords" should "return records with score less or equal to threshold" in {
+
+ val threshold = 10e-5
+
+ val scoredFlowRecords = FlowSuspiciousConnectsAnalysis
+ .filterScoredFlowRecords(testFlowRecords.scoredFlowRecordsDF, threshold)
+
+ scoredFlowRecords.count should be(2)
+ }
+
+ "filterAndSelectCorruptFlowRecords" should "return records where Score is equal to -1" in {
+
+ val corruptFlowRecords = FlowSuspiciousConnectsAnalysis
+ .filterAndSelectCorruptFlowRecords(testFlowRecords.scoredFlowRecordsDF)
+
+ corruptFlowRecords.count should be(1)
+ corruptFlowRecords.schema.size should be(18)
}
+ def testFlowRecords = new {
+ val sqlContext = new SQLContext(sparkContext)
+
+ val inputFlowRecordsRDD = sparkContext.parallelize(wrapRefArray(Array(
+ Seq("2016-05-05 13:54:58", 2016, 5, 5, 24, 54, 58, 0.972, "172.16.0.129", "10.0.2.202", 1024, 80, "TCP", 39l, 12522l, 0, 0),
+ Seq("2016-05-05 13:54:58", 2016, 5, 5, 13, 54, 60, 0.972, "172.16.0.129", "10.0.2.202", 1024, 80, "TCP", 39l, 12522l, 0, 0),
+ Seq("2016-05-05 13:54:58", 2016, 5, 5, 13, 60, 58, 0.972, "172.16.0.129", "10.0.2.202", 1024, 80, "TCP", 39l, 12522l, 0, 0),
+ Seq(null, 2016, 5, 5, 13, 54, 58, 0.972, "172.16.0.129", "10.0.2.202", 1024, 80, "TCP", 39l, 12522l, 0, 0),
+ Seq("2016-05-05 13:54:58", 2016, 5, 5, 13, 54, 58, 0.972, null, "10.0.2.202", 1024, 80, "TCP", 39l, 12522l, 0, 0),
+ Seq("2016-05-05 13:54:58", 2016, 5, 5, 13, 54, 58, 0.972, "172.16.0.129", null, 1024, 80, "TCP", 39l, 12522l, 0, 0),
+ Seq("2016-05-05 13:54:58", 2016, 5, 5, 13, 54, 58, 0.972, "172.16.0.129", "10.0.2.202", null, 80, "TCP", 39l, 12522l, 0, 0),
+ Seq("2016-05-05 13:54:58", 2016, 5, 5, 13, 54, 58, 0.972, "172.16.0.129", "10.0.2.202", 1024, null, "TCP", 39l, 12522l, 0, 0),
+ Seq("2016-05-05 13:54:58", 2016, 5, 5, 13, 54, 58, 0.972, "172.16.0.129", "10.0.2.202", 1024, 80, "TCP", null, 12522l, 0, 0),
+ Seq("2016-05-05 13:54:58", 2016, 5, 5, 13, 54, 58, 0.972, "172.16.0.129", "10.0.2.202", 1024, 80, "TCP", 39l, null, 0, 0),
+ Seq("2016-05-05 13:54:58", 2016, 5, 5, 13, 54, 58, 0.972, "172.16.0.129", "10.0.2.202", 1024, 80, "TCP", 39l, 12522l, null, 0),
+ Seq("2016-05-05 13:54:58", 2016, 5, 5, 13, 54, 58, 0.972, "172.16.0.129", "10.0.2.202", 1024, 80, "TCP", 39l, 12522l, 0, null),
+ Seq("2016-05-05 13:54:58", 2016, 5, 5, 13, 54, 58, 0.972, "172.16.0.129", "10.0.2.202", 1024, 80, "TCP", 39l, 12522l, 0, 0),
+ Seq("2016-05-05 13:54:58", 2016, 5, 5, 13, 54, 58, 0.972, "172.16.0.129", "10.0.2.202", 1024, 80, "TCP", 39l, 12522l, 0, 0),
+ Seq("2016-05-05 13:54:58", 2016, 5, 5, 13, 54, 58, 0.972, "172.16.0.129", "10.0.2.202", 1024, 80, "TCP", 39l, 12522l, 0, 0))
+ .map(row => Row.fromSeq(row))))
+
+ val inputFlowRecordsSchema = StructType(
+ Array(TimeReceivedField,
+ YearField,
+ MonthField,
+ DayField,
+ HourField,
+ MinuteField,
+ SecondField,
+ DurationField,
+ SourceIPField,
+ DestinationIPField,
+ SourcePortField,
+ DestinationPortField,
+ ProtocolField,
+ IpktField,
+ IbytField,
+ OpktField,
+ ObytField))
+
+ val inputFlowRecordsDF = sqlContext.createDataFrame(inputFlowRecordsRDD, inputFlowRecordsSchema)
+
+ val scoredFlowRecordsRDD = sparkContext.parallelize(wrapRefArray(Array(
+ Seq("2016-05-05 13:54:58", 2016, 5, 5, 13, 54, 58, 0.972, "172.16.0.129", "10.0.2.202", 1024, 80, "TCP", 39l, 12522l, 0, 0, -1d),
+ Seq("2016-05-05 13:54:58", 2016, 5, 5, 13, 54, 58, 0.972, "172.16.0.129", "10.0.2.202", 1024, 80, "TCP", 39l, 12522l, 0, 0, 1d),
+ Seq("2016-05-05 13:54:58", 2016, 5, 5, 13, 54, 58, 0.972, "172.16.0.129", "10.0.2.202", 1024, 80, "TCP", 39l, 12522l, 0, 0, 0.0000005),
+ Seq("2016-05-05 13:54:58", 2016, 5, 5, 13, 54, 58, 0.972, "172.16.0.129", "10.0.2.202", 1024, 80, "TCP", 39l, 12522l, 0, 0, 0.05),
+ Seq("2016-05-05 13:54:58", 2016, 5, 5, 13, 54, 58, 0.972, "172.16.0.129", "10.0.2.202", 1024, 80, "TCP", 39l, 12522l, 0, 0, 0.0001))
+ .map(row => Row.fromSeq(row))))
+
+ val scoredFlowRecordsSchema = StructType(
+ Array(TimeReceivedField,
+ YearField,
+ MonthField,
+ DayField,
+ HourField,
+ MinuteField,
+ SecondField,
+ DurationField,
+ SourceIPField,
+ DestinationIPField,
+ SourcePortField,
+ DestinationPortField,
+ ProtocolField,
+ IpktField,
+ IbytField,
+ OpktField,
+ ObytField,
+ ScoreField))
+
+ val scoredFlowRecordsDF = sqlContext.createDataFrame(scoredFlowRecordsRDD, scoredFlowRecordsSchema)
+ }
}