3cab86b
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();
let rpc_timeout = self.service.smart.git.state.config.git.git_rpc_timeout_secs;
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, rpc_timeout).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, rpc_timeout).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,
timeout_secs: u64,
) -> 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)?;
let mut stdin = child
.stdin
.take()
.ok_or_else(|| SshError::Internal("git stdin missing".to_string()))?;
let stdout = child
.stdout
.take()
.ok_or_else(|| SshError::Internal("git stdout missing".to_string()))?;
let stderr = child
.stderr
.take()
.ok_or_else(|| SshError::Internal("git stderr missing".to_string()))?;
let child_arc = Arc::new(tokio::sync::Mutex::new(Some(child)));
{
let child_arc = child_arc.clone();
let input = input.to_vec();
tokio::spawn(async move {
if stdin.write_all(&input).await.is_err() {
if let Some(mut ch) = child_arc.lock().await.take() {
let _ = ch.start_kill();
let _ = ch.wait().await;
}
}
let _ = stdin.shutdown().await;
});
}
let stdout_task = tokio::spawn(async move {
let cap = MAX_GIT_OUTPUT_BUF as u64 + 1;
let mut buf = Vec::new();
let res = stdout.take(cap).read_to_end(&mut buf).await;
(buf, res)
});
let stderr_task = tokio::spawn(async move {
let cap = MAX_GIT_OUTPUT_BUF as u64 + 1;
let mut buf = Vec::new();
let res = stderr.take(cap).read_to_end(&mut buf).await;
(buf, res)
});
let wait_task = {
let child_arc = child_arc.clone();
async move {
let mut guard = child_arc.lock().await;
match guard.as_mut() {
None => None,
Some(ch) => {
match tokio::time::timeout(
std::time::Duration::from_secs(timeout_secs),
ch.wait(),
)
.await
{
Ok(Ok(s)) => Some(s),
_ => {
let _ = ch.start_kill();
let _ = ch.wait().await;
None
}
}
}
}
}
};
let (wait_out, stdout_res, stderr_res) =
tokio::join!(wait_task, stdout_task, stderr_task);
let (stdout_buf, stdout_read) = stdout_res
.map_err(|e| SshError::Internal(format!("stdout task: {e}")))?;
let (stderr_buf, _stderr_read) = stderr_res
.map_err(|e| SshError::Internal(format!("stderr task: {e}")))?;
let status = wait_out.ok_or_else(|| {
SshError::Internal(format!("git {} timed out", op.service_arg()))
})?;
if stdout_buf.len() > MAX_GIT_OUTPUT_BUF || stderr_buf.len() > MAX_GIT_OUTPUT_BUF {
return Err(SshError::Internal(format!(
"git {} output exceeded {} bytes",
op.service_arg(),
MAX_GIT_OUTPUT_BUF
)));
}
stdout_read.map_err(SshError::Io)?;
if !stdout_buf.is_empty() {
let _ = handle.data(channel, Bytes::from(stdout_buf)).await;
}
if !stderr_buf.is_empty() && !status.success() {
let _ = handle
.extended_data(channel, 1, Bytes::from(stderr_buf))
.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))
}