Python paramiko 模块，RejectPolicy() 实例源码

我们从Python开源项目中，提取了以下8个代码示例，用于说明如何使用paramiko.RejectPolicy()。

项目：sqlalchemy-media 作者：pylover | 项目源码 | 文件源码

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

项目：mock-ssh-server 作者：carletes | 项目源码 | 文件源码

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

项目：deb-python-hplefthandclient 作者：openstack | 项目源码 | 文件源码

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)

项目：deb-python-hplefthandclient 作者：openstack | 项目源码 | 文件源码

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)

项目：deb-python-hplefthandclient 作者：openstack | 项目源码 | 文件源码

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)

项目：iosxr-ansible 作者：ios-xr | 项目源码 | 文件源码

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()

项目：DevOps 作者：YoLoveLife | 项目源码 | 文件源码

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()

项目：ansible-provider-docs 作者：alibaba | 项目源码 | 文件源码

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()