anvilsign in

collin/anvil

1// This Source Code Form is subject to the terms of the Mozilla Public
2// License, v. 2.0. If a copy of the MPL was not distributed with this
3// file, You can obtain one at https://mozilla.org/MPL/2.0/.
4//
5// Copyright (c) 2026 WJQSERVER
6
7use std::io::{BufRead, BufReader, Read};
8use std::path::Path;
9use std::sync::atomic::AtomicBool;
10
11use gix::objs::bstr::BString;
12use gix::prelude::ObjectIdExt;
13use gix::progress::Discard;
14use gix::refs::Target;
15use gix::refs::transaction::{Change, LogChange, PreviousValue, RefEdit, RefLog};
16
17use crate::error::{Error, Result};
18use crate::pktline;
19
20const ZERO_ID: &str = "0000000000000000000000000000000000000000";
21const CAPABILITIES: &str = concat!(
22 "report-status report-status-v2 side-band-64k quiet ofs-delta object-format=sha1 agent=gitserver/",
23 env!("CARGO_PKG_VERSION")
24);
25
26pub fn advertise_receive_refs(repo_path: &Path) -> Result<Vec<u8>> {
27 let repo = gix::open(repo_path)?;
28 let mut out = Vec::new();
29 let head_name = repo.head_name().ok().flatten();
30 let mut refs: Vec<(String, gix::ObjectId)> = repo
31 .references()
32 .map_err(|e| Error::Protocol(format!("failed to open refs: {e}")))?
33 .all()
34 .map_err(|e| Error::Protocol(format!("failed to iterate refs: {e}")))?
35 .flatten()
36 .filter_map(|mut reference| {
37 reference
38 .peel_to_id()
39 .ok()
40 .map(|id| (reference.name().as_bstr().to_string(), id.detach()))
41 })
42 .collect();
43 refs.sort_by(|a, b| a.0.cmp(&b.0));
44
45 if refs.is_empty() {
46 out.extend_from_slice(&pktline::encode(
47 format!("{ZERO_ID} capabilities^{{}}\0{CAPABILITIES}\n").as_bytes(),
48 ));
49 } else {
50 let (first_name, first_id) = &refs[0];
51 let mut first = format!("{} {}\0{CAPABILITIES}", first_id, first_name);
52 if head_name
53 .as_ref()
54 .is_some_and(|head| head.as_bstr() == first_name.as_str())
55 {
56 first.push_str(&format!(" symref=HEAD:{first_name}"));
57 }
58 first.push('\n');
59 out.extend_from_slice(&pktline::encode(first.as_bytes()));
60
61 for (name, id) in refs.into_iter().skip(1) {
62 out.extend_from_slice(&pktline::encode(format!("{id} {name}\n").as_bytes()));
63 }
64 }
65
66 out.extend_from_slice(pktline::flush());
67 Ok(out)
68}
69
70pub fn receive_pack<R: Read>(repo_path: &Path, request: R) -> Result<Vec<u8>> {
71 let interrupt = AtomicBool::new(false);
72 receive_pack_with_interrupt(repo_path, request, &interrupt)
73}
74
75pub fn receive_pack_with_interrupt<R: Read>(
76 repo_path: &Path,
77 request: R,
78 interrupt: &AtomicBool,
79) -> Result<Vec<u8>> {
80 let repo = gix::open(repo_path)?;
81 let mut parsed = parse_request(request, interrupt)?;
82 let status = apply_commands(&repo, repo_path, &mut parsed, interrupt)?;
83 Ok(encode_report_status(&parsed.capabilities, &status))
84}
85
86#[derive(Default)]
87struct ReceivePackCapabilities {
88 report_status: bool,
89 report_status_v2: bool,
90}
91
92struct ReceivePackRequest<R> {
93 commands: Vec<UpdateCommand>,
94 pack: R,
95 capabilities: ReceivePackCapabilities,
96}
97
98struct UpdateCommand {
99 old_id: String,
100 new_id: String,
101 refname: String,
102}
103
104enum CommandStatus {
105 Ok(String),
106 Ng(String, String),
107}
108
109fn parse_request<R: Read>(
110 request: R,
111 interrupt: &AtomicBool,
112) -> Result<ReceivePackRequest<BufReader<R>>> {
113 let mut request = BufReader::new(request);
114 let mut commands = Vec::new();
115 let mut capabilities = ReceivePackCapabilities::default();
116
117 loop {
118 check_interrupt(interrupt)?;
119 let mut prefix = [0u8; 4];
120 match request.read_exact(&mut prefix) {
121 Ok(()) => {}
122 Err(err) if err.kind() == std::io::ErrorKind::UnexpectedEof => break,
123 Err(err) => return Err(Error::Io(err)),
124 }
125
126 let len_str = std::str::from_utf8(&prefix)
127 .map_err(|_| Error::Protocol("invalid pkt-line length prefix".into()))?;
128
129 if len_str == "0000" {
130 break;
131 }
132
133 let len = usize::from_str_radix(len_str, 16)
134 .map_err(|_| Error::Protocol("invalid pkt-line length".into()))?;
135 if len < 4 {
136 return Err(Error::Protocol("invalid pkt-line frame length".into()));
137 }
138
139 check_interrupt(interrupt)?;
140 let mut payload = vec![0u8; len - 4];
141 request.read_exact(&mut payload)?;
142
143 let (command_bytes, capability_bytes) =
144 if let Some(nul) = payload.iter().position(|b| *b == 0) {
145 (&payload[..nul], Some(&payload[nul + 1..]))
146 } else {
147 (&payload[..], None)
148 };
149
150 if let Some(capability_bytes) = capability_bytes {
151 let capabilities_line = std::str::from_utf8(capability_bytes)
152 .map_err(|_| Error::Protocol("invalid UTF-8 in receive-pack capabilities".into()))?
153 .trim_end_matches('\n');
154 for capability in capabilities_line.split_ascii_whitespace() {
155 match capability {
156 "report-status" => capabilities.report_status = true,
157 "report-status-v2" => {
158 capabilities.report_status = true;
159 capabilities.report_status_v2 = true;
160 }
161 _ => {}
162 }
163 }
164 }
165
166 let line = std::str::from_utf8(command_bytes)
167 .map_err(|_| Error::Protocol("invalid UTF-8 in update command".into()))?
168 .trim_end_matches('\n');
169 let mut parts = line.split_ascii_whitespace();
170 let Some(old_id) = parts.next() else { continue };
171 let Some(new_id) = parts.next() else { continue };
172 let Some(refname) = parts.next() else {
173 continue;
174 };
175
176 commands.push(UpdateCommand {
177 old_id: old_id.to_owned(),
178 new_id: new_id.to_owned(),
179 refname: refname.to_owned(),
180 });
181 }
182
183 Ok(ReceivePackRequest {
184 commands,
185 pack: request,
186 capabilities,
187 })
188}
189
190fn apply_commands<R: BufRead>(
191 repo: &gix::Repository,
192 repo_path: &Path,
193 request: &mut ReceivePackRequest<R>,
194 interrupt: &AtomicBool,
195) -> Result<Vec<CommandStatus>> {
196 check_interrupt(interrupt)?;
197 if request.pack.fill_buf().map(|buf: &[u8]| !buf.is_empty())? {
198 write_pack(repo_path, &mut request.pack, interrupt)?;
199 }
200
201 let mut edits = Vec::with_capacity(request.commands.len());
202 for (index, command) in request.commands.iter().enumerate() {
203 check_interrupt(interrupt)?;
204 match validate_ref_update(repo, command, interrupt) {
205 Ok(edit) => edits.push((command.refname.clone(), edit)),
206 Err(err) => {
207 return Ok(request
208 .commands
209 .iter()
210 .enumerate()
211 .map(|(cmd_index, cmd)| {
212 if cmd_index == index {
213 CommandStatus::Ng(cmd.refname.clone(), err.to_string())
214 } else {
215 CommandStatus::Ng(
216 cmd.refname.clone(),
217 "transaction aborted due to another command failing validation"
218 .into(),
219 )
220 }
221 })
222 .collect());
223 }
224 }
225 }
226
227 check_interrupt(interrupt)?;
228 match repo.edit_references(edits.into_iter().map(|(_, edit)| edit)) {
229 Ok(_) => Ok(request
230 .commands
231 .iter()
232 .map(|cmd| CommandStatus::Ok(cmd.refname.clone()))
233 .collect()),
234 Err(err) => Ok(request
235 .commands
236 .iter()
237 .map(|cmd| CommandStatus::Ng(cmd.refname.clone(), format!("transaction failed: {err}")))
238 .collect()),
239 }
240}
241
242fn write_pack<R: BufRead>(repo_path: &Path, pack: &mut R, interrupt: &AtomicBool) -> Result<()> {
243 let mut progress = Discard;
244 let outcome = gix_pack::Bundle::write_to_directory(
245 pack,
246 Some(repo_path.join("objects/pack").as_path()),
247 &mut progress,
248 interrupt,
249 None::<&gix::Repository>,
250 Default::default(),
251 );
252 if interrupt.load(std::sync::atomic::Ordering::Relaxed) {
253 return Err(Error::Io(std::io::Error::new(
254 std::io::ErrorKind::TimedOut,
255 "receive-pack timed out",
256 )));
257 }
258 let outcome =
259 outcome.map_err(|e| Error::Protocol(format!("failed to write incoming pack: {e}")))?;
260
261 if let Some(keep) = outcome.keep_path {
262 let _ = std::fs::remove_file(keep);
263 }
264 Ok(())
265}
266
267fn check_interrupt(interrupt: &AtomicBool) -> Result<()> {
268 if interrupt.load(std::sync::atomic::Ordering::Relaxed) {
269 Err(Error::Io(std::io::Error::new(
270 std::io::ErrorKind::TimedOut,
271 "receive-pack timed out",
272 )))
273 } else {
274 Ok(())
275 }
276}
277
278fn validate_ref_update(
279 repo: &gix::Repository,
280 command: &UpdateCommand,
281 interrupt: &AtomicBool,
282) -> Result<RefEdit> {
283 if command.new_id == ZERO_ID {
284 return Err(Error::Protocol(format!(
285 "deletion prohibited for {}",
286 command.refname
287 )));
288 }
289
290 let is_branch = command.refname.starts_with("refs/heads/");
291 let is_tag = command.refname.starts_with("refs/tags/");
292 let new_id = gix::ObjectId::from_hex(command.new_id.as_bytes())
293 .map_err(|_| Error::Protocol(format!("invalid new object id: {}", command.new_id)))?;
294 let new_header = repo
295 .find_header(new_id)
296 .map_err(|e| Error::Protocol(format!("missing new object {}: {e}", command.new_id)))?;
297 if is_branch && new_header.kind() != gix::objs::Kind::Commit {
298 return Err(Error::Protocol(format!(
299 "updates to {} must point to a commit",
300 command.refname
301 )));
302 }
303
304 let name: gix::refs::FullName = command
305 .refname
306 .as_str()
307 .try_into()
308 .map_err(|e| Error::Protocol(format!("invalid ref name {}: {e}", command.refname)))?;
309
310 let (expected, log_message) = if command.old_id == ZERO_ID {
311 (PreviousValue::MustNotExist, BString::from("push create"))
312 } else {
313 if is_tag {
314 return Err(Error::Protocol(format!(
315 "updating existing tag {} is not allowed",
316 command.refname
317 )));
318 }
319
320 let old_id = gix::ObjectId::from_hex(command.old_id.as_bytes())
321 .map_err(|_| Error::Protocol(format!("invalid old object id: {}", command.old_id)))?;
322 if is_branch {
323 ensure_fast_forward(repo, old_id, new_id, &command.refname, interrupt)?;
324 }
325 (
326 PreviousValue::MustExistAndMatch(Target::Object(old_id)),
327 BString::from("push"),
328 )
329 };
330
331 Ok(RefEdit {
332 change: Change::Update {
333 log: LogChange {
334 mode: RefLog::AndReference,
335 force_create_reflog: false,
336 message: log_message,
337 },
338 expected,
339 new: Target::Object(new_id),
340 },
341 name,
342 deref: false,
343 })
344}
345
346fn ensure_fast_forward(
347 repo: &gix::Repository,
348 old_id: gix::ObjectId,
349 new_id: gix::ObjectId,
350 refname: &str,
351 interrupt: &AtomicBool,
352) -> Result<()> {
353 check_interrupt(interrupt)?;
354 if old_id == new_id {
355 return Ok(());
356 }
357
358 let old_commit_time = repo
359 .find_object(old_id)
360 .map_err(|e| Error::Protocol(format!("failed to inspect current tip for {refname}: {e}")))?
361 .try_into_commit()
362 .map_err(|_| Error::Protocol(format!("current tip of {refname} is not a commit")))?
363 .committer()
364 .map_err(|e| Error::Protocol(format!("failed to read commit metadata for {refname}: {e}")))?
365 .seconds();
366
367 let ancestors = new_id
368 .attach(repo)
369 .ancestors()
370 .sorting(gix::revision::walk::Sorting::ByCommitTimeCutoff {
371 order: Default::default(),
372 seconds: old_commit_time,
373 })
374 .all()
375 .map_err(|e| Error::Protocol(format!("failed to walk commits for {refname}: {e}")))?;
376
377 for id in ancestors {
378 check_interrupt(interrupt)?;
379 if id.is_ok_and(|commit| commit.id == old_id) {
380 return Ok(());
381 }
382 }
383
384 Err(Error::Protocol(format!(
385 "non-fast-forward update to {refname} is not allowed"
386 )))
387}
388
389fn encode_report_status(
390 capabilities: &ReceivePackCapabilities,
391 statuses: &[CommandStatus],
392) -> Vec<u8> {
393 if !capabilities.report_status {
394 return pktline::flush().to_vec();
395 }
396
397 let mut status_lines = Vec::new();
398 status_lines.extend_from_slice(&pktline::encode(b"unpack ok\n"));
399
400 for status in statuses {
401 match status {
402 CommandStatus::Ok(refname) => {
403 status_lines
404 .extend_from_slice(&pktline::encode(format!("ok {refname}\n").as_bytes()));
405 }
406 CommandStatus::Ng(refname, message) => {
407 status_lines.extend_from_slice(&pktline::encode(
408 format!("ng {refname} {message}\n").as_bytes(),
409 ));
410 }
411 }
412 }
413 status_lines.extend_from_slice(pktline::flush());
414
415 if capabilities.report_status_v2 {
416 let mut sideband = Vec::new();
417 const MAX_BAND_PAYLOAD: usize = 65519;
418 for chunk in status_lines.chunks(MAX_BAND_PAYLOAD) {
419 let len = 4 + 1 + chunk.len();
420 sideband.extend_from_slice(format!("{len:04x}").as_bytes());
421 sideband.push(0x01);
422 sideband.extend_from_slice(chunk);
423 }
424 sideband.extend_from_slice(pktline::flush());
425 sideband
426 } else {
427 status_lines
428 }
429}
430
431#[cfg(test)]
432mod tests {
433 use std::process::Command;
434
435 use tempfile::TempDir;
436
437 use super::*;
438
439 fn create_repo_with_commit(root: &std::path::Path) -> std::path::PathBuf {
440 let repo_path = root.join("test.git");
441 let work_dir = root.join("work");
442 std::fs::create_dir(&work_dir).unwrap();
443 Command::new("git")
444 .args(["init", "--bare", repo_path.to_str().unwrap()])
445 .output()
446 .unwrap();
447 Command::new("git")
448 .args(["symbolic-ref", "HEAD", "refs/heads/main"])
449 .current_dir(&repo_path)
450 .output()
451 .unwrap();
452 Command::new("git")
453 .args([
454 "clone",
455 repo_path.to_str().unwrap(),
456 work_dir.to_str().unwrap(),
457 ])
458 .output()
459 .unwrap();
460 Command::new("git")
461 .current_dir(&work_dir)
462 .args(["commit", "--allow-empty", "-m", "init"])
463 .env("GIT_AUTHOR_NAME", "Test")
464 .env("GIT_AUTHOR_EMAIL", "t@t.com")
465 .env("GIT_COMMITTER_NAME", "Test")
466 .env("GIT_COMMITTER_EMAIL", "t@t.com")
467 .output()
468 .unwrap();
469 Command::new("git")
470 .current_dir(&work_dir)
471 .args(["push", "origin", "main"])
472 .output()
473 .unwrap();
474 repo_path
475 }
476
477 #[test]
478 fn advertise_receive_pack_refs() {
479 let root = TempDir::new().unwrap();
480 let repo_path = create_repo_with_commit(root.path());
481 let output = advertise_receive_refs(&repo_path).unwrap();
482 let output_str = String::from_utf8_lossy(&output);
483 assert!(output_str.contains("refs/heads/main"));
484 assert!(output_str.contains("report-status"));
485 }
486
487 #[test]
488 fn parse_receive_pack_request_with_capabilities() {
489 let payload = b"0000000000000000000000000000000000000000 1111111111111111111111111111111111111111 refs/heads/main\0 report-status-v2 side-band-64k\n";
490 let mut body = format!("{:04x}", payload.len() + 4).into_bytes();
491 body.extend_from_slice(payload);
492 body.extend_from_slice(b"0000PACK");
493
494 let interrupt = AtomicBool::new(false);
495 let parsed = parse_request(std::io::Cursor::new(&body), &interrupt).unwrap();
496 assert_eq!(parsed.commands.len(), 1);
497 assert!(parsed.capabilities.report_status);
498 assert!(parsed.capabilities.report_status_v2);
499 let mut pack = String::new();
500 let mut reader = parsed.pack;
501 reader.read_to_string(&mut pack).unwrap();
502 assert_eq!(pack.as_bytes(), b"PACK");
503 }
504
505 #[test]
506 fn branch_updates_require_commit_target() {
507 let root = TempDir::new().unwrap();
508 let repo_path = create_repo_with_commit(root.path());
509 let repo = gix::open(repo_path).unwrap();
510 let tree_id = Command::new("git")
511 .args(["rev-parse", "HEAD^{tree}"])
512 .current_dir(root.path().join("work"))
513 .output()
514 .unwrap();
515 let tree_id = String::from_utf8(tree_id.stdout)
516 .unwrap()
517 .trim()
518 .to_string();
519
520 let err = validate_ref_update(
521 &repo,
522 &UpdateCommand {
523 old_id: ZERO_ID.into(),
524 new_id: tree_id,
525 refname: "refs/heads/feature".into(),
526 },
527 &AtomicBool::new(false),
528 )
529 .unwrap_err();
530
531 assert!(err.to_string().contains("must point to a commit"));
532 }
533
534 #[test]
535 fn ensure_fast_forward_respects_interrupt() {
536 let root = TempDir::new().unwrap();
537 let repo_path = create_repo_with_commit(root.path());
538 let repo = gix::open(repo_path).unwrap();
539 let head = repo.head_id().unwrap().detach();
540 let interrupt = AtomicBool::new(true);
541
542 let err =
543 ensure_fast_forward(&repo, head, head, "refs/heads/main", &interrupt).unwrap_err();
544 match err {
545 Error::Io(inner) => assert_eq!(inner.kind(), std::io::ErrorKind::TimedOut),
546 other => panic!("expected timeout io error, got {other}"),
547 }
548 }
549}