Ferramentas de Coleta Ativa de Informações em Redes

  1. Descoberta de Hosts Baseada em ICMP

Implementação em Python para detecção de hosts:

from scapy.all import *
from scapy.layers.inet import IP, ICMP
from random import randint
from argparse import ArgumentParser
import time

def executar_verificacao():
    uso = 'python %(prog)s -t <host alvo>'
    analisador = ArgumentParser(uso=uso)
    analisador.add_argument('-t', '--target', dest='host_alvo', type=str, required=True)
    opcoes = analisador.parse_args()

    print("Relatório de varredura para: " + opcoes.host_alvo + "\n")

    if '-' in opcoes.host_alvo:
        inicio = int(opcoes.host_alvo.split('-')[0].split('.')[3])
        fim = int(opcoes.host_alvo.split('-')[1]) + 1
        rede_base = '.'.join(opcoes.host_alvo.split('.')[:3])
        
        for num in range(inicio, fim):
            verificar_host(f"{rede_base}.{num}")
            time.sleep(0.2)
    else:
        verificar_host(opcoes.host_alvo)

    print("\nVarredura concluída!...\n")

def verificar_host(endereco):
    id_pacote = randint(1, 65535)
    id_icmp = randint(1, 65535)
    seq_icmp = randint(1, 65535)
    pacote = IP(dst=endereco, ttl=64, id=id_pacote)/ICMP(id=id_icmp, seq=seq_icmp)/b'payload_data'

    resposta = sr1(pacote, timeout=1, verbose=False)

    if resposta:
        for recebido in resposta:
            ip_verificado = recebido[IP].src
            print(f"{ip_verificado} ---> Host ativo")
    else:
        print(f"{endereco} ---> Host inativo")

if __name__ == '__main__':
    try:
        executar_verificacao()
    except KeyboardInterrupt:
        print("interrompido pelo usuário, encerrando threads...")

Implementação usendo nmap para detecção de hosts:

import nmap
import argparse

def iniciar_processo():
    uso = 'python %(prog)s -t <host alvo>'
    analisador = argparse.ArgumentParser(uso=uso)
    analisador.add_argument('-t', '--target', dest='host_alvo', required=True)
    opcoes = analisador.parse_args()

    if '-' in opcoes.host_alvo:
        inicio = int(opcoes.host_alvo.split('-')[0].split('.')[3])
        fim = int(opcoes.host_alvo.split('-')[1]) + 1
        rede_base = '.'.join(opcoes.host_alvo.split('.')[:3])
        
        for num in range(inicio, fim):
            escanear_com_nmap(f"{rede_base}.{num}")
    else:
        escanear_com_nmap(opcoes.host_alvo)

def escanear_com_nmap(alvo):
    scanner = nmap.PortScanner()
    try:
        resultado = scanner.scan(hosts=alvo, arguments='-sn -PE')
        estado = resultado['scan'][alvo]['status']['state']
        print(f"[{alvo}] está [{estado}]")
    except Exception as e:
        pass

if __name__ == '__main__':
    try:
        iniciar_processo()
    except KeyboardInterrupt:
        print('interrompido pelo usuário, encerrando threads...')

  1. Descoberta de Hosts Baseada em TCP

Detecção de hosts usando pacotes ACK em Python:

import argparse
from random import randint
from scapy.layers.inet import IP, TCP
from scapy.all import *
import time

def executar_principal():
    uso = 'uso: python %(prog)s -t <host alvo>'
    analisador = argparse.ArgumentParser(uso=uso)
    analisador.add_argument('-t', '--target', dest='host_alvo', required=True)
    opcoes = analisador.parse_args()

    if '-' in opcoes.host_alvo:
        inicio = int(opcoes.host_alvo.split('-')[0].split('.')[3])
        fim = int(opcoes.host_alvo.split('-')[1]) + 1
        rede_base = '.'.join(opcoes.host_alvo.split('.')[:3])
        
        for num in range(inicio, fim):
            analisar_tcp(f"{rede_base}.{num}")
    else:
        analisar_tcp(opcoes.host_alvo)

