SSH honeypot servidor Python segurança cibernética terminal hacker invasão monitoramento

SSH Honeypot em Python Puro: A Armadilha que Grava Cada Comando do Invasor e Transforma Tentativas de Invasão em Inteligência de Segurança

PorOlivetto

Imagine receber um alerta às 3 da manhã: alguém está tentando invadir seu servidor SSH. Seu instinto diz para bloquear o IP e voltar para a cama. Mas e se você pudesse fazer algo melhor? E se, em vez de simplesmente bloquear, você pudesse convidar o invasor para entrar, gravar cada comando que ele digita, descobrir quais ferramentas ele usa, e transformar essa tentativa de invasão em inteligência valiosa?

Isso é um SSH honeypot. E hoje vamos construir um completo, do zero, em Python puro.

O Que é um SSH Honeypot e Por Que Você Precisa de Um

Um honeypot é um sistema deliberadamente vulnerável, projetado para atrair atacantes. Não é um backup, não é um servidor de produção com segurança fraca — é uma armadilha. Um SSH honeypot simula um servidor SSH legítimo, aceita conexões, permite login (geralmente com credenciais fracas ou padrão), e então grava tudo que o invasor faz dentro do sistema falso.

Por que fazer isso? Três razões:

  1. Inteligência de ameaças: Você descobre quais exploits estão sendo testados, quais ferramentas os atacantes usam, quais comandos eles executam primeiro.
  2. Distração: Enquanto o invasor perde tempo explorando o honeypot, ele não está atacando seus sistemas reais.
  3. Detecção precoce: Se alguém está escaneando sua rede e encontra o honeypot, você sabe que um ataque está começando antes dele atingir sistemas críticos.
terminal hacker código colorido segurança invasão
Um terminal SSH falso que parece real é a isca perfeita

A Arquitetura do Nosso SSH Honeypot

Vamos construir um honeypot de baixa interatividade. Isso significa que ele não executa comandos reais — apenas simula respostas plausíveis. É mais seguro que um honeypot de alta interatividade (que executa comandos em um ambiente isolado), e muito mais simples de implementar.

Nossa arquitetura tem quatro componentes:

  1. Servidor SSH: Aceita conexões, autentica usuários (com qualquer senha)
  2. Shell falso: Simula um terminal bash com respostas pré-programadas
  3. Logger: Grava cada comando, timestamp, IP de origem
  4. Analyzer: Processa os logs e gera relatórios de inteligência

Implementando o Servidor SSH com Paramiko

Vamos usar a biblioteca paramiko para criar o servidor SSH. Sim, é uma dependência externa, mas é a forma mais robusta de implementar SSH em Python. Se você quiser zero dependências, teria que implementar o protocolo SSH do zero — e isso é um projeto de meses.

import paramiko
import socket
import threading
import json
from datetime import datetime

# Gerar uma chave RSA para o servidor
# Em produção, use uma chave real, não esta
key = paramiko.RSAKey.generate(2048)

class SSHHoneypot(paramiko.ServerInterface):
    """Servidor SSH que aceita qualquer credencial"""
    
    def __init__(self, client_ip, client_port):
        self.client_ip = client_ip
        self.client_port = client_port
        self.username = None
        
    def check_auth_password(self, username, password):
        """Aceita QUALQUER combinação de usuário/senha"""
        self.username = username
        
        # Log da tentativa de autenticação
        log_auth_attempt(
            ip=self.client_ip,
            port=self.client_port,
            username=username,
            password=password,
            timestamp=datetime.now().isoformat()
        )
        
        # Sempre aceita - queremos ver o que ele faz depois
        return paramiko.AUTH_SUCCESSFUL
    
    def check_channel_request(self, kind, chanid):
        if kind == 'session':
            return paramiko.OPEN_SUCCEEDED
        return paramiko.OPEN_FAILED_ADMINISTRATIVELY_PROHIBITED
    
    def check_channel_shell_request(self, channel):
        return True
    
    def get_allowed_auths(self, username):
        return 'password'

