git_zero_server/upload_pack/

negotiation.rs

//! Stateless v0 negotiation. Parse the complete bounded body before output.
use super::extensions;
use git_zero_core::{
    Error, ObjectFormat, ObjectId, ObjectKind, Repository, Result,
    io::Write,
    pktline::{self, Packet},
};

pub(crate) fn capabilities(selection: bool) -> &'static [u8] {
    if selection {
        git_zero_core::repository::UPLOAD_CAPABILITIES
    } else {
        b" multi_ack multi_ack_detailed side-band side-band-64k no-progress"
    }
}

#[derive(Clone, Copy, Default)]
struct Capabilities(u16);

const CAPABILITY_FLAGS: &[(&[u8], u16)] = &[
    (b"multi_ack", 1),
    (b"multi_ack_detailed", 2),
    (b"side-band", 4),
    (b"side-band-64k", 8),
    (b"no-progress", 16),
    (b"shallow", 64),
    (b"filter", 128),
    (b"deepen-relative", 256),
    (b"ofs-delta", 512),
    (b"include-tag", 1024),
    (b"deepen-since", 2048),
    (b"deepen-not", 4096),
];

fn capability_flag(capability: &[u8], format: ObjectFormat) -> Result<u16> {
    if capability == format.capability() {
        return Ok(32);
    }
    if matches!(
        capability,
        b"object-format=sha1" | b"object-format=sha256" | b""
    ) {
        return Err(Error::InvalidRequest);
    }
    CAPABILITY_FLAGS
        .iter()
        .find_map(|&(name, flag)| (name == capability).then_some(flag))
        .ok_or(Error::Unsupported)
}

impl Capabilities {
    fn parse(bytes: &[u8], format: ObjectFormat) -> Result<Self> {
        let mut seen = 0_u16;
        // libgit2 can leave a trailing separator; empty fields select nothing.
        for capability in bytes
            .split(|&byte| byte == b' ')
            .filter(|part| !part.is_empty())
        {
            let flag = capability_flag(capability, format)?;
            if seen & flag != 0 {
                return Err(Error::InvalidRequest);
            }
            seen |= flag;
        }
        if seen & 12 == 12 {
            return Err(Error::InvalidRequest);
        }
        Ok(Self(seen))
    }

    fn want(&mut self, value: &[u8], first: bool, format: ObjectFormat) -> Result<ObjectId> {
        let mut hex = value;
        if let Some(split) = value.iter().position(|&byte| byte == b' ') {
            if !first {
                return Err(Error::InvalidRequest);
            }
            *self = Self::parse(&value[split + 1..], format)?;
            hex = &value[..split];
        }
        id(hex, format)
    }

    fn ack(self) -> u8 {
        // Clients can select both advertised modes; detailed takes precedence.
        if self.0 & 2 != 0 {
            2
        } else {
            (self.0 & 1) as u8
        }
    }
    fn band(self) -> u8 {
        ((self.0 >> 2) & 3) as u8
    }
}

pub(crate) struct UploadRequest<'a> {
    pub done: bool,
    capabilities: Capabilities,
    haves: &'a [u8],
    v2: bool,
    #[cfg(feature = "read")]
    shallow_body: &'a [u8],
    pub extensions: extensions::Extensions,
    #[cfg(feature = "read")]
    include_tags: bool,
}

pub(crate) fn id(bytes: &[u8], format: ObjectFormat) -> Result<ObjectId> {
    let id = ObjectId::from_hex(bytes)?;
    if id.format() != format || id.is_zero() {
        return Err(Error::InvalidRequest);
    }
    Ok(id)
}

fn request_line(line: &[u8]) -> &[u8] {
    line.strip_suffix(b"\n").unwrap_or(line)
}

fn unexpected_command(line: &[u8]) -> Error {
    if line.starts_with(b"want ") || line.starts_with(b"have ") || line == b"done" {
        Error::InvalidRequest
    } else {
        Error::Unsupported
    }
}

