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

525 lines22,098 bytesCodeBlame
1//! 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 // Requests to g1t's own store are metered (meters.rs); GitHub's are not.
251 if endpoint.authorization.starts_with("Bearer ") {
252 let sent = body.as_ref().map_or(0, |(_, body)| body.len() as u64);
253 let meter = match body.as_ref().map(|(service, _)| *service) {
254 Some("git-receive-pack") => "internal.git.receive_pack",
255 Some(_) => "internal.git.fetch",
256 None => "internal.git.info_refs",
257 };
258 crate::meters::record_remote(meter, &endpoint.url, sent, 0);
259 }
260 let headers = Headers::new();
261 headers.set("user-agent", USER_AGENT)?;
262 headers.set("authorization", &endpoint.authorization)?;
263 let mut init = RequestInit::new();
264 if let Some((service, body)) = body {
265 headers.set("content-type", &format!("application/x-{service}-request"))?;
266 headers.set("accept", &format!("application/x-{service}-result"))?;
267 init.with_body(Some(Uint8Array::from(body.as_slice()).into()));
268 }
269 init.with_method(method).with_headers(headers);
270 Request::new_with_init(url, &init)
271}
272
273/// Asks a server for its refs, as the given service would see them.
274async fn advertise(endpoint: &Endpoint, service: &str) -> Result<std::result::Result<Advertised, String>> {
275 let url = format!("{}/info/refs?service={service}", endpoint.url);
276 let mut response = match Fetch::Request(request(Method::Get, &url, endpoint, None)?).send().await {
277 Ok(response) => response,
278 Err(_) => return Ok(Err("The repository could not be reached.".to_owned())),
279 };
280 match response.status_code() {
281 200 => Ok(Ok(parse_advertisement(&response.bytes().await?))),
282 401 | 403 => Ok(Err("The repository refused g1t's credential.".to_owned())),
283 404 => Ok(Err("The repository was not found, or g1t may not read it.".to_owned())),
284 status => Ok(Err(format!("The repository answered {status}."))),
285 }
286}
287
288/// Fetches one pack holding `wants`, reading in pieces so that one too
289/// large is noticed before it is all held.
290async fn fetch_pack(source: &Endpoint, wants: &[String], haves: &[String]) -> Result<std::result::Result<Vec<u8>, String>> {
291 let body = upload_request(wants, haves);
292 let url = format!("{}/git-upload-pack", source.url);
293 let mut response = Fetch::Request(request(Method::Post, &url, source, Some(("git-upload-pack", body)))?)
294 .send()
295 .await?;
296 if response.status_code() != 200 {
297 return Ok(Err(format!("The source refused to send the repository ({}).", response.status_code())));
298 }
299 let mut received = Vec::new();
300 let mut stream = response.stream()?;
301 while let Some(chunk) = stream.next().await {
302 received.extend_from_slice(&chunk?);
303 if received.len() > MAX_PACK_BYTES {
304 return Ok(Err(format!(
305 "The repository is larger than {} MB, the most g1t copies at once. Push it with git instead.",
306 MAX_PACK_BYTES / 1024 / 1024
307 )));
308 }
309 }
310 Ok(unpack_sideband(&received).map_err(|_| "The source did not send a usable pack.".to_owned()))
311}
312
313/// Makes `target`'s branches and tags match `source`'s. With
314/// `Prune::No`, refs only the target has are kept. `Err` in the inner
315/// result is a reason to show a person.
316pub async fn copy(source: &Endpoint, target: &Endpoint, prune: Prune) -> Result<std::result::Result<Copied, String>> {
317 let theirs = match advertise(source, "git-upload-pack").await? {
318 Ok(refs) => refs,
319 Err(reason) => return Ok(Err(reason)),
320 };
321 let ours = match advertise(target, "git-receive-pack").await? {
322 Ok(refs) => refs,
323 Err(reason) => return Ok(Err(format!("Writing the copy failed: {reason}"))),
324 };
325 let commands = plan(&theirs, &ours, prune);
326 let mut copied = Copied {
327 head: theirs.head.clone(),
328 ..Copied::default()
329 };
330 if commands.is_empty() {
331 return Ok(Ok(copied));
332 }
333 let held: HashSet<&String> = ours.refs.values().collect();
334 let mut wants: Vec<String> = commands
335 .iter()
336 .map(|(_, _, new)| new)
337 .filter(|new| *new != ZERO_ID && !held.contains(new))
338 .cloned()
339 .collect();
340 wants.sort();
341 wants.dedup();
342 let pack = if wants.is_empty() {
343 None
344 } else {
345 let haves: Vec<String> = ours.refs.values().cloned().collect::<HashSet<_>>().into_iter().collect();
346 match fetch_pack(source, &wants, &haves).await? {
347 Ok(pack) => Some(pack),
348 Err(reason) => return Ok(Err(reason)),
349 }
350 };
351 let body = receive_request(&commands, pack);
352 let url = format!("{}/git-receive-pack", target.url);
353 let mut response = Fetch::Request(request(Method::Post, &url, target, Some(("git-receive-pack", body)))?)
354 .send()
355 .await?;
356 let report = response.bytes().await?;
357 if response.status_code() != 200 {
358 return Ok(Err(format!("The push was refused ({}).", response.status_code())));
359 }
360 let problems = refused(&report, &commands);
361 if !problems.is_empty() {
362 return Ok(Err(format!("Some refs were not updated: {}", problems.join("; "))));
363 }
364 for (name, old, new) in commands {
365 if new == ZERO_ID {
366 copied.deleted.push(name);
367 } else {
368 copied.updated.push((name, old, new));
369 }
370 }
371 Ok(Ok(copied))
372}
373
374impl<S: GitStore> Repos<S> {
375 /// `mirror`: a mirror catching up with the host it mirrors, or a
376 /// repository pushing its refs out to one. Each branch moved on g1t is
377 /// announced as a push, so deployments and checks follow it.
378 pub(crate) async fn mirror(&self, a: MirrorArgs) -> Result<Outcome<Mirrored>> {
379 let Some(repo) = self.registry.by_id(&a.repo_id).await? else {
380 return Ok(not_found());
381 };
382 let Some(url) = import::clean_url(&a.url) else {
383 return Ok(Outcome::fail(FailureCode::Invalid, "That is not an https repository address."));
384 };
385 let scope = match a.direction {
386 MirrorDirection::Pull => Scope::Write,
387 MirrorDirection::Push => Scope::Read,
388 };
389 let access = self.store.open(&store_key(&repo)).await?.access(scope).await?;
390 let ours = Endpoint::bearer(&access.remote, &access.token);
391 let theirs = Endpoint::github(&url, &a.token);
392 let copied = match a.direction {
393 MirrorDirection::Pull => {
394 let copied = copy(&theirs, &ours, Prune::Yes).await?;
395 self.refs_moved(&repo.id).await;
396 copied
397 }
398 MirrorDirection::Push => copy(&ours, &theirs, Prune::No).await?,
399 };
400 let copied = match copied {
401 Ok(copied) => copied,
402 Err(reason) => return Ok(Outcome::fail(FailureCode::Conflict, reason)),
403 };
404 if a.direction == MirrorDirection::Pull {
405 for (name, old, new) in &copied.updated {
406 if name.starts_with("refs/heads/") {
407 self.publish_push(&repo, name, old.as_deref(), new, None).await?;
408 }
409 }
410 }
411 Ok(Outcome::Ok(Mirrored {
412 updated: copied.updated.into_iter().map(|(name, _, _)| name).collect(),
413 deleted: copied.deleted,
414 }))
415 }
416}
417
418#[cfg(test)]
419mod tests {
420 use super::*;
421
422 const A: &str = "c71546fcd893ef8b0f57388b65e620d759705dda";
423 const B: &str = "4807077b296e6edbf410d55e72749d3e1170c291";
424 const C: &str = "1111111111111111111111111111111111111111";
425
426 fn advertised(refs: &[(&str, &str)]) -> Advertised {
427 Advertised {
428 refs: refs.iter().map(|(name, hash)| (name.to_string(), hash.to_string())).collect(),
429 head: None,
430 }
431 }
432
433 #[test]
434 fn the_basic_header_carries_the_token_whole() {
435 assert_eq!(base64("x-access-token:abc"), "eC1hY2Nlc3MtdG9rZW46YWJj");
436 assert_eq!(base64("a"), "YQ==");
437 assert_eq!(base64("ab"), "YWI=");
438 // GitHub's stateless installation tokens run to hundreds of
439 // characters; nothing here assumes a length.
440 let token = format!("ghs_{}", "x".repeat(516));
441 assert_eq!(token.len(), 520);
442 let endpoint = Endpoint::github("https://github.com/o/r.git/", &token);
443 assert_eq!(endpoint.url, "https://github.com/o/r.git");
444 let encoded = endpoint.authorization.strip_prefix("Basic ").unwrap();
445 assert_eq!(encoded, base64(&format!("x-access-token:{token}")));
446 assert_eq!(encoded.len(), (15 + 520usize).div_ceil(3) * 4);
447 }
448
449 #[test]
450 fn the_advertisement_keeps_branches_and_tags() {
451 let bytes = [
452 pkt_line("# service=git-upload-pack\n"),
453 b"0000".to_vec(),
454 pkt_line(&format!("{A} HEAD\0multi_ack side-band-64k symref=HEAD:refs/heads/trunk agent=git/x\n")),
455 pkt_line(&format!("{A} refs/heads/trunk\n")),
456 pkt_line(&format!("{B} refs/tags/v1\n")),
457 pkt_line(&format!("{A} refs/tags/v1^{{}}\n")),
458 pkt_line(&format!("{C} refs/pull/1/head\n")),
459 b"0000".to_vec(),
460 ]
461 .concat();
462 let parsed = parse_advertisement(&bytes);
463 assert_eq!(parsed.head.as_deref(), Some("trunk"));
464 assert_eq!(parsed, Advertised {
465 refs: advertised(&[("refs/heads/trunk", A), ("refs/tags/v1", B)]).refs,
466 head: Some("trunk".to_owned()),
467 });
468 assert_eq!(parsed.default_branch().as_deref(), Some("trunk"));
469 assert_eq!(advertised(&[("refs/heads/a", A), ("refs/heads/main", A)]).default_branch().as_deref(), Some("main"));
470 assert_eq!(advertised(&[("refs/tags/v1", A)]).default_branch(), None);
471 let empty = [pkt_line(&format!("{ZERO_ID} capabilities^{{}}\0report-status\n")), b"0000".to_vec()].concat();
472 assert!(parse_advertisement(&empty).refs.is_empty());
473 }
474
475 #[test]
476 fn a_mirror_prunes_and_a_push_out_does_not() {
477 let source = advertised(&[("refs/heads/main", A), ("refs/tags/v1", B)]);
478 let target = advertised(&[("refs/heads/main", B), ("refs/heads/old", C)]);
479 assert_eq!(plan(&source, &target, Prune::Yes), vec![
480 ("refs/heads/main".to_owned(), Some(B.to_owned()), A.to_owned()),
481 ("refs/tags/v1".to_owned(), None, B.to_owned()),
482 ("refs/heads/old".to_owned(), Some(C.to_owned()), ZERO_ID.to_owned()),
483 ]);
484 assert_eq!(plan(&source, &target, Prune::No).len(), 2);
485 assert!(plan(&source, &source, Prune::Yes).is_empty());
486 }
487
488 #[test]
489 fn a_push_of_known_objects_sends_an_empty_pack() {
490 let commands = vec![("refs/tags/v1".to_owned(), None, A.to_owned())];
491 let body = receive_request(&commands, None);
492 assert!(body.ends_with(&EMPTY_PACK));
493 let deletes = vec![("refs/heads/x".to_owned(), Some(A.to_owned()), ZERO_ID.to_owned())];
494 let body = receive_request(&deletes, None);
495 assert!(body.ends_with(b"0000"));
496 assert!(String::from_utf8_lossy(&body).contains("delete-refs"));
497 }
498
499 #[test]
500 fn wants_carry_capabilities_once() {
501 let body = String::from_utf8(upload_request(&[A.to_owned(), B.to_owned()], &[C.to_owned()])).unwrap();
502 assert_eq!(body.matches("side-band-64k").count(), 1);
503 assert!(body.contains(&format!("want {B}\n")));
504 assert!(body.contains(&format!("have {C}\n")));
505 assert!(body.ends_with("0009done\n"));
506 }
507
508 #[test]
509 fn refusals_are_reported_by_ref() {
510 let commands = vec![
511 ("refs/heads/main".to_owned(), None, A.to_owned()),
512 ("refs/heads/x".to_owned(), None, B.to_owned()),
513 ];
514 let report = [
515 pkt_line("unpack ok\n"),
516 pkt_line("ok refs/heads/main\n"),
517 pkt_line("ng refs/heads/x protected branch\n"),
518 b"0000".to_vec(),
519 ]
520 .concat();
521 assert_eq!(refused(&report, &commands), vec!["refs/heads/x: protected branch".to_owned()]);
522 let fine = [pkt_line("unpack ok\n"), pkt_line("ok refs/heads/main\n"), pkt_line("ok refs/heads/x\n")].concat();
523 assert!(refused(&fine, &commands).is_empty());
524 }
525}