使用開源工具ELK可視化 Azure NSG日誌

 

國內的Azure最近上線了網絡觀察程序服務，能夠幫助用戶監控和分析VNET虛擬網絡。其中一個很重要的功能就是能夠記錄NSG的安全訪問日誌了。可是若是用戶設置了NSG流日誌，並下載日誌想要分析一下的話，會發現日誌其實並非很友好，NSG流日誌是以json格式記錄的，能夠看到的內容大體以下圖所示：日誌會記錄NSG規則名，系統時間，源地址，目的地址，源端口，目的端口，協議類型，流量方向和處理規則。

不過全部的記錄都連在一塊兒，若是要查找某個具體的安全訪問記錄，很是困難。

 

不過咱們可使用多個多種開源工具將相關日誌數據可視化。最近正好有客戶但願可使用比較流行的ELK套件來分析NSG安全記錄。在參考了微軟和 Elastic.co的官方文檔，以及同事的文章後，終於實現了使用ELK套件獲取並分析Azure.cn的NSG日誌，特此記錄一下過程和注意事項。

簡單介紹一下ELK，ELK是 Elasticsearch、Logstash、Kibana 三個開源軟件的組合。在實時數據檢索和分析場合，三者一般是配合共用，並且又都前後歸於 Elastic.co 公司名下，故此簡稱ELK。它具備以下幾個優勢：

• 處理方式靈活。Elasticsearch 是實時全文索引，不須要預先編程才能使用；
• 配置簡易上手。Elasticsearch 所有采用 JSON 接口，Logstash 是 Ruby DSL 設計，都是目前業界最通用的配置語法設計；
• 檢索性能高效。雖然每次查詢都是實時計算，可是優秀的設計和實現基本能夠達到全天數據查詢的秒級響應；
• 集羣線性擴展。不論是 Elasticsearch 集羣仍是 Logstash 集羣都是能夠線性擴展的；
• 前端操做炫麗。Kibana 界面上，只須要點擊鼠標，就能夠完成搜索、聚合功能，生成炫麗的儀表板。

 

要部署ELK，咱們如今Azure上部署一臺Centos 7.X的虛機，並把這臺虛機的NSG做爲日誌來源。

經過如下步驟開啓NSG流日誌：

 在Azure的服務裏找到「網絡觀察程序」服務。

 

在網絡觀察程序裏，選中NSG流日誌：

 

在右側用資源組過濾，找到須要分析的NSG。

選中NSG，開啓NSG流日誌，指定日誌存放的存儲帳號和日誌保留時間。

 

完成了Azure NSG流日誌的設置，咱們就能夠開始安裝部署ELK，來獲取數據並進行分析了。

 

一、經過SSH登陸到剛纔創建的Centos虛機。先安裝Java運行環境：

 yum install java-1.8.0-openjdk

 

二、下載可信簽名證書：

 rpm --import https://artifacts.elastic.co/GPG-KEY-elasticsearch

 

三、建立 Elasticsearch 的yum repo文件

 vi /etc/yum.repos.d/elasticsearch.repo

 輸入如下內容

 [elasticsearch-6.x]

name=Elasticsearch repository for 6.x packages

baseurl=https://artifacts.elastic.co/packages/6.x/yum

gpgcheck=1

gpgkey=https://artifacts.elastic.co/GPG-KEY-elasticsearch

enabled=1

autorefresh=1

type=rpm-md

 

四、安裝Elasticsearch包

 yum install elasticsearch

 

五、修改配置參數，容許外部訪問

 vi /etc/elasticsearch/ elasticsearch.yml

 

找到network.host參數，設置爲0.0.0.0

# Set the bind address to a specific IP (IPv4 or IPv6):

#

network.host: 0.0.0.0

#

# Set a custom port for HTTP:

#

http.port: 9200

#

 

六、啓動Elasticsearch並設置系統啓動

 systemctl enable elasticsearch.service

 systemctl start elasticsearch.service 

 

七、建立 Logstash的yum repo文件

 vi /etc/yum.repos.d/logstash.repo

 

輸入如下內容

 [logstash-5.x]

name=Elastic repository for 5.x packages

baseurl=https://artifacts.elastic.co/packages/5.x/yum

gpgcheck=1

gpgkey=https://artifacts.elastic.co/GPG-KEY-elasticsearch

enabled=1

autorefresh=1

type=rpm-md

 

八、安裝Logstash包

 yum install logstash

 

九、安裝讀取Azure Blob存儲的插件

 /usr/share/logstash/bin/logstash-plugin install logstash-input-azureblob

 

十、設置存儲讀取位置爲Azure.cn，默認是讀取WW的Azure

 vi /usr/share/logstash/vendor/bundle/jruby/2.3.0/gems/logstash-input-azureblob-0.9.12-java/lib/logstash/inputs/azureblob.rb

 

找到如下內容

config :endpoint, :validate => :string, :default => 'core.windows.net'

 將endpoint修改成中國Azure的endpoint：core.chinacloudapi.cn

config :endpoint, :validate => :string, :default => 'core.chinacloudapi.cn'

 

十一、建立Logstash的配置文件

 在Azure門戶中中找到所使用存儲帳號的訪問密鑰

 

 

vi /etc/logstash/conf.d/logstash.conf

