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/git.rs

262 lines8,979 bytesCodeBlame
1//! Bridges git's stateful SSH transport to the stateless smart HTTP protocol
2//! spoken by Artifacts.
3
4use anyhow::{Result, bail, ensure};
5use flate2::{Decompress, FlushDecompress, Status};
6use futures_util::StreamExt;
7use tokio::io::{AsyncBufRead, AsyncBufReadExt, AsyncReadExt, AsyncWrite, AsyncWriteExt};
8
9use crate::api::{Access, Service};
10
11const FLUSH: &[u8] = b"0000";
12const ZERO_ID: &str = "0000000000000000000000000000000000000000";
13
14/// A pkt-line carrying `ERR`, which git prints as `fatal: remote error: …`.
15pub fn error_pkt(message: &str) -> Vec<u8> {
16 let payload = format!("ERR {message}\n");
17 format!("{:04x}{payload}", payload.len() + 4).into_bytes()
18}
19
20pub async fn serve<R, W>(
21 http: &reqwest::Client,
22 access: &Access,
23 service: Service,
24 protocol_v2: bool,
25 reader: &mut R,
26 writer: &mut W,
27) -> Result<()>
28where
29 R: AsyncBufRead + Unpin,
30 W: AsyncWrite + Unpin,
31{
32 let bridge = Bridge {
33 http,
34 access,
35 service,
36 // Artifacts only supports version 2 for fetches.
37 protocol_v2: protocol_v2 && service == Service::UploadPack,
38 };
39 match service {
40 Service::UploadPack => bridge.upload_pack(reader, writer).await,
41 Service::ReceivePack => bridge.receive_pack(reader, writer).await,
42 }
43}
44
45struct Bridge<'a> {
46 http: &'a reqwest::Client,
47 access: &'a Access,
48 service: Service,
49 protocol_v2: bool,
50}
51
52impl Bridge<'_> {
53 fn request(&self, method: reqwest::Method, path: &str) -> reqwest::RequestBuilder {
54 let mut request = self
55 .http
56 .request(method, format!("{}/{path}", self.access.remote))
57 .bearer_auth(&self.access.token);
58 if self.protocol_v2 {
59 request = request.header("git-protocol", "version=2");
60 }
61 request
62 }
63
64 /// Sends the ref advertisement, which over SSH comes without the
65 /// `# service=…` section that smart HTTP prepends.
66 async fn advertise<W: AsyncWrite + Unpin>(&self, writer: &mut W) -> Result<()> {
67 let service = self.service.as_str();
68 let response = self
69 .request(
70 reqwest::Method::GET,
71 &format!("info/refs?service={service}"),
72 )
73 .send()
74 .await?
75 .error_for_status()?;
76 let body = response.bytes().await?;
77 let mut advertisement = &body[..];
78 if advertisement
79 .get(4..)
80 .is_some_and(|rest| rest.starts_with(b"# service="))
81 {
82 let length = usize::from_str_radix(std::str::from_utf8(&body[..4])?, 16)?;
83 ensure!(body.len() >= length + 4, "truncated ref advertisement");
84 advertisement = &body[length + 4..];
85 }
86 writer.write_all(advertisement).await?;
87 writer.flush().await?;
88 Ok(())
89 }
90
91 /// Posts one request and streams the response back to the client.
92 async fn rpc<W: AsyncWrite + Unpin>(&self, body: Vec<u8>, writer: &mut W) -> Result<()> {
93 let service = self.service.as_str();
94 let response = self
95 .request(reqwest::Method::POST, service)
96 .header("content-type", format!("application/x-{service}-request"))
97 .header("accept", format!("application/x-{service}-result"))
98 .body(body)
99 .send()
100 .await?
101 .error_for_status()?;
102 let mut stream = response.bytes_stream();
103 while let Some(chunk) = stream.next().await {
104 writer.write_all(&chunk?).await?;
105 }
106 writer.flush().await?;
107 Ok(())
108 }
109
110 /// Protocol v2 is a series of self-contained commands, so each one maps
111 /// onto a single HTTP request.
112 async fn upload_pack<R, W>(&self, reader: &mut R, writer: &mut W) -> Result<()>
113 where
114 R: AsyncBufRead + Unpin,
115 W: AsyncWrite + Unpin,
116 {
117 if !self.protocol_v2 {
118 writer
119 .write_all(&error_pkt(
120 "g1t needs git protocol version 2 over SSH (git 2.26 or newer)",
121 ))
122 .await?;
123 return Ok(());
124 }
125 self.advertise(writer).await?;
126 loop {
127 let mut command = Vec::new();
128 if reader.fill_buf().await?.is_empty() {
129 return Ok(());
130 }
131 read_section(reader, &mut command).await?;
132 if command == FLUSH {
133 return Ok(());
134 }
135 self.rpc(command, writer).await?;
136 }
137 }
138
139 async fn receive_pack<R, W>(&self, reader: &mut R, writer: &mut W) -> Result<()>
140 where
141 R: AsyncBufRead + Unpin,
142 W: AsyncWrite + Unpin,
143 {
144 self.advertise(writer).await?;
145 if reader.fill_buf().await?.is_empty() {
146 return Ok(());
147 }
148 let mut body = Vec::new();
149 let lines = read_section(reader, &mut body).await?;
150 let commands: Vec<&str> = lines
151 .iter()
152 .filter(|line| !line.starts_with("shallow "))
153 .map(String::as_str)
154 .collect();
155 // A bare flush means the client had nothing to push.
156 let Some(first) = commands.first() else {
157 return Ok(());
158 };
159 let capabilities = first.split_once('\0').map_or("", |(_, caps)| caps);
160 let has_push_options = capabilities.split(' ').any(|cap| cap == "push-options");
161 // Commands are "<old id> <new id> <ref>"; only deletions send no pack.
162 let sends_pack = commands
163 .iter()
164 .any(|command| command.split(' ').nth(1) != Some(ZERO_ID));
165
166 if has_push_options {
167 read_section(reader, &mut body).await?;
168 }
169 if sends_pack {
170 read_pack(reader, &mut body).await?;
171 }
172 self.rpc(body, writer).await
173 }
174}
175
176/// Reads pkt-lines up to and including the next flush, appending the raw
177/// bytes to `raw`. Returns the payload of each line.
178async fn read_section<R: AsyncBufRead + Unpin>(
179 reader: &mut R,
180 raw: &mut Vec<u8>,
181) -> Result<Vec<String>> {
182 let mut lines = Vec::new();
183 loop {
184 let mut header = [0u8; 4];
185 reader.read_exact(&mut header).await?;
186 raw.extend_from_slice(&header);
187 let length = usize::from_str_radix(std::str::from_utf8(&header)?, 16)?;
188 match length {
189 0 => return Ok(lines),
190 // Delimiter and response-end packets carry no payload.
191 1 | 2 => continue,
192 3 => bail!("invalid pkt-line length"),
193 _ => {
194 let mut payload = vec![0u8; length - 4];
195 reader.read_exact(&mut payload).await?;
196 raw.extend_from_slice(&payload);
197 lines.push(String::from_utf8_lossy(&payload).trim_end().to_owned());
198 }
199 }
200 }
201}
202
203async fn read_into<R: AsyncBufRead + Unpin>(
204 reader: &mut R,
205 raw: &mut Vec<u8>,
206 count: usize,
207) -> Result<()> {
208 let start = raw.len();
209 raw.resize(start + count, 0);
210 reader.read_exact(&mut raw[start..]).await?;
211 Ok(())
212}
213
214async fn read_byte<R: AsyncBufRead + Unpin>(reader: &mut R, raw: &mut Vec<u8>) -> Result<u8> {
215 let byte = reader.read_u8().await?;
216 raw.push(byte);
217 Ok(byte)
218}
219
220/// Reads exactly one packfile. Over SSH nothing marks where the pack ends
221/// (the client keeps the channel open for the reply), so the only way to
222/// find the end is to walk every object.
223async fn read_pack<R: AsyncBufRead + Unpin>(reader: &mut R, raw: &mut Vec<u8>) -> Result<()> {
224 const OFS_DELTA: u8 = 6;
225 const REF_DELTA: u8 = 7;
226
227 let start = raw.len();
228 read_into(reader, raw, 12).await?;
229 ensure!(&raw[start..start + 4] == b"PACK", "expected a packfile");
230 let objects = u32::from_be_bytes(raw[start + 8..start + 12].try_into()?);
231
232 let mut scratch = vec![0u8; 64 * 1024];
233 for _ in 0..objects {
234 // Type and size header: a varint whose first byte holds the type.
235 let mut byte = read_byte(reader, raw).await?;
236 let kind = (byte >> 4) & 7;
237 while byte & 0x80 != 0 {
238 byte = read_byte(reader, raw).await?;
239 }
240 match kind {
241 OFS_DELTA => while read_byte(reader, raw).await? & 0x80 != 0 {},
242 REF_DELTA => read_into(reader, raw, 20).await?,
243 _ => {}
244 }
245
246 let mut inflate = Decompress::new(true);
247 loop {
248 let input = reader.fill_buf().await?;
249 ensure!(!input.is_empty(), "packfile ended mid-object");
250 let before = inflate.total_in();
251 let status = inflate.decompress(input, &mut scratch, FlushDecompress::None)?;
252 let consumed = (inflate.total_in() - before) as usize;
253 raw.extend_from_slice(&input[..consumed]);
254 reader.consume(consumed);
255 if status == Status::StreamEnd {
256 break;
257 }
258 }
259 }
260 // Trailing SHA-1 checksum.
261 read_into(reader, raw, 20).await
262}