Sign in

flagon-io/g1t

public

g1t/services/repos/src/git_http.rs

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