Python端口检测案例如何检测端口状态

wen python案例 2026-06-29 220

本文目录导读：

基础Socket检测
批量端口扫描
使用高级库 - python-nmap
使用socket的完整检测类
简单的HTTP端口检测
注意事项

我来为您介绍几种Python端口检测的方法：

基础Socket检测

import socket
def check_port(host, port, timeout=2):
    """检测端口是否开放"""
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.settimeout(timeout)
    try:
        result = sock.connect_ex((host, port))
        if result == 0:
            return True, f"端口 {port} 开放"
        else:
            return False, f"端口 {port} 关闭"
    except socket.gaierror:
        return False, "主机名解析失败"
    except socket.error:
        return False, "连接错误"
    finally:
        sock.close()
# 使用示例
host = "192.168.1.1"  # 或 "google.com"
port = 80
status, message = check_port(host, port)
print(f"状态: {status}, 信息: {message}")

批量端口扫描

import socket
from concurrent.futures import ThreadPoolExecutor
def scan_single_port(host, port, timeout=1):
    """扫描单个端口"""
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.settimeout(timeout)
    try:
        result = sock.connect_ex((host, port))
        if result == 0:
            return port, True
        return port, False
    except:
        return port, False
    finally:
        sock.close()
def scan_ports(host, start_port, end_port, max_workers=50):
    """扫描端口范围"""
    open_ports = []
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = [executor.submit(scan_single_port, host, port) 
                  for port in range(start_port, end_port + 1)]
        for future in futures:
            port, is_open = future.result()
            if is_open:
                open_ports.append(port)
    return open_ports
# 使用示例
host = "192.168.1.1"
open_ports = scan_ports(host, 20, 100)
print(f"开放端口: {open_ports}")

使用高级库 - python-nmap

pip install python-nmap

import nmap
def nmap_scan(host, ports="1-1000"):
    """使用nmap进行端口扫描"""
    nm = nmap.PortScanner()
    try:
        nm.scan(host, ports, arguments="-sS -T4")
        results = {}
        if host in nm.all_hosts():
            for proto in nm[host].all_protocols():
                ports = nm[host][proto].keys()
                for port in ports:
                    state = nm[host][proto][port]['state']
                    results[port] = state
        return results
    except nmap.PortScannerError:
        return {"error": "nmap未安装或执行错误"}
# 使用示例
host = "scanme.nmap.org"
results = nmap_scan(host, "22,80,443,8080")
print(f"扫描结果: {results}")

使用socket的完整检测类

import socket
import threading
from queue import Queue
import time
class PortScanner:
    def __init__(self, target_host, timeout=1, max_threads=100):
        self.target_host = target_host
        self.timeout = timeout
        self.max_threads = max_threads
        self.open_ports = []
        self.port_queue = Queue()
    def check_single_port(self, port):
        """检测单个端口"""
        try:
            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            sock.settimeout(self.timeout)
            result = sock.connect_ex((self.target_host, port))
            if result == 0:
                service = self.get_service_name(port)
                self.open_ports.append({
                    'port': port,
                    'service': service,
                    'status': 'open'
                })
            sock.close()
        except Exception as e:
            pass
    def get_service_name(self, port):
        """获取端口对应的服务名"""
        try:
            return socket.getservbyport(port)
        except:
            return "unknown"
    def worker(self):
        """工作线程"""
        while True:
            try:
                port = self.port_queue.get_nowait()
                self.check_single_port(port)
                self.port_queue.task_done()
                # 进度显示
                if port % 100 == 0:
                    print(f"检测进度: 端口 {port}")
            except Queue.Empty:
                break
    def scan(self, start_port=1, end_port=1024):
        """执行扫描"""
        print(f"开始扫描 {self.target_host} 端口 {start_port}-{end_port}")
        start_time = time.time()
        # 填充队列
        for port in range(start_port, end_port + 1):
            self.port_queue.put(port)
        # 创建线程
        threads = []
        for _ in range(min(self.max_threads, end_port - start_port + 1)):
            t = threading.Thread(target=self.worker)
            t.start()
            threads.append(t)
        # 等待所有线程完成
        for t in threads:
            t.join()
        # 等待队列处理完成
        self.port_queue.join()
        elapsed_time = time.time() - start_time
        print(f"扫描完成，耗时: {elapsed_time:.2f}秒")
        return self.open_ports
# 使用示例
host = "192.168.1.1"  # 替换为目标主机
scanner = PortScanner(host)
open_ports = scanner.scan(1, 100)
for port_info in open_ports:
    print(f"端口 {port_info['port']} ({port_info['service']}): {port_info['status']}")

简单的HTTP端口检测

import requests
def check_http_port(host, port, timeout=2):
    """检测HTTP/HTTPS端口"""
    protocols = ['http', 'https']
    for protocol in protocols:
        url = f"{protocol}://{host}:{port}"
        try:
            response = requests.get(url, timeout=timeout)
            return True, f"{protocol.upper()}端口开放，状态码: {response.status_code}"
        except requests.ConnectionError:
            continue
        except requests.Timeout:
            continue
    return False, "HTTP/HTTPS端口不可用"
# 使用示例
host = "example.com"
port = 443
status, message = check_http_port(host, port)
print(f"状态: {status}, 信息: {message}")