Changed communication principes - removed CommunicationDictionary, now all keys is defined by annotations. Added new modules to scan Google docs and other urls.

This commit is contained in:
Zloooy 2019-07-05 17:26:25 +03:00
parent 77ca495fdf
commit e73965365e
27 changed files with 393 additions and 157 deletions

View File

@ -1,7 +1,7 @@
#modules selection #modules selection
config = { config = {
"parser" : "Parser", "parser" : "GDocsHashParser",
"address_generator" : "IpGenerator", "address_generator" : "GDocsAddressGenerator",
"scanner" : "CoreModel", "scanner" : "URLScanner",
"storage" : "JSONStorage" "storage" : "GDocsStorage"
} }

View File

@ -6,30 +6,42 @@ from PyQt5.Qt import QThread, pyqtSignal
from PyQt5.QtCore import QObject, pyqtSlot from PyQt5.QtCore import QObject, pyqtSlot
from config import config from config import config
from inspect import isfunction from inspect import isfunction
from communication.communication_utils import complitable_functions, get_converted_arguments
CoreModel = import_utils.import_class("modules/network_scan/%s.py" % CoreModel = import_utils.import_class("modules/network_scan/%s.py" %
config["scanner"]) config["scanner"])
Parser = import_utils.import_class("modules/address_generation/%s.py" % Parser = import_utils.import_class("modules/address_generation/%s.py" %
config["parser"]) config["parser"]
)
IpGenerator = import_utils.import_class( IpGenerator = import_utils.import_class(
"modules/address_generation/%s.py" % "modules/address_generation/%s.py" %
config["address_generator"] config["address_generator"]
) )
JSONStorage = import_utils.import_class("modules/storage/%s.py" % JSONStorage = import_utils.import_class("modules/storage/%s.py" %
config["storage"]) config["storage"])
convert_table = ConvertTable()
for func in import_utils.import_matching(
"modules/convert_functions/",
lambda name, value:
isfunction(value) and value.__annotations__
):
convert_table.add_function(func)
previous = IpGenerator.get_next_address
for function in [
CoreModel.scan_address,
JSONStorage.put_responce
]:
msg = "%s is complitable with %s"
if not complitable_functions(previous, function, convert_table):
msg = "%s is not complitable with %s"
print(msg % (function, previous))
previous = function
class MainPresenter: class MainPresenter:
def __init__(self, ui): def __init__(self, ui):
self.ui = ui self.ui = ui
self.threads = [] self.threads = []
self.isScanEnabled = False self.isScanEnabled = False
self.convert_table = ConvertTable()
for func in import_utils.import_matching(
"modules/convert_functions/",
lambda name, value:
isfunction(value) and hasattr(value, "__from__")
):
self.convert_table.add_function(func)
self.parser = Parser() self.parser = Parser()
#needed config to specify path #needed config to specify path
self.storage = JSONStorage("results.json") self.storage = JSONStorage("results.json")
@ -37,12 +49,33 @@ class MainPresenter:
def startScan(self, ipRanges, portsStr, threadNumber, timeout): def startScan(self, ipRanges, portsStr, threadNumber, timeout):
timeout = 3 if not timeout else int(timeout) timeout = 3 if not timeout else int(timeout)
addresses = self.parser.parse_address_field(ipRanges) addresses = None
ports = self.parser.parse_port_field(portsStr) parser_args = {'port_field':portsStr, 'address_field':ipRanges}
self.ip_generator = IpGenerator(addresses, ports, self.convert_table) fields = self.parser.parse_fields(
*get_converted_arguments(
self.parser.parse_fields,
parser_args,
convert_table
)
)
self.scanner = CoreModel(timeout) self.scanner = CoreModel(timeout)
if CoreModel.INDEPENDENT_THREAD_MANAGEMENT:
addresses = self.parser.get_all_addresses(ipRanges)
self.ip_generator = PlugAddressGenerator(addresses, ports)
threadNumber = 1
else:
self.ip_generator = IpGenerator()
self.ip_generator.set_parsed_fields(
*get_converted_arguments(
self.ip_generator.set_parsed_fields,
fields,
convert_table
)
)
threadNumber = int(threadNumber) threadNumber = int(threadNumber)
print("thread %i number set" % threadNumber)
for i in range(threadNumber): for i in range(threadNumber):
print(i)
scan_worker = ScanWorker( scan_worker = ScanWorker(
self.ip_generator, self.ip_generator,
self.scanner, self.scanner,
@ -55,10 +88,11 @@ class MainPresenter:
scan_worker.exit_signal.connect(self.on_worker_exit) scan_worker.exit_signal.connect(self.on_worker_exit)
scan_thread.started.connect(scan_worker.work) scan_thread.started.connect(scan_worker.work)
self.threads.append((scan_worker, scan_thread)) self.threads.append((scan_worker, scan_thread))
self.changeThreadLabel(threadNumber)
for thread in self.threads: for thread in self.threads:
scan_worker, scan_thread = thread scan_worker, scan_thread = thread
print("starting")
scan_thread.start() scan_thread.start()
self.changeThreadLabel(threadNumber)
def changeThreadLabel(self, number_of_threads): def changeThreadLabel(self, number_of_threads):
self.number_of_threads = number_of_threads self.number_of_threads = number_of_threads
@ -106,12 +140,33 @@ class ScanWorker(QObject):
@pyqtSlot() @pyqtSlot()
def work(self): def work(self):
while self.isRunning: while self.isRunning:
scan_address = self.ip_generator.get_next_address(self.previous_address) print("worker start")
scan_address = self.ip_generator.get_next_address(
*get_converted_arguments(
self.ip_generator.get_next_address,
self.previous_address,
convert_table
)
)
if not scan_address: if not scan_address:
break break
scan_result = self.scanner.scan_address(
*get_converted_arguments(
self.scanner.scan_address,
scan_address,
convert_table
)
)
print(scan_result)
scan_address.update(scan_result)
self.previous_address = scan_address self.previous_address = scan_address
scan_result = self.scanner.scan_address(scan_address) self.storage.put_responce(
self.storage.put_responce(scan_address, scan_result) *get_converted_arguments(
self.storage.put_responce,
scan_address,
convert_table
)
)
string_scan_address = " ".join(key + ":" + str(scan_address[key]) for string_scan_address = " ".join(key + ":" + str(scan_address[key]) for
key in scan_address.keys()) key in scan_address.keys())
if scan_result == 0: if scan_result == 0:

View File

@ -1,22 +0,0 @@
class CommunicationDictionary(dict):
'''This class is used to provide communication between classes using
key-value interface'''
def __init__(self, convert_table):
'''Convert table stores functions used to get value of unable key if
it's possible'''
super(CommunicationDictionary, self).__init__()
self.convert_table = convert_table
def __getitem__(self, key):
item = None
try:
item = dict.__getitem__(self,key)
except KeyError:
key_to_convert, convert_function = self.convert_table.get_converter(
self.keys(),
key
)
if key_to_convert is not None:
item = convert_function(dict.__getitem__(self,key_to_convert))
return item

View File

@ -1,13 +1,3 @@
def convert_function(_from, _to):
'''This decorator is simple way to declare ability of function to be
converter from name _from to name _to in convert table'''
def real_decorator(function):
function.__from__ = _from
function.__to__ = _to
return function
return real_decorator
class ConvertTable(): class ConvertTable():
'''The class is used to store and find the right function to convert value from '''The class is used to store and find the right function to convert value from
one key to another''' one key to another'''
@ -18,14 +8,29 @@ class ConvertTable():
'''Here you can add function to ConvertTable.''' '''Here you can add function to ConvertTable.'''
#TODO: make this method produce new functions, that will be able to #TODO: make this method produce new functions, that will be able to
#create converter chains #create converter chains
print("adding function", function)
self.convert_functions.append(function) self.convert_functions.append(function)
def all_possible_conversions(self, from_keys):
result = set()
from_keys = set(from_keys)
for function in self.convert_functions:
input_args = set(value for key, value in
function.__annotations__.items() if
key!='return')
if input_args.issubset(from_keys):
result = result.union(set(function.__annotations__['return']))
return result
def get_converter(self, from_keys, to_key): def get_converter(self, from_keys, to_key):
'''This function returns converter function, that can convert one key '''This function returns converter function, that can convert one key
to another''' to another'''
to_key = {to_key}
for function in self.convert_functions: for function in self.convert_functions:
if function.__from__ in from_keys and function.__to__ == to_key: input_args = set(value for key, value in
return function.__from__, function function.__annotations__.items() if
print("Can't find converter!") key!='return')
if input_args.issubset(from_keys) and to_key.issubset(function.__annotations__['return']):
return input_args, function
raise Exception("There is no converter for %s to %s" % (from_keys,
to_key))
return None, None return None, None

View File

@ -0,0 +1,32 @@
def complitable_functions(output_function, input_function, convert_table):
input_keys = set(value for key, value in
input_function.__annotations__.items() if key != 'return')
return_keys = set(output_function.__annotations__["return"])
all_possible_return_keys = return_keys.union(
convert_table.all_possible_conversions(return_keys)
)
print("return:%s\npossible_output:%s" % (return_keys,
all_possible_return_keys))
return input_keys.issubset(all_possible_return_keys)
def get_converted_arguments(function, simple_arg_dict, convert_table):
if simple_arg_dict == None:
return [None for key in function.__annotations__.keys() if key != 'return']
result = []
for key, value in function.__annotations__.items():
if key != 'return':
converted_arg = None
try:
converted_arg = simple_arg_dict[value]
except KeyError:
key_to_convert, convert_function = convert_table.get_converter(
simple_arg_dict.keys(),
value
)
key_to_convert = list(key_to_convert)[0]
converted_arg = convert_function(
simple_arg_dict[key_to_convert]
)[value]
result.append(converted_arg)
return result

View File

@ -0,0 +1,24 @@
from abc import abstractmethod
from core.prototypes.AbstractModuleClass import AbstractModuleClass
class AbstractAddressGenerator(AbstractModuleClass):
'''The class describes addess generation mechanism.'''
@abstractmethod
def set_parsed_fields(self):
'''This method is called after generator initialization. It is used to
store parsing results for futher address generation'''
pass
@abstractmethod
def get_next_address(self, previous_address, scan_result):
'''Address - an only, indivisible object, that describes single scan
target address. This method should return next address to scan based on
previous scanned address and result of scanning previous address, that
can be placed in kwargs.
Scan results is results of scan got from parser'''
pass
@abstractmethod
def get_all_addresses(self):
'''This method is used to return all addresses (ip ranges, url
patterns, etc) at one time to scnners with self thread management'''
pass

View File

@ -0,0 +1,32 @@
def internal(func):
func.is_internal = True
return func
class AbstractModuleClassType(type):
def __new__(self, name, bases, attrs):
print("creating class", name)
if not name.startswith("Abstract"):
for attrname, attrvalue in attrs.items():
if type(attrvalue).__name__ == 'function':
if attrvalue.__name__ not in ["__init__", "save"] and not (hasattr(attrvalue, "is_internal") and attrvalue.is_internal
):
if not name.endswith("Storage"):
try:
attrvalue.__annotations__["return"]
except KeyError:
raise Exception(
"%s.%s: return type is not defined!" %
(name, attrname)
)
if not name.endswith("Parser"):
if not attrvalue.__annotations__:
raise Exception(
"%s.%s: arguments missing annotations!" %
(name, attrname)
)
return super().__new__(self, name, bases, attrs)
class AbstractModuleClass(metaclass = AbstractModuleClassType):
REQUIED_INPUT_KEYS = None
OUTPUT_KEYS = []

View File

@ -0,0 +1,13 @@
from abc import abstractmethod
from core.prototypes.AbstractModuleClass import AbstractModuleClass
class AbstractParser(AbstractModuleClass):
'''The class describes fields parsing mechanisms'''
@abstractmethod
def parse_fields(self, args):
'''In address field can be plased any text, describing address of
scanning target.
In port field only numbers, whitespaces, comma and '-' allowed.'''
pass

View File

@ -0,0 +1,14 @@
from abc import abstractmethod
from core.prototypes.AbstractModuleClass import AbstractModuleClass
class AbstractScanner(AbstractModuleClass):
'''By default the class is used by one thread to scan targets
If it can manage many threads by itself set INDEPENDENT_THREAD_MANAGEMENT
to "True"'''
INDEPENDENT_THREAD_MANAGEMENT = False
@abstractmethod
def scan_address(self, address):
'''This method should contain scanning process of given address. All
items returned will be passed to AbstractStorage and ui'''
pass

View File

@ -0,0 +1,12 @@
from abc import abstractmethod
from core.prototypes.AbstractModuleClass import AbstractModuleClass
class AbstractStorage(AbstractModuleClass):
@abstractmethod
def put_responce(self, address, responce):
pass
@abstractmethod
def save(self):
pass

View File

View File

@ -1,15 +0,0 @@
from abc import abstractmethod, ABC
from core.communication import CommunicationDictionary
class AbstractAddressGenerator(ABC):
'''The class describes addess generation mechanism.
In __init__ method it should get results of parsing fields
and then it returns addresses.'''
@abstractmethod
def get_next_address(self, previous_address, **kwargs) -> CommunicationDictionary:
'''Address - an only, indivisible object, that describes single scan
target address. This method should return next address to scan based on
previous scanned address and result of scanning previous address, that
can be placed in kwargs.'''
pass

View File

@ -1,16 +0,0 @@
from abc import ABC, abstractmethod
class AbstractParser(ABC):
'''The class describes fields parsing mechanisms'''
@abstractmethod
def parse_address_field(self, field):
'''In address field can be plased any text, describing address of
scanning target'''
pass
@abstractmethod
def parse_port_field(self, field):
'''In port field only numbers, whitespaces, comma and '-' allowed'''
pass

View File

@ -0,0 +1,46 @@
from core.prototypes.AbstractAddressGenerator import AbstractAddressGenerator
from core.prototypes.AbstractModuleClass import internal
class GDocsAddressGenerator(AbstractAddressGenerator):
def set_parsed_fields(self, prefix:"gdocs_prefix",
ranges:"gdocs_hash_ranges") -> None:
self.alphabet = list(
'-0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZ_abcdefghijklmnopqrstuvwxyz'
)
self.revsymbols = {symb:i for i, symb in enumerate(self.alphabet)}
self.prefix = prefix
self.ranges = ranges
self.hashlen = len(ranges[0][0])
self.currange = self.ranges.pop(0)
@internal
def hash2int(self, gdhash):
alen = len(self.alphabet)
res = 0
for symb in gdhash:
res *= alen
res += self.revsymbols[symb]
return res
@internal
def int2hash(self, hint):
alen = len(self.alphabet)
reshash = [self.alphabet[0]]*self.hashlen
for i in range(-1, -self.hashlen-1, -1):
hint, rest = divmod(hint, alen)
reshash[i] = self.alphabet[rest]
return "".join(reshash)
def get_next_address(self, prev_url:'url') -> {"url"}:
if not prev_url:
return {'url':self.prefix + self.currange[0]}
prev_hash = prev_url[prev_url.rfind('/') + 1:]
if self.hash2int(self.currange[1]) <= self.hash2int(prev_hash):
if not self.ranges: return None
self.currange = self.ranges.pop(0)
return {'url' : self.prefix + self.currange[0]}
return {'url' : self.prefix + self.int2hash(self.hash2int(prev_hash) +
1)}
def get_all_addresses(self) -> {'gdocs_prefix', 'gdocs_hash_ranges'}:
return {'gdocs_prefix':self.prefix, 'gdocs_hash_ranges': self.ranges}

View File

@ -0,0 +1,19 @@
from core.prototypes.AbstractParser import AbstractParser
class GDocsHashParser(AbstractParser):
def parse_fields(self, afield:"address_field") -> {"gdocs_prefix",
"gdocs_hash_ranges"}:
split_index = afield.find(" ")
prefix = afield[:split_index].strip()
hash_ranges = [hr.strip() for hr in afield[split_index:].split(",")]
result = []
for hr in hash_ranges:
if " - " in hr:
result.append(hr.split(" - "))
else:
result.append([hr, hr])
return {
"gdocs_prefix":prefix,
"gdocs_hash_ranges":result
}

View File

@ -1,32 +1,50 @@
from AbstractAddressGenerator import AbstractAddressGenerator from core.prototypes.AbstractAddressGenerator import AbstractAddressGenerator
from core.prototypes.AbstractModuleClass import internal
from threading import RLock from threading import RLock
from core.communication.CommunicationDictionary import CommunicationDictionary import ipaddress
from types import GeneratorType
class IpGenerator(AbstractAddressGenerator): class IpGenerator(AbstractAddressGenerator):
def __init__(self, ip_generator, ports, convert_table): def set_parsed_fields(self, ips : 'ipv4_ranges', ports : 'ports') -> None:
self.convert_table = convert_table self.ips = ips
self.ip_generator = ip_generator
self.ports = ports self.ports = ports
self.lock = RLock() self.lock = RLock()
@internal
def get_next_port_number(self, previous_port): def get_next_port_number(self, previous_port):
return (self.ports.index(previous_port) + 1) % len(self.ports) return (self.ports.index(previous_port) + 1) % len(self.ports)
def get_next_address(self, previous_address): def get_next_address(self, previous_address : 'ipv4', previous_port :
result = CommunicationDictionary(self.convert_table) 'port') -> {'ipv4', 'port'}:
result = dict()
with self.lock: with self.lock:
portnum = 0 portnum = 0
next_ip = None next_ip = None
if previous_address: if previous_address:
next_ip = previous_address["ipv4"] next_ip = previous_address
port = previous_address["port"] port = previous_port
portnum = self.get_next_port_number(port) portnum = self.get_next_port_number(port)
if (portnum == 0): if (portnum == 0):
next_ip = None
try: try:
next_ip = next(self.ip_generator) while not next_ip:
if isinstance(self.ips[0], ipaddress.IPv4Address):
next_ip = self.ips.pop(0)
else:
if not isinstance(self.ips[0], GeneratorType):
self.ips[0] = self.ips[0].hosts()
try:
next_ip = next(self.ips[0])
except StopIteration: except StopIteration:
self.ips.pop(0)
except IndexError:
return None return None
result["ipv4"] = next_ip result["ipv4"] = next_ip
result["port"] = self.ports[portnum] result["port"] = self.ports[portnum]
return result return result
def get_all_addresses(self) -> {'ipv4_ranges', 'ports'}:
result = dict()
result['ipv4_ranges'] = self.ips
result['ports'] = ports
return result

View File

@ -1,9 +1,18 @@
import ipaddress import ipaddress
from AbstractParser import AbstractParser from core.prototypes.AbstractParser import AbstractParser
from core.prototypes.AbstractModuleClass import internal
class Parser(AbstractParser): class Parser(AbstractParser):
def parse_fields(self, ports:'port_field', ips:'address_field') -> {'ipv4_objects',
'ports'}:
result = dict()
result['ports'] = self.parse_port_field(ports)
result['ipv4_ranges'] = self.get_all_addresses(ips)
return result
@internal
def parse_port_field(self, ports): def parse_port_field(self, ports):
""" """
Parses ports from string, returns them as integers in the list. Parses ports from string, returns them as integers in the list.
@ -31,6 +40,7 @@ class Parser(AbstractParser):
# Change to default ports from constant # Change to default ports from constant
return [21, 22, 23, 25, 80, 443, 110, 111, 135, 139, 445, 8080, 8443, 53, 143, 989, 990, 3306, 1080, 5554, 6667, 2222, 4444, 666, 6666, 1337, 2020, 31337] return [21, 22, 23, 25, 80, 443, 110, 111, 135, 139, 445, 8080, 8443, 53, 143, 989, 990, 3306, 1080, 5554, 6667, 2222, 4444, 666, 6666, 1337, 2020, 31337]
@internal
def parse_address_field(self, ips): def parse_address_field(self, ips):
""" """
Parses ip input string, returns the generator over them. Parses ip input string, returns the generator over them.
@ -43,6 +53,17 @@ class Parser(AbstractParser):
""" """
# A set to contain non repeating ip objects from ipaddress # A set to contain non repeating ip objects from ipaddress
for ip_obj in self.get_all_addresses():
# The object is just one ip, simply yield it:
if isinstance(ip_obj, ipaddress.IPv4Address):
yield ip_obj
# The object is a network, yield every host in it:
else:
for host in ip_obj:
yield host
@internal
def get_all_addresses(self, ips):
ip_objects = set() ip_objects = set()
inputs = [ip.strip() for ip in ips.split(',')] inputs = [ip.strip() for ip in ips.split(',')]
@ -69,12 +90,5 @@ class Parser(AbstractParser):
print(e) print(e)
# If we get any non-ip value just ignore it # If we get any non-ip value just ignore it
pass pass
ip_objects = list(ip_objects)
for ip_obj in ip_objects: return ip_objects
# The object is just one ip, simply yield it:
if isinstance(ip_obj, ipaddress.IPv4Address):
yield ip_obj
# The object is a network, yield every host in it:
else:
for host in ip_obj:
yield host

View File

@ -0,0 +1,13 @@
from core.prototypes.AbstractAddressGenerator import AbstractAddressGenerator
from core.communication.CommunicationDictionary import CommunicationDictionary
class PlugAddressGenerator(AbstractAddressGenerator):
def __init__(self, all_addresses, ports, convert_table):
self.res = CommunicationDictionary(convert_table)
self.res["address_list"] = all_addresses
self.res["ports"] = ports
def get_next_address(self, previous_address):
res = self.res
self.res = None
return res

View File

@ -1,5 +1,2 @@
from core.communication.ConvertTable import convert_function def ipv4tostring(ip:"ipv4") -> {"ipv4_str"}:
return {"ipv4_str":str(ip)}
@convert_function("ipv4","str")
def ipv4tostring(ip):
return str(ip)

View File

@ -0,0 +1,2 @@
def response2status(response:"response") -> {"status"}:
return {"status":responce.status_code}

View File

@ -1,17 +0,0 @@
from abc import ABC, abstractmethod
from core.communication.CommunicationDictionary import CommunicationDictionary
class AbstractScanner(ABC):
'''The class is used by one thread to scan targets'''
@abstractmethod
def __init__(self, **kwargs):
'''In this method you can init some
reusable resources needed for scan'''
pass
@abstractmethod
def scan_address(self, address) -> CommunicationDictionary:
'''This method should contain scanning process of given address. All
items returned will be passed to AbstractStorage and ui'''
pass

View File

@ -1,17 +1,14 @@
import socket import socket
from AbstractScanner import AbstractScanner from core.prototypes.AbstractScanner import AbstractScanner
from core.communication.CommunicationDictionary import CommunicationDictionary
class CoreModel(AbstractScanner): class CoreModel(AbstractScanner):
def __init__(self, timeout): def __init__(self, timeout):
self.defSocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.defSocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.defSocket.settimeout(int(timeout)) self.defSocket.settimeout(int(timeout))
def scan_address(self, address: CommunicationDictionary) -> CommunicationDictionary: def scan_address(self, host:'ipv4_str', port:'port') -> {'scan_result'}:
host = address["str"] result = dict()
port = address["port"]
if not host: raise Exception if not host: raise Exception
result = self.defSocket.connect_ex((host, port)) result["scan_result"] = self.defSocket.connect_ex((host, port))
self.defSocket.close() self.defSocket.close()
self.defSocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
return result return result

View File

@ -0,0 +1,10 @@
from core.prototypes.AbstractScanner import AbstractScanner
import requests
class URLScanner(AbstractScanner):
def __init__(self, timeout):
pass
def scan_address(self, url:"url") -> {"response"}:
return {'response':requests.get(url)}

View File

@ -1,12 +0,0 @@
from abc import ABC, abstractmethod
class AbstractStorage(ABC):
@abstractmethod
def put_responce(self, address, responce):
pass
@abstractmethod
def save(self):
pass

View File

@ -0,0 +1,16 @@
from core.prototypes.AbstractStorage import AbstractStorage
import json
class GDocsStorage(AbstractStorage):
def __init__(self, path):
self.path = path
self.urls = dict()
def put_responce(self, url:'url', status:'status'):
if str(status) not in self.urls.keys():
self.urls[str(status)] = []
self.urls[str(status)].append(url)
def save(self):
print("saving")
with open(self.path, "w") as f:
json.dump(self.urls, f)
self.urls = dict()

View File

@ -1,8 +1,7 @@
from AbstractStorage import AbstractStorage from core.prototypes.AbstractStorage import AbstractStorage
import json import json
from threading import RLock from threading import RLock
class JSONStorage(AbstractStorage): class JSONStorage(AbstractStorage):
def __init__(self, path): def __init__(self, path):
@ -10,11 +9,11 @@ class JSONStorage(AbstractStorage):
self.respdict = dict() self.respdict = dict()
self.lock = RLock() self.lock = RLock()
def put_responce(self, address, responce): def put_responce(self, ip:'ipv4_str', port:'port', scan_result:'scan_result'):
ip, port = address
if ip not in self.respdict.keys(): if ip not in self.respdict.keys():
self.respdict[ip] = {"open": [], "close": []} self.respdict[ip] = {"open": [], "close": []}
self.respdict[ip]["open" if responce != 0 else "close"].append(port) self.respdict[ip]["open" if scan_result == 0
else "close"].append(port)
def save(self): def save(self):
print("saving") print("saving")

View File

@ -1 +1 @@
PyQt5==5.11.3 PyQt5>=5.11.3