AsyncSSH项目中SFTP与Shell会话冲突的解决方案

问题背景

在使用AsyncSSH构建自定义SSH服务器时，开发者可能会遇到一个常见问题：当同时启用SFTP功能和Shell会话功能时，Shell会话会意外失败。具体表现为客户端连接时出现"shell request failed on channel 0"错误，而服务器日志显示认证成功后立即关闭了连接。

问题分析

这种现象的根本原因在于AsyncSSH的会话处理机制。当开发者通过sftp_factory参数启用SFTP功能时，AsyncSSH会默认接管所有会话请求的处理，包括Shell、Exec和Subsystem等。如果此时没有明确指定其他会话类型的处理方式，系统会拒绝这些请求。

解决方案

方案一：使用session_factory或process_factory

最简单的解决方案是在创建服务器时同时指定sftp_factory和process_factory参数：

await asyncssh.create_server(
    lambda: SSHServer(interact),
    "",
    port,
    server_host_keys=["/etc/ssh/ssh_host_ecdsa_key"],
    sftp_factory=SFTPServer,
    process_factory=handle_client_process
)

其中handle_client_process是一个处理客户端进程的函数，可以获取连接信息：

def handle_client_process(process):
    username = process.get_extra_info('username')
    process.stdout.write(f'Welcome, {username}!\n')
    process.exit(0)

方案二：自定义SSHServer类

对于需要更复杂控制的情况，可以创建自定义的SSHServer子类，并实现相应的回调方法：

class CustomSSHServer(asyncssh.SSHServer):
    def __init__(self, interact):
        self.interact = interact
        self.user = None
    
    def session_requested(self):
        return self.handle_session
    
    def handle_session(self, stdin, stdout, stderr):
        # 自定义会话处理逻辑
        stdout.write(f"Hello {self.user}!\n")

方案三：混合使用工厂方法和回调

也可以结合使用工厂方法和回调方法，根据不同的会话类型采取不同的处理方式：

class HybridSSHServer(asyncssh.SSHServer):
    def shell_requested(self):
        return True  # 允许Shell会话
    
    def exec_requested(self, command):
        return self.handle_exec_command(command)
    
    def subsystem_requested(self, subsystem):
        if subsystem == 'sftp':
            return True  # 允许SFTP
        return False

注意事项

1. 会话状态管理：当需要维护会话特定状态时，可以使用functools.partial来传递额外参数。

2. PromptToolkit集成：如果使用PromptToolkit创建交互式Shell，需要注意它可能不支持多个并发SSH连接，这可能导致不可预见的错误。

3. 性能考虑：对于高并发场景，建议使用轻量级的处理函数而非复杂的对象实例化。

最佳实践建议

1. 对于简单用例，优先使用工厂方法模式，代码更简洁。

2. 对于需要复杂状态管理或细粒度控制的场景，使用自定义SSHServer类。

3. 明确区分不同类型的会话处理逻辑，避免功能冲突。

4. 在生产环境中，建议添加详细的日志记录，以便调试会话处理流程。

通过合理配置AsyncSSH的会话处理机制，开发者可以轻松实现同时支持SFTP和Shell会话的SSH服务器，满足各种应用场景的需求。