mozak.tech Engenharia de IA Corporativa 16%

Parte I — Conectividade Padronizada entre Sistemas e Modelos

1.6 — Segurança e Controle em Servidores MCP (MCP Security)

Objetivo da Aula

Ao concluir esta aula, você será capaz de:

Implementar logging de auditoria em JSONL para cada tool call com timestamp e resultado

Construir rate limiting por token com janela de 60 segundos e bloqueio correto

Filtrar tools disponíveis com base nas permissões do token de autenticação

Aplicar validação de schema com Zod em inputs de tools MCP

Avaliar a superfície de ataque de um MCP server e as defesas correspondentes

Por que isso importa

Um MCP server sem segurança é uma porta aberta para sistemas internos. Se o server expõe operações de banco de dados, CRM ou sistema de arquivos, qualquer processo que consiga conectar pode executar essas operações sem restrição.

O vetor de ataque mais comum: um agente LLM em loop infinito (ou um bug que causa loop) que chama uma tool de custo alto (ex: search, email, AI) milhares de vezes até esgotar o orçamento. Sem rate limiting, um único agente mal configurado pode gerar custo de milhares de dólares em minutos.

O segundo vetor: escopo excessivo. Se um agente read-only (que só deveria buscar dados) tem acesso à tool deletar_dado, qualquer prompt injection que induz o LLM a chamar essa tool causa dano real. Principle of least privilege: cada token/agente acessa apenas o que precisa.

O terceiro vetor: falta de auditoria. Quando algo dá errado, você precisa saber: qual agente chamou qual tool, com quais argumentos, e qual foi o resultado. Sem logs, debugging é impossível.

Esta unidade te ensina as 3 defesas fundamentais: auditoria, rate limiting e controle de acesso por permission.

Conceitos Fundamentais

Modelo de Segurança: Token-Based Access Control

O modelo mais simples de auth para MCP servers: cada cliente apresenta um token no momento de conectar, e o server usa esse token para:

Identificar o cliente (autenticação)

Determinar quais tools ele pode usar (autorização)

const TOKENS: Record<string, { nome: string; tools_permitidas: string[] }> = {

  'token-readonly-123': {

    nome: 'agente-leitura',

    tools_permitidas: ['buscar_dado'],          // só leitura

  },

  'token-admin-456': {

    nome: 'agente-admin',

    tools_permitidas: ['buscar_dado', 'modificar_dado', 'deletar_dado'],  // tudo

  },

}

No MCP com transport stdio, o token é passado como argumento de linha de comando ou variável de ambiente. Com HTTP/SSE, é um header Authorization: Bearer <token>.

Rate Limiting em MCP

Rate limiting protege contra uso abusivo — intencional (DoS) ou acidental (loop de agente):

const rateLimits: Record<string, { contagem: number; resetEm: number }> = {}

const RATE_LIMIT = 30  // máximo de calls por minuto



function verificarRateLimit(token: string): boolean {

  const agora = Date.now()

  const entrada = rateLimits[token]



  // Janela nova ou primeira vez

  if (!entrada || agora >= entrada.resetEm) {

    rateLimits[token] = { contagem: 1, resetEm: agora + 60_000 }

    return true  // permitido

  }



  // Dentro da janela atual

  if (entrada.contagem >= RATE_LIMIT) {

    return false  // bloqueado

  }



  entrada.contagem++

  return true  // permitido

}

Como sinalizar bloqueio no MCP: quando a tool é bloqueada por rate limit, retorne um erro com isError: true:

if (!verificarRateLimit(token)) {

  return {

    content: [{ type: 'text', text: `Rate limit excedido para token '${TOKENS[token]?.nome}'. Tente novamente em alguns segundos.` }],

    isError: true,

  }

}

Logging de Auditoria em JSONL

Cada tool call deve ser logada para auditoria posterior:

import * as fs from 'fs'



function registrarAuditoria(

  token: string,

  tool: string,

  args: unknown,

  resultado: string

): void {

  const entrada = {

    timestamp: new Date().toISOString(),

    token,

    agente: TOKENS[token]?.nome ?? 'desconhecido',

    tool,

    args,

    resultado: resultado.slice(0, 200),  // trunca para não logar dados sensíveis longos

  }



  fs.appendFileSync(

    'audit.jsonl',

    JSON.stringify(entrada) + '\n',

    { encoding: 'utf8' }

  )

}

Por que JSONL? JSON Lines permite leitura linha por linha sem parsear o arquivo inteiro — eficiente para arquivos grandes. Compatível com jq, Pandas, e todas as ferramentas de log analysis.

Validação de Schema com Zod

Zod valida os argumentos antes de executar a lógica da tool:

import { z } from 'zod'



