Construa um Port Scanner em Python

Passos

1. Suba alvos de teste locais

   Você precisa de portas abertas reais para escanear. Suba dois
   listeners locais inofensivos em terminais separados para o seu
   scanner ter algo a encontrar.

   # Terminal A — um servidor HTTP simples na porta 8000
   python -m http.server 8000
   
   # Terminal B — um listener TCP eco mínimo e cru na porta 9000
   python -c "
   import socket
   s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
   s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
   s.bind(('127.0.0.1', 9000))
   s.listen(5)
   print('listening on 9000')
   while True:
       conn, addr = s.accept()
       conn.send(b'ECHO-SERVER-1.0 ready\n')
       conn.close()
   "
   

   Deixe os dois rodando pelo resto do lab. Agora você tem duas portas
   abertas conhecidas (8000, 9000) mais qualquer outra coisa já
   aberta na sua máquina (geralmente nada mais nas portas baixas, a menos
   que você tenha SSH, um banco de dados ou um servidor de dev rodando).

   Checkpoint

   Confirme que cada listener responde usando telnet ou um one-liner de
   socket cru (pule se telnet não estiver instalado — o one-liner
   Python funciona em qualquer lugar):

   python -c "
   import socket
   s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
   s.settimeout(2)
   s.connect(('127.0.0.1', 9000))
   print(s.recv(100))
   "
   

   Você deve ver b'ECHO-SERVER-1.0 ready\n' impresso. Se você receber um
   ConnectionRefusedError, o listener do Terminal B não está rodando —
   volte e inicie-o.

   Entregável deste passo: dois listeners rodando localmente nas
   portas 8000 e 9000, confirmados alcançáveis pelo script de checkpoint.

2. Escreva o scanner sequencial

   Crie scanner.py com uma função única que escaneia uma porta e
   reporta seu estado, depois faça um loop sobre um range.

   # scanner.py
   import socket
   import sys
   import time
   
   DEFAULT_TIMEOUT = 0.5  # segundos
   
   
   def scan_port(host: str, port: int, timeout: float = DEFAULT_TIMEOUT) -> str:
       """Retorna 'open', 'closed' ou 'filtered' para uma porta TCP única."""
       sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
       sock.settimeout(timeout)
       try:
           result = sock.connect_ex((host, port))
           # connect_ex retorna 0 no sucesso em vez de levantar exceção
           return "open" if result == 0 else "closed"
       except socket.timeout:
           return "filtered"
       except OSError:
           return "closed"
       finally:
           sock.close()
   
   
   def scan_range(host: str, start_port: int, end_port: int):
       open_ports = []
       start = time.time()
       for port in range(start_port, end_port + 1):
           state = scan_port(host, port)
           if state == "open":
               print(f"Port {port:<6} OPEN")
               open_ports.append(port)
       elapsed = time.time() - start
       print(f"Scan complete: {len(open_ports)} open, "
             f"{end_port - start_port + 1 - len(open_ports)} closed/filtered, "
             f"{elapsed:.1f}s elapsed")
       return open_ports
   
   
   if __name__ == "__main__":
       host = sys.argv[1] if len(sys.argv) > 1 else "127.0.0.1"
       if host not in ("127.0.0.1", "localhost", "::1"):
           print("Recusando escanear um host não local. Esta ferramenta é "
                 "só para testes locais autorizados.")
           sys.exit(1)
       start_port = int(sys.argv[2]) if len(sys.argv) > 2 else 1
       end_port = int(sys.argv[3]) if len(sys.argv) > 3 else 1024
       print(f"Scanning {host} (ports {start_port}-{end_port})...")
       scan_range(host, start_port, end_port)
   

   Repare em duas escolhas de design deliberadas:

   • connect_ex em vez de connect. connect_ex retorna um código
     de erro em vez de levantar uma exceção na recusa, o que mantém o
     loop quente livre da sobrecarga de tratamento de exceção para o caso
     extremamente comum de "fechada".
   • Uma allowlist de host no __main__. O script se recusa a rodar
     contra qualquer coisa que não seja um endereço loopback. Isso não é
     decoração — é a linha de código mais importante deste lab inteiro.
     Mantenha-a em toda versão que você escrever a partir daqui.

   Rode

   python scanner.py 127.0.0.1 1 1024
   

   Checkpoint

   Você deve ver as portas 8000 e 9000 (do Passo 1) reportadas como
   OPEN, mais possivelmente outras poucas já rodando na sua máquina
   (SSH na 22 é comum em macOS/Linux). Cronometre a execução — na maioria
   das máquinas, 1.024 portas sequenciais com timeout de 0.5s leva vários
   segundos para portas abertas e pode levar muito mais se alguma porta
   estiver filtrada (cada porta filtrada queima o timeout completo).
   Anote o tempo decorrido impresso no final — você vai comparar com a
   versão com threads no Passo 3.

   Entregável deste passo: um scanner.py funcionando que reporta
   corretamente as portas 8000 e 9000 como abertas, mais o tempo
   decorrido de um scan de 1–1024.

