Files
awx/awx/main/credential_plugins/conjur.py
Ryan Petrello 310a0f88e5 remove the usage of create_temporary_fifo from credential plugins
this resolves an issue that causes an endless hang on with Cyberark AIM
lookups when a certificate *and* key are specified

the underlying issue here is that we can't rely on the underyling Python
ssl implementation to *only* read from the fifo that stores the pem data
*only once*; in reality, we need to just use *actual* tempfiles for
stability purposes

see: https://github.com/ansible/awx/issues/6986
see: https://github.com/urllib3/urllib3/issues/1880
2020-07-14 16:24:03 -04:00

102 lines
2.7 KiB
Python

from .plugin import CredentialPlugin, CertFiles
import base64
from urllib.parse import urljoin, quote_plus
from django.utils.translation import ugettext_lazy as _
import requests
conjur_inputs = {
'fields': [{
'id': 'url',
'label': _('Conjur URL'),
'type': 'string',
'format': 'url',
}, {
'id': 'api_key',
'label': _('API Key'),
'type': 'string',
'secret': True,
}, {
'id': 'account',
'label': _('Account'),
'type': 'string',
}, {
'id': 'username',
'label': _('Username'),
'type': 'string',
}, {
'id': 'cacert',
'label': _('Public Key Certificate'),
'type': 'string',
'multiline': True
}],
'metadata': [{
'id': 'secret_path',
'label': _('Secret Identifier'),
'type': 'string',
'help_text': _('The identifier for the secret e.g., /some/identifier'),
}, {
'id': 'secret_version',
'label': _('Secret Version'),
'type': 'string',
'help_text': _('Used to specify a specific secret version (if left empty, the latest version will be used).'),
}],
'required': ['url', 'api_key', 'account', 'username'],
}
def conjur_backend(**kwargs):
url = kwargs['url']
api_key = kwargs['api_key']
account = quote_plus(kwargs['account'])
username = quote_plus(kwargs['username'])
secret_path = quote_plus(kwargs['secret_path'])
version = kwargs.get('secret_version')
cacert = kwargs.get('cacert', None)
auth_kwargs = {
'headers': {'Content-Type': 'text/plain'},
'data': api_key,
'allow_redirects': False,
}
with CertFiles(cacert) as cert:
# https://www.conjur.org/api.html#authentication-authenticate-post
auth_kwargs['verify'] = cert
resp = requests.post(
urljoin(url, '/'.join(['authn', account, username, 'authenticate'])),
**auth_kwargs
)
resp.raise_for_status()
token = base64.b64encode(resp.content).decode('utf-8')
lookup_kwargs = {
'headers': {'Authorization': 'Token token="{}"'.format(token)},
'allow_redirects': False,
}
# https://www.conjur.org/api.html#secrets-retrieve-a-secret-get
path = urljoin(url, '/'.join([
'secrets',
account,
'variable',
secret_path
]))
if version:
path = '?'.join([path, version])
with CertFiles(cacert) as cert:
lookup_kwargs['verify'] = cert
resp = requests.get(path, timeout=30, **lookup_kwargs)
resp.raise_for_status()
return resp.text
conjur_plugin = CredentialPlugin(
'CyberArk Conjur Secret Lookup',
inputs=conjur_inputs,
backend=conjur_backend
)