基于開源平臺和威脅情報的自動化攔截技術實踐

金融企業部署了大量的安全設備及平臺對日趨加劇的網絡攻擊進行安全防護，面對海量的告警數據，如何快速從中識別出真正的惡意地址對其實施攔截是安全人員急需解決的問題。國聯證券基于開源組件，通過自建統一日志管理平臺，對各類安全設備告警日志進行統一管理，并采用自動化方式，將告警源地址匯集清洗后，發往第三方威脅情報平臺進行威脅查詢，對于查詢結果明確為惡意的地址，自動在邊界防護設備中實施永久封禁，實現統一日志平臺、威脅情報、安全設備之間的多方聯動。

一、引言

       近年來，網絡攻擊呈現攻擊源多樣化的趨勢，來自云主機、工控設備、智能產品等具有聯網功能的網絡終端發起的攻擊流量占比明顯增多。根據國家計算機網絡應急技術處理協調中心發布的《2019年中國互聯網網絡安全報告》顯示，在DDOS攻擊中參與真實地址攻擊的肉雞達到340余萬個。網絡安全威脅信息共享平臺發布的《2020上半年公共互聯網網絡安全態勢及威脅監測處置報告》統計數據顯示，該平臺收集的惡意IP地址數量約188萬個，以惡意掃描服務器IP地址為主。

       金融企業由于行業特殊性，天然成為網絡攻擊的首選目標，金融行業的重要信息系統無時無刻不面臨來自互聯網的攻擊流量，在這些流量中，又以利用特定漏洞為目的的無差別掃描為甚。同時金融企業大多已經部署了較為全面的安全防護設備，如防火墻、Web應用防火墻、入侵檢測/防御等，對攻擊行為進行攔截并產生攻擊告警，還有部分企業通過購買第三方機構威脅情報服務，對這些攻擊流量的源地址進行威脅分析，采取進一步的安全防護策略，比如安全設備當前僅對當次攻擊流量進行檢測并攔截，若該地址進行下一次攻擊時，安全設備仍需要再次識別，所以對這些地址進行永久封禁是比較好的選擇，這在重保等特殊時期是非常有用的手段。但由于安全設備的多樣性，安全設備之間告警日志無法統一管理，大多安全設備無法直接與威脅情報系統進行集成，面對海量的告警數據，如何快速從中識別出真正的惡意地址，給安全人員帶來了不小的挑戰。另一方面，由于金融信息系統運行實時性要求較高，為了避免變更對業務系統帶來不可預估的影響，通常在夜間執行變更操作，這就壓縮了運維人員的可操作時間，在識別出真正具備威脅的惡意地址后，如何在安全設備中對這些地址進行準確快速地實施封禁策略同時避免頻繁的變更，也是安全運維人員需要著重考慮的問題。

       國聯證券基于開源組件，通過自建統一日志管理平臺，對各類安全設備告警日志進行統一管理，并采用自動化方式，將告警源地址匯集清洗后，發往第三方威脅情報平臺進行威脅查詢，對于查詢結果明確為惡意的地址，自動在邊界防護設備中實施永久封禁，達到不同安全設備、威脅情報、統一日志平臺三者之間多方聯動效果。

二、技術架構

（一）總體設計

       本技術實踐總體設計框架由安全設備、統一日志平臺、威脅情報源和處理引擎四部分組成，技術架構如圖1所示：

圖1

（1）安全設備

       安全設備在企業安全防護中起到攻擊監測告警和攻擊攔截作用，是安全日志的直接輸出者和防護策略生效者。但在企業安全實踐中，部署的安全設備往往存在品牌多樣、類型繁復等問題，不同類型品牌的安全設備提供的安全功能不盡相同，產生的告警日志也無法做到格式統一，需要在統一日志平臺進行結構化處理后再行入庫。另外由于邊界安全設備起到對企業整網入口的防護功能，所以最終經威脅情報判定為惡意的源地址，應在邊界設備進行黑名單寫入。

（2）統一日志平臺

       統一日志平臺是本次技術實踐的核心節點，提供了安全日志處理、存儲、分析、查詢等核心功能，也是安全運營、態勢感知等上層安全系統依賴的基礎。本例直接使用ELK+Kafka的形式搭建自有統一日志平臺。ELK指的是Elastic Stack，由Beat、Logstash、Elasticsearch和Kibana四個開源組件組成，Beat用于接收安全設備的Syslog日志；Logstash 是服務器端數據處理管道，能夠同時從多個來源采集數據，轉換數據；Elasticsearch 是一個基于Lucene的搜索和分析引擎；Kibana可以對Elasticsearch 中存儲的數據以圖形和圖表形式進行可視化；Kafka是一個消息隊列組件，加入Kafka的目的，是為了將安全設備的原始日志通過Beat接收后先發送至Kafka消息隊列，再由Logstash進行消費，避免Logstash在日志轉換過程中由于處理性能問題導致原始日志積壓或丟失。

