Sign in

flagon-io/g1t

public

g1t/crates/sshd/src/git.rs

258 lines8,929 bytesCodeBlame

Pick any line to see why it is the way it is: the commit, the pull request and issue it came from, and what the agent was thinking.

Initial g1t: services, event bus, intents and attempts1//! 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(reqwest::Method::GET, &format!("info/refs?service={service}"))
70 .send()
71 .await?
72 .error_for_status()?;
73 let body = response.bytes().await?;
74 let mut advertisement = &body[..];
75 if advertisement.get(4..).is_some_and(|rest| rest.starts_with(b"# service=")) {
76 let length = usize::from_str_radix(std::str::from_utf8(&body[..4])?, 16)?;
77 ensure!(body.len() >= length + 4, "truncated ref advertisement");
78 advertisement = &body[length + 4..];
79 }
80 writer.write_all(advertisement).await?;
81 writer.flush().await?;
82 Ok(())
83 }
84
85 /// Posts one request and streams the response back to the client.
86 async fn rpc<W: AsyncWrite + Unpin>(&self, body: Vec<u8>, writer: &mut W) -> Result<()> {
87 let service = self.service.as_str();
88 let response = self
89 .request(reqwest::Method::POST, service)
90 .header("content-type", format!("application/x-{service}-request"))
91 .header("accept", format!("application/x-{service}-result"))
92 .body(body)
93 .send()
94 .await?
95 .error_for_status()?;
96 let mut stream = response.bytes_stream();
97 while let Some(chunk) = stream.next().await {
98 writer.write_all(&chunk?).await?;
99 }
100 writer.flush().await?;
101 Ok(())
102 }
103
104 /// Protocol v2 is a series of self-contained commands, so each one maps
105 /// onto a single HTTP request.
106 async fn upload_pack<R, W>(&self, reader: &mut R, writer: &mut W) -> Result<()>
107 where
108 R: AsyncBufRead + Unpin,
109 W: AsyncWrite + Unpin,
110 {
111 if !self.protocol_v2 {
112 writer
113 .write_all(&error_pkt(
114 "g1t needs git protocol version 2 over SSH (git 2.26 or newer)",
115 ))
116 .await?;
117 return Ok(());
118 }
119 self.advertise(writer).await?;
120 loop {
121 let mut command = Vec::new();
122 if reader.fill_buf().await?.is_empty() {
123 return Ok(());
124 }
125 read_section(reader, &mut command).await?;
126 if command == FLUSH {
127 return Ok(());
128 }
129 self.rpc(command, writer).await?;
130 }
131 }
132
133 async fn receive_pack<R, W>(&self, reader: &mut R, writer: &mut W) -> Result<()>
134 where
135 R: AsyncBufRead + Unpin,
136 W: AsyncWrite + Unpin,
137 {
138 self.advertise(writer).await?;
139 if reader.fill_buf().await?.is_empty() {
140 return Ok(());
141 }
142 let mut body = Vec::new();
143 let lines = read_section(reader, &mut body).await?;
144 let commands: Vec<&str> = lines
145 .iter()
146 .filter(|line| !line.starts_with("shallow "))
147 .map(String::as_str)
148 .collect();
149 // A bare flush means the client had nothing to push.
150 let Some(first) = commands.first() else {
151 return Ok(());
152 };
153 let capabilities = first.split_once('\0').map_or("", |(_, caps)| caps);
154 let has_push_options = capabilities.split(' ').any(|cap| cap == "push-options");
155 // Commands are "<old id> <new id> <ref>"; only deletions send no pack.
156 let sends_pack = commands
157 .iter()
158 .any(|command| command.split(' ').nth(1) != Some(ZERO_ID));
159
160 if has_push_options {
161 read_section(reader, &mut body).await?;
162 }
163 if sends_pack {
164 read_pack(reader, &mut body).await?;
165 }
166 self.rpc(body, writer).await
167 }
168}
169
170/// Reads pkt-lines up to and including the next flush, appending the raw
171/// bytes to `raw`. Returns the payload of each line.
172async fn read_section<R: AsyncBufRead + Unpin>(
173 reader: &mut R,
174 raw: &mut Vec<u8>,
175) -> Result<Vec<String>> {
176 let mut lines = Vec::new();
177 loop {
178 let mut header = [0u8; 4];
179 reader.read_exact(&mut header).await?;
180 raw.extend_from_slice(&header);
181 let length = usize::from_str_radix(std::str::from_utf8(&header)?, 16)?;
182 match length {
183 0 => return Ok(lines),
184 // Delimiter and response-end packets carry no payload.
185 1 | 2 => continue,
186 3 => bail!("invalid pkt-line length"),
187 _ => {
188 let mut payload = vec![0u8; length - 4];
189 reader.read_exact(&mut payload).await?;
190 raw.extend_from_slice(&payload);
191 lines.push(String::from_utf8_lossy(&payload).trim_end().to_owned());
192 }
193 }
194 }
195}
196
197async fn read_into<R: AsyncBufRead + Unpin>(
198 reader: &mut R,
199 raw: &mut Vec<u8>,
200 count: usize,
201) -> Result<()> {
202 let start = raw.len();
203 raw.resize(start + count, 0);
204 reader.read_exact(&mut raw[start..]).await?;
205 Ok(())
206}
207
208async fn read_byte<R: AsyncBufRead + Unpin>(reader: &mut R, raw: &mut Vec<u8>) -> Result<u8> {
209 let byte = reader.read_u8().await?;
210 raw.push(byte);
211 Ok(byte)
212}
213
214/// Reads exactly one packfile. Over SSH nothing marks where the pack ends
215/// (the client keeps the channel open for the reply), so the only way to
216/// find the end is to walk every object.
217async fn read_pack<R: AsyncBufRead + Unpin>(reader: &mut R, raw: &mut Vec<u8>) -> Result<()> {
218 const OFS_DELTA: u8 = 6;
219 const REF_DELTA: u8 = 7;
220
221 let start = raw.len();
222 read_into(reader, raw, 12).await?;
223 ensure!(&raw[start..start + 4] == b"PACK", "expected a packfile");
224 let objects = u32::from_be_bytes(raw[start + 8..start + 12].try_into()?);
225
226 let mut scratch = vec![0u8; 64 * 1024];
227 for _ in 0..objects {
228 // Type and size header: a varint whose first byte holds the type.
229 let mut byte = read_byte(reader, raw).await?;
230 let kind = (byte >> 4) & 7;
231 while byte & 0x80 != 0 {
232 byte = read_byte(reader, raw).await?;
233 }
234 match kind {
235 OFS_DELTA => {
236 while read_byte(reader, raw).await? & 0x80 != 0 {}
237 }
238 REF_DELTA => read_into(reader, raw, 20).await?,
239 _ => {}
240 }
241
242 let mut inflate = Decompress::new(true);
243 loop {
244 let input = reader.fill_buf().await?;
245 ensure!(!input.is_empty(), "packfile ended mid-object");
246 let before = inflate.total_in();
247 let status = inflate.decompress(input, &mut scratch, FlushDecompress::None)?;
248 let consumed = (inflate.total_in() - before) as usize;
249 raw.extend_from_slice(&input[..consumed]);
250 reader.consume(consumed);
251 if status == Status::StreamEnd {
252 break;
253 }
254 }
255 }
256 // Trailing SHA-1 checksum.
257 read_into(reader, raw, 20).await
258}