const BuscarDadoSchema = z.object({

  chave: z.string()

    .min(1, 'Chave não pode ser vazia')

    .max(100, 'Chave muito longa')

    .regex(/^[a-z0-9-]+$/, 'Chave deve conter apenas letras minúsculas, números e hífens'),

})



// No handler:

try {

  const { chave } = BuscarDadoSchema.parse(args)

  // ... lógica da tool

} catch (err) {

  if (err instanceof z.ZodError) {

    return {

      content: [{ type: 'text', text: `Input inválido: ${err.errors.map(e => e.message).join(', ')}` }],

      isError: true,

    }

  }

  throw err

}

Isso previne: - Injeção de caracteres especiais na chave - Chaves vazias que causariam comportamento inesperado - Chaves excessivamente longas que poderiam causar problemas em backends

Aprofundamento Técnico

Filtragem de Tools por Permissão

Decisão de Arquitetura: Menor Privilégio no Nível de Tool

O princípio de menor privilégio aplicado a MCP tem uma propriedade especial: se você não listar uma tool na resposta do tools/list, o modelo não sabe que ela existe e não pode chamá-la. Isso é defesa em profundidade contra prompt injection — mesmo que o modelo seja induzido a tentar chamar deletar_banco_de_dados, se o token do usuário atual não tem essa permissão, a tool nunca aparece na lista. Implemente: filtro de tools no handler de tools/list baseado no token/papel do usuário; validação também no tools/call (defesa dupla); rate limiting por token para prevenir abuso. Problema crítico não-resolvido pelo capítulo: propagação de identidade — a ação executada pela tool é em nome do usuário ou do serviço? O log de auditoria deve registrar ambos.

O handler ListToolsRequestSchema pode filtrar as tools retornadas com base no token:

server.setRequestHandler(ListToolsRequestSchema, async () => {

  // Token não registrado: retorna lista vazia (ou lança erro)

  if (!tokenCliente || !TOKENS[tokenCliente]) {

    return { tools: [] }  // cliente não pode ver nenhuma tool

  }



  const toolsPermitidas = TOKENS[tokenCliente].tools_permitidas

  const todasAsTools = [

    { name: 'buscar_dado', description: '...', inputSchema: { ... } },

    { name: 'modificar_dado', description: '...', inputSchema: { ... } },

    { name: 'deletar_dado', description: '...', inputSchema: { ... } },

  ]



  // Filtra: só retorna as tools que o token tem permissão

  return {

    tools: todasAsTools.filter(t => toolsPermitidas.includes(t.name))

  }

})

Isso garante que o LLM nem mesmo sabe que deletar_dado existe se o token não tem permissão — elimina o risco de prompt injection tentando chamar a tool.

Verificação de Permissão no CallTool

Além de filtrar no ListTools, verifique permissão no CallTool também — defesa em profundidade:

server.setRequestHandler(CallToolRequestSchema, async (req) => {

  const { name, arguments: args } = req.params



  // Verifica rate limit antes de qualquer processamento

  if (!verificarRateLimit(tokenCliente ?? '')) {

    registrarAuditoria(tokenCliente ?? '', name, args, 'RATE_LIMIT_EXCEDIDO')

    return {

      content: [{ type: 'text', text: 'Rate limit excedido' }],

      isError: true,

    }

  }



  // Verifica permissão para esta tool específica

  const toolsPermitidas = TOKENS[tokenCliente ?? '']?.tools_permitidas ?? []

  if (!toolsPermitidas.includes(name)) {

    registrarAuditoria(tokenCliente ?? '', name, args, 'PERMISSAO_NEGADA')

    return {

      content: [{ type: 'text', text: `Permissão negada para tool '${name}'` }],

      isError: true,

    }

  }



  // Processa a tool

  let resultado: string

  try {

    // ... lógica da tool ...

    resultado = 'sucesso'

  } catch (err) {

    resultado = `erro: ${err.message}`

    registrarAuditoria(tokenCliente ?? '', name, args, resultado)

    return { content: [{ type: 'text', text: resultado }], isError: true }

  }



  // Loga auditoria bem-sucedida

  registrarAuditoria(tokenCliente ?? '', name, args, resultado)

  return { content: [{ type: 'text', text: resultado }] }

})

Fechamento do tokenCliente no Closure

No starter, o tokenCliente é passado como parâmetro para criarServerSeguro(tokenCliente?). Dentro dos handlers, você acessa via closure:

function criarServerSeguro(tokenCliente?: string): Server {

  const server = new Server({ name: 'server-seguro', version: '1.0.0' }, { capabilities: { tools: {} } })



  // tokenCliente está disponível em todos os handlers por closure

  server.setRequestHandler(ListToolsRequestSchema, async () => {

    // usa tokenCliente do escopo externo

    const perms = TOKENS[tokenCliente ?? '']?.tools_permitidas ?? []

    // ...

  })



  server.setRequestHandler(CallToolRequestSchema, async (req) => {

    // usa tokenCliente do escopo externo

    if (!verificarRateLimit(tokenCliente ?? '')) { ... }

    // ...

  })



  return server

}

