github
/
jumpserver
mirror of https://github.com/jumpserver/jumpserver
1
0
Fork 0
Code Issues Releases Wiki Activity

Revert "fix: 特定key paramiko 测试可连接性失败" 

This reverts commit a4d0e3fd17.
pull/12451/head
feng626 2023-12-28 17:31:44 +08:00 committed by Bryan
parent b72a446bbd
commit 7a708156ee
1 changed files with 3 additions and 23 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

26
apps/ops/ansible/modules_utils/custom_common.py
View File

@ -2,11 +2,8 @@ import re
import time
import paramiko
from paramiko import DSSKey, RSAKey, Ed25519Key, ECDSAKey
from sshtunnel import SSHTunnelForwarder
KEY_CLASSES = (RSAKey, DSSKey, ECDSAKey, Ed25519Key)
def common_argument_spec():
options = dict(
@ -42,38 +39,21 @@ class SSHClient:
self.client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
self.connect_params = self.get_connect_params()
@staticmethod
def get_encrypt_cls(key_path):
for key_cls in KEY_CLASSES:
try:
key_cls.from_private_key_file(key_path)
return key_cls
except paramiko.SSHException:
continue
raise paramiko.SSHException('Invalid key file')
def get_pkey(self, key_path):
if not key_path:
return None
key_cls = self.get_encrypt_cls(key_path)
return key_cls.from_private_key_file(key_path)
def get_connect_params(self):
params = {
'allow_agent': False, 'look_for_keys': False,
'hostname': self.module.params['login_host'],
'port': self.module.params['login_port'],
# TODO: https://github.com/paramiko/paramiko/issues/2048
'pkey': self.get_pkey(self.module.params['login_private_key_path'])
'key_filename': self.module.params['login_private_key_path'] or None
}
if self.module.params['become']:
params['username'] = self.module.params['become_user']
params['password'] = self.module.params['become_password']
params['pkey'] = self.get_pkey(self.module.params['become_private_key_path'])
params['key_filename'] = self.module.params['become_private_key_path'] or None
else:
params['username'] = self.module.params['login_user']
params['password'] = self.module.params['login_password']
params['pkey'] = self.get_pkey(self.module.params['login_private_key_path'])
params['key_filename'] = self.module.params['login_private_key_path'] or None
return params
def _get_channel(self):