fn parse_wants(
    bytes: &mut &[u8],
    format: ObjectFormat,
    mut want: impl FnMut(ObjectId) -> Result<()>,
    extensions: &mut extensions::Extensions,
) -> Result<Capabilities> {
    let mut wanted = false;
    let mut extended = false;
    let mut capabilities = Capabilities::default();
    while let Some(packet) = pktline::read(bytes)? {
        let Packet::Data(line) = packet else {
            if packet != Packet::Flush {
                return Err(Error::InvalidRequest);
            }
            // Shallow commands use the server's advertisement; Git does not
            // echo `shallow` in its want capabilities. Filter is negotiated.
            extensions.options.deepen_relative |= capabilities.0 & 256 != 0;
            extensions.require_filter_capability(capabilities.0 & 128 != 0)?;
            return wanted.then_some(capabilities).ok_or(Error::InvalidRequest);
        };
        let line = request_line(line);
        if let Some(value) = line.strip_prefix(b"want ") {
            if extended {
                return Err(Error::InvalidRequest);
            }
            let id = capabilities.want(value, !wanted, format)?;
            if !extensions.revisions_enabled && capabilities.0 & 6144 != 0 {
                return Err(Error::Unsupported);
            }
            want(id)?;
            wanted = true;
        } else if wanted {
            extensions.argument(line, format)?;
            extended = true;
        } else {
            return Err(unexpected_command(line));
        }
    }
    Err(Error::InvalidRequest)
}

fn parse_haves(mut bytes: &[u8], format: ObjectFormat) -> Result<bool> {
    let empty = bytes.is_empty();
    while let Some(packet) = pktline::read(&mut bytes)? {
        let Packet::Data(line) = packet else {
            return bytes
                .is_empty()
                .then_some(false)
                .ok_or(Error::InvalidRequest);
        };
        let line = request_line(line);
        if line == b"done" && bytes.is_empty() {
            return Ok(true);
        }
        let value = line
            .strip_prefix(b"have ")
            .ok_or_else(|| unexpected_command(line))?;
        id(value, format)?;
    }
    empty.then_some(false).ok_or(Error::InvalidRequest)
}

impl<'a> UploadRequest<'a> {
    pub fn parse(
        bytes: &'a [u8],
        format: ObjectFormat,
        version: crate::route::Version,
        want: impl FnMut(ObjectId) -> Result<()>,
    ) -> Result<Self> {
        Self::parse_mode(bytes, format, version, false, want)
    }

    fn parse_mode(
        mut bytes: &'a [u8],
        format: ObjectFormat,
        version: crate::route::Version,
        revisions_enabled: bool,
        want: impl FnMut(ObjectId) -> Result<()>,
    ) -> Result<Self> {
        let mut extensions = extensions::Extensions::new(revisions_enabled);
        #[cfg(feature = "read")]
        let shallow_body = bytes;
        if version == crate::route::Version::V2 {
            let command = crate::v2::Command::parse(bytes, format)?.ok_or(Error::InvalidRequest)?;
            if command.name != b"fetch" {
                return Err(Error::InvalidRequest);
            }
            let fetch = crate::v2::Fetch::parse(command.args, format, want, &mut extensions)?;
            extensions.validate()?;
            return Ok(Self {
                done: fetch.done,
                capabilities: Capabilities(u16::from(fetch.ofs_delta) * 512),
                haves: command.args,
                v2: true,
                #[cfg(feature = "read")]
                shallow_body: command.args,
                extensions,
                #[cfg(feature = "read")]
                include_tags: fetch.include_tags,
            });
        }
        let capabilities = parse_wants(&mut bytes, format, want, &mut extensions)?;
        extensions.validate()?;
        Ok(Self {
            done: parse_haves(bytes, format)?,
            capabilities,
            haves: bytes,
            v2: false,
            #[cfg(feature = "read")]
            shallow_body,
            extensions,
            #[cfg(feature = "read")]
            include_tags: capabilities.0 & 1024 != 0,
        })
    }

    #[cfg(feature = "read")]
    pub fn ofs_delta(&self) -> bool {
        self.capabilities.0 & 512 != 0
    }

    pub fn require_complete(&self) -> Result<()> {
        // A caller-supplied complete pack cannot promise follow-tag selection.
        // Only the graph-backed API advertises include-tag.
        if self.extensions.active() || self.capabilities.0 & (192 | 1024 | 6144) != 0 {
            return Err(Error::Unsupported);
        }
        Ok(())
    }

