Sign in

flagon-io/g1t

public

Where people and agents ship software together. The open-source git platform for the whole job: issues, agents, checks and deploys to the edge.

gitagentscloudflarerustforgemcp

g1t/crates/sshd/src/main.rs

281 lines9,721 bytesCodeBlame
1//! SSH front end for g1t. Accepts `git@g1t.sh:owner/repo.git`, authenticates
2//! the client's public key against the g1t Worker, and bridges git to
3//! Artifacts over HTTPS.
4//!
5//! Connections arrive either as raw TCP or wrapped in a WebSocket, which is
6//! how they reach a Cloudflare Container when tunnelled through a Worker.
7
8mod api;
9mod git;
10
11use std::collections::HashMap;
12use std::sync::Arc;
13use std::time::Duration;
14
15use anyhow::{Context, Result};
16use futures_util::{SinkExt, StreamExt};
17use russh::keys::PrivateKey;
18use russh::keys::ssh_key::{HashAlg, PublicKey};
19use russh::server::{Auth, Config, Handler, Msg, Session};
20use russh::{Channel, ChannelId, MethodKind, MethodSet};
21use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt, BufReader};
22use tokio::net::TcpListener;
23use tokio_tungstenite::tungstenite::Message;
24
25use api::{Api, Service, User};
26
27const TCP_ADDR: &str = "0.0.0.0:2222";
28const WEBSOCKET_ADDR: &str = "0.0.0.0:8080";
29
30#[tokio::main]
31async fn main() -> Result<()> {
32 let host_key = std::env::var("SSH_HOST_KEY").context("SSH_HOST_KEY is not set")?;
33 let api = Arc::new(Api::new(
34 std::env::var("G1T_API").unwrap_or_else(|_| "https://g1t.sh".into()),
35 std::env::var("INTERNAL_SECRET").context("INTERNAL_SECRET is not set")?,
36 ));
37 let config = Arc::new(Config {
38 keys: vec![PrivateKey::from_openssh(host_key).context("invalid SSH_HOST_KEY")?],
39 methods: MethodSet::from(&[MethodKind::PublicKey][..]),
40 auth_rejection_time: Duration::from_secs(1),
41 auth_rejection_time_initial: Some(Duration::ZERO),
42 inactivity_timeout: Some(Duration::from_secs(300)),
43 ..Default::default()
44 });
45
46 let tcp = TcpListener::bind(TCP_ADDR).await?;
47 let websocket = TcpListener::bind(WEBSOCKET_ADDR).await?;
48 eprintln!("g1t-sshd listening on {TCP_ADDR} (tcp) and {WEBSOCKET_ADDR} (websocket)");
49 loop {
50 tokio::select! {
51 accepted = tcp.accept() => {
52 let (stream, _) = accepted?;
53 tokio::spawn(run_session(config.clone(), api.clone(), stream));
54 }
55 accepted = websocket.accept() => {
56 let (stream, _) = accepted?;
57 let (config, api) = (config.clone(), api.clone());
58 tokio::spawn(async move {
59 match tokio_tungstenite::accept_async(stream).await {
60 Ok(socket) => run_session(config, api, websocket_stream(socket)).await,
61 Err(error) => eprintln!("websocket handshake failed: {error}"),
62 }
63 });
64 }
65 }
66 }
67}
68
69async fn run_session<S>(config: Arc<Config>, api: Arc<Api>, stream: S)
70where
71 S: AsyncRead + AsyncWrite + Unpin + Send + 'static,
72{
73 let handler = Connection {
74 api,
75 user: None,
76 git_protocol: None,
77 channels: HashMap::new(),
78 };
79 let result = match russh::server::run_stream(config, stream, handler).await {
80 Ok(session) => session.await,
81 Err(error) => Err(error),
82 };
83 if let Err(error) = result {
84 eprintln!("session ended: {error:#}");
85 }
86}
87
88/// Presents a WebSocket's binary messages as a plain byte stream.
89fn websocket_stream<S>(socket: tokio_tungstenite::WebSocketStream<S>) -> tokio::io::DuplexStream
90where
91 S: AsyncRead + AsyncWrite + Unpin + Send + 'static,
92{
93 let (ours, theirs) = tokio::io::duplex(64 * 1024);
94 let (mut sink, mut source) = socket.split();
95 let (mut from_ssh, mut to_ssh) = tokio::io::split(theirs);
96 tokio::spawn(async move {
97 while let Some(Ok(message)) = source.next().await {
98 if let Message::Binary(bytes) = message {
99 if to_ssh.write_all(&bytes).await.is_err() {
100 break;
101 }
102 }
103 }
104 let _ = to_ssh.shutdown().await;
105 });
106 tokio::spawn(async move {
107 let mut buffer = vec![0u8; 32 * 1024];
108 loop {
109 match from_ssh.read(&mut buffer).await {
110 Ok(0) | Err(_) => break,
111 Ok(read) => {
112 let message = Message::Binary(buffer[..read].to_vec().into());
113 if sink.send(message).await.is_err() {
114 break;
115 }
116 }
117 }
118 }
119 let _ = sink.close().await;
120 });
121 ours
122}
123
124struct Connection {
125 api: Arc<Api>,
126 user: Option<User>,
127 /// Value of the `GIT_PROTOCOL` environment variable, if the client sent it.
128 git_protocol: Option<String>,
129 channels: HashMap<ChannelId, Channel<Msg>>,
130}
131
132impl Connection {
133 async fn lookup(&self, key: &PublicKey) -> Result<Option<User>> {
134 let fingerprint = key.fingerprint(HashAlg::Sha256).to_string();
135 self.api.user_for_key(&fingerprint).await
136 }
137}
138
139impl Handler for Connection {
140 type Error = anyhow::Error;
141
142 async fn auth_publickey_offered(&mut self, _: &str, key: &PublicKey) -> Result<Auth> {
143 Ok(match self.lookup(key).await? {
144 Some(_) => Auth::Accept,
145 None => Auth::reject(),
146 })
147 }
148
149 /// Called once the client has proven it holds the private key.
150 async fn auth_publickey(&mut self, _: &str, key: &PublicKey) -> Result<Auth> {
151 self.user = self.lookup(key).await?;
152 Ok(match self.user {
153 Some(_) => Auth::Accept,
154 None => Auth::reject(),
155 })
156 }
157
158 async fn channel_open_session(
159 &mut self,
160 channel: Channel<Msg>,
161 reply: russh::server::ChannelOpenHandle,
162 _: &mut Session,
163 ) -> Result<()> {
164 self.channels.insert(channel.id(), channel);
165 reply.accept().await;
166 Ok(())
167 }
168
169 async fn env_request(
170 &mut self,
171 channel: ChannelId,
172 name: &str,
173 value: &str,
174 session: &mut Session,
175 ) -> Result<()> {
176 if name == "GIT_PROTOCOL" {
177 self.git_protocol = Some(value.to_owned());
178 }
179 session.channel_success(channel)?;
180 Ok(())
181 }
182
183 async fn shell_request(&mut self, id: ChannelId, session: &mut Session) -> Result<()> {
184 let (Some(user), Some(channel)) = (&self.user, self.channels.remove(&id)) else {
185 return Ok(session.channel_failure(id)?);
186 };
187 session.channel_success(id)?;
188 let greeting = format!(
189 "Hi {}! You've successfully authenticated, but g1t does not provide shell access.\r\n",
190 user.username
191 );
192 tokio::spawn(async move {
193 let _ = channel.data(greeting.as_bytes()).await;
194 finish(channel, 1).await;
195 });
196 Ok(())
197 }
198
199 async fn exec_request(
200 &mut self,
201 id: ChannelId,
202 command: &[u8],
203 session: &mut Session,
204 ) -> Result<()> {
205 let (Some(user), Some(channel)) = (self.user.clone(), self.channels.remove(&id)) else {
206 return Ok(session.channel_failure(id)?);
207 };
208 session.channel_success(id)?;
209
210 let command = String::from_utf8_lossy(command).into_owned();
211 let protocol_v2 = self
212 .git_protocol
213 .as_deref()
214 .is_some_and(|value| value.split(':').any(|part| part == "version=2"));
215 let api = self.api.clone();
216 tokio::spawn(async move {
217 let (mut read_half, write_half) = channel.split();
218 let mut reader = BufReader::new(read_half.make_reader());
219 let mut writer = write_half.make_writer();
220 let result =
221 run_git(&api, &user, &command, protocol_v2, &mut reader, &mut writer).await;
222 let status = match result {
223 Ok(()) => 0,
224 Err(error) => {
225 eprintln!("{}: `{command}` failed: {error:#}", user.username);
226 let _ = writer.write_all(&git::error_pkt("internal error")).await;
227 1
228 }
229 };
230 let _ = writer.shutdown().await;
231 let _ = write_half.exit_status(status).await;
232 let _ = write_half.eof().await;
233 let _ = write_half.close().await;
234 });
235 Ok(())
236 }
237}
238
239async fn finish(channel: Channel<Msg>, status: u32) {
240 let _ = channel.exit_status(status).await;
241 let _ = channel.eof().await;
242 let _ = channel.close().await;
243}
244
245/// Parses `git-upload-pack 'owner/repo.git'` and friends.
246fn parse_command(command: &str) -> Option<(Service, &str, &str)> {
247 let (program, path) = command.trim().rsplit_once(' ')?;
248 let service = match program {
249 "git-upload-pack" | "git upload-pack" => Service::UploadPack,
250 "git-receive-pack" | "git receive-pack" => Service::ReceivePack,
251 _ => return None,
252 };
253 let path = path.trim_matches(['\'', '"']).trim_start_matches('/');
254 let path = path.strip_suffix(".git").unwrap_or(path);
255 let (owner, repo) = path.split_once('/')?;
256 (!owner.is_empty() && !repo.is_empty() && !repo.contains('/')).then_some((service, owner, repo))
257}
258
259async fn run_git<R, W>(
260 api: &Api,
261 user: &User,
262 command: &str,
263 protocol_v2: bool,
264 reader: &mut R,
265 writer: &mut W,
266) -> Result<()>
267where
268 R: tokio::io::AsyncBufRead + Unpin,
269 W: AsyncWrite + Unpin,
270{
271 let Some((service, owner, repo)) = parse_command(command) else {
272 writer
273 .write_all(&git::error_pkt("g1t only supports git over SSH"))
274 .await?;
275 return Ok(());
276 };
277 match api.access(user, owner, repo, service).await? {
278 Ok(access) => git::serve(&api.http, &access, service, protocol_v2, reader, writer).await,
279 Err(message) => Ok(writer.write_all(&git::error_pkt(&message)).await?),
280 }
281}