def log_auth_attempt(ip, port, username, password, timestamp):
    """Registra tentativas de autenticação"""
    log_entry = {
        'type': 'auth',
        'timestamp': timestamp,
        'ip': ip,
        'port': port,
        'username': username,
        'password': password
    }
    
    with open('/var/log/ssh-honeypot/auth.log', 'a') as f:
        f.write(json.dumps(log_entry) + '\n')

def handle_client(client_socket, client_addr):
    """Manipula uma conexão SSH"""
    client_ip, client_port = client_addr
    
    try:
        transport = paramiko.Transport(client_socket)
        transport.add_server_key(key)
        
        server = SSHHoneypot(client_ip, client_port)
        transport.start_server(server=server)
        
        channel = transport.accept(20)
        if channel is None:
            return
        
        # Inicia o shell falso
        fake_shell(channel, client_ip, server.username)
        
    except Exception as e:
        print(f"Erro na conexão {client_ip}: {e}")
    finally:
        try:
            transport.close()
        except:
            pass

def start_honeypot(host='0.0.0.0', port=2222):
    """Inicia o servidor SSH honeypot"""
    server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    server_socket.bind((host, port))
    server_socket.listen(100)
    
    print(f"SSH Honeypot rodando em {host}:{port}")
    
    while True:
        client_socket, client_addr = server_socket.accept()
        print(f"Conexão de {client_addr[0]}:{client_addr[1]}")
        
        client_thread = threading.Thread(
            target=handle_client,
            args=(client_socket, client_addr)
        )
        client_thread.start()

if __name__ == '__main__':
    start_honeypot()

O Shell Falso: Simulando um Terminal Real

Agora vem a parte divertida: criar um shell que parece real, mas não executa nada. O segredo é ter respostas plausíveis para os comandos mais comuns.

import re

# Comandos simulados com respostas plausíveis
FAKE_RESPONSES = {
    r'^ls$': 'bin  boot  dev  etc  home  lib  media  mnt  opt  proc  root  run  sbin  srv  sys  tmp  usr  var',
    r'^ls -la$': '''total 84
drwxr-xr-x  18 root root  4096 Jul  5 14:23 .
drwxr-xr-x  18 root root  4096 Jul  5 14:23 ..
lrwxrwxrwx   1 root root     7 Jun 12  2024 bin -> usr/bin
drwxr-xr-x   2 root root  4096 Jun 12  2024 boot
drwxr-xr-x   5 root root   360 Jul  5 14:23 dev
drwxr-xr-x  92 root root  4096 Jul  5 14:23 etc
drwxr-xr-x   3 root root  4096 Jun 15  2024 home
lrwxrwxrwx   1 root root     7 Jun 12  2024 lib -> usr/lib''',
    
    r'^pwd$': '/root',
    r'^whoami$': 'root',
    r'^id$': 'uid=0(root) gid=0(root) groups=0(root)',
    r'^uname -a$': 'Linux ubuntu-server 5.15.0-91-generic #101-Ubuntu SMP Tue Nov 14 13:30:08 UTC 2023 x86_64 x86_64 x86_64 GNU/Linux',
    r'^cat /etc/passwd$': '''root:x:0:0:root:/root:/bin/bash
daemon:x:1:1:daemon:/usr/sbin:/usr/sbin/nologin
bin:x:2:2:bin:/bin:/usr/sbin/nologin
sys:x:3:3:sys:/dev:/usr/sbin/nologin
www-data:x:33:33:www-data:/var/www:/usr/sbin/nologin''',
    
    r'^cat /etc/shadow$': 'cat: /etc/shadow: Permission denied',
    r'^sudo cat /etc/shadow$': '''root:$6$xyz$hashedpassword:19543:0:99999:7:::
daemon:*:19543:0:99999:7:::''',
    
    r'^ifconfig$': '''eth0: flags=4163<UP,BROADCAST,RUNNING,MULTICAST>  mtu 1500
        inet 192.168.1.100  netmask 255.255.255.0  broadcast 192.168.1.255
        inet6 fe80::1  prefixlen 64  scopeid 0x20<link>
        ether 00:11:22:33:44:55  txqueuelen 1000  (Ethernet)''',
    
    r'^ip addr$': '''1: lo: <LOOPBACK,UP,LOWER_UP> mtu 65536 qdisc noqueue state UNKNOWN
    inet 127.0.0.1/8 scope host lo
2: eth0: <BROADCAST,MULTICAST,UP,LOWER_UP> mtu 1500 qdisc fq_codel state UP
    inet 192.168.1.100/24 brd 192.168.1.255 scope global eth0''',
    
    r'^netstat -tuln$': '''Active Internet connections (only servers)
Proto Recv-Q Send-Q Local Address           Foreign Address         State
tcp        0      0 0.0.0.0:22              0.0.0.0:*               LISTEN
tcp        0      0 127.0.0.1:3306          0.0.0.0:*               LISTEN
tcp6       0      0 :::80                   :::*                    LISTEN''',
    
    r'^ps aux$': '''USER       PID %CPU %MEM    VSZ   RSS TTY      STAT START   TIME COMMAND
root         1  0.0  0.1 168576 11996 ?        Ss   14:23   0:01 /sbin/init
root       456  0.0  0.0  72296  6012 ?        Ss   14:23   0:00 /usr/sbin/sshd
www-data   892  0.0  0.1 452184 15432 ?        S    14:24   0:02 /usr/sbin/apache2''',
    
    r'^df -h$': '''Filesystem      Size  Used Avail Use% Mounted on
/dev/sda1        20G  8.2G   11G  43% /
tmpfs           2.0G     0  2.0G   0% /dev/shm''',
    
    r'^free -m$': '''              total        used        free      shared  buff/cache   available
Mem:           3948        1234        1890          45         823        2456
Swap:          2047           0        2047''',
}

