使用深度學習檢測DGA（域名生成算法）——LSTM的輸入數據本質上還是詞袋模型

from:http://www.freebuf.com/articles/network/139697.html

DGA（域名生成算法）是一種利用隨機字符來生成C&C域名，從而逃避域名黑名單檢測的技術手段。例如，一個由Cryptolocker創建的DGA生成域xeogrhxquuubt.com，如果我們的進程嘗試其它建立連接，那么我們的機器就可能感染Cryptolocker勒索病毒。域名黑名單通常用於檢測和阻斷這些域的連接，但對於不斷更新的DGA算法並不奏效。我們的團隊也一直在對DGA進行廣泛的研究，並在arxiv發表了一篇關於使用深度學習預測域生成算法的文章。

本文我將為大家介紹一種，簡單而有效的DGA生成域的檢測技術。我們將利用神經網絡（或稱之為深度學習）更具體的來講就是長短期記憶網絡（LSTM），來幫助我們檢測DGA生成域。首先我們會探討深度學習的優勢，然后我將進一步的通過實例來驗證我的論述。

如果你之前對機器學習並不了解，那么我建議你先翻看我之前發布的三篇關於機器學習的文章再來閱讀本文，這樣會更有助於你的理解。

長短期記憶網絡（LSTM）的好處

深度學習近年來在機器學習社區中可以說是占盡風頭。深度學習是機器學習中一種基於對數據進行表征學習的方法。其好處是用非監督式或半監督式的特征學習和分層特征提取高效算法來替代手工獲取特征。隨着數十年的不斷發展，深度學習在過去四五年間一直很受歡迎。再加上硬件的不斷升級優化（如GPU的並行處理改進），也使得培訓復雜網絡成為了可能。LSTM是一種RNN的特殊類型，可以學習長期依賴信息，如文本和語言等。LSTM是實現循環神經網絡的一個這樣的技巧，意味着包含循環的神經網絡。LSTM在長時間的學習模式方面非常擅長如文本和言語。在本文的例子中，我將使用它們來學習字符序列（域名）的模式，從而幫助我們識別哪些是DGA生成域哪些不是。

使用深度學習的一大好處就是我們可以省去特征工程這一繁雜的過程。而如果我們使用常規方法來生成一長串特征列表（例如長度，元音，輔音以及n-gram模型），並使用這些特征來識別DGA生成域和非DGA生成域。那么就需要安全人員實時的更新和創建新的特征庫，這將是一個異常艱巨和痛苦的過程。其次，一旦攻擊者掌握了其中的過濾規則，那么攻擊者就可以輕松地通過更新其DGA來逃避我們的檢測。而深度學習的自動表征學習能力，也讓我們能夠更快的適應不斷變化的對手。同時，也大大減少了我們人力物力的巨大投入。我們技術的另一個優點是僅對域名進行識別而不使用任何上下文功能，如NXDomains

我們技術的另一個優點是，我們僅對域名進行分類而不使用任何上下文功能如NXDomain。上下文功能的生成往往需要額外昂貴的基礎設施（如網絡傳感器和第三方信譽系統）。令人驚訝的是對於沒有上下文信息的LSTM，執行卻明顯優於它們。如果你想了解更多關於LSTM的相關內容，我推薦大家可以關注：colah的博客和blogdeeplearning.net。

什么是DGA？

首先我們要搞清楚什么是DGA？以及DGA檢測的重要性。攻擊者常常會使用域名將惡意程序連接至C&C服務器，從而達到操控受害者機器的目的。這些域名通常會被編碼在惡意程序中，這也使得攻擊者具有了很大的靈活性，他們可以輕松地更改這些域名以及IP。而對於另外一種硬編碼的域名，則往往不被攻擊者所采用，因為其極易遭到黑名單的檢測。

而有了DGA域名生成算法，攻擊者就可以利用它來生成用作域名的偽隨機字符串，這樣就可以有效的避開黑名單列表的檢測。偽隨機意味着字符串序列似乎是隨機的，但由於其結構可以預先確定，因此可以重復產生和復制。該算法常被運用於惡意軟件以及遠程控制軟件上。

