-
Notifications
You must be signed in to change notification settings - Fork 109
/
threads.py
182 lines (146 loc) · 5.65 KB
/
threads.py
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
181
182
# -*- coding: utf-8 -*-
from PySide6.QtCore import QThread, QThreadPool, QRunnable, QObject, Signal
from utils import test_ip, check_ip
from ipaddress import IPv4Network, IPv6Network
import random
__all__ = ['SpeedtestThread', 'ScanThread']
class _SpeedtestTaskSignals(QObject):
foundAvailable = Signal(str, float) # foundAvailable(ip, seconds)
foundUnavailable = Signal(str, str) # foundUnavailable(ip, reason)
class _SpeedtestTask(QRunnable):
def __init__(self, ip, timeout=2.5, repeat=3):
super().__init__()
self.ip = ip
self.timeout = timeout
self.repeat = repeat
self.signals = _SpeedtestTaskSignals()
def run(self):
total = 0
for _ in range(self.repeat):
result = test_ip(self.ip, self.timeout)
if isinstance(result, str):
self.signals.foundUnavailable.emit(self.ip, result)
return
total += result
average = total / self.repeat
self.signals.foundAvailable.emit(self.ip, average)
class SpeedtestThread(QThread):
def __init__(self, parent, ips, available_callback, unavailable_callback,
timeout=2.5, repeat=3, num_workers=8):
super().__init__(parent)
self.ips = ips
self.available_callback = available_callback
self.unavailable_callback = unavailable_callback
self.timeout = timeout
self.repeat = repeat
self.num_workers = num_workers
def run(self):
pool = QThreadPool()
pool.setMaxThreadCount(self.num_workers)
for ip in self.ips:
task = _SpeedtestTask(ip, self.timeout, self.repeat)
task.signals.foundAvailable.connect(self.available_callback)
task.signals.foundUnavailable.connect(self.unavailable_callback)
pool.start(task)
self.msleep(30)
pool.waitForDone()
class _ScanTaskSignals(QObject):
foundAvailable = Signal(str)
class _ScanTask(QRunnable):
def __init__(self, ip, timeout):
super().__init__()
self.ip = ip
self.timeout = timeout
self.signals = _ScanTaskSignals()
def run(self):
if check_ip(self.ip, self.timeout):
self.signals.foundAvailable.emit(self.ip)
class ScanThread(QThread):
foundAvailable = Signal(str)
progressUpdate = Signal(int)
# https://repo.or.cz/gscan_quic.git/blob/89e4b91eb3642b12f7665f7a9f4fa33c403fc318:/iprange_gws_b.txt
net_default = IPv4Network('142.250.0.0/15')
ipv4_extend = [
IPv4Network('108.177.0.0/17'),
IPv4Network('172.217.0.0/16'),
IPv4Network('172.253.0.0/16'),
IPv4Network('216.58.192.0/19'),
IPv4Network('72.14.192.0/18'),
IPv4Network('74.125.0.0/16')
]
# https://repo.or.cz/gscan_quic.git/blob/89e4b91eb3642b12f7665f7a9f4fa33c403fc318:/iprange_gws_6_a.txt
ipv6_extend = [
IPv6Network('2404:6800:4008:c15::0/112'),
IPv6Network('2a00:1450:4001:802::0/112'),
IPv6Network('2a00:1450:4001:803::0/112')
]
def __init__(self, parent, max_ips=5, num_workers=80, timeout=2.5,
enableOptimization=True, extend4=False, extend6=False, randomized=False):
super().__init__(parent)
self.max_ips = max_ips
self.num_workers = num_workers
self.timeout = timeout
self.networks = [
[ip for ip in self.net_default if int(ip) & 0xff == 90]
if enableOptimization else list(self.net_default)
]
if extend4:
self.networks.extend(map(list, self.ipv4_extend))
if extend6:
self.networks.extend(map(list, self.ipv6_extend))
self.currentIndex = 0
self.block_size = max(64, num_workers)
if randomized:
for network in self.networks:
random.shuffle(network)
def __found_available(self, ip):
if self.counter >= self.max_ips:
return
self.counter += 1
self.foundAvailable.emit(ip)
if self.counter >= self.max_ips:
self.pool.clear()
def __add_block(self):
self.num_added = 0
for _ in range(self.block_size):
ok = False
for net in self.networks:
try:
ip = str(net[self.currentIndex])
except IndexError:
continue
ok = True
task = _ScanTask(ip, self.timeout)
task.signals.foundAvailable.connect(self.__found_available)
self.pool.start(task)
self.num_added += 1
if not ok:
return False
self.currentIndex += 1
return True
def cancel(self):
self.running = False
def run(self):
self.running = True
self.pool = QThreadPool()
self.pool.setMaxThreadCount(self.num_workers)
self.counter = 0
while self.running and self.counter < self.max_ips and self.__add_block():
self.pool.waitForDone()
self.progressUpdate.emit(self.num_added)
self.pool.waitForDone()
class DebugThread(QThread):
success = Signal(float) # success(seconds)
fail = Signal(str) # fail(reason)
def __init__(self, parent, ip, host, testip_format, timeout, repeat):
super().__init__(parent)
self.ip = ip
self.host = host
self.testip_format = testip_format
self.timeout = timeout
self.repeat = repeat
def run(self):
for _ in range(self.repeat):
result = test_ip(self.ip, self.timeout, self.host, self.testip_format)
signal = self.fail if isinstance(result, str) else self.success
signal.emit(result)