CIRCL
/
Circlean
mirror of https://github.com/CIRCL/Circlean
2
0
Fork 0
Code Issues Releases Wiki Activity

add comments

pull/33/head
Raphaël Vinot 2015-04-04 18:44:17 +02:00
parent 54ab11b9d4
commit 78b17b013b
3 changed files with 85 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

34
fs/opt/groomer/functions.py
View File

@ -29,6 +29,7 @@ mimes_data = ['octet-stream']
class File(FileBase):
def __init__(self, src_path, dst_path):
''' Init file object, set the mimetype '''
super(File, self).__init__(src_path, dst_path)
mimetype = magic.from_file(src_path, mime=True)
self.main_type, self.sub_type = mimetype.split('/')
@ -37,6 +38,9 @@ class File(FileBase):
self.is_recursive = False
def crosscheck_mime(self):
'''
Set the expected mime and extension variables based on mime type.
'''
# /usr/share/mime has interesting stuff
# guess_type uses the extension to get a mime type
@ -51,12 +55,14 @@ class File(FileBase):
return expected_mimetype, expected_extensions
def verify_extension(self):
'''Check if the extension is the one we expect'''
if self.expected_extensions is None:
return None
path, actual_extension = os.path.splitext(self.src_path)
return actual_extension in self.expected_extensions
def verify_mime(self):
'''Check if the mime is the one we expect'''
if self.expected_mimetype is None:
return None
actual_mimetype = '{}/{}'.format(self.main_type, self.sub_type)
@ -66,6 +72,9 @@ class File(FileBase):
class KittenGroomer(KittenGroomerBase):
def __init__(self, root_src=None, root_dst=None, max_recursive=5):
'''
Initialize the basics of the conversion process
'''
if root_src is None:
root_src = os.path.join(os.sep, 'media', 'src')
if root_dst is None:
@ -100,6 +109,9 @@ class KittenGroomer(KittenGroomerBase):
# ##### Helpers #####
def _init_subtypes_application(self, subtypes_application):
'''
Create the Dict to pick the right function based on the sub mime type
'''
to_return = {}
for list_subtypes, fct in subtypes_application:
for st in list_subtypes:
@ -107,6 +119,9 @@ class KittenGroomer(KittenGroomerBase):
return to_return
def _print_log(self):
'''
Print the logs related to the current file being processed
'''
tmp_log = self.log_name.fields(**self.cur_file.log_details)
if self.cur_file.log_details.get('dangerous'):
tmp_log.warning(self.cur_file.log_string)
@ -116,6 +131,7 @@ class KittenGroomer(KittenGroomerBase):
tmp_log.debug(self.cur_file.log_string)
def _run_process(self, command_line):
'''Run subprocess, wait until it finishes'''
args = shlex.split(command_line)
p = subprocess.Popen(args)
while True:
@ -138,21 +154,25 @@ class KittenGroomer(KittenGroomerBase):
# ##### Threated as malicious, no reason to have it on a USB key ######
def example(self):
'''Way to process example file'''
self.cur_file.log_string += 'Example file'
self.cur_file.make_dangerous()
self._safe_copy()
def message(self):
'''Way to process message file'''
self.cur_file.log_string += 'Message file'
self.cur_file.make_dangerous()
self._safe_copy()
def model(self):
'''Way to process model file'''
self.cur_file.log_string += 'Model file'
self.cur_file.make_dangerous()
self._safe_copy()
def multipart(self):
'''Way to process multipart file'''
self.cur_file.log_string += 'Multipart file'
self.cur_file.make_dangerous()
self._safe_copy()
@ -176,11 +196,13 @@ class KittenGroomer(KittenGroomerBase):
self._unknown_app()
def _executables(self):
'''Way to process executable file'''
self.cur_file.add_log_details('processing_type', 'executable')
self.cur_file.make_dangerous()
self._safe_copy()
def _office_related(self):
'''Way to process all the files LibreOffice can handle'''
self.cur_file.add_log_details('processing_type', 'office')
dst_dir, filename = os.path.split(self.cur_file.dst_path)
tmpdir = os.path.join(dst_dir, 'temp')
@ -194,11 +216,13 @@ class KittenGroomer(KittenGroomerBase):
self._safe_rmtree(tmpdir)
def _pdfa(self, tmpsrcpath):
'''Way to process PDF/A file'''
pdf_command = '{} --dest-dir / {} {}'.format(PDF2HTMLEX, tmpsrcpath,
self.cur_file.dst_path + '.html')
self._run_process(pdf_command)
def _pdf(self):
'''Way to process PDF file'''
self.cur_file.add_log_details('processing_type', 'pdf')
dst_dir, filename = os.path.split(self.cur_file.dst_path)
tmpdir = os.path.join(dst_dir, 'temp')
@ -211,6 +235,7 @@ class KittenGroomer(KittenGroomerBase):
self._safe_rmtree(tmpdir)
def _archive(self):
'''Way to process Archive'''
self.cur_file.add_log_details('processing_type', 'archive')
self.cur_file.is_recursive = True
self.cur_file.log_string += 'Archive extracted, processing content.'
@ -224,10 +249,12 @@ class KittenGroomer(KittenGroomerBase):
self._safe_rmtree(tmpdir)
def _unknown_app(self):
'''Way to process an unknown file'''
self.cur_file.make_unknown()
self._safe_copy()
def _binary_app(self):
'''Way to process an unknown binary file'''
self.cur_file.make_binary()
self._safe_copy()
@ -235,18 +262,22 @@ class KittenGroomer(KittenGroomerBase):
# ##### Not converted, checking the mime type ######
def audio(self):
'''Way to process an audio file'''
self.cur_file.log_string += 'Audio file'
self._media_processing()
def image(self):
'''Way to process an image'''
self.cur_file.log_string += 'Image file'
self._media_processing()
def video(self):
'''Way to process a video'''
self.cur_file.log_string += 'Video file'
self._media_processing()
def _media_processing(self):
'''Generic way to process all the media files'''
self.cur_log.fields(processing_type='media')
if not self.cur_file.verify_mime() or not self.cur_file.verify_extension():
# The extension is unknown or doesn't match the mime type => suspicious
@ -257,6 +288,9 @@ class KittenGroomer(KittenGroomerBase):
#######################
def processdir(self, src_dir=None, dst_dir=None):
'''
Main function doing the processing
'''
if src_dir is None:
src_dir = self.src_root_dir
if dst_dir is None:

