Page MenuHomePhorge
Files F85803179

No OneTemporary
Actions

Size
38 KB
Referenced Files
None
Subscribers
None

View Options

diff --git a/lilybuild/lilybuild/ci_steps.py b/lilybuild/lilybuild/ci_steps.py
index fda1496..fcbf9f1 100644
--- a/lilybuild/lilybuild/ci_steps.py
+++ b/lilybuild/lilybuild/ci_steps.py
@@ -1,488 +1,491 @@
from buildbot.plugins import *
from buildbot.process import buildstep, logobserver
from buildbot.interfaces import IRenderable
from twisted.internet import defer
from .ci_syntax import ci_file
from .ci_syntax import rules as ci_rules
from .helpers import rsync_rules_from_artifacts, get_job_script, normalize_image
from .phorge import SendCoverageToPhorge
import re
import sys
import json
SAFETAR_EXEC = '/lilybuild/lilybuild/safetar.py'
COVERAGE_EXEC = '/lilybuild/lilybuild/coverage.py'
def on_success(step):
return step.build.results == util.SUCCESS
def on_always(_step):
return True
def fill_list(*args):
return list(args)
class RunCIJobStep(steps.BuildStep):
# 200 MiB
artifact_max_size = 200 * 1024 * 1024
default_image = 'alpine'
master_job_artifact_dir_pattern = '%(kw:st)s/repos/%(prop:lilybuild_repo_id)s/builds/%(prop:lilybuild_root_build_id)s/jobs/%(kw:job)s/artifacts'
artifact_file_name = 'artifacts.tar'
master_job_artifact_file_name_pattern = master_job_artifact_dir_pattern + '/' + artifact_file_name
reports_file_name = 'reports.tar'
master_reports_file_name_pattern = master_job_artifact_dir_pattern + '/' + reports_file_name
master_pages_dir_pattern = '%(kw:st)s/repos/%(prop:lilybuild_repo_id)s/pages'
phorge_coverage_file_name = 'coverage-phorge.json'
def __init__(
self,
lbc,
src_relative=None,
src_dir=None,
storage_dir=None,
repo_id=None,
result_relative=None,
result_dir=None,
artifact_stage_relative=None,
artifact_stage_dir=None,
job_prop=None,
artifact_link_base=None,
**kwargs):
self.lbc = lbc
self.src_relative = src_relative
self.src_dir = src_dir
self.work_root_dir = kwargs['workdir']
self.script_dir = 'script'
self.storage_dir = storage_dir
self.repo_id = repo_id
self.artifact_stage_relative = artifact_stage_relative
self.artifact_stage_dir = artifact_stage_dir
self.result_relative = result_relative
self.result_dir = result_dir
self.artifact_link_base = artifact_link_base
super().__init__(name='Run step', **kwargs)
@defer.inlineCallbacks
def run(self):
job_prop = self.getProperty('lilybuild_job_prop')
job = ci_file.CIJob.from_prop(job_prop)
job_index = self.getProperty('lilybuild_job_index')
variables = yield self.get_ci_variables(job)
should_run = True
if len(job.rules):
should_run = False
for r in job.rules:
try:
rule_str = r.get('if')
if not rule_str:
continue
res = ci_rules.evaluate_rule(ci_rules.parse_rule(rule_str), variables)
if not res:
continue
when = r.get('when', 'on_success')
if when == 'never':
should_run = False
else:
should_run = True
break
except SyntaxError:
self.addCompleteLog('error', f'Rule "{rule_str}" has syntax errors')
except:
pass
if should_run:
next_steps = self.job_to_steps(job, job_index, variables)
self.build.addStepsAfterCurrentStep(next_steps)
return util.SUCCESS
else:
self.addCompleteLog('info', 'Job skipped by a rule')
return util.SKIPPED
@defer.inlineCallbacks
def get_ci_variables(self, job):
res = {}
res.update(job.get_predefined_ci_variables())
res.update(self.getProperty('lilybuild_pipeline_vars'))
res['CI_JOB_IMAGE'] = job.image or self.default_image
res['CI_JOB_URL'] = yield self.build.getUrl()
res['CI_JOB_ID'] = self.build.buildid
res['CI_PROJECT_DIR'] = '/build'
try:
repo = self.lbc.repos[self.getProperty('lilybuild_repo_id')]
variables = yield repo['variables_getter'](self.build)
res_vars = {}
for var in variables:
value = variables[var]
if IRenderable.providedBy(value):
value = yield self.build.render(value)
res_vars[var] = value
res.update(res_vars)
except Exception as e:
self.addCompleteLog('exception', f'{e}')
return res
def get_upload_artifacts_jobs(self, short_name, artifact_type, artifact_name, base_dir, paths, exclude, master_pattern, job_index, doStepIf=on_success, has_pages=False):
archive_artifact_step = steps.ShellCommand(
name=f'Archive artifacts: {short_name}',
command=[
SAFETAR_EXEC,
],
initialStdin=json.dumps({
'op': 'create',
'archive_file': artifact_name,
'base_dir': base_dir,
'content': paths,
'items_to_exclude': exclude,
+ 'compression': 'gz',
}),
workdir=self.work_root_dir,
doStepIf=doStepIf,
)
masterdest = util.Interpolate(
master_pattern,
st=self.storage_dir,
job=job_index,
doStepIf=doStepIf,
)
parent_build_id = self.getProperty('lilybuild_pipeline_vars')['CI_PIPELINE_ID']
artifact_url = f'{self.artifact_link_base}/plugins/lilybuild_artifacts/builds/{parent_build_id}/jobs/{job_index}/artifacts/{artifact_type}' if self.artifact_link_base else None
upload_artifact_step = steps.FileUpload(
workersrc=artifact_name,
maxsize=self.artifact_max_size,
name=f'Upload artifacts: {short_name}',
masterdest=masterdest,
workdir=self.work_root_dir,
url=artifact_url,
doStepIf=doStepIf,
)
r = [archive_artifact_step, upload_artifact_step]
if has_pages:
r.append(steps.MasterShellCommand(
command=util.Transform(fill_list,
sys.executable,
'-m', 'lilybuild.pages',
util.Interpolate(
self.master_pages_dir_pattern,
st=self.storage_dir,
),
masterdest,
),
name='Deploy pages',
logEnviron=False,
doStepIf=doStepIf,
))
return r
def job_to_steps(self, job, job_index, variables):
script_name = self.script_dir + '/run.sh'
source_step = self.lbc.create_source_step()
script_step = steps.StringDownload(
get_job_script(variables, job),
name='Set up script',
workerdest=script_name,
workdir=self.work_root_dir,
doStepIf=on_success,
)
chmod_step = steps.ShellCommand(
name='Make script executable',
command=['chmod', '+x', script_name],
workdir=self.work_root_dir,
doStepIf=on_success,
)
artifact_steps = []
dep_job_indices = self.getProperty('lilybuild_dependency_job_indices')
if dep_job_indices:
for i in dep_job_indices:
# The steps may not run or may not have an artifact even if it runs
download_job = steps.FileDownload(
mastersrc=util.Interpolate(
self.master_job_artifact_file_name_pattern,
st=self.storage_dir,
job=i,
),
maxsize=self.artifact_max_size,
name=f'Download artifacts from job #{i}',
workerdest=self.artifact_file_name,
workdir=self.work_root_dir,
doStepIf=on_success,
haltOnFailure=False,
flunkOnFailure=False,
+ # https://github.com/buildbot/buildbot/issues/3709
+ blocksize=256 * 1024,
)
unarchive_job = steps.ShellCommand(
name=f'Unarchive artifacts from job #{i}',
command=[
SAFETAR_EXEC,
],
initialStdin=json.dumps({
'op': 'extract',
'archive_file': self.artifact_file_name,
'target_dir': self.src_relative,
}),
workdir=self.work_root_dir,
doStepIf=on_success,
haltOnFailure=False,
flunkOnFailure=False,
)
artifact_steps += [download_job, unarchive_job]
run_step = steps.ShellCommand(
name='Run script in container',
command=[
'/lilybuild/podman-helper',
normalize_image(job.image or self.default_image),
self.src_relative,
self.script_dir,
self.result_relative,
],
workdir=self.work_root_dir,
doStepIf=on_success,
)
clean_script_step = steps.ShellCommand(
name='Clean script dir',
command=[
'rm',
'-rf',
self.script_dir,
],
workdir=self.work_root_dir,
alwaysRun=True,
)
steps_to_run = [source_step, script_step, chmod_step] + artifact_steps + [run_step, clean_script_step]
if 'paths' in job.artifacts:
steps_to_run += self.get_upload_artifacts_jobs(
'files',
'archive',
self.artifact_file_name,
self.result_relative,
job.artifacts.get('paths', []),
job.artifacts.get('exclude', []),
self.master_job_artifact_file_name_pattern,
job_index,
has_pages=job.is_pages()
)
if job.has_supported_coverage_report():
steps_to_run += [steps.ShellCommand(
name='Process reports',
command=[
COVERAGE_EXEC,
],
initialStdin=json.dumps({
'source_dir': self.src_relative,
'result_dir': self.result_relative,
'untrusted_coverage_file': job.artifacts['reports']['coverage_report']['path'],
'output_dir': self.artifact_stage_relative,
}),
workdir=self.work_root_dir,
doStepIf=on_always,
)] + self.get_upload_artifacts_jobs(
'reports',
'reports',
self.reports_file_name,
self.artifact_stage_relative,
['*'],
[],
self.master_reports_file_name_pattern,
job_index,
doStepIf=on_always
) + [SendCoverageToPhorge(
self.lbc,
self.artifact_stage_relative + '/' + self.phorge_coverage_file_name,
workdir=self.work_root_dir,
)]
clean_stage_dir_again_step = steps.ShellCommand(
name='Clean stage, result and artifact',
command=[
'rm',
'-rf',
self.result_relative,
self.artifact_file_name,
self.artifact_stage_relative,
self.reports_file_name,
],
workdir=self.work_root_dir,
alwaysRun=True,
)
steps_to_run.append(clean_stage_dir_again_step)
return steps_to_run
class TriggerMultipleJobsStep(steps.Trigger):
properties_to_keep = [
'branch',
'revision',
'repository',
'harbormaster_build_target_phid',
'harbormaster_variable_buildable.diff',
'harbormaster_variable_repository.staging.ref',
'harbormaster_variable_repository.staging.uri',
'harbormaster_variable_repository.uri',
'lilybuild_repo',
'lilybuild_repo_id',
'lilybuild_pipeline_vars',
]
def __init__(self, lbc, jobs_with_data, **kwargs):
self.lbc = lbc
self.jobs = jobs_with_data
super().__init__(schedulerNames=[self.lbc.triggerable_scheduler_name], **kwargs)
def getSchedulersAndProperties(self):
ret = []
common_properties = {
'lilybuild_root_build_id': self.build.buildid,
}
for prop in self.properties_to_keep:
if self.hasProperty(prop):
common_properties[prop] = self.getProperty(prop)
for (job, i, dep_job_indices) in self.jobs:
properties = common_properties.copy()
properties['lilybuild_job_prop'] = job.to_prop()
properties['lilybuild_job_index'] = i
properties['virtual_builder_name'] = 'lilybuild-job - ' + common_properties['lilybuild_repo'] + ' - ' + job.name
properties['lilybuild_dependency_job_indices'] = dep_job_indices
ret.append({
'sched_name': self.lbc.triggerable_scheduler_name,
'props_to_set': properties,
'unimportant': False,
})
return ret
class AnalyzeCIFileCommand(buildstep.ShellMixin, steps.BuildStep):
ci_def_file = '.gitlab-ci.yml'
build_target_prop_name = 'harbormaster_build_target_phid'
def __init__(
self,
lbc,
src_relative=None,
src_dir=None,
storage_dir=None,
repo_id=None,
result_relative=None,
result_dir=None,
artifact_stage_relative=None,
artifact_stage_dir=None,
**kwargs):
kwargs['name'] = 'Analyze CI file'
kwargs['command'] = ['cat', self.ci_def_file]
self.lbc = lbc
self.src_relative = src_relative
self.src_dir = src_dir
self.work_root_dir = kwargs['workdir']
self.script_dir = 'script'
self.storage_dir = storage_dir
self.repo_id = repo_id
self.artifact_stage_relative = artifact_stage_relative
self.artifact_stage_dir = artifact_stage_dir
self.result_relative = result_relative
self.result_dir = result_dir
kwargs['workdir'] = self.src_dir
kwargs = self.setupShellMixin(kwargs)
super().__init__(**kwargs)
self.observer = logobserver.BufferLogObserver()
self.addLogObserver('stdio', self.observer)
def stage_to_step(self, stage_name, stage_jobs, job_name_to_index_map, ci_file):
jobs_with_data = []
for job in stage_jobs:
dep_job_names = [
jn
for jn in ci_file.get_jobs_to_pull_artifacts_from(job.name)
if ci_file.jobs[jn].has_artifacts_archive()
]
dep_job_indices = [job_name_to_index_map[jn] for jn in dep_job_names]
jobs_with_data.append((job, job_name_to_index_map[job.name], dep_job_indices))
trigger = TriggerMultipleJobsStep(
name=stage_name,
lbc=self.lbc,
jobs_with_data=jobs_with_data,
waitForFinish=True,
doStepIf=on_success,
)
return trigger
def get_steps(self, stdout):
f = ci_file.CIFile(stdout)
stages = f.get_grouped_jobs()
jobs = [job for (stage, js) in f.get_grouped_jobs() for job in js]
job_names = [job.name for job in jobs]
job_name_to_index_map = {}
for (i, j) in enumerate(jobs):
job_name_to_index_map[j.name] = i
steps = [self.stage_to_step(stage_name, stage_jobs, job_name_to_index_map, f) for (stage_name, stage_jobs) in stages]
print('steps:', steps)
return steps
def get_is_phorge(self):
return not not self.getProperty(self.build_target_prop_name)
def get_ref_and_type(self):
ref_type = 'branch'
ref = self.getProperty('branch')
if self.getProperty('category') == 'tag':
ref_type = 'tag'
if ref is not None:
m = re.match(r'^refs/(heads|tags)/(.+)$', ref)
if m:
ref = m.group(2)
return (ref, ref_type)
@defer.inlineCallbacks
def get_pipeline_ci_vars(self):
url = yield self.build.getUrl()
res = {
'CI_PIPELINE_ID': self.build.buildid,
'CI_PIPELINE_IID': self.build.buildid,
'CI_PIPELINE_URL': url,
'CI_PROJECT_ID': self.getProperty('lilybuild_repo_id'),
'CI_CONFIG_PATH': self.ci_def_file,
}
if not self.get_is_phorge():
res['CI_COMMIT_SHA'] = self.getProperty('got_revision')
res['CI_COMMIT_SHORT_SHA'] = res['CI_COMMIT_SHA'][:8]
(ref, ref_type) = self.get_ref_and_type()
res['CI_COMMIT_REF_NAME'] = ref
res['CI_COMMIT_REF_SLUG'] = ci_file.ci_slugify(ref)
res['CI_COMMIT_REF_PROTECTED'] = 'false'
if ref_type == 'tag':
res['CI_COMMIT_TAG'] = ref
elif ref_type == 'branch':
res['CI_COMMIT_BRANCH'] = ref
return res
@defer.inlineCallbacks
def run(self):
# run './build.sh --list-stages' to generate the list of stages
cmd = yield self.makeRemoteShellCommand()
yield self.runCommand(cmd)
# if the command passes extract the list of stages
result = cmd.results()
if result == util.SUCCESS:
pipeline_vars = yield self.get_pipeline_ci_vars()
self.setProperty('lilybuild_pipeline_vars', pipeline_vars, self.__class__.__name__)
# create a ShellCommand for each stage and add them to the build
self.build.addStepsAfterCurrentStep(self.get_steps(self.observer.getStdout()))
return result
diff --git a/lilybuild/lilybuild/safetar.py b/lilybuild/lilybuild/safetar.py
index a7f4a1c..902c4a4 100755
--- a/lilybuild/lilybuild/safetar.py
+++ b/lilybuild/lilybuild/safetar.py
@@ -1,239 +1,244 @@
#!/usr/bin/env python3
import os
import glob
import re
try:
import tarfile
tarfile.FilterError
except AttributeError:
import backports.tarfile as tarfile
try:
glob_translate = glob.translate
except AttributeError:
# Taken from python 3.13 library
def py313_fnmatch_translate(pat, STAR, QUESTION_MARK):
res = []
add = res.append
i, n = 0, len(pat)
while i < n:
c = pat[i]
i = i+1
if c == '*':
# compress consecutive `*` into one
if (not res) or res[-1] is not STAR:
add(STAR)
elif c == '?':
add(QUESTION_MARK)
elif c == '[':
j = i
if j < n and pat[j] == '!':
j = j+1
if j < n and pat[j] == ']':
j = j+1
while j < n and pat[j] != ']':
j = j+1
if j >= n:
add('\\[')
else:
stuff = pat[i:j]
if '-' not in stuff:
stuff = stuff.replace('\\', r'\\')
else:
chunks = []
k = i+2 if pat[i] == '!' else i+1
while True:
k = pat.find('-', k, j)
if k < 0:
break
chunks.append(pat[i:k])
i = k+1
k = k+3
chunk = pat[i:j]
if chunk:
chunks.append(chunk)
else:
chunks[-1] += '-'
# Remove empty ranges -- invalid in RE.
for k in range(len(chunks)-1, 0, -1):
if chunks[k-1][-1] > chunks[k][0]:
chunks[k-1] = chunks[k-1][:-1] + chunks[k][1:]
del chunks[k]
# Escape backslashes and hyphens for set difference (--).
# Hyphens that create ranges shouldn't be escaped.
stuff = '-'.join(s.replace('\\', r'\\').replace('-', r'\-')
for s in chunks)
# Escape set operations (&&, ~~ and ||).
stuff = re.sub(r'([&~|])', r'\\\1', stuff)
i = j+1
if not stuff:
# Empty range: never match.
add('(?!)')
elif stuff == '!':
# Negated empty range: match any character.
add('.')
else:
if stuff[0] == '!':
stuff = '^' + stuff[1:]
elif stuff[0] in ('^', '['):
stuff = '\\' + stuff
add(f'[{stuff}]')
else:
add(re.escape(c))
assert i == n
return res
def py313_translate(pat, *, recursive=False, include_hidden=False, seps=None):
"""Translate a pathname with shell wildcards to a regular expression.
If `recursive` is true, the pattern segment '**' will match any number of
path segments.
If `include_hidden` is true, wildcards can match path segments beginning
with a dot ('.').
If a sequence of separator characters is given to `seps`, they will be
used to split the pattern into segments and match path separators. If not
given, os.path.sep and os.path.altsep (where available) are used.
"""
if not seps:
if os.path.altsep:
seps = (os.path.sep, os.path.altsep)
else:
seps = os.path.sep
escaped_seps = ''.join(map(re.escape, seps))
any_sep = f'[{escaped_seps}]' if len(seps) > 1 else escaped_seps
not_sep = f'[^{escaped_seps}]'
if include_hidden:
one_last_segment = f'{not_sep}+'
one_segment = f'{one_last_segment}{any_sep}'
any_segments = f'(?:.+{any_sep})?'
any_last_segments = '.*'
else:
one_last_segment = f'[^{escaped_seps}.]{not_sep}*'
one_segment = f'{one_last_segment}{any_sep}'
any_segments = f'(?:{one_segment})*'
any_last_segments = f'{any_segments}(?:{one_last_segment})?'
results = []
parts = re.split(any_sep, pat)
last_part_idx = len(parts) - 1
for idx, part in enumerate(parts):
if part == '*':
results.append(one_segment if idx < last_part_idx else one_last_segment)
elif recursive and part == '**':
if idx < last_part_idx:
if parts[idx + 1] != '**':
results.append(any_segments)
else:
results.append(any_last_segments)
else:
if part:
if not include_hidden and part[0] in '*?':
results.append(r'(?!\.)')
results.extend(py313_fnmatch_translate(part, f'{not_sep}*', not_sep))
if idx < last_part_idx:
results.append(any_sep)
res = ''.join(results)
return fr'(?s:{res})\Z'
glob_translate = py313_translate
def extract(target_dir, archive_file):
with tarfile.open(archive_file) as tf:
tf.errorlevel = 1
tf.extractall(target_dir, filter='data')
class ArchiveFilter:
def __init__(
self,
base_dir,
limit_bytes=None,
items_to_exclude=None
):
self.base_dir = base_dir
self.total_bytes_added = 0
self.limit_bytes = limit_bytes or 100*1024*1024 # 100 MiB
self.exclude_re = [re.compile(glob_translate(i, recursive=True, include_hidden=True)) for i in (items_to_exclude or [])]
def __call__(self, member):
member.name = os.path.relpath('/' + member.name, self.base_dir)
filtered_member = tarfile.data_filter(member, self.base_dir)
if not filtered_member:
return None
if self.total_bytes_added + member.size > self.limit_bytes:
self.total_bytes_added = self.limit_bytes + 1
raise RuntimeError('Limit exceeded')
name = member.name
for r in self.exclude_re:
if r.match(name):
return None
if member.isdir() and r.match(name + '/'):
return None
self.total_bytes_added += member.size
# Assume that data_filter does not do anything else to it besides rejecting
# Any remaining (permissions) will be stripped when the archive is extracted
# Directly using filtered_member will cause errors in further processing,
# as the data_filter seems intended only for extraction.
return member
-def create(archive_file, base_dir, content, limit_bytes, items_to_exclude):
+def create(archive_file, base_dir, content, limit_bytes, items_to_exclude, compression=None):
base_dir = os.path.abspath(base_dir)
+ open_mode = 'w'
+ if compression == 'gz':
+ open_mode = 'w:gz'
+
try:
- with tarfile.open(archive_file, 'w') as tf:
+ with tarfile.open(archive_file, open_mode) as tf:
tf.errorlevel = 1
archive_filter = ArchiveFilter(
base_dir,
limit_bytes,
items_to_exclude=items_to_exclude
)
for g in content:
# iglob is important because once we found one file, we
# add it, and if it does not pass the data filter,
# then we are done with the whole archive. Using glob
# will make it hang here, resulting in DoS.
for f in glob.iglob(g, root_dir=base_dir, recursive=True):
# Specifying 'xxx/**' as the glob will probably make
# this called multiple times on the parent and child
# dirs, so best to avoid it. However, we aren't
# good enough to sanitize this.
tf.add(os.path.join(base_dir, f), filter=archive_filter)
except tarfile.FilterError:
# To prevent exploiting '/**', '../../../**' globs etc., we
# cannot allow the filename to be exposed
raise RuntimeError('Did not pass the data_filter')
if __name__ == '__main__':
import sys
import json
if len(sys.argv) > 1:
a = json.loads(sys.argv[1])
else:
a = json.loads(sys.stdin.read())
op = a.get('op')
if op == 'create':
# Do not remove this try-catch, or it will print out the original
# FilterError, resulting in at least one file name being exposed.
# The file name in the filter error should not be exposed,
# or it allows the attacker to view arbitrary directory structure
# inside the whole worker.
try:
create(
a['archive_file'],
a['base_dir'],
a['content'],
a.get('limit_bytes'),
- a.get('items_to_exclude')
+ a.get('items_to_exclude'),
+ a.get('compression'),
)
except RuntimeError as e:
print('Cannot create archive:', e)
sys.exit(1)
elif op == 'extract':
# Does not need a try-catch, because only the things inside the archive
# or the target dir can be exposed.
extract(a['target_dir'], a['archive_file'])
else:
print('Unknown operation:', op)
sys.exit(1)
diff --git a/lilybuild/lilybuild/tests/safetar_test_worker.py b/lilybuild/lilybuild/tests/safetar_test_worker.py
index 95496f6..68fa518 100644
--- a/lilybuild/lilybuild/tests/safetar_test_worker.py
+++ b/lilybuild/lilybuild/tests/safetar_test_worker.py
@@ -1,208 +1,223 @@
import unittest
import tempfile
import os
import stat
from lilybuild.safetar import (
create, extract
)
try:
import tarfile
tarfile.FilterError
except AttributeError:
import backports.tarfile as tarfile
def make_artifact_dir(root_dir):
os.makedirs(os.path.join(root_dir, 'public'))
with open(os.path.join(root_dir, 'public', 'a'), 'w') as f:
print('test', file=f)
os.makedirs(os.path.join(root_dir, 'other'))
with open(os.path.join(root_dir, 'other', 'a'), 'w') as f:
print('another test', file=f)
def make_artifact_dir_link(root_dir):
os.makedirs(os.path.join(root_dir, 'public'))
with open(os.path.join(root_dir, 'public', 'a'), 'w') as f:
print('test', file=f)
os.makedirs(os.path.join(root_dir, 'other'))
os.symlink('../public/a', os.path.join(root_dir, 'other', 'a'))
def make_bad_artifact_archive(root_dir):
os.makedirs(os.path.join(root_dir, 'public'))
with open(os.path.join(root_dir, 'public', 'a'), 'w') as f:
print('test', file=f)
os.makedirs(os.path.join(root_dir, 'other'))
with open(os.path.join(root_dir, 'other', 'a'), 'w') as f:
print('should not be there', file=f)
archive = os.path.join(root_dir, 'artifacts.tar')
with tarfile.open(archive, 'w') as f:
f.add(os.path.join(root_dir, 'public'), 'public')
f.add(os.path.join(root_dir, 'other'), '../../../other')
return archive
class SafetarTest(unittest.TestCase):
def test_create(self):
with tempfile.TemporaryDirectory() as dir_name:
make_artifact_dir(dir_name)
archive = os.path.join(dir_name, 'artifacts.tar')
create(archive, dir_name, ['public', 'other'], None, None)
with tarfile.open(archive, 'r') as f:
f.getmember('public/a')
f.getmember('other/a')
target = os.path.join(dir_name, 'extracts')
os.makedirs(target)
extract(target, archive)
+ def test_create_compression(self):
+ with tempfile.TemporaryDirectory() as dir_name:
+ make_artifact_dir(dir_name)
+ archive = os.path.join(dir_name, 'artifacts.tar')
+ create(archive, dir_name, ['public', 'other'], None, None, 'gz')
+ with tarfile.open(archive, 'r:gz') as f:
+ f.getmember('public/a')
+ f.getmember('other/a')
+ with tarfile.open(archive, 'r') as f:
+ f.getmember('public/a')
+ f.getmember('other/a')
+ target = os.path.join(dir_name, 'extracts')
+ os.makedirs(target)
+ extract(target, archive)
+
def test_create_glob(self):
with tempfile.TemporaryDirectory() as dir_name:
make_artifact_dir(dir_name)
archive = os.path.join(dir_name, 'artifacts.tar')
create(archive, dir_name, ['*'], None, None)
with tarfile.open(archive, 'r') as f:
f.getmember('public/a')
f.getmember('other/a')
with tempfile.TemporaryDirectory() as dir_name:
make_artifact_dir(dir_name)
archive = os.path.join(dir_name, 'artifacts.tar')
create(archive, dir_name, ['p*/a'], None, None)
with tarfile.open(archive, 'r') as f:
f.getmember('public/a')
with self.assertRaises(KeyError):
f.getmember('other')
with tempfile.TemporaryDirectory() as dir_name:
make_artifact_dir(dir_name)
archive = os.path.join(dir_name, 'artifacts.tar')
create(archive, dir_name, ['**/a'], None, None)
with tarfile.open(archive, 'r') as f:
f.getmember('public/a')
f.getmember('other/a')
def test_create_glob_not_exploitable(self):
# This tests for any traversal attack
with tempfile.TemporaryDirectory() as dir_name:
make_artifact_dir(dir_name)
archive = os.path.join(dir_name, 'artifacts.tar')
with self.assertRaises(RuntimeError):
create(archive, dir_name, ['/**'], None, None)
with tempfile.TemporaryDirectory() as dir_name:
make_artifact_dir(dir_name)
archive = os.path.join(dir_name, 'artifacts.tar')
with self.assertRaises(RuntimeError):
create(archive, os.path.join(dir_name, 'public'), ['../**'], None, None)
def test_create_out_of_scope(self):
with tempfile.TemporaryDirectory() as dir_name:
make_artifact_dir(dir_name)
archive = os.path.join(dir_name, 'artifacts.tar')
with self.assertRaises(RuntimeError):
create(archive, os.path.join(dir_name, 'public'), ['.', '../other'], None, None)
with self.assertRaises(RuntimeError):
create(archive, os.path.join(dir_name, 'public'), ['/home'], None, None)
def test_create_good_link(self):
with tempfile.TemporaryDirectory() as dir_name:
make_artifact_dir_link(dir_name)
archive = os.path.join(dir_name, 'artifacts.tar')
create(archive, dir_name, ['public', 'other'], None, None)
with tarfile.open(archive, 'r') as f:
f.getmember('public/a')
f.getmember('other/a')
def test_create_bad_link(self):
with tempfile.TemporaryDirectory() as dir_name:
make_artifact_dir_link(dir_name)
archive = os.path.join(dir_name, 'artifacts.tar')
with self.assertRaises(RuntimeError):
create(archive, os.path.join(dir_name, 'other'), ['.'], None, None)
def test_create_limited(self):
with tempfile.TemporaryDirectory() as dir_name:
make_artifact_dir(dir_name)
archive = os.path.join(dir_name, 'artifacts.tar')
with self.assertRaises(RuntimeError):
create(archive, dir_name, ['public', 'other'], 8, None)
def test_create_filtered(self):
with tempfile.TemporaryDirectory() as dir_name:
make_artifact_dir(dir_name)
archive = os.path.join(dir_name, 'artifacts.tar')
create(archive, dir_name, ['public', 'other'], None, ['other/a'])
with tarfile.open(archive, 'r') as f:
f.getmember('public/a')
f.getmember('other')
with self.assertRaises(KeyError):
f.getmember('other/a')
with tempfile.TemporaryDirectory() as dir_name:
make_artifact_dir(dir_name)
archive = os.path.join(dir_name, 'artifacts.tar')
create(archive, dir_name, ['public', 'other'], None, ['other'])
with tarfile.open(archive, 'r') as f:
f.getmember('public/a')
with self.assertRaises(KeyError):
f.getmember('other')
with self.assertRaises(KeyError):
f.getmember('other/a')
with tempfile.TemporaryDirectory() as dir_name:
make_artifact_dir(dir_name)
archive = os.path.join(dir_name, 'artifacts.tar')
create(archive, dir_name, ['public', 'other'], None, ['other/'])
with tarfile.open(archive, 'r') as f:
f.getmember('public/a')
with self.assertRaises(KeyError):
f.getmember('other')
with self.assertRaises(KeyError):
f.getmember('other/a')
with tempfile.TemporaryDirectory() as dir_name:
make_artifact_dir(dir_name)
archive = os.path.join(dir_name, 'artifacts.tar')
create(archive, dir_name, ['public', 'other'], None, ['other/a/'])
with tarfile.open(archive, 'r') as f:
f.getmember('public/a')
f.getmember('other')
f.getmember('other/a')
with tempfile.TemporaryDirectory() as dir_name:
make_artifact_dir(dir_name)
archive = os.path.join(dir_name, 'artifacts.tar')
create(archive, dir_name, ['public', 'other'], None, ['oth'])
with tarfile.open(archive, 'r') as f:
f.getmember('public/a')
f.getmember('other')
f.getmember('other/a')
with tempfile.TemporaryDirectory() as dir_name:
make_artifact_dir(dir_name)
archive = os.path.join(dir_name, 'artifacts.tar')
create(archive, dir_name, ['public', 'other'], None, ['**/a'])
with tarfile.open(archive, 'r') as f:
f.getmember('public')
f.getmember('other')
with self.assertRaises(KeyError):
f.getmember('public/a')
with self.assertRaises(KeyError):
f.getmember('other/a')
with tempfile.TemporaryDirectory() as dir_name:
make_artifact_dir(dir_name)
archive = os.path.join(dir_name, 'artifacts.tar')
create(archive, dir_name, ['public', 'other'], None, ['a'])
with tarfile.open(archive, 'r') as f:
f.getmember('public')
f.getmember('other')
f.getmember('public/a')
f.getmember('other/a')
def test_extract_bad(self):
with tempfile.TemporaryDirectory() as root_dir:
archive_file = make_bad_artifact_archive(root_dir)
with self.assertRaises(tarfile.FilterError):
extract(root_dir, archive_file)
if __name__ == '__main__':
unittest.main()

File Metadata

Mime Type
text/x-diff
Expires
Fri, Oct 9, 2:32 AM (1 d, 3 h)
Storage Engine
blob
Storage Format
Raw Data
Storage Handle
1784399
Default Alt Text
(38 KB)

Event Timeline