import socket

# 創(chuàng)建socket對(duì)象
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# 設(shè)置超時(shí)時(shí)間
sock.settimeout(5)
try:
    # 嘗試連接
    result = sock.connect_ex((host, port))
    if result == 0:
        print(f"端口 {port} 開放")
    else:
        print(f"端口 {port} 關(guān)閉")
except socket.error as e:
    print(f"掃描端口 {port} 時(shí)出錯(cuò): {e}")
finally:
    sock.close()

from concurrent.futures import ThreadPoolExecutor, as_completed

def scan_ports(host, ports, max_threads=100):
    open_ports = []
    with ThreadPoolExecutor(max_workers=max_threads) as executor:
        # 提交掃描任務(wù)
        future_to_port = {
            executor.submit(scan_port, host, port, timeout): port 
            for port in ports
        }
        
        # 收集結(jié)果
        for future in as_completed(future_to_port):
            port = future_to_port[future]
            try:
                result = future.result()
                if result:
                    open_ports.append(port)
            except Exception as e:
                print(f"端口 {port} 掃描異常: {e}")
    
    return sorted(open_ports)

#!/usr/bin/env python3
"""
簡(jiǎn)單的多線程端口掃描器
作者: 網(wǎng)絡(luò)安全愛好者
版本: 1.0
"""

import socket
import argparse
import threading
from concurrent.futures import ThreadPoolExecutor, as_completed
import time
import sys
from typing import List, Tuple, Union