def analisar_tcp(alvo):
    try:
        porta_destino = randint(1, 65535)
        pacote = IP(dst=alvo)/TCP(flags='A', dport=porta_destino)
        resposta = sr1(pacote, timeout=1.0, verbose=False)

        if resposta:
            if int(resposta[TCP].flags) == 4:
                time.sleep(0.5)
                print(f"{alvo} está ativo")
            else:
                print(f"{alvo} está inativo")
        else:
            print(f"{alvo} está inativo")
    except:
        pass

if __name__ == '__main__':
    try:
        executar_principal()
    except Exception as e:
        print('interrompido pelo usuário, encerrando threads...')

  1. Descoberta de Hosts Baseada em UDP

Implementação em Python para detecção de hosts via UDP:

import argparse
from random import randint
from scapy.all import *
from scapy.layers.inet import IP, UDP
import time

def principal():
    analisador = argparse.ArgumentParser(uso='uso:python %(prog)s -t <host alvo>')
    analisador.add_argument('-t', '--target', dest='host_alvo', required=True)
    opcoes = analisador.parse_args()

    if '-' in opcoes.host_alvo:
        inicio = int(opcoes.host_alvo.split('-')[0].split('.')[3])
        fim = int(opcoes.host_alvo.split('-')[1]) + 1
        rede_base = '.'.join(opcoes.host_alvo.split('.')[:3])
        
        for num in range(inicio, fim):
            verificar_udp(f"{rede_base}.{num}")
    else:
        verificar_udp(opcoes.host_alvo)

def verificar_udp(endereco):
    try:
        porta_destino = randint(1, 65535)
        pacote = IP(dst=endereco)/UDP(dport=porta_destino)
        resposta = sr1(pacote, timeout=1, verbose=0)

        if resposta:
            if int(resposta[IP].proto) == 1:
                time.sleep(0.5)
                print(f"{endereco} está ativo")
            else:
                print(f"{endereco} está inativo")
        else:
            print(f"{endereco} está inativo")
    except:
        pass

if __name__ == '__main__':
    try:
        principal()
    except Exception as e:
        print('interrompido pelo usuário, encerrando threads...')

  1. Descoberta de Hosts Baseada em ARP

Implementação em Python para detecção de hosts via ARP:

import os
import re
import argparse
from scapy.all import *
from scapy.layers.inet import IP
from scapy.layers.l2 import Ether, ARP

def obter_enderecos(interface):
    dados_interface = os.popen(f'ifconfig {interface}')
    linhas_dados = dados_interface.readlines()
    
    if re.search('\w\w:\w\w:\w\w:\w\w:\w\w:\w\w', str(linhas_dados)):
        endereco_mac = re.search('\w\w:\w\w:\w\w:\w\w:\w\w:\w\w', str(linhas_dados)).group(0)
    
    if re.search(r'((2[0-4]\d|25[0-5]|[01]?\d\d?)\.){3}(2[0-4]\d|25[0-5]|[01]?\d\d?)', str(linhas_dados)):
        endereco_ip = re.search(r'((2[0-4]\d|25[0-5]|[01]?\d\d?)\.){3}(2[0-4]\d|25[0-5]|[01]?\d\d?)', str(linhas_dados))
    
    info_enderecos = (endereco_ip, endereco_mac)
    return info_enderecos

def varredura_arp(interface='eth0'):
    mac_local = obter_enderecos(interface)[1]
    ip_local = obter_enderecos(interface)[0]
    partes_ip = ip_local.split('.')
    
    lista_ips = []
    for i in range(1, 255):
        item_ip = f"{partes_ip[0]}.{partes_ip[1]}.{partes_ip[2]}.{i}"
        lista_ips.append(item_ip)

    pacote = Ether(src=mac_local, dst='FF:FF:FF:FF:FF:FF')/ARP(op=1, hwsrc=mac_local, hwdst='00:00:00:00:00:00', pdst=lista_ips, iface=interface, timeout=2, verbose=False)
    resultado = srp(pacote)
    respostas = resultado[0].res

    hosts_vivos = []
    total_respostas = len(respostas)
    print("==============================")
    print('Resultado da Varredura ARP')
    print(f"IP Local: {ip_local}")
    print(f"MAC Local: {mac_local}")
    print("==============================")
    
    for x in range(total_respostas):
        ip_respondido = respostas[x][1][1].fields['psrc']
        mac_respondido = respostas[x][1][1].fields['hwsrc']
        hosts_vivos.append([ip_respondido, mac_respondido])
        print(f"IP: {ip_respondido}\n\nMAC: {mac_respondido}")
        print("===========================")
    
    arquivo_resultado = open("resultado", "w")
    for i in range(len(hosts_vivos)):
        arquivo_resultado.write(hosts_vivos[i][0] + '\n')
    arquivo_resultado.close()

