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/services/repos/src/mirror.rs

511 lines21,456 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.

Invite-only launch: sign in with GitHub, repository access and lifecycle, many emails, a new look1//! Copying every branch and tag from one git server to another: importing a
2//! repository with a credential, keeping a mirror in step with the host it
3//! mirrors, and pushing a repository's refs out to a host it is mirrored to.
4//!
5//! Like landing (see `land.rs`), this speaks git's smart HTTP protocol and
6//! relays the pack it receives unchanged. Both sides are asked for their
7//! refs; the source is asked for one pack holding what the target lacks;
8//! the target is sent one push that moves every ref that differs.
9
10use std::collections::{BTreeMap, HashSet};
11
12use futures_util::StreamExt;
13use worker::js_sys::Uint8Array;
14use worker::{Fetch, Headers, Method, Request, RequestInit, Result};
15
16use g1t_contracts::repos::{MirrorArgs, MirrorDirection, Mirrored};
17use g1t_contracts::{FailureCode, Outcome};
18
19use crate::land::{read_pkt_lines, unpack_sideband};
20use crate::registry::store_key;
21use crate::store::{GitRepo, GitStore, Scope};
22use crate::{Repos, import, not_found};
23
24const ZERO_ID: &str = "0000000000000000000000000000000000000000";
25/// The most a pack may hold. A Worker holds it in memory while relaying it.
26pub const MAX_PACK_BYTES: usize = 40 * 1024 * 1024;
27/// How many of the target's commits are named to the source as already
28/// had, so a sync only carries what is new.
29const MAX_HAVES: usize = 256;
30const USER_AGENT: &str = "git/2.45.0 (g1t mirror)";
31/// A pack with no objects: for a push whose refs all name objects the
32/// target already holds. The last 20 bytes are the SHA-1 of the first 12.
33const EMPTY_PACK: [u8; 32] = [
34 b'P', b'A', b'C', b'K', 0, 0, 0, 2, 0, 0, 0, 0, 0x02, 0x9d, 0x08, 0x82, 0x3b, 0xd8, 0xa8, 0xea,
35 0xb5, 0x10, 0xad, 0x6a, 0xc7, 0x5c, 0x82, 0x3c, 0xfd, 0x3e, 0xd3, 0x1e,
36];
37
38/// One side of a copy: a repository's smart HTTP address and the
39/// `authorization` header that opens it. The header is never logged.
40pub struct Endpoint {
41 pub url: String,
42 pub authorization: String,
43}
44
45impl Endpoint {
46 /// A g1t repository, with a credential from the git store.
47 pub fn bearer(url: &str, token: &str) -> Self {
48 Endpoint {
49 url: url.trim_end_matches('/').to_owned(),
50 authorization: format!("Bearer {token}"),
51 }
52 }
53
54 /// A GitHub repository, with an installation access token. GitHub takes
55 /// one as the password of the user `x-access-token`. The token is used
56 /// as given: its length and shape are GitHub's to change.
57 pub fn github(url: &str, token: &str) -> Self {
58 Endpoint {
59 url: url.trim_end_matches('/').to_owned(),
60 authorization: format!("Basic {}", base64(&format!("x-access-token:{token}"))),
61 }
62 }
63}
64
65/// What a copy changed.
66#[derive(Debug, Default, PartialEq, Eq)]
67pub struct Copied {
68 /// Refs created or moved, each with where it was and where it is now.
69 pub updated: Vec<(String, Option<String>, String)>,
70 pub deleted: Vec<String>,
71 /// The branch the source's HEAD names, when it said.
72 pub head: Option<String>,
73}
74
75/// Which refs a copy moves.
76#[derive(Clone, Copy, Debug, PartialEq, Eq)]
77pub enum Prune {
78 /// Refs the source no longer has are deleted from the target: the
79 /// target is a mirror of the source.
80 Yes,
81 /// Refs only the target has are left alone: pushing out to a host that
82 /// may have branches of its own.
83 No,
84}
85
86fn base64(text: &str) -> String {
87 const ALPHABET: &[u8; 64] = b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/";
88 let bytes = text.as_bytes();
89 let mut out = String::with_capacity(bytes.len().div_ceil(3) * 4);
90 for chunk in bytes.chunks(3) {
91 let n = (chunk[0] as u32) << 16
92 | (*chunk.get(1).unwrap_or(&0) as u32) << 8
93 | *chunk.get(2).unwrap_or(&0) as u32;
94 for (i, shift) in [18, 12, 6, 0].into_iter().enumerate() {
95 if i <= chunk.len() {
96 out.push(ALPHABET[(n >> shift) as usize & 63] as char);
97 } else {
98 out.push('=');
99 }
100 }
101 }
102 out
103}
104
105fn pkt_line(payload: &str) -> Vec<u8> {
106 format!("{:04x}{payload}", payload.len() + 4).into_bytes()
107}
108
109/// A ref advertisement: branch and tag refs by name, and the branch HEAD
110/// points to when the server said.
111#[derive(Debug, Default, PartialEq, Eq)]
112pub struct Advertised {
113 pub refs: BTreeMap<String, String>,
114 pub head: Option<String>,
115}
116
117impl Advertised {
118 /// The branch to make a copy's default: the one HEAD names, else
119 /// `main`, else the first branch. `None` for a repository with none.
120 pub fn default_branch(&self) -> Option<String> {
121 let branches: Vec<&str> = self.refs.keys().filter_map(|name| name.strip_prefix("refs/heads/")).collect();
122 self.head
123 .clone()
124 .filter(|head| branches.contains(&head.as_str()))
125 .or_else(|| branches.iter().find(|name| **name == "main").map(|name| name.to_string()))
126 .or_else(|| branches.first().map(|name| name.to_string()))
127 }
128}
129
130/// Asks a source for its refs before anything is made from it, so an
131/// address or credential that does not work leaves nothing behind.
132pub async fn probe(source: &Endpoint) -> Result<std::result::Result<Advertised, String>> {
133 advertise(source, "git-upload-pack").await
134}
135
136/// Reads `info/refs`. Peeled tags (`^{}`) and the placeholder an empty
137/// repository sends are skipped; only branches and tags are kept.
138pub fn parse_advertisement(bytes: &[u8]) -> Advertised {
139 let (lines, _) = read_pkt_lines(bytes);
140 let mut advertised = Advertised::default();
141 for line in lines {
142 let mut parts = line.splitn(2, |byte| *byte == 0);
143 let Some(reference) = parts.next().and_then(|bytes| std::str::from_utf8(bytes).ok()) else {
144 continue;
145 };
146 if let Some(capabilities) = parts.next().and_then(|bytes| std::str::from_utf8(bytes).ok()) {
147 advertised.head = capabilities
148 .split(' ')
149 .find_map(|capability| capability.trim().strip_prefix("symref=HEAD:refs/heads/"))
150 .map(str::to_owned)
151 .or(advertised.head);
152 }
153 let Some((hash, name)) = reference.trim_end().split_once(' ') else {
154 continue;
155 };
156 let kept = (name.starts_with("refs/heads/") || name.starts_with("refs/tags/")) && !name.ends_with("^{}");
157 if kept && hash.len() == 40 && hash != ZERO_ID {
158 advertised.refs.insert(name.to_owned(), hash.to_owned());
159 }
160 }
161 advertised
162}
163
164/// One ref command for a push: `(name, old, new)`, `new` the zero id to
165/// delete.
166pub type Command = (String, Option<String>, String);
167
168/// What has to change on `target` so its refs match `source`'s.
169pub fn plan(source: &Advertised, target: &Advertised, prune: Prune) -> Vec<Command> {
170 let mut commands = Vec::new();
171 for (name, hash) in &source.refs {
172 let old = target.refs.get(name);
173 if old != Some(hash) {
174 commands.push((name.clone(), old.cloned(), hash.clone()));
175 }
176 }
177 if prune == Prune::Yes {
178 for (name, hash) in &target.refs {
179 if !source.refs.contains_key(name) {
180 commands.push((name.clone(), Some(hash.clone()), ZERO_ID.to_owned()));
181 }
182 }
183 }
184 commands
185}
186
187/// The body of an upload-pack request for `wants`, naming `haves`.
188pub fn upload_request(wants: &[String], haves: &[String]) -> Vec<u8> {
189 let mut body = Vec::new();
190 for (index, want) in wants.iter().enumerate() {
191 // Capabilities ride on the first want only.
192 let line = if index == 0 { format!("want {want} side-band-64k\n") } else { format!("want {want}\n") };
193 body.extend(pkt_line(&line));
194 }
195 body.extend_from_slice(b"0000");
196 for have in haves.iter().take(MAX_HAVES) {
197 body.extend(pkt_line(&format!("have {have}\n")));
198 }
199 body.extend(pkt_line("done\n"));
200 body
201}
202
203/// The body of a receive-pack request: the commands, then the pack when
204/// anything but deletions is sent.
205pub fn receive_request(commands: &[Command], pack: Option<Vec<u8>>) -> Vec<u8> {
206 let deletes = commands.iter().any(|(_, _, new)| new == ZERO_ID);
207 let capabilities = if deletes { "report-status delete-refs" } else { "report-status" };
208 let mut body = Vec::new();
209 for (index, (name, old, new)) in commands.iter().enumerate() {
210 let old = old.as_deref().unwrap_or(ZERO_ID);
211 let line = if index == 0 { format!("{old} {new} {name}\0 {capabilities}\n") } else { format!("{old} {new} {name}\n") };
212 body.extend(pkt_line(&line));
213 }
214 body.extend_from_slice(b"0000");
215 if commands.iter().any(|(_, _, new)| new != ZERO_ID) {
216 body.extend(pack.unwrap_or_else(|| EMPTY_PACK.to_vec()));
217 }
218 body
219}
220
221/// Which commands a receive-pack report says were refused, with why.
222pub fn refused(report: &[u8], commands: &[Command]) -> Vec<String> {
223 let (lines, _) = read_pkt_lines(report);
224 let lines: Vec<String> = lines
225 .into_iter()
226 .map(|line| {
227 // A report may come inside side-band channel 1.
228 let line = if line.first() == Some(&1) { &line[1..] } else { line };
229 String::from_utf8_lossy(line).trim_end().to_owned()
230 })
231 .collect();
232 let mut problems: Vec<String> = lines
233 .iter()
234 .filter(|line| line.starts_with("unpack ") && *line != "unpack ok")
235 .cloned()
236 .collect();
237 for (name, _, _) in commands {
238 if !lines.iter().any(|line| *line == format!("ok {name}")) {
239 let reason = lines
240 .iter()
241 .find_map(|line| line.strip_prefix(&format!("ng {name} ")))
242 .unwrap_or("not reported");
243 problems.push(format!("{name}: {reason}"));
244 }
245 }
246 problems
247}
248
249fn request(method: Method, url: &str, endpoint: &Endpoint, body: Option<(&str, Vec<u8>)>) -> Result<Request> {
250 let headers = Headers::new();
251 headers.set("user-agent", USER_AGENT)?;
252 headers.set("authorization", &endpoint.authorization)?;
253 let mut init = RequestInit::new();
254 if let Some((service, body)) = body {
255 headers.set("content-type", &format!("application/x-{service}-request"))?;
256 headers.set("accept", &format!("application/x-{service}-result"))?;
257 init.with_body(Some(Uint8Array::from(body.as_slice()).into()));
258 }
259 init.with_method(method).with_headers(headers);
260 Request::new_with_init(url, &init)
261}
262
263/// Asks a server for its refs, as the given service would see them.
264async fn advertise(endpoint: &Endpoint, service: &str) -> Result<std::result::Result<Advertised, String>> {
265 let url = format!("{}/info/refs?service={service}", endpoint.url);
266 let mut response = match Fetch::Request(request(Method::Get, &url, endpoint, None)?).send().await {
267 Ok(response) => response,
268 Err(_) => return Ok(Err("The repository could not be reached.".to_owned())),
269 };
270 match response.status_code() {
271 200 => Ok(Ok(parse_advertisement(&response.bytes().await?))),
272 401 | 403 => Ok(Err("The repository refused g1t's credential.".to_owned())),
273 404 => Ok(Err("The repository was not found, or g1t may not read it.".to_owned())),
274 status => Ok(Err(format!("The repository answered {status}."))),
275 }
276}
277
278/// Fetches one pack holding `wants`, reading in pieces so that one too
279/// large is noticed before it is all held.
280async fn fetch_pack(source: &Endpoint, wants: &[String], haves: &[String]) -> Result<std::result::Result<Vec<u8>, String>> {
281 let body = upload_request(wants, haves);
282 let url = format!("{}/git-upload-pack", source.url);
283 let mut response = Fetch::Request(request(Method::Post, &url, source, Some(("git-upload-pack", body)))?)
284 .send()
285 .await?;
286 if response.status_code() != 200 {
287 return Ok(Err(format!("The source refused to send the repository ({}).", response.status_code())));
288 }
289 let mut received = Vec::new();
290 let mut stream = response.stream()?;
291 while let Some(chunk) = stream.next().await {
292 received.extend_from_slice(&chunk?);
293 if received.len() > MAX_PACK_BYTES {
294 return Ok(Err(format!(
295 "The repository is larger than {} MB, the most g1t copies at once. Push it with git instead.",
296 MAX_PACK_BYTES / 1024 / 1024
297 )));
298 }
299 }
300 Ok(unpack_sideband(&received).map_err(|_| "The source did not send a usable pack.".to_owned()))
301}
302
303/// Makes `target`'s branches and tags match `source`'s. With
304/// `Prune::No`, refs only the target has are kept. `Err` in the inner
305/// result is a reason to show a person.
306pub async fn copy(source: &Endpoint, target: &Endpoint, prune: Prune) -> Result<std::result::Result<Copied, String>> {
307 let theirs = match advertise(source, "git-upload-pack").await? {
308 Ok(refs) => refs,
309 Err(reason) => return Ok(Err(reason)),
310 };
311 let ours = match advertise(target, "git-receive-pack").await? {
312 Ok(refs) => refs,
313 Err(reason) => return Ok(Err(format!("Writing the copy failed: {reason}"))),
314 };
315 let commands = plan(&theirs, &ours, prune);
316 let mut copied = Copied {
317 head: theirs.head.clone(),
318 ..Copied::default()
319 };
320 if commands.is_empty() {
321 return Ok(Ok(copied));
322 }
323 let held: HashSet<&String> = ours.refs.values().collect();
324 let mut wants: Vec<String> = commands
325 .iter()
326 .map(|(_, _, new)| new)
327 .filter(|new| *new != ZERO_ID && !held.contains(new))
328 .cloned()
329 .collect();
330 wants.sort();
331 wants.dedup();
332 let pack = if wants.is_empty() {
333 None
334 } else {
335 let haves: Vec<String> = ours.refs.values().cloned().collect::<HashSet<_>>().into_iter().collect();
336 match fetch_pack(source, &wants, &haves).await? {
337 Ok(pack) => Some(pack),
338 Err(reason) => return Ok(Err(reason)),
339 }
340 };
341 let body = receive_request(&commands, pack);
342 let url = format!("{}/git-receive-pack", target.url);
343 let mut response = Fetch::Request(request(Method::Post, &url, target, Some(("git-receive-pack", body)))?)
344 .send()
345 .await?;
346 let report = response.bytes().await?;
347 if response.status_code() != 200 {
348 return Ok(Err(format!("The push was refused ({}).", response.status_code())));
349 }
350 let problems = refused(&report, &commands);
351 if !problems.is_empty() {
352 return Ok(Err(format!("Some refs were not updated: {}", problems.join("; "))));
353 }
354 for (name, old, new) in commands {
355 if new == ZERO_ID {
356 copied.deleted.push(name);
357 } else {
358 copied.updated.push((name, old, new));
359 }
360 }
361 Ok(Ok(copied))
362}
363
364impl<S: GitStore> Repos<S> {
365 /// `mirror`: a mirror catching up with the host it mirrors, or a
366 /// repository pushing its refs out to one. Each branch moved on g1t is
367 /// announced as a push, so deployments and checks follow it.
368 pub(crate) async fn mirror(&self, a: MirrorArgs) -> Result<Outcome<Mirrored>> {
369 let Some(repo) = self.registry.by_id(&a.repo_id).await? else {
370 return Ok(not_found());
371 };
372 let Some(url) = import::clean_url(&a.url) else {
373 return Ok(Outcome::fail(FailureCode::Invalid, "That is not an https repository address."));
374 };
375 let scope = match a.direction {
376 MirrorDirection::Pull => Scope::Write,
377 MirrorDirection::Push => Scope::Read,
378 };
379 let access = self.store.open(&store_key(&repo)).await?.access(scope).await?;
380 let ours = Endpoint::bearer(&access.remote, &access.token);
381 let theirs = Endpoint::github(&url, &a.token);
382 let copied = match a.direction {
383 MirrorDirection::Pull => copy(&theirs, &ours, Prune::Yes).await?,
384 MirrorDirection::Push => copy(&ours, &theirs, Prune::No).await?,
385 };
386 let copied = match copied {
387 Ok(copied) => copied,
388 Err(reason) => return Ok(Outcome::fail(FailureCode::Conflict, reason)),
389 };
390 if a.direction == MirrorDirection::Pull {
391 for (name, old, new) in &copied.updated {
392 if name.starts_with("refs/heads/") {
393 self.publish_push(&repo, name, old.as_deref(), new, None).await?;
394 }
395 }
396 }
397 Ok(Outcome::Ok(Mirrored {
398 updated: copied.updated.into_iter().map(|(name, _, _)| name).collect(),
399 deleted: copied.deleted,
400 }))
401 }
402}
403
404#[cfg(test)]
405mod tests {
406 use super::*;
407
408 const A: &str = "c71546fcd893ef8b0f57388b65e620d759705dda";
409 const B: &str = "4807077b296e6edbf410d55e72749d3e1170c291";
410 const C: &str = "1111111111111111111111111111111111111111";
411
412 fn advertised(refs: &[(&str, &str)]) -> Advertised {
413 Advertised {
414 refs: refs.iter().map(|(name, hash)| (name.to_string(), hash.to_string())).collect(),
415 head: None,
416 }
417 }
418
419 #[test]
420 fn the_basic_header_carries_the_token_whole() {
421 assert_eq!(base64("x-access-token:abc"), "eC1hY2Nlc3MtdG9rZW46YWJj");
422 assert_eq!(base64("a"), "YQ==");
423 assert_eq!(base64("ab"), "YWI=");
424 // GitHub's stateless installation tokens run to hundreds of
425 // characters; nothing here assumes a length.
426 let token = format!("ghs_{}", "x".repeat(516));
427 assert_eq!(token.len(), 520);
428 let endpoint = Endpoint::github("https://github.com/o/r.git/", &token);
429 assert_eq!(endpoint.url, "https://github.com/o/r.git");
430 let encoded = endpoint.authorization.strip_prefix("Basic ").unwrap();
431 assert_eq!(encoded, base64(&format!("x-access-token:{token}")));
432 assert_eq!(encoded.len(), (15 + 520usize).div_ceil(3) * 4);
433 }
434
435 #[test]
436 fn the_advertisement_keeps_branches_and_tags() {
437 let bytes = [
438 pkt_line("# service=git-upload-pack\n"),
439 b"0000".to_vec(),
440 pkt_line(&format!("{A} HEAD\0multi_ack side-band-64k symref=HEAD:refs/heads/trunk agent=git/x\n")),
441 pkt_line(&format!("{A} refs/heads/trunk\n")),
442 pkt_line(&format!("{B} refs/tags/v1\n")),
443 pkt_line(&format!("{A} refs/tags/v1^{{}}\n")),
444 pkt_line(&format!("{C} refs/pull/1/head\n")),
445 b"0000".to_vec(),
446 ]
447 .concat();
448 let parsed = parse_advertisement(&bytes);
449 assert_eq!(parsed.head.as_deref(), Some("trunk"));
450 assert_eq!(parsed, Advertised {
451 refs: advertised(&[("refs/heads/trunk", A), ("refs/tags/v1", B)]).refs,
452 head: Some("trunk".to_owned()),
453 });
454 assert_eq!(parsed.default_branch().as_deref(), Some("trunk"));
455 assert_eq!(advertised(&[("refs/heads/a", A), ("refs/heads/main", A)]).default_branch().as_deref(), Some("main"));
456 assert_eq!(advertised(&[("refs/tags/v1", A)]).default_branch(), None);
457 let empty = [pkt_line(&format!("{ZERO_ID} capabilities^{{}}\0report-status\n")), b"0000".to_vec()].concat();
458 assert!(parse_advertisement(&empty).refs.is_empty());
459 }
460
461 #[test]
462 fn a_mirror_prunes_and_a_push_out_does_not() {
463 let source = advertised(&[("refs/heads/main", A), ("refs/tags/v1", B)]);
464 let target = advertised(&[("refs/heads/main", B), ("refs/heads/old", C)]);
465 assert_eq!(plan(&source, &target, Prune::Yes), vec![
466 ("refs/heads/main".to_owned(), Some(B.to_owned()), A.to_owned()),
467 ("refs/tags/v1".to_owned(), None, B.to_owned()),
468 ("refs/heads/old".to_owned(), Some(C.to_owned()), ZERO_ID.to_owned()),
469 ]);
470 assert_eq!(plan(&source, &target, Prune::No).len(), 2);
471 assert!(plan(&source, &source, Prune::Yes).is_empty());
472 }
473
474 #[test]
475 fn a_push_of_known_objects_sends_an_empty_pack() {
476 let commands = vec![("refs/tags/v1".to_owned(), None, A.to_owned())];
477 let body = receive_request(&commands, None);
478 assert!(body.ends_with(&EMPTY_PACK));
479 let deletes = vec![("refs/heads/x".to_owned(), Some(A.to_owned()), ZERO_ID.to_owned())];
480 let body = receive_request(&deletes, None);
481 assert!(body.ends_with(b"0000"));
482 assert!(String::from_utf8_lossy(&body).contains("delete-refs"));
483 }
484
485 #[test]
486 fn wants_carry_capabilities_once() {
487 let body = String::from_utf8(upload_request(&[A.to_owned(), B.to_owned()], &[C.to_owned()])).unwrap();
488 assert_eq!(body.matches("side-band-64k").count(), 1);
489 assert!(body.contains(&format!("want {B}\n")));
490 assert!(body.contains(&format!("have {C}\n")));
491 assert!(body.ends_with("0009done\n"));
492 }
493
494 #[test]
495 fn refusals_are_reported_by_ref() {
496 let commands = vec![
497 ("refs/heads/main".to_owned(), None, A.to_owned()),
498 ("refs/heads/x".to_owned(), None, B.to_owned()),
499 ];
500 let report = [
501 pkt_line("unpack ok\n"),
502 pkt_line("ok refs/heads/main\n"),
503 pkt_line("ng refs/heads/x protected branch\n"),
504 b"0000".to_vec(),
505 ]
506 .concat();
507 assert_eq!(refused(&report, &commands), vec!["refs/heads/x: protected branch".to_owned()]);
508 let fine = [pkt_line("unpack ok\n"), pkt_line("ok refs/heads/main\n"), pkt_line("ok refs/heads/x\n")].concat();
509 assert!(refused(&fine, &commands).is_empty());
510 }
511}