Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
if isinstance(config, six.string_types):
if '\n' in config:
config = yaml.safe_load(config)
else:
with open(config) as f:
config = yaml.safe_load(f.read())
config = self._process_config(config)
for duct_config in config:
names = duct_config.pop('name')
protocol = duct_config.pop('protocol')
register_magics = duct_config.pop('register_magics', True)
try:
self.new(names, protocol, register_magics=register_magics, override=override, **duct_config)
except DuctProtocolUnknown as e:
logger.error("Failed to configure `Duct` instance(s) '{}'. {}".format("', '".join(names.split(',')), str(e)))
return self
message = ast.literal_eval(re.match("[^{]*({.*})[^}]*$", message).group(1))
linenumber = message['errorLocation']['lineNumber'] - 1
splt = statement.splitlines()
splt[linenumber] += ' <-- {errorType} ({errorName}) occurred. {message} '.format(**message)
context = '\n\n[Error Context]\n{}\n'.format('\n'.join([splt[l] for l in range(max(linenumber - 1, 0),
min(linenumber + 2, len(splt)))]))
class ErrContext(object):
def __repr__(self):
return context
# logged twice so that both notebook and console users see the error context
exception_args.args = [exception_args, ErrContext()]
logger.error(context)
except:
logger.warn(("Omniduct was unable to parse the database error messages. Refer to the "
"traceback below for full error details."))
if isinstance(exception, type):
exception = exception(exception_args)
raise_with_traceback(exception, traceback)
expect = pexpect.spawn(cmd)
i = expect.expect(expected, timeout=10)
# First phase
if i == 0: # If host identification changed, arrest any further attempts to connect
error_message = (
'Host identification for {} has changed! This is most likely '
'due to the the server being redeployed or reconfigured but '
'may also be due to a man-in-the-middle attack. If you trust '
'your network connection, you should be safe to update the '
'host keys for this host. To do this manually, please remove '
'the line corresponding to this host in ~/.ssh/known_hosts; '
'or call the `update_host_keys` method of this client.'.format(self._host)
)
if self.interactive:
logger.error(error_message)
auto_fix = input('Would you like this client to do this for you? (y/n)')
if auto_fix == 'y':
self.update_host_keys()
return self.connect()
else:
raise RuntimeError("Host keys not updated. Please update keys manually.")
else:
raise RuntimeError(error_message)
if i == 1: # Request to authorize host certificate (i.e. host not in the 'known_hosts' file)
expect.sendline("yes")
i = self.expect(expected)
if i == 2: # Request for password/passphrase
expect.sendline(self.password or getpass.getpass('Password: '))
i = self.expect(expected)
if i == 4: # Request for terminal type
expect.sendline('ascii')