2ae9196
ZhenYi
feat(lib/git): add HTTP handler tree with routes and errors across analysis, archives, blobs, branches, commits, conflicts, content, diffs, hooks, lfs, logs, maintenance, packs, patches, pipelines, refs, remotes, repos, signatures, submodules, tags, trees
2ae9196
use std::sync::Arc;
use bytes::Bytes;
use cache::CacheClient;
use russh::ChannelId;
use russh::server::Session;
use sea_orm::EntityTrait;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use crate::context::facade::GitContext;
use crate::smart::error::SmartError;
use crate::smart::pre_receive::parse_receive_pack_refs;
use crate::ssh::error::SshError;
use crate::ssh::git_service::apply_hardened_env;
use crate::ssh::handler::GitSshHandler;
use crate::sync::post_receive::enqueue_post_receive_push;
pub const MAX_SSH_CHANNEL_BUF: usize = 64 * 1024 * 1024;
pub const MAX_GIT_OUTPUT_BUF: usize = 256 * 1024 * 1024;
impl GitSshHandler {
pub async fn data_impl(
&mut self,
channel: ChannelId,
data: &[u8],
session: &mut Session,
) -> Result<(), russh::Error> {
let state = self.channels.entry(channel).or_default();
if state.buffer_overflow {
return Ok(());
}
let projected = state.buffer.len().saturating_add(data.len());
if projected > MAX_SSH_CHANNEL_BUF {
tracing::warn!(
channel = ?channel,
buffered = state.buffer.len(),
incoming = data.len(),
limit = MAX_SSH_CHANNEL_BUF,
"ssh channel buffer overflow; closing channel"
);
state.buffer_overflow = true;
state.buffer.clear();
let _ = session.close(channel);
let _ = session.flush();
return Ok(());
}
state.buffer.extend_from_slice(data);
Ok(())
}
pub async fn channel_close_impl(&mut self, channel: ChannelId) {
self.channels.remove(&channel);
}
pub async fn channel_eof_impl(
&mut self,
channel: ChannelId,
session: &mut Session,
) -> Result<(), russh::Error> {
let Some(mut state) = self.channels.remove(&channel) else {
let _ = session.close(channel);
let _ = session.flush();
return Ok(());
};
let overflowed = state.buffer_overflow;
let op = state.git_op.take();
let buffer = std::mem::take(&mut state.buffer);
let repo_id = state.repo_id;
let space_id = state.space_id;
let actor_id = state.actor_id;
let push_coordinator = self.service.push_coordinator();
let queue = self.service.smart.queue.clone();
let sync_settings = self.service.sync_settings.clone();
let ssh_service = self.service.clone();
let handle = session.handle();
let db = self.service.smart.git.state.database.clone();
tokio::spawn(async move {
if overflowed {
_ = handle.exit_status_request(channel, 1);
_ = handle.eof(channel);
_ = handle.close(channel);
let _ = handle
.data(
channel,
Bytes::from_static(
b"remote: request rejected: ssh channel buffer cap exceeded\n",
),
)
.await;
return;
}
let is_write = op.as_ref().is_some_and(|o| o.is_write());
let is_receive = matches!(
op.as_ref(),
Some(crate::ssh::handler::git_op::GitOperation::ReceivePack { .. })
);
if is_receive && let (Some(repo_id), Some(actor_uid)) = (repo_id, actor_id) {
match parse_receive_pack_refs(&buffer) {
Ok(updates) if !updates.is_empty() => {
let baby = match build_baby_for_pre_receive(&ssh_service, repo_id).await {
Some(b) => b,
None => {
tracing::error!(
repo_id = %repo_id,
"cannot build baby for pre-receive; refusing push"
);
let _ = handle
.data(
channel,
Bytes::from_static(
b"remote: internal error during pre-receive setup\n",
),
)
.await;
_ = handle.exit_status_request(channel, 1);
_ = handle.eof(channel);
_ = handle.close(channel);
return;
}
};
let verdict = match crate::smart::pre_receive::validate_pre_receive(
db.read(),
&baby,
repo_id,
actor_uid,
&updates,
)
.await
{
Ok(v) => v,
Err(e) => {
tracing::error!(
?e, repo_id = %repo_id,
"pre-receive validation failed; refusing push"
);
let _ = handle
.data(
channel,
Bytes::from_static(
b"remote: internal error during pre-receive validation\n",
),
)
.await;
_ = handle.exit_status_request(channel, 1);
_ = handle.eof(channel);
_ = handle.close(channel);
return;
}
};
if let Err(e) = verdict.accepted_only() {
let msg = format!("remote: {e}\n");
_ = handle.data(channel, Bytes::from(msg.into_bytes())).await;
_ = handle.exit_status_request(channel, 1);
_ = handle.eof(channel);
_ = handle.close(channel);
return;
}
}
Ok(_) => {}
Err(e) => {
tracing::error!(
?e, repo_id = %repo_id,
"receive-pack ref parse failed; refusing push"
);
let _ = handle
.data(
channel,
Bytes::from_static(
b"remote: invalid receive-pack ref update\n",
),
)
.await;
_ = handle.exit_status_request(channel, 1);
_ = handle.eof(channel);
_ = handle.close(channel);
return;
}
}
}
if is_write && let Some(bare) = op.as_ref().map(|o| o.bare().clone()) {
if let Err(e) = crate::sync::post_receive::prepare_post_receive(&bare) {
tracing::warn!(?e, "post-receive hook setup failed");
}
let push_key = format!("git_op:{}", bare.display());
let run = async {
run_git(op.as_ref(), &buffer, &handle, channel).await
};
let result = push_coordinator
.run(&push_key, async {
run.await.map_err(|e| SmartError::Internal(e.to_string()))
})
.await
.map_err(|e: SmartError| SshError::from(e));
if result.is_ok()
&& let (Some(repo_id), Some(space_id), Some(queue)) = (repo_id, space_id, &queue)
{
match enqueue_post_receive_push(
queue,
&sync_settings,
space_id,
repo_id,
actor_id,
&bare,
)
.await
{
Ok(n) if n > 0 => {
tracing::debug!(count = n, "enqueued post-receive ref updates")
}
Ok(_) => tracing::debug!("no refs to enqueue"),
Err(e) => tracing::warn!(?e, "post-receive enqueue failed"),
}
}
let code = if result.is_ok() { 0 } else { 1 };
_ = handle.exit_status_request(channel, code);
tokio::time::sleep(std::time::Duration::from_millis(20)).await;
_ = handle.eof(channel);
_ = handle.close(channel);
} else {
let result = run_git(op.as_ref(), &buffer, &handle, channel).await;
let code = if result.is_ok() { 0 } else { 1 };
_ = handle.exit_status_request(channel, code);
tokio::time::sleep(std::time::Duration::from_millis(20)).await;
_ = handle.eof(channel);
_ = handle.close(channel);
}
});
Ok(())
}
}
async fn run_git(
op: Option<&crate::ssh::handler::git_op::GitOperation>,
input: &[u8],
handle: &russh::server::Handle,
channel: russh::ChannelId,
) -> Result<(), SshError> {
let Some(op) = op else {
return Ok(());
};
let result: Result<(), SshError> = async {
let mut cmd = tokio::process::Command::new("git");
cmd.current_dir(op.bare()).arg(op.service_arg());
if !op.is_upload_archive() {
cmd.arg("--stateless-rpc");
}
cmd.env("GIT_DIR", op.bare())
.env("GIT_TERMINAL_PROMPT", "0")
.stdin(std::process::Stdio::piped())
.stdout(std::process::Stdio::piped())
.stderr(std::process::Stdio::piped());
apply_hardened_env(&mut cmd);
let mut child = cmd.spawn().map_err(SshError::Io)?;
if let Some(mut stdin) = child.stdin.take() {
stdin
.write_all(input)
.await
.map_err(SshError::Io)?;
let _ = stdin.shutdown().await;
}
let stdout_cap = MAX_GIT_OUTPUT_BUF as u64 + 1;
let stdout = if let Some(out) = child.stdout.take() {
let mut buf = Vec::new();
out.take(stdout_cap).read_to_end(&mut buf).await.map_err(SshError::Io)?;
buf
} else {
Vec::new()
};
let stderr = if let Some(err) = child.stderr.take() {
let mut buf = Vec::new();
err.take(stdout_cap).read_to_end(&mut buf).await.map_err(SshError::Io)?;
buf
} else {
Vec::new()
};
let status = child.wait().await.map_err(SshError::Io)?;
if stdout.len() > MAX_GIT_OUTPUT_BUF || stderr.len() > MAX_GIT_OUTPUT_BUF {
return Err(SshError::Internal(format!(
"git {} output exceeded {} bytes",
op.service_arg(),
MAX_GIT_OUTPUT_BUF
)));
}
if !stdout.is_empty() {
_ = handle.data(channel, Bytes::from(stdout)).await;
}
if !stderr.is_empty() && !status.success() {
_ = handle.extended_data(channel, 1, Bytes::from(stderr)).await;
}
if status.success() {
Ok(())
} else {
Err(SshError::Internal(format!(
"git {} failed: exit {:?}",
op.service_arg(),
status.code()
)))
}
}
.await;
result
}
/// Build a GitBaby for `validate_pre_receive`. The SSH path does not have
/// `(space, repo)` strings available — only `repo_id` — so we resolve the
/// repo model directly and assemble a `GitContext` from `AppState`. This
/// mirrors `service::context::context` minus the space-name lookup, which
/// is unnecessary here because `validate_pre_receive` only needs the bare
/// repo metadata + a configured repo_root/objects_root.
async fn build_baby_for_pre_receive(
service: &Arc<crate::ssh::service::GitSshService>,
repo_id: uuid::Uuid,
) -> Option<gitbaby::GitBaby> {
use models::repo::repos;
let db = service.smart.git.state.database.clone();
let repo = repos::Entity::find_by_id(repo_id).one(db.read()).await.ok()??;
let cache = match CacheClient::new(&service.smart.git.state.config.cache).await {
Ok(c) => Arc::new(c),
Err(_) => return None,
};
let git_cfg = &service.smart.git.state.config.git;
let ctx = GitContext {
model: repo,
cache,
database: db,
repo_root: std::path::PathBuf::from(&git_cfg.repo_root),
objects_root: std::path::PathBuf::from(&git_cfg.objects_root),
};
Some(service.smart.git.baby(&ctx))
}