（3）威脅情報源

       根據Gartner對威脅情報的定義，威脅情報是某種基于證據的知識，主要內容為用于識別和檢測威脅的失陷標識，市面上知名的威脅情報源有微步在線情報社區、綠盟威脅情報中心、IBM X-Force Exchange威脅情報共享平臺等，在本例中，主要利用了微步威脅情報的云API接口，實現自動化的情報查詢和獲取。

（4）處理引擎

       處理引擎的作用主要是檢索日志平臺中存儲的結構化日志數據，將源地址進行清洗后批量在威脅情報源中查詢，并將判定為惡意的地址自動寫入邊界防護設備要讀取的黑名單列表。

（二）處理流程

       整體的處理流程大致可分為安全日志轉換與存儲、地址清洗與情報查詢、黑名單寫入與策略下發三個步驟，如圖2所示：

圖2

1、安全日志轉換與存儲

       在本次技術實踐中，采用標準化的Syslog協議，將安全設備的原始日志傳送至統一日志平臺，由日志平臺對原始日志進行統一結構化處理，格式化成由Key-Value構成的JSON鍵值對形式保存在日志平臺中，方便日后對日志的分析和查詢。本例中我們在部署在網絡邊界的下一代防火墻（NGFW）上啟用了動態地址對象讀取功能，并設置一條對動態地址執行DROP的默認策略，實現自動化的地址讀取和策略加載。

2、地址清洗與情報查詢

       處理引擎可以實現讀取日志平臺的結構化日志數據，提取有用的字段，本例中主要是源地址這一字段，在獲取源地址后，需要進行去重，再將地址列表通過請求威脅情報API方式，查詢該部分地址的威脅屬性，確認是否為惡意地址。

3、黑名單寫入與策略下發

       在經過威脅情報源判定后，可以獲得源地址的威脅屬性，對于判定為惡意的源地址，由處理引擎輸出至黑名單列表，該列表以Web資源方式部署在Web服務上供NGFW讀取，NGFW在讀取黑名單列表后，會自動將惡意IP關聯至策略源地址對象，執行阻斷操作。

三、具體實現

       按照核心功能的不同，可以從統一日志平臺和處理引擎兩個模塊進行技術實現，下面主要對這兩個模塊的具體實現細節進行分析。

（一）統一日志平臺

       統一日志平臺在Elastic Stack技術棧的基礎上，加入Kafka消息隊列，實現對安全設備告警日志的統一管理，系統架構如圖3所示。

圖3

1、日志采集

       日志采集主要由Syslog、Beats、Kafka三部分組成，完成了從安全設備源端將告警日志采集至消息隊列的過程。

(1)Syslog

      Syslog廣泛應用于系統日志，是日志傳輸的消息標準，安全設備均支持以Syslog方式將自身日志發送至第三方平臺。這里提供兩種收集Syslog的方式，第一種可以依賴Linux系統中的rsyslog系統，接收安全設備的日志，轉存為文件保存在rsyslog服務器上，再使用Beats家族的Filebeat組件，對日志文本進行采集；第二種可以直接使用Beats的Syslog功能，監聽安全設備發送過來的Syslog，兩種方式的區別在于前一種多保留了一次設備的原始日志。

 (2)Kafka

      Kafka是一種高吞吐的分布式消息發布訂閱系統，引入Kafka的目的，是為了解決下游的Logstash在進行日志轉換時可能產生的日志丟失問題，若直接將Syslog發送至Logstash，因Logstash需要進行大量的正則匹配處理，當Logstash未能及時處理消息時，就可能造成日志丟失。而引入Kafka后，Logstash就可以根據自身的處理能力，順序從Kafka消費日志數據，避免了日志丟失。

       在接收Syslog后，需要通過Beats，將Syslog發送至Kafka。在Kafka的配置上，應考慮建立的Topics數，在本例中，我們根據不同品牌的產品，建立Topic，對于同一品牌的設備，無論部署位置，均發送至同一Topic，便于讀取，如圖4，建立了checkpoint、paloalto、waf三個topic，分別存儲NGFW和WAF的告警日志，經過Beats采集后的日志，安全設備的原始Syslog會被存放在message字段中，如圖5。

圖4

圖5

2、日志轉換

       日志轉換的作用是將Message中的Syslog進行結構化清洗，轉換成鍵值對形式，再存儲至Elasticsearch。這里需要依賴Logstash組件完成，Logstash先從Kafka消息隊列消費日志數據，再對數據進行清洗與轉換，丟棄不需要的字段，對于部分日志內容，添加自定義字段名，然后生成Elasticsearch需要的索引。表1是對WAF日志進行轉換后，最終的字段形式（部分）。