11
fs/opt/groomer/functions_pier9.py
View File

@ -18,6 +18,7 @@ up = ['.upp', '.up3', '.stl', '.obj']
class FilePier9(FileBase):
def __init__(self, src_path, dst_path):
''' Init file object, set the extension '''
super(FilePier9, self).__init__(src_path, dst_path)
a, self.extension = os.path.splitext(self.src_path)
@ -25,7 +26,9 @@ class FilePier9(FileBase):
class KittenGroomerPier9(KittenGroomerBase):
def __init__(self, root_src=None, root_dst=None):
'''
Initialize the basics of the copy
'''
if root_src is None:
root_src = os.path.join(os.sep, 'media', 'src')
if root_dst is None:
@ -36,6 +39,9 @@ class KittenGroomerPier9(KittenGroomerBase):
self.authorized_extensions = printers + cnc + shopbot + omax + epilog_laser + metabeam + up
def _print_log(self):
'''
Print the logs related to the current file being processed
'''
tmp_log = self.log_name.fields(**self.cur_file.log_details)
if not self.cur_file.log_details.get('valid'):
tmp_log.warning(self.cur_file.log_string)
@ -43,6 +49,9 @@ class KittenGroomerPier9(KittenGroomerBase):
tmp_log.debug(self.cur_file.log_string)
def processdir(self):
'''
Main function doing the processing
'''
for srcpath in self._list_all_files(self.src_root_dir):
self.log_name.info('Processing {}', srcpath.replace(self.src_root_dir + '/', ''))
self.cur_file = FilePier9(srcpath, srcpath.replace(self.src_root_dir, self.dst_root_dir))

