def client(self, uid):
private_key_path, _ = self._users[uid]
client = SSHClient()
host_keys = client.get_host_keys()
key = paramiko.RSAKey.from_private_key_file(SERVER_KEY_PATH)
host_keys.add(self.host, "ssh-rsa", key)
host_keys.add("[%s]:%d" % (self.host, self.port), "ssh-rsa", key)
client.set_missing_host_key_policy(paramiko.RejectPolicy())
client.connect(
hostname=self.host,
port=self.port,
username=uid,
key_filename=private_key_path,
allow_agent=False,
look_for_keys=False
)
return client
python类RejectPolicy()的实例源码
def client(self, uid):
private_key_path, _ = self._users[uid]
c = paramiko.SSHClient()
host_keys = c.get_host_keys()
key = paramiko.RSAKey.from_private_key_file(SERVER_KEY_PATH)
host_keys.add(self.host, "ssh-rsa", key)
host_keys.add("[%s]:%d" % (self.host, self.port), "ssh-rsa", key)
c.set_missing_host_key_policy(paramiko.RejectPolicy())
c.connect(hostname=self.host,
port=self.port,
username=uid,
key_filename=private_key_path,
allow_agent=False,
look_for_keys=False)
return c
def _create_ssh(self, **kwargs):
try:
ssh = paramiko.SSHClient()
known_hosts_file = kwargs.get('known_hosts_file', None)
if known_hosts_file is None:
ssh.load_system_host_keys()
else:
# Make sure we can open the file for appending first.
# This is needed to create the file when we run CI tests with
# no existing key file.
open(known_hosts_file, 'a').close()
ssh.load_host_keys(known_hosts_file)
missing_key_policy = kwargs.get('missing_key_policy', None)
if missing_key_policy is None:
missing_key_policy = paramiko.AutoAddPolicy()
elif isinstance(missing_key_policy, basestring):
# To make it configurable, allow string to be mapped to object.
if missing_key_policy == paramiko.AutoAddPolicy().__class__.\
__name__:
missing_key_policy = paramiko.AutoAddPolicy()
elif missing_key_policy == paramiko.RejectPolicy().__class__.\
__name__:
missing_key_policy = paramiko.RejectPolicy()
elif missing_key_policy == paramiko.WarningPolicy().__class__.\
__name__:
missing_key_policy = paramiko.WarningPolicy()
else:
raise exceptions.SSHException(
"Invalid missing_key_policy: %s" % missing_key_policy
)
ssh.set_missing_host_key_policy(missing_key_policy)
self.ssh = ssh
except Exception as e:
msg = "Error connecting via ssh: %s" % e
self._logger.error(msg)
raise paramiko.SSHException(msg)
test_HPELeftHandClient_MockSSH.py 文件源码
项目:deb-python-hplefthandclient
作者: openstack
项目源码
文件源码
阅读 29
收藏 0
点赞 0
评论 0
def test_reject_policy(self):
known_hosts_file = "test_bogus_known_hosts_file"
missing_key_policy = "RejectPolicy"
self.base(known_hosts_file, missing_key_policy)
test_HPELeftHandClient_MockSSH.py 文件源码
项目:deb-python-hplefthandclient
作者: openstack
项目源码
文件源码
阅读 24
收藏 0
点赞 0
评论 0
def test_known_hosts_file_is_none(self):
known_hosts_file = None
missing_key_policy = paramiko.RejectPolicy()
self.base(known_hosts_file, missing_key_policy)
def open(self, host, port=22, username=None, password=None,
key_filename=None, pkey=None, look_for_keys=None,
allow_agent=False, key_policy="loose"):
self.ssh = paramiko.SSHClient()
if key_policy != "ignore":
self.ssh.load_system_host_keys()
try:
self.ssh.load_host_keys(os.path.expanduser('~/.ssh/known_hosts'))
except IOError:
pass
if key_policy == "strict":
self.ssh.set_missing_host_key_policy(paramiko.RejectPolicy())
else:
self.ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
# unless explicitly set, disable look for keys if a password is
# present. this changes the default search order paramiko implements
if not look_for_keys:
look_for_keys = password is None
try:
self.ssh.connect(
host, port=port, username=username, password=password,
timeout=self._timeout, look_for_keys=look_for_keys, pkey=pkey,
key_filename=key_filename, allow_agent=allow_agent,
)
self.shell = self.ssh.invoke_shell()
self.shell.settimeout(self._timeout)
except socket.gaierror:
raise ShellError("unable to resolve host name")
except AuthenticationException:
raise ShellError('Unable to authenticate to remote device')
except socket.timeout:
raise ShellError("timeout trying to connect to remote device")
except socket.error:
exc = get_exception()
if exc.errno == 60:
raise ShellError('timeout trying to connect to host')
raise
if self.kickstart:
self.shell.sendall("\n")
self.receive()
def open(self, host, port=22, username=None, password=None, timeout=10,
key_filename=None, pkey=None, look_for_keys=None,
allow_agent=False, key_policy="loose"):
self.ssh = paramiko.SSHClient()
if key_policy != "ignore":
self.ssh.load_system_host_keys()
try:
self.ssh.load_host_keys(os.path.expanduser('~/.ssh/known_hosts'))
except IOError:
pass
if key_policy == "strict":
self.ssh.set_missing_host_key_policy(paramiko.RejectPolicy())
else:
self.ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
# unless explicitly set, disable look for keys if a password is
# present. this changes the default search order paramiko implements
if not look_for_keys:
look_for_keys = password is None
try:
self.ssh.connect(
host, port=port, username=username, password=password,
timeout=timeout, look_for_keys=look_for_keys, pkey=pkey,
key_filename=key_filename, allow_agent=allow_agent,
)
self.shell = self.ssh.invoke_shell()
self.shell.settimeout(timeout)
except socket.gaierror:
raise ShellError("unable to resolve host name")
except AuthenticationException:
raise ShellError('Unable to authenticate to remote device')
except socket.timeout:
raise ShellError("timeout trying to connect to remote device")
except socket.error:
exc = get_exception()
if exc.errno == 60:
raise ShellError('timeout trying to connect to host')
raise
if self.kickstart:
self.shell.sendall("\n")
self.receive()
def open(self, host, port=22, username=None, password=None,
key_filename=None, pkey=None, look_for_keys=None,
allow_agent=False, key_policy="loose"):
self.ssh = paramiko.SSHClient()
if key_policy != "ignore":
self.ssh.load_system_host_keys()
try:
self.ssh.load_host_keys(os.path.expanduser('~/.ssh/known_hosts'))
except IOError:
pass
if key_policy == "strict":
self.ssh.set_missing_host_key_policy(paramiko.RejectPolicy())
else:
self.ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
# unless explicitly set, disable look for keys if a password is
# present. this changes the default search order paramiko implements
if not look_for_keys:
look_for_keys = password is None
try:
self.ssh.connect(
host, port=port, username=username, password=password,
timeout=self._timeout, look_for_keys=look_for_keys, pkey=pkey,
key_filename=key_filename, allow_agent=allow_agent,
)
self.shell = self.ssh.invoke_shell()
self.shell.settimeout(self._timeout)
except socket.gaierror:
raise ShellError("unable to resolve host name")
except AuthenticationException:
raise ShellError('Unable to authenticate to remote device')
except socket.timeout:
raise ShellError("timeout trying to connect to remote device")
except socket.error as exc:
if exc.errno == 60:
raise ShellError('timeout trying to connect to host')
raise
if self.kickstart:
self.shell.sendall("\n")
self.receive()