p2p internals #4 - Disséquons git

// https://docs.rs/git2/0.13.17/git2/transport/fn.register.html

use bichannel::{ SendError, RecvError };
use git2::Error;
use git2::transport::SmartSubtransportStream;
use git2::transport::{Service, SmartSubtransport, Transport};
use std::io;
use std::io::prelude::*;
use std::sync::{Arc, Mutex};

static HOST_TAG: &str = "host=";

// Note : Ce transport est purement fictif et ne fait que ajouter un header
// de 4 octets: "WOLF"
pub struct WolfChannel
{
    pub channel: bichannel::Channel<Vec<u8>, Vec<u8>>,
}

impl WolfChannel
{
    pub fn recv(&self) -> Result<Vec<u8>, RecvError> {
        let res = self.channel.recv();
        if !res.is_ok() {
            return res;
        }
        let mut res = res.unwrap();
        res.drain(0..4);
        Ok(res)
    }

    pub fn send(&self, data: Vec<u8>) -> Result<(), SendError<Vec<u8>>> {
        let mut to_send = "WOLF".as_bytes().to_vec();
        to_send.extend(data);
        self.channel.send(to_send)
    }
}

pub type Channel = Arc<Mutex<WolfChannel>>;

// Maintenant, écrivons notre smart transport pour git2-rs répondant au scheme wolf://
struct WolfTransport {
    channel: Channel,
}

struct WolfSubTransport {
    action: Service,
    channel: Channel,
    url: String,
    sent_request: bool
}

pub unsafe fn register(channel: Channel) {
    git2::transport::register("wolf", move |remote| factory(remote, channel.clone())).unwrap();
}

fn factory(remote: &git2::Remote<'_>, channel: Channel) -> Result<Transport, Error> {
    Transport::smart(
        remote,
        false, // rpc = false, signifie que notre channel est connecté durant tout le long de la transaction
        WolfTransport {
            channel
        },
    )
}

impl SmartSubtransport for WolfTransport {
    /**
     * Génère un nouveau transport à utiliser. Comme rpc = false, on ne répond que à upload-pack-ls & receive-pack-ls
     */
    fn action(
        &self,
        url: &str,
        action: Service,
    ) -> Result<Box<dyn SmartSubtransportStream>, Error> {
        Ok(Box::new(WolfSubTransport {
            action,
            channel: self.channel.clone(),
            url: String::from(url),
            sent_request: false,
        }))
    }

    fn close(&self) -> Result<(), Error> {
        Ok(())
    }
}

impl WolfTransport {
    fn generate_request(cmd: &str, url: &str) -> Vec<u8> {
        // url format = wolf://host/repo
        // Note: Cette requête n'est envoyé que quand le client démarre sa partie afin de notifier le serveur
        let sep = url.rfind('/').unwrap();
        let host = url.get(7..sep).unwrap();
        let repo = url.get(sep..).unwrap();

        let null_char = '\0';
        let total = 4                                   /* 4 bytes for the len len */
                    + cmd.len()                         /* followed by the command */
                    + 1                                 /* space */
                    + repo.len()                        /* repo to clone */
                    + 1                                 /* \0 */
                    + HOST_TAG.len() + host.len()       /* host=server */
                    + 1                                 /* \0 */;
        let request = format!("{:04x}{} {}{}{}{}{}", total, cmd, repo, null_char, HOST_TAG, host, null_char);
        request.as_bytes().to_vec()
    }
}

impl Read for WolfSubTransport {
    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
        // Envoie de la requête pour le serveur
        if !self.sent_request {
            let cmd = match self.action {
                Service::UploadPackLs => "git-upload-pack",
                Service::UploadPack => "git-upload-pack",
                Service::ReceivePackLs => "git-receive-pack",
                Service::ReceivePack => "git-receive-pack",
            };
            let cmd = WolfTransport::generate_request(cmd, &*self.url);
            let _ = self.channel.lock().unwrap().send(cmd);
            self.sent_request = true;
        }
        // Lis la réponse du serveur
        let mut recv = self.channel.lock().unwrap().recv().unwrap_or(Vec::new());
        let mut iter = recv.drain(..);
        let mut idx = 0;
        while let Some(v) = iter.next() {
            buf[idx] = v;
            idx += 1;
        }
        Ok(idx)
    }
}

