作者：小黑豬（朱文哲）@銀河安全實驗室
公眾號：銀河安全實驗室
之前在《開源工控安全研究框架ISF介紹》這篇文章中，提到了可以利用ISF中的工控協議模塊對設備進行進行Fuzz測試，這篇文章將介紹如何具體的使用KittyFuzzer框架來實現。
由于文章主要描述的是如何利用Kitty框架結合ISF中的工控協議組件進行Fuzz，因此不會對Kitty框架本身進行過多的說明，如果對Kitty框架的參數及如何使用存在困惑的可以參考一下Kitty的官方文檔
1. 工具介紹
1.1 kittyFuzzer
Kitty是一個用python編寫的模塊化及可擴展的開源fuzzing框架，其設計靈感來自OpenRCE的Sulley和Michael Eddington的Peach Fuzzer(現在是Deja Vu Security的)。
Kitty的設計之初是用于幫助我們fuzz一些非常規的目標(例如一些復雜的非TCP/IP通訊的協議)時不用每次都重復的實現一些基礎的功能。因此Kitty被設計成一個通用的抽象的框架，Kitty本身包含了我們所能想到的每個Fuzz測試過程中的公共功能，并且允許用戶根據他們特定的目標輕松的進行擴展。
1.2 ISF
ISF是我之前從事工業控制設備漏洞挖掘工作時積累的工控協議和POC代碼等內容進行整合后的產物，后來將其中的一部分內容進行了開源，項目的地址是https://github.com/dark-lbp/isf。
ISF，是一款基于python編寫的類似[Metasploit]的工控漏洞利用框架。
ISF基于開源項目[routersploit]修改而來，在routersploit基礎框架上針對工控設備增加了工控協議的客戶端、工控協議模塊等功能。
2. Fuzz Modbus協議
Modbus協議是在工業控制領域使用的非常廣泛的一個基礎協議，其協議格式也較為簡單，沒有復雜的協議狀態機。下面會對如何利將Kitty框架與ISF框架中的工控協議組件相結合對Modbus-TCP協議執行fuzzing測試進行說明。
2.1. 導入需要的python庫
在進行Fuzzing測試之前我們需要使用Kitty框架來構造測試用例，首先我們需要導入一下基礎的庫。
# 從Kitty中導入Template等一系列基礎組件
from kitty.model import Template
from kitty.interfaces import WebInterface
from kitty.fuzzers import ServerFuzzer
from kitty.model import GraphModel
# 從Kitty擴展庫katnip中導入TcpTarget用于Fuzz TCP目標
from katnip.targets.tcp import TcpTarget
# 從Kitty擴展庫katnip中導入scapy模塊用于直接使用Scapy的數據結構
from katnip.model.low_level.scapy import *
# 從ISF中導入modbus_tcp相關的數據包結構
from icssploit.protocols.modbus_tcp import *2.2. 定義基礎參數及數據結構
在導入了需要的模塊后，還需要對目標Fuzzing測試對象的IP地址，通訊端口等一些基礎參數進行設置。
# 定義目標Fuzz對象的IP地址
TARGET_IP = '172.16.22.131'
# 定義目標Fuzz對象的通訊端口
TARGET_PORT = 502
# 定義隨機數種子
RANDSEED = int(RandShort())在Modbus-TCP Fuzzing的例子中，我們將使用Modbus-TCP協議中的ReadCoilsRequest請求進行測試，下圖是一個典型的Modbus Read Coils請求數據包。
下面的代碼將介紹如何利用ScapyField將ISF框架中Modbus-TCP協議的ReadCoilsRequest數據包結構直接應用于Kitty中的例子。
# 根據ISF中Modbus-tcp協議的數據結構構造測試數據包,下面例子中將使用RandShort對請求的地址及bit位長度進行測試。
read_coils_request_packet = ModbusHeaderRequest(func_code=0x01)/ReadCoilsRequest(ReferenceNumber=RandShort(), BitCount=RandShort())
# 使用ScapyField直接將Scapy的數據包結構應用于Kitty框架中。
read_coils_request_template = Template(name='Read Coils Request', fields=[
ScapyField(read_coils_request_packet,
name='read_coils_request_packet', # 定義這個Field的名字，用于在報告中顯示
fuzzable=True, # 定義這個Field是否需要Fuzz
seed=RANDSEED, # 定義用于變異的隨機數
fuzz_count=2000 # 這個數據結構的fuzz次數
),
])
# 使用GraphModel進行Fuzz
model = GraphModel()
# 在使用GraphModel中注冊第一個節點，由于Modbus的Read Coils請求是單次的請求/回答形式，因此這里只要注冊簡單的一個節點即可。
model.connect(read_coils_request_packet)2.3. 進行Fuzz模式配置
在完成了基礎的定義后，還需要對Fuzz的模式等進行如下配置。
# 定義一個目標Target, 設置IP、端口及連接超時時間。
modbus_target = TcpTarget(name='modbus target', host=TARGET_IP, port=TARGET_PORT, timeout=2)
# 定義是需要等待Target返回響應，如果設置為True Target不返回數據包則會被識別成異常進行記錄。
modbus_target.set_expect_response(True)
# 定義使用ServerFuzzer的方式進行Fuzz測試
fuzzer = ServerFuzzer()
# 定義fuzzer使用的交互界面為web界面
fuzzer.set_interface(WebInterface(port=26001))
# 在fuzzer中定義使用GraphModel
fuzzer.set_model(model)
# 在fuzzer中定義target為modbus_target
fuzzer.set_target(modbus_target)
# 定義每個測試用例發送之間的延遲
fuzzer.set_delay_between_tests(0.1)
# 開始執行Fuzz
fuzzer.start()完成上述的設置后即可使用python命令執行這個腳本對目標進行測試了，運行后將會在命令行終端看到如下圖所示的輸出。
此時也可以通過開啟的web管理界面來查看Fuzzing測試的狀態。
上面介紹的是最基礎的直接使用ISF中的數據結構來對協議進行Fuzzing的例子，如果想針對性的測試Modbus不同的功能碼，那么就需要修改數據結構定義的部分，下面是Fuzzing寫線圈數據包的例子。
write_coils_request_packet = ModbusHeaderRequest(func_code=0x05)/WriteSingleCoilRequest(ReferenceNumber=RandShort(), Value=RandShort())
# 使用ScapyField直接將Scapy的數據包結構應用于Kitty框架中
write_coils_request_packet_template = Template(name='Write Coils Request', fields=[
ScapyField(write_coils_request_packet,
name='wrire_coils_request_packet', # 定義這個Field的名字，用于在報告中顯示
fuzzable=True, # 定義這個Field是否需要Fuzz
seed=RANDSEED, # 定義用于變異的隨機數
fuzz_count=2000 # 這個數據結構的fuzz次數
),
])
model.connect(write_coils_request_packet_template)2.4. 完整代碼
完整Fuzzing測試用例代碼如下。
# !/usr/bin/env python2
# coding=utf-8
from kitty.model import Template
from kitty.interfaces import WebInterface
from kitty.fuzzers import ServerFuzzer
from kitty.model import GraphModel
from katnip.targets.tcp import TcpTarget
from katnip.model.low_level.scapy import *
from icssploit.protocols.modbus_tcp import *
TARGET_IP = '172.16.22.131'
TARGET_PORT = 502
RANDSEED = int(RandShort())
# 根據ISF中Modbus-tcp協議的數據結構構造測試數據包,下面例子中將使用RandShort對請求的地址及寫入的值進行變異測試。
write_coils_request_packet = ModbusHeaderRequest(func_code=0x05)/WriteSingleCoilRequest(ReferenceNumber=RandShort(), Value=RandShort())
# 使用ScapyField直接將Scapy的數據包結構應用于Kitty框架中
write_coils_request_packet_template = Template(name='Write Coils Request', fields=[
ScapyField(write_coils_request_packet,
name='wrire_coils_request_packet', # 定義這個Field的名字，用于在報告中顯示
fuzzable=True, # 定義這個Field是否需要Fuzz
seed=RANDSEED, # 定義用于變異的隨機數
fuzz_count=2000 # 這個數據結構的fuzz次數
),
])
# 使用GraphModel進行Fuzz
model = GraphModel()
# 在使用GraphModel中注冊第一個節點。
model.connect(write_coils_request_packet_template)
# 定義一個目標Target, 設置IP、端口及連接超時時間。
modbus_target = TcpTarget(name='modbus target', host=TARGET_IP, port=TARGET_PORT, timeout=2)
# 定義是需要等待Target返回響應，如果設置為True Target不返回數據包則會被識別成異常進行記錄。
modbus_target.set_expect_response(True)
# 定義使用基礎的ServerFuzzer進行Fuzz測試
fuzzer = ServerFuzzer()
# 定義fuzzer使用的交互界面為web界面
fuzzer.set_interface(WebInterface(port=26001))
# 在fuzzer中定義使用GraphModel
fuzzer.set_model(model)
# 在fuzzer中定義target為modbus_target
fuzzer.set_target(modbus_target)
# 定義每個測試用例發送之間的延遲
fuzzer.set_delay_between_tests(0.1)
# 開始執行Fuzz
fuzzer.start()3. Fuzz 西門子s7comm協議
西門子s7comm協議是大部分西門子S7-300/400系列PLC默認使用的通訊協議，s7comm協議與Modbus-TCP協議有所不同，s7comm協議由TPKT協議及ISO-COTP協議封裝后進行傳輸，且發送實際控制指令前必須先經過建立COTP連接及配置s7comm通訊參數這兩個步驟。下面會對如何解決這些問題，對西門子S7comm協議執行Fuzzing測試進行說明。
3.1. 導入需要的python庫
此處的操作和進行Modbus-TCP Fuzzing測試時基本相同，只需要額外引入S7comm的協議數據結構即可。
from kitty.model import Template
from kitty.interfaces import WebInterface
from kitty.fuzzers import ServerFuzzer
from kitty.model import GraphModel
from katnip.targets.tcp import TcpTarget
from katnip.model.low_level.scapy import *
# 從ISF中導入s7comm相關的數據包結構
from icssploit.protocols.s7comm import *3.2. 定義基礎參數及數據結構
如之前所說，S7comm協議在發送具體的請求參數前，需要先建立COTP連接并進行通訊參數配置，具體的流程如下圖所示。
因此我們在對請求操作進行Fuzzing測試時也需要先事先建立COTP連接，完成通訊參數配置后才能發送對應的測試數據。
首先需要對目標設備的一些連接信息進行設置，TSAP相關的參數涉及到目標設備的槽位號和連接方式等信息被用于建立COTP-CR連接時使用，具體參數需要根據實際測試設備在編程時的槽位號進行調整。
# snap7 server 配置信息
TARGET_IP = '172.16.22.131'
TARGET_PORT = 102
RANDSEED = int(RandShort())
SRC_TSAP = "0100".encode('hex') # COTP CR請求的參數
DST_TSAP = "0102".encode('hex') # COTP CR請求的參數隨后我們則需要進一步構造用于建立連接的數據包結構及需要進行fuzzing測試的數據包格式， 具體的代碼及說明如下。
# 定義COTP CR建立連接數據包
COTP_CR_PACKET = TPKT()/COTPCR()
COTP_CR_PACKET.Parameters = [COTPOption() for i in range(3)]
COTP_CR_PACKET.PDUType = "CR"
COTP_CR_PACKET.Parameters[0].ParameterCode = "tpdu-size"
COTP_CR_PACKET.Parameters[0].Parameter = "\x0a"
COTP_CR_PACKET.Parameters[1].ParameterCode = "src-tsap"
COTP_CR_PACKET.Parameters[2].ParameterCode = "dst-tsap"
COTP_CR_PACKET.Parameters[1].Parameter = SRC_TSAP
COTP_CR_PACKET.Parameters[2].Parameter = DST_TSAP
# 因為是建立連接使用，因此fuzzable參數需要設置為False避免數據包被變異破壞。
# 如果想對建立連接的過程也進行分Fuzz的話，則可以另行編寫測試用例。
COTP_CR_TEMPLATE = Template(name='cotp_cr', fields=[
ScapyField(COTP_CR_PACKET, name='cotp_cr', fuzzable=False),
])
# 定義通訊參數配置數據結構
SETUP_COMM_PARAMETER_PACKET = TPKT() / COTPDT(EOT=1) / S7Header(ROSCTR="Job", Parameters=S7SetConParameter())
# 因為是建立連接使用，因此fuzzable參數需要設置為False避免數據包被變異破壞。
# 如果想對建立連接的過程也進行分Fuzz的話，則可以另行編寫測試用例。
SETUP_COMM_PARAMETER_TEMPLATE = Template(name='setup comm template', fields=[
ScapyField(SETUP_COMM_PARAMETER_PACKET, name='setup comm', fuzzable=False),
])
# 定義需要Fuzzing的數據包結構, 下面例子中將使用RandShort對請求的SZLId及SZLIndex值進行變異測試。
READ_SZL_PACKET = TPKT() / COTPDT(EOT=1) / S7Header(ROSCTR="UserData", Parameters=S7ReadSZLParameterReq(),Data=S7ReadSZLDataReq(SZLId=RandShort(), SZLIndex=RandShort()))
# 定義READ_SZL_TEMPLATE為可以進行變異的結構，fuzzing的次數為1000次
READ_SZL_TEMPLATE = Template(name='read szl template', fields=[
ScapyField(READ_SZL_PACKET, name='read szl', fuzzable=True, fuzz_count=1000),
])
# 在完成了上述的結構定義后就可以使用GraphModel將這些數據包結構進行前后關聯。
# 使用GraphModel進行Fuzz
model = GraphModel()
# 在GraphModel中注冊第一個節點, 首先發送COTP_CR請求。
model.connect(COTP_CR_TEMPLATE)
# 在GraphModel中注冊第二個節點, 在發送完COTP_CR后發送SETUP_COMM_PARAMETER請求。
model.connect(COTP_CR_TEMPLATE, SETUP_COMM_PARAMETER_TEMPLATE)
# 在GraphModel中注冊第三個節點, 在發送完SETUP_COMM_PARAMETER后發送READ_SZL請求。
model.connect(SETUP_COMM_PARAMETER_TEMPLATE, READ_SZL_TEMPLATE)3.3. 進行Fuzz模式配置
在完成了上述的定義后，剩下的配置和Modbus基本一致，只需要修改一下Target的名稱等即可。
# define target
s7comm_target = TcpTarget(name='s7comm target', host=TARGET_IP, port=TARGET_PORT, timeout=2)
# 定義是需要等待Target返回響應，如果設置為True Target不返回數據包則會被識別成異常進行記錄。
s7comm_target.set_expect_response(True)
# 定義使用基礎的ServerFuzzer進行Fuzz測試
fuzzer = ServerFuzzer()
# 定義fuzzer使用的交互界面為web界面
fuzzer.set_interface(WebInterface(port=26001))
# 在fuzzer中定義使用GraphModel
fuzzer.set_model(model)
# 在fuzzer中定義target為s7comm_target
fuzzer.set_target(s7comm_target)
# 定義每個測試用例發送之間的延遲
fuzzer.set_delay_between_tests(0.1)
# 開始執行Fuzz
fuzzer.start()完成上述的設置后即可使用python命令執行這個腳本對目標進行測試了，運行后將會在命令行終端看到如下圖所示的輸出。
此時也可以通過開啟的web管理界面來查看Fuzzing測試的狀態
上面介紹的是使用Kitty結合ISF中的協議數據結構對西門子S7comm協議進行Fuzzing的例子，如果想針對性的測試S7comm的不同協議功能碼依舊需要修改數據結構定義的部分。
3.4. 完整代碼
下面是Fuzzing Read SZL結構的完整測試用例。
#!/usr/bin/python2
# coding:utf-8
from kitty.model import Template
from kitty.interfaces import WebInterface
from kitty.fuzzers import ServerFuzzer
from kitty.model import GraphModel
from katnip.targets.tcp import TcpTarget
from katnip.model.low_level.scapy import *
# 從ISF中導入cotp相關的數據包結構
from icssploit.protocols.cotp import *
# 從ISF中導入s7comm相關的數據包結構
from icssploit.protocols.s7comm import *
# snap7 server 配置信息
TARGET_IP = '172.16.22.131'
TARGET_PORT = 102
RANDSEED = int(RandShort())
SRC_TSAP = "0100".encode('hex')
DST_TSAP = "0103".encode('hex')
# 定義COTP CR建立連接數據包
COTP_CR_PACKET = TPKT()/COTPCR()
COTP_CR_PACKET.Parameters = [COTPOption() for i in range(3)]
COTP_CR_PACKET.PDUType = "CR"
COTP_CR_PACKET.Parameters[0].ParameterCode = "tpdu-size"
COTP_CR_PACKET.Parameters[0].Parameter = "\x0a"
COTP_CR_PACKET.Parameters[1].ParameterCode = "src-tsap"
COTP_CR_PACKET.Parameters[2].ParameterCode = "dst-tsap"
COTP_CR_PACKET.Parameters[1].Parameter = SRC_TSAP
COTP_CR_PACKET.Parameters[2].Parameter = DST_TSAP
# 因為是建立連接使用，因此fuzzable參數需要設置為False避免數據包被變異破壞。
COTP_CR_TEMPLATE = Template(name='cotp cr template', fields=[
ScapyField(COTP_CR_PACKET, name='cotp cr', fuzzable=False),
])
# 定義通訊參數配置數據結構
SETUP_COMM_PARAMETER_PACKET = TPKT() / COTPDT(EOT=1) / S7Header(ROSCTR="Job", Parameters=S7SetConParameter())
SETUP_COMM_PARAMETER_TEMPLATE = Template(name='setup comm template', fields=[
ScapyField(SETUP_COMM_PARAMETER_PACKET, name='setup comm', fuzzable=False),
])
# 定義需要Fuzzing的數據包結構, 下面例子中將使用RandShort對請求的SZLId及SZLIndex值進行變異測試。
READ_SZL_PACKET = TPKT() / COTPDT(EOT=1) / S7Header(ROSCTR="UserData", Parameters=S7ReadSZLParameterReq(),Data=S7ReadSZLDataReq(SZLId=RandShort(), SZLIndex=RandShort()))
# 定義READ_SZL_TEMPLATE為可以進行變異的結構，fuzzing的次數為1000次
READ_SZL_TEMPLATE = Template(name='read szl template', fields=[
ScapyField(READ_SZL_PACKET, name='read szl', fuzzable=True, fuzz_count=1000),
])
# 使用GraphModel進行Fuzz
model = GraphModel()
# 在使用GraphModel中注冊第一個節點, 首先發送COTP_CR請求。
model.connect(COTP_CR_TEMPLATE)
# 在使用GraphModel中注冊第二個節點, 在發送完COTP_CR后發送SETUP_COMM_PARAMETER請求。
model.connect(COTP_CR_TEMPLATE, SETUP_COMM_PARAMETER_TEMPLATE)
# 在使用GraphModel中注冊第三個節點, 在發送完SETUP_COMM_PARAMETER后發送READ_SZL請求。
model.connect(SETUP_COMM_PARAMETER_TEMPLATE, READ_SZL_TEMPLATE)
# define target
s7comm_target = TcpTarget(name='s7comm target', host=TARGET_IP, port=TARGET_PORT, timeout=2)
# 定義是需要等待Target返回響應，如果設置為True Target不返回數據包則會被識別成異常進行記錄。
s7comm_target.set_expect_response(True)
# 定義使用基礎的ServerFuzzer進行Fuzz測試
fuzzer = ServerFuzzer()
# 定義fuzzer使用的交互界面為web界面
fuzzer.set_interface(WebInterface(port=26001))
# 在fuzzer中定義使用GraphModel
fuzzer.set_model(model)
# 在fuzzer中定義target為s7comm_target
fuzzer.set_target(s7comm_target)
# 定義每個測試用例發送之間的延遲
fuzzer.set_delay_between_tests(0.1)
# 開始執行Fuzz
fuzzer.start()4. 小結
通過將Kitty與基于Scapy的數據包結構進行結合能夠基于一些現有的協議組件(例如ISF或Scapy中原生的協議)快速的構造一個高度自定義的Fuzzing測試用例，特別是在面對復雜的協議時可以減少大量的協議數據結構編寫等大量工作。
Kitty的可擴展性非常強，而且本文只涉及到了其中非常少的一部分功能，通過對Kitty進行擴展可以快速的針對特定目標構造對應的Fuzzing工具。
本文由 Seebug Paper 發布，如需轉載請注明來源。本文地址：http://www.bjnorthway.com/772/
暫無評論