def principal():
    analisador = argparse.ArgumentParser(uso='uso:python %(prog)s -i interface \n'
                                           'Exemplo: python %(prog)s -i eth0\n')
    analisador.add_argument('-i', '--iface', dest='interface', required=True, help='nome da interface')
    opcoes = analisador.parse_args()
    varredura_arp(opcoes.interface)

if __name__ == '__main__':
    try:
        principal()
    except Exception as e:
        print('Erro...')

  1. Detecção de Portas

Implementação em Python para detecção de portas:

import sys
import socket
import argparse
import threading
import queue

class ScannerPortas(threading.Thread):
    def __init__(self, fila_portas, endereco_ip, tempo_limite=3):
        threading.Thread.__init__(self)
        self._fila_portas = fila_portas
        self._endereco_ip = endereco_ip
        self._tempo_limite = tempo_limite

    def run(self):
        while True:
            if self._fila_portas.empty():
                break

            porta = self._fila_portas.get(timeout=0.5)
            try:
                conexao = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
                conexao.settimeout(self._tempo_limite)
                codigo_resultado = conexao.connect_ex((self._endereco_ip, porta))
                
                if codigo_resultado == 0:
                    sys.stdout.write(f"[{porta}] ABERTA\n")
            except Exception as e:
                print(e)
            finally:
                conexao.close()

def iniciar_varredura(alvo, portas, numero_threads):
    lista_portas = []
    if '-' in portas:
        for i in range(int(portas.split('-')[0]), int(portas.split('-')[1]) + 1):
            lista_portas.append(i)
    else:
        lista_portas.append(int(portas))
    
    ip_destino = alvo
    threads_lista = []
    threads_quantidade = numero_threads
    fila_portas = queue.Queue()
    
    for porta in lista_portas:
        fila_portas.put(porta)
    
    for t in range(threads_quantidade):
        threads_lista.append(ScannerPortas(fila_portas, ip_destino, tempo_limite=3))
    
    for thread in threads_lista:
        thread.start()
    
    for thread in threads_lista:
        thread.join()

if __name__ == '__main__':
    try:
        analisador = argparse.ArgumentParser()
        analisador.add_argument('-i', '--ip', dest='alvo_ip', help='IP alvo', required=True)
        analisador.add_argument('-p', '--port', dest='portas', help='Portas para varredura', required=True)
        analisador.add_argument('-t', '--thread', dest='threads', default=100, help='número de threads')
        opcoes = analisador.parse_args()
        iniciar_varredura(opcoes.alvo_ip, opcoes.portas, opcoes.threads)
    except:
        pass

  1. Identificação de Serviços

Implementação em Python para identificação de serviços:

import argparse
import socket
import re
import time

ASSINATURAS = (
    b'FTP|FTP|^220.*FTP',
    b'MySQL|MySQL|mysql_native_password',
    b'oracle-https|^220- ora',
    b'Telnet|Telnet|Telnet',
    b'Telnet|Telnet|^\r\n%conexao fechada por host remoto!\x00$',
    b'VNC|VNC|^RFB',
    b'IMAP|IMAP|^\* OK.*?IMAP',
    b'POP|POP|^\+Ok.*?',
    b'SMTP|SMTP|^220.*?SMTP',
    b'Kangle|Kangle|HTTP.*kangle',
    b'SMTP|SMTP|^554 SMTP',
    b'SSH|SSH|^SSH-',
    b'HTTPS|HTTPS|Location: https',
    b'HTTP|HTTP|HTTP/1.1',
    b'HTTP|HTTP|HTTP/1.0'
)