impl Write for WolfSubTransport {
    fn write(&mut self, data: &[u8]) -> io::Result<usize> {
        let _ = self.channel.lock().unwrap().send(data.to_vec());
        Ok(data.len())
    }
    fn flush(&mut self) -> io::Result<()> {
        // Non utilisé dans notre cas
        Ok(())
    }
}

use crate::wolftransport::Channel;
use git2::{ Buf, Oid, Repository, Sort };
use std::cmp::{ max, min };
use std::i64;
use std::str;

static FLUSH_PKT: &str = "0000";
static NAK_PKT: &str = "0008NAK\n";
static DONE_PKT: &str = "0009done\n";
static WANT_CMD: &str = "want";
static HAVE_CMD: &str = "have";
static UPLOAD_PACK_CMD: &str = "git-upload-pack";

/**
 * Représente un serveur git fonctionnant sur notre serveur personnalisé servant un répertoire
 */
pub struct Server {
    repository: Repository,
    channel: Channel,
    wanted: String,
    common: String,
    have: Vec<String>,
    buf: Vec<u8>,
    stop: bool,
}

impl Server {
    pub fn new(channel: Channel, path: &str) -> Self {
        let repository = Repository::open(path).unwrap();
        Self {
            repository,
            channel,
            wanted: String::new(),
            common: String::new(),
            have: Vec::new(),
            buf: Vec::new(),
            stop: false,
        }
    }

    pub fn run(&mut self) {
        // Stop est mis à true quand le clone est terminé.
        while !self.stop {
            let buf = self.channel.lock().unwrap().recv().unwrap();
            self.recv(buf);
        }
    }

    fn recv(&mut self, buf: Vec<u8>) {
        let mut buf = Some(buf);
        let mut need_more_parsing = true;
        // Comme le client peut envoyer plusieurs pkt-line, on traite
        // jusqu'à ce qu'on soit prêt à recevoir de nouveaux ordres.
        while need_more_parsing {
            need_more_parsing = self.parse(buf.take());
        }
    }

