git_remote_iroh/
server.rs

1use std::path::PathBuf;
2
3use anyhow::Context;
4use iroh::{Endpoint, SecretKey, protocol::Router};
5use iroh_rings::RingGate;
6use tokio::{io::AsyncWriteExt, process::Command};
7use tokio_util::sync::CancellationToken;
8use tracing::{Instrument, info, warn, warn_span};
9
10use crate::{
11    GIT_IROH_ALPN, GitOp,
12    auth::{AuthAdmin, open_db},
13    cli::{PeerCommands, Subcommands},
14    keys,
15};
16
17pub struct Server {
18    git_dir: PathBuf,
19    key: SecretKey,
20    config_dir: PathBuf,
21}
22
23impl Server {
24    pub fn run_command(
25        commands: &Subcommands,
26        config_dir: &PathBuf,
27        git_dir: &PathBuf,
28    ) -> Result<(), anyhow::Error> {
29        let admin = AuthAdmin::new(config_dir, git_dir)?;
30        match commands {
31            Subcommands::Whoami => {
32                println!(
33                    "{} {}",
34                    admin.key.public().to_z32(),
35                    admin.key_file.to_string_lossy()
36                );
37                Ok(())
38            }
39            &Subcommands::Peer {
40                commands:
41                    Some(PeerCommands::Add {
42                        ref ring,
43                        ref name,
44                        ref id,
45                    }),
46                ..
47            } => admin.add_peer(ring, name, id).context("peer add"),
48            &Subcommands::Peer {
49                commands: None,
50                ref ring,
51            } => admin
52                .list_peers(ring.as_ref().unwrap())
53                .context("peer list"),
54            &Subcommands::Ring => admin.list_rings().context("ring list"),
55        }
56    }
57
58    pub fn new(config_dir: PathBuf, git_dir: &PathBuf) -> Result<Self, anyhow::Error> {
59        let (_, key) = keys::load_or_create_key(&config_dir, git_dir)?;
60        Ok(Self {
61            git_dir: git_dir.clone(),
62            key,
63            config_dir,
64        })
65    }
66
67    pub async fn run(&self, cancel: CancellationToken) -> Result<(), anyhow::Error> {
68        let reg = open_db(&self.config_dir)?;
69        let ep = Endpoint::builder(iroh::endpoint::presets::N0)
70            .secret_key(self.key.clone())
71            .alpns(vec![GIT_IROH_ALPN.to_vec()])
72            .bind()
73            .await
74            .context("failed to bind iroh endpoint")?;
75        ep.online().await;
76        let node_id = ep.id();
77        info!(addr = format!("iroh://{node_id}"), "server started");
78
79        let handler = GitIrohTransfer::new(self.git_dir.clone());
80        let gate = RingGate::new(reg, handler);
81        let router = Router::builder(ep.clone())
82            .accept(GIT_IROH_ALPN, gate)
83            .spawn();
84        cancel.cancelled().await;
85        router.shutdown().await?;
86        Ok(())
87    }
88}
89
90/// Git protocol handler for iroh.
91///
92/// Accepts incoming connections, reads the operation, spawns the appropriate
93/// command, connecting stdin/stdout to the iroh bidirectional stream.
94#[derive(Debug, Clone)]
95pub struct GitIrohTransfer {
96    /// the git repo directory to serve
97    repo_dir: PathBuf,
98}
99
100impl GitIrohTransfer {
101    pub fn new(repo_dir: PathBuf) -> Self {
102        Self { repo_dir }
103    }
104}
105
106impl iroh_rings::Transfer for GitIrohTransfer {
107    async fn can_access(&self, peer: &iroh::EndpointId, resource_id: &[u8]) -> bool {
108        // Allow public read access to any peer
109        if let Ok(GitOp::ClientFetchServerPush) = GitOp::from_resource_id(resource_id) {
110            info!(%peer, "allowing public read");
111            true
112        } else {
113            false
114        }
115    }
116
117    async fn transfer(
118        &self,
119        resource_id: &[u8],
120        send: &mut iroh::endpoint::SendStream,
121        recv: &mut iroh::endpoint::RecvStream,
122    ) -> anyhow::Result<()> {
123        let op = GitOp::from_resource_id(resource_id)?;
124        info!(?op);
125
126        // spawn the git command
127        let mut child = Command::new(op.to_string())
128            .arg(&self.repo_dir)
129            .stdin(std::process::Stdio::piped())
130            .stdout(std::process::Stdio::piped())
131            .stderr(std::process::Stdio::inherit())
132            .spawn()
133            .map_err(|e| std::io::Error::other(format!("spawn {op}: {e}")))?;
134
135        let mut child_stdin = child.stdin.take().unwrap();
136        let mut child_stdout = child.stdout.take().unwrap();
137
138        let cancel = CancellationToken::new();
139        {
140            let cancel = cancel.clone();
141            tokio::spawn(
142                async move {
143                    let status = child.wait().await?;
144                    if !status.success() {
145                        warn!(%op, %status);
146                    }
147                    cancel.cancel();
148                    Ok::<(), anyhow::Error>(())
149                }
150                .instrument(warn_span!("process wait")),
151            );
152        }
153
154        let (r1, r2) = tokio::join!(
155            async {
156                tokio::select! {
157                    res = tokio::io::copy(recv, &mut child_stdin) => { res?; Ok(()) }
158                    _ = cancel.cancelled() => { recv.stop(0u8.into())?; Ok::<(), anyhow::Error>(()) }
159                }
160            },
161            async {
162                tokio::select! {
163                    res = tokio::io::copy(&mut child_stdout, send) => { res?; Ok(()) }
164                    _ = cancel.cancelled() => { send.flush().await?; send.finish()?; Ok::<(), anyhow::Error>(()) }
165                }
166            },
167        );
168        r1.context("stdin -> send")?;
169        r2.context("recv -> stdout")?;
170        Ok(())
171    }
172}
git_remote_iroh/server.rs

git_remote_iroh/
server.rs
