gru/snmp_skaner
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Update scan.py

Browse Source
This commit is contained in:
gru 2025-05-06 15:08:39 +02:00
parent abaea01174
commit e1d031e4d7
1 changed files with 162 additions and 80 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

240
scan.py
View File

@ -1,4 +1,3 @@
import ipaddress import ipaddress
import subprocess import subprocess
import argparse import argparse
@ -6,53 +5,96 @@ import csv
import json import json
import time import time
import threading import threading
import itertools import sys
from concurrent.futures import ThreadPoolExecutor, as_completed from concurrent.futures import ThreadPoolExecutor, as_completed
default_oids = { default_oids = {
"sysName": "1.3.6.1.2.1.1.5.0", "sysName": "1.3.6.1.2.1.1.5.0"
#"sysDescr": "1.3.6.1.2.1.1.1.0",
#"sysLocation": "1.3.6.1.2.1.1.6.0",
#"sysContact": "1.3.6.1.2.1.1.4.0"
} }
def get_oid_value(ip, community, oid, timeout=1): class Scanner:
try: def __init__(self):
result = subprocess.run( self.stop_event = threading.Event()
["snmpget", "-v2c", "-c", community, str(ip), oid], self.lock = threading.Lock()
capture_output=True, text=True, timeout=timeout self.counter = [0]
) self.results = []
if result.returncode == 0 and "STRING" in result.stdout: self.last_ip = None
return result.stdout.split("STRING:")[-1].strip() self.errors = [0]
except subprocess.TimeoutExpired:
return "Timeout"
return "Brak danych"
def query_all_oids(ip, community, oids, timeout=1): def get_oid_value(self, ip, community, oid, timeout=1):
values = {name: get_oid_value(ip, community, oid, timeout) for name, oid in oids.items()} if self.stop_event.is_set():
has_data = any(val not in ("Timeout", "Brak danych") for val in values.values()) return "Przerwano"
if has_data: try:
return (str(ip), values) result = subprocess.run(
return None ["snmpget", "-v2c", "-c", community, str(ip), oid],
capture_output=True, text=True, timeout=timeout
)
if result.returncode == 0 and "STRING" in result.stdout:
return result.stdout.split("STRING:")[-1].strip()
else:
with self.lock:
self.errors[0] += 1
except subprocess.TimeoutExpired:
with self.lock:
self.errors[0] += 1
return "Timeout"
except Exception:
with self.lock:
self.errors[0] += 1
return "Błąd"
return "Brak danych"
def scan_subnet(subnet, community, oids, counter, total, lock): def query_all_oids(self, ip, community, oids, timeout=1):
results = [] if self.stop_event.is_set():
for ip in subnet.hosts(): return None
result = query_all_oids(ip, community, oids, timeout=1) values = {name: self.get_oid_value(ip, community, oid, timeout) for name, oid in oids.items()}
if result: has_data = any(val not in ("Timeout", "Brak danych", "Błąd", "Przerwano") for val in values.values())
results.append(result) if has_data:
with lock: return (str(ip), values)
counter[0] += 1 return None
print(f"\rSkanowanie IP: {counter[0]}/{total}", end="", flush=True)
return results
def split_subnet(supernet, new_prefix=24): def scan_subnet(self, subnet, community, oids, oids_order, live_mode):
try: for ip in subnet.hosts():
network = ipaddress.IPv4Network(supernet) if self.stop_event.is_set():
return list(network.subnets(new_prefix=new_prefix)) return
except ValueError as e: result = self.query_all_oids(ip, community, oids)
print(f"Błąd podsieci: {e}") with self.lock:
return [] self.last_ip = str(ip)
self.counter[0] += 1
if result:
self.results.append(result)
if live_mode:
ip_str, values = result
row = f"{ip_str:<15} " + " ".join([f"{values.get(k, ''):<25}" for k in oids_order])
sys.stdout.write("\r" + " " * 120 + "\r")
sys.stdout.write(row + "\n")
sys.stdout.flush()
def show_progress(self, total):
start_time = time.time()
while not self.stop_event.is_set():
with self.lock:
done = self.counter[0]
last_ip = self.last_ip or "-"
errors = self.errors[0]
elapsed = time.time() - start_time
speed = done / elapsed if elapsed > 0 else 0
avg_time_per_host = elapsed / done if done else 0
eta = (total - done) / speed if speed > 0 else float("inf")
eta_str = time.strftime("%H:%M:%S", time.gmtime(eta)) if eta != float("inf") else "N/A"
sys.stdout.write(
f"\rPostęp: {done}/{total} ({(done/total)*100:.2f}%)"
f" | {speed:.2f} hostów/s"
f" | Śr. czas/hosta: {avg_time_per_host:.3f} s"
f" | Błędy: {errors}"
f" | Ostatni IP: {last_ip}"
f" | ETA: {eta_str}"
)
sys.stdout.flush()
time.sleep(0.1)
sys.stdout.write("\n")
def read_subnets_from_file(filename): def read_subnets_from_file(filename):
subnets = [] subnets = []
@ -63,9 +105,17 @@ def read_subnets_from_file(filename):
if line: if line:
subnets.append(line) subnets.append(line)
except Exception as e: except Exception as e:
print(f"Błąd podczas wczytywania pliku podsieci: {e}") print(f"Błąd podczas wczytywania pliku: {e}")
return subnets return subnets
def split_subnet(supernet, new_prefix=24):
try:
network = ipaddress.IPv4Network(supernet)
return list(network.subnets(new_prefix=new_prefix))
except ValueError as e:
print(f"Błąd podsieci: {e}")
return []
def save_to_csv(results, oids, filename="wyniki.csv"): def save_to_csv(results, oids, filename="wyniki.csv"):
with open(filename, "w", newline="", encoding="utf-8") as f: with open(filename, "w", newline="", encoding="utf-8") as f:
writer = csv.writer(f) writer = csv.writer(f)
@ -75,55 +125,87 @@ def save_to_csv(results, oids, filename="wyniki.csv"):
row = [ip] + [values.get(key, "") for key in oids.keys()] row = [ip] + [values.get(key, "") for key in oids.keys()]
writer.writerow(row) writer.writerow(row)
def sort_results(results, key):
return sorted(results, key=lambda x: x[1].get(key, "").lower())
def main(): def main():
parser = argparse.ArgumentParser(description="Skaner SNMP z podziałem podsieci i eksportem do CSV.") parser = argparse.ArgumentParser(description="SNMP skaner https://gitea.linuxiarz.pl/gru/snmp_skaner")
parser.add_argument("subnets", nargs="*", help="Podsieci w formacie CIDR (np. 172.16.0.0/16)") parser.add_argument("subnets", nargs="*", help="Podsieci w formacie CIDR (np. 192.168.1.0/24)")
parser.add_argument("-s", "--subnet-file", help="Plik tekstowy z listą podsieci CIDR (jedna na linię)") parser.add_argument("-s", "--subnet-file", help="Plik z listą podsieci (jedna na linię)")
parser.add_argument("-c", "--community", default="public", help="SNMP community (domyślnie: public)") parser.add_argument("-c", "--community", default="public", help="SNMP community (domyślnie: public)")
parser.add_argument("-o", "--oids", help="OID-y w formacie JSON, np. '{\"sysDescr\":\"1.3.6.1.2.1.1.1.0\"}'") parser.add_argument("-o", "--oids", help="OID-y w formacie JSON, np. '{\"sysName\":\"...\"}'")
parser.add_argument("-w", "--workers", type=int, default=10, help="Liczba równoległych wątków (domyślnie: 10)") parser.add_argument("-w", "--workers", type=int, default=10, help="Liczba wątków (domyślnie: 10)")
parser.add_argument("-p", "--prefix", type=int, default=24, help="Wielkość podsieci do podziału (domyślnie: 24)") parser.add_argument("-p", "--prefix", type=int, default=24, help="Prefix dla podsieci (domyślnie: 24)")
parser.add_argument("-f", "--file", default="wyniki.csv", help="Nazwa pliku CSV do zapisu (domyślnie: wyniki.csv)") parser.add_argument("-f", "--file", default="wyniki.csv", help="Plik CSV wyników")
parser.add_argument("--live", action="store_true", help="Tryb wypisywania wyników na żywo")
parser.add_argument("--sort", help="Pole OID do sortowania wyników (np. sysName)")
args = parser.parse_args() args = parser.parse_args()
oids = json.loads(args.oids) if args.oids else default_oids scanner = Scanner()
all_input_subnets = [] try:
if args.subnet_file: oids = json.loads(args.oids) if args.oids else default_oids
all_input_subnets.extend(read_subnets_from_file(args.subnet_file)) oids_order = list(oids.keys())
if args.subnets:
all_input_subnets.extend(args.subnets)
if not all_input_subnets: all_subnets = []
print("Brak podsieci do przeskanowania (ani z linii poleceń, ani z pliku).") if args.subnet_file:
return all_subnets += read_subnets_from_file(args.subnet_file)
if args.subnets:
all_subnets += args.subnets
if not all_subnets:
print("Brak podsieci do skanowania.")
return
subnets_to_scan = [] subnets_to_scan = []
for subnet in all_input_subnets: for subnet in all_subnets:
subnets_to_scan.extend(split_subnet(subnet, new_prefix=args.prefix)) subnets_to_scan.extend(split_subnet(subnet, args.prefix))
# Oblicz łączną liczbę hostów total_ips = sum(1 for subnet in subnets_to_scan for _ in subnet.hosts())
total_ips = sum(1 for subnet in subnets_to_scan for _ in subnet.hosts())
counter = [0]
lock = threading.Lock()
print(f"{'IP':<15} " + " ".join([f"{name:<25}" for name in oids.keys()])) if args.live:
print("-" * (15 + 26 * len(oids))) print(f"\n{'IP':<15} " + " ".join([f"{name:<25}" for name in oids_order]))
print("-" * (15 + 26 * len(oids_order)))
all_results = [] progress_thread = threading.Thread(target=scanner.show_progress, args=(total_ips,))
with ThreadPoolExecutor(max_workers=args.workers) as executor: progress_thread.daemon = True
futures = { progress_thread.start()
executor.submit(scan_subnet, subnet, args.community, oids, counter, total_ips, lock): subnet
for subnet in subnets_to_scan
}
for future in as_completed(futures):
subnet_results = future.result()
for ip_str, values in subnet_results:
all_results.append((ip_str, values))
print(f"\n{ip_str:<15} " + " ".join([f"{values[name]:<25}" for name in oids.keys()]))
print("\nSkanowanie zakończone.") with ThreadPoolExecutor(max_workers=args.workers) as executor:
save_to_csv(all_results, oids, filename=args.file) futures = []
for subnet in subnets_to_scan:
if scanner.stop_event.is_set():
break
futures.append(executor.submit(
scanner.scan_subnet, subnet, args.community, oids, oids_order, args.live
))
try:
for future in as_completed(futures):
future.result()
except KeyboardInterrupt:
scanner.stop_event.set()
print("\nPrzerwano przez użytkownika. Zapisuję dotychczasowe wyniki...")
if not args.live:
if args.sort and args.sort in oids:
scanner.results = sort_results(scanner.results, args.sort)
print(f"\n{'IP':<15} " + " ".join([f"{name:<25}" for name in oids_order]))
print("-" * (15 + 26 * len(oids_order)))
for ip, values in scanner.results:
print(f"{ip:<15} " + " ".join([f"{values.get(k, ''):<25}" for k in oids_order]))
save_to_csv(scanner.results, oids, filename=args.file)
print(f"\nZapisano do pliku: {args.file}")
except KeyboardInterrupt:
scanner.stop_event.set()
print("\nPrzerwano przez użytkownika. Zapisuję dotychczasowe wyniki...")
save_to_csv(scanner.results, oids, filename=args.file)
except Exception as e:
scanner.stop_event.set()
print(f"\nWystąpił błąd: {str(e)}")
save_to_csv(scanner.results, oids, filename=args.file)
if __name__ == "__main__": if __name__ == "__main__":
main() main()