class PortScanner:
    """
    端口掃描器類
    實(shí)現(xiàn)多線程TCP Connect掃描功能
    """
    
    def __init__(self, timeout: float = 1.0, max_threads: int = 100):
        """
        初始化端口掃描器
        
        Args:
            timeout: 連接超時(shí)時(shí)間（秒）
            max_threads: 最大線程數(shù)
        """
        self.timeout = timeout
        self.max_threads = max_threads
        self.lock = threading.Lock()
        self.scan_results = []
        
    def resolve_host(self, host: str) -> str:
        """
        解析主機(jī)名或IP地址
        
        Args:
            host: 主機(jī)名或IP地址
            
        Returns:
            解析后的IP地址
            
        Raises:
            socket.gaierror: 當(dāng)主機(jī)名無法解析時(shí)
        """
        try:
            # 嘗試解析主機(jī)名
            ip = socket.gethostbyname(host)
            return ip
        except socket.gaierror as e:
            raise socket.gaierror(f"無法解析主機(jī): {host}, 錯(cuò)誤: {e}")
    
    def scan_port(self, host: str, port: int) -> Tuple[int, bool, str]:
        """
        掃描單個(gè)端口
        
        Args:
            host: 目標(biāo)主機(jī)
            port: 目標(biāo)端口
            
        Returns:
            (端口, 是否開放, 服務(wù)名稱)
        """
        try:
            # 創(chuàng)建socket對(duì)象
            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            sock.settimeout(self.timeout)
            
            # 嘗試連接
            result = sock.connect_ex((host, port))
            
            if result == 0:
                # 端口開放，嘗試獲取服務(wù)名稱
                try:
                    service = socket.getservbyport(port, 'tcp')
                except OSError:
                    service = "unknown"
                return (port, True, service)
            else:
                return (port, False, "")
                
        except socket.timeout:
            return (port, False, "timeout")
        except socket.error as e:
            return (port, False, f"error: {e}")
        finally:
            try:
                sock.close()
            except:
                pass
    
    def parse_port_range(self, port_range: str) -> List[int]:
        """
        解析端口范圍字符串
        
        Args:
            port_range: 端口范圍字符串，如 "80", "1-100", "22,80,443"
            
        Returns:
            端口列表
            
        Raises:
            ValueError: 當(dāng)端口范圍格式錯(cuò)誤時(shí)
        """
        ports = []
        
        # 處理逗號(hào)分隔的端口列表
        if ',' in port_range:
            parts = port_range.split(',')
            for part in parts:
                if '-' in part:
                    # 處理范圍
                    start, end = part.split('-')
                    try:
                        start_port = int(start.strip())
                        end_port = int(end.strip())
                        if 1 <= start_port <= end_port <= 65535:
                            ports.extend(range(start_port, end_port + 1))
                        else:
                            raise ValueError(f"端口范圍 {part} 無效")
                    except ValueError as e:
                        raise ValueError(f"無效的端口范圍: {part}, 錯(cuò)誤: {e}")
                else:
                    # 處理單個(gè)端口
                    try:
                        port = int(part.strip())
                        if 1 <= port <= 65535:
                            ports.append(port)
                        else:
                            raise ValueError(f"端口 {port} 超出范圍")
                    except ValueError as e:
                        raise ValueError(f"無效的端口: {part}, 錯(cuò)誤: {e}")
        
        # 處理端口范圍
        elif '-' in port_range:
            start, end = port_range.split('-')
            try:
                start_port = int(start.strip())
                end_port = int(end.strip())
                if 1 <= start_port <= end_port <= 65535:
                    ports = list(range(start_port, end_port + 1))
                else:
                    raise ValueError(f"端口范圍 {port_range} 無效")
            except ValueError as e:
                raise ValueError(f"無效的端口范圍: {port_range}, 錯(cuò)誤: {e}")
        
        # 處理單個(gè)端口
        else:
            try:
                port = int(port_range)
                if 1 <= port <= 65535:
                    ports = [port]
                else:
                    raise ValueError(f"端口 {port} 超出范圍")
            except ValueError as e:
                raise ValueError(f"無效的端口: {port_range}, 錯(cuò)誤: {e}")
        
        return sorted(set(ports))  # 去重并排序
    
    def scan_ports(self, host: str, ports: Union[str, List[int]], 
                   show_closed: bool = False) -> List[Tuple[int, bool, str]]:
        """
        掃描指定主機(jī)的端口
        
        Args:
            host: 目標(biāo)主機(jī)
            ports: 端口范圍或列表
            show_closed: 是否顯示關(guān)閉的端口
            
        Returns:
            掃描結(jié)果列表，每個(gè)元素為 (端口, 是否開放, 服務(wù)名稱)
        """
        # 解析主機(jī)名
        try:
            ip = self.resolve_host(host)
            print(f"掃描主機(jī): {host} ({ip})")
        except socket.gaierror as e:
            print(f"錯(cuò)誤: {e}")
            return []
        
        # 解析端口
        if isinstance(ports, str):
            try:
                port_list = self.parse_port_range(ports)
            except ValueError as e:
                print(f"錯(cuò)誤: {e}")
                return []
        else:
            port_list = ports
        
        print(f"掃描端口數(shù)量: {len(port_list)}")
        print(f"使用線程數(shù): {self.max_threads}")
        print(f"超時(shí)時(shí)間: {self.timeout}秒")
        print("-" * 50)
        
        start_time = time.time()
        open_count = 0
        
        # 使用線程池執(zhí)行掃描
        with ThreadPoolExecutor(max_workers=self.max_threads) as executor:
            # 提交所有掃描任務(wù)
            future_to_port = {
                executor.submit(self.scan_port, ip, port): port 
                for port in port_list
            }
            
            # 處理完成的任務(wù)
            for future in as_completed(future_to_port):
                port = future_to_port[future]
                try:
                    result = future.result()
                    port_num, is_open, service = result
                    
                    with self.lock:
                        if is_open:
                            print(f"端口 {port_num:5d} 開放 - 服務(wù): {service}")
                            open_count += 1
                            self.scan_results.append(result)
                        elif show_closed:
                            print(f"端口 {port_num:5d} 關(guān)閉")
                            self.scan_results.append(result)
                            
                except Exception as e:
                    print(f"掃描端口 {port} 時(shí)發(fā)生異常: {e}")
        
        # 輸出統(tǒng)計(jì)信息
        elapsed_time = time.time() - start_time
        print("-" * 50)
        print(f"掃描完成!")
        print(f"開放端口數(shù)量: {open_count}/{len(port_list)}")
        print(f"掃描耗時(shí): {elapsed_time:.2f} 秒")
        
        return self.scan_results

