Verifique uploads web com Magika e uma API Python

Tim Koschützki

Cofundador · Berlim, Alemanha ·

Um arquivo chamado photo.jpg pode conter um PDF. Para verificar o tipo de um upload, envie os bytes dele para a API Python do Magika e aplique uma lista de permissões ao rótulo retornado. Este passo a passo cria uma aplicação Flask local que serve tanto a página de upload quanto o endpoint de classificação, com um feedback que acompanha a seleção atual.

O navegador envia o arquivo completo para o Python. A classificação acontece nesse servidor, mesmo quando os dois processos rodam no seu notebook. Um tipo aceito não comprova que o arquivo é válido nem que está livre de malware.

Entenda o que o rótulo pode dizer

Extensões e tipos MIME informados pelo cliente descrevem o que um arquivo afirma ser. Detectores de conteúdo inspecionam os bytes, então renomear um arquivo não altera a entrada deles. O Magika usa um modelo treinado para distinguir formatos binários e de texto; mesmo assim, ele pode classificar um arquivo incorretamente.

A API Python expõe identify_bytes(), um indicador de sucesso em result.ok e a previsão final em result.output. Use result.output.label para a lista de permissões. O modo padrão HIGH_CONFIDENCE pode substituir uma previsão do modelo com baixa confiança por um rótulo genérico, que este exemplo rejeita junto com outros tipos fora da lista de permissões.

O Magika e os detectores baseados em assinaturas respondem a uma pergunta de identificação de formato. Nem essa resposta nem uma pontuação de confiança alta comprovam que um decodificador aceitará o arquivo inteiro. As limitações conhecidas do Magika também excluem a detecção confiável de poliglotas: um arquivo pode ser interpretável como mais de um formato.

Integre o Magika a uma aplicação de navegador

Use Linux, Python 3.12 com venv e pip, e um navegador atual. O exemplo fixa o Magika 0.6.1 e o Flask 3.1.0; a instalação precisa de acesso à internet. Depois de instalado, o Magika carrega localmente o modelo incluído no pacote, então a classificação não depende de um serviço em nuvem. Os comandos abaixo usam Bash.

1. Instale o Magika

Cole isto em um terminal, em um diretório onde você quer criar um novo projeto magika-upload. O script se recusa a reutilizar um diretório existente e para se qualquer etapa de configuração falhar. Os parênteses mantêm o seu shell no diretório original.

(
  mkdir magika-upload &&
  cd magika-upload &&
  mkdir static &&
  python3.12 -m venv .venv &&
  .venv/bin/python -m pip install magika==0.6.1 flask==3.1.0
)

Só continue depois que a instalação for concluída com sucesso. Se ela falhar, analise o erro e tente de novo em um diretório novo ou repare o ambiente incompleto antes de prosseguir. Não apague um projeto existente para executar a configuração novamente.

2. Crie uma API mínima de verificação (Flask)

Salve isto como magika-upload/app.py. O modelo é carregado uma única vez, quando a aplicação inicia. A rota / serve a página que você vai criar em seguida; o Flask também serve o diretório static automaticamente.

from flask import Flask, jsonify, request, send_from_directory
from magika import Magika
from werkzeug.exceptions import RequestEntityTooLarge

app = Flask(__name__)
MAX_FILE_BYTES = 5 * 1024 * 1024
app.config['MAX_CONTENT_LENGTH'] = 6 * 1024 * 1024
ALLOWED_TYPES = {'pdf', 'jpeg', 'png'}
magika = Magika()


@app.get('/')
def index():
    return send_from_directory(app.root_path, 'index.html')


@app.errorhandler(RequestEntityTooLarge)
def request_too_large(error):
    return jsonify(error='Upload is too large'), 413


@app.post('/verify')
def verify_file():
    if 'file' not in request.files:
        return jsonify(error='No file provided'), 400

    content = request.files['file'].read(MAX_FILE_BYTES + 1)
    if not content:
        return jsonify(error='File is empty'), 400
    if len(content) > MAX_FILE_BYTES:
        return jsonify(error='Upload is too large'), 413

    result = magika.identify_bytes(content)
    if not result.ok:
        return jsonify(error='File analysis failed'), 500
    if result.output.label not in ALLOWED_TYPES:
        return jsonify(error='Only PDF, JPEG, and PNG files are allowed'), 415

    return jsonify(
        file_type=result.output.label,
        mime_type=result.output.mime_type,
    )

O limite por arquivo é de 5 MiB. O limite separado de 6 MiB por requisição deixa espaço para os cabeçalhos multipart e, ao mesmo tempo, limita a requisição inteira. Como explica o guia de upload do Flask, o processamento de uploads pode usar armazenamento temporário em disco. Esta aplicação não salva arquivos aceitos de forma permanente; ela lê no máximo 5 MiB mais um byte para content antes da classificação.

3. Conecte o front-end

Salve isto como magika-upload/index.html:

<!doctype html>
<html lang="en">
  <head>
    <meta charset="utf-8">
    <meta name="viewport" content="width=device-width, initial-scale=1">
    <title>Classify a file with Magika</title>
    <script src="/static/verify.js" defer></script>
  </head>
  <body>
    <h1>Classify a file with Magika</h1>
    <p>PDF, JPEG, or PNG, up to 5 MiB. The file is sent to the local Python server.</p>
    <form id="uploadForm" action="/verify" method="post" enctype="multipart/form-data">
      <label for="fileInput">File to classify</label>
      <input id="fileInput" name="file" type="file">
      <button type="submit">Classify file</button>
    </form>
    <p id="result" role="status">Choose a file, then classify it.</p>
  </body>
</html>

