Custom git protocol sketch

Signed-off-by: Alexis Sellier <alexis@radicle.xyz>

radicle/src/storage/git.rs

@@ -1,3 +1,5 @@
+pub mod transport;
+
 use std::collections::{BTreeMap, HashMap};
 use std::path::{Path, PathBuf};
 use std::{fmt, fs, io};
@@ -789,6 +791,83 @@ mod tests {
         assert_eq!(bob_master.target().unwrap(), alice_head);
     }
 
+    #[test]
+    fn test_upload_pack() {
+        use std::io::{Read, Write};
+        use std::{io, net, process, thread};
+
+        let socket = net::TcpListener::bind(net::SocketAddr::from(([0, 0, 0, 0], 0))).unwrap();
+        let addr = socket.local_addr().unwrap();
+        let tmp = tempfile::tempdir().unwrap();
+        let source_path = tmp.path().join("source");
+        let target_path = tmp.path().join("target");
+        let (_source, _) = fixtures::repository(&source_path);
+
+        let t = thread::spawn(move || {
+            let (stream, _) = socket.accept().unwrap();
+            // NOTE: `--stateless-rpc` doesn't work, neither does `GIT_PROTOCOL=version=2`.
+            let mut child = process::Command::new("git")
+                .current_dir(source_path.join(".git"))
+                .arg("upload-pack")
+                .arg("--strict")
+                .arg(".")
+                .stdout(process::Stdio::piped())
+                .stdin(process::Stdio::piped())
+                .spawn()
+                .unwrap();
+
+            let mut stdin = child.stdin.take().unwrap();
+            let mut stdout = child.stdout.take().unwrap();
+
+            let mut stream_r = stream.try_clone().unwrap();
+            let mut stream_w = stream;
+
+            let t = thread::spawn(move || {
+                let mut buf = [0u8; 1024];
+
+                while let Ok(n) = stream_r.read(&mut buf) {
+                    if n == 0 {
+                        break;
+                    }
+                    if stdin.write_all(&buf[..n]).is_err() {
+                        break;
+                    }
+                }
+            });
+            io::copy(&mut stdout, &mut stream_w).unwrap();
+
+            t.join().unwrap();
+            child.wait().unwrap();
+        });
+
+        let mut updates = Vec::new();
+        {
+            let mut callbacks = git2::RemoteCallbacks::new();
+            let mut opts = git2::FetchOptions::default();
+
+            callbacks.update_tips(|name, _, _| {
+                updates.push(name.to_owned());
+                true
+            });
+            opts.remote_callbacks(callbacks);
+
+            let target = git2::Repository::init_bare(target_path).unwrap();
+            let refs: &[&str] = &["refs/*:refs/*"];
+
+            // Register the `rad://` transport.
+            transport::register("rad").unwrap();
+            // Fetch with the `rad://` transport.
+            target
+                .remote_anonymous(&format!("rad://{}", addr))
+                .unwrap()
+                .fetch(refs, Some(&mut opts), None)
+                .unwrap();
+
+            t.join().unwrap();
+        }
+        assert_eq!(updates, vec![String::from("refs/heads/master")]);
+    }
+
     #[test]
     fn test_sign_refs() {
         let tmp = tempfile::tempdir().unwrap();

radicle/src/storage/git/transport.rs

@@ -0,0 +1,90 @@
+use std::str::FromStr;
+use std::sync::atomic;
+use std::{io, net};
+
+use crate::git;
+
+/// Git smart protocol over a TCP stream.
+pub struct Smart;
+
+impl git2::transport::SmartSubtransport for Smart {
+    fn action(
+        &self,
+        url: &str,
+        action: git2::transport::Service,
+    ) -> Result<Box<dyn git2::transport::SmartSubtransportStream>, git2::Error> {
+        let url = git::Url::from_bytes(url.as_bytes())
+            .map_err(|e| git2::Error::from_str(e.to_string().as_str()))?;
+
+        let addr = if let (Some(host), Some(port)) = (url.host, url.port) {
+            // TODO: Support hostnames.
+            net::SocketAddr::new(
+                net::IpAddr::from_str(&host)
+                    .map_err(|e| git2::Error::from_str(e.to_string().as_str()))?,
+                port,
+            )
+        } else {
+            return Err(git2::Error::from_str("Git URL must have a host and port"));
+        };
+
+        let stream = std::net::TcpStream::connect(addr)
+            .map_err(|e| git2::Error::from_str(e.to_string().as_str()))?;
+
+        match action {
+            git2::transport::Service::UploadPackLs => {}
+            git2::transport::Service::UploadPack => {}
+            git2::transport::Service::ReceivePack => {
+                return Err(git2::Error::from_str(
+                    "git-receive-pack is not supported with the custom transport",
+                ));
+            }
+            git2::transport::Service::ReceivePackLs => {
+                return Err(git2::Error::from_str(
+                    "git-receive-pack is not supported with the custom transport",
+                ));
+            }
+        }
+        Ok(Box::new(Stream { stream }))
+    }
+
+    fn close(&self) -> Result<(), git2::Error> {
+        Ok(())
+    }
+}
+
+struct Stream {
+    stream: std::net::TcpStream,
+}
+
+impl io::Write for Stream {
+    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
+        self.stream.write(buf)
+    }
+
+    fn flush(&mut self) -> std::io::Result<()> {
+        self.stream.flush()
+    }
+}
+
+impl io::Read for Stream {
+    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
+        self.stream.read(buf)
+    }
+}
+
+/// Register the "smart" transport with `git`.
+pub fn register(prefix: &str) -> Result<(), git2::Error> {
+    static REGISTERED: atomic::AtomicBool = atomic::AtomicBool::new(false);
+
+    if !REGISTERED.swap(true, atomic::Ordering::SeqCst) {
+        unsafe {
+            git2::transport::register(prefix, move |remote| {
+                git2::transport::Transport::smart(remote, false, Smart)
+            })
+        }
+    } else {
+        Err(git2::Error::from_str(
+            "custom git transport is already registered",
+        ))
+    }
+}