3. Acelere com um pool de threads

   Um scan sequencial é dominado pela espera: cada connect() ou retorna
   quase instantaneamente (aberta/fechada) ou bloqueia pelo timeout
   completo (filtrada). I/O de rede como este é exatamente para o que
   threads são boas em Python — o GIL é liberado durante chamadas de
   socket bloqueantes, então threads realmente rodam concorrentemente
   aqui.

   Adicione uma função de scan com threads usando
   concurrent.futures.ThreadPoolExecutor:

   # adicione ao scanner.py
   from concurrent.futures import ThreadPoolExecutor, as_completed
   
   DEFAULT_WORKERS = 100
   
   
   def scan_range_threaded(host: str, start_port: int, end_port: int,
                            workers: int = DEFAULT_WORKERS):
       open_ports = []
       ports = range(start_port, end_port + 1)
       start = time.time()
   
       with ThreadPoolExecutor(max_workers=workers) as pool:
           future_to_port = {
               pool.submit(scan_port, host, port): port for port in ports
           }
           for future in as_completed(future_to_port):
               port = future_to_port[future]
               state = future.result()
               if state == "open":
                   print(f"Port {port:<6} OPEN")
                   open_ports.append(port)
   
       elapsed = time.time() - start
       total = end_port - start_port + 1
       print(f"Scan complete: {len(open_ports)} open, "
             f"{total - len(open_ports)} closed/filtered, "
             f"{elapsed:.1f}s elapsed ({workers} workers)")
       return sorted(open_ports)
   

   Conecte como uma opção:

   # substitua o bloco __main__
   if __name__ == "__main__":
       host = sys.argv[1] if len(sys.argv) > 1 else "127.0.0.1"
       if host not in ("127.0.0.1", "localhost", "::1"):
           print("Recusando escanear um host não local. Esta ferramenta é "
                 "só para testes locais autorizados.")
           sys.exit(1)
       start_port = int(sys.argv[2]) if len(sys.argv) > 2 else 1
       end_port = int(sys.argv[3]) if len(sys.argv) > 3 else 1024
       print(f"Scanning {host} (ports {start_port}-{end_port})...")
       scan_range_threaded(host, start_port, end_port)
   

   Por que não milhares de workers?

   Mais threads não é de graça. Cada socket aberto consome um file
   descriptor (sistemas operacionais limitam isso — geralmente 1024 por
   padrão), e tentativas de conexão ainda custam syscall e overhead real
   de kernel. Um valor de workers na faixa de 50–200 é tipicamente o
   ponto ideal para um scan contra um único host: rápido o suficiente
   para ser dramaticamente melhor que o sequencial, sem esgotar file
   descriptors ou fazer a pilha de rede da sua própria máquina virar o
   gargalo.

   Também existe uma dimensão de cortesia: num alvo real (autorizado),
   um scan extremamente agressivo pode parecer — e se comportar
   funcionalmente como — um denial-of-service. Um scanner bem-comportado
   escolhe um nível de concorrência apropriado ao alvo e à autorização
   que tem, não só "o mais rápido que meu hardware permitir".

   Checkpoint

   Rode a versão com threads e compare o tempo decorrido diretamente com
   a medição do Passo 2:

   python scanner.py 127.0.0.1 1 1024
   

   Você deve ver o mesmo conjunto de portas abertas (8000, 9000, mais o
   que já estivesse aberto), mas um tempo decorrido dramaticamente menor
   — tipicamente 5–20x mais rápido, já que os timeouts das portas
   filtradas agora se sobrepõem em vez de se empilhar sequencialmente.

   Entregável deste passo: um scan com threads que reporta as mesmas
   portas abertas do Passo 2, com a melhoria de tempo decorrido anotada.