表1

        圖6展示了經過Logstash處理后，一串原始Syslog在Elasticsearch中存儲的形式。在對Syslog進行轉換時，需要在Logstash的配置文件中，按照不同的Kafka Topic，編寫不同的解析策略，生成不同的Elasticsearch Index。

圖6

3、日志存儲

(1)Elasticsearch

       經過Logstash轉換的Syslog數據，便可以Kafka Topic加日期為名稱在Elasticsearch建立索引。Elasticsearch是Elastic Stack技術棧中最重要的組件，是一個高擴展的分布式全文檢索引擎，可以近乎實時的存儲、檢索數據，同時支持方便的進行擴展。Elasticsearch屬于非關系型數據庫的一種，本質上存儲的數據就是JSON格式的文檔，這里存儲的就是JSON格式的Syslog數據，Elasticsearch對JSON文檔中的每一個字段都會構建一個對應的倒排索引。    

(2)Kibana

      Kibana則提供了一個友好的Web界面，可以搜索、查看、操作存儲在Elasticsearch索引中的數據，通過使用Kibana能對處理后的數據進行可視化的展示。在將Elasticsearch Index全部匯總至Kibana Index后，便可以繪制儀表盤，對告警日志進行集中展示，如圖7。

圖7

（二）處理引擎

       處理引擎需要實現日志檢索、情報查詢和黑名單寫入三方面的功能，Java或Python均能較為容易的實現上述功能的代碼編寫，這里采用Python語言來編寫處理引擎。

       另一方面，由于金融行業變更操作規范要求，一般不在日間執行變更操作，故處理引擎并不適合實時運行，因此將處理引擎設定為夜間運行是比較好的做法。處理引擎匯總前一天告警日志中的所有源地址，去重后在威脅情報源查詢，再將惡意地址寫入黑名單列表，最后由邊界防護設備自動讀取黑名單，下發攔截策略。

1、日志檢索

       官方提供了Python用于ElasticSearch數據檢索的專用代碼包，只需引入ElasticSearch包，便可方便的操作ElasticSearch存儲的數據。這部分的主要代碼邏輯就是從ElasticSearch中讀取前一天所有安全設備Index的所有日志數據，提取源地址字段的值并進行去重，最終作為要發往威脅情源報進行查詢的地址。

圖8

2、情報查詢

       情報查詢需要連接第三方威脅情報源API，這里采用微步在線情報社區提供的在線云API。微步云API提供了IP 分析、IP 信譽、域名分析等基礎功能接口，支持HTTP方式對這些接口進行調用，由于微步在線對一個地址是否為惡意自有的一套判定機制，這里直接采用微步情報的判定結果，scene/ip_reputation接口的is_malicious響應字段表示該地址是否為惡意。可以將接口返回的is_malicious（是否為惡意IP）、confidence_level（可信度）、severity（嚴重級別）等數據與地址列表合并形成一個Dataframe格式的數據，寫入查詢結果文件，如圖9。

圖9

3、黑名單寫入

       處理引擎的黑名單寫入需要先過濾出is_malicious為Ture的行，再將這些行的源地址，寫入一個文本文件，這也在處理引擎實現。之后邊界防火墻便可以讀取寫有惡意地址的文本文件，以Paloalto防火墻為例，Paloalto防火墻的外部動態列表功能，支持從一個URL讀取地址列表，寫入自身的地址對象，并且可以規定讀取的時間和頻率。為了配合處理引擎的運行時間，通常設定在處理引擎生成每日的封禁地址列表之后進行讀取，隨后便可創建一條默認攔截策略，源地址設置為外部動態列表對象。對于不支持類似外部動態列表功能的防火墻，也可以使用腳本方式，通過命令行將這些惡意地址批量刷入配置文件，以達到相同的功能。

四、總結

       本次技術實踐對傳統安全運維中依賴人工執行惡意地址封禁這一典型場景進行了優化與改進，首先利用Elastic Stack開源平臺對不同品牌類型的安全設備告警日志實現了統一管理，相較于使用某一廠商的特定統一日志管理系統，使用開源組件具有更好的靈活性與可維護性。在統一日志管理的基礎上，本次技術實踐還引入了威脅情報，可以更好地對惡意地址從各維度進行分析，最終結合威脅情報判定結果，對惡意地址實施自動化封禁，實現統一日志平臺、威脅情報、安全設備之間的多方聯動。

       隨著網絡攻擊的日益增多，也催生了更多種類的安全設備、安全技術或安全名詞，對于金融企業來說，如何將不同的安全產品納入統一管理，避免產生安全孤島，真正實現多方聯動，是安全建設的重點也是難點。國聯證券將對此進行更多的探索，提高安全運營水平。

文章來源：上交所技術服務