我找到了这个network scanner in python article 并写了这个短代码。它做你想做的!但是,您确实需要知道设备的可访问端口。端口 22 是 ssh 标准和我正在使用的。我想你可以遍历所有端口。一些默认值是:
linux: [20, 21, 22, 23, 25, 80, 111, 443, 445, 631, 993, 995]
windows: [135, 137, 138, 139, 445]
mac: [22, 445, 548, 631]
import socket
def connect(hostname, port):
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
socket.setdefaulttimeout(1)
result = sock.connect_ex((hostname, port))
sock.close()
return result == 0
for i in range(0,255):
res = connect("192.168.1."+str(i), 22)
if res:
print("Device found at: ", "192.168.1."+str(i) + ":"+str(22))
编辑 by TheLizzard:
使用上面的代码并添加线程:
from threading import Thread, Lock
from time import perf_counter
from sys import stderr
from time import sleep
import socket
# I changed this from "192.168.1.%i" to "192.168.0.%i"
BASE_IP = "192.168.0.%i"
PORT = 80
class Threader:
"""
This is a class that calls a list of functions in a limited number of
threads. It uses locks to make sure the data is thread safe.
Usage:
from time import sleep
def function(i):
sleep(2)
with threader.print_lock:
print(i)
threader = Threader(10) # The maximum number of threads = 10
for i in range(20):
threader.append(function, i)
threader.start()
threader.join()
This class also provides a lock called: `<Threader>.print_lock`
"""
def __init__(self, threads=30):
self.thread_lock = Lock()
self.functions_lock = Lock()
self.functions = []
self.threads = []
self.nthreads = threads
self.running = True
self.print_lock = Lock()
def stop(self) -> None:
# Signal all worker threads to stop
self.running = False
def append(self, function, *args) -> None:
# Add the function to a list of functions to be run
self.functions.append((function, args))
def start(self) -> None:
# Create a limited number of threads
for i in range(self.nthreads):
thread = Thread(target=self.worker, daemon=True)
# We need to pass in `thread` as a parameter so we
# have to use `<threading.Thread>._args` like this:
thread._args = (thread, )
self.threads.append(thread)
thread.start()
def join(self) -> None:
# Joins the threads one by one until all of them are done.
for thread in self.threads:
thread.join()
def worker(self, thread:Thread) -> None:
# While we are running and there are functions to call:
while self.running and (len(self.functions) > 0):
# Get a function
with self.functions_lock:
function, args = self.functions.pop(0)
# Call that function
function(*args)
# Remove the thread from the list of threads.
# This may cause issues if the user calls `<Threader>.join()`
# But I haven't seen this problem while testing/using it.
with self.thread_lock:
self.threads.remove(thread)
start = perf_counter()
# I didn't need a timeout of 1 so I used 0.1
socket.setdefaulttimeout(0.1)
def connect(hostname, port):
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
result = sock.connect_ex((hostname, port))
with threader.print_lock:
if result == 0:
stderr.write(f"[{perf_counter() - start:.5f}] Found {hostname}\n")
threader = Threader(10)
for i in range(255):
threader.append(connect, BASE_IP%i, PORT)
threader.start()
threader.join()
print(f"[{perf_counter() - start:.5f}] Done searching")
input("Press enter to exit.\n? ")