4. Adicione banner grabbing para detecção de serviço

   Saber que uma porta está aberta é útil; saber o que provavelmente
   está escutando é mais útil ainda. Muitos serviços se anunciam no
   momento em que a conexão é feita — SSH, FTP e SMTP enviam um banner de
   saudação sem serem solicitados. Servidores HTTP, em contraste,
   esperam por uma requisição primeiro. Um bom scanner tenta os dois:
   lê passivamente por um instante, e se nada chegar, envia uma sondagem
   HTTP mínima.

   # adicione ao scanner.py
   COMMON_PORTS = {
       21: "FTP", 22: "SSH", 23: "Telnet", 25: "SMTP",
       80: "HTTP", 443: "HTTPS", 3306: "MySQL",
       5432: "PostgreSQL", 6379: "Redis", 8000: "HTTP-alt",
       8080: "HTTP-alt", 9000: "Unknown/custom",
   }
   
   
   def grab_banner(host: str, port: int, timeout: float = 1.0) -> str:
       """Leitura best-effort de um banner de serviço. Retorna 'unknown' se não houver."""
       try:
           sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
           sock.settimeout(timeout)
           sock.connect((host, port))
   
           # Passo 1: muitos serviços (SSH, FTP, SMTP) saúdam imediatamente.
           sock.settimeout(0.8)
           try:
               data = sock.recv(256)
               if data:
                   return data.decode(errors="replace").strip()
           except socket.timeout:
               pass  # sem banner espontâneo — tenta provocar em vez disso
   
           # Passo 2: serviços tipo HTTP precisam de uma requisição antes de falar.
           try:
               sock.sendall(b"HEAD / HTTP/1.0\r\n\r\n")
               sock.settimeout(0.8)
               data = sock.recv(256)
               if data:
                   return data.decode(errors="replace").splitlines()[0]
           except (socket.timeout, OSError):
               pass
   
           return "unknown/no banner"
       except OSError:
           return "unreachable"
       finally:
           sock.close()
   
   
   def scan_with_banners(host: str, start_port: int, end_port: int,
                          workers: int = DEFAULT_WORKERS):
       open_ports = scan_range_threaded(host, start_port, end_port, workers)
       print("\nService detection:")
       for port in open_ports:
           banner = grab_banner(host, port)
           guess = COMMON_PORTS.get(port, "unknown")
           print(f"Port {port:<6} OPEN   guess={guess:<12} banner={banner!r}")
   

   Por que essa ordem de sondagem em dois passos importa

   Se você enviar a sondagem HTTP primeiro para toda porta, você corre
   o risco de confundir ou fazer serviços não-HTTP se comportarem mal
   (alguns vão logar uma requisição lixo, alguns só vão fechar a
   conexão). Leitura-passiva-primeiro, sondagem-ativa-depois é a mesma
   ordem que ferramentas reais usam, e é mais gentil com o que
   realmente está escutando.

   Teste contra seus dois listeners conhecidos

   # checagem manual rápida
   print(grab_banner("127.0.0.1", 9000))  # -> "ECHO-SERVER-1.0 ready"
   print(grab_banner("127.0.0.1", 8000))  # -> uma linha de status HTTP, ex.: "HTTP/1.0 501 ..."
   

   Checkpoint

   Rode o scan completo com banners:

   python -c "
   from scanner import scan_with_banners
   scan_with_banners('127.0.0.1', 1, 1024)
   "
   

   Confirme que a porta 9000 mostra o banner ECHO-SERVER-1.0 ready
   (capturado passivamente) e a porta 8000 mostra uma linha de resposta
   HTTP (capturada via a sondagem ativa). Se o SSH estiver rodando
   localmente, confirme que ele também mostra um banner SSH-2.0-... —
   essa é a linha mais útil que um banner grab pode produzir, já que
   banners SSH geralmente incluem o software e a versão exatos do
   servidor.

   Entregável deste passo: um relatório de scan onde cada porta
   aberta mostra um banner ou um explícito "unknown/no banner",
   distinguindo corretamente o serviço saudado passivamente (9000) do
   sondado ativamente (8000).

