Implement SSH upload-pack fetches

crates/storage/src/ssh.rs

@@ -4,7 +4,7 @@ use std::{
     borrow::Cow,
     fs,
     future::Future,
-    io::{Read, Write},
+    io::{self, Read, Write},
     path::{Path, PathBuf},
     sync::{Arc, Mutex},
     time::Duration,
@@ -14,7 +14,7 @@ use cap_std::{ambient_authority, fs::Dir};
 use cap_tempfile::TempFile;
 use hmac::{Hmac, Mac as _};
 use russh::{
-    Disconnect, client,
+    ChannelMsg, Disconnect, client,
     keys::{
         HashAlg, PrivateKey, PublicKey, agent::client::AgentClient, key::PrivateKeyWithHashAlg,
         ssh_key::Algorithm,
@@ -23,7 +23,7 @@ use russh::{
 use sha1::Sha1;
 
 use crate::{
-    config::{GitRemote, RemoteEndpoint, SshFingerprint, SshIdentitySource},
+    config::{GitRemote, RemoteEndpoint, SshFingerprint, SshIdentitySource, SshRepositoryPath},
     git::{GitError, GitOperationControl, GitProgressPhase, SshHostKey, SshPassphraseProvider},
 };
 
@@ -32,9 +32,69 @@ const MAX_KNOWN_HOSTS_BYTES: u64 = 1024 * 1024;
 const MAX_KNOWN_HOST_LINE_BYTES: usize = 16 * 1024;
 const MAX_AGENT_IDENTITIES: usize = 64;
 const CONNECTION_TIMEOUT: Duration = Duration::from_secs(30);
+const COMMAND_TIMEOUT: Duration = Duration::from_secs(30);
 const CANCELLATION_POLL: Duration = Duration::from_millis(25);
+const CHANNEL_QUEUE_DEPTH: usize = 8;
+const MAX_CHANNEL_CHUNK: usize = 32 * 1024;
+const MAX_SSH_DIAGNOSTIC_BYTES: usize = 8 * 1024;
 static KNOWN_HOSTS_WRITE_LOCK: Mutex<()> = Mutex::new(());
 
+pub(crate) type SshGitTransport =
+    gix::protocol::transport::client::git::blocking_io::Connection<SshStdout, SshStdin>;
+
+pub(crate) struct SshGitCommand {
+    pub(crate) transport: SshGitTransport,
+    pub(crate) completion: SshCommandCompletion,
+}
+
+pub(crate) type SshCommandCompletion = tokio::sync::oneshot::Receiver<Result<(), GitError>>;
+
+pub(crate) struct SshStdout {
+    receiver: tokio::sync::mpsc::Receiver<Vec<u8>>,
+    current: Vec<u8>,
+    offset: usize,
+}
+
+impl Read for SshStdout {
+    fn read(&mut self, output: &mut [u8]) -> io::Result<usize> {
+        if output.is_empty() {
+            return Ok(0);
+        }
+        while self.offset == self.current.len() {
+            let Some(next) = self.receiver.blocking_recv() else {
+                return Ok(0);
+            };
+            self.current = next;
+            self.offset = 0;
+        }
+        let count = output.len().min(self.current.len() - self.offset);
+        output[..count].copy_from_slice(&self.current[self.offset..self.offset + count]);
+        self.offset += count;
+        Ok(count)
+    }
+}
+
+pub(crate) struct SshStdin {
+    sender: tokio::sync::mpsc::Sender<Vec<u8>>,
+}
+
+impl Write for SshStdin {
+    fn write(&mut self, input: &[u8]) -> io::Result<usize> {
+        if input.is_empty() {
+            return Ok(0);
+        }
+        let count = input.len().min(MAX_CHANNEL_CHUNK);
+        self.sender
+            .blocking_send(input[..count].to_vec())
+            .map_err(|_| io::Error::new(io::ErrorKind::BrokenPipe, "SSH channel closed"))?;
+        Ok(count)
+    }
+
+    fn flush(&mut self) -> io::Result<()> {
+        Ok(())
+    }
+}
+
 /// An authenticated SSH connection. Git protocol channels are opened by the
 /// storage transport so callers cannot issue arbitrary commands.
 pub struct SshSession {
@@ -130,6 +190,72 @@ impl SshSession {
         Ok(Self { runtime, handle })
     }
 
+    pub(crate) fn open_upload_pack(
+        &self,
+        remote: &GitRemote,
+        control: &GitOperationControl,
+    ) -> Result<SshGitCommand, GitError> {
+        let RemoteEndpoint::Ssh(endpoint) = remote.endpoint() else {
+            return Err(GitError::ForbiddenRemoteUrl);
+        };
+        let command = upload_pack_command(endpoint.path())?;
+        let channel = self.runtime.block_on(async {
+            controlled(
+                tokio::time::timeout(
+                    COMMAND_TIMEOUT,
+                    open_command_channel(&self.handle, command, control),
+                ),
+                control,
+            )
+            .await
+        })?;
+        let channel = match channel {
+            Ok(Ok(channel)) => channel,
+            Ok(Err(error)) => return Err(error),
+            Err(_) => return Err(GitError::SshProtocolFailed),
+        };
+        let (input_tx, input_rx) = tokio::sync::mpsc::channel(CHANNEL_QUEUE_DEPTH);
+        let (output_tx, output_rx) = tokio::sync::mpsc::channel(CHANNEL_QUEUE_DEPTH);
+        let (completion_tx, completion_rx) = tokio::sync::oneshot::channel();
+        let operation = control.clone();
+        self.runtime.spawn(async move {
+            let result = pump_command(channel, input_rx, output_tx, &operation).await;
+            let _ = completion_tx.send(result);
+        });
+        let transport = gix::protocol::transport::client::git::blocking_io::Connection::new(
+            SshStdout {
+                receiver: output_rx,
+                current: Vec::new(),
+                offset: 0,
+            },
+            SshStdin { sender: input_tx },
+            gix::protocol::transport::Protocol::V1,
+            endpoint.path().as_str().as_bytes().to_vec(),
+            None::<(String, Option<u16>)>,
+            gix::protocol::transport::client::git::ConnectMode::Process,
+            false,
+        )
+        .custom_url(Some(remote.url().into()));
+        Ok(SshGitCommand {
+            transport,
+            completion: completion_rx,
+        })
+    }
+
+    pub(crate) fn finish_command(
+        &self,
+        completion: SshCommandCompletion,
+        control: &GitOperationControl,
+    ) -> Result<(), GitError> {
+        let result = self.runtime.block_on(async {
+            controlled(tokio::time::timeout(COMMAND_TIMEOUT, completion), control).await
+        })?;
+        match result {
+            Ok(Ok(result)) => result,
+            Ok(Err(_)) | Err(_) => Err(GitError::SshProtocolFailed),
+        }
+    }
+
     pub fn close(self) -> Result<(), GitError> {
         self.runtime
             .block_on(self.handle.disconnect(Disconnect::ByApplication, "", ""))
@@ -137,6 +263,147 @@ impl SshSession {
     }
 }
 
+async fn open_command_channel(
+    handle: &client::Handle<HostVerifier>,
+    command: Vec<u8>,
+    control: &GitOperationControl,
+) -> Result<russh::Channel<client::Msg>, GitError> {
+    let mut channel = handle
+        .channel_open_session()
+        .await
+        .map_err(|_| GitError::SshProtocolFailed)?;
+    channel
+        .exec(true, command)
+        .await
+        .map_err(|_| GitError::SshProtocolFailed)?;
+    let mut diagnostic = Vec::new();
+    loop {
+        if control.is_cancelled() {
+            let _ = channel.close().await;
+            return Err(GitError::Cancelled);
+        }
+        match channel.wait().await {
+            Some(ChannelMsg::Success) => return Ok(channel),
+            Some(ChannelMsg::Failure) | Some(ChannelMsg::Close) | None => {
+                return Err(remote_service_error(&diagnostic));
+            }
+            Some(ChannelMsg::ExtendedData { data, .. }) => {
+                extend_diagnostic(&mut diagnostic, &data);
+            }
+            Some(ChannelMsg::ExitStatus { .. }) | Some(ChannelMsg::ExitSignal { .. }) => {
+                return Err(remote_service_error(&diagnostic));
+            }
+            Some(ChannelMsg::Data { .. }) => return Err(GitError::GitProtocolFailed),
+            Some(_) => {}
+        }
+    }
+}
+
+async fn pump_command(
+    mut channel: russh::Channel<client::Msg>,
+    mut input: tokio::sync::mpsc::Receiver<Vec<u8>>,
+    output: tokio::sync::mpsc::Sender<Vec<u8>>,
+    control: &GitOperationControl,
+) -> Result<(), GitError> {
+    let mut diagnostic = Vec::new();
+    let mut exit_status = None;
+    let mut input_open = true;
+    let mut output_open = true;
+    loop {
+        tokio::select! {
+            message = channel.wait() => match message {
+                Some(ChannelMsg::Data { data }) => {
+                    if output_open && output.send(data.to_vec()).await.is_err() {
+                        output_open = false;
+                    }
+                }
+                Some(ChannelMsg::ExtendedData { data, .. }) => {
+                    extend_diagnostic(&mut diagnostic, &data);
+                }
+                Some(ChannelMsg::ExitStatus { exit_status: status }) => exit_status = Some(status),
+                Some(ChannelMsg::ExitSignal { error_message, .. }) => {
+                    extend_diagnostic(&mut diagnostic, error_message.as_bytes());
+                    exit_status = Some(u32::MAX);
+                }
+                Some(ChannelMsg::Eof) => {}
+                Some(ChannelMsg::Close) | None => break,
+                Some(ChannelMsg::Failure) => {
+                    exit_status = Some(u32::MAX);
+                }
+                Some(_) => {}
+            },
+            outbound = input.recv(), if input_open => match outbound {
+                Some(data) => channel
+                    .data_bytes(data)
+                    .await
+                    .map_err(|_| GitError::SshProtocolFailed)?,
+                None => {
+                    input_open = false;
+                    channel.eof().await.map_err(|_| GitError::SshProtocolFailed)?;
+                }
+            },
+            () = tokio::time::sleep(CANCELLATION_POLL) => {
+                if control.is_cancelled() {
+                    let _ = channel.close().await;
+                    return Err(GitError::Cancelled);
+                }
+            }
+        }
+    }
+    if control.is_cancelled() {
+        return Err(GitError::Cancelled);
+    }
+    match exit_status {
+        Some(0) => Ok(()),
+        _ => Err(remote_service_error(&diagnostic)),
+    }
+}
+
+fn upload_pack_command(path: &SshRepositoryPath) -> Result<Vec<u8>, GitError> {
+    let path = shell_quote(path.as_str());
+    let command = format!("git-upload-pack {path}");
+    if command.len() > 64 * 1024 {
+        return Err(GitError::ForbiddenRemoteUrl);
+    }
+    Ok(command.into_bytes())
+}
+
+fn shell_quote(value: &str) -> String {
+    let mut quoted = String::with_capacity(value.len() + 2);
+    quoted.push('\'');
+    for character in value.chars() {
+        if character == '\'' {
+            quoted.push_str("'\\''");
+        } else {
+            quoted.push(character);
+        }
+    }
+    quoted.push('\'');
+    quoted
+}
+
+fn extend_diagnostic(output: &mut Vec<u8>, input: &[u8]) {
+    let remaining = MAX_SSH_DIAGNOSTIC_BYTES.saturating_sub(output.len());
+    output.extend_from_slice(&input[..input.len().min(remaining)]);
+}
+
+fn remote_service_error(diagnostic: &[u8]) -> GitError {
+    let diagnostic = String::from_utf8_lossy(diagnostic)
+        .chars()
+        .map(|character| {
+            if character.is_control() {
+                ' '
+            } else {
+                character
+            }
+        })
+        .collect::<String>()
+        .split_whitespace()
+        .collect::<Vec<_>>()
+        .join(" ");
+    GitError::SshRemoteServiceFailed { diagnostic }
+}
+
 /// Persist a host key only after the caller has confirmed an unknown-key error.
 /// Existing changed keys are never replaced.
 pub fn persist_confirmed_host(
@@ -212,8 +479,21 @@ impl ClientError {
 }
 
 impl From<russh::Error> for ClientError {
-    fn from(_: russh::Error) -> Self {
-        Self::Protocol
+    fn from(error: russh::Error) -> Self {
+        match error {
+            russh::Error::IO(_)
+            | russh::Error::Disconnect
+            | russh::Error::HUP
+            | russh::Error::ConnectionTimeout
+            | russh::Error::KeepaliveTimeout
+            | russh::Error::InactivityTimeout => Self::Git(GitError::NetworkUnavailable),
+            russh::Error::NoCommonAlgo { kind, .. } => {
+                Self::Git(GitError::SshUnsupportedAlgorithm {
+                    algorithm: format!("SSH {kind:?}"),
+                })
+            }
+            _ => Self::Protocol,
+        }
     }
 }
 
@@ -664,12 +944,12 @@ mod tests {
     };
 
     use crate::{
-        config::{GitRemote, SshFingerprint, SshRemoteAuthentication},
+        config::{GitRemote, RemoteEndpoint, SshFingerprint, SshRemoteAuthentication},
         git::{GitError, GitOperationControl, SshPassphraseProvider},
         repository::SecretBytes,
     };
 
-    use super::{SshSession, persist_confirmed_host, ssh_host_key};
+    use super::{SshSession, persist_confirmed_host, ssh_host_key, upload_pack_command};
 
     struct Passphrase(Option<&'static [u8]>);
 
@@ -1001,6 +1281,58 @@ mod tests {
         assert_eq!(fs::read(&known_hosts).expect("known hosts"), b"unchanged\n");
     }
 
+    #[test]
+    fn unavailable_network_is_distinct_from_ssh_protocol_failure() {
+        let temporary = tempfile::tempdir().expect("temporary directory");
+        let identity = key(Algorithm::Ed25519);
+        let identity_path = temporary.path().join("identity");
+        let known_hosts = temporary.path().join("known_hosts");
+        write_key(&identity_path, &identity);
+        fs::write(&known_hosts, b"").expect("known hosts");
+        let listener = std::net::TcpListener::bind(("127.0.0.1", 0)).expect("unused port");
+        let port = listener.local_addr().expect("address").port();
+        drop(listener);
+        assert_eq!(
+            SshSession::connect(
+                &remote(port, &identity_path, &known_hosts),
+                &Passphrase(None),
+                &GitOperationControl::default(),
+            )
+            .expect_err("unavailable network"),
+            GitError::NetworkUnavailable
+        );
+    }
+
+    #[test]
+    fn upload_pack_command_preserves_path_semantics_without_injection() {
+        for (url, expected) in [
+            (
+                "ssh://git@example.test/team/repo%20with%20%27quote%27%3B%24%28touch%20x%29.git",
+                "/team/repo with 'quote';$(touch x).git",
+            ),
+            (
+                "git@example.test:team/repo with 'quote';$(touch x).git",
+                "team/repo with 'quote';$(touch x).git",
+            ),
+            ("ssh://git@example.test/~alice/repo.git", "~alice/repo.git"),
+            (
+                "git@example.test:GIT_PROTOCOL=version=2/repo.git",
+                "GIT_PROTOCOL=version=2/repo.git",
+            ),
+        ] {
+            let endpoint = RemoteEndpoint::parse(url).expect("valid SSH endpoint");
+            let command =
+                upload_pack_command(endpoint.as_ssh().expect("SSH").path()).expect("safe command");
+            let command = std::str::from_utf8(&command).expect("UTF-8 command");
+            assert_eq!(
+                shlex::split(command).expect("shell command"),
+                ["git-upload-pack", expected]
+            );
+        }
+        assert!(RemoteEndpoint::parse("git@example.test:-upload-pack").is_err());
+        assert!(RemoteEndpoint::parse("git@example.test:repo\nsecond").is_err());
+    }
+
     #[cfg(unix)]
     #[test]
     fn agent_authentication_selects_one_configured_fingerprint() {