如何用Rust的russh库处理git-upload-pack实现SSH Git服务器

问题背景 #

我需要搭建一个支持git clone git@localhost:/home/git/repo命令的SSH服务器，已有Go语言的实现代码，现在要改用russh库用Rust重写。当前Rust代码中，当执行克隆命令时exec_request函数会被触发，data参数内容为git-upload-pack /home/git/repo，需要实现该函数的命令处理逻辑。

参考Go语言实现 #

package main

import (
    "fmt"
    "io"
    "log"
    "os/exec"

    "github.com/gliderlabs/ssh"
)

func main() {
    ssh.Handle(func(s ssh.Session) {
        cmd := s.Command()
        if len(cmd) > 0 && cmd[0] == "git-upload-pack" {
            handleGitUploadPack(s, cmd[1])
        } else {
            io.WriteString(s, "Unknown command\n")
        }
    })

    log.Println("Starting SSH server on :22...")
    log.Fatal(ssh.ListenAndServe(":22", nil))
}

func handleGitUploadPack(s ssh.Session, repoPath string) {
    cmd := exec.Command("git-upload-pack", repoPath)
    cmd.Stdout = s
    cmd.Stderr = s
    cmd.Stdin = s

    if err := cmd.Run(); err != nil {
        io.WriteString(s, fmt.Sprintf("Failed to run git-upload-pack: %v\n", err))
    }
}

当前未完成的Rust代码 #

use std::collections::HashMap;
use std::sync::Arc;

use anyhow::Result;
use async_trait::async_trait;
use russh::server::{Msg, Server as _, Session};
use russh::*;
use russh_keys::*;
use tokio::sync::Mutex;

#[tokio::main]
async fn main() -> Result<()> {
    env_logger::builder()
        .filter_level(log::LevelFilter::Debug)
        .init();

    let config = russh::server::Config {
        inactivity_timeout: Some(std::time::Duration::from_secs(3600)),
        auth_rejection_time: std::time::Duration::from_secs(3),
        auth_rejection_time_initial: Some(std::time::Duration::from_secs(0)),
        keys: vec![russh_keys::key::KeyPair::generate_ed25519().unwrap()],
        ..Default::default()
    };
    let config = Arc::new(config);

    let mut sh = Server {
        clients: Arc::new(Mutex::new(HashMap::new())),
        id: 0,
    };

    sh.run_on_address(config, ("0.0.0.0", 22)).await.unwrap();

    Ok(())
}

#[derive(Clone)]
struct Server {
    clients: Arc<Mutex<HashMap<(usize, ChannelId), russh::server::Handle>>>,
    id: usize,
}

impl server::Server for Server {
    type Handler = Self;
    fn new_client(&mut self, _: Option<std::net::SocketAddr>) -> Self {
        let s = self.clone();
        self.id += 1;
        s
    }
}

#[async_trait]
impl server::Handler for Server {
    type Error = anyhow::Error;

    async fn channel_open_session(
        &mut self,
        channel: Channel<Msg>,
        session: &mut Session,
    ) -> Result<bool, Self::Error> {
        {
            let mut clients = self.clients.lock().await;
            clients.insert((self.id, channel.id()), session.handle());
        }
        Ok(true)
    }

    async fn auth_publickey(
        &mut self,
        _: &str,
        _: &key::PublicKey,
    ) -> Result<server::Auth, Self::Error> {
        Ok(server::Auth::Accept)
    }

    async fn exec_request(
        &mut self,
        channel: ChannelId,
        data: &[u8],
        session: &mut Session,
    ) -> Result<(), Self::Error> {
        // How to handle `git-upload-pack ./repo/path` command here.
        Ok(())
    }
}

解决方案 #

要实现git-upload-pack命令的处理，需要完成以下步骤：

1. 解析data中的命令参数，提取仓库路径
2. 使用tokio的异步进程模块启动git-upload-pack
3. 将SSH通道的输入输出与子进程绑定
4. 处理进程执行结果，向通道返回错误信息（如果执行失败）

修改后的完整Rust代码 #

首先需要在Cargo.toml中添加tokio的process特性：

[dependencies]
anyhow = "1.0"
async-trait = "0.1"
russh = "0.45"
russh-keys = "0.45"
tokio = { version = "1.0", features = ["full"] }
env_logger = "0.10"

然后修改exec_request函数及相关逻辑：

use std::collections::HashMap;
use std::sync::Arc;
use std::io::Write;

use anyhow::Result;
use async_trait::async_trait;
use russh::server::{Msg, Server as _, Session, Channel};
use russh::*;
use russh_keys::*;
use tokio::sync::Mutex;
use tokio::process::Command;

