Код: Выделить всё
def check_open_port(host, port):
s = socket.socket()
s.settimeout(0.1)
# the SO_REUSEADDR flag tells the kernel to reuse a local
# socket in TIME_WAIT state, without waiting for its natural
# timeout to expire.
s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
try:
code = s.connect_ex((host, port))
s.close()
if code == 0:
return True
else:
return False
except socket.error:
return False
def get_open_ports(host, max_port=65535):
open_ports = []
def worker(port):
if check_open_port(host, port):
open_ports.append(port)
pool = ThreadPoolExecutor(max_workers=10000)
[pool.submit(worker, port) for port in range(1, max_port + 1)]
pool.shutdown(wait=True)
return open_ports
Код: Выделить всё
[22, 80]Код: Выделить всё
[22, 80, 443]Код: Выделить всё
[22]Я экспериментировал со значениями max_workers и settimeout(), но не смог заставить их работать нормально. Единственный раз, когда это сработало, было без использования потоков, но, очевидно, это заняло много времени, мне нужно их использовать.
Я что-то упустил? Есть ли другой способ реализовать это?
Мобильная версия