Source code for repoman.common.sources.dir
#!/usr/bin/env python
"""
Usage::
dir_path
file_path
dir:repo_name
Will find all the matching artifacts under the specified dir or the given
file.
If relative path passed, it will be relative to the base_repos_path config
value
Configuration values:
* allowed_dir_paths
Comma separated list of paths or empty string.
If set, will not allow using any path/dir/repo from outside of those
paths.
"""
import os
import logging
from . import ArtifactSource
from ..stores import has_store
from ..utils import (
find_recursive,
split,
)
logger = logging.getLogger(__name__)
[docs]class DirSource(ArtifactSource):
__doc__ = __doc__
DEFAULT_CONFIG = {
'allowed_dir_paths': '',
}
CONFIG_SECTION = 'DirSource'
[docs] @staticmethod
def is_allowed(path, allowed_paths):
if allowed_paths and not any(
matched_path
for matched_path in allowed_paths
if path.startswith(matched_path)
):
return False
return True
[docs] def resolve_path(self, path):
if os.path.isabs(path) and os.path.isdir(path):
return path
abs_path = os.path.abspath(path)
if os.path.isdir(abs_path):
return abs_path
allowed_paths = self.config.getarray('allowed_dir_paths')
for allowed_path in allowed_paths:
full_path = os.path.abspath(
os.path.join(allowed_path, path)
)
if os.path.isdir(full_path):
return full_path
raise IOError('No such directory in the path: %s' % path)
[docs] def check_if_allowed(self, path):
allowed_paths = self.config.getarray('allowed_dir_paths')
if not self.is_allowed(path, allowed_paths):
error_msg = 'Source %s outside the base path' % path
logger.error(error_msg + '\nAllowed paths: %s', allowed_paths)
raise IOError(error_msg)
[docs] def expand(self, source_str):
orig_source_str = source_str
if source_str.startswith('dir:'):
source_str = source_str.split(':', 1)[-1]
elif (
not os.path.isdir(source_str)
and has_store(source_str, self.stores)
):
return '', [source_str]
# get rid of any trailing filters
source_path, filters_str = split(source_str, ':', 1)
try:
source_path = self.resolve_path(source_path)
except IOError:
logger.debug('Skipping %s', orig_source_str)
return '', []
logger.debug('Resolved path: %s', source_path)
self.check_if_allowed(source_path)
return (
filters_str,
find_recursive(source_path, lambda x: has_store(x, self.stores))
)