def executar_principal():
    analisador = argparse.ArgumentParser()
    analisador.add_argument('-i', '--ip', dest='endereco_ip', type=str, help='especificar host alvo', required=True)
    analisador.add_argument('-p', '--port', dest='portas', type=str, help='especificar porta alvo', required=True)
    opcoes = analisador.parse_args()
    ip = opcoes.endereco_ip
    portas = opcoes.portas
    print(f"Relatório de varredura para {ip}\n")
    
    for linha in portas.split(','):
        requisitar_servico(ip, linha)
        time.sleep(0.2)
    
    print("\nVarredura concluída!...\n")

def requisitar_servico(ip, porta):
    resposta = ''
    TESTE = 'GET / HTTP/1.0\r\n\r\n'
    socket_cliente = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    socket_cliente.settimeout(10)
    resultado_conexao = socket_cliente.connect((ip, int(porta)))
    
    if resultado_conexao == 0:
        try:
            socket_cliente.sendall(TESTE.encode())
            resposta = socket_cliente.recv(256)
            if resposta:
                analisar_padroes(resposta, porta)
        except(ConnectionResetError, socket.timeout):
            pass
    else:
        pass
    socket_cliente.close()

def analisar_padroes(resposta, porta):
    if re.search(b'<title>502 Bad Gateway', resposta):
        protocolo = {"Serviço falhou ao acessar!!"}
    
    for padrao in ASSINATURAS:
        padrao = padrao.split(b'|')
        if re.search(padrao[-1], resposta, re.IGNORECASE):
            protocolo = f"[{porta}] aberta {padrao[1].decode()}"
            break
        else:
            protocolo = f"[{porta}] aberta Não reconhecida"

    print(protocolo)

if __name__ == '__main__':
    try:
        executar_principal()
    except KeyboardInterrupt:
        print('interrompido pelo usuário, encerrando threads...')

  1. Identificação de Sistemas Operacionasi

Detecção de sistema operacional baseada no valor TTL:

import os
import re
import argparse

def executar_principal():
    analisador = argparse.ArgumentParser()
    analisador.add_argument('-i', '--ip', dest='host_alvo', required=True)
    opcoes = analisador.parse_args()
    ip = opcoes.host_alvo
    analisar_ttl(ip)

def analisar_ttl(ip):
    padrao_ttl_str = re.compile(r'ttl=\d+')
    padrao_ttl_num = re.compile(r'\d+')
    resultado = os.popen(f"ping -c 1 {ip}")
    saida = resultado.read()
    
    for linha in saida.splitlines():
        resultado_padrao = padrao_ttl_str.findall(linha)
        if resultado_padrao:
            ttl_valor = padrao_ttl_num.findall(resultado_padrao[0])
            if int(ttl_valor[0]) <= 64:
                print(f"{ip} é Linux/UNIX")
            else:
                print(f"{ip} é Windows")
        else:
            pass

if __name__ == '__main__':
    executar_principal()

  1. Detecção de Diretórios Sensíveis

Implementação baseada em dicionário para detecção de diretórios sensíveis:

import requests

cabecalhos = {
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/119.0.0.0 Safari/537.36"
}

url_alvo = input("url: ")
arquivo_dicionario = input('arquivo_dicionario.txt')

lista_urls = []
if arquivo_dicionario == "":
    arquivo_dicionario = "arquivo_dicionario.txt"

try:
    with open(arquivo_dicionario, 'r') as arquivo:
        for linha in arquivo:
            linha = linha.replace('\n', '')
            lista_urls.append(linha)
except:
    print("erro! ")

for item in lista_urls:
    url_completa = f"http://{url_alvo}/{item}"
    
    try:
        resposta = requests.get(url_completa, headers=cabecalhos)
        print(f"{url_completa}----------{resposta.status_code}")
    except Exception as e:
        print(f'{url_completa}----------Erro')

Tags: Python network-security information-gathering scapy nmap

Publicado em 9-3 01:51