Skip to content
Sign in

flagon-io/g1t

public
0

g1t/services/repos/src/git_http.rs

1,428 lines56,179 bytesCodeBlame
1//! Git over HTTPS: the smart HTTP remote at `/<namespace>/<repo>.git`,
2//! proxied to the git store with a short-lived token.
3
4use std::cell::RefCell;
5use std::rc::Rc;
6
7use futures_util::StreamExt;
8use g1t_contracts::identity::GitCredentialsArgs;
9use g1t_contracts::repos::{GitAccess, GitService, RepoPath};
10use g1t_contracts::{FailureCode, Outcome, Viewer};
11use worker::js_sys::Uint8Array;
12use worker::wasm_bindgen::JsValue;
13use worker::{Fetch, Fetcher, Headers, Method, Request, RequestInit, Response, Result, Url};
14
15use crate::meters;
16use crate::pack_limits::{PackSizer, Violation};
17use crate::resilience::{self, Busy, Failure};
18
19const ENDPOINTS: [&str; 3] = ["info/refs", "git-upload-pack", "git-receive-pack"];
20const FORWARDED_HEADERS: [&str; 5] = [
21 "accept",
22 "content-encoding",
23 "content-type",
24 "git-protocol",
25 "user-agent",
26];
27
28/// A git request, parsed from its URL.
29pub struct GitRequest {
30 pub path: RepoPath,
31 pub endpoint: &'static str,
32 pub service: GitService,
33}
34
35impl GitRequest {
36 /// The same request for the repository of the same name under
37 /// `namespace`: where a workspace alias leads.
38 pub fn under(&self, namespace: &str) -> GitRequest {
39 GitRequest {
40 path: RepoPath {
41 namespace: namespace.to_owned(),
42 name: self.path.name.clone(),
43 },
44 endpoint: self.endpoint,
45 service: self.service,
46 }
47 }
48}
49
50/// Parses `/<namespace>/<name>[.git]/<endpoint>`, or returns `None` if the
51/// request is not git's.
52pub fn parse(url: &Url) -> Option<GitRequest> {
53 let path = url.path().strip_prefix('/')?;
54 let endpoint = ENDPOINTS
55 .into_iter()
56 .find(|endpoint| path.ends_with(&format!("/{endpoint}")))?;
57 let repo = &path[..path.len() - endpoint.len() - 1];
58 let (namespace, name) = repo.split_once('/')?;
59 let name = name.strip_suffix(".git").unwrap_or(name);
60 if namespace.is_empty() || name.is_empty() || name.contains('/') {
61 return None;
62 }
63 let service = if endpoint == "info/refs" {
64 url.query_pairs()
65 .find(|(key, _)| key == "service")
66 .map(|(_, value)| value.into_owned())?
67 } else {
68 endpoint.to_owned()
69 };
70 let service = match service.as_str() {
71 "git-upload-pack" => GitService::UploadPack,
72 "git-receive-pack" => GitService::ReceivePack,
73 _ => return None,
74 };
75 Some(GitRequest {
76 path: RepoPath {
77 namespace: namespace.to_owned(),
78 name: name.to_owned(),
79 },
80 endpoint,
81 service,
82 })
83}
84
85/// How long each step of a git request took, sent back to git as a
86/// `Server-Timing` header so that a slow clone shows where its time went.
87/// Step names and whole milliseconds only. The Workers clock moves only
88/// while a request waits on something, so each step is the time spent
89/// waiting on the database, another service or the git store. A note
90/// says how a step went without a duration: `refs;desc=hit-colo`.
91pub struct Timing {
92 started: u64,
93 last: u64,
94 steps: Vec<(&'static str, u64)>,
95 notes: Vec<(&'static str, &'static str)>,
96}
97
98impl Timing {
99 pub fn start() -> Self {
100 let now = g1t_kit::now_ms();
101 Self {
102 started: now,
103 last: now,
104 steps: Vec::new(),
105 notes: Vec::new(),
106 }
107 }
108
109 /// Says how `name` went: where an answer or a credential came from.
110 pub fn note(&mut self, name: &'static str, description: &'static str) {
111 self.notes.push((name, description));
112 }
113
114 /// Ends a step, named `step`, that began when the last one ended.
115 pub fn mark(&mut self, step: &'static str) {
116 let now = g1t_kit::now_ms();
117 self.steps.push((step, now.saturating_sub(self.last)));
118 self.last = now;
119 }
120
121 /// `response`, with how long each step took.
122 pub fn apply(&self, response: Response) -> Result<Response> {
123 let total = g1t_kit::now_ms().saturating_sub(self.started);
124 let headers = response.headers().clone();
125 headers.set("server-timing", &server_timing(&self.steps, &self.notes, total))?;
126 Ok(response.with_headers(headers))
127 }
128}
129
130/// A `Server-Timing` value: each step with its duration, the notes, then
131/// the total.
132fn server_timing(steps: &[(&str, u64)], notes: &[(&str, &str)], total: u64) -> String {
133 steps
134 .iter()
135 .map(|(step, ms)| format!("{step};dur={ms}"))
136 .chain(notes.iter().map(|(name, description)| format!("{name};desc={description}")))
137 .chain(std::iter::once(format!("total;dur={total}")))
138 .collect::<Vec<_>>()
139 .join(", ")
140}
141
142/// The user named by an HTTP Basic `Authorization` header, as git sends it.
143pub async fn viewer(request: &Request, identity: &Fetcher) -> Result<Viewer> {
144 let Some(header) = request.headers().get("authorization")? else {
145 return Ok(None);
146 };
147 let Some((scheme, encoded)) = header.split_once(' ') else {
148 return Ok(None);
149 };
150 if !scheme.eq_ignore_ascii_case("basic") {
151 return Ok(None);
152 }
153 let Some(decoded) = decode_base64(encoded.trim()) else {
154 return Ok(None);
155 };
156 let Some((username, secret)) = decoded.split_once(':') else {
157 return Ok(None);
158 };
159 g1t_kit::call(
160 identity,
161 "user_for_git_credentials",
162 &GitCredentialsArgs {
163 username: username.to_owned(),
164 secret: secret.to_owned(),
165 },
166 )
167 .await
168}
169
170/// Standard base64 to a UTF-8 string, or `None` if either step fails.
171fn decode_base64(input: &str) -> Option<String> {
172 let mut bytes = Vec::with_capacity(input.len() * 3 / 4);
173 let mut buffer = 0u32;
174 let mut bits = 0;
175 for byte in input.bytes().filter(|byte| *byte != b'=') {
176 let value = match byte {
177 b'A'..=b'Z' => byte - b'A',
178 b'a'..=b'z' => byte - b'a' + 26,
179 b'0'..=b'9' => byte - b'0' + 52,
180 b'+' => 62,
181 b'/' => 63,
182 _ => return None,
183 };
184 buffer = (buffer << 6) | u32::from(value);
185 bits += 6;
186 if bits >= 8 {
187 bits -= 8;
188 bytes.push((buffer >> bits) as u8);
189 }
190 }
191 String::from_utf8(bytes).ok()
192}
193
194/// The response for a refused git request. Anonymous callers are asked to
195/// authenticate, which is what makes git prompt for credentials.
196pub fn refuse<T>(outcome: Outcome<T>) -> Result<Response> {
197 let Outcome::Fail(failure) = outcome else {
198 return Response::error("Not found", 404);
199 };
200 let mut response = Response::error(failure.message, failure.code.http_status())?;
201 if failure.code == FailureCode::Unauthenticated {
202 response
203 .headers_mut()
204 .set("www-authenticate", "Basic realm=\"g1t\"")?;
205 }
206 Ok(response)
207}
208
209/// `url` with its first path segment, the workspace, replaced by `slug`.
210pub fn with_namespace(url: &Url, slug: &str) -> Option<String> {
211 let rest = url.path().strip_prefix('/')?.split_once('/')?.1;
212 let mut moved = url.clone();
213 moved.set_path(&format!("/{slug}/{rest}"));
214 Some(moved.to_string())
215}
216
217/// Where a git request for a renamed workspace's old address should go
218/// now, if its first segment is an old slug that still redirects.
219pub async fn renamed(url: &Url, identity: &Fetcher) -> Result<Option<String>> {
220 let Some(old) = url.path().strip_prefix('/').and_then(|path| path.split('/').next()) else {
221 return Ok(None);
222 };
223 let current: Option<String> = g1t_kit::call(
224 identity,
225 "resolve_slug",
226 &g1t_contracts::identity::SlugArgs {
227 slug: old.to_owned(),
228 },
229 )
230 .await?;
231 Ok(current.and_then(|slug| with_namespace(url, &slug)))
232}
233
234/// The request under the workspace its first segment is an alias of
235/// (identity's aliases.rs: `g1t` for `flagon-io`), if it is one. Answered
236/// in place rather than redirected: a push does not follow a redirect.
237pub async fn aliased(git: &GitRequest, identity: &Fetcher) -> Result<Option<GitRequest>> {
238 let slug: Option<String> = g1t_kit::call(
239 identity,
240 "resolve_alias",
241 &g1t_contracts::identity::SlugArgs {
242 slug: git.path.namespace.clone(),
243 },
244 )
245 .await?;
246 Ok(slug.map(|slug| git.under(&slug)))
247}
248
249/// `url` with its repository, the first two path segments, replaced by
250/// `to`: where a request for a transferred repository's old path goes.
251/// Keeps whether the old address ended in `.git`.
252pub fn transferred(url: &Url, to: &RepoPath) -> Option<String> {
253 let path = url.path().strip_prefix('/')?;
254 let mut segments = path.splitn(3, '/');
255 let (_, name, rest) = (segments.next()?, segments.next()?, segments.next()?);
256 let suffix = if name.ends_with(".git") { ".git" } else { "" };
257 let mut moved = url.clone();
258 moved.set_path(&format!("/{}/{}{suffix}/{rest}", to.namespace, to.name));
259 Some(moved.to_string())
260}
261
262/// A permanent redirect: 301 for git's first request for refs, which it
263/// follows and then uses the new address for the rest; 308 for the
264/// others, so a POST stays a POST.
265pub fn moved(location: &str, get: bool) -> Result<Response> {
266 let mut response = Response::empty()?.with_status(if get { 301 } else { 308 });
267 response.headers_mut().set("location", location)?;
268 Ok(response)
269}
270
271const ZERO_ID: &str = "0000000000000000000000000000000000000000";
272const HEADS: &str = "refs/heads/";
273const TAGS: &str = "refs/tags/";
274
275/// One ref a push asks to change.
276struct Command {
277 old: String,
278 new: String,
279 name: String,
280}
281
282/// The commands at the start of a receive-pack request, and the
283/// capabilities the client sent with the first of them.
284fn commands(body: &[u8]) -> (Vec<Command>, String) {
285 let mut commands = Vec::new();
286 let mut capabilities = String::new();
287 let mut position = 0;
288 // Commands are pkt-lines; a flush packet ends them and the pack follows.
289 while let Some(length) = body
290 .get(position..position + 4)
291 .and_then(|hex| std::str::from_utf8(hex).ok())
292 .and_then(|hex| usize::from_str_radix(hex, 16).ok())
293 {
294 if length < 4 || position + length > body.len() {
295 break;
296 }
297 let line = &body[position + 4..position + length];
298 position += length;
299 // `<old> <new> <ref>`, and on the first command a NUL then capabilities.
300 let mut halves = line.splitn(2, |byte| *byte == 0);
301 let command = halves.next().unwrap_or_default();
302 if let Some(rest) = halves.next() {
303 capabilities = String::from_utf8_lossy(rest).trim().to_owned();
304 }
305 let Ok(command) = std::str::from_utf8(command) else {
306 continue;
307 };
308 let mut parts = command.trim_end().splitn(3, ' ');
309 if let (Some(old), Some(new), Some(name)) = (parts.next(), parts.next(), parts.next()) {
310 commands.push(Command {
311 old: old.to_owned(),
312 new: new.to_owned(),
313 name: name.to_owned(),
314 });
315 }
316 }
317 (commands, capabilities)
318}
319
320/// The bytes of the pack a receive-pack request carries: everything after
321/// the flush packet that ends its commands. Zero for a push that only
322/// deletes refs.
323pub(crate) fn pack_bytes(body: &[u8]) -> u64 {
324 let mut position = 0;
325 while let Some(length) = body
326 .get(position..position + 4)
327 .and_then(|hex| std::str::from_utf8(hex).ok())
328 .and_then(|hex| usize::from_str_radix(hex, 16).ok())
329 {
330 if length == 0 {
331 return (body.len() - position - 4) as u64;
332 }
333 if length < 4 || position + length > body.len() {
334 break;
335 }
336 position += length;
337 }
338 0
339}
340
341fn pkt_line(payload: &[u8]) -> Vec<u8> {
342 let mut line = format!("{:04x}", payload.len() + 4).into_bytes();
343 line.extend_from_slice(payload);
344 line
345}
346
347/// The branches and tags a push asks to change, as rules see them: the
348/// full ref, where it pointed (`None`: it is created) and where it will
349/// (`None`: it is deleted).
350pub(crate) fn ref_updates(body: &[u8]) -> Vec<(String, Option<String>, Option<String>)> {
351 commands(body)
352 .0
353 .into_iter()
354 .filter(|command| command.name.starts_with(HEADS) || command.name.starts_with(TAGS))
355 .map(|Command { old, new, name }| (name, (old != ZERO_ID).then_some(old), (new != ZERO_ID).then_some(new)))
356 .collect()
357}
358
359/// A report-status answer, as git expects it.
360pub(crate) fn report_response(report: Vec<u8>) -> Result<Response> {
361 let headers = Headers::new();
362 headers.set("content-type", "application/x-git-receive-pack-result")?;
363 headers.set("cache-control", "no-cache")?;
364 Ok(Response::from_bytes(report)?.with_headers(headers))
365}
366
367/// What git is told when a push would change a protected branch: every ref
368/// in it is declined, with the reason against the protected one, so that
369/// git prints it beside the branch. `None` if the push leaves the branch
370/// alone, or creates it in a repository that does not have it yet. Only
371/// where rulesets cannot be read (an installation without the work
372/// service); rulesets decide everywhere else (rules.rs).
373pub(crate) fn refusal(body: &[u8], protected: &str) -> Option<Vec<u8>> {
374 let (commands, capabilities) = commands(body);
375 let reference = format!("{HEADS}{protected}");
376 if !commands
377 .iter()
378 .any(|command| command.name == reference && command.old != ZERO_ID)
379 {
380 return None;
381 }
382 let mut report = pkt_line(b"unpack ok\n");
383 for command in &commands {
384 let reason = if command.name == reference {
385 format!("{protected} is protected: push a branch and open a pull request")
386 } else {
387 format!("not pushed, because the same push would change {protected}")
388 };
389 report.extend(pkt_line(
390 format!("ng {} {reason}\n", command.name).as_bytes(),
391 ));
392 }
393 report.extend_from_slice(b"0000");
394 Some(framed(report, &capabilities, &[]))
395}
396
397/// A report-status as git expects it: inside channel 1 when the client
398/// asked for side-band, after `messages` on channel 2, which git prints as
399/// `remote:` lines. Without side-band the messages cannot be shown.
400fn framed(report: Vec<u8>, capabilities: &str, messages: &[String]) -> Vec<u8> {
401 let sideband = capabilities
402 .split(' ')
403 .any(|capability| capability.starts_with("side-band"));
404 if !sideband {
405 return report;
406 }
407 let mut body = Vec::new();
408 for message in messages {
409 let mut packet = vec![2u8];
410 packet.extend_from_slice(message.as_bytes());
411 packet.push(b'\n');
412 body.extend(pkt_line(&packet));
413 }
414 // side-band (not -64k) packets carry at most 1000 bytes.
415 for chunk in report.chunks(990) {
416 let mut packet = vec![1u8];
417 packet.extend_from_slice(chunk);
418 body.extend(pkt_line(&packet));
419 }
420 body.extend_from_slice(b"0000");
421 body
422}
423
424/// Declines every ref in a push with `reason`, explaining why in
425/// `messages`: what push protection answers when a push adds a secret.
426pub fn declined(body: &[u8], reason: &str, messages: &[String]) -> Result<Response> {
427 let (commands, capabilities) = commands(body);
428 let mut report = pkt_line(b"unpack ok\n");
429 for command in &commands {
430 report.extend(pkt_line(format!("ng {} {reason}\n", command.name).as_bytes()));
431 }
432 report.extend_from_slice(b"0000");
433 let headers = Headers::new();
434 headers.set("content-type", "application/x-git-receive-pack-result")?;
435 headers.set("cache-control", "no-cache")?;
436 Ok(Response::from_bytes(framed(report, &capabilities, messages))?.with_headers(headers))
437}
438
439/// A branch or tag a push asks to move.
440#[derive(Debug, PartialEq, Eq)]
441pub struct Pushed {
442 /// The full ref: `refs/heads/main`, `refs/tags/v1`.
443 pub git_ref: String,
444 /// Where it pointed before; `None` for a new ref.
445 pub before: Option<String>,
446 pub after: String,
447}
448
449impl Pushed {
450 pub fn branch(&self) -> Option<&str> {
451 self.git_ref.strip_prefix(HEADS)
452 }
453}
454
455/// The branches and tags a push asks to move, read from the commands at the
456/// start of a receive-pack request. Deletions and other refs are left out.
457fn pushed_branches(body: &[u8]) -> Vec<Pushed> {
458 commands(body)
459 .0
460 .into_iter()
461 .filter(|command| command.new != ZERO_ID)
462 .filter(|command| command.name.starts_with(HEADS) || command.name.starts_with(TAGS))
463 .map(|Command { old, new, name }| Pushed {
464 git_ref: name,
465 before: (old != ZERO_ID).then_some(old),
466 after: new,
467 })
468 .collect()
469}
470
471/// The git store's answer, and what the request asked it to change.
472pub struct Forwarded {
473 pub response: Response,
474 /// For a push: the branches and tags it asks to move, and the commits
475 /// to move them to. Whether each moved is for the caller to confirm.
476 pub pushed: Vec<Pushed>,
477 /// For a push: the size of the pack it sent, for the storage meter.
478 pub pack_bytes: u64,
479 /// The bytes sent to the store.
480 pub sent: u64,
481 /// Whether the answer is the store's own (not g1t's, for a store that
482 /// was busy).
483 pub from_store: bool,
484 /// For a push: whether it was too large to scan for secrets first and
485 /// was streamed to the store unscanned (`LargePushes::Unscanned`). Its
486 /// `git.push` events say so, and security scans it after it lands.
487 pub unscanned: bool,
488}
489
490/// What became of a git request.
491pub enum Push {
492 Forwarded(Forwarded),
493 /// A push the rules of its branches or tags refuse (rules.rs), answered
494 /// here without reaching the store.
495 Refused(Response),
496 /// A push that adds a secret nobody allowed, answered the same way.
497 Blocked(Response),
498 /// A push the store could not hold: an object or the repository too
499 /// large, or too large to check. With the reason, for the audit log.
500 Declined(Response, String),
501}
502
503/// What a push may bring, checked as it arrives (pack_limits.rs).
504#[derive(Clone, Copy, Debug)]
505pub struct PushLimits {
506 /// The largest object the store holds.
507 pub max_object: u64,
508 /// What the repository holds now, as g1t counts it.
509 pub held: u64,
510 /// The most a repository may hold.
511 pub repo_limit: u64,
512 /// The largest push that is read whole and scanned for secrets.
513 pub scan_cap: usize,
514 /// What happens to a larger one.
515 pub large: LargePushes,
516}
517
518impl Default for PushLimits {
519 fn default() -> Self {
520 PushLimits {
521 max_object: crate::pack_limits::MAX_OBJECT_BYTES,
522 held: 0,
523 repo_limit: crate::pack_limits::DEFAULT_REPO_LIMIT_BYTES,
524 scan_cap: crate::secret_scan::MAX_SCANNED_PUSH,
525 large: LargePushes::Refuse,
526 }
527 }
528}
529
530/// What happens to a push larger than [`PushLimits::scan_cap`]: set by
531/// `LARGE_PUSHES`.
532#[derive(Clone, Copy, Debug, PartialEq, Eq)]
533pub enum LargePushes {
534 /// Declined (the default): push protection cannot read it, so it does
535 /// not let it in.
536 Refuse,
537 /// Streamed to the store without a scan for secrets; the size limits are
538 /// still checked as it passes.
539 Unscanned,
540}
541
542impl LargePushes {
543 pub fn from_var(value: Option<&str>) -> Self {
544 match value.map(str::trim) {
545 Some("unscanned") => LargePushes::Unscanned,
546 _ => LargePushes::Refuse,
547 }
548 }
549}
550
551/// A push the store could not hold.
552#[derive(Clone, Debug, PartialEq, Eq)]
553pub enum SizeViolation {
554 Object { size: u64 },
555 Repository { held: u64, incoming: u64, limit: u64 },
556 Unscannable { size: u64, cap: usize },
557}
558
559/// Why a push is declined for its size, as git shows it: the `ng` reason,
560/// and the lines printed as `remote:`.
561pub fn size_refusal(violation: &SizeViolation) -> (String, Vec<String>) {
562 use crate::pack_limits::{MAX_OBJECT_BYTES, PLATFORM_BODY_LIMIT_BYTES, megabytes};
563 match violation {
564 SizeViolation::Object { size } => (
565 format!("a file of {} is over the {} limit", megabytes(*size), megabytes(MAX_OBJECT_BYTES)),
566 vec![
567 format!("g1t stores files of up to {} each; this push has one of {}.", megabytes(MAX_OBJECT_BYTES), megabytes(*size)),
568 "Take it out of the commits (git rm --cached, then amend or rebase), and keep large".to_owned(),
569 "files elsewhere: https://docs.g1t.sh/guides/git/#size-limits. Nothing was pushed.".to_owned(),
570 ],
571 ),
572 SizeViolation::Repository { held, incoming, limit } => (
573 "the repository would be over its size limit".to_owned(),
574 vec![
575 format!(
576 "This repository holds about {} and the push adds {}, past the {} a repository may hold.",
577 megabytes(*held),
578 megabytes(*incoming),
579 megabytes(*limit)
580 ),
581 "Delete what you no longer need, or split it: https://docs.g1t.sh/guides/git/#size-limits.".to_owned(),
582 "Nothing was pushed.".to_owned(),
583 ],
584 ),
585 SizeViolation::Unscannable { size, cap } => (
586 "the push is too large to check for secrets".to_owned(),
587 vec![
588 format!(
589 "g1t checks every push for secrets and reads up to {} at once; this one is {}.",
590 megabytes(*cap as u64),
591 megabytes(*size)
592 ),
593 "Push in parts, oldest commits first, then push as usual:".to_owned(),
594 " git rev-list --reverse HEAD | awk 'NR % 500 == 0' | xargs -I{} git push origin {}:refs/heads/main".to_owned(),
595 format!("A push over {} is refused by the network before it reaches g1t (HTTP 413).", megabytes(PLATFORM_BODY_LIMIT_BYTES)),
596 "See https://docs.g1t.sh/guides/git/#size-limits. Nothing was pushed.".to_owned(),
597 ],
598 ),
599 }
600}
601
602/// Feeds the next chunk of a push to the size check; the first violation.
603fn check_size(sizer: &mut Option<PackSizer>, chunk: &[u8], limits: &PushLimits) -> Option<SizeViolation> {
604 let walker = sizer.as_mut()?;
605 match walker.feed(chunk) {
606 Ok(()) => {}
607 Err(Violation::ObjectTooLarge { size }) => return Some(SizeViolation::Object { size }),
608 Err(Violation::Malformed(why)) => {
609 // Not for g1t to judge: the store will say.
610 worker::console_error!("push not checked for size: {why}");
611 *sizer = None;
612 return None;
613 }
614 }
615 let incoming = walker.pack_bytes();
616 crate::pack_limits::over_repo_limit(limits.held, incoming, limits.repo_limit).then_some(SizeViolation::Repository {
617 held: limits.held,
618 incoming,
619 limit: limits.repo_limit,
620 })
621}
622
623/// A request body that streams from `stream`, for `fetch`.
624pub(crate) fn stream_body<S>(stream: S) -> Result<JsValue>
625where
626 S: futures_util::TryStream + 'static,
627 S::Ok: Into<Vec<u8>>,
628 S::Error: Into<worker::Error>,
629{
630 let response: worker::web_sys::Response = Response::from_stream(stream)?.into();
631 Ok(response.body().map_or(JsValue::NULL, Into::into))
632}
633
634/// What git is told when the store is busy: 429 or 503, with when to try
635/// again (resilience.rs).
636pub fn busy_response(busy: Busy) -> Result<Response> {
637 let response = Response::error(busy.message(), busy.status())?;
638 response.headers().set("retry-after", &busy.retry_after.to_string())?;
639 Ok(response)
640}
641
642/// The rest of a request's body, read and thrown away so that git hears
643/// the answer; how many bytes it was.
644async fn drain(stream: &mut worker::ByteStream) -> Result<u64> {
645 let mut size = 0;
646 while let Some(chunk) = stream.next().await {
647 size += chunk?.len() as u64;
648 }
649 Ok(size)
650}
651
652/// One packet of a pkt-line stream: data, or a flush (`0000`), delimiter
653/// (`0001`) or response-end (`0002`) packet, kept as its four bytes.
654#[derive(Debug, PartialEq, Eq)]
655enum Packet {
656 Data(Vec<u8>),
657 Special([u8; 4]),
658}
659
660/// The packets in `bytes`, or `None` if it is not a whole pkt-line stream.
661fn packets(bytes: &[u8]) -> Option<Vec<Packet>> {
662 let mut out = Vec::new();
663 let mut position = 0;
664 while position < bytes.len() {
665 let header = bytes.get(position..position + 4)?;
666 let length = usize::from_str_radix(std::str::from_utf8(header).ok()?, 16).ok()?;
667 if length < 4 {
668 out.push(Packet::Special(header.try_into().ok()?));
669 position += 4;
670 continue;
671 }
672 out.push(Packet::Data(bytes.get(position + 4..position + length)?.to_vec()));
673 position += length;
674 }
675 Some(out)
676}
677
678fn encode(packets: &[Packet]) -> Vec<u8> {
679 let mut out = Vec::new();
680 for packet in packets {
681 match packet {
682 Packet::Data(data) => out.extend(pkt_line(data)),
683 Packet::Special(bytes) => out.extend_from_slice(bytes),
684 }
685 }
686 out
687}
688
689/// A ref advertisement (`info/refs` for upload-pack) or a protocol v2
690/// `ls-refs` answer with `HEAD` pointing at `branch`, the repository's
691/// default branch as g1t keeps it, so a clone checks it out. The git store
692/// holds the HEAD it was created with; g1t can change the default branch
693/// since. `None` when there is nothing to change: no `HEAD` line, `HEAD`
694/// already names `branch`, or `branch` is not advertised.
695pub fn with_head(body: &[u8], branch: &str) -> Option<Vec<u8>> {
696 let mut packets = packets(body)?;
697 let target = format!("{HEADS}{branch}");
698 let oid = packets.iter().find_map(|packet| {
699 let Packet::Data(data) = packet else { return None };
700 let line = data.split(|byte| *byte == 0).next()?;
701 let line = std::str::from_utf8(line).ok()?.trim_end();
702 let (oid, name) = line.split_once(' ')?;
703 // v2 lines may carry attributes after the name.
704 let name = name.split(' ').next()?;
705 (name == target).then(|| oid.to_owned())
706 })?;
707 let mut changed = false;
708 for packet in &mut packets {
709 let Packet::Data(data) = packet else { continue };
710 let text = String::from_utf8_lossy(data).into_owned();
711 let Some((_, rest)) = text.split_once(' ') else { continue };
712 if !(rest.starts_with("HEAD\0") || rest.starts_with("HEAD\n") || rest.starts_with("HEAD ") || rest == "HEAD") {
713 continue;
714 }
715 let mut line = format!("{oid} {rest}");
716 // v0: `symref=HEAD:refs/heads/<old>` among the capabilities.
717 // v2: `symref-target:refs/heads/<old>` after the name.
718 for marker in ["symref=HEAD:", "symref-target:"] {
719 if let Some(at) = line.find(marker) {
720 let start = at + marker.len();
721 let end = line[start..]
722 .find([' ', '\n', '\0'])
723 .map_or(line.len(), |offset| start + offset);
724 line.replace_range(start..end, &target);
725 }
726 }
727 changed = line != text;
728 if changed {
729 *data = line.into_bytes();
730 }
731 break;
732 }
733 changed.then(|| encode(&packets))
734}
735
736/// Whether a request to the git store is one whose answer names `HEAD`:
737/// the ref advertisement for a fetch, or a protocol v2 `ls-refs`.
738fn names_head(git: &GitRequest, body: Option<&[u8]>) -> bool {
739 if git.service != GitService::UploadPack {
740 return false;
741 }
742 match body {
743 None => git.endpoint == "info/refs",
744 Some(body) => {
745 git.endpoint == "git-upload-pack"
746 && body.windows(b"command=ls-refs".len()).any(|window| window == b"command=ls-refs")
747 }
748 }
749}
750
751/// Whether a request is a protocol v2 `fetch` still negotiating: it sends
752/// `have` lines and no `done`, so the answer may be acknowledgments only.
753fn negotiating(body: &[u8]) -> bool {
754 let Some(packets) = packets(body) else { return false };
755 let lines: Vec<&[u8]> = packets
756 .iter()
757 .filter_map(|packet| match packet {
758 Packet::Data(data) => Some(data.strip_suffix(b"\n").unwrap_or(data)),
759 Packet::Special(_) => None,
760 })
761 .collect();
762 lines.contains(&b"command=fetch".as_slice())
763 && lines.iter().any(|line| line.starts_with(b"have "))
764 && !lines.contains(&b"done".as_slice())
765}
766
767/// What to do with the start of a store's answer to a negotiating fetch.
768#[derive(Debug, PartialEq, Eq)]
769enum Acknowledged {
770 /// Not enough of it yet to tell.
771 NeedMore,
772 /// Send it on as it is.
773 Whole,
774 /// Acknowledgments without `ready`, followed by more sections: the
775 /// store's answer to keep is these first bytes, ended by a flush.
776 CutAt(usize),
777}
778
779/// How much of the answer to keep. The git store answers a fetch whose
780/// `have`s it does not know with `acknowledgments`, `NAK`, then a pack
781/// anyway; git refuses that ("expected no other sections to be sent after
782/// no 'ready'"), since a server that is not ready must end the response
783/// there and let the client negotiate again. Lines may be `sideband-all`
784/// framed (band 1, `\x01`).
785fn acknowledged(head: &[u8]) -> Acknowledged {
786 let mut position = 0;
787 let mut first = true;
788 loop {
789 let Some(header) = head.get(position..position + 4) else { return Acknowledged::NeedMore };
790 let Some(length) = std::str::from_utf8(header).ok().and_then(|hex| usize::from_str_radix(hex, 16).ok()) else {
791 return Acknowledged::Whole;
792 };
793 if length < 4 {
794 // The acknowledgments section's end: a delimiter means more
795 // sections follow, which only `ready` allows.
796 return match (first, header) {
797 (false, b"0001") => Acknowledged::CutAt(position),
798 _ => Acknowledged::Whole,
799 };
800 }
801 let Some(payload) = head.get(position + 4..position + length) else { return Acknowledged::NeedMore };
802 let line = payload.strip_prefix(b"\x01").unwrap_or(payload);
803 let line = line.strip_suffix(b"\n").unwrap_or(line);
804 if first && line != b"acknowledgments" {
805 return Acknowledged::Whole;
806 }
807 if line == b"ready" {
808 return Acknowledged::Whole;
809 }
810 first = false;
811 position += length;
812 }
813}
814
815/// The answer to a negotiating fetch, with the sections the store sent
816/// after acknowledgments without `ready` left off (see [`acknowledged`]).
817/// Reads only the start of the answer; the rest streams through.
818async fn without_early_pack(mut response: Response) -> Result<Response> {
819 const LOOK: usize = 64 * 1024;
820 let headers = response.headers().clone();
821 headers.delete("content-length")?;
822 let mut stream = response.stream()?;
823 let mut head = Vec::new();
824 loop {
825 match acknowledged(&head) {
826 Acknowledged::CutAt(at) => {
827 head.truncate(at);
828 // A flush ends the acknowledgments and the answer. No
829 // response-end packet: git's HTTP transport adds its own
830 // and refuses one from the server.
831 head.extend_from_slice(b"0000");
832 return Ok(Response::from_bytes(head)?.with_headers(headers));
833 }
834 Acknowledged::Whole => break,
835 Acknowledged::NeedMore if head.len() >= LOOK => break,
836 Acknowledged::NeedMore => match stream.next().await {
837 Some(chunk) => head.extend_from_slice(&chunk?),
838 None => break,
839 },
840 }
841 }
842 let rest = futures_util::stream::once(async move { Ok::<Vec<u8>, worker::Error>(head) }).chain(stream);
843 Ok(Response::from_stream(rest)?.with_headers(headers))
844}
845
846/// Sends the request on to the git store and returns its response as is,
847/// unless it is a push the rules refuse (`rules`, rules.rs), one the
848/// store could not hold (`limits`, pack_limits.rs), or one that `scan`
849/// (push protection) answers itself. A fetch's ref listing has its `HEAD`
850/// pointed at `default_branch` (see [`with_head`]). A POST's body is
851/// `read` when the caller has read it already.
852///
853/// A push is read as it arrives: up to `limits.scan_cap` is kept, to be
854/// scanned and sent on whole; past it, the push is declined, or streamed
855/// to the store unscanned (`LargePushes`), never held. Reads the store
856/// fails for a moment (429, 5xx) are tried again with backoff; a push never
857/// is. A store still busy after that is answered 429 or 503 with
858/// `Retry-After`.
859#[allow(clippy::too_many_arguments)]
860pub async fn forward(
861 mut request: Request,
862 read: Option<Vec<u8>>,
863 git: &GitRequest,
864 access: &GitAccess,
865 rules: impl AsyncFnOnce(&[u8], bool) -> Result<Option<Response>>,
866 default_branch: Option<&str>,
867 limits: PushLimits,
868 scan: impl AsyncFnOnce(&[u8]) -> Result<Option<Response>>,
869) -> Result<Push> {
870 let headers = Headers::new();
871 headers.set("authorization", &format!("Bearer {}", access.token))?;
872 for name in FORWARDED_HEADERS {
873 if let Some(value) = request.headers().get(name)? {
874 headers.set(name, &value)?;
875 }
876 }
877 let query = request.url()?.query().map(|query| format!("?{query}")).unwrap_or_default();
878 let url = format!("{}/{}{query}", access.remote, git.endpoint);
879 let method = request.method();
880 // Its own health and breaker: the fallback store's apart from Artifacts'.
881 let namespace = crate::store::health_namespace(&access.remote);
882
883 if method == Method::Post && git.endpoint == "git-receive-pack" {
884 return push(request, &url, headers, rules, limits, scan, &namespace).await;
885 }
886
887 // A read: the ref advertisement, `ls-refs`, or a fetch of objects.
888 let body = match (&method, read) {
889 (Method::Post, Some(body)) => Some(body),
890 (Method::Post, None) => Some(request.bytes().await?),
891 _ => None,
892 };
893 let lists_head = match &body {
894 None => method == Method::Get && names_head(git, None),
895 Some(body) => names_head(git, Some(body)),
896 };
897 let sent = body.as_ref().map_or(0, |body| body.len() as u64);
898 let mut attempt = 0;
899 let mut response = loop {
900 let mut init = RequestInit::new();
901 init.with_method(method.clone()).with_headers(headers.clone());
902 if let Some(body) = &body {
903 init.with_body(Some(Uint8Array::from(body.as_slice()).into()));
904 }
905 let started = g1t_kit::now_ms();
906 let answered = Fetch::Request(Request::new_with_init(&url, &init)?).send().await;
907 let ms = g1t_kit::now_ms().saturating_sub(started);
908 let failure = match &answered {
909 Ok(response) => resilience::classify_status(response.status_code()),
910 Err(_) => Some(Failure::Transient),
911 };
912 let outcome = match failure {
913 None => meters::Outcome::Ok,
914 Some(Failure::RateLimited) => meters::Outcome::RateLimited,
915 Some(_) => meters::Outcome::Failed,
916 };
917 meters::record_health(&namespace, outcome, ms);
918 match (answered, failure) {
919 (Ok(response), None) => break response,
920 (answered, Some(failure)) if resilience::retry(failure, attempt) => {
921 drop(answered);
922 let wait = resilience::backoff_ms(failure, attempt, worker::js_sys::Math::random());
923 worker::Delay::from(std::time::Duration::from_millis(wait)).await;
924 attempt += 1;
925 }
926 (_, Some(failure)) => {
927 let busy = Busy { rate_limited: failure == Failure::RateLimited, retry_after: 5, read_only: false };
928 return Ok(Push::Forwarded(Forwarded {
929 response: busy_response(busy)?,
930 pushed: Vec::new(),
931 pack_bytes: 0,
932 sent,
933 from_store: false,
934 unscanned: false,
935 }));
936 }
937 (Err(error), None) => return Err(error),
938 }
939 };
940 if let (true, Some(branch)) = (lists_head, default_branch)
941 && response.status_code() == 200
942 {
943 let headers = response.headers().clone();
944 headers.delete("content-length")?;
945 let body = response.bytes().await?;
946 let body = with_head(&body, branch).unwrap_or(body);
947 response = Response::from_bytes(body)?.with_headers(headers);
948 }
949 if git.endpoint == "git-upload-pack"
950 && response.status_code() == 200
951 && body.as_deref().is_some_and(negotiating)
952 {
953 response = without_early_pack(response).await?;
954 }
955 Ok(Push::Forwarded(Forwarded {
956 response,
957 pushed: Vec::new(),
958 pack_bytes: 0,
959 sent,
960 from_store: true,
961 unscanned: false,
962 }))
963}
964
965/// A receive-pack request; see [`forward`].
966#[allow(clippy::too_many_arguments)]
967async fn push(
968 mut request: Request,
969 url: &str,
970 headers: Headers,
971 rules: impl AsyncFnOnce(&[u8], bool) -> Result<Option<Response>>,
972 limits: PushLimits,
973 scan: impl AsyncFnOnce(&[u8]) -> Result<Option<Response>>,
974 namespace: &str,
975) -> Result<Push> {
976 let mut stream = request.stream()?;
977 let mut head: Vec<u8> = Vec::new();
978 let mut sizer = Some(PackSizer::new(limits.max_object));
979 let mut violation = None;
980 let mut ended = false;
981 while head.len() <= limits.scan_cap {
982 match stream.next().await {
983 Some(chunk) => {
984 let chunk = chunk?;
985 if violation.is_none() {
986 violation = check_size(&mut sizer, &chunk, &limits);
987 }
988 head.extend_from_slice(&chunk);
989 }
990 None => {
991 ended = true;
992 break;
993 }
994 }
995 }
996 // The rules of the branches and tags it changes, first: what they
997 // refuse is refused whatever else is wrong with it.
998 if let Some(response) = rules(&head, ended).await? {
999 if !ended {
1000 drain(&mut stream).await?;
1001 }
1002 return Ok(Push::Refused(response));
1003 }
1004 if !ended && limits.large == LargePushes::Refuse && violation.is_none() {
1005 let size = head.len() as u64 + drain(&mut stream).await?;
1006 violation = Some(SizeViolation::Unscannable { size, cap: limits.scan_cap });
1007 ended = true;
1008 }
1009 if let Some(violation) = violation {
1010 if !ended {
1011 drain(&mut stream).await?;
1012 }
1013 let (reason, messages) = size_refusal(&violation);
1014 return Ok(Push::Declined(declined(&head, &reason, &messages)?, reason));
1015 }
1016 let pushed = pushed_branches(&head);
1017 let mut init = RequestInit::new();
1018 init.with_method(Method::Post).with_headers(headers);
1019 let started = g1t_kit::now_ms();
1020 let (answered, pack_bytes, sent, unscanned) = if ended {
1021 if let Some(response) = scan(&head).await? {
1022 return Ok(Push::Blocked(response));
1023 }
1024 let pack = pack_bytes(&head);
1025 let sent = head.len() as u64;
1026 init.with_body(Some(Uint8Array::from(head.as_slice()).into()));
1027 drop(head);
1028 (Fetch::Request(Request::new_with_init(url, &init)?).send().await, pack, sent, false)
1029 } else {
1030 // Larger than can be scanned, and let through unscanned: streamed,
1031 // with the size limits checked as it passes. A violation ends the
1032 // stream before the pack does, so the store refuses it whole.
1033 worker::console_warn!("a push of more than {} bytes goes to the store unscanned", limits.scan_cap);
1034 let commands = head.iter().take(64 * 1024).copied().collect::<Vec<u8>>();
1035 let found: Rc<RefCell<Option<SizeViolation>>> = Rc::default();
1036 let walked = Rc::new(RefCell::new((sizer, 0u64)));
1037 let rest = {
1038 let found = found.clone();
1039 let walked = walked.clone();
1040 stream.map(move |chunk| {
1041 let chunk = chunk?;
1042 let mut walked = walked.borrow_mut();
1043 walked.1 += chunk.len() as u64;
1044 if let Some(violation) = check_size(&mut walked.0, &chunk, &limits) {
1045 *found.borrow_mut() = Some(violation);
1046 return Err(worker::Error::RustError("push over the size limit".into()));
1047 }
1048 Ok(chunk)
1049 })
1050 };
1051 let first = head.len() as u64;
1052 let body = futures_util::stream::once(async move { Ok::<Vec<u8>, worker::Error>(head) }).chain(rest);
1053 init.with_body(Some(stream_body(body)?));
1054 let answered = Fetch::Request(Request::new_with_init(url, &init)?).send().await;
1055 if let Some(violation) = found.borrow_mut().take() {
1056 let (reason, messages) = size_refusal(&violation);
1057 return Ok(Push::Declined(declined(&commands, &reason, &messages)?, reason));
1058 }
1059 let walked = walked.borrow();
1060 let pack = walked.0.as_ref().map_or_else(|| pack_bytes(&commands), PackSizer::pack_bytes);
1061 (answered, pack, first + walked.1, true)
1062 };
1063 let ms = g1t_kit::now_ms().saturating_sub(started);
1064 let failure = match &answered {
1065 Ok(response) => resilience::classify_status(response.status_code()),
1066 Err(_) => Some(Failure::Transient),
1067 };
1068 meters::record_health(
1069 namespace,
1070 match failure {
1071 None => meters::Outcome::Ok,
1072 Some(Failure::RateLimited) => meters::Outcome::RateLimited,
1073 Some(_) => meters::Outcome::Failed,
1074 },
1075 ms,
1076 );
1077 // A push is never tried again: the store may have taken it.
1078 let response = match (answered, failure) {
1079 (Ok(response), None) => response,
1080 (Ok(response), Some(Failure::RateLimited)) => {
1081 drop(response);
1082 busy_response(Busy { rate_limited: true, retry_after: 5, read_only: false })?
1083 }
1084 (Ok(response), Some(_)) => response,
1085 (Err(error), _) => {
1086 worker::console_error!("a push did not reach the store: {error}");
1087 busy_response(Busy { rate_limited: false, retry_after: 5, read_only: false })?
1088 }
1089 };
1090 Ok(Push::Forwarded(Forwarded {
1091 response,
1092 pushed,
1093 pack_bytes,
1094 sent,
1095 from_store: true,
1096 unscanned,
1097 }))
1098}
1099
1100#[cfg(test)]
1101mod tests {
1102 use super::{Acknowledged, GitService, Pushed, RepoPath, Url, ZERO_ID, acknowledged, framed, negotiating, pack_bytes, parse, pushed_branches, refusal, server_timing, transferred, with_head, with_namespace};
1103
1104 #[test]
1105 fn server_timing_names_each_step_and_the_total() {
1106 assert_eq!(
1107 server_timing(&[("repo", 12), ("token", 0), ("store", 140)], &[], 153),
1108 "repo;dur=12, token;dur=0, store;dur=140, total;dur=153"
1109 );
1110 assert_eq!(server_timing(&[], &[], 3), "total;dur=3");
1111 assert_eq!(
1112 server_timing(&[("repo", 1), ("cache", 2)], &[("refs", "hit-colo")], 4),
1113 "repo;dur=1, cache;dur=2, refs;desc=hit-colo, total;dur=4"
1114 );
1115 }
1116
1117 #[test]
1118 fn a_renamed_repository_redirects_to_its_new_name() {
1119 // A rename keeps the old path in the same table as a transfer, so
1120 // the old remote is sent to the new name the same way.
1121 let to = RepoPath {
1122 namespace: "acme".into(),
1123 name: "booster".into(),
1124 };
1125 let url = Url::parse("https://g1t.sh/acme/rocket.git/info/refs?service=git-upload-pack").unwrap();
1126 assert_eq!(
1127 transferred(&url, &to).as_deref(),
1128 Some("https://g1t.sh/acme/booster.git/info/refs?service=git-upload-pack")
1129 );
1130 }
1131
1132 #[test]
1133 fn head_follows_the_default_branch_in_a_v0_advertisement() {
1134 let main = "1111111111111111111111111111111111111111";
1135 let trunk = "2222222222222222222222222222222222222222";
1136 let body = [
1137 pkt("# service=git-upload-pack\n"),
1138 b"0000".to_vec(),
1139 pkt(&format!("{main} HEAD\0multi_ack symref=HEAD:refs/heads/main agent=git/2\n")),
1140 pkt(&format!("{main} refs/heads/main\n")),
1141 pkt(&format!("{trunk} refs/heads/trunk\n")),
1142 b"0000".to_vec(),
1143 ]
1144 .concat();
1145 let changed = String::from_utf8(with_head(&body, "trunk").unwrap()).unwrap();
1146 assert!(changed.contains(&format!("{trunk} HEAD\0multi_ack symref=HEAD:refs/heads/trunk agent=git/2\n")));
1147 assert!(changed.contains(&format!("{main} refs/heads/main\n")));
1148 assert!(changed.starts_with("001e# service=git-upload-pack\n0000"));
1149 // Already right, or a branch it does not have: left alone.
1150 assert!(with_head(&body, "main").is_none());
1151 assert!(with_head(&body, "gone").is_none());
1152 }
1153
1154 #[test]
1155 fn head_follows_the_default_branch_in_a_v2_listing() {
1156 let main = "1111111111111111111111111111111111111111";
1157 let trunk = "2222222222222222222222222222222222222222";
1158 let body = [
1159 pkt(&format!("{main} HEAD symref-target:refs/heads/main\n")),
1160 pkt(&format!("{main} refs/heads/main\n")),
1161 pkt(&format!("{trunk} refs/heads/trunk\n")),
1162 b"0000".to_vec(),
1163 ]
1164 .concat();
1165 let changed = String::from_utf8(with_head(&body, "trunk").unwrap()).unwrap();
1166 assert!(changed.starts_with(&String::from_utf8(pkt(&format!("{trunk} HEAD symref-target:refs/heads/trunk\n"))).unwrap()));
1167 assert!(changed.ends_with("0000"));
1168 // Without symrefs asked for, only the commit changes.
1169 let plain = [pkt(&format!("{main} HEAD\n")), pkt(&format!("{trunk} refs/heads/trunk\n")), b"0000".to_vec()].concat();
1170 let changed = String::from_utf8(with_head(&plain, "trunk").unwrap()).unwrap();
1171 assert!(changed.starts_with(&format!("0032{trunk} HEAD\n")));
1172 }
1173
1174 #[test]
1175 fn a_push_is_measured_by_the_pack_after_its_commands() {
1176 let old = "c71546fcd893ef8b0f57388b65e620d759705dda";
1177 let new = "4807077b296e6edbf410d55e72749d3e1170c291";
1178 let pack = b"PACK\0\0\0\x02\0\0\0\0rest-of-pack";
1179 let body = [
1180 pkt(&format!("{old} {new} refs/heads/main\0 report-status\n")),
1181 b"0000".to_vec(),
1182 pack.to_vec(),
1183 ]
1184 .concat();
1185 assert_eq!(pack_bytes(&body), pack.len() as u64);
1186 // Only deletions: no pack.
1187 let body = [pkt(&format!("{old} {ZERO_ID} refs/heads/gone\n")), b"0000".to_vec()].concat();
1188 assert_eq!(pack_bytes(&body), 0);
1189 assert_eq!(pack_bytes(b"garbage"), 0);
1190 }
1191
1192 #[test]
1193 fn a_transferred_repository_keeps_the_rest_of_the_address() {
1194 let to = RepoPath {
1195 namespace: "flagon-io".into(),
1196 name: "g1t".into(),
1197 };
1198 let url = Url::parse("https://g1t.sh/syntaqx/g1t.git/info/refs?service=git-receive-pack").unwrap();
1199 assert_eq!(
1200 transferred(&url, &to).as_deref(),
1201 Some("https://g1t.sh/flagon-io/g1t.git/info/refs?service=git-receive-pack")
1202 );
1203 let url = Url::parse("https://g1t.sh/syntaqx/g1t/git-upload-pack").unwrap();
1204 assert_eq!(
1205 transferred(&url, &to).as_deref(),
1206 Some("https://g1t.sh/flagon-io/g1t/git-upload-pack")
1207 );
1208 }
1209
1210 #[test]
1211 fn an_alias_is_answered_as_its_workspaces_repository() {
1212 for (address, service) in [
1213 ("https://g1t.sh/g1t/g1t.git/info/refs?service=git-upload-pack", GitService::UploadPack),
1214 ("https://g1t.sh/g1t/g1t.git/git-receive-pack", GitService::ReceivePack),
1215 ("https://g1t.sh/g1t/g1t/git-upload-pack", GitService::UploadPack),
1216 ] {
1217 let git = parse(&Url::parse(address).unwrap()).unwrap();
1218 assert_eq!(git.path.namespace, "g1t", "{address}");
1219 let canonical = git.under("flagon-io");
1220 assert_eq!(
1221 canonical.path,
1222 RepoPath {
1223 namespace: "flagon-io".into(),
1224 name: "g1t".into(),
1225 },
1226 "{address}"
1227 );
1228 assert_eq!(canonical.service, service);
1229 assert_eq!(canonical.endpoint, git.endpoint);
1230 }
1231 }
1232
1233 #[test]
1234 fn a_renamed_workspace_keeps_the_rest_of_the_address() {
1235 let url = worker::Url::parse(
1236 "https://g1t.sh/acme/rocket.git/info/refs?service=git-upload-pack",
1237 )
1238 .unwrap();
1239 assert_eq!(
1240 with_namespace(&url, "acme-inc").as_deref(),
1241 Some("https://g1t.sh/acme-inc/rocket.git/info/refs?service=git-upload-pack")
1242 );
1243 let bare = worker::Url::parse("https://g1t.sh/acme").unwrap();
1244 assert_eq!(with_namespace(&bare, "acme-inc"), None);
1245 }
1246
1247 fn pkt(payload: &str) -> Vec<u8> {
1248 format!("{:04x}{payload}", payload.len() + 4).into_bytes()
1249 }
1250
1251 fn joined(parts: &[&[u8]]) -> Vec<u8> {
1252 parts.concat()
1253 }
1254
1255 #[test]
1256 fn a_fetch_with_haves_and_no_done_is_negotiating() {
1257 let request = |lines: &[&str]| {
1258 let mut body = joined(&[&pkt("command=fetch\n"), &pkt("object-format=sha1\n"), b"0001"]);
1259 for line in lines {
1260 body.extend(pkt(&format!("{line}\n")));
1261 }
1262 body.extend(b"0000");
1263 body
1264 };
1265 let want = "want 8407eba58b925619274d012258c2b474a5dbf012";
1266 let have = "have 55cd670a89a80df4fa9d9f0244c44fbd2ed1db8b";
1267 assert!(negotiating(&request(&["deepen 1", want, have])));
1268 assert!(!negotiating(&request(&[want, have, "done"])));
1269 assert!(!negotiating(&request(&[want, "done"])));
1270 let ls_refs = joined(&[&pkt("command=ls-refs\n"), b"0001", &pkt("have nothing\n"), b"0000"]);
1271 assert!(!negotiating(&ls_refs));
1272 }
1273
1274 #[test]
1275 fn acknowledgments_without_ready_end_the_answer() {
1276 // What the store sent a shallow fetch whose only `have` it did not
1277 // know, sideband-all framed: a NAK, then a pack anyway.
1278 let answer = joined(&[
1279 &pkt("\x01acknowledgments\n"),
1280 &pkt("\x01NAK\n"),
1281 b"0001",
1282 &pkt("\x01shallow-info\n"),
1283 &pkt("\x01shallow 8407eba58b925619274d012258c2b474a5dbf012\n"),
1284 b"0001",
1285 &pkt("\x01packfile\n"),
1286 ]);
1287 let cut = joined(&[&pkt("\x01acknowledgments\n"), &pkt("\x01NAK\n")]).len();
1288 assert_eq!(acknowledged(&answer), Acknowledged::CutAt(cut));
1289 // Not yet at the section's end.
1290 assert_eq!(acknowledged(&answer[..cut - 2]), Acknowledged::NeedMore);
1291 assert_eq!(acknowledged(&answer[..cut]), Acknowledged::NeedMore);
1292 // Without sideband framing too.
1293 let plain = joined(&[&pkt("acknowledgments\n"), &pkt("ACK abc\n"), b"0001", &pkt("packfile\n")]);
1294 assert!(matches!(acknowledged(&plain), Acknowledged::CutAt(_)));
1295 }
1296
1297 #[test]
1298 fn a_ready_store_or_a_plain_pack_streams_through() {
1299 let ready = joined(&[
1300 &pkt("\x01acknowledgments\n"),
1301 &pkt("\x01ACK bab14ff1b6d9c4918100098009747d776759a967\n"),
1302 &pkt("\x01ready\n"),
1303 b"0001",
1304 &pkt("\x01packfile\n"),
1305 ]);
1306 assert_eq!(acknowledged(&ready), Acknowledged::Whole);
1307 // Acknowledgments only, ended by a flush: already right.
1308 let only = joined(&[&pkt("acknowledgments\n"), &pkt("NAK\n"), b"0000"]);
1309 assert_eq!(acknowledged(&only), Acknowledged::Whole);
1310 let pack = joined(&[&pkt("\x01packfile\n"), b"0000"]);
1311 assert_eq!(acknowledged(&pack), Acknowledged::Whole);
1312 assert_eq!(acknowledged(b"00"), Acknowledged::NeedMore);
1313 }
1314
1315 #[test]
1316 fn pushed_branches_are_read_from_the_commands() {
1317 let old = "c71546fcd893ef8b0f57388b65e620d759705dda";
1318 let new = "4807077b296e6edbf410d55e72749d3e1170c291";
1319 let body = [
1320 pkt(&format!(
1321 "{old} {new} refs/heads/main\0 report-status side-band-64k\n"
1322 )),
1323 pkt(&format!("{ZERO_ID} {new} refs/heads/feature/x\n")),
1324 pkt(&format!("{old} {ZERO_ID} refs/heads/gone\n")),
1325 pkt(&format!("{ZERO_ID} {new} refs/tags/v1\n")),
1326 b"0000".to_vec(),
1327 b"PACK\0\0\0\x02\0\0\0\0".to_vec(),
1328 ]
1329 .concat();
1330 assert_eq!(
1331 pushed_branches(&body),
1332 [
1333 Pushed {
1334 git_ref: "refs/heads/main".to_owned(),
1335 before: Some(old.to_owned()),
1336 after: new.to_owned()
1337 },
1338 Pushed {
1339 git_ref: "refs/heads/feature/x".to_owned(),
1340 before: None,
1341 after: new.to_owned()
1342 },
1343 Pushed {
1344 git_ref: "refs/tags/v1".to_owned(),
1345 before: None,
1346 after: new.to_owned()
1347 },
1348 ]
1349 );
1350 }
1351
1352 #[test]
1353 fn a_push_to_a_protected_branch_is_declined_with_the_reason() {
1354 let old = "c71546fcd893ef8b0f57388b65e620d759705dda";
1355 let new = "4807077b296e6edbf410d55e72749d3e1170c291";
1356 let body = [
1357 pkt(&format!("{old} {new} refs/heads/main\0 report-status\n")),
1358 pkt(&format!("{ZERO_ID} {new} refs/heads/feature\n")),
1359 b"0000".to_vec(),
1360 ]
1361 .concat();
1362 let report = String::from_utf8(refusal(&body, "main").unwrap()).unwrap();
1363 assert!(report.starts_with("000eunpack ok\n"));
1364 assert!(report.contains("ng refs/heads/main main is protected"));
1365 assert!(report.contains("ng refs/heads/feature not pushed"));
1366 assert!(report.ends_with("0000"));
1367 }
1368
1369 #[test]
1370 fn the_report_is_framed_for_a_client_that_asked_for_side_band() {
1371 let old = "c71546fcd893ef8b0f57388b65e620d759705dda";
1372 let body = [
1373 pkt(&format!(
1374 "{old} {ZERO_ID} refs/heads/main\0 report-status side-band-64k\n"
1375 )),
1376 b"0000".to_vec(),
1377 ]
1378 .concat();
1379 let report = refusal(&body, "main").unwrap();
1380 // A length, then channel 1, then the report itself.
1381 assert_eq!(report[4], 1);
1382 assert_eq!(&report[5..18], b"000eunpack ok");
1383 assert!(report.ends_with(b"00000000"));
1384 }
1385
1386 #[test]
1387 fn other_branches_and_a_first_push_are_let_through() {
1388 let old = "c71546fcd893ef8b0f57388b65e620d759705dda";
1389 let new = "4807077b296e6edbf410d55e72749d3e1170c291";
1390 let feature = [
1391 pkt(&format!("{old} {new} refs/heads/feature\0 report-status\n")),
1392 b"0000".to_vec(),
1393 ]
1394 .concat();
1395 assert!(refusal(&feature, "main").is_none());
1396 // An empty repository has to be able to receive its first commits.
1397 let first = [
1398 pkt(&format!(
1399 "{ZERO_ID} {new} refs/heads/main\0 report-status\n"
1400 )),
1401 b"0000".to_vec(),
1402 ]
1403 .concat();
1404 assert!(refusal(&first, "main").is_none());
1405 }
1406
1407 #[test]
1408 fn a_blocked_push_explains_itself_on_the_progress_channel() {
1409 let report = b"000eunpack ok\n0000".to_vec();
1410 let messages = vec!["g1t found a secret in this push, so nothing was pushed.".to_owned()];
1411 let body = framed(report.clone(), "report-status side-band-64k", &messages);
1412 // Channel 2 first, which git prints as `remote:` lines.
1413 assert_eq!(body[4], 2);
1414 assert!(String::from_utf8_lossy(&body).contains("so nothing was pushed.\n"));
1415 let at = body.windows(5).position(|w| w == b"000eu").unwrap();
1416 assert_eq!(body[at - 1], 1);
1417 assert!(body.ends_with(b"0000"));
1418 // A client without side-band gets the bare report.
1419 assert_eq!(framed(report.clone(), "report-status", &messages), report);
1420 }
1421
1422 #[test]
1423 fn a_fetch_request_names_no_branches() {
1424 assert!(
1425 pushed_branches(b"0032want c71546fcd893ef8b0f57388b65e620d759705dda\n0000").is_empty()
1426 );
1427 }
1428}