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 scp_like in [false, true] { for old_kind in [None, Some("base")] { 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(), ReceiveBehavior::Success, false, 1, ); let remote = remote(temporary.path(), &server, &user_key, scp_like); 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"); let expected = if scp_like { b"git-receive-pack 'team/store.git'".to_vec() } else { b"git-receive-pack '/team/store.git'".to_vec() }; assert_eq!(observed.commands, [expected]); assert_eq!(observed.receive_requests.len(), 1); assert_pack_request(&observed.receive_requests[0], old.as_deref(), &new); } } for scp_like in [false, true] { 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, scp_like); 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")); }