42
fs/opt/groomer/helpers.py
View File

@ -7,36 +7,64 @@ from twiggy import quickSetup, log
class KittenGroomerError(Exception):
def __init__(self, message):
'''
Base KittenGroomer exception handler.
'''
super(KittenGroomerError, self).__init__(message)
self.message = message
class ImplementationRequired(KittenGroomerError):
'''
Implementation required error
'''
pass
class FileBase(object):
def __init__(self, src_path, dst_path):
'''
Contains base information for a file on the source USB key,
initialised with expected src and dest path
'''
self.src_path = src_path
self.dst_path = dst_path
self.log_details = {'filepath': self.src_path}
self.log_string = ''
def add_log_details(self, key, value):
'''
Add an entry in the log dictionary
'''
self.log_details[key] = value
def make_dangerous(self):
'''
This file should be considered as dangerous and never run.
Prepending and appending DANGEROUS to the destination
file name avoid double-click of death
'''
self.log_details['dangerous'] = True
path, filename = os.path.split(self.dst_path)
self.dst_path = os.path.join(path, 'DANGEROUS_{}_DANGEROUS'.format(filename))
def make_unknown(self):
'''
This file has an unknown type and it was not possible to take
a decision. Theuser will have to decide what to do.
Prepending UNKNOWN
'''
self.log_details['unknown'] = True
path, filename = os.path.split(self.dst_path)
self.dst_path = os.path.join(path, 'UNKNOWN_{}'.format(filename))
def make_binary(self):
'''
This file is a binary, and should probably not be run.
Appending .bin avoir double click of death but the user
will have to decide by itself.
'''
self.log_details['binary'] = True
path, filename = os.path.split(self.dst_path)
self.dst_path = os.path.join(path, '{}.bin'.format(filename))
@ -45,6 +73,9 @@ class FileBase(object):
class KittenGroomerBase(object):
def __init__(self, root_src, root_dst):
'''
Setup the base options of the copy/convert setup
'''
self.src_root_dir = root_src
self.dst_root_dir = root_dst
self.log_root_dir = os.path.join(self.dst_root_dir, 'logs')
@ -58,19 +89,22 @@ class KittenGroomerBase(object):
# ##### Helpers #####
def _safe_rmtree(self, directory):
'''Remove a directory tree if it exists'''
if os.path.exists(directory):
shutil.rmtree(directory)
def _safe_remove(self, filepath):
'''Remove a file if it exists'''
if os.path.exists(filepath):
os.remove(filepath)
def _safe_mkdir(self, directory):
'''Remove a directory if it exists'''
if not os.path.exists(directory):
os.makedirs(directory)
def _safe_copy(self):
''' Create dir if needed '''
''' Copy a file and create directory if needed '''
try:
dst_path, filename = os.path.split(self.cur_file.dst_path)
self._safe_mkdir(dst_path)
@ -82,6 +116,7 @@ class KittenGroomerBase(object):
return False
def _list_all_files(self, directory):
''' Generate an iterator over all the files in a directory tree '''
for root, dirs, files in os.walk(directory):
for filename in files:
filepath = os.path.join(root, filename)
@ -89,6 +124,8 @@ class KittenGroomerBase(object):
def _print_log(self):
'''
Print log, should be called after each file.
You probably want to reimplement it in the subclass
'''
tmp_log = self.log_name.fields(**self.cur_file.log_details)
@ -97,4 +134,7 @@ class KittenGroomerBase(object):
#######################
def processdir(self, src_dir=None, dst_dir=None):
'''
Main function doing the work, you have to implement it yourself.
'''
raise ImplementationRequired('You have to implement the result processdir.')