我們來簡單了解下攻擊者和受害者端都做了哪些操作。首先攻擊者運行算法並隨機選擇少量的域（可能只有一個），然后攻擊者將該域注冊並指向其C2服務器。在受害者端惡意軟件運行DGA並檢查輸出的域是否存在，如果檢測為該域已注冊，那么惡意軟件將選擇使用該域作為其命令和控制（C2）服務器。如果當前域檢測為未注冊，那么程序將繼續檢查其它域。

安全人員可以通過收集樣本以及對DGA進行逆向，來預測哪些域將來會被生成和預注冊並將它們列入黑名單中。但DGA可以在一天內生成成千上萬的域，因此我們不可能每天都重復收集和更新我們的列表。

圖1展示了許多類型的惡意軟件的工作流程。如圖所示惡意軟件會嘗試連接三個域：asdfg.com，wedcf.com和bjgkre.com。前兩個域未被注冊，並從DNS服務器接收到NXDomain響應。第三個域已被注冊，因此惡意軟件會使用該域名來建立連接。

見原文。。。

我們提出了使用神經網絡來檢測DGA的簡單技術。該技術不需要使用任何上下文信息（如NXDomains和第三方信譽系統），並且檢測效果也遠遠優於現有的一些技術。本文只是我們關於DGA研究的部分摘要，你可以點擊閱讀我們在arxiv的完整文章。同時，你也可以對我們發布在github上的代碼做進一步的研究學習。

 

 

核心代碼如下：

"""Train and test LSTM classifier"""
import dga_classifier.data as data
import numpy as np
from keras.preprocessing import sequence
from keras.models import Sequential
from keras.layers.core import Dense, Dropout, Activation
from keras.layers.embeddings import Embedding
from keras.layers.recurrent import LSTM
import sklearn
from sklearn.cross_validation import train_test_split


def build_model(max_features, maxlen):
    """Build LSTM model"""
    model = Sequential()
    model.add(Embedding(max_features, 128, input_length=maxlen))
    model.add(LSTM(128))
    model.add(Dropout(0.5))
    model.add(Dense(1))
    model.add(Activation('sigmoid'))

    model.compile(loss='binary_crossentropy',
                  optimizer='rmsprop')

    return model

def run(max_epoch=25, nfolds=10, batch_size=128):
    """Run train/test on logistic regression model"""
    indata = data.get_data()

    # Extract data and labels
    X = [x[1] for x in indata] labels = [x[0] for x in indata] # Generate a dictionary of valid characters
    valid_chars = {x:idx+1 for idx, x in enumerate(set(''.join(X)))} max_features = len(valid_chars) + 1 maxlen = np.max([len(x) for x in X]) # Convert characters to int and pad
    X = [[valid_chars[y] for y in x] for x in X] X = sequence.pad_sequences(X, maxlen=maxlen) # Convert labels to 0-1
    y = [0 if x == 'benign' else 1 for x in labels]

    final_data = []

    for fold in range(nfolds):
        print "fold %u/%u" % (fold+1, nfolds)
        X_train, X_test, y_train, y_test, _, label_test = train_test_split(X, y, labels, 
                                                                           test_size=0.2)

        print 'Build model...'
        model = build_model(max_features, maxlen)

        print "Train..."
        X_train, X_holdout, y_train, y_holdout = train_test_split(X_train, y_train, test_size=0.05)
        best_iter = -1
        best_auc = 0.0
        out_data = {}

        for ep in range(max_epoch):
            model.fit(X_train, y_train, batch_size=batch_size, nb_epoch=1)

            t_probs = model.predict_proba(X_holdout)
            t_auc = sklearn.metrics.roc_auc_score(y_holdout, t_probs)

            print 'Epoch %d: auc = %f (best=%f)' % (ep, t_auc, best_auc)

            if t_auc > best_auc:
                best_auc = t_auc
                best_iter = ep

                probs = model.predict_proba(X_test)

                out_data = {'y':y_test, 'labels': label_test, 'probs':probs, 'epochs': ep,
                            'confusion_matrix': sklearn.metrics.confusion_matrix(y_test, probs > .5)}

                print sklearn.metrics.confusion_matrix(y_test, probs > .5)
            else:
                # No longer improving...break and calc statistics
                if (ep-best_iter) > 2:
                    break

        final_data.append(out_data)

return final_data