    #[cfg(feature = "read")]
    pub(crate) fn exclusion_ids<'r>(
        &'r self,
        repository: &'r impl Repository,
    ) -> impl Iterator<Item = Result<ObjectId>> + 'r {
        packet_values(self.shallow_body, b"deepen-not ")
            .map(move |name| exclusion_ref(repository, name))
    }

    #[cfg(feature = "read")]
    pub(crate) fn revision_info(
        &self,
        format: ObjectFormat,
        changes: impl Iterator<Item = git_zero_core::ShallowChange>,
        output: &mut impl Write,
    ) -> Result<()> {
        let mut updates = ShallowUpdates::new(self, format);
        if updates.enabled {
            for change in changes {
                let (id, unshallow) = match change {
                    git_zero_core::ShallowChange::Shallow(id) => (id, false),
                    git_zero_core::ShallowChange::Unshallow(id) => (id, true),
                };
                updates.change(id, unshallow, output)?;
            }
        }
        updates.finish(output)
    }

    #[cfg(feature = "read")]
    pub fn shallow_info(
        &self,
        format: ObjectFormat,
        objects: &[Option<git_zero_core::ValidationMetadata>],
        output: &mut impl Write,
    ) -> Result<()> {
        self.revision_info(
            format,
            objects.iter().flatten().filter_map(shallow_change),
            output,
        )
    }

    #[cfg(feature = "read")]
    pub(crate) fn parse_selected<R: Repository + ?Sized>(
        repository: &git_zero_core::ValidatedRepository<'_, '_, R>,
        body: &'a [u8],
        version: crate::route::Version,
        wants: &mut [ObjectId],
        options: crate::WantOptions,
        revisions_enabled: bool,
    ) -> Result<(Self, usize)> {
        let mut count = 0;
        let mut remaining = options.max_want_lines;
        let upload = Self::parse_mode(
            body,
            repository.object_format(),
            version,
            revisions_enabled,
            |id| {
                remaining = remaining.checked_sub(1).ok_or(Error::LimitExceeded)?;
                let allowed = match options.policy {
                    crate::WantPolicy::AdvertisedOnly => super::advertised(repository, id),
                    crate::WantPolicy::AllVisibleReachable => repository.find_index(id).is_some(),
                };
                if !allowed {
                    return Err(Error::InvalidRequest);
                }
                if !wants[..count].contains(&id) {
                    *wants.get_mut(count).ok_or(Error::BufferTooSmall)? = id;
                    count += 1;
                }
                Ok(())
            },
        )?;
        Ok((upload, count))
    }

    /// Revisit validated packets without retaining an array of have IDs. Only
    /// present commits are acknowledged or used to exclude complete history.
    pub fn common<'r>(
        &'r self,
        repository: &'r impl Repository,
    ) -> impl Iterator<Item = Result<ObjectId>> + 'r {
        packet_values(self.haves, b"have ")
            .filter_map(|hex| ObjectId::from_hex(hex).ok())
            .filter_map(move |id| common_commit(repository, id).transpose())
    }

    #[cfg(feature = "read")]
    pub fn shallow(&self) -> impl Iterator<Item = Result<ObjectId>> + '_ {
        packet_values(self.shallow_body, b"shallow ").map(ObjectId::from_hex)
    }

    #[cfg(feature = "read")]
    pub fn validate_shallow(&self, repository: &impl Repository) -> Result<()> {
        for id in self.shallow() {
            if common_commit(repository, id?)?.is_none() {
                return Err(Error::InvalidRequest);
            }
        }
        Ok(())
    }

    #[cfg(feature = "read")]
    pub fn include_tags(&self) -> bool {
        self.include_tags
    }

    pub fn acknowledge(&self, repository: &impl Repository, output: &mut impl Write) -> Result<()> {
        let mut acknowledgments = Acknowledgments::new(self);
        acknowledgments.start(output)?;
        for id in self.common(repository) {
            acknowledgments.common(id?, output)?;
        }
        acknowledgments.finish(output)
    }

    pub fn output<'w, W: Write>(&self, output: &'w mut W) -> PackOutput<'w, W> {
        PackOutput {
            output,
            band: self.band(),
            buffer: [0; 1024],
            length: 0,
        }
    }

    pub(crate) fn band(&self) -> u8 {
        // V2 always frames its packfile section on sideband channel 1.
        if self.v2 { 2 } else { self.capabilities.band() }
    }
}

#[cfg(feature = "read")]
fn exclusion_matches(full: &[u8], name: &[u8]) -> bool {
    [
        (b"".as_slice(), b"".as_slice()),
        (b"refs/", b""),
        (b"refs/tags/", b""),
        (b"refs/heads/", b""),
        (b"refs/remotes/", b""),
        (b"refs/remotes/", b"/HEAD"),
    ]
    .iter()
    .any(|&(prefix, suffix)| {
        full.strip_prefix(prefix)
            .and_then(|rest| rest.strip_suffix(suffix))
            == Some(name)
    })
}

#[cfg(feature = "read")]
fn exclusion_ref(repository: &impl Repository, name: &[u8]) -> Result<ObjectId> {
    let mut found = if name == b"HEAD" {
        repository.head_state().id()
    } else {
        None
    };
    for index in 0..repository.reference_count() {
        let full = repository
            .reference_name(index)
            .ok_or(Error::InvalidReference)?
            .as_bytes();
        if !exclusion_matches(full, name) {
            continue;
        }
        if found.is_some() {
            return Err(Error::InvalidRequest);
        }
        // Validation already authenticated symbolic aliases and their final ID.
        found = Some(
            repository
                .reference_id(index)
                .ok_or(Error::InvalidReference)?,
        );
    }
    found.ok_or(Error::InvalidRequest)
}