Em um server real com HTTP transport, o token viria do header da request. Com stdio, vem de variável de ambiente ou argumento de linha de comando.

Exemplos Anotados

Exemplo 1: Todos os 5 TODOs Implementados

import { z } from 'zod'

import * as fs from 'fs'

import { Server } from '@modelcontextprotocol/sdk/server/index.js'

import { CallToolRequestSchema, ListToolsRequestSchema } from '@modelcontextprotocol/sdk/types.js'



const TOKENS: Record<string, { nome: string; tools_permitidas: string[] }> = {

  'token-readonly-123': { nome: 'agente-leitura', tools_permitidas: ['buscar_dado'] },

  'token-admin-456': { nome: 'agente-admin', tools_permitidas: ['buscar_dado', 'modificar_dado', 'deletar_dado'] },

}



const rateLimits: Record<string, { contagem: number; resetEm: number }> = {}

const RATE_LIMIT = 30

const dados: Record<string, string> = { 'chave-1': 'valor-1', 'chave-2': 'valor-2' }



const BuscarSchema = z.object({

  chave: z.string().min(1).max(100).regex(/^[a-z0-9-]+$/),

})

const ModificarSchema = z.object({

  chave: z.string().min(1).max(100),

  valor: z.string().max(1000),

})



// TODO 1: Logging de auditoria

function registrarAuditoria(token: string, tool: string, args: unknown, resultado: string): void {

  const entrada = {

    timestamp: new Date().toISOString(),

    token,

    agente: TOKENS[token]?.nome ?? 'desconhecido',

    tool,

    args,

    resultado: resultado.slice(0, 200),

  }

  fs.appendFileSync('audit.jsonl', JSON.stringify(entrada) + '\n', { encoding: 'utf8' })

}



// TODO 2: Rate limiting

function verificarRateLimit(token: string): boolean {

  const agora = Date.now()

  const entrada = rateLimits[token]



  if (!entrada || agora >= entrada.resetEm) {

    rateLimits[token] = { contagem: 1, resetEm: agora + 60_000 }

    return true

  }



  if (entrada.contagem >= RATE_LIMIT) return false

  entrada.contagem++

  return true

}



function criarServerSeguro(tokenCliente?: string): Server {

  const server = new Server({ name: 'server-seguro', version: '1.0.0' }, { capabilities: { tools: {} } })



  // TODO 3: Filtra tools por permissão do token

  server.setRequestHandler(ListToolsRequestSchema, async () => {

    if (!tokenCliente || !TOKENS[tokenCliente]) {

      return { tools: [] }

    }



    const perms = TOKENS[tokenCliente].tools_permitidas

    const todasTools = [

      { name: 'buscar_dado', description: 'Busca valor por chave', inputSchema: { type: 'object', properties: { chave: { type: 'string' } }, required: ['chave'] } },

      { name: 'modificar_dado', description: 'Modifica valor de uma chave', inputSchema: { type: 'object', properties: { chave: { type: 'string' }, valor: { type: 'string' } }, required: ['chave', 'valor'] } },

      { name: 'deletar_dado', description: 'DESTRUTIVO: Remove uma chave permanentemente', inputSchema: { type: 'object', properties: { chave: { type: 'string' } }, required: ['chave'] } },

    ]



    return { tools: todasTools.filter(t => perms.includes(t.name)) }

  })



  server.setRequestHandler(CallToolRequestSchema, async (req) => {

    const { name, arguments: args } = req.params



    // TODO 4: Verifica rate limit antes de processar

    if (!verificarRateLimit(tokenCliente ?? 'anonimo')) {

      registrarAuditoria(tokenCliente ?? '', name, args, 'RATE_LIMIT_EXCEDIDO')

      return { content: [{ type: 'text', text: 'Rate limit excedido. Tente novamente em alguns segundos.' }], isError: true }

    }



    // TODO 5: Verifica permissão para modificar_dado

    if (name === 'modificar_dado') {

      const perms = TOKENS[tokenCliente ?? '']?.tools_permitidas ?? []

      if (!perms.includes('modificar_dado')) {

        registrarAuditoria(tokenCliente ?? '', name, args, 'PERMISSAO_NEGADA')

        return { content: [{ type: 'text', text: 'Permissão negada: seu token não permite modificar dados' }], isError: true }

      }

    }



    if (name === 'buscar_dado') {

      try {

        const { chave } = BuscarSchema.parse(args)

        const valor = dados[chave]

        const resultado = valor ? JSON.stringify({ chave, valor }) : `Chave '${chave}' não encontrada`

        registrarAuditoria(tokenCliente ?? '', name, args, resultado)

        return { content: [{ type: 'text', text: resultado }] }

      } catch (err) {

        return { content: [{ type: 'text', text: `Input inválido: ${err.message}` }], isError: true }

      }

    }



    if (name === 'modificar_dado') {

      try {

        const { chave, valor } = ModificarSchema.parse(args)

        dados[chave] = valor

        const resultado = JSON.stringify({ sucesso: true, chave, valor_novo: valor })

        registrarAuditoria(tokenCliente ?? '', name, args, resultado)

        return { content: [{ type: 'text', text: resultado }] }

      } catch (err) {

        return { content: [{ type: 'text', text: `Input inválido: ${err.message}` }], isError: true }

      }

    }



    throw new Error(`Tool desconhecida: ${name}`)

  })



  return server

}

