mirror of
https://github.com/ansible/awx.git
synced 2026-05-20 07:17:40 -02:30
Merge pull request #6404 from ryanpetrello/fix-6384
fix a few SSH key data validation issues for credential creation
This commit is contained in:
@@ -501,7 +501,10 @@ class CredentialInputField(JSONSchemaField):
|
|||||||
|
|
||||||
# `ssh_key_unlock` requirements are very specific and can't be
|
# `ssh_key_unlock` requirements are very specific and can't be
|
||||||
# represented without complicated JSON schema
|
# represented without complicated JSON schema
|
||||||
if model_instance.credential_type.kind == 'ssh':
|
if (
|
||||||
|
model_instance.credential_type.managed_by_tower is True and
|
||||||
|
'ssh_key_unlock' in model_instance.credential_type.defined_fields
|
||||||
|
):
|
||||||
if model_instance.has_encrypted_ssh_key_data and not value.get('ssh_key_unlock'):
|
if model_instance.has_encrypted_ssh_key_data and not value.get('ssh_key_unlock'):
|
||||||
errors['ssh_key_unlock'] = [_('must be set when SSH key is encrypted.')]
|
errors['ssh_key_unlock'] = [_('must be set when SSH key is encrypted.')]
|
||||||
if not model_instance.has_encrypted_ssh_key_data and value.get('ssh_key_unlock'):
|
if not model_instance.has_encrypted_ssh_key_data and value.get('ssh_key_unlock'):
|
||||||
|
|||||||
@@ -644,6 +644,7 @@ def scm(cls):
|
|||||||
'id': 'ssh_key_data',
|
'id': 'ssh_key_data',
|
||||||
'label': 'SCM Private Key',
|
'label': 'SCM Private Key',
|
||||||
'type': 'string',
|
'type': 'string',
|
||||||
|
'format': 'ssh_private_key',
|
||||||
'secret': True,
|
'secret': True,
|
||||||
'multiline': True
|
'multiline': True
|
||||||
}, {
|
}, {
|
||||||
@@ -694,6 +695,7 @@ def net(cls):
|
|||||||
'id': 'ssh_key_data',
|
'id': 'ssh_key_data',
|
||||||
'label': 'SSH Private Key',
|
'label': 'SSH Private Key',
|
||||||
'type': 'string',
|
'type': 'string',
|
||||||
|
'format': 'ssh_private_key',
|
||||||
'secret': True,
|
'secret': True,
|
||||||
'multiline': True
|
'multiline': True
|
||||||
}, {
|
}, {
|
||||||
@@ -871,6 +873,7 @@ def gce(cls):
|
|||||||
'id': 'ssh_key_data',
|
'id': 'ssh_key_data',
|
||||||
'label': 'RSA Private Key',
|
'label': 'RSA Private Key',
|
||||||
'type': 'string',
|
'type': 'string',
|
||||||
|
'format': 'ssh_private_key',
|
||||||
'secret': True,
|
'secret': True,
|
||||||
'multiline': True
|
'multiline': True
|
||||||
}]
|
}]
|
||||||
@@ -893,6 +896,7 @@ def azure(cls):
|
|||||||
'id': 'ssh_key_data',
|
'id': 'ssh_key_data',
|
||||||
'label': 'Management Certificate',
|
'label': 'Management Certificate',
|
||||||
'type': 'string',
|
'type': 'string',
|
||||||
|
'format': 'ssh_private_key',
|
||||||
'secret': True,
|
'secret': True,
|
||||||
'multiline': True
|
'multiline': True
|
||||||
}]
|
}]
|
||||||
|
|||||||
@@ -5,6 +5,9 @@ from awx.main.models.credential import Credential, CredentialType
|
|||||||
from awx.main.utils.common import decrypt_field
|
from awx.main.utils.common import decrypt_field
|
||||||
from awx.api.versioning import reverse
|
from awx.api.versioning import reverse
|
||||||
|
|
||||||
|
EXAMPLE_PRIVATE_KEY = '-----BEGIN PRIVATE KEY-----\nxyz==\n-----END PRIVATE KEY-----'
|
||||||
|
EXAMPLE_ENCRYPTED_PRIVATE_KEY = '-----BEGIN PRIVATE KEY-----\nProc-Type: 4,ENCRYPTED\nxyz==\n-----END PRIVATE KEY-----'
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.django_db
|
@pytest.mark.django_db
|
||||||
@pytest.mark.parametrize('kind, total', [
|
@pytest.mark.parametrize('kind, total', [
|
||||||
@@ -664,7 +667,7 @@ def test_inputs_cannot_contain_extra_fields(get, post, organization, admin, cred
|
|||||||
'name': 'Best credential ever',
|
'name': 'Best credential ever',
|
||||||
'username': 'some_username',
|
'username': 'some_username',
|
||||||
'password': 'some_password',
|
'password': 'some_password',
|
||||||
'ssh_key_data': 'some_key_data',
|
'ssh_key_data': EXAMPLE_ENCRYPTED_PRIVATE_KEY,
|
||||||
'ssh_key_unlock': 'some_key_unlock',
|
'ssh_key_unlock': 'some_key_unlock',
|
||||||
}],
|
}],
|
||||||
['v2', {
|
['v2', {
|
||||||
@@ -673,7 +676,7 @@ def test_inputs_cannot_contain_extra_fields(get, post, organization, admin, cred
|
|||||||
'inputs': {
|
'inputs': {
|
||||||
'username': 'some_username',
|
'username': 'some_username',
|
||||||
'password': 'some_password',
|
'password': 'some_password',
|
||||||
'ssh_key_data': 'some_key_data',
|
'ssh_key_data': EXAMPLE_ENCRYPTED_PRIVATE_KEY,
|
||||||
'ssh_key_unlock': 'some_key_unlock',
|
'ssh_key_unlock': 'some_key_unlock',
|
||||||
}
|
}
|
||||||
}]
|
}]
|
||||||
@@ -693,7 +696,7 @@ def test_scm_create_ok(post, organization, admin, version, params):
|
|||||||
cred = Credential.objects.all()[:1].get()
|
cred = Credential.objects.all()[:1].get()
|
||||||
assert cred.inputs['username'] == 'some_username'
|
assert cred.inputs['username'] == 'some_username'
|
||||||
assert decrypt_field(cred, 'password') == 'some_password'
|
assert decrypt_field(cred, 'password') == 'some_password'
|
||||||
assert decrypt_field(cred, 'ssh_key_data') == 'some_key_data'
|
assert decrypt_field(cred, 'ssh_key_data') == EXAMPLE_ENCRYPTED_PRIVATE_KEY
|
||||||
assert decrypt_field(cred, 'ssh_key_unlock') == 'some_key_unlock'
|
assert decrypt_field(cred, 'ssh_key_unlock') == 'some_key_unlock'
|
||||||
|
|
||||||
|
|
||||||
@@ -796,7 +799,7 @@ def test_vault_create_ok(post, organization, admin, version, params):
|
|||||||
'name': 'Best credential ever',
|
'name': 'Best credential ever',
|
||||||
'username': 'some_username',
|
'username': 'some_username',
|
||||||
'password': 'some_password',
|
'password': 'some_password',
|
||||||
'ssh_key_data': 'some_key_data',
|
'ssh_key_data': EXAMPLE_ENCRYPTED_PRIVATE_KEY,
|
||||||
'ssh_key_unlock': 'some_key_unlock',
|
'ssh_key_unlock': 'some_key_unlock',
|
||||||
'authorize': True,
|
'authorize': True,
|
||||||
'authorize_password': 'some_authorize_password',
|
'authorize_password': 'some_authorize_password',
|
||||||
@@ -807,7 +810,7 @@ def test_vault_create_ok(post, organization, admin, version, params):
|
|||||||
'inputs': {
|
'inputs': {
|
||||||
'username': 'some_username',
|
'username': 'some_username',
|
||||||
'password': 'some_password',
|
'password': 'some_password',
|
||||||
'ssh_key_data': 'some_key_data',
|
'ssh_key_data': EXAMPLE_ENCRYPTED_PRIVATE_KEY,
|
||||||
'ssh_key_unlock': 'some_key_unlock',
|
'ssh_key_unlock': 'some_key_unlock',
|
||||||
'authorize': True,
|
'authorize': True,
|
||||||
'authorize_password': 'some_authorize_password',
|
'authorize_password': 'some_authorize_password',
|
||||||
@@ -829,7 +832,7 @@ def test_net_create_ok(post, organization, admin, version, params):
|
|||||||
cred = Credential.objects.all()[:1].get()
|
cred = Credential.objects.all()[:1].get()
|
||||||
assert cred.inputs['username'] == 'some_username'
|
assert cred.inputs['username'] == 'some_username'
|
||||||
assert decrypt_field(cred, 'password') == 'some_password'
|
assert decrypt_field(cred, 'password') == 'some_password'
|
||||||
assert decrypt_field(cred, 'ssh_key_data') == 'some_key_data'
|
assert decrypt_field(cred, 'ssh_key_data') == EXAMPLE_ENCRYPTED_PRIVATE_KEY
|
||||||
assert decrypt_field(cred, 'ssh_key_unlock') == 'some_key_unlock'
|
assert decrypt_field(cred, 'ssh_key_unlock') == 'some_key_unlock'
|
||||||
assert decrypt_field(cred, 'authorize_password') == 'some_authorize_password'
|
assert decrypt_field(cred, 'authorize_password') == 'some_authorize_password'
|
||||||
assert cred.inputs['authorize'] is True
|
assert cred.inputs['authorize'] is True
|
||||||
@@ -885,7 +888,7 @@ def test_cloudforms_create_ok(post, organization, admin, version, params):
|
|||||||
'name': 'Best credential ever',
|
'name': 'Best credential ever',
|
||||||
'username': 'some_username',
|
'username': 'some_username',
|
||||||
'project': 'some_project',
|
'project': 'some_project',
|
||||||
'ssh_key_data': 'XYZ'
|
'ssh_key_data': EXAMPLE_PRIVATE_KEY,
|
||||||
}],
|
}],
|
||||||
['v2', {
|
['v2', {
|
||||||
'credential_type': 1,
|
'credential_type': 1,
|
||||||
@@ -893,7 +896,7 @@ def test_cloudforms_create_ok(post, organization, admin, version, params):
|
|||||||
'inputs': {
|
'inputs': {
|
||||||
'username': 'some_username',
|
'username': 'some_username',
|
||||||
'project': 'some_project',
|
'project': 'some_project',
|
||||||
'ssh_key_data': 'XYZ'
|
'ssh_key_data': EXAMPLE_PRIVATE_KEY,
|
||||||
}
|
}
|
||||||
}]
|
}]
|
||||||
])
|
])
|
||||||
@@ -912,7 +915,7 @@ def test_gce_create_ok(post, organization, admin, version, params):
|
|||||||
cred = Credential.objects.all()[:1].get()
|
cred = Credential.objects.all()[:1].get()
|
||||||
assert cred.inputs['username'] == 'some_username'
|
assert cred.inputs['username'] == 'some_username'
|
||||||
assert cred.inputs['project'] == 'some_project'
|
assert cred.inputs['project'] == 'some_project'
|
||||||
assert decrypt_field(cred, 'ssh_key_data') == 'XYZ'
|
assert decrypt_field(cred, 'ssh_key_data') == EXAMPLE_PRIVATE_KEY
|
||||||
|
|
||||||
|
|
||||||
#
|
#
|
||||||
@@ -924,14 +927,14 @@ def test_gce_create_ok(post, organization, admin, version, params):
|
|||||||
'kind': 'azure',
|
'kind': 'azure',
|
||||||
'name': 'Best credential ever',
|
'name': 'Best credential ever',
|
||||||
'username': 'some_username',
|
'username': 'some_username',
|
||||||
'ssh_key_data': 'XYZ'
|
'ssh_key_data': EXAMPLE_PRIVATE_KEY
|
||||||
}],
|
}],
|
||||||
['v2', {
|
['v2', {
|
||||||
'credential_type': 1,
|
'credential_type': 1,
|
||||||
'name': 'Best credential ever',
|
'name': 'Best credential ever',
|
||||||
'inputs': {
|
'inputs': {
|
||||||
'username': 'some_username',
|
'username': 'some_username',
|
||||||
'ssh_key_data': 'XYZ'
|
'ssh_key_data': EXAMPLE_PRIVATE_KEY
|
||||||
}
|
}
|
||||||
}]
|
}]
|
||||||
])
|
])
|
||||||
@@ -949,7 +952,7 @@ def test_azure_create_ok(post, organization, admin, version, params):
|
|||||||
assert Credential.objects.count() == 1
|
assert Credential.objects.count() == 1
|
||||||
cred = Credential.objects.all()[:1].get()
|
cred = Credential.objects.all()[:1].get()
|
||||||
assert cred.inputs['username'] == 'some_username'
|
assert cred.inputs['username'] == 'some_username'
|
||||||
assert decrypt_field(cred, 'ssh_key_data') == 'XYZ'
|
assert decrypt_field(cred, 'ssh_key_data') == EXAMPLE_PRIVATE_KEY
|
||||||
|
|
||||||
|
|
||||||
#
|
#
|
||||||
|
|||||||
@@ -211,6 +211,7 @@ def test_credential_creation_validation_failure(organization_factory, inputs):
|
|||||||
|
|
||||||
|
|
||||||
@pytest.mark.django_db
|
@pytest.mark.django_db
|
||||||
|
@pytest.mark.parametrize('kind', ['ssh', 'net', 'scm'])
|
||||||
@pytest.mark.parametrize('ssh_key_data, ssh_key_unlock, valid', [
|
@pytest.mark.parametrize('ssh_key_data, ssh_key_unlock, valid', [
|
||||||
[EXAMPLE_PRIVATE_KEY, None, True], # unencrypted key, no unlock pass
|
[EXAMPLE_PRIVATE_KEY, None, True], # unencrypted key, no unlock pass
|
||||||
[EXAMPLE_PRIVATE_KEY, 'super-secret', False], # unencrypted key, unlock pass
|
[EXAMPLE_PRIVATE_KEY, 'super-secret', False], # unencrypted key, unlock pass
|
||||||
@@ -221,14 +222,16 @@ def test_credential_creation_validation_failure(organization_factory, inputs):
|
|||||||
['INVALID-KEY-DATA', None, False], # invalid key data
|
['INVALID-KEY-DATA', None, False], # invalid key data
|
||||||
[EXAMPLE_PRIVATE_KEY.replace('=', '\u003d'), None, True], # automatically fix JSON-encoded GCE keys
|
[EXAMPLE_PRIVATE_KEY.replace('=', '\u003d'), None, True], # automatically fix JSON-encoded GCE keys
|
||||||
])
|
])
|
||||||
def test_ssh_key_data_validation(credentialtype_ssh, organization, ssh_key_data, ssh_key_unlock, valid):
|
def test_ssh_key_data_validation(organization, kind, ssh_key_data, ssh_key_unlock, valid):
|
||||||
inputs = {}
|
inputs = {}
|
||||||
if ssh_key_data:
|
if ssh_key_data:
|
||||||
inputs['ssh_key_data'] = ssh_key_data
|
inputs['ssh_key_data'] = ssh_key_data
|
||||||
if ssh_key_unlock:
|
if ssh_key_unlock:
|
||||||
inputs['ssh_key_unlock'] = ssh_key_unlock
|
inputs['ssh_key_unlock'] = ssh_key_unlock
|
||||||
|
cred_type = CredentialType.defaults[kind]()
|
||||||
|
cred_type.save()
|
||||||
cred = Credential(
|
cred = Credential(
|
||||||
credential_type=credentialtype_ssh,
|
credential_type=cred_type,
|
||||||
name="Best credential ever",
|
name="Best credential ever",
|
||||||
inputs=inputs,
|
inputs=inputs,
|
||||||
organization=organization
|
organization=organization
|
||||||
|
|||||||
Reference in New Issue
Block a user