#[tokio::main]
async fn main() -> Result<()> {
    env_logger::builder()
        .filter_level(log::LevelFilter::Debug)
        .init();

    let config = russh::server::Config {
        inactivity_timeout: Some(std::time::Duration::from_secs(3600)),
        auth_rejection_time: std::time::Duration::from_secs(3),
        auth_rejection_time_initial: Some(std::time::Duration::from_secs(0)),
        keys: vec![russh_keys::key::KeyPair::generate_ed25519().unwrap()],
        ..Default::default()
    };
    let config = Arc::new(config);

    let mut sh = Server {
        clients: Arc::new(Mutex::new(HashMap::new())),
        id: 0,
    };

    sh.run_on_address(config, ("0.0.0.0", 22)).await.unwrap();

    Ok(())
}

#[derive(Clone)]
struct Server {
    clients: Arc<Mutex<HashMap<(usize, ChannelId), russh::server::Handle>>>,
    id: usize,
}

impl server::Server for Server {
    type Handler = Self;
    fn new_client(&mut self, _: Option<std::net::SocketAddr>) -> Self {
        let s = self.clone();
        self.id += 1;
        s
    }
}

#[async_trait]
impl server::Handler for Server {
    type Error = anyhow::Error;

    async fn channel_open_session(
        &mut self,
        channel: Channel<Msg>,
        session: &mut Session,
    ) -> Result<bool, Self::Error> {
        {
            let mut clients = self.clients.lock().await;
            clients.insert((self.id, channel.id()), session.handle());
        }
        Ok(true)
    }

    async fn auth_publickey(
        &mut self,
        _: &str,
        _: &key::PublicKey,
    ) -> Result<server::Auth, Self::Error> {
        Ok(server::Auth::Accept)
    }

    async fn exec_request(
        &mut self,
        channel: ChannelId,
        data: &[u8],
        session: &mut Session,
    ) -> Result<(), Self::Error> {
        // 解析命令内容
        let cmd_str = String::from_utf8_lossy(data).trim().to_string();
        let mut parts = cmd_str.split_whitespace();
        
        match parts.next() {
            Some("git-upload-pack") => {
                let repo_path = parts.next().ok_or_else(|| anyhow::anyhow!("Missing repository path"))?;
                
                // 获取通道句柄
                let handle = session.handle();
                
                // 启动git-upload-pack进程
                let mut cmd = Command::new("git-upload-pack");
                cmd.arg(repo_path);
                
                // 绑定标准输入输出
                let mut child = cmd.spawn()?;
                
                // 异步处理输入输出转发
                let stdin = child.stdin.take().ok_or_else(|| anyhow::anyhow!("Failed to get stdin"))?;
                let stdout = child.stdout.take().ok_or_else(|| anyhow::anyhow!("Failed to get stdout"))?;
                let stderr = child.stderr.take().ok_or_else(|| anyhow::anyhow!("Failed to get stderr"))?;
                
                // 转发SSH通道输入到进程stdin
                tokio::spawn(async move {
                    let mut reader = handle.channel_reader(channel);
                    let mut stdin = stdin;
                    let _ = tokio::io::copy(&mut reader, &mut stdin).await;
                    let _ = stdin.flush();
                });
                
                // 转发进程stdout到SSH通道
                tokio::spawn(async move {
                    let mut stdout = stdout;
                    let mut writer = handle.channel_writer(channel);
                    let _ = tokio::io::copy(&mut stdout, &mut writer).await;
                });
                
                // 转发进程stderr到SSH通道
                tokio::spawn(async move {
                    let mut stderr = stderr;
                    let mut writer = handle.channel_writer(channel);
                    let _ = tokio::io::copy(&mut stderr, &mut writer).await;
                });
                
                // 等待进程结束，发送退出状态
                let status = child.wait().await?;
                handle.channel_close(channel).await?;
                handle.channel_eof(channel).await?;
                handle.exit_status_request(channel, status.code().unwrap_or(1)).await?;
            }
            _ => {
                // 处理未知命令
                let err_msg = "Unknown command\n";
                session.handle().channel_data(channel, err_msg.as_bytes()).await?;
                session.handle().channel_close(channel).await?;
                session.handle().channel_eof(channel).await?;
                session.handle().exit_status_request(channel, 1).await?;
            }
        }
        
        Ok(())
    }
}

关键逻辑说明 #

• 命令解析：将data转为字符串后拆分，判断是否为git-upload-pack命令，并提取仓库路径
• 进程启动：使用tokio::process::Command异步启动git-upload-pack，确保与SSH服务器的异步模型兼容
• IO转发：通过tokio的异步IO，将SSH通道的输入输出与子进程的标准输入输出绑定，实现数据双向传输
• 状态处理：等待进程结束后，向SSH通道发送退出状态码，完成会话清理

内容的提问来源于stack exchange，提问作者crabrs