use std::ops::ControlFlow;
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};
use alkcall::core::BiStream;
use bstr::BString;
use tokio::io::{ReadHalf, WriteHalf};
use tokio_util::compat::{Compat, TokioAsyncReadCompatExt, TokioAsyncWriteCompatExt};
use gix_transport::client::async_io::SetServiceResponse;
use gix_transport::client::git::ConnectMode;
use gix_transport::client::{Capabilities, Error as ClientError};
use gix_transport::{Protocol, Service as GixService};
use crate::limits::Limits;
use crate::registry::validate_repo_id;
use crate::substrate::Service;
pub use crate::session_fetch::{FetchRound, FetchRoundReport};
pub use crate::session_push::{PushAdvertisedRef, PushReport, PushSendOptions};
pub type GitTransport = gix_transport::client::git::async_io::Connection<
Compat<ReadHalf<BiStream>>,
Compat<WriteHalf<BiStream>>,
>;
const HANDSHAKE_EXTRA_PARAMETERS: &[(&str, Option<&str>)] = &[];
pub(crate) const CLIENT_AGENT: &str =
concat!(env!("CARGO_PKG_NAME"), "/", env!("CARGO_PKG_VERSION"));
fn desired_version(service: Service) -> Protocol {
match service {
Service::UploadPack => Protocol::V2,
Service::ReceivePack => Protocol::V1,
}
}
fn gix_service(service: Service) -> GixService {
match service {
Service::UploadPack => GixService::UploadPack,
Service::ReceivePack => GixService::ReceivePack,
}
}
pub(crate) fn map_client_error(error: ClientError) -> SessionError {
match error {
ClientError::Io(e) => {
let err_line = e
.get_ref()
.and_then(|inner| inner.downcast_ref::<gix_packetline::read::Error>());
match err_line {
Some(line) => SessionError::PeerRefused(line.message.to_string()),
None => SessionError::Transport(e.to_string()),
}
}
ClientError::LineDecode { .. } => SessionError::Protocol(format!(
"malformed pkt-line in the handshake reply: {error}"
)),
ClientError::MissingHandshake
| ClientError::Capabilities { .. }
| ClientError::ExpectedLine(_)
| ClientError::ExpectedDataLine
| ClientError::UnsupportedProtocolVersion(_)
| ClientError::AuthenticationUnsupported
| ClientError::AuthenticationRefused(_) => {
SessionError::Protocol(format!("undecodable handshake reply: {error}"))
}
other => SessionError::Transport(other.to_string()),
}
}
#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
pub enum SessionError {
#[error("the peer refused the git session: {0}")]
PeerRefused(String),
#[error("protocol error in the git session handshake: {0}")]
Protocol(String),
#[error(
"the peer negotiated protocol {negotiated:?} for {service} — the service's pinned \
version contradicts it; fail closed (ADR-003/ADR-013)"
)]
VersionRefused {
service: Service,
negotiated: Protocol,
},
#[error(
"the git session exceeded the session_wall_clock budget ({budget:?}) — fail closed (ADR-009)"
)]
WallClock {
budget: Duration,
},
#[error(
"client-side budget breach — the peer's {what} exceeded the {budget_name} budget \
({limit}) — fail closed (ADR-009, client-side)"
)]
BudgetExceeded {
what: String,
budget_name: &'static str,
limit: u64,
},
#[error("git session transport failure: {0}")]
Transport(String),
#[error("invalid git session input: {0}")]
InvalidInput(String),
}
#[derive(Debug, Clone)]
pub struct HandshakeResponse {
pub protocol: Protocol,
pub capabilities: Capabilities,
pub refs: Vec<BString>,
}
pub struct GitSession {
transport: GitTransport,
pub(crate) service: Service,
repo: String,
pub(crate) limits: Limits,
deadline: Instant,
handshake: Option<HandshakeResponse>,
pub(crate) terminal: Option<SessionError>,
}
impl std::fmt::Debug for GitSession {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("GitSession")
.field("service", &self.service)
.field("repo", &self.repo)
.field("handshaken", &self.handshake.is_some())
.field("terminal", &self.terminal.is_some())
.field("remaining_wall_clock", &self.remaining_wall_clock())
.finish_non_exhaustive()
}
}
impl GitSession {
pub fn connect_direct(
stream: BiStream,
repo: &str,
service: Service,
host: Option<&str>,
limits: Limits,
) -> Result<Self, SessionError> {
let started = Instant::now();
if let Err(cause) = validate_repo_id(repo) {
return Err(SessionError::InvalidInput(format!(
"invalid repo id {repo:?}: {cause}"
)));
}
if let Some(host) = host {
if host.bytes().any(|byte| byte == 0 || byte == b'\n') {
return Err(SessionError::InvalidInput(format!(
"the host carries the request line's framing bytes (NUL/LF), ADR-016 §2: {host:?}"
)));
}
}
let (read_half, write_half) = tokio::io::split(stream);
let transport = gix_transport::client::git::async_io::Connection::new(
read_half.compat(),
write_half.compat_write(),
desired_version(service),
repo.to_owned(),
host.map(|host| (host.to_owned(), None::<u16>)),
ConnectMode::Daemon,
false,
);
Ok(Self::assemble(transport, repo, service, limits, started))
}
pub fn open_via_channels(
stream: BiStream,
repo: &str,
service: Service,
limits: Limits,
) -> Result<Self, SessionError> {
if let Err(cause) = validate_repo_id(repo) {
return Err(SessionError::InvalidInput(format!(
"invalid repo id {repo:?}: {cause}"
)));
}
let (read_half, write_half) = tokio::io::split(stream);
let transport = gix_transport::client::git::async_io::Connection::new(
read_half.compat(),
write_half.compat_write(),
desired_version(service),
repo.to_owned(),
None::<(String, Option<u16>)>,
ConnectMode::Process,
false,
);
Ok(Self::assemble(
transport,
repo,
service,
limits,
Instant::now(),
))
}
fn assemble(
transport: GitTransport,
repo: &str,
service: Service,
limits: Limits,
started: Instant,
) -> Self {
GitSession {
transport,
service,
repo: repo.to_owned(),
deadline: started + limits.session_wall_clock,
limits,
handshake: None,
terminal: None,
}
}
pub fn service(&self) -> Service {
self.service
}
pub fn repo(&self) -> &str {
&self.repo
}
pub fn limits(&self) -> &Limits {
&self.limits
}
pub fn remaining_wall_clock(&self) -> Duration {
self.deadline.saturating_duration_since(Instant::now())
}
pub(crate) fn transport_mut(&mut self) -> &mut GitTransport {
&mut self.transport
}
pub async fn handshake(&mut self) -> Result<HandshakeResponse, SessionError> {
if let Some(cached) = &self.handshake {
return Ok(cached.clone());
}
if let Some(cached) = &self.terminal {
return Err(cached.clone());
}
let bounds = AdvertisementBounds {
max_refs: self.limits.max_advertisement_refs,
max_bytes: self.limits.max_request_body,
};
match self.handshake_once(bounds).await {
Ok(response) => {
self.handshake = Some(response.clone());
Ok(response)
}
Err(error) => {
self.terminal = Some(error.clone());
Err(error)
}
}
}
async fn handshake_once(
&mut self,
bounds: AdvertisementBounds,
) -> Result<HandshakeResponse, SessionError> {
use gix_transport::client::async_io::Transport as _;
let service = self.service;
let remaining = self.remaining_wall_clock();
let exchange = tokio::time::timeout(remaining, async {
let response = self
.transport_mut()
.handshake(gix_service(service), HANDSHAKE_EXTRA_PARAMETERS)
.await
.map_err(map_client_error)?;
let SetServiceResponse {
actual_protocol,
capabilities,
refs,
} = response;
match service {
Service::UploadPack => {
if actual_protocol != Protocol::V2 {
return Err(SessionError::VersionRefused {
service,
negotiated: actual_protocol,
});
}
Ok(HandshakeResponse {
protocol: actual_protocol,
capabilities,
refs: Vec::new(),
})
}
Service::ReceivePack => {
if actual_protocol == Protocol::V2 {
return Err(SessionError::VersionRefused {
service,
negotiated: actual_protocol,
});
}
let refs = match refs {
Some(mut reader) => drain_advertisement(&mut reader, bounds).await?,
None => Vec::new(),
};
Ok(HandshakeResponse {
protocol: actual_protocol,
capabilities,
refs,
})
}
}
});
match exchange.await {
Ok(inner) => inner,
Err(_elapsed) => Err(SessionError::WallClock {
budget: self.limits.session_wall_clock,
}),
}
}
pub(crate) fn typed_wall_clock_breach(&mut self) -> SessionError {
let error = SessionError::WallClock {
budget: self.limits.session_wall_clock,
};
self.terminal = Some(error.clone());
error
}
}
pub(crate) struct AdvertisementBounds {
pub(crate) max_refs: usize,
pub(crate) max_bytes: u64,
}
pub(crate) async fn drain_bounded_lines<R, T>(
reader: &mut R,
bounds: AdvertisementBounds,
what: &'static str,
mut on_line: impl FnMut(&str) -> Result<T, SessionError>,
) -> Result<Vec<T>, SessionError>
where
R: gix_transport::client::async_io::ReadlineBufRead + Unpin,
{
let mut items = Vec::new();
let mut bytes = 0u64;
let mut line = String::new();
loop {
line.clear();
let n = match reader.readline_str(&mut line).await {
Ok(n) => n,
Err(e) => {
let err_line = e
.get_ref()
.and_then(|inner| inner.downcast_ref::<gix_packetline::read::Error>());
return Err(match err_line {
Some(line) => SessionError::PeerRefused(line.message.to_string()),
None => SessionError::Transport(e.to_string()),
});
}
};
if n == 0 {
return Ok(items);
}
bytes += n as u64;
if bytes > bounds.max_bytes {
return Err(SessionError::BudgetExceeded {
what: what.to_owned(),
budget_name: "max_request_body",
limit: bounds.max_bytes,
});
}
if items.len() >= bounds.max_refs {
return Err(SessionError::BudgetExceeded {
what: what.to_owned(),
budget_name: "max_advertisement_refs",
limit: bounds.max_refs as u64,
});
}
items.push(on_line(&line)?);
}
}
async fn drain_advertisement<R>(
reader: &mut R,
bounds: AdvertisementBounds,
) -> Result<Vec<BString>, SessionError>
where
R: gix_transport::client::async_io::ReadlineBufRead + Unpin,
{
drain_bounded_lines(reader, bounds, "the advertisement", |line| {
Ok(BString::from(line.as_bytes()))
})
.await
}
pub(crate) async fn drain_banded_payload<S>(
reader: &mut Box<dyn gix_transport::client::async_io::ExtendedBufRead<'_> + Unpin + '_>,
sink: &mut S,
band_handler: bool,
max_pack_size: u64,
what: &'static str,
section: &'static str,
) -> Result<u64, SessionError>
where
S: futures::io::AsyncWrite + Unpin,
{
use futures::io::{AsyncBufReadExt as _, AsyncWriteExt as _};
let band_error: Arc<Mutex<Option<String>>> = Arc::default();
if band_handler {
reader.set_progress_handler(Some(Box::new({
let band_error = Arc::clone(&band_error);
move |is_error: bool, text: &[u8]| {
if is_error {
*band_error.lock().unwrap_or_else(|e| e.into_inner()) =
Some(String::from_utf8_lossy(text).into_owned());
ControlFlow::Break(())
} else {
ControlFlow::Continue(())
}
}
})));
}
let base: &mut (dyn futures::io::AsyncBufRead + Unpin) = reader.as_mut();
let mut total = 0u64;
loop {
let delivered = base.fill_buf().await;
match delivered {
Ok(ready) => {
if ready.is_empty() {
break;
}
let len = ready.len();
if total + len as u64 > max_pack_size {
return Err(SessionError::BudgetExceeded {
what: what.to_owned(),
budget_name: "max_pack_size",
limit: max_pack_size,
});
}
sink.write_all(ready).await.map_err(|e| {
SessionError::Transport(format!("the pack consumer failed: {e}"))
})?;
total += len as u64;
base.consume_unpin(len);
}
Err(_) => {
return Err(
match band_error.lock().unwrap_or_else(|e| e.into_inner()).take() {
Some(text) => SessionError::Protocol(format!(
"{section} aborted by the peer's error band: {text}"
)),
None => SessionError::Transport(format!("{section}'s stream failed")),
},
);
}
}
}
sink.flush()
.await
.map_err(|e| SessionError::Transport(format!("the pack consumer failed: {e}")))?;
Ok(total)
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn the_session_wires_no_credentials_surface() {
assert!(
HANDSHAKE_EXTRA_PARAMETERS.is_empty(),
"no hand-wired handshake parameters: the ADR-016 grammar is closed and no credentials layer rides it"
);
}
#[test]
fn the_request_line_version_encoding_is_the_pinned_grammar() {
assert_eq!(desired_version(Service::UploadPack), Protocol::V2);
assert_eq!(desired_version(Service::ReceivePack), Protocol::V1);
}
#[test]
fn handshake_errors_map_typed() {
let refused = ClientError::Io(std::io::Error::other(gix_packetline::read::Error {
message: "access denied: owner/doubled".into(),
}));
assert_eq!(
map_client_error(refused),
SessionError::PeerRefused("access denied: owner/doubled".to_owned()),
"the peer's decline text survives the mapping"
);
assert!(matches!(
map_client_error(ClientError::Io(std::io::Error::other("gone"))),
SessionError::Transport(_)
));
assert!(matches!(
map_client_error(ClientError::ExpectedLine("text")),
SessionError::Protocol(_)
));
}
}