python 端口扫描

import socket
import threading
from datetime import datetime
import argparse
import sys

class PortScanner:
    def __init__(self, target, ports, timeout=1, threads=100):
        self.target = target
        self.ports = ports
        self.timeout = timeout
        self.threads = threads
        self.open_ports = []
        self.lock = threading.Lock()
        self.completed_ports = 0
        self.start_time = None
        self.total_ports = ports[1] - ports[0] + 1
        
    def scan_port(self, port):
        try:
            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            sock.settimeout(self.timeout)
            result = sock.connect_ex((self.target, port))
            if result == 0:
                try:
                    service = socket.getservbyport(port)
                except:
                    service = "unknown"
                with self.lock:
                    self.open_ports.append((port, service))
                    print(f"\n[+] 端口 {port} 开放 - {service}")
                    self._update_progress()
            sock.close()
        except Exception as e:
            pass
        finally:
            with self.lock:
                self.completed_ports += 1
                if self.completed_ports % 50 == 0 or self.completed_ports == self.total_ports:
                    self._update_progress()
    
    def _update_progress(self):
        if self.start_time is None:
            return
        
        elapsed = datetime.now() - self.start_time
        elapsed_seconds = elapsed.total_seconds()
        
        if self.completed_ports > 0:
            progress = (self.completed_ports / self.total_ports) * 100
            rate = self.completed_ports / elapsed_seconds
            remaining_seconds = (self.total_ports - self.completed_ports) / rate
            estimated_total = elapsed_seconds + remaining_seconds
        else:
            progress = 0
            remaining_seconds = 0
            estimated_total = 0
        
        bar_length = 40
        filled_length = int(bar_length * progress / 100)
        bar = '█' * filled_length + '░' * (bar_length - filled_length)
        
        elapsed_str = self._format_time(elapsed_seconds)
        remaining_str = self._format_time(remaining_seconds)
        total_str = self._format_time(estimated_total)
        
        status = f"\r[{bar}] {progress:.1f}% | 已扫描: {self.completed_ports}/{self.total_ports} | " \
                 f"已用: {elapsed_str} | 剩余: {remaining_str} | 预计: {total_str}"
        sys.stdout.write(status)
        sys.stdout.flush()
    
    def _format_time(self, seconds):
        if seconds < 60:
            return f"{seconds:.1f}秒"
        elif seconds < 3600:
            minutes = int(seconds // 60)
            secs = int(seconds % 60)
            return f"{minutes}分{secs}秒"
        else:
            hours = int(seconds // 3600)
            minutes = int((seconds % 3600) // 60)
            secs = int(seconds % 60)
            return f"{hours}时{minutes}分{secs}秒"
    
    def scan(self):
        print(f"\n开始扫描目标: {self.target}")
        print(f"扫描端口范围: {self.ports[0]}-{self.ports[1]}")
        print(f"线程数: {self.threads}")
        print(f"超时时间: {self.timeout}秒")
        print(f"开始时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
        print("-" * 60)
        
        self.start_time = datetime.now()
        
        threads = []
        for port in range(self.ports[0], self.ports[1] + 1):
            t = threading.Thread(target=self.scan_port, args=(port,))
            threads.append(t)
            t.start()
            
            if len(threads) >= self.threads:
                for thread in threads:
                    thread.join()
                threads = []
        
        for thread in threads:
            thread.join()
        
        print("\n")
        print("-" * 60)
        end_time = datetime.now()
        total_time = (end_time - self.start_time).total_seconds()
        print(f"扫描完成时间: {end_time.strftime('%Y-%m-%d %H:%M:%S')}")
        print(f"总耗时: {self._format_time(total_time)}")
        print(f"\n发现 {len(self.open_ports)} 个开放端口:")
        self.open_ports.sort(key=lambda x: x[0])
        for port, service in self.open_ports:
            print(f"  端口 {port}: {service}")
        
        return self.open_ports

def main():
    parser = argparse.ArgumentParser(description='端口扫描工具')
    parser.add_argument('target', help='目标IP地址或域名')
    parser.add_argument('-p', '--ports', default='1-1024', 
                        help='端口范围，格式: start-end (默认: 1-1024)')
    parser.add_argument('-t', '--timeout', type=int, default=1,
                        help='连接超时时间(秒) (默认: 1)')
    parser.add_argument('-T', '--threads', type=int, default=100,
                        help='线程数 (默认: 100)')
    
    args = parser.parse_args()
    
    try:
        port_range = args.ports.split('-')
        start_port = int(port_range[0])
        end_port = int(port_range[1])
        
        if start_port < 1 or end_port > 65535 or start_port > end_port:
            print("错误: 端口范围无效 (1-65535)")
            sys.exit(1)
            
    except:
        print("错误: 端口范围格式不正确，应为 start-end")
        sys.exit(1)
    
    try:
        target_ip = socket.gethostbyname(args.target)
    except socket.gaierror:
        print(f"错误: 无法解析主机名 {args.target}")
        sys.exit(1)
    
    scanner = PortScanner(
        target=target_ip,
        ports=(start_port, end_port),
        timeout=args.timeout,
        threads=args.threads
    )
    
    scanner.scan()

if __name__ == "__main__":
    main()

# 扫描指定端口范围
python main.py 192.168.1.1 -p 1-65535

# 设置超时时间 (秒)
python main.py 192.168.1.1 -t 2

# 设置线程数
python main.py 192.168.1.1 -T 200

# 组合使用
python main.py example.com -p 1-10000 -t 2 -T 150