Something went wrong. Try again.
Monorepo for Tangled
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540use std::io::Write;
use axum::http::{HeaderMap, HeaderValue, Method, header};use knot_git::{Filter, RefRecord, Repo};use knot_runtime::{HttpRequest, HttpResponse, HttpTransport, NetworkError};use knot_types::{HttpStatus, ObjectFormat, Oid, RefName};use url::Url;
use crate::error::PackError;use crate::pkt::{self, Frame};use crate::{HaveOids, WantOids};
#[derive(Debug, thiserror::Error)]pub enum FetchError { #[error("upstream url: {0}")] Url(String), #[error("upstream network: {0}")] Network(#[from] NetworkError), #[error("upstream returned http status {0}")] Status(HttpStatus), #[error("upstream protocol: {0}")] Protocol(String), #[error("upstream reported: {0}")] Remote(String), #[error("fetched pack exceeds {limit} bytes")] PackTooLarge { limit: u64 }, #[error(transparent)] Pack(#[from] PackError),}
#[derive(Debug, Clone, PartialEq, Eq)]pub struct UpstreamRefs { pub object_format: ObjectFormat, pub head_symref: Option<RefName>, pub refs: Vec<RefRecord>,}
impl UpstreamRefs { pub fn tips(&self) -> Vec<Oid> { let mut tips: Vec<Oid> = self.refs.iter().map(|record| record.target).collect(); tips.sort_unstable(); tips.dedup(); tips }
pub fn find(&self, name: &RefName) -> Option<Oid> { self.refs .iter() .find(|record| record.name == *name) .map(|record| record.target) }}
fn protocol(message: impl Into<String>) -> FetchError { FetchError::Protocol(message.into())}
fn endpoint(base: &Url, suffix: &str) -> Result<Url, FetchError> { let trimmed = base.as_str().trim_end_matches('/'); Url::parse(&format!("{trimmed}/{suffix}")).map_err(|error| FetchError::Url(error.to_string()))}
fn headers_v2(content_type: Option<&'static str>) -> HeaderMap { let mut headers = HeaderMap::new(); headers.insert("git-protocol", HeaderValue::from_static("version=2")); if let Some(content_type) = content_type { headers.insert(header::CONTENT_TYPE, HeaderValue::from_static(content_type)); } headers}
async fn execute( http: &dyn HttpTransport, request: HttpRequest,) -> Result<HttpResponse, FetchError> { let response = http.execute(request).await?; if !response.status.is_success() { return Err(FetchError::Status(HttpStatus::new( response.status.as_u16(), ))); } Ok(response)}
pub fn parse_advertisement(body: &[u8]) -> Result<ObjectFormat, FetchError> { let lines = pkt::data_payloads_all(body).map_err(|error| protocol(error.to_string()))?; let lines: Vec<&str> = lines .iter() .map(|line| std::str::from_utf8(line).unwrap_or_default().trim_end()) .collect(); let has = |name: &str| { lines .iter() .any(|line| *line == name || line.starts_with(&format!("{name}="))) }; if !has("version 2") { return Err(protocol("upstream doesn't speak git protocol v2")); } if !has("ls-refs") || !has("fetch") { return Err(protocol("upstream is missing ls-refs or fetch v2 command")); } lines .iter() .find_map(|line| line.strip_prefix("object-format=")) .map_or(Ok(ObjectFormat::SHA1), |token| { ObjectFormat::from_capability(token).ok_or_else(|| { protocol(format!("upstream advertises unknown object format {token}")) }) })}
pub fn ls_refs_request(prefixes: &[&str]) -> Result<Vec<u8>, PackError> { let mut buf = Vec::new(); pkt::write_data(&mut buf, b"command=ls-refs\n")?; pkt::write_data(&mut buf, b"agent=knot/0\n")?; pkt::write_delim(&mut buf)?; pkt::write_data(&mut buf, b"symrefs\n")?; prefixes.iter().try_for_each(|prefix| { pkt::write_data(&mut buf, format!("ref-prefix {prefix}\n").as_bytes()) })?; pkt::write_flush(&mut buf)?; Ok(buf)}
pub fn parse_ls_refs(body: &[u8], object_format: ObjectFormat) -> Result<UpstreamRefs, FetchError> { let lines = pkt::data_payloads(body).map_err(|error| protocol(error.to_string()))?; lines.iter().try_fold( UpstreamRefs { object_format, head_symref: None, refs: Vec::new(), }, |mut refs, line| { let text = std::str::from_utf8(line) .map_err(|_| protocol("ref line isn't utf-8"))? .trim_end(); if let Some(message) = text.strip_prefix("ERR ") { return Err(FetchError::Remote(message.to_string())); } let (oid, rest) = text .split_once(' ') .ok_or_else(|| protocol(format!("malformed ref line: {text}")))?; let target = Oid::from_hex(oid).map_err(|_| protocol(format!("malformed ref oid: {oid}")))?; if target.object_id().kind() != object_format.kind() { return Err(protocol(format!( "ref oid {oid} doesn't match advertised object format {}", object_format.capability() ))); } let mut attributes = rest.split(' '); match attributes.next() { Some("HEAD") => { refs.head_symref = attributes .find_map(|attribute| attribute.strip_prefix("symref-target:")) .and_then(|symref| RefName::new(symref).ok()); } Some(name) => { if let Ok(name) = RefName::new(name) { refs.refs.push(RefRecord { name, target }); } } None => return Err(protocol(format!("malformed ref line: {text}"))), } Ok(refs) }, )}
pub fn fetch_request(wants: &WantOids, haves: &HaveOids) -> Result<Vec<u8>, PackError> { let mut buf = Vec::new(); pkt::write_data(&mut buf, b"command=fetch\n")?; pkt::write_data(&mut buf, b"agent=knot/0\n")?; pkt::write_delim(&mut buf)?; pkt::write_data(&mut buf, b"no-progress\n")?; pkt::write_data(&mut buf, b"ofs-delta\n")?; wants .iter() .try_for_each(|want| pkt::write_data(&mut buf, format!("want {want}\n").as_bytes()))?; haves .iter() .try_for_each(|have| pkt::write_data(&mut buf, format!("have {have}\n").as_bytes()))?; pkt::write_data(&mut buf, b"done\n")?; pkt::write_flush(&mut buf)?; Ok(buf)}
pub fn parse_fetch_response(body: &[u8], max_pack_bytes: u64) -> Result<Vec<u8>, FetchError> { let (pack, in_packfile) = pkt::frames(body, None) .map(|frame| frame.map_err(|error| protocol(error.to_string()))) .try_fold( (Vec::new(), false), |(mut pack, in_packfile), frame| match (frame?.0, in_packfile) { (Frame::Data(payload), false) => { if let Some(message) = payload .strip_prefix(b"ERR ".as_slice()) .map(|rest| String::from_utf8_lossy(rest).trim_end().to_string()) { return Err(FetchError::Remote(message)); } let entered = payload.strip_suffix(b"\n".as_slice()).unwrap_or(payload) == b"packfile"; Ok((pack, entered)) } (Frame::Data(payload), true) => match payload.split_first() { Some((1, data)) => { if pack.len() as u64 + data.len() as u64 > max_pack_bytes { return Err(FetchError::PackTooLarge { limit: max_pack_bytes, }); } pack.extend_from_slice(data); Ok((pack, true)) } Some((2, _)) => Ok((pack, true)), Some((3, message)) => Err(FetchError::Remote( String::from_utf8_lossy(message).trim_end().to_string(), )), _ => Err(protocol("empty sideband frame in packfile section")), }, (_, in_packfile) => Ok((pack, in_packfile)), }, )?; if !in_packfile { return Err(protocol("upstream response has no packfile section")); } Ok(pack)}
pub async fn remote_refs( http: &dyn HttpTransport, base: &Url, prefixes: &[&str],) -> Result<UpstreamRefs, FetchError> { let advertise = endpoint(base, "info/refs?service=git-upload-pack")?; let response = execute( http, HttpRequest { method: Method::GET, url: advertise, headers: headers_v2(None), body: None, }, ) .await?; let object_format = parse_advertisement(&response.body)?;
let upload = endpoint(base, "git-upload-pack")?; let response = execute( http, HttpRequest { method: Method::POST, url: upload, headers: headers_v2(Some("application/x-git-upload-pack-request")), body: Some(ls_refs_request(prefixes)?.into()), }, ) .await?; parse_ls_refs(&response.body, object_format)}
pub async fn remote_pack( http: &dyn HttpTransport, base: &Url, wants: &WantOids, haves: &HaveOids, max_pack_bytes: u64,) -> Result<Vec<u8>, FetchError> { if wants.is_empty() { return Ok(Vec::new()); } let upload = endpoint(base, "git-upload-pack")?; let response = execute( http, HttpRequest { method: Method::POST, url: upload, headers: headers_v2(Some("application/x-git-upload-pack-request")), body: Some(fetch_request(wants, haves)?.into()), }, ) .await?; parse_fetch_response(&response.body, max_pack_bytes)}
pub fn local_refs(source: &Repo, prefixes: &[&str]) -> Result<UpstreamRefs, FetchError> { let refs = source .advertised_refs() .map_err(PackError::from)? .iter() .filter(|record| crate::upload::matches_prefix(record.name.as_str(), prefixes)) .cloned() .collect(); Ok(UpstreamRefs { object_format: source.object_format(), head_symref: source.head().map(|head| head.name), refs, })}
struct BoundedPack { buf: Vec<u8>, limit: u64, overflowed: bool,}
impl Write for BoundedPack { fn write(&mut self, data: &[u8]) -> std::io::Result<usize> { if self.buf.len() as u64 + data.len() as u64 > self.limit { self.overflowed = true; return Err(std::io::Error::other("pack byte limit exceeded")); } self.buf.extend_from_slice(data); Ok(data.len()) }
fn flush(&mut self) -> std::io::Result<()> { Ok(()) }}
pub fn local_pack( source: &Repo, wants: &WantOids, haves: &HaveOids, max_pack_bytes: u64,) -> Result<Vec<u8>, FetchError> { if wants.is_empty() { return Ok(Vec::new()); } let oids = source .select_pack_objects_filtered( wants.wants(), haves.haves(), Filter::None, crate::upload::selection_budget(), ) .map_err(PackError::from)? .send; let mut out = BoundedPack { buf: Vec::new(), limit: max_pack_bytes, overflowed: false, }; match crate::objects::write_pack( &source.objects_dir(), oids, None, &mut out, source.object_format().kind(), ) { Ok(()) => Ok(out.buf), Err(_) if out.overflowed => Err(FetchError::PackTooLarge { limit: max_pack_bytes, }), Err(error) => Err(FetchError::Pack(error)), }}
#[cfg(test)]mod tests { use super::*;
fn data(buf: &mut Vec<u8>, line: &[u8]) { pkt::write_data(buf, line).unwrap(); }
#[test] fn the_v2_advertisement_parses_to_its_object_format_and_v0_is_refused() { let scan = tempfile::tempdir().unwrap(); let repo = knot_git::Layout::new(scan.path()) .create(&knot_types::RepoDid::new("did:plc:squid").unwrap()) .unwrap(); let v2 = crate::upload::advertise(&repo).unwrap(); assert_eq!(parse_advertisement(&v2).unwrap(), ObjectFormat::SHA1);
let advertise = |extra: Option<&str>| { let mut buf = Vec::new(); data(&mut buf, b"version 2\n"); data(&mut buf, b"ls-refs\n"); data(&mut buf, b"fetch\n"); if let Some(extra) = extra { data(&mut buf, format!("{extra}\n").as_bytes()); } pkt::write_flush(&mut buf).unwrap(); parse_advertisement(&buf) }; assert_eq!( advertise(Some("object-format=sha256")).unwrap(), ObjectFormat::SHA256 ); assert_eq!( advertise(None).unwrap(), ObjectFormat::SHA1, "protocol v2 treats an absent object-format as sha1" ); assert!(matches!( advertise(Some("object-format=sha3")), Err(FetchError::Protocol(_)) ));
let mut v0 = Vec::new(); data(&mut v0, b"# service=git-upload-pack\n"); pkt::write_flush(&mut v0).unwrap(); data( &mut v0, b"95d09f2b10159347eece71399a7e2e907ea3df4f HEAD\0side-band-64k\n", ); pkt::write_flush(&mut v0).unwrap(); assert!(matches!( parse_advertisement(&v0), Err(FetchError::Protocol(_)) )); }
#[test] fn ls_refs_lines_parse_with_symref_skip_head_and_enforce_the_object_format() { let mut body = Vec::new(); data( &mut body, b"95d09f2b10159347eece71399a7e2e907ea3df4f HEAD symref-target:refs/heads/main\n", ); data( &mut body, b"95d09f2b10159347eece71399a7e2e907ea3df4f refs/heads/main\n", ); pkt::write_flush(&mut body).unwrap(); let refs = parse_ls_refs(&body, ObjectFormat::SHA1).unwrap(); assert_eq!( refs.head_symref.as_ref().map(RefName::as_str), Some("refs/heads/main") ); assert_eq!(refs.refs.len(), 1); assert_eq!(refs.refs[0].name.as_str(), "refs/heads/main"); assert_eq!(refs.tips().len(), 1);
let single = |oid: &str| { let mut buf = Vec::new(); data(&mut buf, format!("{oid} refs/heads/main\n").as_bytes()); pkt::write_flush(&mut buf).unwrap(); buf }; let sha1 = "95d09f2b10159347eece71399a7e2e907ea3df4f"; let sha256 = "6ef19b41225c5369f1c104d45d8d85efa9b057b53b14b4b9b939dd74decc5321"; assert!(parse_ls_refs(&single(sha256), ObjectFormat::SHA256).is_ok()); assert!(matches!( parse_ls_refs(&single(sha256), ObjectFormat::SHA1), Err(FetchError::Protocol(_)) )); assert!(matches!( parse_ls_refs(&single(sha1), ObjectFormat::SHA256), Err(FetchError::Protocol(_)) )); }
#[test] fn a_malformed_ref_oid_is_a_protocol_error() { let mut body = Vec::new(); data(&mut body, b"zzzz refs/heads/main\n"); pkt::write_flush(&mut body).unwrap(); assert!(matches!( parse_ls_refs(&body, ObjectFormat::SHA1), Err(FetchError::Protocol(_)) )); }
#[test] fn an_err_line_is_surfaced_as_remote() { let mut body = Vec::new(); data(&mut body, b"ERR access denied\n"); pkt::write_flush(&mut body).unwrap(); assert!(matches!( parse_ls_refs(&body, ObjectFormat::SHA1), Err(FetchError::Remote(message)) if message == "access denied" )); }
#[test] fn the_packfile_section_demuxes_data_and_drops_progress() { let mut body = Vec::new(); data(&mut body, b"packfile\n"); data(&mut body, b"\x01PACKDATA"); data(&mut body, b"\x02counting objects\n"); data(&mut body, b"\x01MORE"); pkt::write_flush(&mut body).unwrap(); let pack = parse_fetch_response(&body, 1024).unwrap(); assert_eq!(pack, b"PACKDATAMORE"); }
#[test] fn a_sideband_error_band_is_remote_and_the_limit_holds() { let mut body = Vec::new(); data(&mut body, b"packfile\n"); data(&mut body, b"\x03out of disk\n"); pkt::write_flush(&mut body).unwrap(); assert!(matches!( parse_fetch_response(&body, 1024), Err(FetchError::Remote(message)) if message == "out of disk" ));
let mut big = Vec::new(); data(&mut big, b"packfile\n"); data(&mut big, b"\x01PACKDATA"); pkt::write_flush(&mut big).unwrap(); assert!(matches!( parse_fetch_response(&big, 4), Err(FetchError::PackTooLarge { limit: 4 }) )); }
#[test] fn a_response_without_a_packfile_section_is_refused() { let mut body = Vec::new(); data(&mut body, b"acknowledgments\n"); data(&mut body, b"NAK\n"); pkt::write_flush(&mut body).unwrap(); assert!(matches!( parse_fetch_response(&body, 1024), Err(FetchError::Protocol(_)) )); }
#[test] fn the_fetch_request_includes_wants_haves_and_done() { let want = Oid::from_hex("95d09f2b10159347eece71399a7e2e907ea3df4f").unwrap(); let have = Oid::from_hex("2222222222222222222222222222222222222222").unwrap(); let body = fetch_request(&WantOids::new(vec![want]), &HaveOids::new(vec![have])).unwrap(); let lines = pkt::data_payloads_all(&body).unwrap(); let text: Vec<&str> = lines .iter() .map(|line| std::str::from_utf8(line).unwrap().trim_end()) .collect(); assert!(text.contains(&"command=fetch")); assert!(text.contains(&"want 95d09f2b10159347eece71399a7e2e907ea3df4f")); assert!(text.contains(&"have 2222222222222222222222222222222222222222")); assert!(text.contains(&"done")); assert!(text.contains(&"no-progress")); }}