    fn parse(&mut self, buf: Option<Vec<u8>>) -> bool {
        // Parse la taille du pjt-line
        // Référence : https://github.com/git/git/blob/master/Documentation/technical/protocol-common.txt#L51
        // Les 4 premiers octets stockent la taille, sauf pour 0000 qui est le paquet FLUSH
        if buf.is_some() {
            self.buf.append(&mut buf.unwrap());
        }
        let pkt_len = str::from_utf8(&self.buf[0..4]).unwrap();
        let pkt_len = max(4 as usize, i64::from_str_radix(pkt_len, 16).unwrap() as usize);
        let pkt : Vec<u8> = self.buf.drain(0..pkt_len).collect();
        let pkt = str::from_utf8(&pkt[0..pkt_len]).unwrap();
        println!("RECV: {}", pkt);

        if pkt.find(UPLOAD_PACK_CMD) == Some(4) {
            // Cf: https://github.com/git/git/blob/master/Documentation/technical/pack-protocol.txt#L166
            // Envoi des références
            println!("Upload pack command detected");
            // Note: la commande peut aussi contenir des paramètres comme version=1.
            // Pour cet article, on ignore ce cas.
            self.send_references_capabilities();
        } else if pkt.find(WANT_CMD) == Some(4) {
            // Référence :
            // https://github.com/git/git/blob/master/Documentation/technical/pack-protocol.txt#L229
            // NOTE: Un client peut envoyer plusieurs want. À noter aussi que le premier want est suivi des
            // options que le client souhaite. Pour simplifier, ces cas sont ignorés ici.
            self.wanted = String::from(pkt.get(9..49).unwrap()); // on prend juste le commit id
            println!("Detected wanted commit: {}", self.wanted);
        } else if pkt.find(HAVE_CMD) == Some(4) {
            // Référence :
            // https://github.com/git/git/blob/master/Documentation/technical/pack-protocol.txt#L390
            // NOTE: possible d'améliorer cette partie pour le multi-ack
            let have_commit = String::from(pkt.get(9..49).unwrap()); // on prend juste le commit id
            if self.common.is_empty() {
                if self.repository.find_commit(Oid::from_str(&*have_commit).unwrap()).is_ok() {
                    self.common = have_commit.clone();
                    println!("Set common commit to: {}", self.common);
                }
            }
            self.have.push(have_commit);
        } else if pkt == DONE_PKT {
            // Référence :
            // https://github.com/git/git/blob/master/Documentation/technical/pack-protocol.txt#L390
            // NOTE: multi-ack ignoré ici. Si pas de base commune, on envoie seulement NAK.
            println!("Peer negotiation is done. Answering to want order");
            let send_data = match self.common.is_empty() {
                true => self.nak(),
                false => self.ack_first(),
            };
            if send_data {
                self.send_pack_data();
            }
            self.stop = true;
        } else if pkt == FLUSH_PKT {
            if !self.have.is_empty() {
                // Référence :
                // https://github.com/git/git/blob/master/Documentation/technical/pack-protocol.txt#L390
                self.ack_common();
                self.nak();
            }
        } else {
            println!("Unwanted packet received: {}", pkt);
        }
        self.buf.len() != 0
    }

    fn send_references_capabilities(&self) {
        let current_head = self.repository.refname_to_id("HEAD").unwrap();
        let mut capabilities = format!("{} HEAD\0side-band side-band-64k shallow no-progress include-tag", current_head);
        capabilities = format!("{:04x}{}\n", capabilities.len() + 5 /* taille + \n */, capabilities);

        for name in self.repository.referencess().unwrap().names() {
            let reference : &str = name.unwrap();
            let oid = self.repository.refname_to_id(reference).unwrap();
            capabilities += &*format!("{:04x}{} {}\n", 6 /* taille + espace + \n */ + 40 /* oid */ + reference.len(), oid, reference);
        }

        print!("Send: {}", capabilities);
        self.channel.lock().unwrap().send(capabilities.as_bytes().to_vec()).unwrap();
        println!("Send: {}", FLUSH_PKT);
        self.channel.lock().unwrap().send(FLUSH_PKT.as_bytes().to_vec()).unwrap();
    }

    fn nak(&self) -> bool {
        print!("Send: {}", NAK_PKT);
        self.channel.lock().unwrap().send(NAK_PKT.as_bytes().to_vec()).is_ok()
    }

    fn ack_common(&self) -> bool {
        let length = 18 /* taille + ACK + espace * 2 + continue + \n */ + self.common.len();
        let msg = format!("{:04x}ACK {} continue\n", length, self.common);
        print!("Send: {}", msg);
        self.channel.lock().unwrap().send(msg.as_bytes().to_vec()).is_ok()
    }

    fn ack_first(&self) -> bool {
        let length = 9 /* taille + ACK + espace + \n */ + self.common.len();
        let msg = format!("{:04x}ACK {}\n", length, self.common);
        print!("Send: {}", msg);
        self.channel.lock().unwrap().send(msg.as_bytes().to_vec()).is_ok()
    }

