use std::collections::HashSet;
use std::io::{BufReader, BufWriter, Read, Write};
use super::{
advertised_refs, check_object_format, head_info, parse_oid, read_packet, write_flush,
write_line, Packet, Peel, ProtocolVersion, Result, ServeError, ServeOptions,
};
use crate::config::ConfigSet;
use crate::hooks::{run_hook_in_git_dir, HookResult};
use crate::index_pack::{install_pack_path, IngestPackOptions};
use crate::objects::pack_stream_scan::{PackStreamScanError, PackStreamScanner};
use crate::objects::ObjectId;
use crate::odb::Odb;
use crate::receive_pack::{max_input_size_from_config, should_use_unpack_objects};
use crate::receive_quarantine::ReceiveQuarantine;
use crate::repo::Repository;
use crate::unpack_objects::{unpack_objects, UnpackOptions};
const SPOOL_CHUNK: usize = 256 * 1024;
const CAPABILITIES: &[&str] = &[
"report-status",
"report-status-v2",
"delete-refs",
"side-band-64k",
"quiet",
"atomic",
"ofs-delta",
];
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct ReceivePolicy {
pub deny_non_fast_forwards: bool,
pub deny_deletes: bool,
pub deny_current_branch: bool,
}
impl Default for ReceivePolicy {
fn default() -> Self {
Self {
deny_non_fast_forwards: false,
deny_deletes: false,
deny_current_branch: true,
}
}
}
impl ReceivePolicy {
#[must_use]
pub fn from_config(config: &ConfigSet) -> Self {
let flag = |key: &str| config.get_bool(key).and_then(|r| r.ok()).unwrap_or(false);
let deny_current_branch = match config.get("receive.denyCurrentBranch") {
None => true,
Some(v) => !matches!(
v.to_ascii_lowercase().as_str(),
"ignore" | "warn" | "false" | "no" | "off" | "0" | "updateinstead"
),
};
Self {
deny_non_fast_forwards: flag("receive.denyNonFastForwards"),
deny_deletes: flag("receive.denyDeletes"),
deny_current_branch,
}
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RefCommand {
pub refname: String,
pub old: Option<ObjectId>,
pub new: Option<ObjectId>,
}
impl RefCommand {
fn refused(&self, reason: RefUpdateRefusal) -> RefUpdateResult {
RefUpdateResult {
refname: self.refname.clone(),
old: self.old,
new: self.new,
error: Some(reason),
}
}
fn accepted(&self) -> RefUpdateResult {
RefUpdateResult {
refname: self.refname.clone(),
old: self.old,
new: self.new,
error: None,
}
}
}
#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
#[non_exhaustive]
pub enum RefUpdateRefusal {
#[error("funny refname")]
InvalidRefName,
#[error("deny updating a hidden ref")]
HiddenRef,
#[error("stale info")]
StaleInfo,
#[error("deletion prohibited")]
DeletionProhibited,
#[error("deletion of the current branch prohibited")]
CurrentBranchDeletion,
#[error("branch is currently checked out")]
CurrentBranch,
#[error("missing necessary objects")]
MissingObjects,
#[error("non-fast-forward")]
NonFastForward,
#[error("bad ref")]
BadRef,
#[error("unable to verify fast-forward")]
FastForwardCheckFailed,
#[error("unpacker error")]
UnpackerError,
#[error("atomic transaction failed")]
AtomicTransactionFailed,
#[error("pre-receive hook declined")]
PreReceiveHookDeclined,
#[error("hook declined")]
UpdateHookDeclined,
#[error("{0}")]
Policy(String),
#[error("failed to update ref: {0}")]
RefStore(String),
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RefUpdateResult {
pub refname: String,
pub old: Option<ObjectId>,
pub new: Option<ObjectId>,
pub error: Option<RefUpdateRefusal>,
}
#[derive(Debug, thiserror::Error)]
#[non_exhaustive]
pub enum UnpackError {
#[error("unexpected end of pack stream")]
Truncated,
#[error("pack exceeds maximum input size of {limit} bytes")]
TooLarge {
limit: u64,
},
#[error(transparent)]
Malformed(#[from] PackStreamScanError),
#[error(transparent)]
Ingest(Box<crate::error::Error>),
#[error(transparent)]
Io(#[from] std::io::Error),
}
impl From<crate::error::Error> for UnpackError {
fn from(err: crate::error::Error) -> Self {
Self::Ingest(Box::new(err))
}
}
pub trait ReceiveHook {
fn pre_receive(
&self,
commands: &[RefCommand],
odb: &Odb,
) -> std::result::Result<(), RefUpdateRefusal> {
let _ = (commands, odb);
Ok(())
}
fn update(&self, command: &RefCommand, odb: &Odb) -> std::result::Result<(), RefUpdateRefusal> {
let _ = (command, odb);
Ok(())
}
}
impl ReceiveHook for () {}
impl<F> ReceiveHook for F
where
F: Fn(&RefCommand, &Odb) -> std::result::Result<(), RefUpdateRefusal>,
{
fn update(&self, command: &RefCommand, odb: &Odb) -> std::result::Result<(), RefUpdateRefusal> {
self(command, odb)
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct HookRunRecord {
pub name: String,
pub ran: bool,
pub exit_code: Option<i32>,
}
#[derive(Debug, Default)]
pub struct ReceiveOutcome {
pub unpack_error: Option<UnpackError>,
pub updates: Vec<RefUpdateResult>,
pub hooks: Vec<HookRunRecord>,
pub push_options: Vec<String>,
}
pub fn receive_pack(
repo: &Repository,
input: &mut dyn Read,
output: &mut dyn Write,
opts: &ServeOptions,
policy: &ReceivePolicy,
) -> Result<ReceiveOutcome> {
receive_pack_with_hook(repo, input, output, opts, policy, &())
}
pub fn receive_pack_with_hook(
repo: &Repository,
input: &mut dyn Read,
output: &mut dyn Write,
opts: &ServeOptions,
policy: &ReceivePolicy,
hook: &dyn ReceiveHook,
) -> Result<ReceiveOutcome> {
if opts.advertise_refs || !opts.stateless_rpc {
if opts.protocol == ProtocolVersion::V1 {
write_line(output, "version 1")?;
}
write_advertisement(repo, output, opts)?;
output.flush()?;
if opts.advertise_refs {
return Ok(ReceiveOutcome::default());
}
}
let Some(request) = read_commands(repo, input)? else {
return Ok(ReceiveOutcome::default());
};
let cfg = repo
.config()
.unwrap_or_else(|_| std::sync::Arc::new(ConfigSet::new()));
let max_input = max_input_size_from_config(cfg.as_ref());
let sideband = request.caps.contains("side-band-64k");
let needs_pack = request.commands.iter().any(|c| c.new.is_some());
let mut quarantine = if needs_pack {
Some(ReceiveQuarantine::create(repo.odb(), repo.git_dir())?)
} else {
None
};
let unpack_error = match quarantine.as_ref() {
Some(q) => {
let odb = q.odb();
spool_push_pack(input, &odb, max_input)
.and_then(|pack| ingest_spooled_pack(&odb, pack, cfg.as_ref()))
.err()
}
None => None,
};
let mut hook_records = Vec::new();
let updates = if unpack_error.is_some() {
request
.commands
.iter()
.map(|c| c.refused(RefUpdateRefusal::UnpackerError))
.collect()
} else {
apply_commands_with_hooks(
repo,
&request,
opts,
policy,
hook,
quarantine.as_mut(),
sideband.then_some(&mut *output),
&mut hook_records,
)?
};
if request.caps.contains("report-status") || request.caps.contains("report-status-v2") {
let report = build_report(unpack_error.as_ref(), &updates)?;
if sideband {
crate::pkt_line::write_sideband_channel1_64k(&mut &mut *output, &report)?;
write_flush(output)?;
} else {
output.write_all(&report)?;
}
}
output.flush()?;
Ok(ReceiveOutcome {
unpack_error,
updates,
hooks: hook_records,
push_options: request.push_options.clone(),
})
}
fn write_advertisement(
repo: &Repository,
output: &mut dyn Write,
opts: &ServeOptions,
) -> Result<()> {
let mut caps: Vec<String> = CAPABILITIES.iter().map(|c| (*c).to_owned()).collect();
let config = repo
.config()
.unwrap_or_else(|_| std::sync::Arc::new(ConfigSet::new()));
if config
.get_bool("receive.advertisepushoptions")
.and_then(|r| r.ok())
.unwrap_or(false)
{
caps.push("push-options".to_owned());
}
caps.push(format!("object-format={}", repo.odb().hash_algo().name()));
if !opts.agent.is_empty() {
caps.push(format!("agent={}", opts.agent));
}
let caps = caps.join(" ");
let refs = advertised_refs(repo, &opts.hidden_refs, &[], Peel::No)?;
if refs.is_empty() {
let zero = ObjectId::null(repo.odb().hash_algo());
write_line(
output,
&format!("{} capabilities^{{}}\0{caps}", zero.to_hex()),
)?;
}
for (i, r) in refs.iter().enumerate() {
if i == 0 {
write_line(output, &format!("{} {}\0{caps}", r.oid.to_hex(), r.name))?;
} else {
write_line(output, &format!("{} {}", r.oid.to_hex(), r.name))?;
}
}
write_flush(output)
}
struct PushRequest {
commands: Vec<RefCommand>,
caps: HashSet<String>,
push_options: Vec<String>,
}
fn read_commands(repo: &Repository, input: &mut dyn Read) -> Result<Option<PushRequest>> {
let mut commands = Vec::new();
let mut caps = HashSet::new();
loop {
match read_packet(input)? {
None | Some(Packet::Flush) => break,
Some(Packet::Data(line)) => {
let (line, line_caps) = line.split_once('\0').unwrap_or((line.as_str(), ""));
if commands.is_empty() {
caps.extend(
line_caps
.split(' ')
.filter(|c| !c.is_empty())
.map(str::to_owned),
);
}
let mut parts = line.splitn(3, ' ');
let (Some(old), Some(new), Some(refname)) =
(parts.next(), parts.next(), parts.next())
else {
return Err(ServeError::Protocol(format!("malformed command '{line}'")));
};
let old = parse_oid(repo, old)?;
let new = parse_oid(repo, new)?;
commands.push(RefCommand {
old: (!old.is_zero()).then_some(old),
new: (!new.is_zero()).then_some(new),
refname: refname.to_owned(),
});
}
Some(other) => {
return Err(ServeError::Protocol(format!(
"unexpected packet in command list: {other:?}"
)))
}
}
}
if commands.is_empty() {
return Ok(None);
}
let mut push_options = Vec::new();
if caps.contains("push-options") {
loop {
match read_packet(input)? {
None | Some(Packet::Flush) => break,
Some(Packet::Data(line)) => {
push_options.push(line);
}
Some(other) => {
return Err(ServeError::Protocol(format!(
"unexpected packet in push options: {other:?}"
)))
}
}
}
}
let format = caps.iter().find_map(|c| c.strip_prefix("object-format="));
check_object_format(repo, format)?;
Ok(Some(PushRequest {
commands,
caps,
push_options,
}))
}
#[allow(clippy::too_many_arguments)]
fn apply_commands_with_hooks(
repo: &Repository,
request: &PushRequest,
opts: &ServeOptions,
policy: &ReceivePolicy,
hook: &dyn ReceiveHook,
quarantine: Option<&mut ReceiveQuarantine>,
mut sideband_out: Option<&mut dyn Write>,
hook_records: &mut Vec<HookRunRecord>,
) -> Result<Vec<RefUpdateResult>> {
let lookup_odb = quarantine.as_ref().map(|q| q.odb());
let current_branch = if repo.is_bare() {
None
} else {
head_info(repo).target
};
let mut results: Vec<RefUpdateResult> = request
.commands
.iter()
.map(|c| {
Ok(
match check_command(
repo,
lookup_odb.as_ref(),
c,
opts,
policy,
current_branch.as_deref(),
)? {
Some(reason) => c.refused(reason),
None => c.accepted(),
},
)
})
.collect::<Result<_>>()?;
let atomic = request.caps.contains("atomic");
if atomic && results.iter().any(|r| r.error.is_some()) {
refuse_remaining(&mut results, &RefUpdateRefusal::AtomicTransactionFailed);
return Ok(results);
}
let accepted: Vec<RefCommand> = request
.commands
.iter()
.zip(&results)
.filter(|(_, r)| r.error.is_none())
.map(|(c, _)| c.clone())
.collect();
if !accepted.is_empty() {
if let Err(refusal) = hook.pre_receive(&accepted, lookup_odb.as_ref().unwrap_or(repo.odb()))
{
refuse_remaining(&mut results, &refusal);
return Ok(results);
}
}
let algo = repo.odb().hash_algo();
let push_option_env = push_option_env_owned(&request.push_options);
let mut pre_hook_env_owned = push_option_env.clone();
if let Some(q) = quarantine.as_ref() {
pre_hook_env_owned.extend(q.hook_env());
}
let pre_hook_env: Vec<(&str, &str)> = pre_hook_env_owned
.iter()
.map(|(k, v)| (k.as_str(), v.as_str()))
.collect();
let update_hook_env: &[(&str, &str)] = &[];
let post_receive_env_owned = push_option_env;
let post_receive_env: Vec<(&str, &str)> = post_receive_env_owned
.iter()
.map(|(k, v)| (k.as_str(), v.as_str()))
.collect();
let all_cmds: Vec<&RefCommand> = request.commands.iter().collect();
let pre_stdin = commands_stdin(&all_cmds, algo);
let pre_result = run_hook_in_git_dir(
repo,
"pre-receive",
&[],
Some(pre_stdin.as_bytes()),
&pre_hook_env,
);
relay_hook_sideband(&mut sideband_out, &pre_result)?;
hook_records.push(hook_record("pre-receive", &pre_result));
if !pre_result.is_ok() {
refuse_remaining(&mut results, &RefUpdateRefusal::PreReceiveHookDeclined);
return Ok(results);
}
if results.iter().all(|r| r.error.is_some()) {
return Ok(results);
}
if let Some(q) = quarantine {
q.migrate().map_err(ServeError::Repository)?;
repo.odb().invalidate_packs();
}
for (cmd, result) in request.commands.iter().zip(results.iter_mut()) {
if result.error.is_some() {
continue;
}
if let Err(refusal) = hook.update(cmd, repo.odb()) {
result.error = Some(refusal);
continue;
}
let zero = null_oid_hex(algo);
let old_hex = cmd.old.map(|o| o.to_hex()).unwrap_or_else(|| zero.clone());
let new_hex = cmd.new.map(|o| o.to_hex()).unwrap_or(zero.clone());
let args = [cmd.refname.as_str(), old_hex.as_str(), new_hex.as_str()];
let update_result = run_hook_in_git_dir(repo, "update", &args, None, update_hook_env);
relay_hook_sideband(&mut sideband_out, &update_result)?;
hook_records.push(hook_record("update", &update_result));
if !update_result.is_ok() {
result.error = Some(RefUpdateRefusal::UpdateHookDeclined);
}
}
if atomic && results.iter().any(|r| r.error.is_some()) {
refuse_remaining(&mut results, &RefUpdateRefusal::AtomicTransactionFailed);
return Ok(results);
}
let pending: Vec<(usize, crate::refs::store::RefUpdate)> = request
.commands
.iter()
.zip(results.iter())
.enumerate()
.filter(|(_, (_, r))| r.error.is_none())
.map(|(idx, (cmd, _))| (idx, receive_command_store_update(cmd)))
.collect();
if atomic {
let batch: Vec<_> = pending.iter().map(|(_, u)| u.clone()).collect();
if let Err(e) = crate::refs::commit_ref_store_batch(repo.git_dir(), &batch) {
let refusal = RefUpdateRefusal::RefStore(e.to_string());
for (idx, _) in pending {
if results[idx].error.is_none() {
results[idx].error = Some(refusal.clone());
}
}
}
} else {
for (idx, update) in pending {
if let Err(e) = crate::refs::commit_store_update(repo.git_dir(), update) {
results[idx].error = Some(RefUpdateRefusal::RefStore(e.to_string()));
}
}
}
let successful: Vec<&RefCommand> = request
.commands
.iter()
.zip(results.iter())
.filter(|(_, r)| r.error.is_none())
.map(|(c, _)| c)
.collect();
if !successful.is_empty() {
let post_stdin = commands_stdin(&successful, algo);
let post_result = run_hook_in_git_dir(
repo,
"post-receive",
&[],
Some(post_stdin.as_bytes()),
&post_receive_env,
);
relay_hook_sideband(&mut sideband_out, &post_result)?;
hook_records.push(hook_record("post-receive", &post_result));
let mut post_update_args: Vec<&str> = Vec::new();
for result in results.iter().filter(|r| r.error.is_none()) {
post_update_args.push(result.refname.as_str());
}
let pu_result = run_hook_in_git_dir(
repo,
"post-update",
&post_update_args,
None,
update_hook_env,
);
relay_hook_sideband(&mut sideband_out, &pu_result)?;
hook_records.push(hook_record("post-update", &pu_result));
}
Ok(results)
}
fn refuse_remaining(results: &mut [RefUpdateResult], reason: &RefUpdateRefusal) {
for r in results.iter_mut().filter(|r| r.error.is_none()) {
r.error = Some(reason.clone());
}
}
fn hook_record(name: &str, result: &HookResult) -> HookRunRecord {
HookRunRecord {
name: name.to_owned(),
ran: result.was_executed(),
exit_code: result.status(),
}
}
fn push_option_env_owned(options: &[String]) -> Vec<(String, String)> {
let mut pairs = vec![(
"GIT_PUSH_OPTION_COUNT".to_owned(),
options.len().to_string(),
)];
for (i, opt) in options.iter().enumerate() {
pairs.push((format!("GIT_PUSH_OPTION_{i}"), opt.clone()));
}
pairs
}
fn commands_stdin(commands: &[&RefCommand], algo: crate::objects::HashAlgo) -> String {
let zero = null_oid_hex(algo);
commands
.iter()
.map(|c| {
format!(
"{} {} {}\n",
c.old.map(|o| o.to_hex()).unwrap_or_else(|| zero.clone()),
c.new.map(|o| o.to_hex()).unwrap_or_else(|| zero.clone()),
c.refname
)
})
.collect()
}
fn null_oid_hex(algo: crate::objects::HashAlgo) -> String {
ObjectId::null(algo).to_hex()
}
fn receive_command_store_update(cmd: &RefCommand) -> crate::refs::store::RefUpdate {
let mut flags = crate::refs::store::RefUpdateFlags::default();
if cmd.refname == "HEAD" {
flags.no_deref = true;
}
crate::refs::store::RefUpdate {
name: cmd.refname.clone(),
new_value: cmd.new.map(crate::refs::store::RawRef::Direct),
expected: match cmd.old {
Some(oid) => crate::refs::store::Expected::Oid(oid),
None => crate::refs::store::Expected::Missing,
},
reflog: None,
flags,
}
}
fn relay_hook_sideband(out: &mut Option<&mut dyn Write>, result: &HookResult) -> Result<()> {
let (Some(w), Some(output)) = (out.as_mut(), result.output()) else {
return Ok(());
};
for chunk in output
.stdout
.chunks(65515)
.chain(output.stderr.chunks(65515))
{
crate::pkt_line::write_sideband_packet(*w, 2, chunk).map_err(ServeError::Io)?;
}
Ok(())
}
fn check_command(
repo: &Repository,
lookup: Option<&Odb>,
cmd: &RefCommand,
opts: &ServeOptions,
policy: &ReceivePolicy,
current_branch: Option<&str>,
) -> Result<Option<RefUpdateRefusal>> {
if !cmd.refname.starts_with("refs/") || !crate::refs::is_valid_storable_ref_name(&cmd.refname) {
return Ok(Some(RefUpdateRefusal::InvalidRefName));
}
if crate::hide_refs::ref_is_hidden(&cmd.refname, &cmd.refname, &opts.hidden_refs) {
return Ok(Some(RefUpdateRefusal::HiddenRef));
}
let current = crate::refs::resolve_ref(repo.git_dir(), &cmd.refname).ok();
if current != cmd.old {
return Ok(Some(RefUpdateRefusal::StaleInfo));
}
let is_current_branch = current_branch == Some(cmd.refname.as_str());
let Some(new) = cmd.new else {
if policy.deny_deletes {
return Ok(Some(RefUpdateRefusal::DeletionProhibited));
}
if is_current_branch && policy.deny_current_branch {
return Ok(Some(RefUpdateRefusal::CurrentBranchDeletion));
}
return Ok(None);
};
let odb = lookup.unwrap_or(repo.odb());
if !odb.contains(&new)? {
return Ok(Some(RefUpdateRefusal::MissingObjects));
}
if is_current_branch && policy.deny_current_branch {
return Ok(Some(RefUpdateRefusal::CurrentBranch));
}
if let (Some(old), true) = (current, policy.deny_non_fast_forwards) {
if cmd.refname.starts_with("refs/heads/") {
return Ok(check_fast_forward(repo, odb, old, new));
}
}
Ok(None)
}
fn check_fast_forward(
repo: &Repository,
odb: &crate::odb::Odb,
old: ObjectId,
new: ObjectId,
) -> Option<RefUpdateRefusal> {
let is_commit = |oid: &ObjectId| {
odb.read_info(oid)
.is_ok_and(|info| info.kind == crate::objects::ObjectKind::Commit)
};
if !is_commit(&old) || !is_commit(&new) {
return Some(RefUpdateRefusal::BadRef);
}
let graph = repo
.caches()
.commit_graph_chain(&repo.git_dir().join("objects"));
match crate::merge_base::is_ancestor_odb_walk_with_graph(odb, graph.as_deref(), old, new) {
Ok(true) => None,
Ok(false) => Some(RefUpdateRefusal::NonFastForward),
Err(_) => Some(RefUpdateRefusal::FastForwardCheckFailed),
}
}
fn build_report(
unpack_error: Option<&UnpackError>,
updates: &[RefUpdateResult],
) -> Result<Vec<u8>> {
let mut report = Vec::new();
match unpack_error {
None => write_line(&mut report, "unpack ok")?,
Some(e) => write_line(&mut report, &format!("unpack {}", one_line(&e.to_string())))?,
}
for u in updates {
match &u.error {
None => write_line(&mut report, &format!("ok {}", u.refname))?,
Some(reason) => write_line(
&mut report,
&format!("ng {} {}", u.refname, one_line(&reason.to_string())),
)?,
}
}
write_flush(&mut report)?;
Ok(report)
}
fn one_line(msg: &str) -> String {
msg.lines().next().unwrap_or_default().trim().to_owned()
}
struct SpooledPack {
file: tempfile::NamedTempFile,
header: [u8; 12],
}
fn spool_push_pack(
input: &mut dyn Read,
odb: &Odb,
max_bytes: Option<u64>,
) -> std::result::Result<SpooledPack, UnpackError> {
let hash = odb.hash_algo();
let mut file = tempfile::Builder::new()
.prefix("tmp_pack_")
.tempfile_in(odb.objects_dir().join("pack"))?;
let mut scanner = PackStreamScanner::new(hash);
let mut header = [0u8; 12];
let mut chunk = vec![0u8; SPOOL_CHUNK];
let mut written = 0u64;
let mut pack_len: Option<u64> = None;
let mut out = BufWriter::with_capacity(SPOOL_CHUNK, file.as_file_mut());
loop {
let want = match pack_len {
Some(len) if written >= len => break,
Some(len) => usize::try_from(len - written).map_or(chunk.len(), |n| n.min(chunk.len())),
None => chunk.len(),
};
let n = read_some(input, &mut chunk[..want])?;
if n == 0 {
return Err(UnpackError::Truncated);
}
let mut data = &chunk[..n];
if pack_len.is_none() {
scanner.feed(data)?;
if scanner.is_complete() {
let len = scanner
.pack_body_byte_len()
.saturating_add(hash.len() as u64);
let keep = usize::try_from(len.saturating_sub(written)).map_or(n, |k| k.min(n));
data = &data[..keep];
pack_len = Some(len);
}
}
if let Some(limit) = max_bytes {
let total = pack_len.unwrap_or(written.saturating_add(data.len() as u64));
if total > limit {
return Err(UnpackError::TooLarge { limit });
}
}
if let Some(missing) = usize::try_from(written)
.ok()
.and_then(|start| header.get_mut(start..))
{
let take = missing.len().min(data.len());
missing[..take].copy_from_slice(&data[..take]);
}
out.write_all(data)?;
written += data.len() as u64;
}
out.flush()?;
drop(out);
Ok(SpooledPack { file, header })
}
fn read_some(input: &mut dyn Read, buf: &mut [u8]) -> std::io::Result<usize> {
loop {
match input.read(buf) {
Err(e) if e.kind() == std::io::ErrorKind::Interrupted => continue,
other => return other,
}
}
}
fn ingest_spooled_pack(
odb: &Odb,
pack: SpooledPack,
cfg: &ConfigSet,
) -> std::result::Result<(), UnpackError> {
if should_use_unpack_objects(&pack.header, cfg) {
let mut reader = BufReader::with_capacity(SPOOL_CHUNK, pack.file.reopen()?);
unpack_objects(
&mut reader,
odb,
&UnpackOptions {
quiet: true,
strict: true,
..UnpackOptions::default()
},
)?;
return Ok(());
}
let path = pack.file.into_temp_path();
install_pack_path(
&path,
odb,
&IngestPackOptions {
fix_thin: true,
skip_post_index_verify: true,
..Default::default()
},
)?;
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn report_lists_unpack_status_then_each_ref() {
let oid = ObjectId::from_hex("1111111111111111111111111111111111111111")
.unwrap_or_else(|_| ObjectId::zero());
let updates = vec![
RefUpdateResult {
refname: "refs/heads/main".into(),
old: None,
new: Some(oid),
error: None,
},
RefUpdateResult {
refname: "refs/heads/topic".into(),
old: Some(oid),
new: None,
error: Some(RefUpdateRefusal::DeletionProhibited),
},
];
let report = build_report(None, &updates).unwrap_or_default();
assert_eq!(
String::from_utf8_lossy(&report),
"000eunpack ok\n0017ok refs/heads/main\n002cng refs/heads/topic deletion prohibited\n0000"
);
}
#[test]
fn report_renders_typed_unpack_error_and_refusal() {
let updates = vec![RefUpdateResult {
refname: "refs/heads/main".into(),
old: None,
new: None,
error: Some(RefUpdateRefusal::UnpackerError),
}];
let report = build_report(Some(&UnpackError::Truncated), &updates).unwrap_or_default();
assert_eq!(
String::from_utf8_lossy(&report),
"0029unpack unexpected end of pack stream\n0026ng refs/heads/main unpacker error\n0000"
);
}
#[test]
fn refusals_render_the_reasons_git_clients_expect() {
let cases = [
(RefUpdateRefusal::InvalidRefName, "funny refname"),
(RefUpdateRefusal::HiddenRef, "deny updating a hidden ref"),
(RefUpdateRefusal::StaleInfo, "stale info"),
(RefUpdateRefusal::DeletionProhibited, "deletion prohibited"),
(
RefUpdateRefusal::CurrentBranchDeletion,
"deletion of the current branch prohibited",
),
(
RefUpdateRefusal::CurrentBranch,
"branch is currently checked out",
),
(
RefUpdateRefusal::MissingObjects,
"missing necessary objects",
),
(RefUpdateRefusal::NonFastForward, "non-fast-forward"),
(RefUpdateRefusal::BadRef, "bad ref"),
(
RefUpdateRefusal::FastForwardCheckFailed,
"unable to verify fast-forward",
),
(RefUpdateRefusal::UnpackerError, "unpacker error"),
(
RefUpdateRefusal::AtomicTransactionFailed,
"atomic transaction failed",
),
(
RefUpdateRefusal::PreReceiveHookDeclined,
"pre-receive hook declined",
),
(RefUpdateRefusal::UpdateHookDeclined, "hook declined"),
(RefUpdateRefusal::Policy("frozen".into()), "frozen"),
(
RefUpdateRefusal::RefStore("locked".into()),
"failed to update ref: locked",
),
];
for (refusal, wire) in cases {
assert_eq!(refusal.to_string(), wire);
}
}
struct NoReadPastEnd<'a> {
data: &'a [u8],
}
impl Read for NoReadPastEnd<'_> {
fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
if self.data.is_empty() {
return Err(std::io::Error::other("read past the end of the pack"));
}
let n = buf.len().min(self.data.len());
buf[..n].copy_from_slice(&self.data[..n]);
self.data = &self.data[n..];
Ok(n)
}
}
fn sample_pack() -> Vec<u8> {
use grit_test_support::objects::{HashAlgo as FixtureAlgo, ObjectKind as PK, PackBuilder};
let mut pb = PackBuilder::new(FixtureAlgo::Sha1);
let mut seed = 0x2545_f491_u32;
let big: Vec<u8> = (0..(SPOOL_CHUNK * 3))
.map(|_| {
seed ^= seed << 13;
seed ^= seed >> 17;
seed ^= seed << 5;
seed as u8
})
.collect();
pb.add_full(PK::Blob, &big);
pb.add_full(PK::Blob, b"small\n");
pb.build().bytes
}
fn test_odb(dir: &std::path::Path) -> Odb {
let objects = dir.join("objects");
std::fs::create_dir_all(objects.join("pack")).unwrap_or_default();
Odb::new(&objects)
}
#[test]
fn spool_stops_at_the_trailer_without_reading_further() {
let tmp = tempfile::tempdir().unwrap_or_else(|e| panic!("{e}"));
let odb = test_odb(tmp.path());
let pack = sample_pack();
let mut input = NoReadPastEnd { data: &pack };
let spooled = spool_push_pack(&mut input, &odb, None).unwrap_or_else(|e| panic!("{e}"));
assert_eq!(std::fs::read(spooled.file.path()).unwrap_or_default(), pack);
assert_eq!(&spooled.header[..], &pack[..12]);
}
#[test]
fn spool_keeps_only_the_pack_when_more_bytes_follow() {
let tmp = tempfile::tempdir().unwrap_or_else(|e| panic!("{e}"));
let odb = test_odb(tmp.path());
let pack = sample_pack();
let mut stream = pack.clone();
stream.extend_from_slice(b"trailing junk");
let spooled =
spool_push_pack(&mut stream.as_slice(), &odb, None).unwrap_or_else(|e| panic!("{e}"));
assert_eq!(std::fs::read(spooled.file.path()).unwrap_or_default(), pack);
}
#[test]
fn spool_reports_typed_errors() {
let tmp = tempfile::tempdir().unwrap_or_else(|e| panic!("{e}"));
let odb = test_odb(tmp.path());
let pack = sample_pack();
let truncated = &pack[..pack.len() - 5];
assert!(matches!(
spool_push_pack(&mut &truncated[..], &odb, None),
Err(UnpackError::Truncated)
));
let limit = (pack.len() - 1) as u64;
assert!(matches!(
spool_push_pack(&mut pack.as_slice(), &odb, Some(limit)),
Err(UnpackError::TooLarge { limit: l }) if l == limit
));
assert!(spool_push_pack(&mut pack.as_slice(), &odb, Some(pack.len() as u64)).is_ok());
assert!(matches!(
spool_push_pack(&mut &b"JUNK\0\0\0\x02\0\0\0\x01"[..], &odb, None),
Err(UnpackError::Malformed(PackStreamScanError::BadSignature))
));
let leftovers: Vec<_> = std::fs::read_dir(odb.objects_dir().join("pack"))
.map(|rd| rd.filter_map(|e| e.ok()).collect())
.unwrap_or_default();
assert!(
leftovers.is_empty(),
"failed spools must not leave temp files"
);
}
#[test]
fn deny_current_branch_values() {
let policy_for = |value: Option<&str>| {
let mut config = ConfigSet::new();
if let Some(v) = value {
let _ = config.add_command_override("receive.denyCurrentBranch", v);
}
ReceivePolicy::from_config(&config).deny_current_branch
};
assert!(policy_for(None));
assert!(!policy_for(Some("ignore")));
assert!(!policy_for(Some("updateInstead")));
assert!(policy_for(Some("refuse")));
}
}