Implement SSH receive-pack pushes

crates/storage/src/git.rs

@@ -23,6 +23,9 @@ use gix::{
 use sha1::{Digest as _, Sha1};
 use zeroize::Zeroize as _;
 
+#[cfg(feature = "ssh")]
+use std::io::{Read as _, Write as _};
+
 use crate::{
     config::{ApplicationId, GitRemote, RemoteEndpoint, RemoteTransport, ServerId, SshFingerprint},
     crypto::{KeyHandle, KeyStore, SecretProvider},
@@ -494,6 +497,15 @@ pub enum GitError {
     },
     GitProtocolFailed,
     MalformedGitPack,
+    RemoteUnpackFailed {
+        diagnostic: String,
+    },
+    RemoteRefRejected {
+        diagnostic: String,
+    },
+    PushOutcomeUnknown {
+        diagnostic: String,
+    },
     CredentialsUnavailable,
     CredentialAccessDenied,
     CredentialCancelled,
@@ -620,6 +632,24 @@ impl fmt::Display for GitError {
             }
             Self::GitProtocolFailed => formatter.write_str("the Git wire protocol failed"),
             Self::MalformedGitPack => formatter.write_str("the remote sent a malformed Git pack"),
+            Self::RemoteUnpackFailed { diagnostic } => {
+                write!(
+                    formatter,
+                    "the remote could not unpack the Git objects: {diagnostic}"
+                )
+            }
+            Self::RemoteRefRejected { diagnostic } => {
+                write!(
+                    formatter,
+                    "the remote rejected the Git update: {diagnostic}"
+                )
+            }
+            Self::PushOutcomeUnknown { diagnostic } if diagnostic.is_empty() => formatter
+                .write_str("the remote Git update outcome is unknown; fetch before retrying"),
+            Self::PushOutcomeUnknown { diagnostic } => write!(
+                formatter,
+                "the remote Git update outcome is unknown; fetch before retrying: {diagnostic}"
+            ),
             Self::CredentialsUnavailable => {
                 formatter.write_str("HTTPS Git credentials are unavailable")
             }
@@ -973,6 +1003,200 @@ impl GitSmartHttpTransport for ReqwestGitTransport {
     }
 }
 
+struct PushContext {
+    remote: String,
+    branch: String,
+    reference: String,
+    new: gix::hash::ObjectId,
+}
+
+enum ReceivePackClient<'a> {
+    Https {
+        url: &'a url::Url,
+        credential: GitCredential,
+        transport: &'a dyn GitSmartHttpTransport,
+    },
+    #[cfg(feature = "ssh")]
+    Ssh {
+        session: Option<crate::ssh::SshSession>,
+        command: Option<crate::ssh::SshRawGitCommand>,
+    },
+}
+
+impl ReceivePackClient<'_> {
+    fn advertisement(&mut self, control: &GitOperationControl) -> Result<Vec<u8>, GitError> {
+        match self {
+            Self::Https {
+                url,
+                credential,
+                transport,
+            } => transport.advertise_receive_pack_controlled(url, credential, control),
+            #[cfg(feature = "ssh")]
+            Self::Ssh { command, .. } => {
+                control.checkpoint(GitProgressPhase::Receiving)?;
+                let result = read_receive_pack_advertisement(
+                    &mut command.as_mut().ok_or(GitError::SshProtocolFailed)?.stdout,
+                );
+                if control.is_cancelled() {
+                    return Err(GitError::Cancelled);
+                }
+                let result = result?;
+                control.checkpoint(GitProgressPhase::Receiving)?;
+                Ok(result)
+            }
+        }
+    }
+
+    fn finish_without_update(&mut self, _control: &GitOperationControl) -> Result<(), GitError> {
+        match self {
+            Self::Https { .. } => Ok(()),
+            #[cfg(feature = "ssh")]
+            Self::Ssh { .. } => {
+                self.exchange_ssh(b"0000".to_vec(), _control, false)?;
+                Ok(())
+            }
+        }
+    }
+
+    fn receive(
+        &mut self,
+        request: Vec<u8>,
+        control: &GitOperationControl,
+    ) -> Result<Vec<u8>, GitError> {
+        match self {
+            Self::Https {
+                url,
+                credential,
+                transport,
+            } => transport.receive_pack_controlled(url, credential, request, control),
+            #[cfg(feature = "ssh")]
+            Self::Ssh { .. } => self.exchange_ssh(request, control, true),
+        }
+    }
+
+    #[cfg(feature = "ssh")]
+    fn exchange_ssh(
+        &mut self,
+        request: Vec<u8>,
+        control: &GitOperationControl,
+        update_may_apply: bool,
+    ) -> Result<Vec<u8>, GitError> {
+        control.checkpoint(GitProgressPhase::Sending)?;
+        let Self::Ssh { session, command } = self else {
+            return Err(GitError::SshProtocolFailed);
+        };
+        let mut command = command.take().ok_or(GitError::SshProtocolFailed)?;
+        let session = session.take().ok_or(GitError::SshProtocolFailed)?;
+        let write = command.stdin.write_all(&request);
+        drop(command.stdin);
+        if write.is_err() {
+            let _ = session.close();
+            return Err(if update_may_apply {
+                unknown_push_outcome("SSH channel closed while sending")
+            } else {
+                GitError::SshProtocolFailed
+            });
+        }
+        let response = read_bounded_to_end(&mut command.stdout);
+        let completion = session.finish_command(command.completion, control);
+        let close = session.close();
+        let response = if update_may_apply {
+            let response =
+                response.map_err(|_| unknown_push_outcome("SSH channel closed after sending"))?;
+            completion.map_err(map_ambiguous_push_error)?;
+            close.map_err(map_ambiguous_push_error)?;
+            response
+        } else {
+            let response = response.map_err(|_| GitError::SshProtocolFailed)?;
+            completion?;
+            close?;
+            response
+        };
+        if update_may_apply {
+            control
+                .checkpoint(GitProgressPhase::Sending)
+                .map_err(map_ambiguous_push_error)?;
+        } else {
+            control.checkpoint(GitProgressPhase::Sending)?;
+        }
+        Ok(response)
+    }
+}
+
+#[cfg(feature = "ssh")]
+fn read_receive_pack_advertisement(input: &mut impl std::io::Read) -> Result<Vec<u8>, GitError> {
+    let mut output = Vec::new();
+    for _ in 0..1_000_000 {
+        let mut prefix = [0_u8; 4];
+        input
+            .read_exact(&mut prefix)
+            .map_err(|_| GitError::GitProtocolFailed)?;
+        let length = std::str::from_utf8(&prefix)
+            .ok()
+            .and_then(|value| usize::from_str_radix(value, 16).ok())
+            .ok_or(GitError::GitProtocolFailed)?;
+        output.extend_from_slice(&prefix);
+        if length == 0 {
+            return Ok(output);
+        }
+        if !(4..=65_520).contains(&length) || output.len() + length - 4 > 128 * 1024 * 1024 {
+            return Err(GitError::GitProtocolFailed);
+        }
+        let start = output.len();
+        output.resize(start + length - 4, 0);
+        input
+            .read_exact(&mut output[start..])
+            .map_err(|_| GitError::GitProtocolFailed)?;
+    }
+    Err(GitError::GitProtocolFailed)
+}
+
+#[cfg(feature = "ssh")]
+fn read_bounded_to_end(input: &mut impl std::io::Read) -> Result<Vec<u8>, GitError> {
+    let mut output = Vec::new();
+    input
+        .take(128 * 1024 * 1024 + 1)
+        .read_to_end(&mut output)
+        .map_err(|_| GitError::SshProtocolFailed)?;
+    if output.len() > 128 * 1024 * 1024 {
+        return Err(GitError::GitProtocolFailed);
+    }
+    Ok(output)
+}
+
+fn protocol_diagnostic(input: &[u8]) -> String {
+    String::from_utf8_lossy(&input[..input.len().min(8 * 1024)])
+        .chars()
+        .map(|character| {
+            if character.is_control() {
+                ' '
+            } else {
+                character
+            }
+        })
+        .collect::<String>()
+        .split_whitespace()
+        .collect::<Vec<_>>()
+        .join(" ")
+}
+
+fn unknown_push_outcome(diagnostic: &str) -> GitError {
+    GitError::PushOutcomeUnknown {
+        diagnostic: protocol_diagnostic(diagnostic.as_bytes()),
+    }
+}
+
+#[cfg(feature = "ssh")]
+fn map_ambiguous_push_error(error: GitError) -> GitError {
+    match error {
+        GitError::SshRemoteServiceFailed { diagnostic } if diagnostic.is_empty() => {
+            unknown_push_outcome("SSH service ended without confirmed status")
+        }
+        GitError::SshRemoteServiceFailed { diagnostic } => unknown_push_outcome(&diagnostic),
+        error => unknown_push_outcome(&error.to_string()),
+    }
+}
+
 fn map_reqwest_error(error: reqwest::Error) -> GitError {
     if error.is_builder() {
         return invalid(error);
@@ -1831,9 +2055,57 @@ impl GitRepository {
         &self,
         configured: &GitRemote,
         branch: Option<&str>,
-        credentials: &impl GitCredentialProvider,
+        credentials: &impl GitRemoteCredentialProvider,
     ) -> Result<PushOutcome, GitError> {
-        self.push_with_transport(configured, branch, credentials, &ReqwestGitTransport)
+        self.push_controlled(
+            configured,
+            branch,
+            credentials,
+            &GitOperationControl::default(),
+        )
+    }
+
+    fn push_controlled(
+        &self,
+        configured: &GitRemote,
+        branch: Option<&str>,
+        credentials: &impl GitRemoteCredentialProvider,
+        control: &GitOperationControl,
+    ) -> Result<PushOutcome, GitError> {
+        let context = self.prepare_push(configured, branch, control)?;
+        control.checkpoint(GitProgressPhase::Authenticating)?;
+        match configured.endpoint() {
+            RemoteEndpoint::Https(_) => {
+                let (url, server_id, application_id) = require_https_remote(configured)?;
+                let credential = credentials.credential(server_id, application_id)?;
+                let transport = ReqwestGitTransport;
+                let mut client = ReceivePackClient::Https {
+                    url,
+                    credential,
+                    transport: &transport,
+                };
+                self.push_with_receive_pack(context, &mut client, control)
+            }
+            RemoteEndpoint::Ssh(_) => {
+                #[cfg(feature = "ssh")]
+                {
+                    let session =
+                        crate::ssh::SshSession::connect(configured, credentials, control)?;
+                    let command = session.open_receive_pack(configured, control)?;
+                    let mut client = ReceivePackClient::Ssh {
+                        session: Some(session),
+                        command: Some(command),
+                    };
+                    self.push_with_receive_pack(context, &mut client, control)
+                }
+                #[cfg(not(feature = "ssh"))]
+                {
+                    Err(GitError::UnsupportedRemoteTransport {
+                        transport: RemoteTransport::Ssh,
+                    })
+                }
+            }
+        }
     }
 
     pub fn push_with_transport(
@@ -1860,6 +2132,24 @@ impl GitRepository {
         transport: &impl GitSmartHttpTransport,
         control: &GitOperationControl,
     ) -> Result<PushOutcome, GitError> {
+        let context = self.prepare_push(configured, branch, control)?;
+        let (url, server_id, application_id) = require_https_remote(configured)?;
+        control.checkpoint(GitProgressPhase::Authenticating)?;
+        let credential = credentials.credential(server_id, application_id)?;
+        let mut client = ReceivePackClient::Https {
+            url,
+            credential,
+            transport,
+        };
+        self.push_with_receive_pack(context, &mut client, control)
+    }
+
+    fn prepare_push(
+        &self,
+        configured: &GitRemote,
+        branch: Option<&str>,
+        control: &GitOperationControl,
+    ) -> Result<PushContext, GitError> {
         control.checkpoint(GitProgressPhase::Validating)?;
         if !self.status()?.is_clean() {
             return Err(GitError::DirtyWorktree);
@@ -1869,7 +2159,7 @@ impl GitRepository {
         if !same_remote_endpoint(&actual_url, configured.url())? {
             return Err(GitError::ForbiddenRemoteUrl);
         }
-        let (url, server_id, application_id) = require_https_remote(configured)?;
+        ensure_remote_transport_available(configured.endpoint())?;
         let branch = branch.map_or_else(
             || self.current_branch(),
             |branch| {
@@ -1883,55 +2173,92 @@ impl GitRepository {
             .head_id()
             .map_err(|_| GitError::UnbornHead)?
             .detach();
-        control.checkpoint(GitProgressPhase::Authenticating)?;
-        let credential = credentials.credential(server_id, application_id)?;
-        let advertisement =
-            transport.advertise_receive_pack_controlled(url, &credential, control)?;
-        let advertised = parse_receive_pack_advertisement(&advertisement)?;
-        let old = advertised.refs.get(&reference).copied();
+        Ok(PushContext {
+            remote: name.to_owned(),
+            branch,
+            reference,
+            new,
+        })
+    }
+
+    fn push_with_receive_pack(
+        &self,
+        context: PushContext,
+        client: &mut ReceivePackClient<'_>,
+        control: &GitOperationControl,
+    ) -> Result<PushOutcome, GitError> {
+        let advertisement = client.advertisement(control)?;
+        let advertised = match parse_receive_pack_advertisement(&advertisement) {
+            Ok(advertised) => advertised,
+            Err(error) => {
+                let _ = client.finish_without_update(control);
+                return Err(error);
+            }
+        };
+        let old = advertised.refs.get(&context.reference).copied();
         if let Some(old) = old {
-            if old == new {
+            if old == context.new {
+                client.finish_without_update(control)?;
+                self.update_remote_tracking(&context.remote, &context.branch, context.new)?;
                 return Ok(PushOutcome {
-                    remote: name.to_owned(),
-                    branch,
+                    remote: context.remote,
+                    branch: context.branch,
                     old: Some(old.to_string()),
-                    new: new.to_string(),
+                    new: context.new.to_string(),
                 });
             }
-            let base = self
-                .repository
-                .merge_base(old, new)
-                .map_err(|_| GitError::NonFastForward)?
-                .detach();
+            let base = match self.repository.merge_base(old, context.new) {
+                Ok(base) => base.detach(),
+                Err(_) => {
+                    let _ = client.finish_without_update(control);
+                    return Err(GitError::NonFastForward);
+                }
+            };
             if base != old {
+                let _ = client.finish_without_update(control);
                 return Err(GitError::NonFastForward);
             }
         }
         if !advertised.capabilities.contains("report-status") {
+            let _ = client.finish_without_update(control);
             return Err(GitError::InvalidRepository(
                 "server does not support receive-pack status reports".to_owned(),
             ));
         }
-        let pack = build_pack(&self.repository, new, old)?;
+        let pack = match build_pack(&self.repository, context.new, old) {
+            Ok(pack) => pack,
+            Err(error) => {
+                let _ = client.finish_without_update(control);
+                return Err(error);
+            }
+        };
         let old_hex = old.map_or_else(
             || "0000000000000000000000000000000000000000".to_owned(),
             |id| id.to_string(),
         );
+        let capabilities = if advertised.capabilities.contains("agent") {
+            format!(
+                "report-status agent=ironstorage/{}",
+                env!("CARGO_PKG_VERSION")
+            )
+        } else {
+            "report-status".to_owned()
+        };
         let command = format!(
-            "{old_hex} {new} {reference}\0report-status agent=ironstorage/{}\n",
-            env!("CARGO_PKG_VERSION")
+            "{old_hex} {} {}\0{capabilities}\n",
+            context.new, context.reference,
         );
         let mut request = encode_pkt_line(command.as_bytes())?;
         request.extend_from_slice(b"0000");
         request.extend_from_slice(&pack);
-        let response = transport.receive_pack_controlled(url, &credential, request, control)?;
-        parse_receive_pack_result(&response, &reference)?;
-        self.update_remote_tracking(name, &branch, new)?;
+        let response = client.receive(request, control)?;
+        parse_receive_pack_result(&response, &context.reference)?;
+        self.update_remote_tracking(&context.remote, &context.branch, context.new)?;
         Ok(PushOutcome {
-            remote: name.to_owned(),
-            branch,
+            remote: context.remote,
+            branch: context.branch,
             old: old.map(|id| id.to_string()),
-            new: new.to_string(),
+            new: context.new.to_string(),
         })
     }
 
@@ -1951,13 +2278,15 @@ impl GitRepository {
         credentials: &impl GitRemoteCredentialProvider,
         control: &GitOperationControl,
     ) -> Result<(PullOutcome, PushOutcome), GitError> {
-        self.sync_with_transports_controlled(
+        let pull = self.pull_with_transport_controlled(
             configured,
+            None,
             credentials,
             &EmbeddedFetchTransport,
-            &ReqwestGitTransport,
             control,
-        )
+        )?;
+        let push = self.push_controlled(configured, None, credentials, control)?;
+        Ok((pull, push))
     }
 
     pub fn sync_with_transports_controlled(
@@ -3336,7 +3665,8 @@ fn parse_receive_pack_advertisement(input: &[u8]) -> Result<ReceivePackAdvertise
 }
 
 fn parse_receive_pack_result(input: &[u8], reference: &str) -> Result<(), GitError> {
-    let packets = decode_pkt_lines(input)?;
+    let packets = decode_pkt_lines(input)
+        .map_err(|_| unknown_push_outcome("malformed receive-pack status"))?;
     let mut unpacked = false;
     let mut updated = false;
     for packet in packets.into_iter().flatten() {
@@ -3348,10 +3678,9 @@ fn parse_receive_pack_result(input: &[u8], reference: &str) -> Result<(), GitErr
             if line == b"unpack ok" {
                 unpacked = true;
             } else if let Some(reason) = line.strip_prefix(b"unpack ") {
-                return Err(GitError::InvalidRepository(format!(
-                    "remote could not unpack objects: {}",
-                    String::from_utf8_lossy(reason)
-                )));
+                return Err(GitError::RemoteUnpackFailed {
+                    diagnostic: protocol_diagnostic(reason),
+                });
             } else if line == format!("ok {reference}").as_bytes() {
                 updated = true;
             } else if let Some(reason) = line.strip_prefix(format!("ng {reference} ").as_bytes()) {
@@ -3359,16 +3688,16 @@ fn parse_receive_pack_result(input: &[u8], reference: &str) -> Result<(), GitErr
                 return if reason.to_ascii_lowercase().contains("non-fast-forward") {
                     Err(GitError::NonFastForward)
                 } else {
-                    Err(GitError::InvalidRepository(format!(
-                        "remote rejected update: {reason}"
-                    )))
+                    Err(GitError::RemoteRefRejected {
+                        diagnostic: protocol_diagnostic(reason.as_bytes()),
+                    })
                 };
             }
         }
     }
     if !unpacked || !updated {
-        return Err(GitError::InvalidRepository(
-            "receive-pack response omitted update status".to_owned(),
+        return Err(unknown_push_outcome(
+            "receive-pack response omitted update status",
         ));
     }
     Ok(())
@@ -3892,6 +4221,9 @@ fn io(operation: &'static str, path: &Path) -> GitError {
 #[cfg(all(test, feature = "ssh"))]
 mod ssh_tests;
 
+#[cfg(all(test, feature = "ssh"))]
+mod ssh_push_tests;
+
 #[cfg(test)]
 mod tests {
     use super::{GitIdentity, GitRepository};