Связано с моим первоначальным вопросом здесь: Python3 threading, пытаясь пинговать несколько IP-адресов/тестовых портов одновременно
По сути, я расширил первоначальные требования и сделал это объектно-ориентированным. Итак, у меня есть класс Client с именем клиента, местоположением и сетевым cidr. Мой сценарий port_test.py запрашивает базу данных для клиентов, инициирует клиентский объект и вызывает его метод display_results, который определяет ThreadPool и вызывает scan_network, который сканирует диапазоны IP-адресов для определенного порта, определенного в моем файле конфигурации.
Я думаю, что это близко к работе, но я получаю эту ошибку при вызове range(len(all_hosts)) (см. код ниже): TypeError: object of type 'IPv4Address' has no len()
Я понимаю ошибку, я просто не понимаю, почему я ее получаю. Это работало до того, как я создал класс, и я не понимаю, почему это не работает сейчас.
Какие-нибудь мысли?
EDIT: см. мой ответ для рабочего кода
port_test.py:
#! /usr/bin/python
# Import modules
import mysql.connector
import logging
import configparser
import coloredlogs
from Client import Client
#logging.basicConfig(format='%(levelname)s:%(message)s',level=logging.INFO)
coloredlogs.install(level='DEBUG')
coloredlogs.DEFAULT_LOG_FORMAT = '%(asctime)s -- %(message)s'
# read from the config file
config = configparser.ConfigParser()
config.read('config.ini')
db=config['mysql']
net=config['network']
port = int(net['port'])
# create the connection, connect, and setup the query
cnx = mysql.connector.connect(host=db['host'], user=db['user'], database=db['database'], password=db['password'])
cursor = cnx.cursor()
query = ("select c.client_name as name, l.location_name as location, fw.net_cidr as cidr "
"from firewalls fw "
"left join clients c on c.id = fw.client_id "
"left join locations l on l.id = fw.location_id "
"where fw.net_cidr <> '' and c.active = '1' and fw.active = '1'")
cursor.execute(query)
results = cursor.fetchall()
for (name, location, cidr) in results:
client_obj = Client(name, location, cidr)
net_addr = str(cidr)
client_obj.display_results(net_addr)
# close the database connection
cursor.close()
cnx.close()
Клиент.py:
#! /usr/bin/python3
# Client.py
import logging
from multiprocessing.dummy import Pool as ThreadPool
import configparser
import coloredlogs
import ipaddress
import socket
import subprocess
# read from the config file
config = configparser.ConfigParser()
config.read('config.ini')
net=config['network']
port = int(net['port'])
class Client:
def __init__(self, name, location, net_cidr):
self.name = name
self.location = location
self.net_cidr = net_cidr
def display_results(self, net_addr):
# execute and display the results
logging.info("{} -> {} ---> {} ".format(self.name, self.location, self.net_cidr))
try:
ip_net = ipaddress.ip_network(net_addr)
# Get all hosts on that network
all_hosts = list(ip_net.hosts())
# make a pool of workers
pool = ThreadPool(16)
# ping the cidrs in their own thread
pool.map(self.scan_network, all_hosts)
pool.close()
pool.join()
except ValueError as e:
logging.warning(e)
def scan_network(self, all_hosts):
logging.info("List Length: " + str(len(all_hosts)))
try:
for i in range(len(all_hosts)):
# For each IP address in the subnet, test to see if port 3389 is open
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(.25)
result = sock.connect_ex((str(all_hosts[i]),port))
if result == 0:
logging.info(str(all_hosts[i]) + ": " + net['port'] + " is open")
except TypeError as e:
logging.critical(e)
pass