2019-08-07 07:44:03 +00:00
|
|
|
from functools import wraps
|
2019-05-20 12:52:02 +00:00
|
|
|
from glob import glob
|
|
|
|
from fnmatch import fnmatchcase
|
2019-08-07 07:44:03 +00:00
|
|
|
from os.path import dirname, isdir, isfile, realpath
|
2019-05-20 12:52:02 +00:00
|
|
|
|
|
|
|
|
2019-08-07 07:44:03 +00:00
|
|
|
def _with_is_allowed(func):
|
|
|
|
@wraps(func)
|
|
|
|
def wrapper(self, *args, **kwargs):
|
|
|
|
if self._is_allowed(args[0]): # pylint: disable=protected-access
|
|
|
|
return func(self, *args, **kwargs)
|
|
|
|
raise ValueError('Forbidden path.')
|
|
|
|
return wrapper
|
2019-05-20 12:52:02 +00:00
|
|
|
|
|
|
|
|
2019-08-07 07:44:03 +00:00
|
|
|
class FileManager: # pylint: disable=too-many-instance-attributes
|
|
|
|
def __init__(self, patterns):
|
|
|
|
self.patterns = patterns
|
2019-05-20 12:52:02 +00:00
|
|
|
|
|
|
|
@property
|
|
|
|
def files(self):
|
2019-08-07 07:44:03 +00:00
|
|
|
return list(set(
|
|
|
|
path
|
|
|
|
for pattern in self.patterns
|
|
|
|
for path in glob(pattern, recursive=True)
|
|
|
|
if isfile(path) and self._is_allowed(path)
|
|
|
|
))
|
2019-05-20 12:52:02 +00:00
|
|
|
|
|
|
|
@property
|
2019-08-07 07:44:03 +00:00
|
|
|
def parents(self):
|
|
|
|
return list(set(
|
|
|
|
self._find_directory(pattern)
|
|
|
|
for pattern in self.patterns
|
|
|
|
))
|
|
|
|
|
|
|
|
@staticmethod
|
|
|
|
def _find_directory(pattern):
|
|
|
|
while pattern and not isdir(pattern):
|
|
|
|
pattern = dirname(pattern)
|
|
|
|
return pattern
|
|
|
|
|
|
|
|
def _is_allowed(self, filepath):
|
2019-05-20 12:52:02 +00:00
|
|
|
return any(
|
2019-08-07 07:44:03 +00:00
|
|
|
fnmatchcase(realpath(filepath), pattern)
|
|
|
|
for pattern in self.patterns
|
2019-05-20 12:52:02 +00:00
|
|
|
)
|
|
|
|
|
2019-08-07 07:44:03 +00:00
|
|
|
@_with_is_allowed
|
|
|
|
def read_file(self, filepath): # pylint: disable=no-self-use
|
|
|
|
with open(filepath, 'rb', buffering=0) as ifile:
|
|
|
|
return ifile.read().decode(errors='surrogateescape')
|
2019-05-20 12:52:02 +00:00
|
|
|
|
2019-08-07 07:44:03 +00:00
|
|
|
@_with_is_allowed
|
|
|
|
def write_file(self, filepath, contents): # pylint: disable=no-self-use
|
|
|
|
with open(filepath, 'wb', buffering=0) as ofile:
|
|
|
|
ofile.write(contents.encode())
|