我有以下代码:
import paramiko
policy = paramiko.client.WarningPolicy()
client = paramiko.client.SSHClient()
client.set_missing_host_key_policy(policy)
username = '...'
password = '...'
file_path = '...'
pkey = paramiko.RSAKey.from_private_key_file(file_path)
client.connect('...', username=username, password=password, pkey=key)
sftp = client.open_sftp()
从文档来看,它似乎应该有效。一切都成功,但是当代码点击
client.open_sftp
时,它会发出 SSHException: Unable to open channel.
的轰炸,并且传输(来自 client.get_transport
)处于活动状态,但未经过身份验证。我也无法为此启用调试日志记录(我正在尝试 logging.getLogger('paramiko').setLevel(logging.DEBUG)
但没有成功。)
关于我可以从哪里开始调试这个非常模糊的错误消息有什么想法吗?
很抱歉回复晚了,但这个问题真的很难找到任何信息,所以我想为其他陷入此问题的人发布解决方案。
在绞尽脑汁试图解决这个问题之后,我找到了一个解决方案,这要归功于 Doug Ellwanger 和 Daniel Brownridge 发布的一些代码。该问题似乎是由使用更多交互风格处理多重身份验证的方式引起的。
import paramiko
import threading
...
username = '...'
password = '...'
file_path = '...'
pkey = paramiko.RSAKey.from_private_key_file(file_path)
sftpClient = multifactor_auth('...', 22, username, pkey, password)
...
def multifactor_auth_sftp_client(host, port, username, key, password):
#Create an SSH transport configured to the host
transport = paramiko.Transport((host, port))
#Negotiate an SSH2 session
transport.connect()
#Attempt authenticating using a private key
transport.auth_publickey(username, key)
#Create an event for password auth
password_auth_event = threading.Event()
#Create password auth handler from transport
password_auth_handler = paramiko.auth_handler.AuthHandler(transport)
#Set transport auth_handler to password handler
transport.auth_handler = password_auth_handler
#Aquire lock on transport
transport.lock.acquire()
#Register the password auth event with handler
password_auth_handler.auth_event = password_auth_event
#Set the auth handler method to 'password'
password_auth_handler.auth_method = 'password'
#Set auth handler username
password_auth_handler.username = username
#Set auth handler password
password_auth_handler.password = password
#Create an SSH user auth message
userauth_message = paramiko.message.Message()
userauth_message.add_string('ssh-userauth')
userauth_message.rewind()
#Make the password auth attempt
password_auth_handler._parse_service_accept(userauth_message)
#Release lock on transport
transport.lock.release()
#Wait for password auth response
password_auth_handler.wait_for_response(password_auth_event)
#Create an open SFTP client channel
return transport.open_sftp_client()
我希望这有帮助,它对我的项目有用。
在你的脚本中声明
pkey = paramiko.RSAKey.from_private_key_file(file_path)
然后您将拥有
pkey = key
,而不是 pkey。
不确定密钥来自哪个,但这可能是一个问题。
我能够在 paramiko 的 github 上使用这个解决方案,该解决方案是从一个仍然悬而未决的问题中找到的(在撰写本文时)。我将包含该 github 问题的原始解决方案,以及我在自己的代码中使用的轻微变体。
来自 jacky15 的 github 问题:
import paramiko
paramiko.util.log_to_file('/path/to/log')
hostname = 'server.name'
port = 12345
username = 'username'
password = 'password'
pkey = paramiko.RSAKey.from_private_key_file('/path/to/key')
transport = paramiko.Transport((hostname, port))
transport.connect()
# auth the public key as usual, auth service now is activated on server
transport.auth_publickey(username=username, key=pkey)
# try to send another userauth request without request auth service
m = paramiko.Message()
m.add_byte(paramiko.common.cMSG_USERAUTH_REQUEST)
m.add_string(username)
m.add_string('ssh-connection')
m.add_string('password')
m.add_boolean(False)
py3_password = paramiko.py3compat.bytestring(password)
m.add_string(py3_password)
transport._send_message(m)
# now it works! : )
sftp_client = paramiko.SFTPClient.from_transport(transport)
来自我的代码:
transport = paramiko.Transport((self.hostname, self.port))
transport.start_client(event=None, timeout=30)
transport.get_remote_server_key()
my_key = paramiko.RSAKey.from_private_key_file(self.hostkey)
transport.auth_publickey(self.username, my_key)
m = paramiko.Message()
m.add_byte(paramiko.common.cMSG_USERAUTH_REQUEST)
m.add_string(self.username)
m.add_string('ssh-connection')
m.add_string('password')
m.add_boolean(False)
py3_password = paramiko.py3compat.bytestring(self.password)
m.add_string(py3_password)
transport._send_message(m)
self.connection = paramiko.SFTPClient.from_transport(transport)
这是一个示例代码,首先使用“密码”然后使用“公钥”进行身份验证
import threading
import paramiko
# Define server credentials
hostname = '127.0.0.1'
port = 22
username = 'username'
password = 'password'
private_key_path = '/path/to/id_rsa'
try:
# Initialize transport
transport = paramiko.Transport((hostname, port), disabled_algorithms={'pubkeys': ['rsa-sha2-256', 'rsa-sha2-512']})
transport.connect()
# First authenticate with username and password
transport.auth_password(username=username, password=password)
print("Password authentication successful.")
# Next authenticate with the ssh public key
# Load the private key
private_key = paramiko.RSAKey.from_private_key_file(private_key_path)
publickey_auth_event = threading.Event()
publickey_auth_handler = paramiko.auth_handler.AuthHandler(transport)
transport.auth_handler = publickey_auth_handler
transport.lock.acquire()
publickey_auth_handler.auth_event = publickey_auth_event
publickey_auth_handler.auth_method = "publickey"
publickey_auth_handler.username = username
publickey_auth_handler.private_key = private_key
userauth_message = paramiko.message.Message()
userauth_message.add_string('ssh-userauth')
userauth_message.rewind()
publickey_auth_handler._parse_service_accept(userauth_message)
transport.lock.release()
publickey_auth_handler.wait_for_response(publickey_auth_event)
print("Public key authentication successful.")
# Open an SFTP session
sftp = paramiko.SFTPClient.from_transport(transport)
print("SFTP session established.")
# List files in the remote directory
print("Remote files:", sftp.listdir())
# Close the connection
sftp.close()
transport.close()
except Exception as e:
print(f"Authentication failed: {e}")
Paramiko 高级 API
SSHClient
可以自行处理常见的双因素身份验证。例如,对于密钥和密码身份验证,请使用:
ssh = paramiko.SSHClient()
ssh.connect(
"example.com", username="username", password="password",
key_filename="/path/to/key")
因此,@osekmedia 答案中的复杂代码通常是不需要的。
我知道只有两种情况可以“提供帮助”:
SSHClient
默认情况下验证主机密钥。您可能会将主机密钥验证失败误认为是双因素身份验证失败。他们没有关系。只是 @osekmedia 代码使用的低级 Transport
API 不会验证主机密钥,这避免了您的实际问题。但这是一个安全缺陷。正确的解决方案请参阅Paramiko“未知服务器”。
您可能认为您正在使用密码身份验证,而实际上您使用的是键盘交互式身份验证。通常情况下,Paramiko 可以处理键盘交互式身份验证,即使您错误地要求密码身份验证。但对于一些不起眼的服务器,这不起作用,请参阅Python Paramiko 中的密码身份验证失败,但相同的凭据在 SSH/SFTP 客户端中工作。在这种情况下,应该执行以下代码:
username = "username"
transport = paramiko.Transport('example.com')
transport.connect(username=username)
key = paramiko.RSAKey.from_private_key_file("/path/to/key")
transport.auth_publickey(username, key)
def handler(title, instructions, fields):
if len(fields) > 1:
raise SSHException("Expecting one field only.")
return ["password"]
transport.auth_interactive(username, handler)
请注意,上面的代码使用了
Transport
,因此它默认绕过主机密钥验证。使用 hostkey
的 Transport.connect
参数来纠正这一点。