```python
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
网络端口扫描与安全检测系统
功能：多线程端口扫描、服务识别、弱口令检测、报告生成
技术栈：socket, threading, [Link], json, datetime
作者：AI 助手
版本：3.0
"""
import socket
import threading
import time
import json
import re
from datetime import datetime
from typing import List, Dict, Optional, Tuple
from [Link] import ThreadPoolExecutor, as_completed
from dataclasses import dataclass, asdict
from collections import defaultdict
import ipaddress
# ==================== 配置常量 ====================
COMMON_PORTS = {
21: "FTP", 22: "SSH", 23: "Telnet", 25: "SMTP",
53: "DNS", 80: "HTTP", 110: "POP3", 111: "RPC",
135: "MSRPC", 139: "NetBIOS", 143: "IMAP", 443: "HTTPS",
445: "SMB", 993: "IMAPS", 995: "POP3S", 1433: "MSSQL",
1521: "Oracle", 3306: "MySQL", 3389: "RDP", 5432: "PostgreSQL",
5900: "VNC", 6379: "Redis", 27017: "MongoDB"
}
WEAK_PASSWORDS = [
"admin", "root", "password", "123456", "admin123",
"root123", "password123", "12345678", "adminadmin",
"rootroot", "test", "test123", "qwerty", "abc123"
]
DEFAULT_TIMEOUT = 2.0
MAX_WORKERS = 100
REPORT_DIR = "scan_reports"
# ==================== 数据模型 ====================
@dataclass
class PortInfo:
"""端口信息"""
port: int
service: str
state: str # open, closed, filtered
banner: str = ""
version: str = ""
vulnerability: List[str] = None
def __post_init__(self):
if [Link] is None:
[Link] = []
def to_dict(self) -> Dict:
return asdict(self)
@dataclass
class ScanResult:
"""扫描结果"""
target: str
start_time: str
end_time: str
total_ports: int
open_ports: List[PortInfo]
closed_count: int = 0
filtered_count: int = 0
def to_dict(self) -> Dict:
return {
"target": [Link],
"start_time": self.start_time,
"end_time": self.end_time,
"total_ports": self.total_ports,
"open_ports": [p.to_dict() for p in self.open_ports],
"closed_count": self.closed_count,
"filtered_count": self.filtered_count
}
# ==================== 端口扫描器核心 ====================
class PortScanner:
"""端口扫描器"""
def __init__(self, target: str, timeout: float = DEFAULT_TIMEOUT):
[Link] = target
[Link] = timeout
[Link] = ScanResult(
target=target,
start_time=[Link]().isoformat(),
end_time="",
total_ports=0,
open_ports=[]
)
self._lock = [Link]()
self._progress = 0
def _is_ip_valid(self) -> bool:
"""验证 IP 地址格式"""
try:
ipaddress.ip_address([Link])
return True
except ValueError:
return False
def _resolve_hostname(self) -> Optional[str]:
"""解析主机名"""
try:
return [Link]([Link])
except [Link]:
return None
def _connect_port(self, port: int) -> Tuple[int, bool, str]:
"""
尝试连接端口
返回: (端口, 是否开放, 服务标识)
"""
try:
sock = [Link](socket.AF_INET, socket.SOCK_STREAM)
[Link]([Link])
start = [Link]()
result = sock.connect_ex(([Link], port))
elapsed = [Link]() - start
if result == 0:
# 端口开放，尝试获取 banner
banner = self._get_banner(sock, port)
[Link]()
return (port, True, banner)
else:
[Link]()
return (port, False, "")
except [Link]:
return (port, False, "")
except Exception:
return (port, False, "")
def _get_banner(self, sock: [Link], port: int, timeout: float = 2.0) -> str:
"""获取服务 banner"""
try:
[Link](timeout)
# 根据端口发送不同的探测请求
probes = {
21: "USER anonymous\r\n",
22: "",
23: "\r\n",
25: "EHLO test\r\n",
80: "HEAD / HTTP/1.0\r\n\r\n",
443: "",
3306: "\x00\x00\x00\x01\xff\x15\x04\x00\x00\x00\x00",
6379: "PING\r\n"
}
probe = [Link](port, "\r\n")
if probe:
[Link]([Link]())
# 读取响应
banner = [Link](1024).decode('utf-8', errors='ignore')
banner = [Link]().replace('\r\n', ' ').replace('\n', ' ')
# 清理并截断
banner = [Link](r'\s+', ' ', banner)[:500]
return banner
except Exception:
return ""
def _identify_service(self, port: int, banner: str) -> Tuple[str, str]:
"""识别服务和版本"""
service = COMMON_PORTS.get(port, "unknown")
version = ""
# 基于 banner 的版本识别
if service == "SSH" and banner:
version_match = [Link](r'SSH-([0-9.]+)', banner)
if version_match:
version = version_match.group(1)
elif service == "HTTP" and banner:
version_match = [Link](r'Server:\s*([^\s]+)', banner)
if version_match:
version = version_match.group(1)
elif service == "FTP" and banner:
version_match = [Link](r'FTP[\s-]+([^\s]+)', banner)
if version_match:
version = version_match.group(1)
elif service == "MySQL" and banner:
version_match = [Link](r'mysql[\s-]+([0-9.]+)', banner, re.I)
if version_match:
version = version_match.group(1)
return service, version
def _check_vulnerabilities(self, port: int, service: str, banner: str) -> List[str]:
"""简易漏洞