Python實現(xiàn)快速掃描目標主機的開放端口和服務
更新時間：2025年12月04日 15:45:39 作者：零日失眠者
這篇文章主要為大家詳細介紹了如何使用Python編寫一個功能強大的端口掃描器腳本,實現(xiàn)快速掃描目標主機的開放端口和服務,感興趣的小伙伴可以了解下
功能介紹
這是一個功能強大的端口掃描器腳本，能夠快速掃描目標主機的開放端口和服務。該腳本具備以下核心功能：
- 多種掃描模式：支持TCP連接掃描、SYN掃描、UDP掃描等多種掃描方式
- 端口范圍自定義：支持掃描單個端口、端口范圍或常見端口列表
- 服務識別：識別常見端口對應的服務類型和版本信息
- 并發(fā)掃描：使用多線程技術提高掃描效率
- 詳細報告生成：生成詳細的掃描報告，包括開放端口、服務信息、響應時間等
- 結果導出：支持將掃描結果導出為JSON、CSV、XML等多種格式
- 靈活配置：支持自定義超時時間、并發(fā)線程數(shù)、掃描延遲等參數(shù)
- 安全防護：內(nèi)置掃描速率限制，避免對目標系統(tǒng)造成過大壓力
場景應用
1. 網(wǎng)絡安全審計
- 識別網(wǎng)絡設備和服務器的開放端口
- 發(fā)現(xiàn)潛在的安全漏洞和未授權服務
- 驗證防火墻規(guī)則的有效性
- 為滲透測試提供基礎信息
2. 系統(tǒng)管理維護
- 檢查服務器端口配置是否符合安全策略
- 確認服務是否正常運行在指定端口
- 排查網(wǎng)絡連接問題
- 監(jiān)控端口變化情況
3. 網(wǎng)絡故障排查
- 診斷網(wǎng)絡連通性問題
- 確定服務是否可達
- 分析端口阻塞原因
- 驗證網(wǎng)絡設備狀態(tài)
4. 合規(guī)性檢查
- 滿足等保測評中的端口檢查要求
- 符合行業(yè)安全規(guī)范
- 提供合規(guī)性審計證據(jù)
- 支持安全管理制度落實
報錯處理
1. 網(wǎng)絡連接異常
try:
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(timeout)
result = sock.connect_ex((host, port))
sock.close()
except socket.gaierror:
logger.error(f"無法解析主機名: {host}")
except socket.timeout:
logger.warning(f"連接 {host}:{port} 超時")
except ConnectionRefusedError:
logger.info(f"端口 {port} 被拒絕連接")
except PermissionError:
logger.error(f"無權限掃描端口 {port}，可能需要管理員權限")
2. 參數(shù)驗證錯誤
if not is_valid_ip(target):
raise ValueError(f"無效的IP地址: {target}")
if port < 1 or port > 65535:
raise ValueError(f"端口號超出有效范圍: {port}")
3. 文件操作異常
try:
with open(output_file, 'w') as f:
json.dump(scan_results, f, indent=2)
except PermissionError:
logger.error(f"無權限寫入文件: {output_file}")
except IOError as e:
logger.error(f"文件寫入錯誤: {str(e)}")
4. 內(nèi)存不足異常
try:
# 大規(guī)模掃描時的內(nèi)存管理
if len(ports) > 10000:
logger.warning("掃描端口數(shù)量過多，可能消耗大量內(nèi)存")
except MemoryError:
logger.error("內(nèi)存不足，無法完成掃描")
sys.exit(1)
代碼實現(xiàn)
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
端口掃描器
功能：掃描目標主機的開放端口和服務
作者：Cline
版本：1.0
"""
import socket
import argparse
import sys
import json
import csv
import xml.etree.ElementTree as ET
from datetime import datetime
import threading
from concurrent.futures import ThreadPoolExecutor, as_completed
import time
import logging
import ipaddress
# 配置日志
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('port_scanner.log'),
logging.StreamHandler(sys.stdout)
]
)
logger = logging.getLogger(__name__)
# 常見端口和服務映射
COMMON_PORTS = {
21: 'FTP',
22: 'SSH',
23: 'Telnet',
25: 'SMTP',
53: 'DNS',
80: 'HTTP',
110: 'POP3',
143: 'IMAP',
443: 'HTTPS',
993: 'IMAPS',
995: 'POP3S',
1433: 'MSSQL',
1521: 'Oracle',
3306: 'MySQL',
3389: 'RDP',
5432: 'PostgreSQL',
6379: 'Redis',
8080: 'HTTP-Alt',
8443: 'HTTPS-Alt',
27017: 'MongoDB'
}
class PortScanner:
def __init__(self, config):
self.target = config['target']
self.ports = config['ports']
self.scan_type = config.get('scan_type', 'tcp') # tcp, syn, udp
self.timeout = config.get('timeout', 1.0)
self.threads = config.get('threads', 100)
self.delay = config.get('delay', 0) # 掃描延遲（毫秒）
self.output_format = config.get('output_format', 'json')
self.output_file = config.get('output_file', 'scan_results.json')
self.verbose = config.get('verbose', False)
# 掃描結果
self.results = {
'target': self.target,
'scan_time': datetime.now().isoformat(),
'scan_type': self.scan_type,
'open_ports': [],
'closed_ports': [],
'filtered_ports': []
}
def is_valid_ip(self, ip):
"""驗證IP地址格式"""
try:
ipaddress.ip_address(ip)
return True
except ValueError:
return False
def resolve_hostname(self, hostname):
"""解析主機名到IP地址"""
try:
ip = socket.gethostbyname(hostname)
return ip
except socket.gaierror:
logger.error(f"無法解析主機名: {hostname}")
return None
def scan_port(self, port):
"""掃描單個端口"""
host = self.target
if not self.is_valid_ip(host):
host = self.resolve_hostname(host)
if not host:
return None
# 添加掃描延遲
if self.delay > 0:
time.sleep(self.delay / 1000.0)
try:
if self.scan_type == 'tcp':
return self.tcp_connect_scan(host, port)
elif self.scan_type == 'syn':
return self.syn_scan(host, port)
elif self.scan_type == 'udp':
return self.udp_scan(host, port)
else:
logger.error(f"不支持的掃描類型: {self.scan_type}")
return None
except Exception as e:
logger.error(f"掃描端口 {port} 時發(fā)生錯誤: {str(e)}")
return None
def tcp_connect_scan(self, host, port):
"""TCP連接掃描"""
try:
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(self.timeout)
start_time = time.time()
result = sock.connect_ex((host, port))
end_time = time.time()
response_time = round((end_time - start_time) * 1000, 2) # 毫秒
sock.close()
service = COMMON_PORTS.get(port, 'Unknown')
if result == 0:
# 端口開放，嘗試獲取服務banner
banner = self.get_service_banner(host, port)
return {
'port': port,
'state': 'open',
'service': service,
'response_time': response_time,
'banner': banner
}
else:
return {
'port': port,
'state': 'closed',
'service': service,
'response_time': response_time
}
except socket.timeout:
return {
'port': port,
'state': 'filtered',
'service': COMMON_PORTS.get(port, 'Unknown'),
'response_time': self.timeout * 1000
}
except Exception as e:
logger.debug(f"TCP掃描端口 {port} 時發(fā)生錯誤: {str(e)}")
return {
'port': port,
'state': 'unknown',
'service': COMMON_PORTS.get(port, 'Unknown')
}
def syn_scan(self, host, port):
"""SYN掃描（需要root權限）"""
try:
# 這里簡化實現(xiàn)，實際SYN掃描需要使用原始套接字
logger.warning("SYN掃描需要root權限，降級為TCP連接掃描")
return self.tcp_connect_scan(host, port)
except Exception as e:
logger.error(f"SYN掃描端口 {port} 時發(fā)生錯誤: {str(e)}")
return None
def udp_scan(self, host, port):
"""UDP掃描"""
try:
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
sock.settimeout(self.timeout)
# 發(fā)送空數(shù)據(jù)包
sock.sendto(b'', (host, port))
try:
# 嘗試接收響應
data, addr = sock.recvfrom(1024)
sock.close()
service = COMMON_PORTS.get(port, 'Unknown')
banner = data.decode('utf-8', errors='ignore')[:100]
return {
'port': port,
'state': 'open',
'service': service,
'banner': banner
}
except socket.timeout:
# UDP端口可能開放但無響應
sock.close()
return {
'port': port,
'state': 'open|filtered',
'service': COMMON_PORTS.get(port, 'Unknown')
}
except Exception as e:
logger.debug(f"UDP掃描端口 {port} 時發(fā)生錯誤: {str(e)}")
return {
'port': port,
'state': 'closed',
'service': COMMON_PORTS.get(port, 'Unknown')
}
def get_service_banner(self, host, port):
"""獲取服務banner信息"""
try:
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(2.0)
sock.connect((host, port))
# 發(fā)送簡單的探測數(shù)據(jù)
if port in [21, 22, 23, 25, 80, 110, 143]:
sock.send(b'\r\n')
# 接收響應
banner = sock.recv(1024).decode('utf-8', errors='ignore')
sock.close()
# 清理banner信息
banner = banner.strip().replace('\r\n', ' ').replace('\n', ' ')
return banner[:200] # 限制長度
except Exception as e:
logger.debug(f"獲取端口 {port} 的banner時發(fā)生錯誤: {str(e)}")
return ""
def scan_ports(self):
"""掃描所有端口"""
logger.info(f"開始掃描 {self.target} 的端口...")
logger.info(f"掃描類型: {self.scan_type}, 超時時間: {self.timeout}s, 線程數(shù): {self.threads}")
start_time = time.time()
with ThreadPoolExecutor(max_workers=self.threads) as executor:
# 提交所有掃描任務
future_to_port = {
executor.submit(self.scan_port, port): port
for port in self.ports
}
# 收集結果
for future in as_completed(future_to_port):
port = future_to_port[future]
try:
result = future.result()
if result:
if result['state'] == 'open':
self.results['open_ports'].append(result)
if self.verbose:
logger.info(f"端口 {result['port']} ({result['service']}) 開放")
elif result['state'] == 'closed':
self.results['closed_ports'].append(result)
elif result['state'] in ['filtered', 'open|filtered']:
self.results['filtered_ports'].append(result)
except Exception as e:
logger.error(f"處理端口 {port} 的結果時出錯: {str(e)}")
end_time = time.time()
self.results['duration'] = round(end_time - start_time, 2)
# 按端口號排序結果
self.results['open_ports'].sort(key=lambda x: x['port'])
self.results['closed_ports'].sort(key=lambda x: x['port'])
self.results['filtered_ports'].sort(key=lambda x: x['port'])
logger.info(f"掃描完成，耗時 {self.results['duration']} 秒")
logger.info(f"開放端口: {len(self.results['open_ports'])}")
logger.info(f"關閉端口: {len(self.results['closed_ports'])}")
logger.info(f"過濾端口: {len(self.results['filtered_ports'])}")
return self.results
def print_results(self):
"""打印掃描結果"""
print("\n" + "="*60)
print(f"端口掃描報告 - 目標: {self.target}")
print("="*60)
print(f"掃描時間: {self.results['scan_time']}")
print(f"掃描類型: {self.results['scan_type']}")
print(f"掃描耗時: {self.results['duration']} 秒")
print(f"開放端口: {len(self.results['open_ports'])}")
print(f"關閉端口: {len(self.results['closed_ports'])}")
print(f"過濾端口: {len(self.results['filtered_ports'])}")
if self.results['open_ports']:
print("\n開放端口詳情:")
print("-" * 80)
print(f"{'端口':<8} {'服務':<15} {'響應時間(ms)':<15} {'Banner信息'}")
print("-" * 80)
for port_info in self.results['open_ports']:
banner = port_info.get('banner', '')[:50] + ('...' if len(port_info.get('banner', '')) > 50 else '')
print(f"{port_info['port']:<8} {port_info['service']:<15} {port_info['response_time']:<15} {banner}")
if not self.results['open_ports']:
print("\n未發(fā)現(xiàn)開放端口")
def save_results(self):
"""保存掃描結果"""
try:
# 確保輸出目錄存在
import os
output_dir = os.path.dirname(self.output_file) if os.path.dirname(self.output_file) else '.'
os.makedirs(output_dir, exist_ok=True)
if self.output_format == 'json':
self._save_json()
elif self.output_format == 'csv':
self._save_csv()
elif self.output_format == 'xml':
self._save_xml()
else:
logger.error(f"不支持的輸出格式: {self.output_format}")
except Exception as e:
logger.error(f"保存掃描結果時出錯: {str(e)}")
def _save_json(self):
"""保存為JSON格式"""
with open(self.output_file, 'w', encoding='utf-8') as f:
json.dump(self.results, f, indent=2, ensure_ascii=False)
logger.info(f"掃描結果已保存到 {self.output_file}")
def _save_csv(self):
"""保存為CSV格式"""
all_ports = self.results['open_ports'] + self.results['closed_ports'] + self.results['filtered_ports']
with open(self.output_file, 'w', newline='', encoding='utf-8') as f:
writer = csv.writer(f)
writer.writerow(['Port', 'State', 'Service', 'Response Time (ms)', 'Banner'])
for port_info in all_ports:
writer.writerow([
port_info['port'],
port_info['state'],
port_info['service'],
port_info.get('response_time', ''),
port_info.get('banner', '')
])
logger.info(f"掃描結果已保存到 {self.output_file}")
def _save_xml(self):
"""保存為XML格式"""
root = ET.Element("port_scan")
root.set("target", self.target)
root.set("scan_time", self.results['scan_time'])
root.set("duration", str(self.results['duration']))
# 添加開放端口
open_ports_elem = ET.SubElement(root, "open_ports")
for port_info in self.results['open_ports']:
port_elem = ET.SubElement(open_ports_elem, "port")
port_elem.set("number", str(port_info['port']))
port_elem.set("service", port_info['service'])
port_elem.set("response_time", str(port_info.get('response_time', '')))
if port_info.get('banner'):
banner_elem = ET.SubElement(port_elem, "banner")
banner_elem.text = port_info['banner']
tree = ET.ElementTree(root)
tree.write(self.output_file, encoding='utf-8', xml_declaration=True)
logger.info(f"掃描結果已保存到 {self.output_file}")
def parse_ports(port_str):
"""解析端口參數(shù)"""
ports = set()
# 處理逗號分隔的端口列表
for part in port_str.split(','):
part = part.strip()
if '-' in part:
# 處理端口范圍
try:
start, end = map(int, part.split('-'))
if 1 <= start <= 65535 and 1 <= end <= 65535 and start <= end:
ports.update(range(start, end + 1))
else:
raise ValueError(f"無效的端口范圍: {part}")
except ValueError as e:
logger.error(f"解析端口范圍時出錯: {str(e)}")
sys.exit(1)
else:
# 處理單個端口
try:
port = int(part)
if 1 <= port <= 65535:
ports.add(port)
else:
raise ValueError(f"端口號超出有效范圍: {port}")
except ValueError as e:
logger.error(f"解析端口時出錯: {str(e)}")
sys.exit(1)
return sorted(list(ports))
def main():
parser = argparse.ArgumentParser(description='端口掃描器')
parser.add_argument('target', help='目標主機IP地址或域名')
parser.add_argument('-p', '--ports', default='1-1000', help='要掃描的端口，支持范圍(如1-1000)和列表(如22,80,443)')
parser.add_argument('-t', '--type', choices=['tcp', 'syn', 'udp'], default='tcp', help='掃描類型')
parser.add_argument('--timeout', type=float, default=1.0, help='連接超時時間(秒)')
parser.add_argument('--threads', type=int, default=100, help='并發(fā)線程數(shù)')
parser.add_argument('--delay', type=int, default=0, help='掃描延遲(毫秒)')
parser.add_argument('-o', '--output', help='輸出文件路徑')
parser.add_argument('-f', '--format', choices=['json', 'csv', 'xml'], default='json', help='輸出格式')
parser.add_argument('-v', '--verbose', action='store_true', help='詳細輸出')
parser.add_argument('--top-ports', type=int, help='掃描最常見的N個端口')
args = parser.parse_args()
# 解析端口
if args.top_ports:
# 使用最常見的端口
common_ports = sorted(COMMON_PORTS.keys())[:args.top_ports]
ports = common_ports
else:
ports = parse_ports(args.ports)
# 配置輸出文件
output_file = args.output
if not output_file:
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
output_file = f"scan_results_{timestamp}.{args.format}"
# 配置掃描參數(shù)
config = {
'target': args.target,
'ports': ports,
'scan_type': args.type,
'timeout': args.timeout,
'threads': args.threads,
'delay': args.delay,
'output_format': args.format,
'output_file': output_file,
'verbose': args.verbose
}
# 創(chuàng)建掃描器實例
scanner = PortScanner(config)
# 執(zhí)行掃描
try:
scanner.scan_ports()
scanner.print_results()
scanner.save_results()
except KeyboardInterrupt:
logger.info("掃描被用戶中斷")
sys.exit(1)
except Exception as e:
logger.error(f"掃描過程中發(fā)生錯誤: {str(e)}")
sys.exit(1)
if __name__ == '__main__':
main()
使用說明
1. 基本使用
# 掃描單個主機的默認端口范圍(1-1000) python port_scanner.py 192.168.1.1 # 掃描特定端口 python port_scanner.py 192.168.1.1 -p 22,80,443 # 掃描端口范圍 python port_scanner.py 192.168.1.1 -p 1-10000 # 掃描最常見的100個端口 python port_scanner.py 192.168.1.1 --top-ports 100
2. 掃描類型
# TCP連接掃描（默認） python port_scanner.py 192.168.1.1 -t tcp # UDP掃描 python port_scanner.py 192.168.1.1 -t udp # SYN掃描（需要root權限） sudo python port_scanner.py 192.168.1.1 -t syn
3. 性能調(diào)優(yōu)
# 調(diào)整超時時間 python port_scanner.py 192.168.1.1 --timeout 2.0 # 調(diào)整并發(fā)線程數(shù) python port_scanner.py 192.168.1.1 --threads 200 # 添加掃描延遲（毫秒） python port_scanner.py 192.168.1.1 --delay 100
4. 輸出格式
# JSON格式輸出（默認） python port_scanner.py 192.168.1.1 -o results.json # CSV格式輸出 python port_scanner.py 192.168.1.1 -f csv -o results.csv # XML格式輸出 python port_scanner.py 192.168.1.1 -f xml -o results.xml
5. 詳細輸出
# 顯示詳細掃描過程 python port_scanner.py 192.168.1.1 -v
高級特性
1. 批量掃描
可以通過腳本擴展實現(xiàn)批量掃描多個主機：
targets = ['192.168.1.1', '192.168.1.2', '192.168.1.3']
for target in targets:
config = {'target': target, 'ports': [22, 80, 443]}
scanner = PortScanner(config)
scanner.scan_ports()
scanner.save_results()
2. 定期掃描
可以結合cron定時任務實現(xiàn)定期掃描：
# 每小時掃描一次關鍵服務器 0 * * * * /usr/bin/python3 /path/to/port_scanner.py 192.168.1.1 -p 22,80,443 -o /var/log/scan_$(date +\%Y\%m\%d_\%H\%M\%S).json
3. 結果分析
掃描結果可以用于進一步分析：
import json
# 加載掃描結果
with open('scan_results.json', 'r') as f:
results = json.load(f)
# 分析開放端口
open_ports = results['open_ports']
for port in open_ports:
print(f"發(fā)現(xiàn)開放端口: {port['port']} ({port['service']})")
安全考慮
1. 權限控制
- 普通端口掃描不需要特殊權限
- SYN掃描需要root權限
- 應謹慎使用高權限運行腳本
2. 掃描頻率
- 避免過于頻繁的掃描
- 添加適當延遲保護目標系統(tǒng)
- 遵守網(wǎng)絡安全政策
3. 日志記錄
- 記錄所有掃描活動
- 保護敏感信息不被泄露
- 定期審查掃描日志
性能優(yōu)化
1. 并發(fā)控制
- 合理設置線程數(shù)避免系統(tǒng)過載
- 根據(jù)網(wǎng)絡狀況調(diào)整超時時間
- 使用連接池復用socket連接
2. 內(nèi)存管理
- 及時釋放不需要的資源
- 控制掃描規(guī)模避免內(nèi)存溢出
- 使用生成器處理大數(shù)據(jù)集
3. 網(wǎng)絡優(yōu)化
- 本地DNS緩存提高解析速度
- 復用連接減少握手開銷
- 批量處理減少網(wǎng)絡往返
這個端口掃描器是一個功能完整、安全可靠的網(wǎng)絡探測工具，能夠幫助網(wǎng)絡管理員和安全專業(yè)人員快速識別網(wǎng)絡資產(chǎn)和服務狀態(tài)。
以上就是Python實現(xiàn)快速掃描目標主機的開放端口和服務的詳細內(nèi)容，更多關于Python掃描端口的資料請關注腳本之家其它相關文章！
相關文章
這篇文章主要介紹了python自定義解析簡單xml格式文件的方法,涉及Python解析XML文件的相關技巧,非常具有實用價值,需要的朋友可以參考下2015-05-05
這篇文章主要用示例簡單介紹了Python下線程之間的共享和釋放,線程一直是Python開發(fā)者中的熱點問題,需要的朋友可以參考下2015-05-05
下面小編就為大家?guī)硪黄蒙善鱽砀膶懼苯臃祷亓斜淼暮瘮?shù)方法。小編覺得挺不錯的，現(xiàn)在就分享給大家，也給大家做個參考。一起跟隨小編過來看看吧2017-05-05
python 線程池順序執(zhí)行的方法實現(xiàn)
在Python中,線程池默認是并發(fā)執(zhí)行任務的,但若需要實現(xiàn)任務的順序執(zhí)行,文中通過示例代碼介紹的非常詳細,對大家的學習或者工作具有一定的參考學習價值,需要的朋友們下面隨著小編來一起學習學習吧2025-08-08
今天小編就為大家分享一篇PyQT5 emit 和 connect的用法詳解，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧2019-12-12
Python漏洞驗證程序Poc利用入門到實戰(zhàn)編寫
這篇文章主要為大家介紹了Python?Poc利用入門到實戰(zhàn)編寫實現(xiàn)示例，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進步2022-02-02
今天小編就為大家分享一篇在PyCharm下使用 ipython 交互式編程的方法，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧2019-01-01
在本篇文章中我們給大家整理了關于python為什么要設置環(huán)境變量的相關知識點，有興趣的朋友們參考下。2019-06-06
OpenCV實現(xiàn)從灰度圖像切出Mask前景區(qū)域
本文主要介紹了如何利用OpenCV實現(xiàn)從灰度圖像，根據(jù)閾值，切出多個前景區(qū)域，過濾面積太小的圖像。文中的示例代碼講解詳細，需要的可以參考一下2022-06-06
詳解PyQt5中textBrowser顯示print語句輸出的簡單方法
這篇文章主要介紹了詳解PyQt5中textBrowser顯示print語句輸出的簡單方法，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧2020-08-08