Docker/pandora-box
Docker/pandora-box
1
0
Fork 0
mirror of https://github.com/dbarzin/pandora-box.git synced 2025-07-18 21:09:41 +02:00
Code Issues Activity

new tests

Browse source
This commit is contained in:
dbarzin 2022-09-25 10:30:14 +02:00
parent 816179cfa4
commit d57d498f5a
1 changed files with 33 additions and 31 deletions
Download patch file Download diff file Expand all files Collapse all files

64
pandora-box.py Executable file → Normal file
View file

@ -45,22 +45,18 @@ def config():
global FAKE_SCAN, QUARANTINE, QUARANTINE_FOLDER global FAKE_SCAN, QUARANTINE, QUARANTINE_FOLDER
global CURSES global CURSES
# intantiate a ConfirParser # intantiate a ConfirParser
try : config = configparser.ConfigParser()
config = configparser.ConfigParser() # read the config file
# read the config file config.read('pandora-box.ini')
config.read('pandora-box.ini') # set values
# set values FAKE_SCAN=config['DEFAULT']['FAKE_SCAN'].lower()=="true"
FAKE_SCAN = config['DEFAULT']['FAKE_SCAN'].lower()=="true" USB_AUTO_MOUNT=config['DEFAULT']['USB_AUTO_MOUNT'].lower()=="true"
USB_AUTO_MOUNT = config['DEFAULT']['USB_AUTO_MOUNT'].lower()=="true" PANDORA_ROOT_URL=config['DEFAULT']['PANDORA_ROOT_URL']
PANDORA_ROOT_URL = config['DEFAULT']['PANDORA_ROOT_URL'] # Quarantine
# Quarantine QUARANTINE = config['DEFAULT']['QUARANTINE'].lower()=="true"
QUARANTINE = config['DEFAULT']['QUARANTINE'].lower()=="true" QUARANTINE_FOLDER = config['DEFAULT']['QUARANTINE_FOLDER']
QUARANTINE_FOLDER = config['DEFAULT']['QUARANTINE_FOLDER'] # Curses
# Curses CURSES = config['DEFAULT']['CURSES'].lower()=="true"
CURSES = config['DEFAULT']['CURSES'].lower()=="true"
except Exception as e :
log("Could not read config file: %s" % e)
raise e
# ---------------------------------------------------------- # ----------------------------------------------------------
@ -96,10 +92,10 @@ def display_image(status):
# display image # display image
if "*" in image: if "*" in image:
# slide show # slide show
os.system("fim -qa -c 'while(1){display;sleep 1;next;}' %s </dev/null 2>/dev/null &" % image) os.system("fim -qa -c 'while(1){display;sleep 1;next;}' %s </dev/null 2>/dev/null >/dev/null &" % image)
else : else :
# only one image # only one image
os.system("fim -qa %s </dev/null 2>/dev/null &" % image) os.system("fim -qa %s </dev/null 2>/dev/null >/dev/null &" % image)
# ----------------------------------------------------------- # -----------------------------------------------------------
@ -290,8 +286,8 @@ def log(str):
for i in range(min(curses.LINES-22,len(logs))): for i in range(min(curses.LINES-22,len(logs))):
log_win.addstr(i+1,1,logs[i][:curses.COLS-2],curses.color_pair(3)) log_win.addstr(i+1,1,logs[i][:curses.COLS-2],curses.color_pair(3))
log_win.refresh() log_win.refresh()
else: # else:
print(str,end="\n\r") # print(str,end="\n\r")
# ----------------------------------------------------------- # -----------------------------------------------------------
# Device # Device
@ -320,7 +316,7 @@ def mount_device(device):
if not os.path.exists("/media/box"): if not os.path.exists("/media/box"):
log("folder /media/box does not exists") log("folder /media/box does not exists")
return None return None
res = os.system("pmount " + device.device_node + " /media/box") res = os.system("pmount " + device.device_node + " /media/box >/dev/null 2>/dev/null")
found = False found = False
loop = 0 loop = 0
while (not found) and (loop < 10): while (not found) and (loop < 10):
@ -335,17 +331,18 @@ def mount_device(device):
return "/media/box" return "/media/box"
"""Unmount USB device""" """Unmount USB device"""
def umount_device(mount_point): def umount_device():
if USB_AUTO_MOUNT: if USB_AUTO_MOUNT:
os.system("umount " + mount_point) log("Sync partitions")
res = os.system("sync")
else: else:
os.system("punmount /media/box") log("Unmount partitions")
res = os.system("pumount /media/box 2>/dev/null >/dev/null")
"""Main device loop""" """Main device loop"""
def device_loop(): def device_loop():
# First unmount remaining device # First unmount remaining device
if not USB_AUTO_MOUNT: umount_device()
umount_device("/media/box")
# Loop # Loop
context = pyudev.Context() context = pyudev.Context()
monitor = pyudev.Monitor.from_netlink(context) monitor = pyudev.Monitor.from_netlink(context)
@ -376,6 +373,7 @@ def device_loop():
statvfs=os.statvfs(mount_point) statvfs=os.statvfs(mount_point)
except Exception as e : except Exception as e :
log("Unexpected error: %s" % e) log("Unexpected error: %s" % e)
logging.info("An exception was thrown!", exc_info=True)
if not CURSES: if not CURSES:
display_image("WAIT") display_image("WAIT")
continue continue
@ -402,6 +400,7 @@ def device_loop():
log('%s removed' % file) log('%s removed' % file)
except Exception as e : except Exception as e :
log("Unexpected error: %s" % str(e)) log("Unexpected error: %s" % str(e))
logging.info("An exception was thrown!", exc_info=True)
os.system("sync") os.system("sync")
log("Clean done.") log("Clean done.")
if not CURSES: if not CURSES:
@ -409,7 +408,7 @@ def device_loop():
else: else:
if not CURSES: if not CURSES:
display_image("OK") display_image("OK")
umount_device(mount_point) umount_device()
if device.action == "remove": if device.action == "remove":
log("Device removed") log("Device removed")
@ -425,6 +424,7 @@ def device_loop():
update_bar(0) update_bar(0)
except Exception as e: except Exception as e:
log("Unexpected error: %s" % str(e) ) log("Unexpected error: %s" % str(e) )
logging.info("An exception was thrown!", exc_info=True)
finally: finally:
log("Done.") log("Done.")
@ -449,7 +449,7 @@ def log_device_info(dev):
"""Scan a mount point with Pandora""" """Scan a mount point with Pandora"""
def scan(mount_point, used): def scan(mount_point, used):
global infected_filed, FAKE_SCAN global infected_filed
infected_files = [] infected_files = []
scanned = 0 scanned = 0
file_count = 0 file_count = 0
@ -466,13 +466,14 @@ def scan(mount_point, used):
file_size = os.path.getsize(full_path) file_size = os.path.getsize(full_path)
# log("Check %s [%s]" % (file, human_readable_size(file_size))) # log("Check %s [%s]" % (file, human_readable_size(file_size)))
file_scan_start_time = time.time() file_scan_start_time = time.time()
if FAKE_SCAN: if FAKE_SCAN :
time.sleep(0.1) time.sleep(0.1)
status = "SKIPPED" status = "SKIPPED"
else: else:
if file_size > (1024*1024*1024): if file_size > (1024*1024*1024):
status = "TOO BIG" status = "TOO BIG"
else: else:
log("ppypandora : [%s] " % full_path)
res = pandora.submit_from_disk(full_path) res = pandora.submit_from_disk(full_path)
time.sleep(0.1) time.sleep(0.1)
loop = 0 loop = 0
@ -528,7 +529,8 @@ def main(stdscr):
while True: while True:
device_loop() device_loop()
except Exception as e : except Exception as e :
log("Unexpected error: %s" % e) log("Unexpected error: %s" % e)
logging.info("An exception was thrown!", exc_info=True)
finally: finally:
end_curses() end_curses()