💻 源代码
import socket
import requests
from concurrent.futures import ThreadPoolExecutor
def scan_port(ip, port):
"""扫描单个端口"""
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(1)
result = sock.connect_ex((ip, port))
sock.close()
return port if result == 0 else None
def scan_ports(ip, ports):
"""批量扫描端口"""
with ThreadPoolExecutor(max_workers=100) as executor:
results = list(executor.map(lambda p: scan_port(ip, p), ports))
return [r for r in results if r]
def scan_subdomain(domain):
"""子域名探测"""
subdomains = ['www', 'mail', 'ftp', 'blog', 'dev']
found = []
for sub in subdomains:
url = f'http://{sub}.{domain}'
try:
r = requests.get(url, timeout=3)
if r.status_code < 500:
found.append(url)
except:
pass
return found
def scan_directory(url):
"""目录扫描"""
dirs = ['admin', 'backup', 'config', 'upload', 'api']
found = []
for d in dirs:
r = requests.get(f'{url}/{d}/', timeout=3)
if r.status_code == 200:
found.append(f'{url}/{d}/')
return found
# 示例
open_ports = scan_ports('127.0.0.1', range(1, 1000))
print(f"开放端口: {open_ports}")