1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
| import json
import socket
import threading
from concurrent.futures import ThreadPoolExecutor, as_completed
class Plugin:
def __init__(self):
self.name = "端口扫描器"
self.description = "TCP端口扫描工具"
self.version = "1.0.0"
self.author = "CTFTool Team"
self.initialized = False
self.running = False
def Init(self):
"""初始化插件"""
self.initialized = True
return True
def Run(self, context_json):
"""执行端口扫描"""
context = json.loads(context_json)
target = context.get("target", "127.0.0.1")
scan_type = context.get("scan_type", "common")
timeout = context.get("timeout", 1)
threads = context.get("threads", 50)
self.running = True
open_ports = []
try:
if scan_type == "common":
ports = [21, 22, 23, 25, 53, 80, 81, 110, 135, 139, 143, 443, 445, 3306, 3389, 5900, 8080]
else:
ports = range(1, 1001)
with ThreadPoolExecutor(max_workers=threads) as executor:
futures = {executor.submit(self.scan_port, target, port, timeout): port for port in ports}
for future in as_completed(futures):
if not self.running:
executor.shutdown(wait=False, cancel_futures=True)
break
port = futures[future]
try:
result = future.result()
if result:
service = self.get_service_name(port)
open_ports.append({"port": port, "status": "open", "service": service})
except Exception as e:
continue
return json.dumps({
"success": True,
"target": target,
"scan_type": scan_type,
"open_ports": open_ports,
"total_open": len(open_ports)
})
except Exception as e:
return json.dumps({
"success": False,
"error": str(e)
})
finally:
self.running = False
def scan_port(self, target, port, timeout):
"""扫描单个端口"""
if not self.running:
return False
try:
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
s.settimeout(timeout)
result = s.connect_ex((target, port))
return result == 0
except:
return False
def get_service_name(self, port):
"""获取端口常见服务名称"""
common_services = {
21: "FTP",
22: "SSH",
23: "Telnet",
25: "SMTP",
53: "DNS",
80: "HTTP",
110: "POP3",
135: "MSRPC",
139: "NetBIOS",
143: "IMAP",
443: "HTTPS",
445: "SMB",
3306: "MySQL",
3389: "RDP",
5900: "VNC",
8080: "HTTP-Proxy"
}
return common_services.get(port, "Unknown")
def GetUI(self):
"""返回UI定义"""
return {
"type": "panel",
"title": "端口扫描器",
"controls": [
{
"type": "textbox",
"name": "target",
"label": "目标IP",
"default": "127.0.0.1"
},
{
"type": "radio",
"name": "scan_type",
"label": "扫描类型",
"options": [
{"id": "common", "label": "常见端口", "selected": True},
{"id": "full", "label": "全端口(1-1000)"}
]
},
{
"type": "number",
"name": "timeout",
"label": "超时时间(秒)",
"default": 1,
"min": 0.1,
"max": 10,
"step": 0.1
},
{
"type": "number",
"name": "threads",
"label": "线程数",
"default": 50,
"min": 10,
"max": 200,
"step": 10
},
{
"type": "button",
"name": "start_scan",
"label": "开始扫描"
},
{
"type": "button",
"name": "stop_scan",
"label": "停止扫描",
"disabled": true
},
{
"type": "progress",
"name": "scan_progress",
"label": "扫描进度"
},
{
"type": "table",
"name": "results_table",
"label": "扫描结果",
"columns": [
{"name": "port", "label": "端口"},
{"name": "status", "label": "状态"},
{"name": "service", "label": "服务"}
]
}
]
}
def Dispose(self):
"""停止扫描并清理资源"""
self.running = False
self.initialized = False
return True
|
