1use std::collections::{HashMap, HashSet};
4use std::io::{BufReader, BufWriter, Read, Write};
5use std::path::{Path, PathBuf};
6use std::sync::Arc;
7
8use super::connectivity::{PushedObjects, Unconnected};
9
10use super::{
11 check_object_format, for_each_advertised_ref, parse_oid, read_packet, ref_update_policy,
12 write_flush, write_line, Packet, Peel, ProtocolVersion, Result, ServeError, ServeOptions,
13};
14use crate::config::ConfigSet;
15use crate::fsck_msg_policy::{ObjectFsckPolicy, SkipListError};
16use crate::fsck_standalone::FsckError;
17use crate::hooks::{run_hook_in_git_dir_with_cancel, HookError, HookResult};
18use crate::index_pack::{install_pack_path, IngestPackOptions};
19use crate::objects::odb::store::{MemoryStore, ObjectStore, WritableObjectStore};
20use crate::objects::odb::OdbBuilder;
21use crate::objects::pack_stream_scan::{PackStreamScanError, PackStreamScanner};
22use crate::objects::{ObjectId, ObjectKind};
23use crate::odb::Odb;
24use crate::odb::WriteOptions;
25use crate::parallel::{try_par_map, ParallelHashError, ParallelOp};
26use crate::progress::{Cancel, CancelSignal, NeverCancel};
27use crate::receive_pack::{max_input_size_from_config, should_use_unpack_objects};
28use crate::receive_quarantine::{MigratedPackKeepGuard, ReceiveQuarantine};
29use crate::repo::Repository;
30use crate::shallow::{add_shallow_boundaries, load_shallow_boundaries, PushShallowInfo};
31use crate::unpack_objects::{unpack_objects, UnpackOptions};
32
33const SPOOL_CHUNK: usize = 256 * 1024;
35
36enum ReceivePackIngest {
38 Quarantine(ReceiveQuarantine),
39 MemoryOverlay {
40 overlay: Arc<MemoryStore>,
41 objects_dir: PathBuf,
42 read: Arc<dyn ObjectStore>,
43 git_dir: PathBuf,
44 received: HashSet<ObjectId>,
45 },
46}
47
48impl ReceivePackIngest {
49 fn create(repo: &Repository) -> crate::error::Result<Self> {
50 let local = repo.odb();
51 if local.uses_files_primary() {
52 return Ok(Self::Quarantine(ReceiveQuarantine::create(
53 local,
54 repo.git_dir(),
55 )?));
56 }
57 let overlay = Arc::new(MemoryStore::new(local.hash_algo()));
58 Ok(Self::MemoryOverlay {
59 overlay,
60 objects_dir: local.objects_dir().to_path_buf(),
61 read: local.read_store(),
62 git_dir: repo.git_dir().to_path_buf(),
63 received: HashSet::new(),
64 })
65 }
66
67 fn odb(&self) -> Odb {
68 match self {
69 Self::Quarantine(q) => q.odb(),
70 Self::MemoryOverlay {
71 overlay,
72 objects_dir,
73 read,
74 git_dir,
75 ..
76 } => {
77 let mut odb = OdbBuilder::files(objects_dir)
78 .primary(overlay.clone())
79 .push_read_source(read.clone())
80 .alternates(false)
81 .build();
82 odb = odb.with_config_git_dir(git_dir.clone());
83 odb
84 }
85 }
86 }
87
88 fn spool_dir(&self) -> PathBuf {
89 match self {
90 Self::Quarantine(q) => q.odb().objects_dir().join("pack"),
91 Self::MemoryOverlay { .. } => std::env::temp_dir(),
92 }
93 }
94
95 fn files_quarantine(&self) -> Option<&ReceiveQuarantine> {
96 match self {
97 Self::Quarantine(q) => Some(q),
98 Self::MemoryOverlay { .. } => None,
99 }
100 }
101
102 fn set_received(&mut self, ids: HashSet<ObjectId>) {
103 if let Self::MemoryOverlay { received, .. } = self {
104 *received = ids;
105 }
106 }
107
108 fn migrate(&mut self, repo: &Repository) -> crate::error::Result<Vec<MigratedPackKeepGuard>> {
109 match self {
110 Self::Quarantine(q) => q.migrate(),
111 Self::MemoryOverlay {
112 overlay, received, ..
113 } => {
114 let primary = repo.odb().primary_writable()?;
115 for oid in received.iter() {
116 if primary.contains(oid)? {
117 continue;
118 }
119 let Some(obj) = overlay.read(oid)? else {
120 return Err(crate::error::Error::ObjectNotFound(oid.to_hex()));
121 };
122 primary.write(obj.kind, &obj.data, WriteOptions::default())?;
123 }
124 repo.odb().flush_for_publish()?;
125 Ok(Vec::new())
126 }
127 }
128 }
129}
130
131const CAPABILITIES: &[&str] = &[
134 "report-status",
135 "report-status-v2",
136 "delete-refs",
137 "side-band-64k",
138 "quiet",
139 "atomic",
140 "ofs-delta",
141];
142
143#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
146pub enum DenyCurrentBranch {
147 #[default]
149 Refuse,
150 Warn,
152 Ignore,
154 UpdateInstead,
157}
158
159impl DenyCurrentBranch {
160 #[must_use]
162 pub fn refuses_checked_out_update(self) -> bool {
163 matches!(self, Self::Refuse)
164 }
165}
166
167#[derive(Debug, Clone, PartialEq, Eq)]
169pub struct ReceivePolicy {
170 pub deny_non_fast_forwards: bool,
173 pub deny_deletes: bool,
175 pub deny_current_branch: DenyCurrentBranch,
178 pub deny_delete_current: bool,
181 pub fsck_objects: Option<ObjectFsckPolicy>,
186 pub shallow_update: bool,
194}
195
196impl Default for ReceivePolicy {
197 fn default() -> Self {
198 Self {
199 deny_non_fast_forwards: false,
200 deny_deletes: false,
201 deny_current_branch: DenyCurrentBranch::Refuse,
202 deny_delete_current: true,
203 fsck_objects: None,
204 shallow_update: false,
205 }
206 }
207}
208
209impl ReceivePolicy {
210 #[must_use]
233 pub fn from_config(config: &ConfigSet) -> Self {
234 let flag = |key: &str| config.get_bool(key).and_then(|r| r.ok()).unwrap_or(false);
235 let allows_delete = |key: &str| {
237 config.get(key).is_some_and(|v| {
238 let v = v.to_ascii_lowercase();
239 matches!(v.as_str(), "ignore" | "warn" | "false" | "no" | "off" | "0")
240 })
241 };
242 let deny_current_branch = match config.get("receive.denyCurrentBranch") {
243 None => DenyCurrentBranch::Refuse,
244 Some(v) => {
245 let v = v.to_ascii_lowercase();
246 if matches!(v.as_str(), "refuse" | "true" | "1" | "yes" | "on") {
247 DenyCurrentBranch::Refuse
248 } else if v == "warn" {
249 DenyCurrentBranch::Warn
250 } else if v == "updateinstead" {
251 DenyCurrentBranch::UpdateInstead
252 } else if matches!(v.as_str(), "ignore" | "false" | "no" | "off" | "0") {
253 DenyCurrentBranch::Ignore
254 } else {
255 DenyCurrentBranch::Refuse
256 }
257 }
258 };
259 let explicit = |key: &str| config.get_bool(key).map(|r| r.unwrap_or(true));
260 let fsck = explicit("receive.fsckObjects")
261 .or_else(|| explicit("transfer.fsckObjects"))
262 .unwrap_or(false);
263 Self {
264 deny_non_fast_forwards: flag("receive.denyNonFastForwards"),
265 deny_deletes: flag("receive.denyDeletes"),
266 deny_current_branch,
267 deny_delete_current: !allows_delete("receive.denyDeleteCurrent"),
268 fsck_objects: fsck.then(|| ObjectFsckPolicy::from_config(config, "receive.fsck")),
269 shallow_update: flag("receive.shallowUpdate"),
270 }
271 }
272}
273
274#[derive(Debug, Clone, PartialEq, Eq)]
276pub struct RejectedObject {
277 pub oid: ObjectId,
279 pub kind: ObjectKind,
281 pub error: FsckError,
283}
284
285impl std::fmt::Display for RejectedObject {
286 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
287 write!(
288 f,
289 "{} {} refused: {}",
290 self.kind,
291 self.oid.to_hex(),
292 self.error.report_line()
293 )
294 }
295}
296
297#[derive(Debug, Clone, PartialEq, Eq)]
299pub struct RefCommand {
300 pub refname: String,
302 pub old: Option<ObjectId>,
304 pub new: Option<ObjectId>,
306}
307
308impl RefCommand {
309 fn refused(&self, reason: RefUpdateRefusal) -> RefUpdateResult {
310 RefUpdateResult {
311 refname: self.refname.clone(),
312 old: self.old,
313 new: self.new,
314 error: Some(reason),
315 }
316 }
317
318 fn accepted(&self) -> RefUpdateResult {
319 RefUpdateResult {
320 refname: self.refname.clone(),
321 old: self.old,
322 new: self.new,
323 error: None,
324 }
325 }
326}
327
328#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
334#[non_exhaustive]
335pub enum RefUpdateRefusal {
336 #[error("funny refname")]
338 InvalidRefName,
339 #[error("deny updating a hidden ref")]
341 HiddenRef,
342 #[error("stale info")]
344 StaleInfo,
345 #[error("deletion prohibited")]
347 DeletionProhibited,
348 #[error("deletion of the current branch prohibited")]
351 CurrentBranchDeletion,
352 #[error("branch is currently checked out")]
355 CurrentBranch,
356 #[error("Up-to-date check failed")]
358 PushCheckoutUpToDateFailed,
359 #[error("Working directory has unstaged changes")]
361 PushCheckoutUnstagedChanges,
362 #[error("Working directory has staged changes")]
364 PushCheckoutStagedChanges,
365 #[error("Could not update working tree to new HEAD")]
367 PushCheckoutFailed,
368 #[error("push-to-checkout hook declined")]
370 PushCheckoutHookDeclined,
371 #[error("missing necessary objects")]
373 MissingObjects,
374 #[error("missing necessary objects")]
381 Unconnected {
382 missing: ObjectId,
384 referrer: ObjectId,
386 },
387 #[error("non-fast-forward")]
389 NonFastForward,
390 #[error("bad ref")]
393 BadRef,
394 #[error("unable to verify fast-forward")]
398 FastForwardCheckFailed,
399 #[error("shallow update not allowed")]
403 ShallowUpdateNotAllowed,
404 #[error("unpacker error")]
407 UnpackerError,
408 #[error("atomic transaction failed")]
411 AtomicTransactionFailed,
412 #[error("pre-receive hook declined")]
414 PreReceiveHookDeclined,
415 #[error("hook declined")]
417 UpdateHookDeclined,
418 #[error("{0}")]
420 Policy(String),
421 #[error("{target} is the target branch; it changes only through integrate")]
425 TargetBranch {
426 target: String,
428 },
429 #[error("reference already exists")]
431 ReferenceAlreadyExists,
432 #[error("failed to update ref: {0}")]
434 RefStore(String),
435}
436
437#[derive(Debug, Clone, PartialEq, Eq)]
439pub struct RefUpdateResult {
440 pub refname: String,
442 pub old: Option<ObjectId>,
444 pub new: Option<ObjectId>,
446 pub error: Option<RefUpdateRefusal>,
448}
449
450#[derive(Debug, thiserror::Error)]
456#[non_exhaustive]
457pub enum UnpackError {
458 #[error("unexpected end of pack stream")]
460 Truncated,
461 #[error("pack exceeds maximum input size of {limit} bytes")]
463 TooLarge {
464 limit: u64,
466 },
467 #[error(transparent)]
469 Malformed(#[from] PackStreamScanError),
470 #[error(transparent)]
473 Ingest(Box<crate::error::Error>),
474 #[error("{0}")]
476 Rejected(RejectedObject),
477 #[error(transparent)]
480 FsckSkipList(#[from] SkipListError),
481 #[error(transparent)]
483 Io(#[from] std::io::Error),
484 #[error("cancelled")]
486 Cancelled,
487}
488
489impl From<crate::error::Error> for UnpackError {
490 fn from(err: crate::error::Error) -> Self {
491 Self::Ingest(Box::new(err))
492 }
493}
494
495pub trait ReceiveHook {
510 fn pre_receive(
526 &self,
527 commands: &[RefCommand],
528 odb: &Odb,
529 ) -> std::result::Result<(), RefUpdateRefusal> {
530 let _ = (commands, odb);
531 Ok(())
532 }
533
534 fn pre_receive_ctx(
536 &self,
537 commands: &[RefCommand],
538 odb: &Odb,
539 ctx: &ReceiveHookContext<'_>,
540 ) -> std::result::Result<(), RefUpdateRefusal> {
541 let _ = ctx;
542 self.pre_receive(commands, odb)
543 }
544
545 fn update(&self, command: &RefCommand, odb: &Odb) -> std::result::Result<(), RefUpdateRefusal> {
560 let _ = (command, odb);
561 Ok(())
562 }
563
564 fn update_ctx(
566 &self,
567 command: &RefCommand,
568 odb: &Odb,
569 ctx: &ReceiveHookContext<'_>,
570 ) -> std::result::Result<(), RefUpdateRefusal> {
571 let _ = ctx;
572 self.update(command, odb)
573 }
574
575 fn message(&self, updates: &[RefUpdateResult]) -> Option<String> {
591 let _ = updates;
592 None
593 }
594
595 fn post_receive(
602 &self,
603 commands: &[&RefCommand],
604 odb: &Odb,
605 ctx: &ReceiveHookContext<'_>,
606 ) -> std::result::Result<(), RefUpdateRefusal> {
607 let _ = (commands, odb, ctx);
608 Ok(())
609 }
610
611 fn post_update(
616 &self,
617 commands: &[&RefCommand],
618 odb: &Odb,
619 ) -> std::result::Result<(), RefUpdateRefusal> {
620 let _ = (commands, odb);
621 Ok(())
622 }
623}
624
625#[derive(Debug, Clone, Copy, Default)]
627pub struct ReceiveHookContext<'a> {
628 pub push_options: &'a [String],
630}
631
632impl ReceiveHook for () {}
633
634impl<F> ReceiveHook for F
635where
636 F: Fn(&RefCommand, &Odb) -> std::result::Result<(), RefUpdateRefusal>,
637{
638 fn update(&self, command: &RefCommand, odb: &Odb) -> std::result::Result<(), RefUpdateRefusal> {
639 self(command, odb)
640 }
641}
642
643#[derive(Debug, Clone, PartialEq, Eq)]
645pub struct HookRunRecord {
646 pub name: String,
648 pub ran: bool,
650 pub exit_code: Option<i32>,
652}
653
654#[derive(Debug, Default)]
656pub struct ReceiveOutcome {
657 pub unpack_error: Option<UnpackError>,
659 pub updates: Vec<RefUpdateResult>,
661 pub hooks: Vec<HookRunRecord>,
663 pub push_options: Vec<String>,
665 pub bytes_received: u64,
667 pub objects_received: u32,
669 pub post_receive_callback_error: Option<RefUpdateRefusal>,
671 pub post_update_callback_error: Option<RefUpdateRefusal>,
673}
674
675struct AppliedPush {
676 updates: Vec<RefUpdateResult>,
677 post_receive_callback_error: Option<RefUpdateRefusal>,
678 post_update_callback_error: Option<RefUpdateRefusal>,
679}
680
681fn applied_push(updates: Vec<RefUpdateResult>) -> AppliedPush {
682 AppliedPush {
683 updates,
684 post_receive_callback_error: None,
685 post_update_callback_error: None,
686 }
687}
688
689pub fn receive_pack(
698 repo: &Repository,
699 input: &mut dyn Read,
700 output: &mut dyn Write,
701 opts: &ServeOptions,
702 policy: &ReceivePolicy,
703) -> Result<ReceiveOutcome> {
704 receive_pack_with_hook(repo, input, output, opts, policy, &())
705}
706
707pub fn receive_pack_with_hook(
761 repo: &Repository,
762 input: &mut dyn Read,
763 output: &mut dyn Write,
764 opts: &ServeOptions,
765 policy: &ReceivePolicy,
766 hook: &dyn ReceiveHook,
767) -> Result<ReceiveOutcome> {
768 receive_pack_with_hook_cancellable(repo, input, output, opts, policy, hook, &NeverCancel)
769}
770
771pub fn receive_pack_with_hook_cancellable(
781 repo: &Repository,
782 input: &mut dyn Read,
783 output: &mut dyn Write,
784 opts: &ServeOptions,
785 policy: &ReceivePolicy,
786 hook: &dyn ReceiveHook,
787 cancel: &dyn Cancel,
788) -> Result<ReceiveOutcome> {
789 let stop = CancelSignal::new(cancel);
790 stop.poll().map_err(|_| ServeError::Cancelled)?;
791 if opts.advertise_refs || !opts.stateless_rpc {
792 if opts.protocol == ProtocolVersion::V1 {
793 write_line(output, "version 1")?;
794 }
795 write_advertisement(repo, output, opts)?;
796 output.flush()?;
797 if opts.advertise_refs {
798 return Ok(ReceiveOutcome::default());
799 }
800 }
801
802 let Some(request) = read_commands(repo, input)? else {
803 return Ok(ReceiveOutcome::default());
804 };
805
806 let cfg = repo
807 .config()
808 .unwrap_or_else(|_| std::sync::Arc::new(ConfigSet::new()));
809 let max_input = max_input_size_from_config(cfg.as_ref());
810 let sideband = request.caps.contains("side-band-64k");
811
812 let needs_pack = request.commands.iter().any(|c| c.new.is_some());
813 let mut bytes_received = 0u64;
814 let mut objects_received = 0u32;
815 let mut ingest = if needs_pack {
816 Some(ReceivePackIngest::create(repo)?)
817 } else {
818 None
819 };
820
821 let receiver_grafts = if request.shallow.is_empty() {
822 HashSet::new()
823 } else {
824 load_shallow_boundaries(repo.git_dir())
825 };
826 let mut shallow_info =
827 PushShallowInfo::classify(&request.shallow, repo.odb(), &receiver_grafts)?;
828 let client_boundaries: HashSet<ObjectId> = request.shallow.iter().copied().collect();
829
830 let promises = if crate::promisor::repo_treats_promisor_packs(repo.git_dir(), cfg.as_ref()) {
833 Some(super::promised_objects::PromisorObjects::new(repo)?)
834 } else {
835 None
836 };
837 let existing = if needs_pack {
839 existing_ref_values(repo)?
840 } else {
841 HashSet::new()
842 };
843 let mut unconnected: HashMap<ObjectId, Unconnected> = HashMap::new();
845 let unpack_error = match ingest.as_mut() {
846 Some(ing) => {
847 let odb = ing.odb();
848 let spool_dir = ing.spool_dir();
849 spool_push_pack(input, &odb, &spool_dir, max_input, &stop)
850 .and_then(|pack| {
851 bytes_received = pack.bytes_received();
852 objects_received = pack.object_count();
853 ingest_spooled_pack(
854 &odb,
855 pack,
856 cfg.as_ref(),
857 &client_boundaries,
858 promises.as_ref(),
859 )
860 })
861 .and_then(|indexed| {
862 if let Some(ids) = &indexed {
863 ing.set_received(ids.clone());
864 }
865 if let Some(fsck) = &policy.fsck_objects {
866 check_received_objects(repo, &odb, indexed.as_ref(), cfg.as_ref(), fsck)?;
867 }
868 let received = match indexed {
871 Some(received) => received,
872 None if odb.uses_files_primary() => odb
873 .enumerate_local_loose_objects()?
874 .into_iter()
875 .map(|(oid, _)| oid)
876 .collect(),
877 None => HashSet::new(),
878 };
879 unconnected = unconnected_tips(
880 repo,
881 &odb,
882 &received,
883 &existing,
884 &request.commands,
885 &client_boundaries,
886 promises.as_ref(),
887 )?;
888 Ok(())
889 })
890 .err()
891 }
892 None => None,
893 };
894 if matches!(unpack_error, Some(UnpackError::Cancelled)) {
895 return Err(ServeError::Cancelled);
896 }
897
898 let shallow = match ingest.as_ref() {
899 Some(ing) if unpack_error.is_none() => {
900 let lookup = ing.odb();
901 shallow_info.retain_received(&lookup)?;
902 ShallowPlan::new(
903 repo,
904 &lookup,
905 &request,
906 policy,
907 &shallow_info,
908 &receiver_grafts,
909 )?
910 }
911 _ => ShallowPlan::none(request.commands.len()),
912 };
913
914 let mut hook_records = Vec::new();
915 let mut post_receive_callback_error = None;
916 let mut post_update_callback_error = None;
917 let updates = if unpack_error.is_some() {
918 request
919 .commands
920 .iter()
921 .map(|c| c.refused(RefUpdateRefusal::UnpackerError))
922 .collect()
923 } else {
924 stop.poll().map_err(|_| ServeError::Cancelled)?;
925 let applied = apply_commands_with_hooks(
926 repo,
927 &request,
928 opts,
929 policy,
930 hook,
931 ingest.as_mut(),
932 &unconnected,
933 &shallow,
934 sideband.then_some(&mut *output),
935 &mut hook_records,
936 cancel,
937 )?;
938 post_receive_callback_error = applied.post_receive_callback_error;
939 post_update_callback_error = applied.post_update_callback_error;
940 applied.updates
941 };
942
943 if sideband {
944 if let Some(text) = hook.message(&updates) {
945 write_progress_text(output, &text)?;
946 }
947 }
948 if request.caps.contains("report-status") || request.caps.contains("report-status-v2") {
949 let report = build_report(unpack_error.as_ref(), &updates)?;
950 if sideband {
951 crate::pkt_line::write_sideband_channel1_64k(&mut &mut *output, &report)?;
952 write_flush(output)?;
953 } else {
954 output.write_all(&report)?;
955 }
956 }
957 output.flush()?;
958
959 Ok(ReceiveOutcome {
960 unpack_error,
961 updates,
962 hooks: hook_records,
963 push_options: request.push_options.clone(),
964 bytes_received,
965 objects_received,
966 post_receive_callback_error,
967 post_update_callback_error,
968 })
969}
970
971fn write_progress_text(output: &mut dyn Write, text: &str) -> Result<()> {
974 for line in text.lines() {
975 let line = format!("{line}\n");
976 for chunk in line.as_bytes().chunks(65515) {
977 crate::pkt_line::write_sideband_packet(&mut *output, 2, chunk)
978 .map_err(ServeError::Io)?;
979 }
980 }
981 Ok(())
982}
983
984fn write_advertisement(
986 repo: &Repository,
987 output: &mut dyn Write,
988 opts: &ServeOptions,
989) -> Result<()> {
990 let mut caps: Vec<String> = CAPABILITIES.iter().map(|c| (*c).to_owned()).collect();
991 let config = repo
992 .config()
993 .unwrap_or_else(|_| std::sync::Arc::new(ConfigSet::new()));
994 if config
995 .get_bool("receive.advertisepushoptions")
996 .and_then(|r| r.ok())
997 .unwrap_or(false)
998 {
999 caps.push("push-options".to_owned());
1000 }
1001 caps.push(format!("object-format={}", repo.odb().hash_algo().name()));
1002 if let Some(path) = &opts.push_bundle {
1003 caps.push(format!("{}={path}", super::PUSH_BUNDLE_CAPABILITY));
1004 }
1005 if !opts.agent.is_empty() {
1006 caps.push(format!("agent={}", opts.agent));
1007 }
1008 let caps = caps.join(" ");
1009
1010 let mut first_line = true;
1011 for_each_advertised_ref(repo, &opts.hidden_refs, &[], Peel::No, |r| {
1012 if first_line {
1013 write_line(output, &format!("{} {}\0{caps}", r.oid.to_hex(), r.name))?;
1014 first_line = false;
1015 } else {
1016 write_line(output, &format!("{} {}", r.oid.to_hex(), r.name))?;
1017 }
1018 Ok(std::ops::ControlFlow::Continue(()))
1019 })?;
1020 if first_line {
1021 let zero = ObjectId::null(repo.odb().hash_algo());
1022 write_line(
1023 output,
1024 &format!("{} capabilities^{{}}\0{caps}", zero.to_hex()),
1025 )?;
1026 }
1027 write_flush(output)
1028}
1029
1030struct PushRequest {
1032 commands: Vec<RefCommand>,
1033 caps: HashSet<String>,
1034 push_options: Vec<String>,
1035 shallow: Vec<ObjectId>,
1037}
1038
1039fn read_commands(repo: &Repository, input: &mut dyn Read) -> Result<Option<PushRequest>> {
1045 let mut commands = Vec::new();
1046 let mut caps = HashSet::new();
1047 let mut shallow = Vec::new();
1048 loop {
1049 match read_packet(input)? {
1050 None | Some(Packet::Flush) => break,
1051 Some(Packet::Data(line)) => {
1052 if let Some(hex) = line.strip_prefix("shallow ") {
1053 shallow.push(parse_oid(repo, hex)?);
1054 continue;
1055 }
1056 let (line, line_caps) = line.split_once('\0').unwrap_or((line.as_str(), ""));
1057 if commands.is_empty() {
1058 caps.extend(
1059 line_caps
1060 .split(' ')
1061 .filter(|c| !c.is_empty())
1062 .map(str::to_owned),
1063 );
1064 }
1065 let mut parts = line.splitn(3, ' ');
1066 let (Some(old), Some(new), Some(refname)) =
1067 (parts.next(), parts.next(), parts.next())
1068 else {
1069 return Err(ServeError::Protocol(format!("malformed command '{line}'")));
1070 };
1071 let old = parse_oid(repo, old)?;
1072 let new = parse_oid(repo, new)?;
1073 commands.push(RefCommand {
1074 old: (!old.is_zero()).then_some(old),
1075 new: (!new.is_zero()).then_some(new),
1076 refname: refname.to_owned(),
1077 });
1078 }
1079 Some(other) => {
1080 return Err(ServeError::Protocol(format!(
1081 "unexpected packet in command list: {other:?}"
1082 )))
1083 }
1084 }
1085 }
1086 if commands.is_empty() {
1087 return Ok(None);
1088 }
1089 let mut push_options = Vec::new();
1090 if caps.contains("push-options") {
1091 loop {
1092 match read_packet(input)? {
1093 None | Some(Packet::Flush) => break,
1094 Some(Packet::Data(line)) => {
1095 push_options.push(line);
1096 }
1097 Some(other) => {
1098 return Err(ServeError::Protocol(format!(
1099 "unexpected packet in push options: {other:?}"
1100 )))
1101 }
1102 }
1103 }
1104 }
1105 let format = caps.iter().find_map(|c| c.strip_prefix("object-format="));
1106 check_object_format(repo, format)?;
1107 Ok(Some(PushRequest {
1108 commands,
1109 caps,
1110 push_options,
1111 shallow,
1112 }))
1113}
1114
1115struct ShallowPlan {
1117 roots: Vec<Vec<ObjectId>>,
1120 hook_file: Option<tempfile::NamedTempFile>,
1124}
1125
1126impl ShallowPlan {
1127 fn none(commands: usize) -> Self {
1129 Self {
1130 roots: vec![Vec::new(); commands],
1131 hook_file: None,
1132 }
1133 }
1134
1135 fn new(
1138 repo: &Repository,
1139 odb: &Odb,
1140 request: &PushRequest,
1141 policy: &ReceivePolicy,
1142 info: &PushShallowInfo,
1143 receiver_grafts: &HashSet<ObjectId>,
1144 ) -> Result<Self> {
1145 if info.is_empty() {
1146 return Ok(Self::none(request.commands.len()));
1147 }
1148 let mut receiver_tips: HashSet<ObjectId> = HashSet::new();
1149 super::for_each_advertised_ref(repo, &[], &[], Peel::Yes, |r| {
1150 receiver_tips.insert(r.peeled.unwrap_or(r.oid));
1151 Ok(std::ops::ControlFlow::Continue(()))
1152 })?;
1153 receiver_tips.extend(crate::refs::resolve_ref_in(repo, "HEAD").ok());
1154 let graph = repo.odb().commit_graph_chain();
1155 let roots = request
1156 .commands
1157 .iter()
1158 .map(|c| match c.new {
1159 Some(new) => {
1160 info.new_roots(odb, graph.as_deref(), new, receiver_grafts, &receiver_tips)
1161 }
1162 None => Ok(Vec::new()),
1163 })
1164 .collect::<crate::error::Result<_>>()?;
1165 let hook_file = if policy.shallow_update {
1166 Some(crate::shallow::temporary_push_shallow_file(
1167 repo.git_dir(),
1168 receiver_grafts,
1169 &request.shallow,
1170 )?)
1171 } else {
1172 None
1173 };
1174 Ok(Self { roots, hook_file })
1175 }
1176
1177 fn hook_env(&self) -> Result<Vec<(String, String)>> {
1179 let Some(file) = &self.hook_file else {
1180 return Ok(Vec::new());
1181 };
1182 crate::shallow::git_shallow_hook_env(file.path()).map_err(ServeError::Repository)
1183 }
1184}
1185
1186#[allow(clippy::too_many_arguments)]
1188fn apply_commands_with_hooks(
1189 repo: &Repository,
1190 request: &PushRequest,
1191 opts: &ServeOptions,
1192 policy: &ReceivePolicy,
1193 hook: &dyn ReceiveHook,
1194 ingest: Option<&mut ReceivePackIngest>,
1195 unconnected: &HashMap<ObjectId, Unconnected>,
1196 shallow: &ShallowPlan,
1197 mut sideband_out: Option<&mut dyn Write>,
1198 hook_records: &mut Vec<HookRunRecord>,
1199 cancel: &dyn Cancel,
1200) -> Result<AppliedPush> {
1201 let stop = CancelSignal::new(cancel);
1202 let lookup_odb = ingest.as_ref().map(|ing| ing.odb());
1203 let head_branch = ref_update_policy::head_branch(repo);
1204 let checked_out = ref_update_policy::checked_out_branch(repo);
1205 let branches = ref_update_policy::HeadBranches {
1206 head: head_branch.as_deref(),
1207 checked_out: checked_out.as_deref(),
1208 };
1209 let mut results: Vec<RefUpdateResult> = request
1210 .commands
1211 .iter()
1212 .zip(&shallow.roots)
1213 .map(|(c, roots)| {
1214 if !roots.is_empty() && !policy.shallow_update {
1215 return Ok(c.refused(RefUpdateRefusal::ShallowUpdateNotAllowed));
1216 }
1217 Ok(
1218 match ref_update_policy::validate_ref_update(
1219 repo,
1220 lookup_odb.as_ref(),
1221 c,
1222 policy,
1223 &opts.hidden_refs,
1224 branches,
1225 )? {
1226 Some(reason) => c.refused(reason),
1227 None => match c.new.and_then(|new| unconnected.get(&new)) {
1228 Some(gap) => c.refused(RefUpdateRefusal::Unconnected {
1229 missing: gap.oid,
1230 referrer: gap.referrer,
1231 }),
1232 None => c.accepted(),
1233 },
1234 },
1235 )
1236 })
1237 .collect::<Result<_>>()?;
1238
1239 let atomic = request.caps.contains("atomic");
1240 if atomic && results.iter().any(|r| r.error.is_some()) {
1241 refuse_remaining(&mut results, &RefUpdateRefusal::AtomicTransactionFailed);
1242 return Ok(applied_push(results));
1243 }
1244
1245 let accepted: Vec<RefCommand> = request
1246 .commands
1247 .iter()
1248 .zip(&results)
1249 .filter(|(_, r)| r.error.is_none())
1250 .map(|(c, _)| c.clone())
1251 .collect();
1252 let hook_ctx = ReceiveHookContext {
1253 push_options: &request.push_options,
1254 };
1255 if !accepted.is_empty() {
1256 if let Err(refusal) = hook.pre_receive_ctx(
1257 &accepted,
1258 lookup_odb.as_ref().unwrap_or(repo.odb()),
1259 &hook_ctx,
1260 ) {
1261 refuse_remaining(&mut results, &refusal);
1262 return Ok(applied_push(results));
1263 }
1264 }
1265
1266 let scripts = ReceiveHookScripts::new(repo, &request.push_options, shallow.hook_env()?, cancel);
1267 let all_cmds: Vec<&RefCommand> = request.commands.iter().collect();
1268 stop.poll().map_err(|_| ServeError::Cancelled)?;
1269 let pre_result = scripts.pre_receive(
1270 &all_cmds,
1271 ingest.as_ref().and_then(|i| i.files_quarantine()),
1272 )?;
1273 relay_hook_sideband(&mut sideband_out, &pre_result)?;
1274 hook_records.push(hook_record("pre-receive", &pre_result));
1275 if !pre_result.is_ok() {
1276 refuse_remaining(&mut results, &RefUpdateRefusal::PreReceiveHookDeclined);
1277 return Ok(applied_push(results));
1278 }
1279
1280 if results.iter().all(|r| r.error.is_some()) {
1281 return Ok(applied_push(results));
1282 }
1283
1284 let mut migrated_pack_keeps = Vec::new();
1285 if let Some(ing) = ingest {
1286 migrated_pack_keeps = ing.migrate(repo).map_err(ServeError::Repository)?;
1287 repo.odb().invalidate_packs();
1288 }
1289
1290 for (cmd, result) in request.commands.iter().zip(results.iter_mut()) {
1291 if result.error.is_some() {
1292 continue;
1293 }
1294 if let Err(refusal) = hook.update_ctx(cmd, repo.odb(), &hook_ctx) {
1295 result.error = Some(refusal);
1296 continue;
1297 }
1298 stop.poll().map_err(|_| ServeError::Cancelled)?;
1299 let update_result = scripts.update(cmd)?;
1300 relay_hook_sideband(&mut sideband_out, &update_result)?;
1301 hook_records.push(hook_record("update", &update_result));
1302 if !update_result.is_ok() {
1303 result.error = Some(RefUpdateRefusal::UpdateHookDeclined);
1304 }
1305 }
1306
1307 for (cmd, result) in request.commands.iter().zip(results.iter_mut()) {
1308 if result.error.is_some() {
1309 continue;
1310 }
1311 match super::push_to_checkout::deploy_for_update_instead(
1312 repo,
1313 cmd,
1314 policy,
1315 &mut sideband_out,
1316 )? {
1317 None => {}
1318 Some(reason) => result.error = Some(reason),
1319 }
1320 }
1321
1322 if atomic && results.iter().any(|r| r.error.is_some()) {
1323 refuse_remaining(&mut results, &RefUpdateRefusal::AtomicTransactionFailed);
1324 return Ok(applied_push(results));
1325 }
1326
1327 let new_roots: HashSet<ObjectId> = shallow
1328 .roots
1329 .iter()
1330 .zip(&results)
1331 .filter(|(_, r)| r.error.is_none())
1332 .flat_map(|(roots, _)| roots.iter().copied())
1333 .collect();
1334 if let Err(e) = add_shallow_boundaries(repo.git_dir(), &new_roots) {
1335 let refusal = RefUpdateRefusal::RefStore(e.to_string());
1336 for (roots, result) in shallow.roots.iter().zip(results.iter_mut()) {
1337 if !roots.is_empty() && result.error.is_none() {
1338 result.error = Some(refusal.clone());
1339 }
1340 }
1341 if atomic {
1342 refuse_remaining(&mut results, &RefUpdateRefusal::AtomicTransactionFailed);
1343 return Ok(applied_push(results));
1344 }
1345 }
1346
1347 let pending: Vec<(usize, crate::refs::store::RefUpdate)> = request
1348 .commands
1349 .iter()
1350 .zip(results.iter())
1351 .enumerate()
1352 .filter(|(_, (_, r))| r.error.is_none())
1353 .map(|(idx, (cmd, _))| (idx, receive_command_store_update(cmd)))
1354 .collect();
1355
1356 let store = repo.refs();
1357 let ref_lock_wait = store.ref_lock_wait();
1358 if atomic {
1359 let batch: Vec<_> = pending.iter().map(|(_, u)| u.clone()).collect();
1360 if let Err(crate::error::Error::RefStore(e)) =
1361 crate::refs::commit_ref_updates_on_store_with_wait(store, &batch, || ref_lock_wait)
1362 {
1363 let refusal = ref_update_refusal_from_store(&e);
1364 for (idx, _) in pending {
1365 if results[idx].error.is_none() {
1366 results[idx].error = Some(refusal.clone());
1367 }
1368 }
1369 }
1370 } else {
1371 for (idx, refusal) in commit_receive_ref_updates_on_store(store, pending, ref_lock_wait) {
1372 results[idx].error = Some(refusal);
1373 }
1374 }
1375
1376 let successful: Vec<&RefCommand> = request
1377 .commands
1378 .iter()
1379 .zip(results.iter())
1380 .filter(|(_, r)| r.error.is_none())
1381 .map(|(c, _)| c)
1382 .collect();
1383 let mut post_receive_callback_error = None;
1384 let mut post_update_callback_error = None;
1385 if !successful.is_empty() {
1386 stop.poll().map_err(|_| ServeError::Cancelled)?;
1387 let post_result = scripts.post_receive(&successful)?;
1388 relay_hook_sideband(&mut sideband_out, &post_result)?;
1389 hook_records.push(hook_record("post-receive", &post_result));
1390 post_receive_callback_error = hook
1391 .post_receive(
1392 &successful,
1393 repo.odb(),
1394 &ReceiveHookContext {
1395 push_options: &request.push_options,
1396 },
1397 )
1398 .err();
1399
1400 stop.poll().map_err(|_| ServeError::Cancelled)?;
1401 let pu_result = scripts.post_update(&successful)?;
1402 relay_hook_sideband(&mut sideband_out, &pu_result)?;
1403 hook_records.push(hook_record("post-update", &pu_result));
1404 post_update_callback_error = hook.post_update(&successful, repo.odb()).err();
1405 }
1406
1407 drop(migrated_pack_keeps);
1408
1409 Ok(AppliedPush {
1410 updates: results,
1411 post_receive_callback_error,
1412 post_update_callback_error,
1413 })
1414}
1415
1416fn refuse_remaining(results: &mut [RefUpdateResult], reason: &RefUpdateRefusal) {
1418 for r in results.iter_mut().filter(|r| r.error.is_none()) {
1419 r.error = Some(reason.clone());
1420 }
1421}
1422
1423fn hook_record(name: &str, result: &HookResult) -> HookRunRecord {
1424 HookRunRecord {
1425 name: name.to_owned(),
1426 ran: result.was_executed(),
1427 exit_code: result.status(),
1428 }
1429}
1430
1431pub(crate) struct ReceiveHookScripts<'a> {
1440 repo: &'a Repository,
1441 cancel: &'a dyn Cancel,
1442 receive_env: Vec<(String, String)>,
1444 update_env: Vec<(String, String)>,
1446}
1447
1448impl<'a> ReceiveHookScripts<'a> {
1449 pub fn new(
1452 repo: &'a Repository,
1453 push_options: &[String],
1454 extra_env: Vec<(String, String)>,
1455 cancel: &'a dyn Cancel,
1456 ) -> Self {
1457 let mut receive_env = Vec::with_capacity(push_options.len() + 1 + extra_env.len());
1458 receive_env.push((
1459 "GIT_PUSH_OPTION_COUNT".to_owned(),
1460 push_options.len().to_string(),
1461 ));
1462 receive_env.extend(
1463 push_options
1464 .iter()
1465 .enumerate()
1466 .map(|(i, opt)| (format!("GIT_PUSH_OPTION_{i}"), opt.clone())),
1467 );
1468 receive_env.extend(extra_env.iter().cloned());
1469 Self {
1470 repo,
1471 cancel,
1472 receive_env,
1473 update_env: extra_env,
1474 }
1475 }
1476
1477 pub(crate) fn pre_receive(
1482 &self,
1483 commands: &[&RefCommand],
1484 quarantine: Option<&ReceiveQuarantine>,
1485 ) -> std::result::Result<HookResult, ServeError> {
1486 let mut env = self.receive_env.clone();
1487 if let Some(q) = quarantine {
1488 env.extend(q.hook_env());
1489 }
1490 self.run_with_commands("pre-receive", commands, &env)
1491 }
1492
1493 pub(crate) fn update(&self, cmd: &RefCommand) -> std::result::Result<HookResult, ServeError> {
1495 let algo = self.repo.odb().hash_algo();
1496 let hex = |oid: Option<ObjectId>| oid.unwrap_or_else(|| ObjectId::null(algo)).to_hex();
1497 let (old, new) = (hex(cmd.old), hex(cmd.new));
1498 let args = [cmd.refname.as_str(), old.as_str(), new.as_str()];
1499 self.run_git_dir_hook("update", &args, None, &pairs(&self.update_env))
1500 }
1501
1502 pub(crate) fn post_receive(
1504 &self,
1505 commands: &[&RefCommand],
1506 ) -> std::result::Result<HookResult, ServeError> {
1507 self.run_with_commands("post-receive", commands, &self.receive_env)
1508 }
1509
1510 pub(crate) fn post_update(
1512 &self,
1513 commands: &[&RefCommand],
1514 ) -> std::result::Result<HookResult, ServeError> {
1515 let args: Vec<&str> = commands.iter().map(|c| c.refname.as_str()).collect();
1516 self.run_git_dir_hook("post-update", &args, None, &pairs(&self.update_env))
1517 }
1518
1519 fn run_git_dir_hook(
1520 &self,
1521 hook_name: &str,
1522 args: &[&str],
1523 stdin_data: Option<&[u8]>,
1524 env: &[(&str, &str)],
1525 ) -> std::result::Result<HookResult, ServeError> {
1526 match run_hook_in_git_dir_with_cancel(
1527 self.repo,
1528 hook_name,
1529 args,
1530 stdin_data,
1531 env,
1532 self.cancel,
1533 ) {
1534 Ok(result) => Ok(result),
1535 Err(HookError::Cancelled) => Err(ServeError::Cancelled),
1536 Err(_) => Ok(hook_declined()),
1537 }
1538 }
1539
1540 fn run_with_commands(
1541 &self,
1542 hook_name: &str,
1543 commands: &[&RefCommand],
1544 env: &[(String, String)],
1545 ) -> std::result::Result<HookResult, ServeError> {
1546 let zero = ObjectId::null(self.repo.odb().hash_algo());
1547 let stdin: String = commands
1548 .iter()
1549 .map(|c| {
1550 format!(
1551 "{} {} {}\n",
1552 c.old.unwrap_or(zero).to_hex(),
1553 c.new.unwrap_or(zero).to_hex(),
1554 c.refname
1555 )
1556 })
1557 .collect();
1558 self.run_git_dir_hook(hook_name, &[], Some(stdin.as_bytes()), &pairs(env))
1559 }
1560}
1561
1562fn hook_declined() -> HookResult {
1563 HookResult::Failed {
1564 status: 1,
1565 output: crate::hooks::HookOutput::default(),
1566 }
1567}
1568
1569fn pairs(env: &[(String, String)]) -> Vec<(&str, &str)> {
1570 env.iter().map(|(k, v)| (k.as_str(), v.as_str())).collect()
1571}
1572
1573fn receive_updates_eligible_for_initial_packed(updates: &[crate::refs::store::RefUpdate]) -> bool {
1574 use crate::refs::store::{Expected, RawRef};
1575 updates.iter().all(|update| {
1576 matches!(update.new_value, Some(RawRef::Direct(_)))
1577 && matches!(update.expected, Expected::Missing)
1578 && !update.flags.log_only
1579 && update.reflog.is_none()
1580 })
1581}
1582
1583fn ref_update_refusal_from_store(err: &crate::refs::store::RefStoreError) -> RefUpdateRefusal {
1589 match err {
1590 crate::refs::store::RefStoreError::CreateExists { .. } => {
1591 RefUpdateRefusal::ReferenceAlreadyExists
1592 }
1593 other => RefUpdateRefusal::RefStore(other.to_string()),
1594 }
1595}
1596
1597fn commit_receive_ref_updates_on_store(
1598 store: &dyn crate::refs::store::RefStore,
1599 pending: Vec<(usize, crate::refs::store::RefUpdate)>,
1600 ref_lock_wait: crate::lockfile::LockWait,
1601) -> Vec<(usize, RefUpdateRefusal)> {
1602 if pending.is_empty() {
1603 return Vec::new();
1604 }
1605 let updates: Vec<_> = pending.iter().map(|(_, u)| u.clone()).collect();
1606 let batch_ok = if receive_updates_eligible_for_initial_packed(&updates) {
1607 crate::refs::commit_initial_ref_updates_on_store_with_wait(store, updates.clone(), || {
1608 ref_lock_wait
1609 })
1610 } else {
1611 crate::refs::commit_ref_updates_on_store_with_wait(store, &updates, || ref_lock_wait)
1612 };
1613 if batch_ok.is_ok() {
1614 return Vec::new();
1615 }
1616 let indices: Vec<_> = pending.iter().map(|(idx, _)| *idx).collect();
1617 match crate::refs::commit_ref_updates_non_atomic_on_store(store, updates) {
1618 Ok(outcomes) => indices
1619 .into_iter()
1620 .zip(outcomes)
1621 .filter_map(|(idx, outcome)| {
1622 outcome
1623 .result
1624 .err()
1625 .map(|e| (idx, ref_update_refusal_from_store(&e)))
1626 })
1627 .collect(),
1628 Err(e) => indices
1629 .into_iter()
1630 .map(|idx| {
1631 (
1632 idx,
1633 match &e {
1634 crate::error::Error::RefStore(store) => {
1635 ref_update_refusal_from_store(store)
1636 }
1637 other => RefUpdateRefusal::RefStore(other.to_string()),
1638 },
1639 )
1640 })
1641 .collect(),
1642 }
1643}
1644
1645fn receive_command_store_update(cmd: &RefCommand) -> crate::refs::store::RefUpdate {
1646 let mut flags = crate::refs::store::RefUpdateFlags::default();
1647 if cmd.refname == "HEAD" {
1648 flags.no_deref = true;
1649 }
1650 crate::refs::store::RefUpdate {
1651 name: cmd.refname.clone(),
1652 new_value: cmd.new.map(crate::refs::store::RawRef::Direct),
1653 expected: match cmd.old {
1654 Some(oid) => crate::refs::store::Expected::Oid(oid),
1655 None => crate::refs::store::Expected::Missing,
1656 },
1657 reflog: None,
1658 flags,
1659 }
1660}
1661
1662fn relay_hook_sideband(out: &mut Option<&mut dyn Write>, result: &HookResult) -> Result<()> {
1664 let (Some(w), Some(output)) = (out.as_mut(), result.output()) else {
1665 return Ok(());
1666 };
1667 for chunk in output
1668 .stdout
1669 .chunks(65515)
1670 .chain(output.stderr.chunks(65515))
1671 {
1672 crate::pkt_line::write_sideband_packet(*w, 2, chunk).map_err(ServeError::Io)?;
1673 }
1674 Ok(())
1675}
1676
1677fn build_report(
1680 unpack_error: Option<&UnpackError>,
1681 updates: &[RefUpdateResult],
1682) -> Result<Vec<u8>> {
1683 let mut report = Vec::new();
1684 match unpack_error {
1685 None => write_line(&mut report, "unpack ok")?,
1686 Some(e) => write_line(&mut report, &format!("unpack {}", one_line(&e.to_string())))?,
1687 }
1688 for u in updates {
1689 match &u.error {
1690 None => write_line(&mut report, &format!("ok {}", u.refname))?,
1691 Some(reason) => write_line(
1692 &mut report,
1693 &format!("ng {} {}", u.refname, one_line(&reason.to_string())),
1694 )?,
1695 }
1696 }
1697 write_flush(&mut report)?;
1698 Ok(report)
1699}
1700
1701fn one_line(msg: &str) -> String {
1703 msg.lines().next().unwrap_or_default().trim().to_owned()
1704}
1705
1706struct SpooledPack {
1708 file: tempfile::NamedTempFile,
1709 header: [u8; 12],
1711 bytes_received: u64,
1713}
1714
1715impl SpooledPack {
1716 fn bytes_received(&self) -> u64 {
1717 self.bytes_received
1718 }
1719
1720 fn object_count(&self) -> u32 {
1721 u32::from_be_bytes([
1722 self.header[8],
1723 self.header[9],
1724 self.header[10],
1725 self.header[11],
1726 ])
1727 }
1728}
1729
1730fn spool_push_pack(
1738 input: &mut dyn Read,
1739 odb: &Odb,
1740 spool_dir: &Path,
1741 max_bytes: Option<u64>,
1742 stop: &CancelSignal<'_>,
1743) -> std::result::Result<SpooledPack, UnpackError> {
1744 let hash = odb.hash_algo();
1745 let mut file = tempfile::Builder::new()
1746 .prefix("tmp_pack_")
1747 .tempfile_in(spool_dir)?;
1748 let mut scanner = PackStreamScanner::new(hash);
1749 let mut header = [0u8; 12];
1750 let mut chunk = vec![0u8; SPOOL_CHUNK];
1751 let mut written = 0u64;
1752 let mut pack_len: Option<u64> = None;
1753 let mut out = BufWriter::with_capacity(SPOOL_CHUNK, file.as_file_mut());
1754 loop {
1755 stop.poll().map_err(|_| UnpackError::Cancelled)?;
1756 let want = match pack_len {
1757 Some(len) if written >= len => break,
1758 Some(len) => usize::try_from(len - written).map_or(chunk.len(), |n| n.min(chunk.len())),
1759 None => chunk.len(),
1760 };
1761 let n = read_some(input, &mut chunk[..want])?;
1762 if n == 0 {
1763 return Err(UnpackError::Truncated);
1764 }
1765 let mut data = &chunk[..n];
1766 if pack_len.is_none() {
1767 scanner.feed(data)?;
1768 if scanner.is_complete() {
1769 let len = scanner
1770 .pack_body_byte_len()
1771 .saturating_add(hash.len() as u64);
1772 let keep = usize::try_from(len.saturating_sub(written)).map_or(n, |k| k.min(n));
1773 data = &data[..keep];
1774 pack_len = Some(len);
1775 }
1776 }
1777 if let Some(limit) = max_bytes {
1778 let total = pack_len.unwrap_or(written.saturating_add(data.len() as u64));
1779 if total > limit {
1780 return Err(UnpackError::TooLarge { limit });
1781 }
1782 }
1783 if let Some(missing) = usize::try_from(written)
1784 .ok()
1785 .and_then(|start| header.get_mut(start..))
1786 {
1787 let take = missing.len().min(data.len());
1788 missing[..take].copy_from_slice(&data[..take]);
1789 }
1790 out.write_all(data)?;
1791 written += data.len() as u64;
1792 }
1793 out.flush()?;
1794 drop(out);
1795 Ok(SpooledPack {
1796 file,
1797 header,
1798 bytes_received: written,
1799 })
1800}
1801
1802fn read_some(input: &mut dyn Read, buf: &mut [u8]) -> std::io::Result<usize> {
1804 loop {
1805 match input.read(buf) {
1806 Err(e) if e.kind() == std::io::ErrorKind::Interrupted => continue,
1807 other => return other,
1808 }
1809 }
1810}
1811
1812fn check_received_objects(
1817 repo: &Repository,
1818 odb: &Odb,
1819 indexed: Option<&HashSet<ObjectId>>,
1820 cfg: &ConfigSet,
1821 fsck: &ObjectFsckPolicy,
1822) -> std::result::Result<(), UnpackError> {
1823 let checker = fsck.checker(odb.hash_algo(), repo.git_dir(), repo.environment())?;
1824 let mut received: Vec<ObjectId> = match indexed {
1825 Some(ids) => ids.iter().copied().collect(),
1826 None => odb
1827 .enumerate_local_loose_objects()?
1828 .into_iter()
1829 .map(|(oid, _)| oid)
1830 .collect(),
1831 };
1832 received.sort_unstable();
1833 let parallel = repo
1834 .thread_policy()
1835 .try_resolve(ParallelOp::IndexPack, cfg)
1836 .map_err(crate::error::Error::from)?;
1837 let verdicts = try_par_map(&received, parallel, received.len(), |oid| {
1838 if odb.read_info(oid)?.kind == ObjectKind::Blob {
1839 return Ok(None);
1840 }
1841 let obj = odb.read(oid)?;
1842 Ok(checker
1843 .check(oid, obj.kind, &obj.data)
1844 .err()
1845 .map(|error| RejectedObject {
1846 oid: *oid,
1847 kind: obj.kind,
1848 error,
1849 }))
1850 })
1851 .map_err(|ParallelHashError::Task(e): ParallelHashError<crate::error::Error>| e)?;
1852 match verdicts.into_iter().flatten().next() {
1853 Some(rejected) => Err(UnpackError::Rejected(rejected)),
1854 None => Ok(()),
1855 }
1856}
1857
1858fn unconnected_tips(
1864 repo: &Repository,
1865 odb: &Odb,
1866 received: &HashSet<ObjectId>,
1867 existing: &HashSet<ObjectId>,
1868 commands: &[RefCommand],
1869 shallow_boundaries: &HashSet<ObjectId>,
1870 promises: Option<&super::promised_objects::PromisorObjects<'_>>,
1871) -> std::result::Result<HashMap<ObjectId, Unconnected>, UnpackError> {
1872 let mut tips: Vec<ObjectId> = commands.iter().filter_map(|c| c.new).collect();
1873 tips.sort_unstable();
1874 tips.dedup();
1875 if tips.is_empty() {
1876 return Ok(HashMap::new());
1877 }
1878 let receiver_grafts = load_shallow_boundaries(repo.git_dir());
1879 let graph = repo.odb().commit_graph_chain();
1880 let pushed = PushedObjects {
1881 odb,
1882 repo_odb: repo.odb(),
1883 graph: graph.as_deref(),
1884 received,
1885 existing,
1886 shallow_boundaries,
1887 receiver_grafts: &receiver_grafts,
1888 promises,
1889 };
1890 let gaps = pushed.unconnected_tips(&tips)?;
1891 Ok(tips
1892 .into_iter()
1893 .zip(gaps)
1894 .filter_map(|(tip, gap)| Some((tip, gap?)))
1895 .collect())
1896}
1897
1898fn existing_ref_values(repo: &Repository) -> Result<HashSet<ObjectId>> {
1900 let mut existing = super::advertised_object_ids_from_repo(repo, &[], &[], super::Peel::Yes)?;
1901 existing.extend(crate::refs::resolve_ref_in(repo, "HEAD").ok());
1902 Ok(existing)
1903}
1904
1905fn ingest_spooled_pack(
1917 odb: &Odb,
1918 pack: SpooledPack,
1919 cfg: &ConfigSet,
1920 shallow_boundaries: &HashSet<ObjectId>,
1921 promises: Option<&super::promised_objects::PromisorObjects<'_>>,
1922) -> std::result::Result<Option<HashSet<ObjectId>>, UnpackError> {
1923 if !odb.uses_files_primary() {
1924 let mut reader = BufReader::with_capacity(SPOOL_CHUNK, pack.file.reopen()?);
1925 let opts = UnpackOptions {
1926 quiet: true,
1927 strict: true,
1928 shallow_boundaries: shallow_boundaries.clone(),
1929 ..UnpackOptions::default()
1930 };
1931 let ingested = match promises {
1932 Some(promises) => {
1933 let promised =
1934 |oid: &ObjectId, referrer: &ObjectId| promises.is_promised(oid, Some(referrer));
1935 crate::unpack_objects::unpack_objects_with_promises(
1936 &mut reader,
1937 odb,
1938 &opts,
1939 &promised,
1940 )?;
1941 odb.flush_for_publish()?;
1942 let mut ids = HashSet::new();
1943 odb.for_each_object(&mut |oid| {
1944 ids.insert(*oid);
1945 std::ops::ControlFlow::Continue(())
1946 })?;
1947 ids
1948 }
1949 None => {
1950 let ingested = crate::unpack_objects::ingest_pack_into_odb(
1951 &mut reader,
1952 odb,
1953 &opts,
1954 None,
1955 None,
1956 )?;
1957 ingested.object_ids.into_iter().collect()
1958 }
1959 };
1960 return Ok(Some(ingested));
1961 }
1962 if should_use_unpack_objects(&pack.header, cfg) {
1963 let mut reader = BufReader::with_capacity(SPOOL_CHUNK, pack.file.reopen()?);
1964 let opts = UnpackOptions {
1965 quiet: true,
1966 strict: true,
1967 shallow_boundaries: shallow_boundaries.clone(),
1968 ..UnpackOptions::default()
1969 };
1970 match promises {
1971 Some(promises) => {
1972 let promised =
1973 |oid: &ObjectId, referrer: &ObjectId| promises.is_promised(oid, Some(referrer));
1974 crate::unpack_objects::unpack_objects_with_promises(
1975 &mut reader,
1976 odb,
1977 &opts,
1978 &promised,
1979 )?;
1980 }
1981 None => {
1982 unpack_objects(&mut reader, odb, &opts)?;
1983 }
1984 }
1985 return Ok(None);
1986 }
1987 let path = pack.file.into_temp_path();
1988 let installed = install_pack_path(
1989 &path,
1990 odb,
1991 &IngestPackOptions {
1992 fix_thin: true,
1993 skip_post_index_verify: true,
1996 merged_config: Some(cfg.clone()),
1997 ..Default::default()
1998 },
1999 )?;
2000 Ok(Some(installed.object_ids))
2001}
2002
2003#[cfg(test)]
2004mod tests {
2005 use super::*;
2006
2007 #[test]
2008 fn report_lists_unpack_status_then_each_ref() {
2009 let oid = ObjectId::from_hex("1111111111111111111111111111111111111111")
2010 .unwrap_or_else(|_| ObjectId::zero());
2011 let updates = vec![
2012 RefUpdateResult {
2013 refname: "refs/heads/main".into(),
2014 old: None,
2015 new: Some(oid),
2016 error: None,
2017 },
2018 RefUpdateResult {
2019 refname: "refs/heads/topic".into(),
2020 old: Some(oid),
2021 new: None,
2022 error: Some(RefUpdateRefusal::DeletionProhibited),
2023 },
2024 ];
2025 let report = build_report(None, &updates).unwrap_or_default();
2026 assert_eq!(
2027 String::from_utf8_lossy(&report),
2028 "000eunpack ok\n0017ok refs/heads/main\n002cng refs/heads/topic deletion prohibited\n0000"
2029 );
2030 }
2031
2032 #[test]
2033 fn report_renders_typed_unpack_error_and_refusal() {
2034 let updates = vec![RefUpdateResult {
2035 refname: "refs/heads/main".into(),
2036 old: None,
2037 new: None,
2038 error: Some(RefUpdateRefusal::UnpackerError),
2039 }];
2040 let report = build_report(Some(&UnpackError::Truncated), &updates).unwrap_or_default();
2041 assert_eq!(
2042 String::from_utf8_lossy(&report),
2043 "0029unpack unexpected end of pack stream\n0026ng refs/heads/main unpacker error\n0000"
2044 );
2045 }
2046
2047 #[test]
2048 fn refusals_render_the_reasons_git_clients_expect() {
2049 let cases = [
2050 (RefUpdateRefusal::InvalidRefName, "funny refname"),
2051 (RefUpdateRefusal::HiddenRef, "deny updating a hidden ref"),
2052 (RefUpdateRefusal::StaleInfo, "stale info"),
2053 (RefUpdateRefusal::DeletionProhibited, "deletion prohibited"),
2054 (
2055 RefUpdateRefusal::CurrentBranchDeletion,
2056 "deletion of the current branch prohibited",
2057 ),
2058 (
2059 RefUpdateRefusal::CurrentBranch,
2060 "branch is currently checked out",
2061 ),
2062 (
2063 RefUpdateRefusal::MissingObjects,
2064 "missing necessary objects",
2065 ),
2066 (
2067 RefUpdateRefusal::Unconnected {
2068 missing: ObjectId::zero(),
2069 referrer: ObjectId::zero(),
2070 },
2071 "missing necessary objects",
2072 ),
2073 (RefUpdateRefusal::NonFastForward, "non-fast-forward"),
2074 (RefUpdateRefusal::BadRef, "bad ref"),
2075 (
2076 RefUpdateRefusal::FastForwardCheckFailed,
2077 "unable to verify fast-forward",
2078 ),
2079 (
2080 RefUpdateRefusal::ShallowUpdateNotAllowed,
2081 "shallow update not allowed",
2082 ),
2083 (RefUpdateRefusal::UnpackerError, "unpacker error"),
2084 (
2085 RefUpdateRefusal::AtomicTransactionFailed,
2086 "atomic transaction failed",
2087 ),
2088 (
2089 RefUpdateRefusal::PreReceiveHookDeclined,
2090 "pre-receive hook declined",
2091 ),
2092 (RefUpdateRefusal::UpdateHookDeclined, "hook declined"),
2093 (RefUpdateRefusal::Policy("frozen".into()), "frozen"),
2094 (
2095 RefUpdateRefusal::TargetBranch {
2096 target: "main".into(),
2097 },
2098 "main is the target branch; it changes only through integrate",
2099 ),
2100 (
2101 RefUpdateRefusal::RefStore("locked".into()),
2102 "failed to update ref: locked",
2103 ),
2104 (
2105 RefUpdateRefusal::ReferenceAlreadyExists,
2106 "reference already exists",
2107 ),
2108 ];
2109 for (refusal, wire) in cases {
2110 assert_eq!(refusal.to_string(), wire);
2111 }
2112 }
2113
2114 struct NoReadPastEnd<'a> {
2117 data: &'a [u8],
2118 }
2119
2120 impl Read for NoReadPastEnd<'_> {
2121 fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
2122 if self.data.is_empty() {
2123 return Err(std::io::Error::other("read past the end of the pack"));
2124 }
2125 let n = buf.len().min(self.data.len());
2126 buf[..n].copy_from_slice(&self.data[..n]);
2127 self.data = &self.data[n..];
2128 Ok(n)
2129 }
2130 }
2131
2132 fn sample_pack() -> Vec<u8> {
2133 use grit_test_support::objects::{HashAlgo as FixtureAlgo, ObjectKind as PK, PackBuilder};
2134 let mut pb = PackBuilder::new(FixtureAlgo::Sha1);
2135 let mut seed = 0x2545_f491_u32;
2136 let big: Vec<u8> = (0..(SPOOL_CHUNK * 3))
2138 .map(|_| {
2139 seed ^= seed << 13;
2140 seed ^= seed >> 17;
2141 seed ^= seed << 5;
2142 seed as u8
2143 })
2144 .collect();
2145 pb.add_full(PK::Blob, &big);
2146 pb.add_full(PK::Blob, b"small\n");
2147 pb.build().bytes
2148 }
2149
2150 fn test_odb(dir: &std::path::Path) -> Odb {
2151 let objects = dir.join("objects");
2152 std::fs::create_dir_all(objects.join("pack")).unwrap_or_default();
2153 Odb::new(&objects)
2154 }
2155
2156 #[test]
2157 fn spool_stops_at_the_trailer_without_reading_further() {
2158 let tmp = tempfile::tempdir().unwrap_or_else(|e| panic!("{e}"));
2159 let odb = test_odb(tmp.path());
2160 let pack = sample_pack();
2161 let mut input = NoReadPastEnd { data: &pack };
2162 let spool = odb.objects_dir().join("pack");
2163 let spooled = spool_push_pack(&mut input, &odb, &spool, None, &CancelSignal::never())
2164 .unwrap_or_else(|e| panic!("{e}"));
2165 assert_eq!(std::fs::read(spooled.file.path()).unwrap_or_default(), pack);
2166 assert_eq!(&spooled.header[..], &pack[..12]);
2167 }
2168
2169 #[test]
2170 fn spool_keeps_only_the_pack_when_more_bytes_follow() {
2171 let tmp = tempfile::tempdir().unwrap_or_else(|e| panic!("{e}"));
2172 let odb = test_odb(tmp.path());
2173 let pack = sample_pack();
2174 let mut stream = pack.clone();
2175 stream.extend_from_slice(b"trailing junk");
2176 let spool = odb.objects_dir().join("pack");
2177 let spooled = spool_push_pack(
2178 &mut stream.as_slice(),
2179 &odb,
2180 &spool,
2181 None,
2182 &CancelSignal::never(),
2183 )
2184 .unwrap_or_else(|e| panic!("{e}"));
2185 assert_eq!(std::fs::read(spooled.file.path()).unwrap_or_default(), pack);
2186 }
2187
2188 #[test]
2189 fn spool_reports_typed_errors() {
2190 let tmp = tempfile::tempdir().unwrap_or_else(|e| panic!("{e}"));
2191 let odb = test_odb(tmp.path());
2192 let pack = sample_pack();
2193
2194 let truncated = &pack[..pack.len() - 5];
2195 let never = CancelSignal::never();
2196 let spool = odb.objects_dir().join("pack");
2197 assert!(matches!(
2198 spool_push_pack(&mut &truncated[..], &odb, &spool, None, &never),
2199 Err(UnpackError::Truncated)
2200 ));
2201
2202 let limit = (pack.len() - 1) as u64;
2203 assert!(matches!(
2204 spool_push_pack(&mut pack.as_slice(), &odb, &spool, Some(limit), &never),
2205 Err(UnpackError::TooLarge { limit: l }) if l == limit
2206 ));
2207 assert!(spool_push_pack(
2208 &mut pack.as_slice(),
2209 &odb,
2210 &spool,
2211 Some(pack.len() as u64),
2212 &never
2213 )
2214 .is_ok());
2215
2216 assert!(matches!(
2217 spool_push_pack(
2218 &mut &b"JUNK\0\0\0\x02\0\0\0\x01"[..],
2219 &odb,
2220 &spool,
2221 None,
2222 &never
2223 ),
2224 Err(UnpackError::Malformed(PackStreamScanError::BadSignature))
2225 ));
2226
2227 let leftovers: Vec<_> = std::fs::read_dir(odb.objects_dir().join("pack"))
2228 .map(|rd| rd.filter_map(|e| e.ok()).collect())
2229 .unwrap_or_default();
2230 assert!(
2231 leftovers.is_empty(),
2232 "failed spools must not leave temp files"
2233 );
2234 }
2235
2236 #[test]
2237 fn deny_current_branch_values() {
2238 let policy_for = |value: Option<&str>| {
2239 let mut config = ConfigSet::new();
2240 if let Some(v) = value {
2241 let _ = config.add_command_override("receive.denyCurrentBranch", v);
2242 }
2243 ReceivePolicy::from_config(&config).deny_current_branch
2244 };
2245 use DenyCurrentBranch as D;
2246 assert_eq!(policy_for(None), D::Refuse);
2247 assert_eq!(policy_for(Some("ignore")), D::Ignore);
2248 assert_eq!(policy_for(Some("updateInstead")), D::UpdateInstead);
2249 assert_eq!(policy_for(Some("refuse")), D::Refuse);
2250 assert_eq!(policy_for(Some("warn")), D::Warn);
2251 }
2252
2253 #[test]
2254 fn shallow_update_is_off_unless_configured() {
2255 assert!(!ReceivePolicy::default().shallow_update);
2256 let mut config = ConfigSet::new();
2257 let _ = config.add_command_override("receive.shallowUpdate", "true");
2258 assert!(ReceivePolicy::from_config(&config).shallow_update);
2259 }
2260
2261 #[test]
2262 fn deny_delete_current_values() {
2263 let policy_for = |value: Option<&str>| {
2264 let mut config = ConfigSet::new();
2265 if let Some(v) = value {
2266 let _ = config.add_command_override("receive.denyDeleteCurrent", v);
2267 }
2268 ReceivePolicy::from_config(&config).deny_delete_current
2269 };
2270 assert!(policy_for(None));
2271 assert!(!policy_for(Some("ignore")));
2272 assert!(!policy_for(Some("warn")));
2273 assert!(!policy_for(Some("false")));
2274 assert!(policy_for(Some("updateInstead")));
2276 assert!(policy_for(Some("refuse")));
2277 assert!(ReceivePolicy::default().deny_delete_current);
2278 }
2279
2280 #[test]
2281 fn existing_ref_values_includes_injected_head() {
2282 use std::sync::Arc;
2283
2284 use crate::config::environment::RepositoryOptions;
2285 use crate::objects::ObjectKind;
2286 use crate::refs::store::{
2287 Expected, MemoryRefStore, RawRef, RefStore, RefTransaction, RefUpdate,
2288 };
2289 use crate::repo::{InitOptions, Repository, RepositoryBuilder};
2290
2291 let dir = tempfile::tempdir().expect("tempdir");
2292 Repository::init(
2293 dir.path(),
2294 &InitOptions::new()
2295 .with_bare(true)
2296 .with_initial_branch("master"),
2297 )
2298 .expect("init");
2299 let store = Arc::new(MemoryRefStore::new());
2300 let repo = RepositoryBuilder::new(RepositoryOptions::empty())
2301 .ref_store(store.clone())
2302 .initial_branch("master")
2303 .open(dir.path(), None)
2304 .expect("open");
2305 let tree = repo.odb().write(ObjectKind::Tree, b"").expect("tree");
2306 let commit = repo
2307 .odb()
2308 .write(
2309 ObjectKind::Commit,
2310 format!(
2311 "tree {tree}\nauthor T <t@e.com> 1 +0000\ncommitter T <t@e.com> 1 +0000\n\ninit\n"
2312 )
2313 .as_bytes(),
2314 )
2315 .expect("commit");
2316 store
2317 .prepare(
2318 RefTransaction::new()
2319 .update(RefUpdate {
2320 name: "refs/heads/master".into(),
2321 new_value: Some(RawRef::Direct(commit)),
2322 expected: Expected::Missing,
2323 reflog: None,
2324 flags: Default::default(),
2325 })
2326 .expect("txn"),
2327 )
2328 .expect("prepare")
2329 .commit()
2330 .expect("commit");
2331
2332 let values = existing_ref_values(&repo).expect("values");
2333 assert!(values.contains(&commit));
2334 }
2335}