    fn send_pack_data(&self) {
        println!("Send: [PACKFILE]");
        // Note: Ici, on ajoute tous les commits tant qu'on à pas
        // trouvé un commit annoncé par le client et qu'on ne sois
        // pas sûr d'avoir parcouru toutes les branches (ou jusqu'au
        // commit initial).
        let mut pb = self.repository.packbuilder().unwrap();
        let fetched = Oid::from_str(&*self.wanted).unwrap();
        let mut revwalk = self.repository.revwalk().unwrap();
        let _ = revwalk.push(fetched);
        let _ = revwalk.set_sorting(Sort::TOPOLOGICAL);

        let mut parents : Vec<String> = Vec::new();
        let mut have = false;

        while let Some(oid) = revwalk.next() {
            let oid = oid.unwrap();
            let oid_str = oid.to_string();
            have |= self.have.iter().find(|&o| *o == oid_str).is_some();
            if let Some(pos) = parents.iter().position(|p| *p == oid_str) {
                parents.remove(pos);
            }
            if have && parents.is_empty() {
                // Commit initial
                break;
            }
            let _ = pb.insert_commit(oid);
            let commit = self.repository.find_commit(oid).unwrap();
            let mut commit_parents = commit.parents();
            // On doit être sûr de parcourir tous les chemins
            while let Some(p) = commit_parents.next() {
                parents.push(p.id().to_string());
            }
        }

        // Note: Packbuilder possède des fonctions pour écrire des chunks et pas tout le packfile d'un coup.
        let mut data = Buf::new();
        let _ = pb.write_buf(&mut data);

        let len = data.len();
        let data : Vec<u8> = data.to_vec();
        let mut sent = 0;
        while sent < len {
            // cf https://github.com/git/git/blob/master/Documentation/technical/pack-protocol.txt#L166
            // Si side-band-64k est spécifié, on doit envoyer jusqu'à 65519 octets + 1 de contrôle par pkt-line.
            let pkt_size = min(65515, len - sent);
            // Le paquet: Size (4 octets), Control byte (0x01 pour les données), pack data.
            let pkt = format!("{:04x}", pkt_size + 5 /* taille + control */);
            self.channel.lock().unwrap().send(pkt.as_bytes().to_vec()).unwrap();
            self.channel.lock().unwrap().send(b"\x01".to_vec()).unwrap();
            self.channel.lock().unwrap().send(data[sent..(sent+pkt_size)].to_vec()).unwrap();
            sent += pkt_size;
        }

        println!("Send: {}", FLUSH_PKT);
        // Et on fini par un FLUSH
        self.channel.lock().unwrap().send(FLUSH_PKT.as_bytes().to_vec()).unwrap();
    }
}

mod wolftransport;
mod server;

use bichannel::channel;
use wolftransport::WolfChannel;
use git2::build::RepoBuilder;
use server::Server;
use std::env;
use std::sync::{Arc, Mutex};
use std::thread;
use std::path::Path;

fn main() {
    let args: Vec<String> = env::args().collect();
    if args.len() != 3 {
        println!("Usage: ./p2p-internal-git <src_dir> <dest_dir>");
        return;
    }
    
    let src_dir = args[1].clone();
    let dest_dir = Path::new(&args[2]);

    if dest_dir.is_dir() {
        println!("Can't clone into an existing directory");
        return;
    }

    // On utilise ici bichannel::channel() à des fins de démos
    // mais on peut le remplacer par le transport que l'on souhaite
    // par exemple un transport TLS.
    let (server_channel, transport_channel) = channel();
    let transport_channel = Arc::new(Mutex::new(WolfChannel {
        channel: transport_channel
    }));
    let server_channel = Arc::new(Mutex::new(WolfChannel {
        channel: server_channel
    }));
    unsafe {
        wolftransport::register(transport_channel);
    }

    let server = thread::spawn(move || {
        println!("Starting server for {}", src_dir);
        let mut server = Server::new(server_channel, &*src_dir);
        server.run();
    });

    // Note: "wolf://" est pour utiliser notre transport. localhost/zds n'est pas utilisé
    // parce que notre serveur ne gère qu'un seul répertoire.
    RepoBuilder::new().clone("wolf://localhost/zds", dest_dir).unwrap();
    println!("Cloned into {:?}!", dest_dir);

    server.join().expect("The server panicked");
}