def get_fake_response(command):
    """Retorna resposta plausível para um comando"""
    for pattern, response in FAKE_RESPONSES.items():
        if re.match(pattern, command):
            return response
    
    # Comandos desconhecidos
    if command.startswith('cat '):
        return f'cat: {command.split()[1]}: No such file or directory'
    elif command.startswith('cd '):
        return ''  # cd não produz output
    elif command.startswith('rm '):
        return ''  # rm não produz output (assustador)
    elif command.startswith('wget ') or command.startswith('curl '):
        return 'Connection timed out'
    elif command:
        return f'bash: {command}: command not found'
    
    return ''

def fake_shell(channel, client_ip, username):
    """Shell falso que grava todos os comandos"""
    channel.send(f'{username}@ubuntu-server:~$ ')
    
    command_buffer = ''
    
    while True:
        try:
            data = channel.recv(1024)
            if not data:
                break
            
            char = data.decode('utf-8', errors='ignore')
            
            if char == '\r' or char == '\n':
                # Enter pressionado
                command = command_buffer.strip()
                
                # Log do comando
                log_command(
                    ip=client_ip,
                    username=username,
                    command=command,
                    timestamp=datetime.now().isoformat()
                )
                
                # Executa comando especial
                if command == 'exit' or command == 'logout':
                    channel.send('\r\nlogout\r\n')
                    break
                
                # Obtém resposta falsa
                response = get_fake_response(command)
                if response:
                    channel.send(f'\r\n{response}\r\n')
                
                # Prompt
                channel.send(f'{username}@ubuntu-server:~$ ')
                command_buffer = ''
                
            elif char == '\x7f':  # Backspace
                if command_buffer:
                    command_buffer = command_buffer[:-1]
                    channel.send('\b \b')
            else:
                command_buffer += char
                channel.send(char)
                
        except Exception as e:
            break

def log_command(ip, username, command, timestamp):
    """Registra comandos executados"""
    if not command:
        return
        
    log_entry = {
        'type': 'command',
        'timestamp': timestamp,
        'ip': ip,
        'username': username,
        'command': command
    }
    
    with open('/var/log/ssh-honeypot/commands.log', 'a') as f:
        f.write(json.dumps(log_entry) + '\n')
    
    print(f"[{ip}] {username}$ {command}")
hacker digitando código escuro segurança cibernética
Enquanto o invasor digita, cada comando é gravado para análise posterior

Analisando os Logs: Extraindo Inteligência