#[cfg(feature = "read")]
fn shallow_change(
    object: &git_zero_core::ValidationMetadata,
) -> Option<git_zero_core::ShallowChange> {
    if object.is_shallow() {
        Some(git_zero_core::ShallowChange::Shallow(object.id))
    } else if object.is_unshallow() {
        Some(git_zero_core::ShallowChange::Unshallow(object.id))
    } else {
        None
    }
}

// The request parser already checked packet framing and ID syntax. These
// helpers keep packet filtering separate from fallible repository metadata.
fn packet_values<'a>(mut bytes: &'a [u8], prefix: &'static [u8]) -> impl Iterator<Item = &'a [u8]> {
    core::iter::from_fn(move || pktline::read(&mut bytes).ok().flatten()).filter_map(
        move |packet| {
            let Packet::Data(line) = packet else {
                return None;
            };
            request_line(line).strip_prefix(prefix)
        },
    )
}

pub(crate) fn common_commit(
    repository: &impl Repository,
    id: ObjectId,
) -> Result<Option<ObjectId>> {
    let Some(index) = repository.find_index(id) else {
        return Ok(None);
    };
    Ok((repository.object_info(index)?.kind == ObjectKind::Commit).then_some(id))
}

/// Complete-pack sink passed to host pack writers. It forwards raw pack bytes
/// or frames channel 1 according to the client's negotiated sideband mode.
/// Uses a fixed 1 KiB buffer; packet sizes stay within both protocol limits.
/// The server flushes the final packet and terminator after the writer succeeds.
pub struct PackOutput<'a, W: Write> {
    output: &'a mut W,
    band: u8,
    buffer: [u8; 1024],
    length: usize,
}

impl<W: Write> PackOutput<'_, W> {
    fn flush(&mut self) -> Result<()> {
        if self.length != 0 {
            pktline::write_parts(self.output, &[b"\x01", &self.buffer[..self.length]])?;
            self.length = 0;
        }
        Ok(())
    }

    pub(crate) fn finish(&mut self) -> Result<()> {
        if self.band != 0 {
            self.flush()?;
            self.output.write_all(b"0000")?;
        }
        Ok(())
    }
}

impl<W: Write> Write for PackOutput<'_, W> {
    fn write_all(&mut self, mut bytes: &[u8]) -> Result<()> {
        if self.band == 0 {
            return self.output.write_all(bytes);
        }
        let capacity = if self.band == 1 {
            995
        } else {
            self.buffer.len()
        };
        while !bytes.is_empty() {
            let count = bytes.len().min(capacity - self.length);
            self.buffer[self.length..self.length + count].copy_from_slice(&bytes[..count]);
            self.length += count;
            bytes = &bytes[count..];
            if self.length == capacity {
                self.flush()?;
            }
        }
        Ok(())
    }
}

/// Write a terminal v0 ERR packet as a replacement for a failed buffered upload
/// response. Discard the failed response first. No backend details are exposed.
/// Hosts own the HTTP status; never append this to a partial raw pack.
pub fn write_upload_error(error: Error, output: &mut impl Write) -> Result<()> {
    let message: &[u8] = match error {
        Error::BufferTooSmall | Error::LimitExceeded => b"request exceeds resource limits",
        Error::Backend => b"repository unavailable",
        Error::Unsupported => b"unsupported upload request",
        _ => b"invalid upload request",
    };
    pktline::write_parts(output, &[b"ERR ", message, b"\n"])
}

/// Stream host-computed shallow changes before acknowledgments/pack framing.
/// The host must validate commit IDs and may only unshallow a client boundary.
#[derive(Clone, Copy, Eq, PartialEq)]
enum Phase {
    Initial,
    Started,
    Finished,
}