def main():
    """主函數(shù)"""
    parser = argparse.ArgumentParser(
        description="簡(jiǎn)單的多線程端口掃描器",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
使用示例:
  %(prog)s 192.168.1.1 80
  %(prog)s example.com 1-100
  %(prog)s localhost 22,80,443,8000-9000
  %(prog)s 10.0.0.1 1-1000 -t 5 -T 2.0 -v
        """
    )
    
    parser.add_argument("host", help="要掃描的目標(biāo)主機(jī)名或IP地址")
    parser.add_argument("ports", help="要掃描的端口范圍，如: 80, 1-100, 22,80,443")
    parser.add_argument("-t", "--threads", type=int, default=100,
                       help="最大線程數(shù) (默認(rèn): 100)")
    parser.add_argument("-T", "--timeout", type=float, default=1.0,
                       help="連接超時(shí)時(shí)間，單位秒 (默認(rèn): 1.0)")
    parser.add_argument("-v", "--verbose", action="store_true",
                       help="顯示所有端口（包括關(guān)閉的端口）")
    
    args = parser.parse_args()
    
    # 創(chuàng)建掃描器實(shí)例
    scanner = PortScanner(timeout=args.timeout, max_threads=args.threads)
    
    try:
        # 執(zhí)行掃描
        results = scanner.scan_ports(args.host, args.ports, args.verbose)
        
        # 輸出開放端口摘要
        if results:
            open_ports = [(port, service) for port, is_open, service in results if is_open]
            if open_ports:
                print("\n開放端口摘要:")
                for port, service in sorted(open_ports):
                    print(f"  {port}/tcp - {service}")
            else:
                print("\n未發(fā)現(xiàn)開放端口")
                
    except KeyboardInterrupt:
        print("\n掃描被用戶中斷")
        sys.exit(1)
    except Exception as e:
        print(f"掃描過程中發(fā)生錯(cuò)誤: {e}")
        sys.exit(1)

if __name__ == "__main__":
    main()

# 掃描單個(gè)端口
python port_scanner.py example.com 80

# 掃描端口范圍
python port_scanner.py 192.168.1.1 1-100

# 掃描多個(gè)指定端口
python port_scanner.py localhost 22,80,443,8000

# 使用更多線程和更長(zhǎng)超時(shí)時(shí)間
python port_scanner.py target.com 1-1000 -t 200 -T 2.0

# 顯示所有端口（包括關(guān)閉的）
python port_scanner.py 10.0.0.1 20-30 -v

def adaptive_timeout(self, host: str, sample_ports: List[int] = [80, 443, 22]) -> float:
    """
    根據(jù)樣本端口響應(yīng)時(shí)間計(jì)算合適的超時(shí)時(shí)間
    
    Args:
        host: 目標(biāo)主機(jī)
        sample_ports: 用于測(cè)試的樣本端口
        
    Returns:
        計(jì)算得到的超時(shí)時(shí)間
    """
    response_times = []
    
    for port in sample_ports:
        try:
            start = time.time()
            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            sock.settimeout(2.0)
            result = sock.connect_ex((host, port))
            elapsed = time.time() - start
            
            if result == 0:  # 端口開放
                response_times.append(elapsed)
            sock.close()
        except:
            pass
    
    if response_times:
        avg_time = sum(response_times) / len(response_times)
        return min(max(avg_time * 3, 0.5), 5.0)  # 限制在0.5-5秒之間
    else:
        return 2.0  # 默認(rèn)超時(shí)時(shí)間

class ConnectionPool:
    """連接池管理類"""
    
    def __init__(self, max_size=10):
        self.max_size = max_size
        self.pool = []
        self.lock = threading.Lock()
    
    def get_connection(self, host, port, timeout):
        """從池中獲取或創(chuàng)建連接"""
        with self.lock:
            if self.pool:
                sock = self.pool.pop()
                try:
                    # 測(cè)試連接是否仍然有效
                    sock.getsockopt(socket.SOL_SOCKET, socket.SO_ERROR)
                    return sock
                except socket.error:
                    sock.close()
            
        # 創(chuàng)建新連接
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.settimeout(timeout)
        return sock
    
    def return_connection(self, sock):
        """將連接返回池中"""
        with self.lock:
            if len(self.pool) < self.max_size:
                self.pool.append(sock)
            else:
                sock.close()

import random

def randomize_scan_order(self, ports: List[int]) -> List[int]:
    """
    隨機(jī)化端口掃描順序
    
    Args:
        ports: 原始端口列表
        
    Returns:
        隨機(jī)排序后的端口列表
    """
    randomized_ports = ports.copy()
    random.shuffle(randomized_ports)
    return randomized_ports

import time

class RateLimiter:
    """速率限制器"""
    
    def __init__(self, max_requests_per_second):
        self.max_requests = max_requests_per_second
        self.interval = 1.0 / max_requests_per_second
        self.last_request_time = 0
        self.lock = threading.Lock()
    
    def acquire(self):
        """獲取執(zhí)行許可"""
        with self.lock:
            current_time = time.time()
            elapsed = current_time - self.last_request_time
            
            if elapsed < self.interval:
                time.sleep(self.interval - elapsed)
            
            self.last_request_time = time.time()

import unittest
from unittest.mock import patch, MagicMock

class TestPortScanner(unittest.TestCase):
    """端口掃描器測(cè)試類"""
    
    def setUp(self):
        self.scanner = PortScanner(timeout=0.1, max_threads=10)
    
    def test_resolve_host_valid(self):
        """測(cè)試有效主機(jī)名解析"""
        ip = self.scanner.resolve_host("localhost")
        self.assertEqual(ip, "127.0.0.1")
    
    def test_resolve_host_invalid(self):
        """測(cè)試無效主機(jī)名解析"""
        with self.assertRaises(socket.gaierror):
            self.scanner.resolve_host("invalid-hostname-that-should-not-exist")
    
    def test_parse_port_range_single(self):
        """測(cè)試單個(gè)端口解析"""
        ports = self.scanner.parse_port_range("80")
        self.assertEqual(ports, [80])
    
    def test_parse_port_range_range(self):
        """測(cè)試端口范圍解析"""
        ports = self.scanner.parse_port_range("1-3")
        self.assertEqual(ports, [1, 2, 3])
    
    def test_parse_port_range_mixed(self):
        """測(cè)試混合端口解析"""
        ports = self.scanner.parse_port_range("80,443,1-3")
        self.assertEqual(ports, [1, 2, 3, 80, 443])
    
    @patch('socket.socket')
    def test_scan_port_open(self, mock_socket):
        """測(cè)試開放端口掃描"""
        mock_sock = MagicMock()
        mock_socket.return_value = mock_sock
        mock_sock.connect_ex.return_value = 0
        
        port, is_open, service = self.scanner.scan_port("localhost", 80)
        
        self.assertEqual(port, 80)
        self.assertTrue(is_open)
        mock_sock.connect_ex.assert_called_with(("localhost", 80))

if __name__ == '__main__':
    unittest.main()

def integration_test():
    """集成測(cè)試函數(shù)"""
    scanner = PortScanner(timeout=1.0, max_threads=50)
    
    # 測(cè)試本地主機(jī)
    print("測(cè)試本地主機(jī)掃描...")
    results = scanner.scan_ports("localhost", "80,443,22,8080")
    
    # 驗(yàn)證結(jié)果
    open_ports = [port for port, is_open, service in results if is_open]
    print(f"發(fā)現(xiàn)的開放端口: {open_ports}")
    
    # 預(yù)期至少本地回環(huán)地址應(yīng)該有某些端口開放
    assert len(open_ports) >= 0, "應(yīng)該能夠掃描本地主機(jī)"

if __name__ == "__main__":
    integration_test()