零、前言
机器学习和大数据早已经火的一塌糊涂，将它们应用到安全防护也是近几年的热点话题。本文旨在介绍几种机器学习算法在安全上的应用，尽量用通俗的语言解释机器学习是怎么回事，不说数学原理，只将具体应用。代码实现基于Spark，因为Spark是大数据计算引擎，有自带机器学习包（MLlib），可以满足大数据的计算要求。
一、环境搭建
0x00：Spark安装
本文的实验环境是windows10，Spark的安装去官网下载即可，
http://spark.apache.org/downloads.html
解压到任意目录，然后将安装目录添加到环境变量，Spark主要由Java、Scale和Python实现，本文使用Python版本即PySpark。想要在Python中调用PySpark，还需将spark目录中的PySpark文件夹spark-2.1.0-bin-hadoop2.7\python\PySpark，复制到Python的安装目录Python27\Lib\site-packages中。如果安装成功你可以在cmd启动PySpark，并看到如下界面：
在浏览器打开http://127.0.0.1:4040/jobs/，你可以看到Spark的Web-UI界面：
Linux下的安装或集群配置可以参考这两篇文章：
http://www.tuicool.com/articles/QBRJnen
http://blog.csdn.net/oopsoom/article/details/24257981
0x01：代码示例
现在就来测试一个Wordcount代码示例，
from pyspark import SparkContext
if __name__ == "__main__":
sc = SparkContext(appName="PythonWordCount")
lines = sc.textFile('words.txt')
counts = lines.flatMap(lambda x: x.split(' ')) \
.map(lambda x: (x, 1)) \
.reduceByKey(lambda x,y:x+y)
output = counts.collect()
for (word, count) in output:
print "%s: %i" % (word, count)
sc.stop()
运行结果：
bad: 2 spark: 3 mlib: 2 good: 2 hadoop: 1 cool: 2
想要理解这几行代码需要一点点Spark编程基础，这里我就不废话了，可能一时半会也说不清楚，建议去看《Spark快速大数据分析》，入门超棒的一本书。
二、机器学习
首先先来解释一下什么是机器学习，机器学习的宗旨是让机器学会人类识别事物的方法，我们希望机器能够学会人类的思维过程。这其中比较典型的例子就是图片识别，比如下面两张图片，我们一眼就能认出这是一只猫和一只二哈
那么问题来了，为什么人能在一张包含各种颜色的图片中识别出一只动物呢？可能你会说：因为我见过这些动物。没错，这就是经验，经验告诉我们图片中的一些特征组合表示里面有一只二哈，那么我么就认为这是一个包含二哈的图片。可是人类的视觉系统经过了数万年的进化才有了现在的能力，怎么将它传授给机器（代码）呢？于是牛逼的人类开始尝试这件事情，这就是机器学习。
无论是人类还是机器，识别的本质是分类，我们将某个物体成功地归为一类即是成功地识别。识别过程大体可以分为四个步骤：学习、提取特征、识别、分类，学习的本质也是对特征的学习。人类的学习很多时候是无意识的，比如当你第一次见到二哈，有人告诉你了它的品种，当你再见到一只狗的时候你大概就能区分一下这是不是二哈，次数多了变成了你的经验，这就是人类的不断的学习过程。但是机器没有意识，我们必须给机器灌输一些“意识”，告诉他该干什么，该怎么学习，这就是训练。
机器学习主要分为三种：监督学习、半监督学习和无监督学习。那么什么是监督呢？举个栗子，你从未见过猫和狗，然后给你一组猫和狗的图片，图片的背面写着对应的动物的名称（猫或狗）；当你看过数遍以后再给你一组图片，区分一下哪些是猫哪些是狗，这就是监督学习后的分类测试。但是如果你没有第一组带着标签的（Label）图片，你能否直接去识别第二组图片呢？我想也是可以的，你会根据它们的毛发、眼睛、鼻子等特征大体的分成两种不同的图片（虽然你不知道它们叫什么），这个时候我们一般称之为聚类。简单的讲，监督学习就是先用带标签的训练集进行训练后再对测试集分类，无监督学习是直接对测试集分类（聚类）。半监督学习则是两者的结合，通常不带标签的数据远大于带标签的数据。
最后说一下特征向量，向量这个词最早可能是在数学里见到的，给人的感觉可能是深奥、高大上。在机器学习里我们可以把它看成是一个数组，维度既是数组的长度。放到算法里跑的一定是数值型的数据，比如文本分类，用以分类的向量一定是词频或单词数量等这类数据，而不是字符串。比如一行文本通过N-gram分词后建立的TF-IDF向量为：
三、URL分类
本节主要参考freebuf的一篇文章（见参考文献），以URL分类为例，尝试运行一个真正的机器学习算法。本示例使用监督学习算法–逻辑回归、支持向量机和朴素贝叶斯，通过学习两组已打标的训练集去预测一组已打标的测试集。测试集打标是为了将标签与预测结果比较，以计算预测的准确率。
机器学习算法的实现首先是要构造好的特征向量，在类似的分类算法中比较好的是首先通过N-Gram将文本数据向量化，比如对于下面的例子,如果N取3，步长为1，则：
[htt,ttp,tp:,p:/,://,//w,/ww,www,ww.,w.a,.ab,abc,bc.,c.c,.co,com]
然后去计算TF-IDF，词频—逆文档频率（简称TF-IDF）是一种用来从文本文档（例如URL）中生成特征向量的简单方法。它为文档中的每个词计算两个统计值：一个是词频（TF），也就是每个词在文档中出现的次数，另一个是逆文档频率（IDF），用来衡量一个词在整个文档语料库中出现的（逆）频繁程度。这些值的积，也就是TF×IDF，展示了一个词与特定文档的相关程度（比如这个词在某文档中很常见，但在整个语料库中却很少见）。如上图所示。
N-Gram的分词方法乍一看好像没什么道理，因为一般的特征向量的构造是提取的特征关键词。比如如果我们将向量定义为script、select、union、eval等词出现的词频，那就很好理解，因为那些词都是恶意关键词，在正则匹配中一般也会拦截。但其实N-Gram也是一样的效果，一个特定的关键词会被切分成特定的序列，比如select被分成[sel,ele,lec,ect],而其他的正常的词一般不会出现这样的序列。
算法代码如下：
#-*-coding:utf-8-*-
#Author:Xman21
from pyspark import SparkContext
from pyspark.mllib.regression import LabeledPoint
from pyspark.mllib.feature import HashingTF,IDF
from pyspark.mllib.classification import LogisticRegressionWithSGD,SVMWithSGD,NaiveBayes
import sys
reload(sys)
sys.setdefaultencoding('utf8')
def split2(line,distance,step):
'''分词函数，按照一定的步长和距离分词,生成N-gram序列'''
result = []
length = len(line)
for i in xrange(0,length,step):
result.append(line[i:i+distance])
return result
def check(test,model):
'''训练检测函数'''
countAll = 16447
lines = test.collect()
correctCount = 0
for line in lines:
temp = line[0:line.rfind(" ")]
testExample = tf.transform(split2(temp,3,1))
try:
if int(line.split(" ")[-1]) == model.predict(testExample):
correctCount = correctCount+1
except:
print line
C = float(correctCount)/countAll
print "正确率："+str(C)
def TFIDF(badData,goodData,distance,step):
'''IT-IDF函数，根据不同的分词方法生成TF-IDF向量'''
badFeatures = badData.map(lambda line: tf.transform(split2(line,distance,step)))
goodFeatures = goodData.map(lambda line: tf.transform(split2(line,distance,step)))
badFeatures.cache()
goodFeatures.cache()
idf = IDF()
idfModel = idf.fit(badFeatures)
badVectors = idfModel.transform(badFeatures)
idfModel = idf.fit(goodFeatures)
goodVectors = idfModel.transform(goodFeatures)
# 这里对训练集打标，bad标记为1，good标记为0
badExamples = badVectors.map(lambda features: LabeledPoint(1, features))
goodExamples = goodVectors.map(lambda features: LabeledPoint(0, features))
dataAll = badExamples.union(goodExamples)
return dataAll
if __name__=="__main__":
sc = SparkContext(appName="webshellCheck")
# bad.txt是恶意URL
bad = sc.textFile("bad.txt")
# good.txt是正常URL
good = sc.textFile("good.txt")
# test.txt是测试集
test = sc.textFile("test.txt")
# 创建一个HashingTF实例来把http文本映射为包含10000个特征的向量
tf = HashingTF(numFeatures = 10000)
# 这里有三种算法，使用某种算法去掉相应的注释即可
# 生成Logistic算法数据
dataLogistic = TFIDF(bad,good,3,1)
# #生成SVMWithSGD算法数据
# dataSVMWithSGD = TFIDF(bad,good,5,5)
# #生成NaiveBayes算法数据
# dataNaiveBayes = TFIDF(bad,good,6,6)
# 使用分类算法进行训练,iterations位迭代次数,step为迭代步长
modelLogistic = LogisticRegressionWithSGD.train(data=dataLogistic,iterations=10000,step=6)
print "LogisticRegressionWithSGD train success"
# modelSVMWithSGD = SVMWithSGD.train(data=dataSVMWithSGD,iterations=100000,step=5)
# print "SVMWithSGD train success"
# modelNaiveBayes = NaiveBayes.train(data=dataNaiveBayes,lambda_=0.1)
# print "NaiveBayes train success"
check(test,modelLogistic)
sc.stop()
运行结果：
LogisticRegressionWithSGD train success 正确率：0.990697391622
我们换一个只有10条数据的测试集test2.txt并修改下测试函数，输出一下测试效果
def check(test,model):
'''训练检测函数'''
countAll = 16447
lines = test.collect()
correctCount = 0
for line in lines:
temp = line[0:line.rfind(" ")]
testExample = tf.transform(split2(temp,3,1))
#try:
if model.predict(testExample)==1:
print line+" ====> "+"恶意URL"
else:
print line+" ====> "+"正常URL"
# except:
# print line
# C = float(correctCount)/countAll
# print "正确率："+str(C)
运行结果：
可以看到识别效果非常完美。
四、日志处理
这一节主要写一写Spark的威力，如何用Spark快速的数据统计并进行机器学习分类，以Apache默认日志为例（Apache详细日志格式大家可以百度一下），如下：
115.28.44.151 - - [28/Mar/2014:00:26:10 +0800] "GET /manager/html HTTP/1.1" 404 162 "-" "Mozilla/3.0 (compatible; Indy Library)"
第一步要对原始日志进行预处理，将其分割成一个列表。使用如下正则
log_Pattern = r'^(?P<remote_addr>.*?) - (?P<remote_user>.*) \[(?P<time_local>.*?)\] "(?P<request>.*?)" '\ '(?P<status>.*?) (?P<body_bytes_sent>.*?) "(?P<http_referer>.*?)" "(?P<http_user_agent>.*?)"$'
可以将日志分割为这样的列表
[u'115.28.44.151', u'-', u'28/03/2014:00:26:10', u'GET', u'/manager/html', u'404', u'162', u'-', u'Mozilla/3.0 (compatible; Indy Library)']
然后就可以很方便的运用RDD操作求出访问最多的前10个IP、前10个URL之类的信息，也可以根据日志中的request部分利用机器学算法进行日志分类，根据上文URL分类的代码，再封装两个类，全部代码如下：
#-*-coding:utf-8-*-
#Author:Xman21
from pyspark import SparkContext
from pyspark.mllib.regression import LabeledPoint
from pyspark.mllib.feature import HashingTF,IDF
from pyspark.mllib.classification import LogisticRegressionWithSGD,SVMWithSGD,NaiveBayes
import re
import time
import sys
reload(sys)
sys.setdefaultencoding('utf8')
class log_Apache():
"""Apache日志切分类，切分后格式为
[ip,user,time,method,request,status,body_bytes,referer,user_agent]"""
def __init__(self):
pass
def re_Split(self,line):
month = {"Jan":"01","Feb":"02","Mar":"03","Apr":"04","May":"05","Jun":"06","Jul":"07","Aug":"08","Sep":"09"
,"Oct":"10","Nov":"11","Dec":"12"}
line_Result = []
log_Pattern = r'^(?P<remote_addr>.*?) - (?P<remote_user>.*) \[(?P<time_local>.*?)\] "(?P<request>.*?)" '\
'(?P<status>.*?) (?P<body_bytes_sent>.*?) "(?P<http_referer>.*?)" "(?P<http_user_agent>.*?)"$'
log_Pattern_Obj = re.compile(log_Pattern).search(line)
try:
remote_addr = log_Pattern_Obj.group('remote_addr')
except:
remote_addr = ''
try:
remote_user = log_Pattern_Obj.group('remote_user')
except:
remote_user = ''
try:
time_local = log_Pattern_Obj.group('time_local').split(" ")[0]
for i in month:
if i in time_local:
time_local = time_local.replace(i,month[i])
break
except:
time_local = ''
try:
temp = log_Pattern_Obj.group('request').split(" ")
if len(temp)>1:
request_mothed = temp[0]
request_url = temp[1]
elif len(temp) == 1:
request_mothed = ''
request_url = temp[0]
else:
request_mothed = ''
request_url = ''
except:
request_mothed = ''
request_url = ''
try:
status = log_Pattern_Obj.group('status')
except:
status = ''
try:
body_bytes_sent = log_Pattern_Obj.group('body_bytes_sent')
except:
body_bytes_sent = ''
try:
http_referer = log_Pattern_Obj.group('http_referer')
except:
http_referer = ''
try:
http_user_agent = log_Pattern_Obj.group('http_user_agent')
except:
http_user_agent = ''
line_Result += [remote_addr,remote_user,time_local,request_mothed,request_url,
status,body_bytes_sent,http_referer,http_user_agent]
return line_Result
def log_Split(self,sc,log_Lines):
re_Split = self.re_Split
log_Result = log_Lines.map(lambda line:re_Split(line))
return log_Result
class Statistics():
"""统计类，获取日志的一些信息"""
def __init__(self):
pass
def IP(self,dataRDD):
"""访问前10的IP统计"""
temp = dataRDD.map(lambda line: (line[0],1)).reduceByKey(lambda x,y:x+y).map(lambda line:(line[1],line[0])).\
sortByKey(False).map(lambda line:(line[1],line[0])).take(10)
return temp
def URL(self,dataRDD):
"""访问前10的页面统计"""
temp = dataRDD.map(lambda line: (line[4],1)).reduceByKey(lambda x,y:x+y).map(lambda line:(line[1],line[0])).\
sortByKey(False).map(lambda line:(line[1],line[0])).take(10)
return temp
class model_Check():
"""机器学习算法类，用以日志分类"""
def __init__(self):
pass
def split2(self,line,distance,step):
'''分词函数，按照一定的步长和距离分词,生成N-gram序列'''
result = []
length = len(line)
for i in xrange(0,length,step):
result.append(line[i:i+distance])
return result
def check(self,tf,test,test_list,model):
'''训练检测函数'''
count = 0
lines_test = test.collect()
lines_test_list = test_list.collect()
correctCount = 0
for line1,line2 in zip(lines_test,lines_test_list):
temp = line2[4]
testExample = tf.transform(self.split2(temp,3,1))
#try:
if model.predict(testExample)==1:
count += 1
print line1+" ====> "+"恶意log"
print count
def TFIDF(self,tf,badData,goodData,distance,step):
'''IT-IDF函数，根据不同的分词方法生成TF-IDF向量'''
split2 = self.split2
badFeatures = badData.map(lambda line: tf.transform(split2(line,distance,step)))
goodFeatures = goodData.map(lambda line: tf.transform(split2(line,distance,step)))
badFeatures.cache()
goodFeatures.cache()
idf = IDF()
idfModel = idf.fit(badFeatures)
badVectors = idfModel.transform(badFeatures)
idfModel = idf.fit(goodFeatures)
goodVectors = idfModel.transform(goodFeatures)
badExamples = badVectors.map(lambda features: LabeledPoint(1, features))
goodExamples = goodVectors.map(lambda features: LabeledPoint(0, features))
dataAll = badExamples.union(goodExamples)
return dataAll
def model_check(self,sc,bad,good,test,test_list):
# 创建一个HashingTF实例来把http文本映射为包含10000个特征的向量
tf = HashingTF(numFeatures = 10000)
# 生成Logistic算法数据
dataLogistic = self.TFIDF(tf,bad,good,3,1)
# 使用分类算法进行训练,iterations位迭代次数,step为迭代步长
modelLogistic = LogisticRegressionWithSGD.train(data=dataLogistic,iterations=10000,step=6)
print "LogisticRegressionWithSGD train success"
self.check(tf,test,test_list,modelLogistic)
sc.stop()
if __name__ == '__main__':
sc = SparkContext(appName = "logCheck")
bad = sc.textFile("bad.txt")
good = sc.textFile("good.txt")
test = sc.textFile("Apache.log")
#日志预处理
step1 = log_Apache()
result1 = step1.log_Split(sc,test)
step2 = Statistics()
print "访问次数最高的前十个IP及其次数："
print step2.IP(result1)
print "访问次数最高的前十个URL及其次数："
print step2.URL(result1)
step3 = model_Check()
step3.model_check(sc,bad,good,test,result1)
可以看到程序能将日志中访问最高的前十个IP和URL统计了出来
也能将日志中的恶意记录识别出来
但同时也有较多的误报，这也是现在机器学习比较棘手的一个问题。
五、总结
说了这么多，那机器学习到底有什么好的呢？这不也存在误报么。误报这个问题我觉得任何入侵检测系统都会存在，但是个人认为机器学习的最大好处之一就是能够识别未知攻击。最常见的例子，当我们去尝试绕过一个WAF去进行SQL注入的时候，只需要找到那么一个字符串组合使其正则无法匹配就好了，比如将select变为sel/**/ect。人的肉眼能直接看出这是个变形的注入语句，机器学习同样具备这种能力，虽然没有见过新型的注入语句，但是在训练集里训练过类似的，那就能识别出来。
当然，机器学习是有弊端的，最大的一点就是速度慢，往往很难达到实时预警的效果。另外就是模型不好训练，因为训练集的构造是很困难的。
六、参考文献
http://www.cnblogs.com/shishanyuan/p/4699644.html
http://www.freebuf.com/articles/web/134334.html
http://www.freebuf.com/articles/web/126543.html
http://www.freebuf.com/articles/network/131279.html
http://blog.csdn.net/v_july_v/article/details/7624837
赞