Gravar logs é fácil. O difícil é transformar esses logs em inteligência acionável. Vamos criar um analisador que identifica padrões, agrupa ataques por origem, e destaca os comandos mais perigosos.

import json
from collections import defaultdict, Counter
from datetime import datetime, timedelta

class HoneypotAnalyzer:
    """Analisa logs do honeypot e gera relatórios"""
    
    def __init__(self, auth_log_path, commands_log_path):
        self.auth_log_path = auth_log_path
        self.commands_log_path = commands_log_path
        
    def load_logs(self):
        """Carrega logs de autenticação e comandos"""
        auth_logs = []
        with open(self.auth_log_path, 'r') as f:
            for line in f:
                auth_logs.append(json.loads(line))
        
        command_logs = []
        with open(self.commands_log_path, 'r') as f:
            for line in f:
                command_logs.append(json.loads(line))
        
        return auth_logs, command_logs
    
    def analyze(self, hours=24):
        """Analisa logs das últimas N horas"""
        auth_logs, command_logs = self.load_logs()
        
        cutoff = datetime.now() - timedelta(hours=hours)
        
        # Filtrar logs recentes
        recent_auth = [
            log for log in auth_logs
            if datetime.fromisoformat(log['timestamp']) > cutoff
        ]
        
        recent_commands = [
            log for log in command_logs
            if datetime.fromisoformat(log['timestamp']) > cutoff
        ]
        
        # Agrupar por IP
        attacks_by_ip = defaultdict(lambda: {
            'auth_attempts': 0,
            'commands': [],
            'usernames': set(),
            'passwords': set()
        })
        
        for log in recent_auth:
            ip = log['ip']
            attacks_by_ip[ip]['auth_attempts'] += 1
            attacks_by_ip[ip]['usernames'].add(log['username'])
            attacks_by_ip[ip]['passwords'].add(log['password'])
        
        for log in recent_commands:
            ip = log['ip']
            attacks_by_ip[ip]['commands'].append(log['command'])
        
        # Comandos mais comuns
        all_commands = [log['command'] for log in recent_commands]
        command_freq = Counter(all_commands)
        
        # Usernames mais comuns
        all_usernames = [log['username'] for log in recent_auth]
        username_freq = Counter(all_usernames)
        
        # Senhas mais comuns
        all_passwords = [log['password'] for log in recent_auth]
        password_freq = Counter(all_passwords)
        
        return {
            'period': f'Últimas {hours} horas',
            'total_auth_attempts': len(recent_auth),
            'total_commands': len(recent_commands),
            'unique_ips': len(attacks_by_ip),
            'top_ips': sorted(
                attacks_by_ip.items(),
                key=lambda x: x[1]['auth_attempts'],
                reverse=True
            )[:10],
            'top_commands': command_freq.most_common(20),
            'top_usernames': username_freq.most_common(10),
            'top_passwords': password_freq.most_common(10),
            'attacks_by_ip': dict(attacks_by_ip)
        }
    
    def generate_report(self, hours=24):
        """Gera relatório legível"""
        analysis = self.analyze(hours)
        
        report = []
        report.append("=" * 60)
        report.append("RELATÓRIO DE INTELIGÊNCIA - SSH HONEYPOT")
        report.append("=" * 60)
        report.append(f"Período: {analysis['period']}")
        report.append(f"Total de tentativas de autenticação: {analysis['total_auth_attempts']}")
        report.append(f"Total de comandos executados: {analysis['total_commands']}")
        report.append(f"IPs únicos: {analysis['unique_ips']}")
        report.append("")
        
        report.append("TOP 10 IPs ATACANTES:")
        report.append("-" * 60)
        for ip, data in analysis['top_ips']:
            report.append(f"{ip}: {data['auth_attempts']} tentativas, {len(data['commands'])} comandos")
            if data['commands']:
                report.append(f"  Últimos comandos: {', '.join(data['commands'][-3:])}")
        report.append("")
        
        report.append("USERNAMES MAIS USADOS:")
        report.append("-" * 60)
        for username, count in analysis['top_usernames']:
            report.append(f"{username}: {count} tentativas")
        report.append("")
        
        report.append("SENHAS MAIS USADAS:")
        report.append("-" * 60)
        for password, count in analysis['top_passwords']:
            report.append(f"{password}: {count} tentativas")
        report.append("")
        
        report.append("COMANDOS MAIS EXECUTADOS:")
        report.append("-" * 60)
        for cmd, count in analysis['top_commands']:
            report.append(f"{cmd}: {count} vezes")
        
        return '\n'.join(report)