Salve o código a seguir como magika-upload/static/verify.js. Alterar a seleção limpa o resultado anterior imediatamente. Enviar de novo substitui a requisição anterior, então uma resposta lenta não consegue sobrescrever o feedback de um trabalho mais recente.

const form = document.getElementById('uploadForm')
const input = document.getElementById('fileInput')
const result = document.getElementById('result')
const errors = {
  400: 'Choose a non-empty file.',
  413: 'File exceeds the upload limit.',
  415: 'Only PDF, JPEG, and PNG files are allowed.',
}
let requestId = 0
let activeRequest

input.addEventListener('change', () => {
  requestId += 1
  activeRequest?.abort()
  result.textContent = 'Selection changed. Click Classify file to check it.'
})

form.addEventListener('submit', async (event) => {
  event.preventDefault()
  const id = ++requestId
  activeRequest?.abort()
  const file = input.files[0]
  if (!file) {
    result.textContent = 'Choose a file, then classify it.'
    return
  }

  const controller = new AbortController()
  activeRequest = controller
  const timeout = setTimeout(() => controller.abort(), 30_000)
  const data = new FormData()
  data.append('file', file)
  result.textContent = `Classifying ${file.name}…`

  try {
    const response = await fetch('/verify', {
      method: 'POST',
      body: data,
      signal: controller.signal,
    })
    if (id !== requestId) return
    if (!response.ok) {
      result.textContent = errors[response.status] ??
        `Classification failed (HTTP ${response.status}).`
      return
    }

    const analysis = await response.json()
    if (id !== requestId) return
    result.textContent = `${file.name}: ${analysis.file_type} (${analysis.mime_type}). ` +
      'Accepted by the type allowlist.'
  } catch {
    if (id !== requestId) return
    result.textContent = controller.signal.aborted
      ? 'Request timed out. Click Classify file to retry.'
      : 'Could not reach or read the server. Check the terminal, then retry.'
  } finally {
    clearTimeout(timeout)
  }
})

O ID da requisição protege tanto a resposta quanto o corpo JSON dela, lido de forma assíncrona. AbortController para de aguardar requisições substituídas, e um temporizador de 30 segundos limita cada tentativa do navegador. Abortar a requisição do navegador não garante que o Python pare de processar os bytes que já recebeu.

A partir do mesmo diretório pai usado na configuração, inicie a aplicação:

(
  cd magika-upload &&
  .venv/bin/python -m flask --app app run --host 127.0.0.1 --port 5050
)

Aguarde a mensagem de inicialização do Flask e abra http://127.0.0.1:5050/. Abra essa URL em vez do arquivo HTML no disco: a página, o script e o endpoint /verify precisam compartilhar a mesma origem. Selecione um arquivo em File to classify e clique em Classify file. Você pode clicar no botão de novo para tentar o mesmo arquivo outra vez. Quando terminar, pare o servidor com Ctrl+C.

4. Compare os resultados com uma lista de permissões

Teste um PNG pequeno e válido renomeado para photo.txt. Você deve ver photo.txt: png (image/png). Accepted by the type allowlist. O nome do arquivo não autoriza o upload: somente os rótulos pdf, jpeg e png do servidor fazem isso. Para arquivos JPEG, o rótulo é jpeg, mesmo quando o nome do arquivo termina em .jpg.

Em seguida, teste um arquivo de texto não vazio renomeado para photo.png. Ele deve produzir Only PDF, JPEG, and PNG files are allowed. Teste também um arquivo vazio e um arquivo com pouco mais de 5 MiB. Eles são rejeitados antes de o Magika ser executado. Enquanto uma requisição estiver pendente, selecione outro arquivo: a página deve limpar o resultado pendente e aguardar até que você classifique a nova seleção.

Resolva uma requisição com falha

A instalação ou o carregamento do modelo falha: leia primeiro o erro no terminal. Confirme que a instalação foi concluída em .venv e que você está iniciando o executável Python desse ambiente. A página só carrega depois que a aplicação inicia com sucesso.
A porta está ocupada: o Flask encerra com um erro de endereço em uso. Escolha outra porta no comando de inicialização e abra a URL dessa porta. Este é um servidor de desenvolvimento local, não uma configuração de implantação.
O servidor rejeita a requisição: HTTP 400 indica um arquivo ausente ou vazio, 413 indica um limite de tamanho e 415 indica um rótulo fora da lista de permissões. Requisições maiores podem causar uma redefinição de conexão no servidor de desenvolvimento em vez de uma resposta 413 legível.
O navegador não consegue ler uma resposta: verifique o terminal do servidor e a URL e tente de novo. Falhas HTTP, um servidor parado e tempos limite esgotados não contam como aceitação. A página exibe mensagens de erro fixas em vez de renderizar uma página de erro do servidor ou um stack trace.

Use o resultado em um fluxo de trabalho de upload

Se você estender esta aplicação para armazenar arquivos, faça a verificação da lista de permissões e o armazenamento sobre os mesmos bytes recebidos. Uma resposta de classificação bem-sucedida não deve virar permissão para um upload posterior e não verificado. O exemplo retorna apenas uma classificação; ele não cria nenhum upload salvo nem uma aprovação reutilizável.

Um rótulo permitido pode ajudar a escolher um decodificador de imagens, um parser de documentos ou um serviço de moderação na etapa seguinte. Essas operações ainda precisam de validação e limites de recursos próprios. A varredura de malware é uma tarefa separada, e uma previsão de formato nunca deve ser exibida como um veredito de segurança.

Para um fluxo de trabalho de upload gerenciado, consulte a referência de verificação de arquivos. Mantenha a mesma distinção em qualquer um dos designs: reconhecer um formato é uma decisão no tratamento de um upload, não uma prova de que o conteúdo dele é seguro.