mirror of
https://github.com/ChronosX88/PyNesca.git
synced 2024-10-18 06:31:00 +00:00
Removed duplicate directories.
This commit is contained in:
Zloooy
parent
b16e248712
commit
77ca495fdf
106
MainPresenter.py
106
MainPresenter.py
@ -1,106 +0,0 @@
|
||||
from network_scan.CoreModel import CoreModel
|
||||
from address_generation.Parser import Parser
|
||||
from threading import RLock
|
||||
import datetime
|
||||
from PyQt5.Qt import QThread, pyqtSignal
|
||||
from PyQt5.QtCore import QObject, pyqtSlot
|
||||
from address_generation.IpGenerator import IpGenerator
|
||||
from storage.JSONStorage import JSONStorage
|
||||
|
||||
|
||||
class MainPresenter:
|
||||
def __init__(self, ui):
|
||||
self.ui = ui
|
||||
self.threads = []
|
||||
self.isScanEnabled = False
|
||||
self.parser = Parser()
|
||||
#needed config to specify path
|
||||
self.storage = JSONStorage("results.json")
|
||||
self.exit_lock = RLock()
|
||||
|
||||
def startScan(self, ipRanges, portsStr, threadNumber, timeout):
|
||||
timeout = 3 if not timeout else int(timeout)
|
||||
addresses = self.parser.parse_address_field(ipRanges)
|
||||
ports = self.parser.parse_port_field(portsStr)
|
||||
self.ip_generator = IpGenerator(addresses, ports)
|
||||
self.scanner = CoreModel(timeout)
|
||||
threadNumber = int(threadNumber)
|
||||
for i in range(threadNumber):
|
||||
scan_worker = ScanWorker(
|
||||
self.ip_generator,
|
||||
self.scanner,
|
||||
self.storage
|
||||
)
|
||||
scan_thread = QThread()
|
||||
scan_worker.log_signal.connect(self.log_text)
|
||||
scan_worker.moveToThread(scan_thread)
|
||||
scan_worker.exit_signal.connect(scan_thread.exit)
|
||||
scan_worker.exit_signal.connect(self.on_worker_exit)
|
||||
scan_thread.started.connect(scan_worker.work)
|
||||
self.threads.append((scan_worker, scan_thread))
|
||||
self.changeThreadLabel(threadNumber)
|
||||
for thread in self.threads:
|
||||
scan_worker, scan_thread = thread
|
||||
scan_thread.start()
|
||||
|
||||
def changeThreadLabel(self, number_of_threads):
|
||||
self.number_of_threads = number_of_threads
|
||||
self.ui.currentThreadsLabel.setText(str(number_of_threads))
|
||||
|
||||
def on_worker_exit(self):
|
||||
self.changeThreadLabel(self.number_of_threads - 1)
|
||||
with self.exit_lock:
|
||||
for num, thread in enumerate(self.threads):
|
||||
scan_worker, scan_thread = thread
|
||||
if not scan_worker.isRunning:
|
||||
self.threads.pop(num)
|
||||
break
|
||||
if self.number_of_threads == 0:
|
||||
self.on_end_scanning()
|
||||
|
||||
def on_end_scanning(self):
|
||||
self.isScanEnabled = False
|
||||
self.ui.startButton.setText("Start")
|
||||
self.storage.save()
|
||||
|
||||
def stopScan(self):
|
||||
while self.threads:
|
||||
scan_worker, scan_thread = self.threads[0]
|
||||
if scan_worker.isRunning:
|
||||
scan_worker.stop()
|
||||
|
||||
def log_text(self, string):
|
||||
self.ui.dataText.append("[" + str(datetime.datetime.now()) + "] " + str(string))
|
||||
|
||||
|
||||
class ScanWorker(QObject):
|
||||
|
||||
log_signal = pyqtSignal(str)
|
||||
exit_signal = pyqtSignal()
|
||||
|
||||
def __init__(self, ip_generator, scanner, storage, **kwargs):
|
||||
super().__init__()
|
||||
self.ip_generator = ip_generator
|
||||
self.storage = storage
|
||||
self.scanner = scanner
|
||||
self.previous_address = None
|
||||
self.isRunning = True
|
||||
|
||||
@pyqtSlot()
|
||||
def work(self):
|
||||
while self.isRunning:
|
||||
scan_address = self.ip_generator.get_next_address(self.previous_address)
|
||||
if not scan_address:
|
||||
break
|
||||
self.previous_address = scan_address
|
||||
scan_result = self.scanner.scan_address(scan_address)
|
||||
self.storage.put_responce(scan_address, scan_result)
|
||||
if scan_result == 0:
|
||||
self.log_signal.emit('%s has open port: %s' % scan_address)
|
||||
else:
|
||||
self.log_signal.emit('%s has closed port: %s' % scan_address)
|
||||
self.stop()
|
||||
|
||||
def stop(self):
|
||||
self.isRunning = False
|
||||
self.exit_signal.emit()
|
||||
98
RawTCP.py
98
RawTCP.py
@ -1,98 +0,0 @@
|
||||
from struct import pack, unpack
|
||||
import socket
|
||||
|
||||
|
||||
class TCPHeader():
|
||||
# TCP header class. Thanks to Silver Moon for the flags calculation and packing order
|
||||
# This was designed to be re-used. You might want to randomize the seq number
|
||||
# get_struct performs packing based on if you have a valid checksum or not
|
||||
def __init__(self, src_port=47123, dst_port=80, seqnum=1000, acknum=0, data_offset=80, fin=0, syn=1, rst=0, psh=0, ack=0, urg=0, window=5840, check=0, urg_ptr=0):
|
||||
# !=network(big-endian), H=short(2), L=long(4),B=char(1)
|
||||
self.order = "!HHLLBBHHH"
|
||||
self.src_port = src_port
|
||||
self.dst_port = dst_port
|
||||
self.seqnum = seqnum
|
||||
self.acknum = acknum
|
||||
# size of tcp header; size is specified by 4-byte words; This is 80
|
||||
# decimal, which is 0x50, which is 20bytes (5words*4bytes).
|
||||
self.data_offset = data_offset
|
||||
self.fin = fin
|
||||
self.syn = syn
|
||||
self.rst = rst
|
||||
self.psh = psh
|
||||
self.ack = ack
|
||||
self.urg = urg
|
||||
self.window = socket.htons(window)
|
||||
self.check = check
|
||||
self.urg_ptr = urg_ptr
|
||||
|
||||
def flags(self):
|
||||
return self.fin + (self.syn << 1) + (self.rst << 2) + (self.psh << 3) + (self.ack << 4) + (self.urg << 5)
|
||||
|
||||
def get_struct(self, check=False, checksummed=False):
|
||||
if check is not False:
|
||||
self.check = check
|
||||
if checksummed:
|
||||
return pack('!HHLLBBH', self.src_port, self.dst_port, self.seqnum, self.acknum, self.data_offset, self.flags(), self.window) + pack('H', self.check) + pack('!H', self.urg_ptr)
|
||||
else:
|
||||
return pack(self.order, self.src_port, self.dst_port, self.seqnum, self.acknum, self.data_offset, self.flags(), self.window, self.check, self.urg_ptr)
|
||||
|
||||
|
||||
def checksum(msg):
|
||||
# Shoutout to Silver Moon @ binarytides for this checksum algo.
|
||||
sum = 0
|
||||
for i in range(0, len(msg), 2):
|
||||
w = msg[i] + (msg[i + 1] << 8)
|
||||
sum = sum + w
|
||||
|
||||
sum = (sum >> 16) + (sum & 0xffff)
|
||||
sum = sum + (sum >> 16)
|
||||
sum = ~sum & 0xffff
|
||||
return sum
|
||||
|
||||
|
||||
def tcp_checksum(source_ip, dest_ip, tcp_header, user_data=b''):
|
||||
# Calculates the correct checksum for the tcp header
|
||||
tcp_length = len(tcp_header) + len(user_data)
|
||||
# This is an IP header w/ TCP as protocol.
|
||||
ip_header = pack('!4s4sBBH', socket.inet_aton(source_ip), socket.inet_aton(
|
||||
dest_ip), 0, socket.IPPROTO_TCP, tcp_length)
|
||||
# Assemble the packet (IP Header + TCP Header + data, and then send it to
|
||||
# checksum function)
|
||||
packet = ip_header + tcp_header + user_data
|
||||
return checksum(packet)
|
||||
|
||||
|
||||
def handle_packet(raw_packet):
|
||||
# Now we need to unpack the packet. It will be an IP/TCP packet
|
||||
# We are looking for SYN-ACKs from our SYN scan
|
||||
# Fields to check: IP - src addr; TCP - src port, flags
|
||||
# We want to pull out and compare only these three
|
||||
# Heres the math for unpacking: B=1, H=2, L=4, 4s=4 (those are bytes)
|
||||
packet = raw_packet[0]
|
||||
# This is the IP header, not including any self.options OR THE DST ADDR.
|
||||
# Normal length is 20!! Im parsing as little as possible
|
||||
ip_header = unpack('!BBHHHBBH4s', packet[0:16])
|
||||
# If there are any self.options, the length of the IP header will be >20. We
|
||||
# dont care about self.options
|
||||
ip_header_length = (ip_header[0] & 0xf) * 4
|
||||
# This is the source address (position 8, or the first "4s" in our
|
||||
# unpack)
|
||||
src_addr = socket.inet_ntoa(ip_header[8])
|
||||
|
||||
# We had to get the proper IP Header length to find the TCP header
|
||||
# offset.
|
||||
tcp_header_raw = packet[ip_header_length:ip_header_length + 14]
|
||||
# TCP header structure is pretty straight-forward. We want PORTS and
|
||||
# FLAGS, so we partial unpack it
|
||||
tcp_header = unpack('!HHLLBB', tcp_header_raw)
|
||||
|
||||
src_port = tcp_header[0] # self-explanatory
|
||||
dst_port = tcp_header[1] # self-explanatory FIXME: notused
|
||||
# We only care about syn-ack, which will be 18 (0x12)
|
||||
flag = tcp_header[5]
|
||||
|
||||
if flag == 18:
|
||||
return (src_addr, src_port)
|
||||
else:
|
||||
return None
|
||||
89
TCPScan.py
89
TCPScan.py
@ -1,89 +0,0 @@
|
||||
from RawTCP import TCPHeader, handle_packet, tcp_checksum
|
||||
import socket
|
||||
import sys
|
||||
|
||||
|
||||
class TCPScanner:
|
||||
def __init__(self):
|
||||
self.source_ips = {}
|
||||
# TODO: options, threading, input/output queues
|
||||
self.options = None
|
||||
return
|
||||
|
||||
def in_scope(self):
|
||||
"""Check that IP is in scanning scope"""
|
||||
# TODO
|
||||
pass
|
||||
|
||||
def tcp_listener(self):
|
||||
# Raw socket listener for when send_raw_syn() is used. This will catch
|
||||
# return SYN-ACKs
|
||||
listen = socket.socket(socket.AF_INET, socket.SOCK_RAW, socket.IPPROTO_TCP)
|
||||
while True:
|
||||
# packet = ('E \x00(\x1f\xaa@\x00w\x06\x99w2\xe0\xc8\xa2\xa2\xf3\xac\x18\xdf\xb3\x00\x16\xb6\x80\xc1\xa0/\xa6=$P\x10\xce\xab\xd1\xe4\x00\x00', ('50.XXX.200.162', 0))
|
||||
raw_packet = listen.recvfrom(65565)
|
||||
ret = handle_packet(raw_packet)
|
||||
if ret is None:
|
||||
continue
|
||||
src_addr, src_port = ret
|
||||
if self.in_scope(src_addr) and src_port in self.ports:
|
||||
self.output_queue.put((src_addr, src_port))
|
||||
|
||||
def send_raw_syn(self, dest_ip, dst_port):
|
||||
# Use raw sockets to send a SYN packet.
|
||||
# If you want, you could use the IP header assembled in the tcp_checksum
|
||||
# function to have a fully custom TCP/IP stack
|
||||
try:
|
||||
# Using IPPROTO_TCP so the kernel will deal with the IP packet for us.
|
||||
# Change to IPPROTO_IP if you want control of IP header as well
|
||||
s = socket.socket(socket.AF_INET, socket.SOCK_RAW, socket.IPPROTO_TCP)
|
||||
except Exception:
|
||||
sys.stderr.write("Error creating socket in send_raw_syn\n")
|
||||
return
|
||||
if self.options.source == "auto":
|
||||
# This gets the correct source IP. Just in case of multiple interfaces,
|
||||
# it will pick the right one
|
||||
src_addr = self.get_source_ip(dest_ip)
|
||||
else:
|
||||
src_addr = self.options.source
|
||||
src_port = 54321
|
||||
make_tcpheader = TCPHeader(src_port, dst_port)
|
||||
tcp_header = make_tcpheader.get_struct()
|
||||
packet = make_tcpheader.get_struct(check=tcp_checksum(
|
||||
src_addr, dest_ip, tcp_header), checksummed=True)
|
||||
try:
|
||||
s.sendto(packet, (dest_ip, 0))
|
||||
except Exception as e:
|
||||
sys.stderr.write("Error utilizing raw socket in send_raw_syn: {}\n".format(e))
|
||||
|
||||
def get_source_ip(self, dst_addr):
|
||||
# Credit: 131264/alexander from stackoverflow. This gets the correct IP for sending. Useful if you have multiple interfaces
|
||||
# NOTE: This will send an additional packet for every single IP to confirm
|
||||
# the route. (but just one packet)
|
||||
try:
|
||||
if dst_addr in self.source_ips:
|
||||
return self.source_ips[dst_addr]
|
||||
else:
|
||||
self.source_ips[dst_addr] = [(s.connect((dst_addr, 53)), s.getsockname()[0], s.close(
|
||||
)) for s in [socket.socket(socket.AF_INET, socket.SOCK_DGRAM)]][0][1]
|
||||
return self.source_ips[dst_addr]
|
||||
except Exception:
|
||||
sys.stderr.write(
|
||||
"Something went wrong in get_source_ip, results might be wrong\n")
|
||||
|
||||
|
||||
def send_full_connect_syn(ip, port, timeout):
|
||||
# Normal scan using socket to connect. Does 3-way handshack, then graceful
|
||||
# teardown using FIN
|
||||
try:
|
||||
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||||
s.settimeout(timeout)
|
||||
except Exception as e:
|
||||
sys.stderr.write("Error creating socket in send_full_connect_syn: {}\n".format(e))
|
||||
return False
|
||||
try:
|
||||
s.connect((ip, port))
|
||||
return True
|
||||
s.close()
|
||||
except Exception:
|
||||
return False
|
||||
@ -1,15 +0,0 @@
|
||||
from abc import abstractmethod, ABC
|
||||
|
||||
|
||||
class AbstractAddressGenerator(ABC):
|
||||
'''The class describes addess generation mechanism.
|
||||
In __init__ method it should get results of parsing fields
|
||||
and then it returns addresses.'''
|
||||
@classmethod
|
||||
@abstractmethod
|
||||
def get_next_address(self, previous_address, **kwargs):
|
||||
'''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
|
||||
@ -1,18 +0,0 @@
|
||||
from abc import ABC, abstractmethod
|
||||
|
||||
|
||||
class AbstractParser(ABC):
|
||||
'''The class describes fields parsing mechanisms'''
|
||||
|
||||
@classmethod
|
||||
@abstractmethod
|
||||
def parse_address_field(self, field):
|
||||
'''In address field can be plased any text, describing address of
|
||||
scanning target'''
|
||||
pass
|
||||
|
||||
@classmethod
|
||||
@abstractmethod
|
||||
def parse_port_field(self, field):
|
||||
'''In port field only numbers, whitespaces, comma and '-' allowed'''
|
||||
pass
|
||||
@ -1,28 +0,0 @@
|
||||
from AbstractAddressGenerator import AbstractAddressGenerator
|
||||
from threading import RLock
|
||||
|
||||
|
||||
class IpGenerator(AbstractAddressGenerator):
|
||||
|
||||
def __init__(self, ip_generator, ports):
|
||||
self.ip_generator = ip_generator
|
||||
self.ports = ports
|
||||
self.lock = RLock()
|
||||
|
||||
def get_next_port_number(self, previous_port):
|
||||
return (self.ports.index(previous_port) + 1) % len(self.ports)
|
||||
|
||||
def get_next_address(self, previous_address):
|
||||
with self.lock:
|
||||
portnum = 0
|
||||
next_ip = None
|
||||
if previous_address:
|
||||
next_ip, port = previous_address
|
||||
portnum = self.get_next_port_number(port)
|
||||
if (portnum == 0):
|
||||
try:
|
||||
next_ip = str(next(self.ip_generator))
|
||||
except StopIteration:
|
||||
return None
|
||||
result = (next_ip, self.ports[portnum])
|
||||
return result
|
||||
@ -1,80 +0,0 @@
|
||||
import ipaddress
|
||||
from AbstractParser import AbstractParser
|
||||
|
||||
|
||||
class Parser(AbstractParser):
|
||||
|
||||
def parse_port_field(self, ports):
|
||||
"""
|
||||
Parses ports from string, returns them as integers in the list.
|
||||
Handles non-existent ports and non-port values.
|
||||
"""
|
||||
if ports:
|
||||
# Using set to avoid repetitions
|
||||
parsed = set()
|
||||
ports = ports.split(",")
|
||||
for port in ports:
|
||||
try:
|
||||
# Input is in range form ("100-200"):
|
||||
if '-' in port:
|
||||
start, end = map(int, port.split('-'))
|
||||
parsed.update(
|
||||
[p for p in range(start, end + 1) if 65355 >= p > 0])
|
||||
# Input is a single port ("80"):
|
||||
else:
|
||||
parsed.add(int(port))
|
||||
except ValueError as e:
|
||||
# If we get any not integer just ignore it
|
||||
pass
|
||||
return sorted(list(parsed))
|
||||
else:
|
||||
# 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]
|
||||
|
||||
def parse_address_field(self, ips):
|
||||
"""
|
||||
Parses ip input string, returns the generator over them.
|
||||
|
||||
Supports next inputs:
|
||||
1) 1.2.3.4
|
||||
2) 192.168.0.0/24
|
||||
3) 1.2.3.4 - 5.6.7.8
|
||||
Any non-ip value will be ignored.
|
||||
"""
|
||||
|
||||
# A set to contain non repeating ip objects from ipaddress
|
||||
ip_objects = set()
|
||||
inputs = [ip.strip() for ip in ips.split(',')]
|
||||
|
||||
for input_ in inputs:
|
||||
try:
|
||||
# Input is in range form ("1.2.3.4 - 5.6.7.8"):
|
||||
if '-' in input_:
|
||||
input_ips = input_.split('-')
|
||||
ranges = set(
|
||||
ipaddress.summarize_address_range(
|
||||
*map(lambda x: ipaddress.IPv4Address(x.strip()), input_ips)
|
||||
)
|
||||
)
|
||||
ip_objects.update(ranges)
|
||||
# Input is in CIDR form ("192.168.0.0/24"):
|
||||
elif '/' in input_:
|
||||
network = ipaddress.ip_network(input_)
|
||||
ip_objects.add(network)
|
||||
# Input is a single ip ("1.1.1.1"):
|
||||
else:
|
||||
ip = ipaddress.ip_address(input_)
|
||||
ip_objects.add(ip)
|
||||
except ValueError as e:
|
||||
print(e)
|
||||
# If we get any non-ip value just ignore it
|
||||
pass
|
||||
|
||||
for ip_obj in 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
|
||||
@ -1,6 +0,0 @@
|
||||
import sys
|
||||
import os
|
||||
|
||||
|
||||
fil = __file__[:__file__.rindex(os.sep)]
|
||||
sys.path.insert(0,fil)
|
||||
@ -1,27 +0,0 @@
|
||||
from unittest import TestCase, main
|
||||
from ipaddress import IPv4Address
|
||||
from IpGenerator import IpGenerator
|
||||
from Parser import Parser
|
||||
|
||||
|
||||
class testIpGenerator(TestCase):
|
||||
|
||||
def setUp(self):
|
||||
p = Parser()
|
||||
self.ipgen = IpGenerator(
|
||||
p.parse_address_field("192.168.1.1 - 192.168.1.10"), [80, 90])
|
||||
|
||||
def testIpGeneration(self):
|
||||
'''self.assertEqual(
|
||||
self.ipgen.get_next_address(None),
|
||||
("192.168.1.1", 80))'''
|
||||
previous_address = None
|
||||
a = True
|
||||
while previous_address or a:
|
||||
previous_address=self.ipgen.get_next_address(previous_address)
|
||||
a = False
|
||||
self.assertEqual(self.ipgen.get_next_address(None), None)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@ -1,37 +0,0 @@
|
||||
from unittest import main, TestCase
|
||||
from Parser import Parser
|
||||
import ipaddress
|
||||
|
||||
|
||||
class TestParser(TestCase):
|
||||
|
||||
def setUp(self):
|
||||
self.parser = Parser()
|
||||
|
||||
def test_port_parsing(self):
|
||||
self.assertEqual(self.parser.parse_port_field("80,90"), [80, 90])
|
||||
|
||||
def test_ip_parsing(self):
|
||||
self.assertEqual(
|
||||
set(self.parser.parse_address_field("192.168.1.1 - 192.168.1.3")),
|
||||
{ipaddress.IPv4Address(ip) for ip in
|
||||
[
|
||||
"192.168.1.1",
|
||||
"192.168.1.2",
|
||||
"192.168.1.3"
|
||||
]})
|
||||
self.assertEqual(
|
||||
set(self.parser.parse_address_field("192.168.1.1")),
|
||||
{ipaddress.IPv4Address("192.168.1.1")}
|
||||
)
|
||||
self.assertEqual(
|
||||
set(self.parser.parse_address_field("192.168.1.0/31")),
|
||||
{ipaddress.IPv4Address(ip) for ip in
|
||||
[
|
||||
'192.168.1.1',
|
||||
'192.168.1.0'
|
||||
]})
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@ -1,19 +0,0 @@
|
||||
from abc import ABC, abstractmethod
|
||||
|
||||
|
||||
class AbstractScanner(ABC):
|
||||
'''The class is used by one thread to scan targets'''
|
||||
|
||||
@classmethod
|
||||
@abstractmethod
|
||||
def __init__(self, **kwargs):
|
||||
'''In this method you can init some
|
||||
reusable resources needed for scan'''
|
||||
pass
|
||||
|
||||
@classmethod
|
||||
@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
|
||||
@ -1,15 +0,0 @@
|
||||
import socket
|
||||
from AbstractScanner import AbstractScanner
|
||||
|
||||
|
||||
class CoreModel(AbstractScanner):
|
||||
def __init__(self, timeout):
|
||||
self.defSocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||||
self.defSocket.settimeout(int(timeout))
|
||||
|
||||
def scan_address(self, address):
|
||||
host, port = address
|
||||
result = self.defSocket.connect_ex((host, port))
|
||||
self.defSocket.close()
|
||||
self.defSocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||||
return result
|
||||
@ -1,6 +0,0 @@
|
||||
import sys
|
||||
import os
|
||||
|
||||
|
||||
fil = __file__[:__file__.rindex(os.sep)]
|
||||
sys.path.insert(0,fil)
|
||||
@ -1,14 +0,0 @@
|
||||
from abc import ABC, abstractmethod
|
||||
|
||||
|
||||
class AbstractStorage(ABC):
|
||||
|
||||
@classmethod
|
||||
@abstractmethod
|
||||
def put_responce(self, address, responce):
|
||||
pass
|
||||
|
||||
@classmethod
|
||||
@abstractmethod
|
||||
def save(self):
|
||||
pass
|
||||
@ -1,23 +0,0 @@
|
||||
from AbstractStorage import AbstractStorage
|
||||
import json
|
||||
from threading import RLock
|
||||
|
||||
|
||||
class JSONStorage(AbstractStorage):
|
||||
|
||||
def __init__(self, path):
|
||||
self.path = path
|
||||
self.respdict = dict()
|
||||
self.lock = RLock()
|
||||
|
||||
def put_responce(self, address, responce):
|
||||
ip, port = address
|
||||
if ip not in self.respdict.keys():
|
||||
self.respdict[ip] = {"open": [], "close": []}
|
||||
self.respdict[ip]["open" if responce != 0 else "close"].append(port)
|
||||
|
||||
def save(self):
|
||||
print("saving")
|
||||
with open(self.path, "w") as f:
|
||||
json.dump(self.respdict, f)
|
||||
self.respdict = {}
|
||||
@ -1,6 +0,0 @@
|
||||
import sys
|
||||
import os
|
||||
|
||||
|
||||
fil = __file__[:__file__.rindex(os.sep)]
|
||||
sys.path.insert(0,fil)
|
||||
Loading…
Reference in New Issue
Block a user