# Uso
if __name__ == '__main__':
    analyzer = HoneypotAnalyzer(
        '/var/log/ssh-honeypot/auth.log',
        '/var/log/ssh-honeypot/commands.log'
    )
    
    print(analyzer.generate_report(hours=24))

Detectando Padrões de Ataque

Com os logs em mãos, podemos identificar padrões interessantes. A maioria dos ataques segue uma sequência previsível:

  1. Reconhecimento: uname -a, cat /etc/passwd, ifconfig
  2. Escalada de privilégios: sudo, su, chmod
  3. Persistência: Criação de usuários, modificação de crontab
  4. Exfiltração: wget, curl para baixar malware

Vamos criar um detector de padrões que classifica cada sessão de ataque:

class AttackPatternDetector:
    """Detecta padrões de ataque em sessões SSH"""
    
    RECON_COMMANDS = ['uname', 'whoami', 'id', 'ifconfig', 'ip addr', 'cat /etc/passwd']
    PRIVILEGE_COMMANDS = ['sudo', 'su', 'chmod', 'chown', 'passwd']
    PERSISTENCE_COMMANDS = ['useradd', 'adduser', 'crontab', 'systemctl enable']
    EXFIL_COMMANDS = ['wget', 'curl', 'scp', 'ftp', 'nc']
    
    def classify_session(self, commands):
        """Classifica uma sessão de ataque"""
        recon_count = sum(1 for cmd in commands if any(r in cmd for r in self.RECON_COMMANDS))
        priv_count = sum(1 for cmd in commands if any(p in cmd for p in self.PRIVILEGE_COMMANDS))
        persist_count = sum(1 for cmd in commands if any(p in cmd for p in self.PERSISTENCE_COMMANDS))
        exfil_count = sum(1 for cmd in commands if any(e in cmd for e in self.EXFIL_COMMANDS))
        
        classifications = []
        
        if recon_count > 2:
            classifications.append('RECONHECIMENTO')
        if priv_count > 0:
            classifications.append('ESCALADA DE PRIVILÉGIOS')
        if persist_count > 0:
            classifications.append('PERSISTÊNCIA')
        if exfil_count > 0:
            classifications.append('EXFILTRAÇÃO/INSTALAÇÃO')
        
        if not classifications:
            classifications.append('SONDAGEM BÁSICA')
        
        return classifications
    
    def analyze_all_sessions(self, attacks_by_ip):
        """Analisa todas as sessões"""
        results = []
        
        for ip, data in attacks_by_ip.items():
            commands = data['commands']
            if not commands:
                continue
            
            classifications = self.classify_session(commands)
            
            results.append({
                'ip': ip,
                'classifications': classifications,
                'command_count': len(commands),
                'commands': commands
            })
        
        return sorted(results, key=lambda x: x['command_count'], reverse=True)

Integração com Alertas em Tempo Real

De nada adianta ter logs se você só vai olhar para eles uma semana depois. Vamos integrar o honeypot com um sistema de alertas que notifica você em tempo real quando algo interessante acontece.

import smtplib
from email.mime.text import MIMEText
import requests

