diff --git a/Examples/encode_pack.rs b/Examples/encode_pack.rs index ab53b071..7c159eac 100644 --- a/Examples/encode_pack.rs +++ b/Examples/encode_pack.rs @@ -1,7 +1,7 @@ //! An example demonstrating how to encode Git objects into a pack file using git-internal crate. //! This example creates several Blob objects from string data, encodes them into a pack file, //! and writes the resulting pack file to disk. -//! +//! //! Make sure to check the output directory for the generated pack file after running this example. //! The example assumes SHA-1 hashing for simplicity. diff --git a/Examples/http_server.rs b/Examples/http_server.rs new file mode 100644 index 00000000..30816a82 --- /dev/null +++ b/Examples/http_server.rs @@ -0,0 +1,536 @@ +//! Git smart HTTP server using git-internal protocol handlers. +//! +//! This example serves repositories from `GIT_REPO_ROOT` and uses the local `git` +//! binary for repository plumbing (show-ref, cat-file, update-ref). +//! Repo names map to: +//! - `/` if it is a bare repo (contains `objects/`) +//! - `/.git` if bare with `.git` suffix +//! - `//.git` for non-bare repos +//! +//! Quick test (two terminals): +//! A) Prepare a bare repo and start the server: +//! ```bash +//! mkdir -p /tmp/git-http-demo && git init --bare /tmp/git-http-demo/demo.git +//! GIT_REPO_ROOT=/tmp/git-http-demo cargo run --example http_server +//! ``` +//! This creates a server-side repo and starts the HTTP server on port 3000. +//! +//! B) Verify info/refs, push, then clone: +//! ```bash +//! curl -i "http://127.0.0.1:3000/demo.git/info/refs?service=git-receive-pack" +//! mkdir -p /tmp/demo-src && cd /tmp/demo-src +//! git init +//! git config user.name demo +//! git config user.email demo@example.com +//! echo hello > README.md +//! git add README.md +//! git commit -m "init" +//! git remote add origin http://127.0.0.1:3000/demo.git +//! git push -u origin main +//! git clone http://127.0.0.1:3000/demo.git /tmp/demo-clone +//! ``` +//! - The curl call checks the Git smart HTTP advertisement. +//! - The push exercises `receive-pack`; replace `main` with `master` if needed. +//! - The clone exercises `upload-pack`. + +use async_trait::async_trait; +use axum::{ + Router, + body::{Body, Bytes}, + extract::{Path, Query, State}, + http::{HeaderMap, StatusCode}, + response::{IntoResponse, Response}, + routing::{get, post}, +}; +use flate2::{Compression, write::ZlibEncoder}; +use futures::StreamExt; +use git_internal::{ + hash::{ObjectHash, get_hash_kind}, + internal::object::{ + ObjectTrait, + blob::Blob, + commit::Commit, + tree::{Tree, TreeItem, TreeItemMode}, + types::ObjectType, + }, + protocol::{ + core::{AuthenticationService, RepositoryAccess}, + http::HttpGitHandler, + types::{ProtocolError, ProtocolStream}, + }, +}; +use std::{ + collections::HashMap, + io::Write, + path::{Path as StdPath, PathBuf}, + str::FromStr, + sync::Arc, +}; +use tokio::process::Command; + +/// Repository Access +#[derive(Clone)] +struct FsRepository { + git_dir: Arc, +} + +impl FsRepository { + /// Create a new FsRepository for the given git directory. + fn new(git_dir: PathBuf) -> Self { + Self { + git_dir: Arc::new(git_dir), + } + } + /// Create a git command with the appropriate git-dir. + fn git_cmd(&self) -> Command { + let mut cmd = Command::new("git"); + cmd.arg("--git-dir").arg(&*self.git_dir); + cmd + } + /// Run a git command with the given arguments. + async fn run_git(&self, args: I) -> Result + where + I: IntoIterator, + S: AsRef, + { + self.git_cmd() + .args(args) + .output() + .await + .map_err(ProtocolError::Io) + } + /// Get the path to the objects directory. + fn objects_dir(&self) -> PathBuf { + self.git_dir.join("objects") + } + /// Write a loose object to the objects directory. + fn write_loose_object( + &self, + obj_type: ObjectType, + data: &[u8], + ) -> Result { + let hash = ObjectHash::from_type_and_data(obj_type, data); + let hex = hash.to_string(); + let (dir, file) = hex.split_at(2); + let obj_dir = self.objects_dir().join(dir); + let obj_path = obj_dir.join(file); + + if obj_path.exists() { + return Ok(hash); + } + + std::fs::create_dir_all(&obj_dir).map_err(ProtocolError::Io)?; + + let mut encoder = ZlibEncoder::new(Vec::new(), Compression::default()); + let header = format!("{obj_type} {}\0", data.len()); + encoder + .write_all(header.as_bytes()) + .map_err(ProtocolError::Io)?; + encoder.write_all(data).map_err(ProtocolError::Io)?; + let compressed = encoder.finish().map_err(ProtocolError::Io)?; + + std::fs::write(&obj_path, compressed).map_err(ProtocolError::Io)?; + Ok(hash) + } + /// Parse a raw tree listing into TreeItems. + fn parse_tree_listing(&self, raw: &[u8]) -> Result, ProtocolError> { + let mut items = Vec::new(); + let text = String::from_utf8_lossy(raw); + for line in text.lines() { + if line.is_empty() { + continue; + } + let (meta, name) = line + .split_once('\t') + .ok_or_else(|| ProtocolError::invalid_request("Invalid tree entry"))?; + let mut parts = meta.split_whitespace(); + let mode_raw = parts + .next() + .ok_or_else(|| ProtocolError::invalid_request("Missing tree mode"))?; + let _kind = parts + .next() + .ok_or_else(|| ProtocolError::invalid_request("Missing tree kind"))?; + let hash_str = parts + .next() + .ok_or_else(|| ProtocolError::invalid_request("Missing tree hash"))?; + + let mode_norm = mode_raw.trim_start_matches('0'); + let mode_bytes = if mode_norm.is_empty() { + b"0" + } else { + mode_norm.as_bytes() + }; + let mode = TreeItemMode::tree_item_type_from_bytes(mode_bytes) + .map_err(|e| ProtocolError::repository_error(e.to_string()))?; + let id = + ObjectHash::from_str(hash_str).map_err(|e| ProtocolError::repository_error(e))?; + + items.push(TreeItem::new(mode, id, name.to_string())); + } + Ok(items) + } +} + +#[async_trait] +impl RepositoryAccess for FsRepository { + /// Get all refs in the repository. + async fn get_repository_refs(&self) -> Result, ProtocolError> { + let output = self.run_git(["show-ref", "--head"]).await?; + if !output.status.success() && output.stdout.is_empty() { + return Ok(Vec::new()); + } + let text = String::from_utf8_lossy(&output.stdout); + let mut refs = Vec::new(); + for line in text.lines() { + if line.is_empty() { + continue; + } + let mut parts = line.split_whitespace(); + let hash = parts.next().unwrap_or_default().to_string(); + let name = parts.next().unwrap_or_default().to_string(); + if !hash.is_empty() && !name.is_empty() { + refs.push((name, hash)); + } + } + Ok(refs) + } + + /// Check if an object exists by hash. + async fn has_object(&self, object_hash: &str) -> Result { + let status = self + .git_cmd() + .arg("cat-file") + .arg("-e") + .arg(object_hash) + .status() + .await + .map_err(ProtocolError::Io)?; + Ok(status.success()) + } + + /// Get raw object data by hash. + async fn get_object(&self, object_hash: &str) -> Result, ProtocolError> { + let output = self.run_git(["cat-file", "-p", object_hash]).await?; + if !output.status.success() { + return Err(ProtocolError::ObjectNotFound(object_hash.to_string())); + } + Ok(output.stdout) + } + + /// Store pack data (not implemented in this example). + async fn store_pack_data(&self, _pack_data: &[u8]) -> Result<(), ProtocolError> { + Ok(()) + } + + /// Update a reference to point to a new hash. + async fn update_reference( + &self, + ref_name: &str, + old_hash: Option<&str>, + new_hash: &str, + ) -> Result<(), ProtocolError> { + let zero = ObjectHash::zero_str(get_hash_kind()); + if new_hash == zero { + let mut cmd = self.git_cmd(); + cmd.arg("update-ref").arg("-d").arg(ref_name); + if let Some(old) = old_hash { + cmd.arg(old); + } + let out = cmd.output().await.map_err(ProtocolError::Io)?; + if !out.status.success() { + return Err(ProtocolError::repository_error( + String::from_utf8_lossy(&out.stderr).to_string(), + )); + } + return Ok(()); + } + + let mut cmd = self.git_cmd(); + cmd.arg("update-ref").arg(ref_name).arg(new_hash); + if let Some(old) = old_hash { + cmd.arg(old); + } + let out = cmd.output().await.map_err(ProtocolError::Io)?; + if !out.status.success() { + return Err(ProtocolError::repository_error( + String::from_utf8_lossy(&out.stderr).to_string(), + )); + } + Ok(()) + } + + /// Get objects needed for a pack (not implemented in this example). + async fn get_objects_for_pack( + &self, + _wants: &[String], + _haves: &[String], + ) -> Result, ProtocolError> { + Ok(Vec::new()) + } + + /// Check if the repository has a default branch (main or master). + async fn has_default_branch(&self) -> Result { + let refs = self.get_repository_refs().await?; + Ok(refs + .iter() + .any(|(name, _)| name == "refs/heads/main" || name == "refs/heads/master")) + } + + /// Post-receive hook (not implemented in this example). + async fn post_receive_hook(&self) -> Result<(), ProtocolError> { + Ok(()) + } + + /// Get a Commit object by hash. + async fn get_commit(&self, commit_hash: &str) -> Result { + let output = self.run_git(["cat-file", "-p", commit_hash]).await?; + if !output.status.success() { + return Err(ProtocolError::ObjectNotFound(commit_hash.to_string())); + } + let hash = + ObjectHash::from_str(commit_hash).map_err(|e| ProtocolError::repository_error(e))?; + Commit::from_bytes(&output.stdout, hash) + .map_err(|e| ProtocolError::repository_error(e.to_string())) + } + + /// Get a Tree object by hash. + async fn get_tree(&self, tree_hash: &str) -> Result { + let output = self.run_git(["cat-file", "-p", tree_hash]).await?; + if !output.status.success() { + return Err(ProtocolError::ObjectNotFound(tree_hash.to_string())); + } + let id = ObjectHash::from_str(tree_hash).map_err(|e| ProtocolError::repository_error(e))?; + let items = self.parse_tree_listing(&output.stdout)?; + if items.is_empty() { + return Ok(Tree { + id, + tree_items: Vec::new(), + }); + } + Tree::from_tree_items(items).map_err(|e| ProtocolError::repository_error(e.to_string())) + } + + /// Get a Blob object by hash. + async fn get_blob(&self, blob_hash: &str) -> Result { + let output = self.run_git(["cat-file", "-p", blob_hash]).await?; + if !output.status.success() { + return Err(ProtocolError::ObjectNotFound(blob_hash.to_string())); + } + let hash = + ObjectHash::from_str(blob_hash).map_err(|e| ProtocolError::repository_error(e))?; + Blob::from_bytes(&output.stdout, hash) + .map_err(|e| ProtocolError::repository_error(e.to_string())) + } + + /// Handle unpacking of received pack objects. + async fn handle_pack_objects( + &self, + commits: Vec, + trees: Vec, + blobs: Vec, + ) -> Result<(), ProtocolError> { + // Store unpacked objects as loose objects (enough for this example server). + for blob in blobs { + let data = blob + .to_data() + .map_err(|e| ProtocolError::repository_error(format!("serialize blob: {e}")))?; + self.write_loose_object(ObjectType::Blob, &data)?; + } + for tree in trees { + let data = tree + .to_data() + .map_err(|e| ProtocolError::repository_error(format!("serialize tree: {e}")))?; + self.write_loose_object(ObjectType::Tree, &data)?; + } + for commit in commits { + let data = commit + .to_data() + .map_err(|e| ProtocolError::repository_error(format!("serialize commit: {e}")))?; + self.write_loose_object(ObjectType::Commit, &data)?; + } + Ok(()) + } +} + +/// Authentication Service that allows all requests. +#[derive(Clone)] +struct AllowAllAuth; + +#[async_trait] +impl AuthenticationService for AllowAllAuth { + async fn authenticate_http( + &self, + _headers: &HashMap, + ) -> Result<(), ProtocolError> { + Ok(()) + } + + async fn authenticate_ssh( + &self, + _username: &str, + _public_key: &[u8], + ) -> Result<(), ProtocolError> { + Ok(()) + } +} + +/// GitHTTP Handlers +#[derive(Clone)] +struct AppState { + repo_root: PathBuf, + auth: AllowAllAuth, +} + +#[tokio::main] +async fn main() { + // Use GIT_REPO_ROOT to locate repositories on disk. + let repo_root = std::env::var("GIT_REPO_ROOT").unwrap_or_else(|_| "./repos".to_string()); + + let state = AppState { + repo_root: PathBuf::from(repo_root), + auth: AllowAllAuth, + }; + + // Routes match Git smart HTTP endpoints. + let app = Router::new() + .route("/{repo}/info/refs", get(info_refs)) + .route("/{repo}/git-upload-pack", post(upload_pack)) + .route("/{repo}/git-receive-pack", post(receive_pack)) + .with_state(Arc::new(state)); + + let addr = "0.0.0.0:3000"; + println!("HTTP Git server on http://{addr}"); + println!( + "Repo root: {}", + std::env::var("GIT_REPO_ROOT").unwrap_or_else(|_| "./repos".into()) + ); + + let listener = tokio::net::TcpListener::bind(addr).await.unwrap(); + axum::serve(listener, app).await.unwrap(); +} + +async fn info_refs( + State(state): State>, + Path(repo_name): Path, + Query(params): Query>, + headers: HeaderMap, +) -> Response { + // info/refs advertises refs + capabilities for the requested service. + let Some(service) = params.get("service") else { + return (StatusCode::BAD_REQUEST, "missing service").into_response(); + }; + + let git_dir = match resolve_repo_path(&state.repo_root, &repo_name) { + Ok(p) => p, + Err(resp) => return resp, + }; + + let repo = FsRepository::new(git_dir); + let mut handler = HttpGitHandler::new(repo, state.auth.clone()); + + let request_path = format!("/{}/info/refs", repo_name); + let query = format!("service={}", service); + + if let Err(e) = handler.authenticate_http(&headers_to_map(&headers)).await { + return (StatusCode::UNAUTHORIZED, e.to_string()).into_response(); + } + + match handler.handle_info_refs(&request_path, &query).await { + Ok((data, content_type)) => { + ([(axum::http::header::CONTENT_TYPE, content_type)], data).into_response() + } + Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()).into_response(), + } +} + +async fn upload_pack( + State(state): State>, + Path(repo_name): Path, + headers: HeaderMap, + body: Bytes, +) -> Response { + let git_dir = match resolve_repo_path(&state.repo_root, &repo_name) { + Ok(p) => p, + Err(resp) => return resp, + }; + + let repo = FsRepository::new(git_dir); + let mut handler = HttpGitHandler::new(repo, state.auth.clone()); + let request_path = format!("/{}/git-upload-pack", repo_name); + + if let Err(e) = handler.authenticate_http(&headers_to_map(&headers)).await { + return (StatusCode::UNAUTHORIZED, e.to_string()).into_response(); + } + + // upload-pack returns a stream (pack data) so we stream it back to the client. + match handler.handle_upload_pack(&request_path, &body).await { + Ok((stream, content_type)) => { + let body = Body::from_stream(stream); + ([(axum::http::header::CONTENT_TYPE, content_type)], body).into_response() + } + Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()).into_response(), + } +} + +async fn receive_pack( + State(state): State>, + Path(repo_name): Path, + headers: HeaderMap, + body: Body, +) -> Response { + let git_dir = match resolve_repo_path(&state.repo_root, &repo_name) { + Ok(p) => p, + Err(resp) => return resp, + }; + + let repo = FsRepository::new(git_dir); + let mut handler = HttpGitHandler::new(repo, state.auth.clone()); + let request_path = format!("/{}/git-receive-pack", repo_name); + + if let Err(e) = handler.authenticate_http(&headers_to_map(&headers)).await { + return (StatusCode::UNAUTHORIZED, e.to_string()).into_response(); + } + + // Convert Axum body into ProtocolStream for git-internal. + let stream: ProtocolStream = Box::pin(body.into_data_stream().map(|r| { + r.map_err(|e| ProtocolError::Io(std::io::Error::new(std::io::ErrorKind::Other, e))) + })); + + match handler.handle_receive_pack(&request_path, stream).await { + Ok((stream, content_type)) => { + let body = Body::from_stream(stream); + ([(axum::http::header::CONTENT_TYPE, content_type)], body).into_response() + } + Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()).into_response(), + } +} + +fn resolve_repo_path(repo_root: &StdPath, repo: &str) -> Result { + // Reject traversal and map repo name to bare or non-bare layout. + if repo.is_empty() || repo.contains("..") || repo.contains('\\') || repo.contains('/') { + return Err((StatusCode::BAD_REQUEST, "invalid repo").into_response()); + } + + let direct = repo_root.join(repo); + let bare = repo_root.join(format!("{repo}.git")); + let non_bare = repo_root.join(repo).join(".git"); + + if direct.is_dir() && direct.join("objects").exists() { + return Ok(direct); + } + if bare.is_dir() && bare.join("objects").exists() { + return Ok(bare); + } + if non_bare.is_dir() && non_bare.join("objects").exists() { + return Ok(non_bare); + } + + Err((StatusCode::NOT_FOUND, "repo not found").into_response()) +} + +fn headers_to_map(headers: &HeaderMap) -> HashMap { + headers + .iter() + .filter_map(|(k, v)| v.to_str().ok().map(|s| (k.to_string(), s.to_string()))) + .collect() +} diff --git a/Examples/ssh_server.rs b/Examples/ssh_server.rs new file mode 100644 index 00000000..3adbf2ea --- /dev/null +++ b/Examples/ssh_server.rs @@ -0,0 +1,547 @@ +//! SSH exec handler for git-upload-pack / git-receive-pack. +//! +//! This example uses git-internal's `SshGitHandler` and the same `FsRepository` +//! as the HTTP example. It relies on an external SSH transport (OpenSSH or a +//! local wrapper) to provide stdin/stdout. +//! +//! Quick test without real SSHD (two terminals): +//! +//! Server side: +//! ```bash +//! cargo build --example ssh_server --manifest-path git-internal/Cargo.toml +//! rm -rf /tmp/git-ssh-demo +//! mkdir -p /tmp/git-ssh-demo +//! git init --bare /tmp/git-ssh-demo/demo.git +//! cat > /tmp/git-ssh-wrapper <<'EOF' +//! #!/bin/sh +//! shift +//! export SSH_ORIGINAL_COMMAND="$*" +//! export GIT_REPO_ROOT=/tmp/git-ssh-demo +//! exec /path/to/git-internal/target/debug/examples/ssh_server +//! EOF +//! chmod +x /tmp/git-ssh-wrapper +//! ``` +//! - Builds the example binary, prepares a bare repo, and creates a wrapper +//! that simulates an SSH server. +//! +//! Client side (push + clone): +//! ```bash +//! rm -rf /tmp/ssh-src +//! mkdir -p /tmp/ssh-src && cd /tmp/ssh-src +//! git init +//! git config user.name demo +//! git config user.email demo@example.com +//! echo hello > README.md +//! git add README.md +//! git commit -m "init" +//! git remote add origin ssh://dummy@dummy/demo.git +//! GIT_SSH_COMMAND=/tmp/git-ssh-wrapper git push -u origin main +//! rm -rf /tmp/ssh-clone +//! GIT_SSH_COMMAND=/tmp/git-ssh-wrapper git clone ssh://dummy@dummy/demo.git /tmp/ssh-clone +//! ``` +//! - `GIT_SSH_COMMAND` points Git at the wrapper instead of a real ssh binary. +//! - `main` can be replaced with `master` depending on your default branch. +//! +//! OpenSSH usage (real server): +//! - Install the binary on the server and wire it in `~/.ssh/authorized_keys`: +//! `command="/path/to/ssh_server" ssh-ed25519 AAAA...` +//! - Then clients can run: `git clone ssh://user@host/demo.git`. + +use async_trait::async_trait; +use bytes::{Bytes, BytesMut}; +use flate2::{Compression, write::ZlibEncoder}; +use futures::StreamExt; +use git_internal::{ + hash::{ObjectHash, get_hash_kind}, + internal::object::{ + ObjectTrait, + blob::Blob, + commit::Commit, + tree::{Tree, TreeItem, TreeItemMode}, + types::ObjectType, + }, + protocol::{ + core::{AuthenticationService, RepositoryAccess}, + ssh::{SshGitHandler, parse_ssh_command}, + types::{ProtocolError, ProtocolStream}, + utils::{read_pkt_line, read_until_white_space}, + }, +}; +use std::{ + collections::HashMap, + io::Write, + path::{Component, Path as StdPath, PathBuf}, + str::FromStr, + sync::Arc, +}; +use tokio::{ + io::{AsyncReadExt, AsyncWriteExt}, + process::Command, +}; +use tokio_util::io::ReaderStream; + +/// Repo implementation (same as HTTP example) +#[derive(Clone)] +struct FsRepository { + git_dir: Arc, +} + +impl FsRepository { + /// Create a new FsRepository for the given git directory. + fn new(git_dir: PathBuf) -> Self { + Self { + git_dir: Arc::new(git_dir), + } + } + + /// Create a git command with the appropriate git-dir. + fn git_cmd(&self) -> Command { + let mut cmd = Command::new("git"); + cmd.arg("--git-dir").arg(&*self.git_dir); + cmd + } + + /// Run a git command with the given arguments. + async fn run_git(&self, args: I) -> Result + where + I: IntoIterator, + S: AsRef, + { + self.git_cmd() + .args(args) + .output() + .await + .map_err(ProtocolError::Io) + } + + /// Get the path to the objects directory. + fn objects_dir(&self) -> PathBuf { + self.git_dir.join("objects") + } + + /// Write a loose object to the objects directory. + fn write_loose_object( + &self, + obj_type: ObjectType, + data: &[u8], + ) -> Result { + let hash = ObjectHash::from_type_and_data(obj_type, data); + let hex = hash.to_string(); + let (dir, file) = hex.split_at(2); + let obj_dir = self.objects_dir().join(dir); + let obj_path = obj_dir.join(file); + + if obj_path.exists() { + return Ok(hash); + } + + std::fs::create_dir_all(&obj_dir).map_err(ProtocolError::Io)?; + + let mut encoder = ZlibEncoder::new(Vec::new(), Compression::default()); + let header = format!("{obj_type} {}\0", data.len()); + encoder + .write_all(header.as_bytes()) + .map_err(ProtocolError::Io)?; + encoder.write_all(data).map_err(ProtocolError::Io)?; + let compressed = encoder.finish().map_err(ProtocolError::Io)?; + + std::fs::write(&obj_path, compressed).map_err(ProtocolError::Io)?; + Ok(hash) + } + + /// Parse a raw tree listing into TreeItems. + fn parse_tree_listing(&self, raw: &[u8]) -> Result, ProtocolError> { + let mut items = Vec::new(); + let text = String::from_utf8_lossy(raw); + for line in text.lines() { + if line.is_empty() { + continue; + } + let (meta, name) = line + .split_once('\t') + .ok_or_else(|| ProtocolError::invalid_request("Invalid tree entry"))?; + let mut parts = meta.split_whitespace(); + let mode_raw = parts + .next() + .ok_or_else(|| ProtocolError::invalid_request("Missing tree mode"))?; + let _kind = parts + .next() + .ok_or_else(|| ProtocolError::invalid_request("Missing tree kind"))?; + let hash_str = parts + .next() + .ok_or_else(|| ProtocolError::invalid_request("Missing tree hash"))?; + + let mode_norm = mode_raw.trim_start_matches('0'); + let mode_bytes = if mode_norm.is_empty() { + b"0" + } else { + mode_norm.as_bytes() + }; + let mode = TreeItemMode::tree_item_type_from_bytes(mode_bytes) + .map_err(|e| ProtocolError::repository_error(e.to_string()))?; + let id = + ObjectHash::from_str(hash_str).map_err(|e| ProtocolError::repository_error(e))?; + + items.push(TreeItem::new(mode, id, name.to_string())); + } + Ok(items) + } +} + +#[async_trait] +impl RepositoryAccess for FsRepository { + async fn get_repository_refs(&self) -> Result, ProtocolError> { + let output = self.run_git(["show-ref", "--head"]).await?; + if !output.status.success() && output.stdout.is_empty() { + return Ok(Vec::new()); + } + let text = String::from_utf8_lossy(&output.stdout); + let mut refs = Vec::new(); + for line in text.lines() { + if line.is_empty() { + continue; + } + let mut parts = line.split_whitespace(); + let hash = parts.next().unwrap_or_default().to_string(); + let name = parts.next().unwrap_or_default().to_string(); + if !hash.is_empty() && !name.is_empty() { + refs.push((name, hash)); + } + } + Ok(refs) + } + + async fn has_object(&self, object_hash: &str) -> Result { + let status = self + .git_cmd() + .arg("cat-file") + .arg("-e") + .arg(object_hash) + .status() + .await + .map_err(ProtocolError::Io)?; + Ok(status.success()) + } + + async fn get_object(&self, object_hash: &str) -> Result, ProtocolError> { + let output = self.run_git(["cat-file", "-p", object_hash]).await?; + if !output.status.success() { + return Err(ProtocolError::ObjectNotFound(object_hash.to_string())); + } + Ok(output.stdout) + } + + async fn store_pack_data(&self, _pack_data: &[u8]) -> Result<(), ProtocolError> { + Ok(()) + } + + async fn update_reference( + &self, + ref_name: &str, + old_hash: Option<&str>, + new_hash: &str, + ) -> Result<(), ProtocolError> { + let zero = ObjectHash::zero_str(get_hash_kind()); + if new_hash == zero { + let mut cmd = self.git_cmd(); + cmd.arg("update-ref").arg("-d").arg(ref_name); + if let Some(old) = old_hash { + cmd.arg(old); + } + let out = cmd.output().await.map_err(ProtocolError::Io)?; + if !out.status.success() { + return Err(ProtocolError::repository_error( + String::from_utf8_lossy(&out.stderr).to_string(), + )); + } + return Ok(()); + } + + let mut cmd = self.git_cmd(); + cmd.arg("update-ref").arg(ref_name).arg(new_hash); + if let Some(old) = old_hash { + cmd.arg(old); + } + let out = cmd.output().await.map_err(ProtocolError::Io)?; + if !out.status.success() { + return Err(ProtocolError::repository_error( + String::from_utf8_lossy(&out.stderr).to_string(), + )); + } + Ok(()) + } + + async fn get_objects_for_pack( + &self, + _wants: &[String], + _haves: &[String], + ) -> Result, ProtocolError> { + Ok(Vec::new()) + } + + async fn has_default_branch(&self) -> Result { + let refs = self.get_repository_refs().await?; + Ok(refs + .iter() + .any(|(name, _)| name == "refs/heads/main" || name == "refs/heads/master")) + } + + async fn post_receive_hook(&self) -> Result<(), ProtocolError> { + Ok(()) + } + + async fn get_commit(&self, commit_hash: &str) -> Result { + let output = self.run_git(["cat-file", "-p", commit_hash]).await?; + if !output.status.success() { + return Err(ProtocolError::ObjectNotFound(commit_hash.to_string())); + } + let hash = + ObjectHash::from_str(commit_hash).map_err(|e| ProtocolError::repository_error(e))?; + Commit::from_bytes(&output.stdout, hash) + .map_err(|e| ProtocolError::repository_error(e.to_string())) + } + + async fn get_tree(&self, tree_hash: &str) -> Result { + let output = self.run_git(["cat-file", "-p", tree_hash]).await?; + if !output.status.success() { + return Err(ProtocolError::ObjectNotFound(tree_hash.to_string())); + } + let id = ObjectHash::from_str(tree_hash).map_err(|e| ProtocolError::repository_error(e))?; + let items = self.parse_tree_listing(&output.stdout)?; + if items.is_empty() { + return Ok(Tree { + id, + tree_items: Vec::new(), + }); + } + Tree::from_tree_items(items).map_err(|e| ProtocolError::repository_error(e.to_string())) + } + + async fn get_blob(&self, blob_hash: &str) -> Result { + let output = self.run_git(["cat-file", "-p", blob_hash]).await?; + if !output.status.success() { + return Err(ProtocolError::ObjectNotFound(blob_hash.to_string())); + } + let hash = + ObjectHash::from_str(blob_hash).map_err(|e| ProtocolError::repository_error(e))?; + Blob::from_bytes(&output.stdout, hash) + .map_err(|e| ProtocolError::repository_error(e.to_string())) + } + + async fn handle_pack_objects( + &self, + commits: Vec, + trees: Vec, + blobs: Vec, + ) -> Result<(), ProtocolError> { + // Store unpacked objects as loose objects (enough for this example server). + for blob in blobs { + let data = blob + .to_data() + .map_err(|e| ProtocolError::repository_error(format!("serialize blob: {e}")))?; + self.write_loose_object(ObjectType::Blob, &data)?; + } + for tree in trees { + let data = tree + .to_data() + .map_err(|e| ProtocolError::repository_error(format!("serialize tree: {e}")))?; + self.write_loose_object(ObjectType::Tree, &data)?; + } + for commit in commits { + let data = commit + .to_data() + .map_err(|e| ProtocolError::repository_error(format!("serialize commit: {e}")))?; + self.write_loose_object(ObjectType::Commit, &data)?; + } + Ok(()) + } +} + +/// Auth +#[derive(Clone)] +struct AllowAllAuth; + +#[async_trait] +impl AuthenticationService for AllowAllAuth { + async fn authenticate_http( + &self, + _headers: &HashMap, + ) -> Result<(), ProtocolError> { + Ok(()) + } + + async fn authenticate_ssh( + &self, + _username: &str, + _public_key: &[u8], + ) -> Result<(), ProtocolError> { + Ok(()) + } +} + +/// SSH Exec Handler +#[tokio::main] +async fn main() { + // Repo root is the base directory for served repositories. + let repo_root = std::env::var("GIT_REPO_ROOT").unwrap_or_else(|_| "./repos".to_string()); + // SSH daemon passes the requested command here. + let command_line = std::env::var("SSH_ORIGINAL_COMMAND") + .ok() + .or_else(|| { + let args = std::env::args().skip(1).collect::>(); + if args.is_empty() { + None + } else { + Some(args.join(" ")) + } + }) + .unwrap_or_else(|| { + eprintln!("missing SSH_ORIGINAL_COMMAND"); + std::process::exit(1); + }); + + let Some((command, args)) = parse_ssh_command(&command_line) else { + eprintln!("invalid command"); + std::process::exit(1); + }; + + // Extract repo argument (first non-flag token). + let repo_arg = args + .iter() + .find(|a| !a.starts_with('-')) + .map(|s| s.trim_matches('\'').trim_matches('"').to_string()) + .unwrap_or_else(|| { + eprintln!("missing repo argument"); + std::process::exit(1); + }); + + let git_dir = match resolve_repo_path(&PathBuf::from(repo_root), &repo_arg) { + Ok(p) => p, + Err(msg) => { + eprintln!("{msg}"); + std::process::exit(1); + } + }; + + let repo = FsRepository::new(git_dir); + let auth = AllowAllAuth; + let mut handler = SshGitHandler::new(repo, auth); + + let username = std::env::var("USER").unwrap_or_else(|_| "unknown".to_string()); + if let Err(e) = handler.authenticate_ssh(&username, &[]).await { + eprintln!("auth failed: {e}"); + std::process::exit(1); + } + + let mut stdout = tokio::io::stdout(); + let service = match command.as_str() { + "git-upload-pack" => "git-upload-pack", + "git-receive-pack" => "git-receive-pack", + _ => { + eprintln!("unsupported command: {command}"); + std::process::exit(1); + } + }; + + // SSH protocol starts by advertising refs/capabilities. + let info_refs = handler.handle_info_refs(service).await.unwrap(); + stdout.write_all(&info_refs).await.unwrap(); + stdout.flush().await.unwrap(); + + let output: ProtocolStream = match command.as_str() { + "git-upload-pack" => { + // Read request until "done" or flush to avoid blocking. + let request = read_upload_pack_request().await.unwrap(); + handler.handle_upload_pack(&request).await.unwrap() + } + "git-receive-pack" => { + // Stream pack data directly from stdin. + let stream = ReaderStream::new(tokio::io::stdin()) + .map(|result| result.map_err(ProtocolError::Io)); + let stream: ProtocolStream = Box::pin(stream); + handler.handle_receive_pack(stream).await.unwrap() + } + _ => unreachable!("unsupported command: {command}"), + }; + + let mut stream = output; + while let Some(chunk) = stream.next().await { + let data = chunk.unwrap(); + stdout.write_all(&data).await.unwrap(); + } + if let Err(err) = stdout.flush().await { + eprintln!("failed to flush stdout to client: {err}"); + std::process::exit(1); + } +} + +fn upload_pack_request_complete(buf: &BytesMut) -> bool { + // Scan pkt-lines; request ends at flush or "done". + let mut view = buf.clone().freeze(); + loop { + let (consumed, pkt_line) = read_pkt_line(&mut view); + if consumed == 0 { + return false; + } + if pkt_line.is_empty() { + return true; + } + let mut pkt_line = pkt_line; + let command = read_until_white_space(&mut pkt_line); + if command == "done" { + return true; + } + } +} + +async fn read_upload_pack_request() -> Result { + // Read stdin until the upload-pack request is complete. + let mut stdin = tokio::io::stdin(); + let mut buf = BytesMut::with_capacity(8192); + let mut temp = [0u8; 8192]; + + loop { + let n = stdin.read(&mut temp).await.map_err(ProtocolError::Io)?; + if n == 0 { + break; + } + buf.extend_from_slice(&temp[..n]); + if upload_pack_request_complete(&buf) { + break; + } + } + + Ok(buf.freeze()) +} + +fn resolve_repo_path(repo_root: &StdPath, repo: &str) -> Result { + // Reject traversal and map repo name to bare or non-bare layout. + let repo = repo.trim_start_matches('/'); + if repo.is_empty() || repo.contains('\\') { + return Err("invalid repo".to_string()); + } + + let repo_path = StdPath::new(repo); + if repo_path + .components() + .any(|c| matches!(c, Component::ParentDir)) + { + return Err("invalid repo".to_string()); + } + + let direct = repo_root.join(repo); + let bare = repo_root.join(format!("{repo}.git")); + let non_bare = repo_root.join(repo).join(".git"); + + if direct.is_dir() && direct.join("objects").exists() { + return Ok(direct); + } + if bare.is_dir() && bare.join("objects").exists() { + return Ok(bare); + } + if non_bare.is_dir() && non_bare.join("objects").exists() { + return Ok(non_bare); + } + + Err("repo not found".to_string()) +} diff --git a/src/protocol/core.rs b/src/protocol/core.rs index 1041f4d4..810640af 100644 --- a/src/protocol/core.rs +++ b/src/protocol/core.rs @@ -5,7 +5,7 @@ use std::{collections::HashMap, str::FromStr}; use async_trait::async_trait; -use bytes::Bytes; +use bytes::{BufMut, Bytes, BytesMut}; use futures::stream::StreamExt; use crate::{ @@ -13,7 +13,7 @@ use crate::{ internal::object::ObjectTrait, protocol::{ smart::SmartProtocol, - types::{ProtocolError, ProtocolStream, ServiceType}, + types::{Capability, ProtocolError, ProtocolStream, ServiceType, SideBand}, }, }; @@ -252,9 +252,51 @@ impl GitProtocol { &mut self, request_data: &[u8], ) -> Result { + const SIDE_BAND_PACKET_LEN: usize = 1000; + const SIDE_BAND_64K_PACKET_LEN: usize = 65520; + const SIDE_BAND_HEADER_LEN: usize = 5; // 4-byte length + 1-byte band + let request_bytes = bytes::Bytes::from(request_data.to_vec()); - let (stream, _) = self.smart_protocol.git_upload_pack(request_bytes).await?; - Ok(Box::pin(stream.map(|data| Ok(Bytes::from(data))))) + let (pack_stream, protocol_buf) = + self.smart_protocol.git_upload_pack(request_bytes).await?; + let ack_bytes = protocol_buf.freeze(); + + let ack_stream: ProtocolStream = if ack_bytes.is_empty() { + Box::pin(futures::stream::empty::>()) + } else { + Box::pin(futures::stream::once(async move { Ok(ack_bytes) })) + }; + + let sideband_max = if self + .smart_protocol + .capabilities + .contains(&Capability::SideBand64k) + { + Some(SIDE_BAND_64K_PACKET_LEN - SIDE_BAND_HEADER_LEN) + } else if self + .smart_protocol + .capabilities + .contains(&Capability::SideBand) + { + Some(SIDE_BAND_PACKET_LEN - SIDE_BAND_HEADER_LEN) + } else { + None + }; + + let data_stream: ProtocolStream = if let Some(max_payload) = sideband_max { + let stream = pack_stream.flat_map(move |chunk| { + let packets = build_side_band_packets(&chunk, max_payload); + futures::stream::iter(packets.into_iter().map(Ok)) + }); + let stream = stream.chain(futures::stream::once(async { + Ok(Bytes::from_static(b"0000")) + })); + Box::pin(stream) + } else { + Box::pin(pack_stream.map(|data| Ok(Bytes::from(data)))) + }; + + Ok(Box::pin(ack_stream.chain(data_stream))) } /// Handle git-receive-pack request (for push) @@ -262,21 +304,83 @@ impl GitProtocol { &mut self, request_stream: ProtocolStream, ) -> Result { + const SIDE_BAND_PACKET_LEN: usize = 1000; + const SIDE_BAND_64K_PACKET_LEN: usize = 65520; + const SIDE_BAND_HEADER_LEN: usize = 5; // 4-byte length + 1-byte band + let result_bytes = self .smart_protocol .git_receive_pack_stream(request_stream) .await?; - // Return the report status as a single-chunk stream - Ok(Box::pin(futures::stream::once(async { Ok(result_bytes) }))) + + let sideband_max = if self + .smart_protocol + .capabilities + .contains(&Capability::SideBand64k) + { + Some(SIDE_BAND_64K_PACKET_LEN - SIDE_BAND_HEADER_LEN) + } else if self + .smart_protocol + .capabilities + .contains(&Capability::SideBand) + { + Some(SIDE_BAND_PACKET_LEN - SIDE_BAND_HEADER_LEN) + } else { + None + }; + + // Wrap report-status in side-band if negotiated by the client. + if let Some(max_payload) = sideband_max { + let packets = build_side_band_packets(result_bytes.as_ref(), max_payload); + let stream = futures::stream::iter(packets.into_iter().map(Ok)).chain( + futures::stream::once(async { Ok(Bytes::from_static(b"0000")) }), + ); + Ok(Box::pin(stream)) + } else { + // Return the report status as a single-chunk stream + Ok(Box::pin(futures::stream::once(async { Ok(result_bytes) }))) + } } } +fn build_side_band_packets(chunk: &[u8], max_payload: usize) -> Vec { + if chunk.is_empty() { + return Vec::new(); + } + + let mut out = Vec::new(); + let mut offset = 0; + + while offset < chunk.len() { + let end = (offset + max_payload).min(chunk.len()); + let payload = &chunk[offset..end]; + let length = payload.len() + 5; // 4-byte length + 1-byte band + let mut pkt = BytesMut::with_capacity(length); + pkt.put(Bytes::from(format!("{length:04x}"))); + pkt.put_u8(SideBand::PackfileData.value()); + pkt.put(payload); + out.push(pkt.freeze()); + offset = end; + } + + out +} + #[cfg(test)] mod tests { use super::*; - use crate::hash::HashKind; + use crate::hash::{HashKind, set_hash_kind_for_test}; + use crate::internal::object::{ + blob::Blob, + commit::Commit, + signature::{Signature, SignatureType}, + tree::{Tree, TreeItem, TreeItemMode}, + }; use crate::protocol::types::TransportProtocol; + use crate::protocol::utils; use async_trait::async_trait; + use bytes::{Bytes, BytesMut}; + use futures::StreamExt; /// Simple mock repository that serves fixed refs and echoes wants. #[derive(Clone)] @@ -356,6 +460,168 @@ mod tests { ) } + /// Mock repo that serves a single commit, tree, and blobs. + #[derive(Clone)] + struct SideBandRepo { + commit: Commit, + tree: Tree, + blobs: Vec, + } + #[async_trait] + impl RepositoryAccess for SideBandRepo { + async fn get_repository_refs(&self) -> Result, ProtocolError> { + Ok(vec![( + "refs/heads/main".to_string(), + self.commit.id.to_string(), + )]) + } + + async fn has_object(&self, object_hash: &str) -> Result { + let known = object_hash == self.commit.id.to_string() + || object_hash == self.tree.id.to_string() + || self.blobs.iter().any(|b| b.id.to_string() == object_hash); + Ok(known) + } + + async fn get_object(&self, _object_hash: &str) -> Result, ProtocolError> { + Ok(Vec::new()) + } + + async fn store_pack_data(&self, _pack_data: &[u8]) -> Result<(), ProtocolError> { + Ok(()) + } + + async fn update_reference( + &self, + _ref_name: &str, + _old_hash: Option<&str>, + _new_hash: &str, + ) -> Result<(), ProtocolError> { + Ok(()) + } + + async fn get_objects_for_pack( + &self, + _wants: &[String], + _haves: &[String], + ) -> Result, ProtocolError> { + Ok(Vec::new()) + } + + async fn has_default_branch(&self) -> Result { + Ok(true) + } + + async fn post_receive_hook(&self) -> Result<(), ProtocolError> { + Ok(()) + } + + async fn get_commit(&self, commit_hash: &str) -> Result { + if commit_hash == self.commit.id.to_string() { + Ok(self.commit.clone()) + } else { + Err(ProtocolError::ObjectNotFound(commit_hash.to_string())) + } + } + + async fn get_tree(&self, tree_hash: &str) -> Result { + if tree_hash == self.tree.id.to_string() { + Ok(self.tree.clone()) + } else { + Err(ProtocolError::ObjectNotFound(tree_hash.to_string())) + } + } + + async fn get_blob(&self, blob_hash: &str) -> Result { + self.blobs + .iter() + .find(|b| b.id.to_string() == blob_hash) + .cloned() + .ok_or_else(|| ProtocolError::ObjectNotFound(blob_hash.to_string())) + } + } + + fn build_repo_with_objects() -> (SideBandRepo, Commit) { + let blob = Blob::from_content("hello"); + let item = TreeItem::new(TreeItemMode::Blob, blob.id, "hello.txt".to_string()); + let tree = Tree::from_tree_items(vec![item]).unwrap(); + let author = Signature::new( + SignatureType::Author, + "tester".to_string(), + "tester@example.com".to_string(), + ); + let committer = Signature::new( + SignatureType::Committer, + "tester".to_string(), + "tester@example.com".to_string(), + ); + let commit = Commit::new(author, committer, tree.id, vec![], "init commit"); + + let repo = SideBandRepo { + commit: commit.clone(), + tree, + blobs: vec![blob], + }; + + (repo, commit) + } + + /// upload-pack should emit NAK before sending pack data. + #[tokio::test] + async fn upload_pack_emits_ack_before_pack() { + let _guard = set_hash_kind_for_test(HashKind::Sha1); + let (repo, commit) = build_repo_with_objects(); + let mut proto = GitProtocol::new(repo, MockAuth); + let mut request = BytesMut::new(); + utils::add_pkt_line_string(&mut request, format!("want {}\n", commit.id)); + utils::add_pkt_line_string(&mut request, "done\n".to_string()); + + let mut stream = proto.upload_pack(&request).await.expect("upload-pack"); + let mut out = BytesMut::new(); + while let Some(chunk) = stream.next().await { + out.extend_from_slice(&chunk.expect("stream chunk")); + } + + let mut out_bytes = out.freeze(); + let (_len, line) = utils::read_pkt_line(&mut out_bytes); + assert_eq!(line, Bytes::from_static(b"NAK\n")); + assert!( + out_bytes.as_ref().starts_with(b"PACK"), + "pack should follow ack" + ); + } + + /// upload-pack with side-band should wrap pack data in side-band packets. + #[tokio::test] + async fn upload_pack_sideband_frames_pack() { + let _guard = set_hash_kind_for_test(HashKind::Sha1); + let (repo, commit) = build_repo_with_objects(); + + let mut proto = GitProtocol::new(repo, MockAuth); + let mut request = BytesMut::new(); + utils::add_pkt_line_string(&mut request, format!("want {} side-band-64k\n", commit.id)); + utils::add_pkt_line_string(&mut request, "done\n".to_string()); + + let mut stream = proto.upload_pack(&request).await.expect("upload-pack"); + let mut out = BytesMut::new(); + while let Some(chunk) = stream.next().await { + out.extend_from_slice(&chunk.expect("stream chunk")); + } + + let mut out_bytes = out.freeze(); + let (_len, line) = utils::read_pkt_line(&mut out_bytes); + assert_eq!(line, Bytes::from_static(b"NAK\n")); + + let raw = out_bytes.as_ref(); + assert!(raw.len() > 9, "side-band packet should include PACK header"); + let len_hex = std::str::from_utf8(&raw[..4]).expect("hex length"); + let pkt_len = usize::from_str_radix(len_hex, 16).expect("parse length"); + assert!(pkt_len > 5, "side-band packet should contain data"); + assert_eq!(raw[4], SideBand::PackfileData.value()); + assert_eq!(&raw[5..9], b"PACK"); + assert!(raw.ends_with(b"0000"), "side-band stream should flush"); + } + /// info_refs should include refs, capabilities, and object-format. #[tokio::test] async fn info_refs_includes_refs_and_caps() { diff --git a/src/protocol/smart.rs b/src/protocol/smart.rs index cc0f7a40..5f0e4654 100644 --- a/src/protocol/smart.rs +++ b/src/protocol/smart.rs @@ -158,6 +158,8 @@ where &mut self, upload_request: Bytes, ) -> Result<(ReceiverStream>, BytesMut), ProtocolError> { + self.capabilities.clear(); + self.set_wire_hash_kind(self.local_hash_kind); let mut upload_request = upload_request; let mut want: Vec = Vec::new(); let mut have: Vec = Vec::new(); @@ -238,9 +240,6 @@ where &mut protocol_buf, format!("ACK {last_common_commit} ready\n"), ); - protocol_buf.put(&PKT_LINE_END_MARKER[..]); - - add_pkt_line_string(&mut protocol_buf, format!("ACK {last_common_commit} \n")); let pack_stream = pack_generator.generate_incremental_pack(want, have).await?; @@ -249,6 +248,10 @@ where /// Parse receive pack commands from protocol bytes pub fn parse_receive_pack_commands(&mut self, mut protocol_bytes: Bytes) { + self.command_list.clear(); + self.capabilities.clear(); + self.set_wire_hash_kind(self.local_hash_kind); + let mut first_line = true; loop { let (bytes_take, pkt_line) = read_pkt_line(&mut protocol_bytes); @@ -260,6 +263,14 @@ where break; } + if first_line { + if let Some(pos) = pkt_line.iter().position(|b| *b == b'\0') { + let caps = String::from_utf8_lossy(&pkt_line[(pos + 1)..]).to_string(); + self.parse_capabilities(&caps); + } + first_line = false; + } + let ref_command = self.parse_ref_command(&mut pkt_line.clone()); self.command_list.push(ref_command); } @@ -270,29 +281,78 @@ where &mut self, data_stream: ProtocolStream, ) -> Result { - // Collect all pack data from stream - let mut pack_data = BytesMut::new(); + // Collect all request data from stream + let mut request_data = BytesMut::new(); let mut stream = data_stream; while let Some(chunk_result) = futures::StreamExt::next(&mut stream).await { let chunk = chunk_result .map_err(|e| ProtocolError::invalid_request(&format!("Stream error: {e}")))?; - pack_data.extend_from_slice(&chunk); + request_data.extend_from_slice(&chunk); } - // Create pack generator for unpacking - let pack_generator = PackGenerator::new(&self.repo_storage); + let mut protocol_bytes = request_data.freeze(); + self.command_list.clear(); + self.capabilities.clear(); + self.set_wire_hash_kind(self.local_hash_kind); + let mut first_line = true; + let mut saw_flush = false; + loop { + let (bytes_take, pkt_line) = read_pkt_line(&mut protocol_bytes); - // Unpack the received data - let (commits, trees, blobs) = pack_generator.unpack_stream(pack_data.freeze()).await?; + if bytes_take == 0 { + if protocol_bytes.is_empty() { + break; + } + return Err(ProtocolError::invalid_request( + "Invalid pkt-line in receive-pack request", + )); + } - // Store the unpacked objects via the repository access trait - self.repo_storage - .handle_pack_objects(commits, trees, blobs) - .await - .map_err(|e| { - ProtocolError::repository_error(format!("Failed to store pack objects: {e}")) - })?; + if pkt_line.is_empty() { + saw_flush = true; + break; + } + + if first_line { + if let Some(pos) = pkt_line.iter().position(|b| *b == b'\0') { + let caps = String::from_utf8_lossy(&pkt_line[(pos + 1)..]).to_string(); + self.parse_capabilities(&caps); + } + first_line = false; + } + + let ref_command = self.parse_ref_command(&mut pkt_line.clone()); + self.command_list.push(ref_command); + } + + if !saw_flush { + return Err(ProtocolError::invalid_request( + "Missing flush before pack data", + )); + } + + // Remaining bytes (if any) are pack data. + let pack_data = if protocol_bytes.is_empty() { + None + } else { + Some(protocol_bytes) + }; + + if let Some(pack_data) = pack_data { + // Create pack generator for unpacking + let pack_generator = PackGenerator::new(&self.repo_storage); + // Unpack the received data + let (commits, trees, blobs) = pack_generator.unpack_stream(pack_data).await?; + + // Store the unpacked objects via the repository access trait + self.repo_storage + .handle_pack_objects(commits, trees, blobs) + .await + .map_err(|e| { + ProtocolError::repository_error(format!("Failed to store pack objects: {e}")) + })?; + } // Build status report let mut report_status = BytesMut::new(); @@ -405,12 +465,11 @@ mod tests { }; use async_trait::async_trait; - use bytes::{Bytes, BytesMut}; + use bytes::BytesMut; use futures; use tokio::sync::mpsc; use super::*; - use crate::protocol::types::RefCommand; // import sibling types use crate::protocol::utils; // import sibling module use crate::{ hash::{HashKind, ObjectHash, set_hash_kind_for_test}, @@ -615,19 +674,25 @@ mod tests { pack_bytes.extend_from_slice(&chunk); } - // Prepare protocol and command + // Prepare protocol and request let repo_access = TestRepoAccess::new(); let auth = TestAuth; let mut smart = SmartProtocol::new(TransportProtocol::Http, repo_access.clone(), auth); smart.set_wire_hash_kind(HashKind::Sha1); - smart.command_list.push(RefCommand::new( - smart.zero_id.to_string(), - commit.id.to_string(), - "refs/heads/main".to_string(), - )); + + let mut request = BytesMut::new(); + add_pkt_line_string( + &mut request, + format!( + "{} {} refs/heads/main\0report-status\n", + smart.zero_id, commit.id + ), + ); + request.put(&PKT_LINE_END_MARKER[..]); + request.extend_from_slice(&pack_bytes); // Create request stream - let request_stream = Box::pin(futures::stream::once(async { Ok(Bytes::from(pack_bytes)) })); + let request_stream = Box::pin(futures::stream::once(async { Ok(request.freeze()) })); // Execute receive-pack let result_bytes = smart @@ -786,4 +851,24 @@ mod tests { assert_eq!(smart.command_list[0].ref_name, "refs/heads/main"); assert_eq!(smart.command_list[1].ref_name, "refs/tags/v1.0"); } + + /// receive-pack should error if ref commands are not terminated by a flush. + #[tokio::test] + async fn receive_pack_missing_flush_errors() { + let _guard = set_hash_kind_for_test(HashKind::Sha1); + let repo_access = TestRepoAccess::new(); + let auth = TestAuth; + let mut smart = SmartProtocol::new(TransportProtocol::Http, repo_access, auth); + + let zero = ObjectHash::zero_str(HashKind::Sha1); + let mut pkt = BytesMut::new(); + add_pkt_line_string(&mut pkt, format!("{zero} {zero} refs/heads/main\n")); + + let request_stream = Box::pin(futures::stream::once(async { Ok(pkt.freeze()) })); + let err = smart + .git_receive_pack_stream(request_stream) + .await + .unwrap_err(); + assert!(matches!(err, ProtocolError::InvalidRequest(_))); + } } diff --git a/src/protocol/utils.rs b/src/protocol/utils.rs index ee749742..9c487848 100644 --- a/src/protocol/utils.rs +++ b/src/protocol/utils.rs @@ -20,7 +20,7 @@ pub fn read_pkt_line(bytes: &mut Bytes) -> (usize, Bytes) { return (0, Bytes::new()); } - let pkt_length = bytes.copy_to_bytes(4); + let pkt_length = bytes.slice(0..4); let pkt_length_str = match core::str::from_utf8(&pkt_length) { Ok(s) => s, Err(_) => { @@ -38,6 +38,7 @@ pub fn read_pkt_line(bytes: &mut Bytes) -> (usize, Bytes) { }; if pkt_length == 0 { + bytes.advance(4); return (4, Bytes::new()); // Consumed 4 bytes for the "0000" marker } @@ -46,17 +47,18 @@ pub fn read_pkt_line(bytes: &mut Bytes) -> (usize, Bytes) { return (0, Bytes::new()); } - let data_length = pkt_length - 4; - if bytes.len() < data_length { + if bytes.len() < pkt_length { tracing::warn!( "Insufficient data: need {} bytes, have {}", - data_length, + pkt_length, bytes.len() ); return (0, Bytes::new()); } // this operation will change the original bytes + bytes.advance(4); + let data_length = pkt_length - 4; let pkt_line = bytes.copy_to_bytes(data_length); tracing::debug!("pkt line: {:?}", pkt_line); @@ -120,3 +122,20 @@ pub fn search_subsequence(haystack: &[u8], needle: &[u8]) -> Option { .windows(needle.len()) .position(|window| window == needle) } + +#[cfg(test)] +mod tests { + use super::*; + use bytes::Bytes; + + /// Test that read_pkt_line correctly reads a complete packet line + #[test] + fn read_pkt_line_incomplete_does_not_consume() { + let mut buf = Bytes::from_static(b"0009do"); + let before = buf.len(); + let (len, data) = read_pkt_line(&mut buf); + assert_eq!(len, 0); + assert!(data.is_empty()); + assert_eq!(buf.len(), before); + } +}