From 76e707f6457f3a0d8b46f3a49beb9b5d16162046 Mon Sep 17 00:00:00 2001 From: Chili Palmer Date: Tue, 25 Aug 2026 20:56:06 +0200 Subject: [PATCH] Implement SSH receive-pack pushes --- crates/storage/src/git.rs | 412 ++++++++++- crates/storage/src/git/ssh_push_tests.rs | 904 +++++++++++++++++++++++ crates/storage/src/ssh.rs | 106 ++- docs/configuration.md | 3 +- docs/git-synchronization.md | 12 + 5 files changed, 1370 insertions(+), 67 deletions(-) create mode 100644 crates/storage/src/git/ssh_push_tests.rs diff --git a/crates/storage/src/git.rs b/crates/storage/src/git.rs index 3ca8cba..94a30e2 100644 --- a/crates/storage/src/git.rs +++ b/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, + command: Option, + }, +} + +impl ReceivePackClient<'_> { + fn advertisement(&mut self, control: &GitOperationControl) -> Result, 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, + control: &GitOperationControl, + ) -> Result, 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, + control: &GitOperationControl, + update_may_apply: bool, + ) -> Result, 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, 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, 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::() + .split_whitespace() + .collect::>() + .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 { - 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 { + 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 { + 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 { 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 { + 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 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}; diff --git a/crates/storage/src/git/ssh_push_tests.rs b/crates/storage/src/git/ssh_push_tests.rs new file mode 100644 index 0000000..c0cfe1a --- /dev/null +++ b/crates/storage/src/git/ssh_push_tests.rs @@ -0,0 +1,904 @@ +use std::{ + fs, + path::Path, + sync::{Arc, Mutex, mpsc}, + thread, + time::Duration, +}; + +use russh::{ + Channel, ChannelId, Sig, + keys::{PrivateKey, PublicKey, ssh_key::Algorithm}, + server, +}; +use sha1::{Digest as _, Sha1}; + +use super::{ + GitCredential, GitCredentialProvider, GitError, GitIdentity, GitOperationControl, + GitRepository, PullOutcome, SshPassphraseProvider, build_pack, +}; +use crate::{ + config::{GitRemote, SshFingerprint, SshRemoteAuthentication}, + repository::{Repository, SecretBytes}, +}; + +#[derive(Clone, Copy)] +enum ReceiveBehavior { + Success, + NonFastForward, + Reject, + UnpackFailure, + MalformedStatus, + EarlyEof, + NonZero, + Signal, + Disconnect, + SlowSuccess, +} + +#[derive(Clone)] +struct UploadFixture { + advertisement: Vec, + pack: Vec, +} + +#[derive(Default)] +struct Observed { + commands: Vec>, + receive_requests: Vec>, +} + +#[derive(Clone, Copy)] +enum Service { + Upload, + Receive, +} + +struct GitServerHandler { + user_key: PublicKey, + upload: Option, + advertised_receive_tip: Option, + receive_behavior: ReceiveBehavior, + malformed_upload: bool, + observed: Arc>, + service: Option, + input: Vec, + saw_want: bool, + saw_have: bool, +} + +impl Clone for GitServerHandler { + fn clone(&self) -> Self { + Self { + user_key: self.user_key.clone(), + upload: self.upload.clone(), + advertised_receive_tip: self.advertised_receive_tip.clone(), + receive_behavior: self.receive_behavior, + malformed_upload: self.malformed_upload, + observed: Arc::clone(&self.observed), + service: None, + input: Vec::new(), + saw_want: false, + saw_have: false, + } + } +} + +impl server::Handler for GitServerHandler { + type Error = russh::Error; + + async fn auth_publickey( + &mut self, + user: &str, + public_key: &PublicKey, + ) -> Result { + Ok(if user == "git" && public_key == &self.user_key { + server::Auth::Accept + } else { + server::Auth::reject() + }) + } + + async fn channel_open_session( + &mut self, + _channel: Channel, + reply: server::ChannelOpenHandle, + _session: &mut server::Session, + ) -> Result<(), Self::Error> { + reply.accept().await; + Ok(()) + } + + async fn exec_request( + &mut self, + channel: ChannelId, + command: &[u8], + session: &mut server::Session, + ) -> Result<(), Self::Error> { + self.observed + .lock() + .expect("observed") + .commands + .push(command.to_vec()); + if command.starts_with(b"git-upload-pack '") && command.ends_with(b"'") { + self.service = Some(Service::Upload); + session.channel_success(channel)?; + if self.malformed_upload { + session.data(channel, b"zzzz".to_vec())?; + session.close(channel)?; + } else if let Some(upload) = &self.upload { + session.data(channel, upload.advertisement.clone())?; + } else { + session.channel_failure(channel)?; + } + } else if command.starts_with(b"git-receive-pack '") && command.ends_with(b"'") { + self.service = Some(Service::Receive); + session.channel_success(channel)?; + session.data( + channel, + receive_advertisement(self.advertised_receive_tip.as_deref()), + )?; + } else { + session.channel_failure(channel)?; + } + Ok(()) + } + + async fn data( + &mut self, + channel: ChannelId, + data: &[u8], + session: &mut server::Session, + ) -> Result<(), Self::Error> { + match self.service { + Some(Service::Receive) => self.input.extend_from_slice(data), + Some(Service::Upload) => self.upload_data(channel, data, session)?, + None => session.close(channel)?, + } + Ok(()) + } + + async fn channel_eof( + &mut self, + channel: ChannelId, + session: &mut server::Session, + ) -> Result<(), Self::Error> { + if matches!(self.service, Some(Service::Receive)) { + self.observed + .lock() + .expect("observed") + .receive_requests + .push(self.input.clone()); + self.finish_receive(channel, session).await?; + } + Ok(()) + } +} + +impl GitServerHandler { + fn upload_data( + &mut self, + channel: ChannelId, + data: &[u8], + session: &mut server::Session, + ) -> Result<(), russh::Error> { + self.input.extend_from_slice(data); + while self.input.len() >= 4 { + let Some(length) = std::str::from_utf8(&self.input[..4]) + .ok() + .and_then(|value| usize::from_str_radix(value, 16).ok()) + else { + session.close(channel)?; + return Ok(()); + }; + if length == 0 { + self.input.drain(..4); + if !self.saw_want { + session.exit_status_request(channel, 0)?; + session.eof(channel)?; + session.close(channel)?; + return Ok(()); + } else if self.saw_have { + session.data(channel, packet(b"NAK\n"))?; + self.saw_have = false; + } + continue; + } + if length < 4 || self.input.len() < length { + break; + } + let line = self.input[4..length].to_vec(); + self.input.drain(..length); + if line.starts_with(b"want ") { + self.saw_want = true; + } else if line.starts_with(b"have ") { + self.saw_have = true; + } else if line == b"done\n" || line == b"done" { + session.data(channel, packet(b"NAK\n"))?; + if let Some(upload) = &self.upload { + for chunk in upload.pack.chunks(997) { + let mut sideband = Vec::with_capacity(chunk.len() + 1); + sideband.push(1); + sideband.extend_from_slice(chunk); + session.data(channel, packet(&sideband))?; + } + } + session.data(channel, b"0000".to_vec())?; + session.exit_status_request(channel, 0)?; + session.eof(channel)?; + session.close(channel)?; + return Ok(()); + } + } + Ok(()) + } + + async fn finish_receive( + &self, + channel: ChannelId, + session: &mut server::Session, + ) -> Result<(), russh::Error> { + let reference = receive_reference(&self.input).unwrap_or("refs/heads/main"); + match self.receive_behavior { + ReceiveBehavior::Success => { + if self.input == b"0000" { + session.exit_status_request(channel, 0)?; + } else { + send_status( + session, + channel, + b"unpack ok\n", + format!("ok {reference}\n"), + )?; + } + } + ReceiveBehavior::NonFastForward => send_status( + session, + channel, + b"unpack ok\n", + format!("ng {reference} non-fast-forward\n"), + )?, + ReceiveBehavior::Reject => send_status( + session, + channel, + b"unpack ok\n", + format!("ng {reference} protected branch\n"), + )?, + ReceiveBehavior::UnpackFailure => { + send_status(session, channel, b"unpack corrupt pack\n", String::new())? + } + ReceiveBehavior::MalformedStatus => { + session.data(channel, b"zzzz".to_vec())?; + session.exit_status_request(channel, 0)?; + } + ReceiveBehavior::EarlyEof => { + session.eof(channel)?; + session.close(channel)?; + return Ok(()); + } + ReceiveBehavior::Disconnect => { + session.close(channel)?; + return Ok(()); + } + ReceiveBehavior::NonZero => { + session.extended_data(channel, 1, b"receive-pack failed\n".to_vec())?; + session.exit_status_request(channel, 7)?; + } + ReceiveBehavior::Signal => { + session.exit_signal_request(channel, Sig::TERM, false, "killed", "")?; + } + ReceiveBehavior::SlowSuccess => { + tokio::time::sleep(Duration::from_millis(500)).await; + send_status( + session, + channel, + b"unpack ok\n", + format!("ok {reference}\n"), + )?; + } + } + session.eof(channel)?; + session.close(channel) + } +} + +fn send_status( + session: &mut server::Session, + channel: ChannelId, + unpack: &[u8], + reference: String, +) -> Result<(), russh::Error> { + session.data(channel, packet(unpack))?; + if !reference.is_empty() { + session.data(channel, packet(reference.as_bytes()))?; + } + session.data(channel, b"0000".to_vec())?; + session.exit_status_request(channel, 0) +} + +struct TestServer { + port: u16, + host_key: PrivateKey, + observed: Arc>, + join: thread::JoinHandle<()>, +} + +fn start_server( + user_key: PublicKey, + upload: Option, + receive_tip: Option, + receive_behavior: ReceiveBehavior, + malformed_upload: bool, + connections: usize, +) -> TestServer { + let host_key = key(); + let observed = Arc::new(Mutex::new(Observed::default())); + let handler = GitServerHandler { + user_key, + upload, + advertised_receive_tip: receive_tip, + receive_behavior, + malformed_upload, + observed: Arc::clone(&observed), + service: None, + input: Vec::new(), + saw_want: false, + saw_have: false, + }; + let server_key = host_key.clone(); + let (port_tx, port_rx) = mpsc::channel(); + let join = thread::spawn(move || { + tokio::runtime::Runtime::new() + .expect("server runtime") + .block_on(async move { + let listener = tokio::net::TcpListener::bind(("127.0.0.1", 0)) + .await + .expect("bind server"); + port_tx + .send(listener.local_addr().expect("address").port()) + .expect("send port"); + let config = Arc::new(server::Config { + keys: vec![server_key], + auth_rejection_time: Duration::ZERO, + auth_rejection_time_initial: Some(Duration::ZERO), + ..server::Config::default() + }); + for _ in 0..connections { + let (stream, _) = listener.accept().await.expect("accept client"); + if let Ok(session) = + server::run_stream(Arc::clone(&config), stream, handler.clone()).await + { + let _ = session.await; + } + } + }); + }); + TestServer { + port: port_rx + .recv_timeout(Duration::from_secs(5)) + .expect("server port"), + host_key, + observed, + join, + } +} + +struct Credentials; + +impl GitCredentialProvider for Credentials { + fn credential( + &self, + _server: &crate::config::ServerId, + _application: &crate::config::ApplicationId, + ) -> Result { + Err(GitError::CredentialsUnavailable) + } +} + +impl SshPassphraseProvider for Credentials { + fn ssh_key_passphrase(&self, fingerprint: &SshFingerprint) -> Result { + Err(GitError::SshKeyPassphraseUnavailable { + fingerprint: fingerprint.clone(), + }) + } +} + +fn key() -> PrivateKey { + PrivateKey::random(&mut russh::keys::key::safe_rng(), Algorithm::Ed25519).expect("generate key") +} + +fn packet(data: &[u8]) -> Vec { + let mut output = format!("{:04x}", data.len() + 4).into_bytes(); + output.extend_from_slice(data); + output +} + +fn identity() -> GitIdentity { + GitIdentity::new("SSH Push Test", "ssh-push@ironstorage.invalid").expect("identity") +} + +fn commit(git: &GitRepository, path: &str, contents: &[u8], message: &str) -> String { + fs::write(git.root().join(path), contents).expect("write worktree"); + git.stage(&[path.into()]).expect("stage"); + git.commit(message).expect("commit") +} + +fn upload_fixture(repository: &GitRepository) -> UploadFixture { + let head = repository.repository.head_commit().expect("head").id; + let capabilities = "multi_ack_detailed side-band-64k thin-pack ofs-delta include-tag symref=HEAD:refs/heads/main"; + let mut advertisement = packet(format!("{head} HEAD\0{capabilities}\n").as_bytes()); + advertisement.extend_from_slice(&packet(format!("{head} refs/heads/main\n").as_bytes())); + advertisement.extend_from_slice(b"0000"); + UploadFixture { + advertisement, + pack: build_pack(&repository.repository, head, None).expect("pack"), + } +} + +fn receive_advertisement(tip: Option<&str>) -> Vec { + let mut output = match tip { + Some(tip) => packet( + format!("{tip} refs/heads/main\0report-status delete-refs ofs-delta\n").as_bytes(), + ), + None => packet(b"0000000000000000000000000000000000000000 capabilities^{}\0report-status delete-refs ofs-delta\n"), + }; + output.extend_from_slice(b"0000"); + output +} + +fn remote(root: &Path, server: &TestServer, identity: &PrivateKey, scp_like: bool) -> GitRemote { + let identity_file = root.join("identity"); + fs::write( + &identity_file, + identity + .to_openssh(russh::keys::ssh_key::LineEnding::LF) + .expect("identity"), + ) + .expect("write identity"); + let known_hosts = root.join("known_hosts"); + fs::write( + &known_hosts, + format!( + "[127.0.0.1]:{} {}\n", + server.port, + server.host_key.public_key().to_openssh().expect("host key") + ), + ) + .expect("known hosts"); + let url = if scp_like { + "git@127.0.0.1:team/store.git".to_owned() + } else { + format!("ssh://git@127.0.0.1:{}/team/store.git", server.port) + }; + let mut remote = GitRemote::ssh_with_authentication( + "origin", + url, + SshRemoteAuthentication::key_file(identity_file, known_hosts).expect("authentication"), + ) + .expect("remote"); + remote.set_ssh_test_port(server.port); + remote +} + +fn local_repository(root: &Path) -> GitRepository { + fs::create_dir(root).expect("local root"); + let store = Repository::open(root).expect("store"); + GitRepository::init(&store, identity()).expect("Git") +} + +fn receive_reference(input: &[u8]) -> Option<&str> { + let length = std::str::from_utf8(input.get(..4)?) + .ok() + .and_then(|value| usize::from_str_radix(value, 16).ok())?; + let command = std::str::from_utf8(input.get(4..length)?).ok()?; + command.split_once('\0')?.0.split_whitespace().nth(2) +} + +fn tracking_id(repository: &GitRepository) -> Option { + repository + .repository + .try_find_reference("refs/remotes/origin/main") + .expect("tracking lookup") + .map(|reference| reference.id().to_string()) +} + +fn assert_pack_request(request: &[u8], old: Option<&str>, new: &str) { + let pack_offset = request + .windows(4) + .position(|window| window == b"PACK") + .expect("pack payload"); + let command = &request[..pack_offset]; + assert!( + command + .windows(new.len()) + .any(|window| window == new.as_bytes()) + ); + assert!( + command + .windows(b"refs/heads/main".len()) + .any(|window| window == b"refs/heads/main") + ); + if let Some(old) = old { + assert!( + command + .windows(old.len()) + .any(|window| window == old.as_bytes()) + ); + } + let pack = &request[pack_offset..]; + assert_eq!(&pack[..4], b"PACK"); + assert_eq!( + u32::from_be_bytes(pack[4..8].try_into().expect("version")), + 2 + ); + assert!(u32::from_be_bytes(pack[8..12].try_into().expect("count")) >= 3); + assert_eq!( + Sha1::digest(&pack[..pack.len() - 20]).as_slice(), + &pack[pack.len() - 20..] + ); + assert!( + !request + .windows(b"PRIVATE KEY".len()) + .any(|window| window == b"PRIVATE KEY") + ); +} + +#[test] +fn new_branch_fast_forward_and_already_current_push_are_confirmed() { + for (old_kind, behavior) in [ + (None, ReceiveBehavior::Success), + (Some("base"), ReceiveBehavior::Success), + ] { + let temporary = tempfile::tempdir().expect("temporary directory"); + let local = local_repository(&temporary.path().join("local")); + let base = commit(&local, ".gpg-id", b"ALICE\n", "Initialize"); + let new = commit(&local, "entry.gpg", b"ciphertext", "Add entry"); + let old = old_kind.map(|_| base.clone()); + let user_key = key(); + let server = start_server( + user_key.public_key().clone(), + None, + old.clone(), + behavior, + false, + 1, + ); + let remote = remote(temporary.path(), &server, &user_key, false); + let mut local = local; + local.add_remote("origin", remote.url()).expect("remote"); + let outcome = local + .push(&remote, Some("main"), &Credentials) + .expect("push"); + assert_eq!(outcome.new_id(), new); + assert_eq!(tracking_id(&local), Some(new.clone())); + server.join.join().expect("server"); + let observed = server.observed.lock().expect("observed"); + assert_eq!( + observed.commands, + [b"git-receive-pack '/team/store.git'".to_vec()] + ); + assert_eq!(observed.receive_requests.len(), 1); + assert_pack_request(&observed.receive_requests[0], old.as_deref(), &new); + } + + let temporary = tempfile::tempdir().expect("temporary directory"); + let local = local_repository(&temporary.path().join("local")); + let head = commit(&local, ".gpg-id", b"ALICE\n", "Initialize"); + let user_key = key(); + let server = start_server( + user_key.public_key().clone(), + None, + Some(head.clone()), + ReceiveBehavior::Success, + false, + 1, + ); + let remote = remote(temporary.path(), &server, &user_key, false); + let mut local = local; + local.add_remote("origin", remote.url()).expect("remote"); + local + .push(&remote, Some("main"), &Credentials) + .expect("current push"); + assert_eq!(tracking_id(&local), Some(head)); + server.join.join().expect("server"); + assert_eq!( + server.observed.lock().expect("observed").receive_requests, + [b"0000".to_vec()] + ); +} + +#[test] +fn receive_pack_failures_never_advance_tracking_or_change_local_state() { + for (behavior, expected) in [ + (ReceiveBehavior::NonFastForward, GitError::NonFastForward), + ( + ReceiveBehavior::Reject, + GitError::RemoteRefRejected { + diagnostic: "protected branch".to_owned(), + }, + ), + ( + ReceiveBehavior::UnpackFailure, + GitError::RemoteUnpackFailed { + diagnostic: "corrupt pack".to_owned(), + }, + ), + ( + ReceiveBehavior::MalformedStatus, + GitError::PushOutcomeUnknown { + diagnostic: "malformed receive-pack status".to_owned(), + }, + ), + ( + ReceiveBehavior::EarlyEof, + GitError::PushOutcomeUnknown { + diagnostic: "SSH service ended without confirmed status".to_owned(), + }, + ), + ( + ReceiveBehavior::NonZero, + GitError::PushOutcomeUnknown { + diagnostic: "receive-pack failed".to_owned(), + }, + ), + ( + ReceiveBehavior::Signal, + GitError::PushOutcomeUnknown { + diagnostic: "killed".to_owned(), + }, + ), + ( + ReceiveBehavior::Disconnect, + GitError::PushOutcomeUnknown { + diagnostic: "SSH service ended without confirmed status".to_owned(), + }, + ), + ] { + let temporary = tempfile::tempdir().expect("temporary directory"); + let local = local_repository(&temporary.path().join("local")); + commit(&local, ".gpg-id", b"ALICE\n", "Initialize"); + let head = commit(&local, "entry.gpg", b"ciphertext", "Add entry"); + let before_status = local.status().expect("status"); + let before_entry = fs::read(local.root().join("entry.gpg")).expect("entry"); + let user_key = key(); + let server = start_server( + user_key.public_key().clone(), + None, + None, + behavior, + false, + 1, + ); + let remote = remote(temporary.path(), &server, &user_key, false); + let mut local = local; + local.add_remote("origin", remote.url()).expect("remote"); + assert_eq!( + local + .push(&remote, Some("main"), &Credentials) + .expect_err("push failure"), + expected + ); + assert_eq!(tracking_id(&local), None); + assert_eq!(local.status().expect("status"), before_status); + assert_eq!( + fs::read(local.root().join("entry.gpg")).expect("entry"), + before_entry + ); + assert_eq!(local.log(Some(1)).expect("log")[0].id(), head); + server.join.join().expect("server"); + } +} + +#[test] +fn non_fast_forward_and_cancellation_fail_without_blind_replay() { + let temporary = tempfile::tempdir().expect("temporary directory"); + let local = local_repository(&temporary.path().join("local")); + commit(&local, ".gpg-id", b"ALICE\n", "Initialize"); + let unrelated = "1111111111111111111111111111111111111111".to_owned(); + let user_key = key(); + let server = start_server( + user_key.public_key().clone(), + None, + Some(unrelated), + ReceiveBehavior::Success, + false, + 1, + ); + let non_ff_remote = remote(temporary.path(), &server, &user_key, false); + let mut local = local; + local + .add_remote("origin", non_ff_remote.url()) + .expect("remote"); + assert_eq!( + local.push(&non_ff_remote, Some("main"), &Credentials), + Err(GitError::NonFastForward) + ); + server.join.join().expect("server"); + assert_eq!( + server.observed.lock().expect("observed").receive_requests, + [b"0000".to_vec()] + ); + + fs::write(local.root().join("staged.gpg"), b"staged").expect("staged entry"); + local.stage(&["staged.gpg".into()]).expect("stage entry"); + let staged = local.status().expect("staged status"); + assert_eq!( + local.push(&non_ff_remote, Some("main"), &Credentials), + Err(GitError::DirtyWorktree) + ); + assert_eq!(local.status().expect("status after refusal"), staged); + + let control = GitOperationControl::default(); + control.cancel(); + assert_eq!( + local.push_controlled(&non_ff_remote, Some("main"), &Credentials, &control), + Err(GitError::Cancelled) + ); + + let temporary = tempfile::tempdir().expect("temporary directory"); + let local = local_repository(&temporary.path().join("local")); + commit(&local, ".gpg-id", b"ALICE\n", "Initialize"); + let user_key = key(); + let server = start_server( + user_key.public_key().clone(), + None, + None, + ReceiveBehavior::SlowSuccess, + false, + 1, + ); + let remote = remote(temporary.path(), &server, &user_key, false); + let mut local = local; + local.add_remote("origin", remote.url()).expect("remote"); + let control = GitOperationControl::default(); + let cancel = control.clone(); + let cancellation = thread::spawn(move || { + thread::sleep(Duration::from_millis(100)); + cancel.cancel(); + }); + assert!(matches!( + local.push_controlled(&remote, Some("main"), &Credentials, &control), + Err(GitError::PushOutcomeUnknown { .. }) + )); + cancellation.join().expect("cancellation"); + assert_eq!(tracking_id(&local), None); + server.join.join().expect("server"); + assert_eq!( + server + .observed + .lock() + .expect("observed") + .receive_requests + .len(), + 1 + ); +} + +#[test] +fn untrusted_and_unauthenticated_clients_never_request_receive_pack() { + let temporary = tempfile::tempdir().expect("temporary directory"); + let local = local_repository(&temporary.path().join("local")); + commit(&local, ".gpg-id", b"ALICE\n", "Initialize"); + let accepted_identity = key(); + let server = start_server( + accepted_identity.public_key().clone(), + None, + None, + ReceiveBehavior::Success, + false, + 2, + ); + let changed = remote(temporary.path(), &server, &accepted_identity, false); + fs::write( + temporary.path().join("known_hosts"), + format!( + "[127.0.0.1]:{} {}\n", + server.port, + key().public_key().to_openssh().expect("wrong host key") + ), + ) + .expect("changed known hosts"); + let mut local = local; + local.add_remote("origin", changed.url()).expect("remote"); + assert!(matches!( + local.push(&changed, Some("main"), &Credentials), + Err(GitError::ChangedSshHostKey { .. }) + )); + + let rejected = remote(temporary.path(), &server, &key(), false); + assert_eq!( + local.push(&rejected, Some("main"), &Credentials), + Err(GitError::SshAuthenticationRejected) + ); + server.join.join().expect("server"); + assert!( + server + .observed + .lock() + .expect("observed") + .commands + .is_empty() + ); +} + +#[test] +fn synchronization_pulls_then_pushes_over_both_ssh_url_forms() { + for scp_like in [false, true] { + let temporary = tempfile::tempdir().expect("temporary directory"); + let source = local_repository(&temporary.path().join("source")); + let base = commit(&source, ".gpg-id", b"ALICE\n", "Initialize"); + let user_key = key(); + let server = start_server( + user_key.public_key().clone(), + Some(upload_fixture(&source)), + Some(base.clone()), + ReceiveBehavior::Success, + false, + 3, + ); + let remote = remote(temporary.path(), &server, &user_key, scp_like); + let destination = temporary.path().join("clone"); + let clone = GitRepository::clone_into(&destination, identity(), &remote, &Credentials) + .expect("clone"); + let new = commit(&clone, "local.gpg", b"local", "Local change"); + let (pull, push) = clone.sync(&remote, &Credentials).expect("synchronize"); + assert_eq!(pull, PullOutcome::UpToDate); + assert_eq!(push.new_id(), new); + assert_eq!(tracking_id(&clone), Some(new)); + server.join.join().expect("server"); + let observed = server.observed.lock().expect("observed"); + assert_eq!( + observed + .commands + .iter() + .filter(|command| command.starts_with(b"git-upload-pack")) + .count(), + 2 + ); + assert_eq!( + observed + .commands + .iter() + .filter(|command| command.starts_with(b"git-receive-pack")) + .count(), + 1 + ); + let path = if scp_like { + "'team/store.git'" + } else { + "'/team/store.git'" + }; + let upload = format!("git-upload-pack {path}").into_bytes(); + let receive = format!("git-receive-pack {path}").into_bytes(); + assert!( + observed + .commands + .iter() + .all(|command| command == &upload || command == &receive) + ); + } +} + +#[test] +fn synchronization_never_pushes_after_pull_failure() { + let temporary = tempfile::tempdir().expect("temporary directory"); + let local = local_repository(&temporary.path().join("local")); + commit(&local, ".gpg-id", b"ALICE\n", "Initialize"); + let user_key = key(); + let server = start_server( + user_key.public_key().clone(), + None, + None, + ReceiveBehavior::Success, + true, + 1, + ); + let remote = remote(temporary.path(), &server, &user_key, false); + let mut local = local; + local.add_remote("origin", remote.url()).expect("remote"); + assert!(local.sync(&remote, &Credentials).is_err()); + server.join.join().expect("server"); + let commands = &server.observed.lock().expect("observed").commands; + assert_eq!(commands.len(), 1); + assert!(commands[0].starts_with(b"git-upload-pack")); +} diff --git a/crates/storage/src/ssh.rs b/crates/storage/src/ssh.rs index 842b809..1841ffc 100644 --- a/crates/storage/src/ssh.rs +++ b/crates/storage/src/ssh.rs @@ -47,6 +47,12 @@ pub(crate) struct SshGitCommand { pub(crate) completion: SshCommandCompletion, } +pub(crate) struct SshRawGitCommand { + pub(crate) stdout: SshStdout, + pub(crate) stdin: SshStdin, + pub(crate) completion: SshCommandCompletion, +} + pub(crate) type SshCommandCompletion = tokio::sync::oneshot::Receiver>; pub(crate) struct SshStdout { @@ -195,10 +201,50 @@ impl SshSession { remote: &GitRemote, control: &GitOperationControl, ) -> Result { - let RemoteEndpoint::Ssh(endpoint) = remote.endpoint() else { - return Err(GitError::ForbiddenRemoteUrl); - }; - let command = upload_pack_command(endpoint.path())?; + let endpoint = remote + .endpoint() + .as_ssh() + .ok_or(GitError::ForbiddenRemoteUrl)?; + let SshRawGitCommand { + stdout, + stdin, + completion, + } = self.open_git_command(remote, GitService::UploadPack, control)?; + let transport = gix::protocol::transport::client::git::blocking_io::Connection::new( + stdout, + stdin, + gix::protocol::transport::Protocol::V1, + endpoint.path().as_str().as_bytes().to_vec(), + None::<(String, Option)>, + gix::protocol::transport::client::git::ConnectMode::Process, + false, + ) + .custom_url(Some(remote.url().into())); + Ok(SshGitCommand { + transport, + completion, + }) + } + + pub(crate) fn open_receive_pack( + &self, + remote: &GitRemote, + control: &GitOperationControl, + ) -> Result { + self.open_git_command(remote, GitService::ReceivePack, control) + } + + fn open_git_command( + &self, + remote: &GitRemote, + service: GitService, + control: &GitOperationControl, + ) -> Result { + let endpoint = remote + .endpoint() + .as_ssh() + .ok_or(GitError::ForbiddenRemoteUrl)?; + let command = git_service_command(service, endpoint.path())?; let channel = self.runtime.block_on(async { controlled( tokio::time::timeout( @@ -222,22 +268,13 @@ impl SshSession { 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 { + Ok(SshRawGitCommand { + stdout: 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)>, - gix::protocol::transport::client::git::ConnectMode::Process, - false, - ) - .custom_url(Some(remote.url().into())); - Ok(SshGitCommand { - transport, + stdin: SshStdin { sender: input_tx }, completion: completion_rx, }) } @@ -359,9 +396,19 @@ async fn pump_command( } } -fn upload_pack_command(path: &SshRepositoryPath) -> Result, GitError> { +#[derive(Clone, Copy)] +enum GitService { + UploadPack, + ReceivePack, +} + +fn git_service_command(service: GitService, path: &SshRepositoryPath) -> Result, GitError> { + let service = match service { + GitService::UploadPack => "git-upload-pack", + GitService::ReceivePack => "git-receive-pack", + }; let path = shell_quote(path.as_str()); - let command = format!("git-upload-pack {path}"); + let command = format!("{service} {path}"); if command.len() > 64 * 1024 { return Err(GitError::ForbiddenRemoteUrl); } @@ -949,7 +996,9 @@ mod tests { repository::SecretBytes, }; - use super::{SshSession, persist_confirmed_host, ssh_host_key, upload_pack_command}; + use super::{ + GitService, SshSession, git_service_command, persist_confirmed_host, ssh_host_key, + }; struct Passphrase(Option<&'static [u8]>); @@ -1321,13 +1370,18 @@ mod tests { ), ] { 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] - ); + for (service, expected_service) in [ + (GitService::UploadPack, "git-upload-pack"), + (GitService::ReceivePack, "git-receive-pack"), + ] { + let command = git_service_command(service, 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"), + [expected_service, expected] + ); + } } assert!(RemoteEndpoint::parse("git@example.test:-upload-pack").is_err()); assert!(RemoteEndpoint::parse("git@example.test:repo\nsecond").is_err()); diff --git a/docs/configuration.md b/docs/configuration.md index 7e42c62..0f83a03 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -97,7 +97,8 @@ configuration, try additional keys, prompt for passwords or keyboard-interactive authentication, launch an agent, or invoke proxy/helper commands. In builds with the `ssh` feature, the same configuration drives branch discovery, clone, fetch, and pull over the embedded upload-pack channel; -push support is enabled separately when receive-pack is available. +push and full pull-then-push synchronization use the matching embedded +receive-pack channel. The typed endpoint model is always available so an SSH remote remains readable through the Rust API even when the binary was built without SSH. Such a build diff --git a/docs/git-synchronization.md b/docs/git-synchronization.md index 5767262..4c3904f 100644 --- a/docs/git-synchronization.md +++ b/docs/git-synchronization.md @@ -80,6 +80,18 @@ verification, object limits, and atomic ref updates, while the existing IronStorage code continues to own clone staging, checkout, merge, conflict, and rollback behavior. +SSH push opens the same verified and authenticated session-channel boundary for +`git-receive-pack ''`. Storage validates the advertisement, rejects +non-fast-forward updates before sending, constructs the complete reachable +object pack, then half-closes channel input and drains the status and bounded +stderr streams. The remote-tracking ref advances only after `unpack ok`, an +`ok` for the selected ref, a zero service exit, and clean channel completion. +Unpack and ref-policy rejections remain typed failures; a malformed response, +disconnect, or cancellation after sending is an unknown outcome that requires +a fresh fetch before retrying and is never replayed automatically. Synchronize +always completes pull first and cannot open receive-pack after a failed, +cancelled, or conflicted pull. + Pull refuses a dirty worktree. It fast-forwards when possible and otherwise uses the embedded three-way tree merge. Unresolved paths are returned as typed `MergeConflicts`; no conflict markers or partial checkout are written. Checkout