class AlertSystem:
    """Sistema de alertas em tempo real"""
    
    def __init__(self, config):
        self.config = config
    
    def send_email(self, subject, body):
        """Envia alerta por email"""
        msg = MIMEText(body)
        msg['Subject'] = subject
        msg['From'] = self.config['email_from']
        msg['To'] = self.config['email_to']
        
        with smtplib.SMTP(self.config['smtp_server'], self.config['smtp_port']) as server:
            server.starttls()
            server.login(self.config['email_user'], self.config['email_pass'])
            server.send_message(msg)
    
    def send_telegram(self, message):
        """Envia alerta via Telegram"""
        url = f"https://api.telegram.org/bot{self.config['telegram_token']}/sendMessage"
        data = {
            'chat_id': self.config['telegram_chat_id'],
            'text': message,
            'parse_mode': 'Markdown'
        }
        requests.post(url, json=data)
    
    def alert_high_risk_command(self, ip, username, command):
        """Alerta para comandos de alto risco"""
        high_risk = ['wget http', 'curl http', 'nc -', 'bash -i', 'python -c']
        
        if any(risk in command for risk in high_risk):
            message = f"🚨 *COMANDO DE ALTO RISCO*\n\n"
            message += f"*IP:* `{ip}`\n"
            message += f"*Usuário:* `{username}`\n"
            message += f"*Comando:* `{command}`\n"
            
            self.send_telegram(message)
    
    def alert_brute_force(self, ip, attempts):
        """Alerta para brute force"""
        if attempts > 100:
            message = f"⚠️ *BRUTE FORCE DETECTADO*\n\n"
            message += f"*IP:* `{ip}`\n"
            message += f"*Tentativas:* {attempts}\n"
            
            self.send_telegram(message)

# Uso no servidor SSH
alert_system = AlertSystem({
    'telegram_token': 'SEU_TOKEN',
    'telegram_chat_id': 'SEU_CHAT_ID',
    'email_from': 'honeypot@seudominio.com',
    'email_to': 'admin@seudominio.com',
    'email_user': 'honeypot@seudominio.com',
    'email_pass': 'sua_senha',
    'smtp_server': 'smtp.seudominio.com',
    'smtp_port': 587
})

# No handler de comandos:
def log_command(ip, username, command, timestamp):
    # ... código anterior ...
    
    # Alerta em tempo real
    alert_system.alert_high_risk_command(ip, username, command)

Gerando Relatórios Diários Automatizados

Vamos criar um script que roda diariamente via cron e gera um relatório resumido:

#!/usr/bin/env python3
import json
from datetime import datetime, timedelta
from collections import Counter

def generate_daily_report():
    """Gera relatório diário"""
    analyzer = HoneypotAnalyzer(
        '/var/log/ssh-honeypot/auth.log',
        '/var/log/ssh-honeypot/commands.log'
    )
    
    report = analyzer.generate_report(hours=24)
    
    # Salvar relatório
    date = datetime.now().strftime('%Y-%m-%d')
    filename = f'/var/log/ssh-honeypot/reports/report-{date}.txt'
    
    with open(filename, 'w') as f:
        f.write(report)
    
    print(f"Relatório salvo em {filename}")
    
    # Opcionalmente, enviar por email
    alert_system = AlertSystem({...})
    alert_system.send_email(
        f'Relatório SSH Honeypot - {date}',
        report
    )

if __name__ == '__main__':
    generate_daily_report()

Adicione ao crontab:

0 8 * * * /usr/bin/python3 /opt/ssh-honeypot/daily_report.py

Conclusão: Transformando Invasões em Inteligência

Um SSH honeypot não é apenas uma ferramenta de segurança — é uma fonte de inteligência. Cada tentativa de invasão te ensina algo sobre as táticas, técnicas e procedimentos (TTPs) dos atacantes reais.

Com o que construímos hoje, você tem:

  • Um servidor SSH que aceita qualquer credencial
  • Um shell falso com respostas plausíveis
  • Logging completo de autenticações e comandos
  • Analisador que identifica padrões de ataque
  • Sistema de alertas em tempo real
  • Relatórios diários automatizados

O próximo passo? Rode isso em uma VPS barata (R$10/mês), aponte um domínio para ela, e espere. Em 24 horas você terá dados reais sobre quem está escaneando a internet e o que eles fazem quando encontram um servidor vulnerável.

A pergunta que fica: qual automação de segurança você quer ver no próximo post? Um detector de port scanning? Um WAF minimalista? Um analisador de tráfego DNS? Me conta nos comentários que eu trago no próximo artigo.

Até a próxima, e lembre-se: a melhor forma de se defender é entender como o inimigo ataca. 🛡️

Posts Similares