External-Mirrors/awx
2
0
Fork 0
mirror of https://github.com/ansible/awx.git synced 2026-01-15 11:50:42 -03:30
Code Issues Packages Projects Releases Wiki Activity

only allow facts to cache in the proper file system location

Browse Source
This commit is contained in:
Ryan Petrello 2018-01-15 11:38:02 -05:00
parent 983b192a45
commit e1d50a43fd
No known key found for this signature in database
GPG Key ID: F2AA5F2122351777
2 changed files with 17 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
awx/main/models/jobs.py
View File

@ -731,6 +731,9 @@ class Job(UnifiedJob, JobOptions, SurveyJobMixin, JobNotificationMixin, TaskMana
hosts = hosts.filter(ansible_facts_modified__gte=timeout)
for host in hosts:
filepath = os.sep.join(map(six.text_type, [destination, host.name]))
if not os.path.realpath(filepath).startswith(destination):
system_tracking_logger.error('facts for host {} could not be cached'.format(smart_str(host.name)))
continue
with codecs.open(filepath, 'w', encoding='utf-8') as f:
os.chmod(f.name, 0600)
json.dump(host.ansible_facts, f)
@ -741,6 +744,9 @@ class Job(UnifiedJob, JobOptions, SurveyJobMixin, JobNotificationMixin, TaskMana
destination = os.path.join(destination, 'facts')
for host in self._get_inventory_hosts():
filepath = os.sep.join(map(six.text_type, [destination, host.name]))
if not os.path.realpath(filepath).startswith(destination):
system_tracking_logger.error('facts for host {} could not be cached'.format(smart_str(host.name)))
continue
if os.path.exists(filepath):
# If the file changed since we wrote it pre-playbook run...
modified = os.path.getmtime(filepath)

11
awx/main/tests/unit/models/test_jobs.py
View File

@ -47,6 +47,17 @@ def test_start_job_fact_cache(hosts, job, inventory, tmpdir):
assert filepath in modified_times
def test_fact_cache_with_invalid_path_traversal(job, inventory, tmpdir, mocker):
job._get_inventory_hosts = mocker.Mock(return_value=[
Host(name='../foo', ansible_facts={"a": 1, "b": 2},),
])
fact_cache = str(tmpdir)
job.start_job_fact_cache(fact_cache, {}, 0)
# a file called "foo" should _not_ be written outside the facts dir
assert os.listdir(os.path.join(fact_cache, 'facts', '..')) == ['facts']
def test_finish_job_fact_cache_with_existing_data(job, hosts, inventory, mocker, tmpdir):
fact_cache = str(tmpdir)
modified_times = {}