External-Mirrors/awx
2
0
Fork 0
mirror of https://github.com/ansible/awx.git synced 2026-01-13 19:10:07 -03:30
Code Issues Packages Projects Releases Wiki Activity

AC-672 Fixed responding to passphrase prompts for project updates, re-added tests removed in earlier update.

Browse Source
This commit is contained in:
Chris Church 2013-11-19 13:18:04 -05:00
parent ee44f1ea76
commit 216ab5cf41
2 changed files with 59 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

34
awx/main/tasks.py
View File

@ -27,6 +27,7 @@ from celery import Task
# Django
from django.conf import settings
from django.db import transaction
from django.utils.datastructures import SortedDict
from django.utils.timezone import now
# AWX
@ -152,10 +153,7 @@ class BaseTask(Task):
Return a dictionary of prompt regular expressions and password lookup
keys.
'''
return {
r'Enter passphrase for .*:': 'ssh_key_unlock',
r'Bad passphrase, try again for .*:': '',
}
return SortedDict()
def run_pexpect(self, instance, args, cwd, env, passwords,
output_replacements=None):
@ -180,6 +178,7 @@ class BaseTask(Task):
expect_list.extend([pexpect.TIMEOUT, pexpect.EOF])
while child.isalive():
result_id = child.expect(expect_list, timeout=pexpect_timeout)
#print 'pexpect result_id', result_id, expect_list[result_id], expect_passwords.get(result_id, None)
if result_id in expect_passwords:
child.sendline(expect_passwords[result_id])
updates = {'status': 'running',
@ -407,11 +406,11 @@ class RunJob(BaseTask):
def get_password_prompts(self):
d = super(RunJob, self).get_password_prompts()
d.update({
r'sudo password.*:': 'sudo_password',
r'SSH password:': 'password',
r'Password:': 'password',
})
d[re.compile(r'^Enter passphrase for .*:\s*?$', re.M)] = 'ssh_key_unlock'
d[re.compile(r'^Bad passphrase, try again for .*:\s*?$', re.M)] = ''
d[re.compile(r'^sudo password.*:\s*?$', re.M)] = 'sudo_password'
d[re.compile(r'^SSH password:\s*?$', re.M)] = 'password'
d[re.compile(r'^Password:\s*?$', re.M)] = 'password'
return d
def pre_run_check(self, job, **kwargs):
@ -663,15 +662,14 @@ class RunProjectUpdate(BaseTask):
def get_password_prompts(self):
d = super(RunProjectUpdate, self).get_password_prompts()
d.update({
re.compile(r'^Username for.*:\s*?$', re.M): 'scm_username',
re.compile(r'^Password for.*:\s*?$', re.M): 'scm_password',
re.compile(r'^Password:\s*?$', re.M): 'scm_password',
re.compile(r'^\S+?@\S+?\'s\s+?password:\s*?$', re.M): 'scm_password',
re.compile(r'^Enter passphrase for .*:\s*?$', re.M): 'scm_key_unlock',
# FIXME: Configure whether we should auto accept host keys?
re.compile(r'^Are you sure you want to continue connecting \(yes/no\)\?\s*?$', re.M): 'yes',
})
d[re.compile(r'^Username for.*:\s*?$', re.M)] = 'scm_username'
d[re.compile(r'^Password for.*:\s*?$', re.M)] = 'scm_password'
d[re.compile(r'^Password:\s*?$', re.M)] = 'scm_password'
d[re.compile(r'^\S+?@\S+?\'s\s+?password:\s*?$', re.M)] = 'scm_password'
d[re.compile(r'^Enter passphrase for .*:\s*?$', re.M)] = 'scm_key_unlock'
d[re.compile(r'^Bad passphrase, try again for .*:\s*?$', re.M)] = ''
# FIXME: Configure whether we should auto accept host keys?
d[re.compile(r'^Are you sure you want to continue connecting \(yes/no\)\?\s*?$', re.M)] = 'yes'
return d
def get_idle_timeout(self):

43
awx/main/tests/projects.py
View File

@ -1220,6 +1220,49 @@ class ProjectUpdatesTest(BaseTransactionTest):
self.check_project_update(project2, should_fail=None)#,
#should_error=should_error)
def test_scm_key_unlock_on_project_update(self):
scm_url = 'git@github.com:ansible/ansible.github.com.git'
project = self.create_project(
name='my git project over ssh with encrypted key',
scm_type='git',
scm_url=scm_url,
scm_key_data=TEST_SSH_KEY_DATA_LOCKED,
scm_key_unlock=TEST_SSH_KEY_DATA_UNLOCK,
)
url = reverse('api:project_update_view', args=(project.pk,))
with self.current_user(self.super_django_user):
response = self.get(url, expect=200)
self.assertTrue(response['can_update'])
with self.current_user(self.super_django_user):
response = self.post(url, {}, expect=202)
project_update = project.project_updates.order_by('-pk')[0]
self.check_project_update(project, should_fail=None,
project_update=project_update)
# Verify that we responded to ssh-agent prompt.
self.assertTrue('Identity added' in project_update.result_stdout,
project_update.result_stdout)
# Try again with a bad unlock password.
project = self.create_project(
name='my git project over ssh with encrypted key and bad pass',
scm_type='git',
scm_url=scm_url,
scm_key_data=TEST_SSH_KEY_DATA_LOCKED,
scm_key_unlock='not the right password',
)
with self.current_user(self.super_django_user):
response = self.get(url, expect=200)
self.assertTrue(response['can_update'])
with self.current_user(self.super_django_user):
response = self.post(url, {}, expect=202)
project_update = project.project_updates.order_by('-pk')[0]
self.check_project_update(project, should_fail=None,
project_update=project_update)
# Verify response to ssh-agent prompt, did not accept password.
self.assertTrue('Bad passphrase' in project_update.result_stdout,
project_update.result_stdout)
self.assertFalse('Identity added' in project_update.result_stdout,
project_update.result_stdout)
def create_local_git_repo(self):
repo_dir = tempfile.mkdtemp()
self._temp_project_dirs.append(repo_dir)