輸入如下內容

 input {

azureblob

  {

      storage_account_name => "此處更改成NSG流日誌所使用的存儲帳戶"

      storage_access_key => "此處更改成存儲帳戶的訪問密鑰"

      container => "insights-logs-networksecuritygroupflowevent"

      codec => "json"

      # Refer https://docs.microsoft.com/en-us/azure/network-watcher/network-watcher-read-nsg-flow-logs

      # Typical numbers could be 21/9 or 12/2 depends on the nsg log file types

      file_head_bytes => 12

      file_tail_bytes => 2

      # Enable / tweak these settings when event is too big for codec to handle.

      # break_json_down_policy => "with_head_tail"

      # break_json_batch_count => 2

  }

}

 

filter {

  split { field => "[records]" }

  split { field => "[records][properties][flows]"}

  split { field => "[records][properties][flows][flows]"}

  split { field => "[records][properties][flows][flows][flowTuples]"}

 

mutate{

split => { "[records][resourceId]" => "/"}

add_field => {"Subscription" => "%{[records][resourceId][2]}"

              "ResourceGroup" => "%{[records][resourceId][4]}"

              "NetworkSecurityGroup" => "%{[records][resourceId][8]}"}

convert => {"Subscription" => "string"}

convert => {"ResourceGroup" => "string"}

convert => {"NetworkSecurityGroup" => "string"}

split => { "[records][properties][flows][flows][flowTuples]" => ","}

add_field => {

            "unixtimestamp" => "%{[records][properties][flows][flows][flowTuples][0]}"

            "srcIp" => "%{[records][properties][flows][flows][flowTuples][1]}"

            "destIp" => "%{[records][properties][flows][flows][flowTuples][2]}"

            "srcPort" => "%{[records][properties][flows][flows][flowTuples][3]}"

            "destPort" => "%{[records][properties][flows][flows][flowTuples][4]}"

            "protocol" => "%{[records][properties][flows][flows][flowTuples][5]}"

            "trafficflow" => "%{[records][properties][flows][flows][flowTuples][6]}"

            "traffic" => "%{[records][properties][flows][flows][flowTuples][7]}"

             }

convert => {"unixtimestamp" => "integer"}

convert => {"srcPort" => "integer"}

convert => {"destPort" => "integer"}       

}

 

date{

 match => ["unixtimestamp" , "UNIX"]

}

}

output {

stdout { codec => rubydebug }

elasticsearch {

 hosts => "localhost"

 index => "nsg-flow-logs"

}

}

 

十二、啓動Logstash並設置系統啓動

 systemctl enable logstash.service

 systemctl start logstash.service

 

1三、建立 Kibana的yum repo文件

 vi /etc/yum.repos.d/kibana.repo

 

輸入如下內容

 [kibana-4.5]

name=Kibana repository for 4.5.x packages

baseurl=http://packages.elastic.co/kibana/4.5/centos

gpgcheck=1

gpgkey=http://packages.elastic.co/GPG-KEY-elasticsearch

enabled=1

 

1四、安裝Kibana包

 yum install kibana

 

1五、設置Kibana參數，容許外部訪問

 vi /etc/kibana/kibana.yml

 

找到如下server.host參數，修改成0.0.0.0

 # Specifies the address to which the Kibana server will bind. IP addresses and host names are both valid values.

# The default is 'localhost', which usually means remote machines will not be able to connect.

# To allow connections from remote users, set this parameter to a non-loopback address.

server.host: "0.0.0.0"

 

1六、啓動Kibana並設置系統啓動

 systemctl enable kibana.service

 systemctl start kibana.service

 

1七、檢查ELK服務狀態

 systemctl status elasticsearch

 

systemctl status logstash

 

systemctl status kibana

 

 

1八、設置虛機的NSG策略，容許外部訪問kibana門戶

 

 

 1九、登陸Kibana門戶

 http://hospip:5601

 

 

 20、設置state:storeInSessionStorage參數

 在「Management」選項卡下的 Advanced Setting裏找到state:storeInSessionStorage參數，並enable

 

 

 2一、在「Management」選項卡下，經過「Index Patterns」 建立索引。

 

 

 2二、微軟已經很貼心的爲你們準備了示例的Dashboard，能夠直接導入，之後你們還能夠按照本身的須要設計

在此處下載儀表板文件，在此處下載可視化效果文件。

在 Kibana 的「Management」選項卡下，經過「Saved Objects」導入文件。 而後,可從「Dashboard」選項卡打開並加載示例Dashboard。

 

 

 進入Dashboard就能夠查看當前設置的一些示例報表

 

在Dashboard的右上角能夠設置日誌分析的時間段，能夠幫助用戶迅速定位到須要分析的時間點

 

 

 

最後感謝康老師的支持。他的文檔給了我很大幫助。

康老師的博客地址：https://www.azure.cn/blog/2017/12/27/AZURE-NSG-FLOW-LOG-Analysis/

Azure官方配置文檔說明：https://docs.microsoft.com/zh-cn/azure/network-watcher/network-watcher-visualize-nsg-flow-logs-open-source-tools

若是須要搭建ELK集羣服務，能夠參考此文檔：https://docs.azure.cn/zh-cn/articles/training/open-source-azure-virtual-machines-create-elk-cluster

ELK官方文檔：https://www.elastic.co/guide/index.html

Azure存儲插件的說明：https://github.com/Azure/azure-diagnostics-tools/tree/master/Logstash/logstash-input-azureblob

 

固然用戶也可使用PowerBI服務來分析NSG的流日誌，這樣就能夠不用本身安裝和部署了。文檔位置：https://docs.microsoft.com/zh-cn/azure/network-watcher/network-watcher-visualize-nsg-flow-logs-power-bi