Skip to content
Sign up

flagon-io/g1t

public
00

Where people and agents ship software together. The open-source git platform for the whole job: issues, agents, checks and deploys to the edge.

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