Cách làm một công cụ quét cổng trên Python sử dụng thư viện Socket

Trong bài hướng dẫn này, tôi sẽ hướng dẫn bạn tạo cho riêng mình một công cụ quét cổng trên Python sử dụng thư viện Socket. Nguyên lý đằng sau công cụ quét cổng đơn giản này đó là cố gắng thực hiện kết nối đến một cổng nhất định (trang web, máy chủ hoặc bất kỳ thiết bị nào kết nối đến mạng Internet/mạng nội bộ) thông qua một danh sách các cổng, nếu một kết nối được thiết lập thành công, điều đó có nghĩa cổng đang mở.
Ví dụ, khi bạn tải trang web này, bạn phải tạo ra một kết nối đến với máy chủ thông qua cổng 80, tương tự, đoạn code này sẽ cố gắng kết nối đến một máy chủ nhưng trên nhiều cổng. Loại công cụ này hữu ích đối với các hacker và người thực hiện kiểm thử bảo mật, do vậy đừng dùng công cụ này đối với một máy chủ mà bạn không có quyền để kiểm tra!
Bạn có thể cài đặt thư viện colorama để in ra màn hình màu mè:

pip3 install colorama

Hãy bắt đầu bằng việc import module socket vào:

import socket # for connecting
from colorama import init, Fore
# some colors
init()
GREEN = Fore.GREEN
RESET = Fore.RESET
GRAY = Fore.LIGHTBLACK_EX

Lưu ý: module socket đã được cài đặt sẵn trên máy của bạn, nó là module tích hợp sẵn trong bộ thư viện tiêu chuẩn của Python, do đó bạn không phải cài bất cứ thứ gì cả.
Socket module cung cấp cho chúng ta các công việc liên quan đến socket, các hàm thực hiện các chức năng liên quan đến hệ thống mạng, … Chúng được sử dụng rộng rãi trên mạng Internet, bởi chúng đứng đằng sau tất cả các kết nối đến tất cả các hệ thống mạng. Bất cứ liên lạc nào trên hệ thống mạng đều thông qua một socket, để biết thêm chi tiết hãy truy cập official Python documentation.
Mình sẽ sử dụng thư viện colorama để in ra màu xanh biểu thị trạng thái cổng mở, màu xám cho trạng thái cổng đóng.
Hãy định nghĩa hàm có nhiệm vụ xác định cổng đóng hay mở:

def is_port_open(host, port):
    """
    determine whether `host` has the `port` open
    """
    # creates a new socket
    s = socket.socket()
    try:
        # tries to connect to host using that port
        s.connect((host, port))
        # make timeout if you want it a little faster ( less accuracy )
        # s.settimeout(0.2)
    except:
        # cannot connect, port is closed
        # return false
        return False
    else:
        # the connection was established, port is open!
        return True

Hàm s.connect((host, port)) cố gắng thực hiện kết nối socket đến một địa chỉ từ xa (host,port), nó sẽ tạo ra ngoại lệ khi kết nối đến máy chủ đó thất bại, đó là lý do tại sao mình bọc dòng code đó bên trong cấu trúc try-except, do vậy bất cứ khi nào một ngoại lệ được tạo ra, đó là dấu hiệu cho chúng ta thấy rằng thực ra cổng đã đóng, còn ngược lại thì cổng vẫn mở.
Bây giờ hãy nhét hàm đó vào vòng lặp:

# get the host from the user
host = input("Enter the host:")
# iterate over ports, from 1 to 1024
for port in range(1, 1025):
    if is_port_open(host, port):
        print(f"{GREEN}[+] {host}:{port} is open      {RESET}")
    else:
        print(f"{GRAY}[!] {host}:{port} is closed    {RESET}", end="\r")

Đó là tất cả, đây là hình chụp khi mình cố gắng quét một máy nội bộ trong mạng của mình:

Bây giờ khi bạn chạy đoạn mã này, bạn sẽ ngay lập tức để ý rằng nó chạy khá chậm, chúng ta có thể giải quyết bằng cách đặt giá trị timeout khoảng 200 mili giây. Tuy nhiên, điều này thực chất làm giảm tính chính xác của quá trình quét, một cách tốt hơn để thưc hiện điều này đó là sử dụng kỹ thuật đa luồng, đừng lo lắng nếu bạn không hiểu gì về nó, mình đã tạo ra phiên bản chạy đa luồng của công cụ này.
Lưu ý rằng đoạn code này dành cho những người thử nghiệm trên chính thiết bị của họ và mình sẽ không có trách nhiệm nếu nó bị sử dụng sai mục đích.
Happy Coding ♥

Mã nguồn

simple_port_scanner.py

import socket # for connecting
from colorama import init, Fore
# some colors
init()
GREEN = Fore.GREEN
RESET = Fore.RESET
GRAY = Fore.LIGHTBLACK_EX
def is_port_open(host, port):
    """
    determine whether `host` has the `port` open
    """
    # creates a new socket
    s = socket.socket()
    try:
        # tries to connect to host using that port
        s.connect((host, port))
        # make timeout if you want it a little faster ( less accuracy )
        s.settimeout(0.2)
    except:
        # cannot connect, port is closed
        # return false
        return False
    else:
        # the connection was established, port is open!
        return True
# get the host from the user
host = input("Enter the host:")
# iterate over ports, from 1 to 1024
for port in range(1, 1025):
    if is_port_open(host, port):
        print(f"{GREEN}[+] {host}:{port} is open      {RESET}")
    else:
        print(f"{GRAY}[!] {host}:{port} is closed    {RESET}", end="\r")

fast_port_scanner.py

import argparse
import socket # for connecting
from colorama import init, Fore
 
from threading import Thread, Lock
from queue import Queue
 
# some colors
init()
GREEN = Fore.GREEN
RESET = Fore.RESET
GRAY = Fore.LIGHTBLACK_EX

# number of threads, feel free to tune this parameter as you wish
N_THREADS = 200
# thread queue
q = Queue()
print_lock = Lock()
 
def port_scan(port):
    """
    Scan a port on the global variable `host`
    """
    try:
        s = socket.socket()
        s.connect((host, port))
    except:
        with print_lock:
            print(f"{GRAY}{host:15}:{port:5} is closed  {RESET}", end='\r')
    else:
        with print_lock:
            print(f"{GREEN}{host:15}:{port:5} is open    {RESET}")
    finally:
        s.close()
 
def scan_thread():
    global q
    while True:
        # get the port number from the queue
        worker = q.get()
        # scan that port number
        port_scan(worker)
        # tells the queue that the scanning for that port 
        # is done
        q.task_done()
 
def main(host, ports):
    global q
    for t in range(N_THREADS):
        # for each thread, start it
        t = Thread(target=scan_thread)
        # when we set daemon to true, that thread will end when the main thread ends
        t.daemon = True
        # start the daemon thread
        t.start()
 
    for worker in ports:
        # for each port, put that port into the queue
        # to start scanning
        q.put(worker)
     
    # wait the threads ( port scanners ) to finish
    q.join()
 
if __name__ == "__main__":
    # parse some parameters passed
    parser = argparse.ArgumentParser(description="Simple port scanner")
    parser.add_argument("host", help="Host to scan.")
    parser.add_argument("--ports", "-p", dest="port_range", default="1-65535", help="Port range to scan, default is 1-65535 (all ports)")
    args = parser.parse_args()
    host, port_range = args.host, args.port_range
 
    start_port, end_port = port_range.split("-")
    start_port, end_port = int(start_port), int(end_port)
 
    ports = [ p for p in range(start_port, end_port)]
 
    main(host, ports)