Firmadyne `extractor.py`
===
---
###### tags: `firmadyne` `security`
本篇紀錄閱讀 `extractor.py` source code 筆記
---
**目錄**
[TOC]
# `/sources/extractor/extractor.py`
1. Source Code:
https://github.com/firmadyne/extractor/blob/481fe2850076743d06db052ee4934ffb9a312426/extractor.py
2. Explaining:
在 Line 731~732，定義了進入點為 `main()`
---
main() 中使用了 `argparse.ArgumentParser`
用法參考 https://docs.python.org/3/library/argparse.html
定義了參數 `input` `output` `-sql` `-nf` `-nk` `-np` `-b`
---
在官方 Usage 中是如此呼叫 `extractor.py`:
- `./sources/extractor/extractor.py -b Netgear -sql 127.0.0.1 -np -nk "WNAP320 Firmware Version 2.0.3.zip" images`
所以 `main()` 中 Line 726 行執行 `Extractor` 的 `__init__()` 的參數如下:
- ` extract = Extractor(indir='WNAP320 Firmware Version 2.0.3.zip', outdir='images', rootfs=True, kernel=False, numproc=False, server=127.0.0.1, brand='Netgear')`
---
`Extractor` `__init__()` 執行動作如下:
- ```python==39
self._input = os.path.abspath(indir)
```
self._input 將存著 indir 絕對路徑，此例子為 `${FIRMADYNE}/WNAP320 Firmware Version 2.0.3.zip`
- ```python==41
self.output_dir = os.path.abspath(outdir) if outdir else None
```
若 outdir 有設定，就會存著 outdir 的絕對路徑，否則存 None，此例子為 `${FIRMADYNE}/images`
- ```python==44
self.do_kernel = kernel
```
```python==47
self.do_rootfs = rootfs
```
```python==50
self.brand = brand
```
```python==53
self.database = server
```
將參數存到物件的 member 中
- ```python==56
self._pool = multiprocessing.Pool() if numproc else None
```
若 numproc 為 True，就會將物件 member _pool 設定為 a pool of worker processes，否則就是 None，此例子為 None
- ```python==59
self.visited = set()
```
```python==63
self._list = list()
```
將 visited 和 _list 分別初始化為 class set 和 class list
---
回到 `main()` 中 Line 729，執行了剛剛 construct 的 `Extractor` 的 member function `extract()`，執行動作如下:
- ```python==175
if os.path.isdir(self._input):
```
此例子中 self._input 是 `${FIRMADYNE}/WNAP320 Firmware Version 2.0.3.zip`
他是一個檔案，不是一個目錄， os.path.isdir 回傳 false，不會進入 if
- ```python==179
elif os.path.isfile(self._input):
```
進入這個 elif，執行
- ```python==180
self._list.append(self._input)
```
- ```python==184
if self.output_dir and not os.path.isdir(self.output_dir):
```
此例子中，self.output_dir 有設定，但它是一個目錄，os.path.isdir 回傳 True，不會進入 if
- ```python==187
if self._pool:
```
此例子中，_pool 為 None，進入 else 區域執行
- ```python==190
for item in self._list:
self._extract_item(item)
```
此例子中， _list 只有一個 object，也就是 _input，會執行
`self._extract_item('${FIRMADYNE}/WNAP320 Firmware Version 2.0.3.zip')`
---
`Extractor` `_extract_item()` 執行 `ExtractionItem(self, path, 0).extract()`
作用只是傳遞 `Extractor` object 本身跟 path 給 `ExtractionItem` `__init__()` 初始化，並且執行 `ExtractionItem` `extract()`
此例中 `ExtractionItem` `__init__()` 的實際參數如下:
`__init__(self, extractor=此Extractor物件, path='${FIRMADYNE}/WNAP320 Firmware Version 2.0.3.zip', depth=0, tag=None)`
---
`ExtractionItem` `__init__()` 進行以下動作:
- ```python==212
self.temp = None
```
```python==215
self.depth = depth
```
```python==218
self.extractor = extractor
```
```python==221
self.item = path
```
一些 member 的 initialization
此例子中，這些參數會是
```
.temp=None
.depth=0
.extractor=剛剛的 Extractor 物件
.item='${FIRMADYNE}/WNAP320 Firmware Version 2.0.3.zip'
```
- ```python==224
if self.extractor.database:
```
此例子中， extractor.database 是 `127.0.0.1`，進入 if
連接至 database，並存放在 member self.database 中
- ```python==234
self.checksum = Extractor.io_md5(path)
```
呼叫 static method `Extractor.io_md5()`，並將結果存回 member checksum，`Extractor.io_md5()`可自行去看，不是重點
- ```python==237
self.tag = tag if tag else self.generate_tag()
```
此例子中，tag 是 None，會執行 member function `generate_tag()` 並存回 member tag 中，`generate_tag()` 也可自行去看，簡單來說，那 method 會執行以下事情:
- 查詢 table brand 是否已有此 brand，沒有就 insert 一個，並存回 brand_id
- 查詢 table image 是否有同 checksum 的 image，存回 image_id，沒有就 insert 一個 image
- ```python==303
return str(image_id[0]) if \
image_id else os.path.basename(self.item) + "_" + self.checksum
```
此例子中 image_id 有值，所以會回傳 str(image_id[0])
結果就是此 firmware 在 table image 的 id，此例子為 `"1"`
- ```python==240
self.output = os.path.join(self.extractor.output_dir, self.tag) if \
self.extractor.output_dir else None
```
此例子中， member output 會是 `${FIRMADYNE}/images/1`
- ```python==244
self.terminate = False
self.status = None
self.update_status()
```
設定 member terminate, status，並執行 member function `update_status()`，此 function 做以下事情:
- ```python==322
kernel_done = os.path.isfile(self.get_kernel_path()) if \
self.extractor.do_kernel and self.output else \
not self.extractor.do_kernel
```
此例子中，extractor.do_kernel 是 False，所以存回 `not self.extractor.do_kernel`，也就是將 True 存回 kernel_done
- ```python==325
rootfs_done = os.path.isfile(self.get_rootfs_path()) if \
self.extractor.do_rootfs and self.output else \
not self.extractor.do_rootfs
```
此例子中，extractor.do_rootfs 是 True，且 sele.out 是 `${FIRMADYNE}/images/1`，所以會將 `os.path.isfile(self.get_rootfs_path())` 結果存回 rootfs_done，簡單來說，會去查看有沒有 `${FIRMADYNE}/images/1.tar.gz` 這個檔案，目前是沒有，所以會存 false 到 rootfs_done
- ```python==328
self.status = (kernel_done, rootfs_done)
```
建一個 member tuple status，最終存 `(True, False)`
- ```python==330
if self.database and kernel_done and self.extractor.do_kernel:
```
此例中 `extractor.do_kernel` 為 False，不會進 if
- ```python==333
if self.database and rootfs_done and self.extractor.do_rootfs:
```
此例中 `rootfs_done` 為 False，不會進 if
- ```python==336
return self.get_status()
```
`get_status()` 執行 `return True if self.terminate or all(i for i in self.status) else False`
此例目前 self.terminate 為 False，且 tuple self.status中不是全部元素都是 True，所以回傳 False
---
執行完 `ExtractionItem` `__init__()` 回到 `ExtractionItem` `extract()` 進行以下動作:
- ```python==397
with Extractor.visited_lock:
```
參考 https://docs.python.org/2/library/threading.html#using-locks-conditions-and-semaphores-in-the-with-statement
用 `with` 可以自動為此 block 開關鎖
重點在此
- ```python==420
for analysis in [self._check_archive, self._check_firmware,
self._check_kernel, self._check_rootfs,
self._check_compressed]:
```
```python==424
os.chdir(self.temp)
```
```python==427
if analysis():
```
將當前目錄切換到 temp，注意等等會有一個 binwalk 的參數 `module.extractor.directory` 就會是 temp
此迴圈依序執行 member function `_check_archive()`, `_check_firmware()`, `_check_kernel()`, `_check_rootfs()`, `_check_compressed`
## `_check_archive()`
```python==474
return self._check_recursive("archive")
```
在 `_check_recursive(self, fmt=archive)` 中:
- ```python==638
for module in binwalk.scan(self.item, "-e", "-r", "-y", fmt,
signature=True, quiet=True):
```
此例中，如同下了指令:
- `binwalk --signature --quiet -e -r -y archive '${FIRMADYNE}/WNAP320 Firmware Version 2.0.3.zip'`
- ```python==640
for entry in module.results:
```
```python==647
desc = entry.description
self.printf(">>>> %s" % entry.description)
break
```
將第一個 entry 的 description 存進 desc 後就跳出
- ```python==651
if module.extractor.directory:
```
`module.extractor.directory` 為 member temp 的值，進入 if
- ```python==652
unix = Extractor.io_find_rootfs(module.extractor.directory)
```
static method `Extractor.io_find_rootfs` 簡單來說，會去尋找 filesystem，會做的事情大概如下:
- 如果當前目錄底下，只有一個子目錄，就把當前目錄改成那個子目錄
所以走到最後當前目錄底下不會只有一個子目錄
子目錄要嘛 0 個，要嘛很多個
- 看看目前子目錄中，名字有在 UNIX_DIRS 裡的目錄有幾個
並存到 count 中
- count 若大於 UNIX_THRESHOLD，就回傳 (True, 當前目錄位置)，表示成功找到 filesystem
- 否則就回傳 (False, 最初的目錄位置)
- 中間有略過一部分，但其實不是重點
重點是這個 function 是拿來找 filesystem
此例中是沒找到 filesystem 的，所以變數 unix 會是 `(False, self.temp)`
- ```python==655
if unix[0]:
```
不會進入，改進 Line 663 的 else
- ```python==666
for root, _, files in os.walk(module.extractor.directory):
```
Line 666 ~ 702 會走過所有在 module.extractor.directory 底下的目錄，請參考 os.walk()
- ```python==674
if desc and "original file name:" in desc:
```
此例子是不會有 `"original file name:"` 字串，故不會進入 if
- ```python==683
for filename in files:
```
Line 683 ~ 702 會 iterate 過目前目錄 `_` (定義在 Line 666) 底下所有的 filename。
- ```python==684
if count > ExtractionItem.RECURSION_BREADTH:
```
一開始不會進入此 if，改進 Line 689 的 else。
- ```python==690
new_item = ExtractionItem(self.extractor,
os.path.join(root,
filename),
self.depth + 1,
self.tag)
if new_item.extract():
```
以此 file 為 root path 遞迴嘗試 Extraction。
## `_check_firmware()`
```python=481
for module in binwalk.scan(self.item, "-y", "header", signature=True,
quiet=True):
```
此例中，如同下了指令:
- `binwalk --signature --quiet -y header '${FIRMADYNE}/WNAP320 Firmware Version 2.0.3.zip'`
```python=483
for entry in module.results:
```
Loop 每一條結果
```python=485
if "uImage header" in entry.description:
```
若在 description 欄位中有 match pattern `uImage header`
```python=486
if not self.get_kernel_status() and \
"OS Kernel Image" in entry.description:
```
`get_kernel_status` 只是用來回傳 member status[0] 的內容，此例子為 True
所以若 pattern `OS Kernel Image` 也有在 description 中，就會進入 if，如果真的進了:
- ```python=488
kernel_offset = entry.offset + 64
kernel_size = 0
for stmt in entry.description.split(','):
if "image size:" in stmt:
kernel_size = int(''.join(
i for i in stmt if i.isdigit()), 10)
if kernel_size != 0 and kernel_offset + kernel_size \
<= os.path.getsize(self.item):
self.printf(">>>> %s" % entry.description)
tmp_fd, tmp_path = tempfile.mkstemp(dir=self.temp)
os.close(tmp_fd)
Extractor.io_dd(self.item, kernel_offset,
kernel_size, tmp_path)
kernel = ExtractionItem(self.extractor, tmp_path,
self.depth, self.tag)
return kernel.extract()
```
Line 491 ~ 494 抓取 kernel_size 的資訊
Line 502 執行 `Extractor` static method `io_dd`
`io_dd` 主要做的事情就是把 `self.item` 從 `kernel_offset` 開始抓 `kernel_size` bytes 存到 `tmp_path`
Line 504 ~ 507 將 `tmp_path` 設為 root path 再 extract 一次，並且 return
```python=515
elif not self.get_kernel_status() and \
not self.get_rootfs_status() and \
"rootfs offset: " in entry.description and \
"kernel offset: " in entry.description:
```
若剛剛沒找到 pattern `OS Kernel Image`
那就改找 pattern `rootfs offset: ` 和 `kernel offset: `
並且 member status[0]、[1] 都要是 False
但此例中 `get_kernel_status` 回傳 status[0]，也就是 True，所以不會進此 elif。
若進了此 elif:
- ```python=519
kernel_offset = 0
kernel_size = 0
rootfs_offset = 0
rootfs_size = 0
```
初始一些參數
- ```python=524
for stmt in entry.description.split(','):
if "kernel offset:" in stmt:
kernel_offset = int(stmt.split(':')[1], 16)
elif "kernel length:" in stmt:
kernel_size = int(stmt.split(':')[1], 16)
elif "rootfs offset:" in stmt:
rootfs_offset = int(stmt.split(':')[1], 16)
elif "rootfs length:" in stmt:
rootfs_size = int(stmt.split(':')[1], 16)
```
- ```python=535
if kernel_offset != rootfs_size and kernel_size == 0 and \
rootfs_size == 0:
kernel_size = rootfs_offset - kernel_offset
rootfs_size = os.path.getsize(self.item) - rootfs_offset
```
抓取、設定參數 `kernel_offset` `kernel_size` `rootfs_offset` `rootfs_size`
- ```python=541
if (kernel_size > 0 and kernel_offset + kernel_size \
<= os.path.getsize(self.item)) and \
(rootfs_size != 0 and rootfs_offset + rootfs_size \
<= os.path.getsize(self.item)):
self.printf(">>>> %s" % entry.description)
tmp_fd, tmp_path = tempfile.mkstemp(dir=self.temp)
os.close(tmp_fd)
Extractor.io_dd(self.item, kernel_offset, kernel_size,
tmp_path)
kernel = ExtractionItem(self.extractor, tmp_path,
self.depth, self.tag)
kernel.extract()
tmp_fd, tmp_path = tempfile.mkstemp(dir=self.temp)
os.close(tmp_fd)
Extractor.io_dd(self.item, rootfs_offset, rootfs_size,
tmp_path)
rootfs = ExtractionItem(self.extractor, tmp_path,
self.depth, self.tag)
rootfs.extract()
return self.update_status()
```
確定 size + offset 在檔案整個 size 範圍內，才做以下事情
- 將 kernel dd 出來，並且當作 root path 做 extraction
- 將 rootfs dd 出來，並且當作 root path 做 extraction
- update status
```python=564
return False
```
如果每個 entry 都不會進上面兩個條件式其中之一，`_check_firmware` 即回傳 False。
## `_check_kernel()`
```python=571
if not self.get_kernel_status():
for module in binwalk.scan(self.item, "-y", "kernel",
signature=True, quiet=True):
for entry in module.results:
if "kernel version" in entry.description:
self.update_database("kernel_version",
entry.description)
if "Linux" in entry.description:
if self.get_kernel_path():
shutil.copy(self.item, self.get_kernel_path())
else:
self.extractor.do_kernel = False
self.printf(">>>> %s" % entry.description)
return True
# VxWorks, etc
else:
self.printf(">>>> Ignoring: %s" % entry.description)
return False
return False
return False
```
Line 571 若 member status[0] 為 False，表示還未抓到 kernel，進入 if 嘗試抓 kernel
Line 572 此例中，如同下了指令:
- `binwalk --signature --quiet -y kernel '${FIRMADYNE}/WNAP320 Firmware Version 2.0.3.zip'`
Line 575 若搜尋到 pattern `kernel version` 做以下事情:
- 設定 database 中，此 iid 的 kernel_version 欄位等於整個 entry.description
- 若搜尋到 pattern `Linux`
- 若有設定 output 位置，member function `get_kernel_path` 回傳要輸出 kernel 的位置。
- 若有 output 位置，則複製原檔案到該輸出的位置
- 沒有設定 output 位置的話，設定 Extractor 的 do_kernel 為 False
- 回傳 True
- 沒搜到 `Linux` 就回傳 False
沒搜到 pattern 就回傳 False
## `_check_rootfs()`
```python=597
if not self.get_rootfs_status():
for module in binwalk.scan(self.item, "-e", "-r", "-y",
"filesystem", signature=True,
quiet=True):
for entry in module.results:
self.printf(">>>> %s" % entry.description)
break
if module.extractor.directory:
unix = Extractor.io_find_rootfs(module.extractor.directory)
if not unix[0]:
self.printf(">>>> Extraction failed!")
return False
self.printf(">>>> Found Linux filesystem in %s!" % unix[1])
if self.output:
shutil.make_archive(self.output, "gztar",
root_dir=unix[1])
else:
self.extractor.do_rootfs = False
return True
return False
```
前面有解釋 `io_find_rootfs` 會做什麼
簡單來說，若用 binwalk 發現含有 `filesystem` 的 entry
將這個路徑丟到 `io_find_rootfs` 找 filesystem
沒找到 fs 或根本沒有這個 entry 都會 return False
其他狀況 return True
Line 614 若有設定 output 路徑，將剛剛 `io_find_rootfs` 抓出來的 fs 路徑輸出到該輸出的位置，這個就是主要產生 `iid.tar.gz` 的 code
沒設定 output 路徑的話，設定 do_rootfs 為 False
## `_check_compressed`
```python=626
return self._check_recursive("compressed")
```
前面有解釋 `_check_recursive` 會做什麼
這裡只是把 binwalk -y 的參數設定為 `compressed`
接下來做的事情的流程跟前面解釋的一樣
Sign in with Wallet
Connect another wallet