/// A v0/v1 depth probe ends with a flush even without changes; v2 emits a
/// section only for changes in a final response. Errors are terminal.
#[cfg(feature = "read")]
pub struct ShallowUpdates {
    format: ObjectFormat,
    v2: bool,
    enabled: bool,
    phase: Phase,
    failure: Option<Error>,
}
#[cfg(feature = "read")]
impl ShallowUpdates {
    pub(crate) fn new(request: &UploadRequest<'_>, format: ObjectFormat) -> Self {
        Self {
            format,
            v2: request.v2,
            enabled: request.extensions.history() && (!request.v2 || request.done),
            phase: Phase::Initial,
            failure: None,
        }
    }
    /// Whether this protocol round carries depth-boundary announcements.
    pub const fn is_enabled(&self) -> bool {
        self.enabled
    }
    pub fn change(&mut self, id: ObjectId, unshallow: bool, output: &mut impl Write) -> Result<()> {
        let result = self.write_change(id, unshallow, output);
        self.record(result)
    }
    fn write_change(
        &mut self,
        id: ObjectId,
        unshallow: bool,
        output: &mut impl Write,
    ) -> Result<()> {
        self.ready()?;
        if self.phase == Phase::Finished || !self.enabled {
            return Err(Error::InvalidRequest);
        }
        if id.is_zero() || id.format() != self.format {
            return Err(Error::InvalidObjectId);
        }
        if self.phase == Phase::Initial && self.v2 {
            pktline::write(output, b"shallow-info\n")?;
        }
        self.phase = Phase::Started;
        let marker: &[u8] = if unshallow {
            b"unshallow "
        } else {
            b"shallow "
        };
        pktline::write_parts(output, &[marker, id.to_hex().as_bytes(), b"\n"])
    }
    pub fn finish(&mut self, output: &mut impl Write) -> Result<()> {
        let result = self.write_finish(output);
        self.record(result)
    }
    fn write_finish(&mut self, output: &mut impl Write) -> Result<()> {
        self.ready()?;
        if self.phase == Phase::Finished {
            return Err(Error::InvalidRequest);
        }
        let started = self.phase == Phase::Started;
        self.phase = Phase::Finished;
        if self.enabled && (!self.v2 || started) {
            output.write_all(if self.v2 { b"0001" } else { b"0000" })?;
        }
        Ok(())
    }
    fn ready(&self) -> Result<()> {
        self.failure.map_or(Ok(()), Err)
    }
    fn record(&mut self, result: Result<()>) -> Result<()> {
        if let Err(error) = result {
            self.failure.get_or_insert(error);
        }
        result
    }
}

/// Feed only host-verified common commit IDs. Sinks may change between calls.
/// The host must not acknowledge private/unreachable IDs or claim an early cut.
pub struct Acknowledgments {
    v2: bool,
    done: bool,
    mode: u8,
    suppress: bool,
    last: Option<ObjectId>,
    phase: Phase,
}
impl Acknowledgments {
    pub(crate) fn new(request: &UploadRequest<'_>) -> Self {
        Self {
            v2: request.v2,
            done: request.done,
            mode: request.capabilities.ack(),
            suppress: !request.v2 && request.extensions.history() && request.haves.is_empty(),
            last: None,
            phase: Phase::Initial,
        }
    }
    pub fn start(&mut self, output: &mut impl Write) -> Result<()> {
        if self.phase != Phase::Initial {
            return Err(Error::InvalidRequest);
        }
        self.phase = Phase::Started;
        if self.v2 {
            pktline::write(
                output,
                if self.done {
                    b"packfile\n"
                } else {
                    b"acknowledgments\n"
                },
            )?;
        }
        Ok(())
    }
    pub fn common(&mut self, id: ObjectId, output: &mut impl Write) -> Result<()> {
        if self.phase != Phase::Started || id.is_zero() {
            return Err(Error::InvalidRequest);
        }
        if self.suppress
            || (self.v2 && self.done)
            || (!self.v2 && self.mode == 0 && self.last.is_some())
        {
            return Ok(());
        }
        let suffix: &[u8] = if self.v2 || self.mode == 0 {
            b"\n"
        } else if self.mode == 1 {
            b" continue\n"
        } else {
            b" common\n"
        };
        pktline::write_parts(output, &[b"ACK ", id.to_hex().as_bytes(), suffix])?;
        self.last = Some(id);
        Ok(())
    }
    pub fn finish(&mut self, output: &mut impl Write) -> Result<()> {
        if self.phase != Phase::Started {
            return Err(Error::InvalidRequest);
        }
        self.phase = Phase::Finished;
        if self.v2 {
            if !self.done {
                if self.last.is_none() {
                    pktline::write(output, b"NAK\n")?;
                }
                output.write_all(b"0000")?;
            }
        } else if !self.suppress && !(self.mode == 0 && self.last.is_some()) {
            if let Some(id) = self.last.filter(|_| self.done) {
                pktline::write_parts(output, &[b"ACK ", id.to_hex().as_bytes(), b"\n"])?;
            } else {
                pktline::write(output, b"NAK\n")?;
            }
        }
        Ok(())
    }
}