import nmap import socket import requests import paramiko import threading import argparse from datetime import datetime from bs4 import BeautifulSoup from urllib.parse import urljoin class VulnerabilityScanner: def __init__(self, target, ports=None, threads=10): """ 初始化漏洞扫描器 参数: target (str): 目标IP或域名 ports (str): 要扫描的端口范围，如 '1-1000' threads (int): 线程数 """ self.target = target self.ports = ports or '1-1000' self.threads = threads self.results = { 'target': target, 'start_time': datetime.now().isoformat(), 'open_ports': [], 'services': {}, 'vulnerabilities': [] } self.lock = threading.Lock() def validate_target(self): """验证目标地址是否有效""" try: socket.gethostbyname(self.target) return True except socket.gaierror: print(f"[!] 无法解析目标: {self.target}") return False def port_scan(self): """使用nmap进行端口扫描""" if not self.validate_target(): return False print(f"[*] 开始扫描 {self.target} 的端口 {self.ports}...") try: nm = nmap.PortScanner() nm.scan(hosts=self.target, ports=self.ports, arguments=f'-T4 --min-parallelism {self.threads}') for host in nm.all_hosts(): for proto in nm[host].all_protocols(): ports = nm[host][proto].keys() for port in sorted(ports): service = nm[host][proto][port]['name'] state = nm[host][proto][port]['state'] if state == 'open': with self.lock: self.results['open_ports'].append(port) self.results['services'][port] = { 'protocol': proto, 'service': service, 'product': nm[host][proto][port].get('product', ''), 'version': nm[host][proto][port].get('version', '') } print(f"[+] 发现开放端口: {port}/{proto} - {service}") return True except nmap.PortScannerError as e: print(f"[!] Nmap扫描错误: {e}") return False except Exception as e: print(f"[!] 扫描过程中发生错误: {e}") return False def check_weak_passwords(self, port, service): """检测常见服务的弱口令""" common_credentials = { 'ssh': [('root', 'root'), ('admin', 'admin'), ('root', 'password')], 'ftp': [('anonymous', ''), ('admin', 'admin')], 'telnet': [('root', 'root'), ('admin', 'admin')], 'mysql': [('root', ''), ('root', 'root')] } if service not in common_credentials: return print(f"[*] 正在检测 {service.upper()} 弱口令...") if service == 'ssh': ssh = paramiko.SSHClient() ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) for username, password in common_credentials[service]: try: ssh.connect(self.target, port=port, username=username, password=password, timeout=5) ssh.close() with self.lock: self.results['vulnerabilities'].append({ 'type': 'weak_password', 'port': port, 'service': service, 'credentials': f"{username}:{password}", 'severity': 'high' }) print(f"[!] 发现弱口令: {service} - {username}:{password}") break except: continue # 可以添加其他服务的弱口令检测逻辑 def check_sql_injection(self, url): """检测SQL注入漏洞""" if not url.startswith(('http://', 'https://')): url = f"http://{url}" print(f"[*] 正在检测 {url} 的SQL注入漏洞...") test_payloads = [ "'", "\"", "' OR '1'='1", "\" OR \"1\"=\"1", "' OR 1=1--", ] 怎么使用