Padrões e Armadilhas

Padrões

Padrão 1: Defesa em profundidade — verifique permissão em ListTools E CallTool ListTools filtra o que o LLM vê (principal). CallTool verifica novamente antes de executar (fallback). Nunca dependa de só uma camada.

Padrão 2: Log antes de retornar, não só em sucesso

// Loga tanto sucesso quanto erro

registrarAuditoria(token, tool, args, isError ? `ERRO: ${resultado}` : resultado)

Padrão 3: Zod parse em todos os args de tools críticas Tools que fazem mutações, deletam dados, ou chamam sistemas externos devem ter validação Zod. Tools de leitura simples podem usar cast direto.

Armadilhas

⚠️ Armadilha 1: Rate limit por nome de token, não por token real

// PROBLEMA: token inválido também incrementa o rate limit

rateLimits[tokenCliente ?? 'anonimo']



// MELHOR: só rate-limita tokens válidos (evita DoS por tokens inexistentes)

if (!TOKENS[tokenCliente ?? '']) {

  return { content: [{ type: 'text', text: 'Token inválido' }], isError: true }

}

// Depois do check, rate-limita por token válido

⚠️ Armadilha 2: fs.appendFileSync bloqueante em servidor de alto volume

// PROBLEMA: appendFileSync bloqueia o event loop

fs.appendFileSync('audit.jsonl', linha + '\n')



// MELHOR para alto volume: use stream assíncrono

const logStream = fs.createWriteStream('audit.jsonl', { flags: 'a' })

logStream.write(linha + '\n')  // non-blocking

⚠️ Armadilha 3: Logar dados sensíveis completos

// PERIGO: loga senha, token, CPF, etc.

registrarAuditoria(token, tool, args, resultado)  // args pode ter dados sensíveis



// SEGURO: projeta campos sensíveis antes de logar

const argsSeguros = { ...args, senha: '[REDACTED]', cpf: '[REDACTED]' }

registrarAuditoria(token, tool, argsSeguros, resultado)
⚗ Laboratório prático — mozak.tech
Se não for realizar o laboratório, pule para o próximo capítulo.

Ponte para o Lab

O starter tem 5 TODOs:

TODO 1 — Em registrarAuditoria(): implemente logging em audit.jsonl. Crie um objeto com timestamp (ISO string), token, tool, args, resultado. Use fs.appendFileSync('audit.jsonl', JSON.stringify(entrada) + '\n', { encoding: 'utf8' }). Seção de referência: Conceitos Fundamentais → “Logging de Auditoria em JSONL”.

TODO 2 — Em verificarRateLimit(): implemente fixed window rate limiting. Se não há entrada ou resetEm < Date.now(): crie nova entrada {contagem: 1, resetEm: Date.now() + 60_000} e retorne true. Se contagem >= RATE_LIMIT: retorne false. Senão: incremente contagem e retorne true. Seção de referência: Conceitos Fundamentais → “Rate Limiting em MCP”.

TODO 3 — No ListToolsRequestSchema handler: filtre as tools baseado em TOKENS[tokenCliente]?.tools_permitidas. Se o token não estiver em TOKENS, retorne { tools: [] }. Seção de referência: Aprofundamento Técnico → “Filtragem de Tools por Permissão”.

TODO 4 — No CallToolRequestSchema handler: adicione verificação de rate limit no início, antes de processar qualquer tool. Se !verificarRateLimit(tokenCliente): chame registrarAuditoria e retorne erro 429-equivalente. Seção de referência: Aprofundamento Técnico → “Verificação de Permissão no CallTool”.

TODO 5 — Para a tool modificar_dado: verifique se tokenCliente tem 'modificar_dado' na lista de tools_permitidas. Se não: registre e retorne erro de permissão. Seção de referência: Aprofundamento Técnico → “Verificação de Permissão no CallTool”.

Agora você está pronto para o lab.