Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def __init__(self, config):
"""Set up platform."""
super(LXDPlatform, self).__init__(config)
# TODO: allow configuration of remote lxd host via env variables
# set up lxd connection
self.client = Client()
def get_client():
""" Returns a PyLXD client to be used to orchestrate containers. """
return pylxd.Client()
pool_key = '|'.join((six.text_type(remote_addr),
six.text_type(cert),
six.text_type(key),
six.text_type(verify_cert),))
if pool_key in _connection_pool:
log.debug((
'Returning the client "{0}" from our connection pool'
).format(remote_addr))
return _connection_pool[pool_key]
try:
if remote_addr is None or remote_addr == '/var/lib/lxd/unix.socket':
log.debug('Trying to connect to the local unix socket')
client = pylxd.Client()
else:
if remote_addr.startswith('/'):
client = pylxd.Client(remote_addr)
else:
if cert is None or key is None:
raise SaltInvocationError(
('You have to give a Cert and '
'Key file for remote endpoints.')
)
cert = os.path.expanduser(cert)
key = os.path.expanduser(key)
if not os.path.isfile(cert):
raise SaltInvocationError(
('You have given an invalid cert path: "{0}", '
if not os.path.isfile(key):
raise SaltInvocationError(
('You have given an invalid key path: "{0}", '
'the file does not exists or is not a file.').format(
key
)
)
log.debug((
'Trying to connecto to "{0}" '
'with cert "{1}", key "{2}" and '
'verify_cert "{3!s}"'.format(
remote_addr, cert, key, verify_cert)
))
client = pylxd.Client(
endpoint=remote_addr,
cert=(cert, key,),
verify=verify_cert
)
except pylxd.exceptions.ClientConnectionFailed:
raise CommandExecutionError(
"Failed to connect to '{0}'".format(remote_addr)
)
except TypeError as e:
# Happens when the verification failed.
raise CommandExecutionError(
('Failed to connect to "{0}",'
' looks like the SSL verification failed, error was: {1}'
).format(remote_addr, six.text_type(e))
)
def __init__(self, app, job, db, log_file, config):
ImageBuilder.__init__(self, app, job, db, log_file, config)
# Always use localhost to publish built Images and run containers
self._client = LXDClient()
self._source_hooks_path = ''
self._container_name = self._ami_name.replace('.', '-')
self._container_config = self._config.get('container', {
'endpoint': self._config.get('endpoint', 'https://lxd.ghost.morea.fr:8443'),
'debug': self._config.get('debug', False),
})
self._config['ghost_venv'] = sys.exec_prefix
provisioners_config = get_provisioners_config()
self.provisioners = []
for key, provisioner_config in provisioners_config.iteritems():
if key == 'salt':
self.provisioners.append('salt')
elif key == 'ansible':
self.provisioners.append('ansible')
else:
def lxd_is_available(config=None):
"""
Test if lxd is available on system and test if remote lxd endpoint is available
"""
config = config or {}
try:
lxd_local = LXDClient()
container_config = config.get('container', {'endpoint': config.get('endpoint',
DEFAULT_LXD_REMOTE_ENDPOINT)})
if container_config.get('endpoint', 'localhost') != "localhost":
lxd = LXDClient(endpoint=container_config.get('endpoint', DEFAULT_LXD_REMOTE_ENDPOINT), verify=True)
except:
return False
return True
six.text_type(key),
six.text_type(verify_cert),))
if pool_key in _connection_pool:
log.debug((
'Returning the client "{0}" from our connection pool'
).format(remote_addr))
return _connection_pool[pool_key]
try:
if remote_addr is None or remote_addr == '/var/lib/lxd/unix.socket':
log.debug('Trying to connect to the local unix socket')
client = pylxd.Client()
else:
if remote_addr.startswith('/'):
client = pylxd.Client(remote_addr)
else:
if cert is None or key is None:
raise SaltInvocationError(
('You have to give a Cert and '
'Key file for remote endpoints.')
)
cert = os.path.expanduser(cert)
key = os.path.expanduser(key)
if not os.path.isfile(cert):
raise SaltInvocationError(
('You have given an invalid cert path: "{0}", '
'the file does not exists or is not a file.').format(
cert
)