Skip to content
Sign up

flagon-io/g1t

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