5. Blinde, documente os limites e submeta

   Termine apertando os freios de segurança e escrevendo, explicitamente,
   as regras que esta ferramenta deve seguir.

   Adicione um rate limit e uma CLI adequada

   Um scanner sem nenhum ritmo é indistinguível de uma ferramenta de
   ataque agressiva. Adicione um botão pequeno e opcional de delay e uma
   interface argparse adequada para que a ferramenta se
   autodocumente:

   # substitua o bloco __main__ mais uma vez
   import argparse
   
   if __name__ == "__main__":
       parser = argparse.ArgumentParser(
           description="Scanner de portas TCP educacional. SÓ HOSTS LOCAIS."
       )
       parser.add_argument("host", nargs="?", default="127.0.0.1")
       parser.add_argument("start_port", nargs="?", type=int, default=1)
       parser.add_argument("end_port", nargs="?", type=int, default=1024)
       parser.add_argument("--workers", type=int, default=DEFAULT_WORKERS)
       parser.add_argument("--banners", action="store_true",
                            help="tenta detecção de serviço/banner nas portas abertas")
       args = parser.parse_args()
   
       if args.host not in ("127.0.0.1", "localhost", "::1"):
           print("Recusando escanear um host não local. Esta ferramenta é "
                 "só para testes locais autorizados. Veja o README.")
           sys.exit(1)
   
       print(f"Scanning {args.host} (ports {args.start_port}-{args.end_port})...")
       if args.banners:
           scan_with_banners(args.host, args.start_port, args.end_port, args.workers)
       else:
           scan_range_threaded(args.host, args.start_port, args.end_port, args.workers)
   

   Escreva o limite onde qualquer um que use esta ferramenta vai ver

   Crie um README.md curto ao lado do scanner.py:

   # scanner.py — port scanner TCP educacional
   
   Construído para um lab de segurança do DARE. Demonstra como um TCP
   connect scan, concorrência com threads e banner grabbing funcionam.
   
   ## Uso autorizado apenas
   
   Esta ferramenta se recusa a rodar contra qualquer coisa que não seja
   127.0.0.1 / localhost / ::1 por design (veja a checagem em
   `__main__`). Não remova essa checagem para apontar isto a um host que
   você não possui ou não tem autorização explícita por escrito para
   testar. Port scanning não autorizado pode violar lei de crime
   computacional e políticas de uso aceitável.
   
   ## Uso
   
   python scanner.py [host] [start_port] [end_port] [--workers N] [--banners]
   

   Checkpoint

   Rode a ferramenta finalizada de ponta a ponta com o conjunto completo
   de funcionalidades:

   python scanner.py 127.0.0.1 1 1024 --workers 150 --banners
   

   Confirme: ela recusa um host não local (teste com
   python scanner.py 8.8.8.8 e confirme que imprime a recusa e sai com
   código diferente de zero em vez de escanear), ela encontra suas portas
   abertas conhecidas, e reporta banners para pelo menos os dois
   listeners do Passo 1.

   Critério de submissão

   Submeta quando tudo o que segue for verdade:

   1. scanner.py implementando os três estágios: sequencial
      (scan_port/scan_range), com threads (scan_range_threaded) e
      ciente de banner (grab_banner/scan_with_banners).
   2. A checagem de allowlist de host local presente e não modificada no
      __main__, confirmada a recusar um host não-loopback.
   3. Uma nota de comparação de tempo (dos Passos 2 e 3) mostrando o
      ganho de velocidade com threads na sua máquina.
   4. Um README.md declarando o limite de uso autorizado da ferramenta
      com suas próprias palavras.