149bec6
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::ChannelState;
use crate::ssh::handler::GitSshHandler;
use crate::ssh::handler::git_op::GitOperation;
use crate::sync::post_receive::enqueue_post_receive_push_with_ci_logged;
pub const MAX_SSH_CHANNEL_BUF: usize = 64 * 1024 * 1024;
pub const MAX_GIT_OUTPUT_BUF: usize = 256 * 1024 * 1024;
fn hex_preview(buf: &[u8]) -> String {
let n = buf.len().min(96);
let mut s = String::with_capacity(n * 2);
for b in &buf[..n] {
s.push_str(&format!("{b:02x}"));
}
s
}
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(());
}
if state.dispatched {
if let Some(tx) = &state.stdin_tx {
let _ = tx.send(data.to_vec());
}
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);
if state.git_op.is_some() && state.buffer.len() == data.len() {
tracing::debug!(
?channel,
first_hex = %hex_preview(&state.buffer),
"ssh: first data chunk on armed channel"
);
}
// Push-only: receive-pack runs after EOF, so its negotiation reply has
// to be synthesized here. Upload services must not receive it, their
// own process emits the real NAK/ACK.
let is_receive = matches!(state.git_op, Some(GitOperation::ReceivePack { .. }));
if is_receive && !state.nak_sent && Self::wants_section_complete(&state.buffer) {
// Over SSH the request and response share one stream: send-pack
// stops after the wants section and waits for NAK before it sends
// the pack. We only spawn receive-pack after channel EOF, so
// synthesize the negotiation reply now; the real process' own NAK
// is stripped from stdout later (see run_git).
tracing::debug!(?channel, "ssh push: synthesized NAK for wants section");
state.nak_sent = true;
let _ = session.data(channel, Bytes::from_static(b"0008NAK\n0000"));
let _ = session.flush();
}
Ok(())
}
/// True when the buffer head is a complete `want` negotiation section
/// terminated by a flush packet.
fn wants_section_complete(buf: &[u8]) -> bool {
let mut i = 0usize;
let mut saw_want = false;
while i + 4 <= buf.len() {
let Ok(len) =
usize::from_str_radix(std::str::from_utf8(&buf[i..i + 4]).unwrap_or_default(), 16)
else {
return false;
};
if len == 0 {
return saw_want;
}
if len < 4 || i + len > buf.len() {
return false;
}
if !buf[i + 4..i + len].starts_with(b"want ") {
return false;
}
saw_want = true;
i += len;
}
false
}
pub async fn channel_close_impl(&mut self, channel: ChannelId) {
tracing::debug!(?channel, "ssh channel closed");
if let Some(state) = self.channels.remove(&channel)
&& let Some(abort) = state.stream_abort
{
abort.abort();
}
}
pub async fn channel_eof_impl(
&mut self,
channel: ChannelId,
session: &mut Session,
) -> Result<(), russh::Error> {
let Some(state) = self.channels.get_mut(&channel) else {
let _ = session.close(channel);
let _ = session.flush();
return Ok(());
};
if state.dispatched {
// Half-close only: the git process may still have data to write
// back (upload-pack answers with the packfile after `done`), so the
// session is torn down when the pump task finishes, not here.
if let Some(tx) = state.stdin_tx.take() {
drop(tx);
}
tracing::debug!(?channel, "ssh: client half-closed duplex channel");
return Ok(());
}
let state = self.channels.remove(&channel).expect("channel state");
self.dispatch_state(channel, state, session.handle(), "eof");
Ok(())
}
/// Detach buffered channel state and run its git operation in the
/// background. `reason` is only used for tracing.
fn dispatch_state(
&mut self,
channel: ChannelId,
state: ChannelState,
handle: russh::server::Handle,
reason: &str,
) {
let mut state = state;
let overflowed = state.buffer_overflow;
let nak_sent = state.nak_sent;
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;
tracing::debug!(
?channel,
op = op.as_ref().map(|o| o.service_arg()),
buffered_bytes = buffer.len(),
repo_id = ?repo_id,
actor_id = ?actor_id,
overflowed,
%reason,
"ssh dispatching git operation"
);
let push_coordinator = self.service.push_coordinator();
let queue = self.service.smart.queue.clone();
let sync_settings = self.service.sync_settings.clone();
let ci_topic = self.service.smart.ci_event_topic().map(|t| t.to_string());
let ai_ctx =
self.service
.smart
.ai_config()
.map(|cfg| crate::sync::post_receive::AiPushCtx {
db: self.service.smart.git.state.database.clone(),
redis: self.service.smart.git.state.redis.clone(),
config: cfg.clone(),
});
let ssh_service = self.service.clone();
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;
}
tracing::debug!(repo_id = %repo_id, "ssh pre-receive validation passed");
}
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());
tracing::debug!(push_key = %push_key, repo_id = ?repo_id, "ssh push: acquiring push lock");
let run = async {
run_git(
op.as_ref(),
&buffer,
&handle,
channel,
rpc_timeout,
nak_sent,
)
.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 let Err(e) = &result {
tracing::warn!(repo_id = ?repo_id, error = %e, "ssh push failed");
}
if result.is_ok()
&& let (Some(repo_id), Some(space_id), Some(queue)) =
(repo_id, space_id, &queue)
{
enqueue_post_receive_push_with_ci_logged(
queue,
&sync_settings,
ci_topic.as_deref(),
ai_ctx,
&crate::sync::PushTarget {
space_id,
repo_id,
actor_id,
bare: &bare,
},
)
.await;
}
let code = if result.is_ok() { 0 } else { 1 };
tracing::debug!(
?channel,
code,
nak_sent,
"ssh: sending exit-status for push"
);
let r1 = handle.exit_status_request(channel, code).await;
// Windows OpenSSH can drop buffered channel data if CLOSE
// races ahead of the client's read loop; give the client time
// to drain the report before tearing the channel down.
tokio::time::sleep(std::time::Duration::from_millis(250)).await;
let r2 = handle.eof(channel).await;
let r3 = handle.close(channel).await;
tracing::debug!(exit_status = ?r1, eof = ?r2, close = ?r3, "ssh: channel shutdown results");
} else {
let result = run_git(
op.as_ref(),
&buffer,
&handle,
channel,
rpc_timeout,
nak_sent,
)
.await;
let code = if result.is_ok() { 0 } else { 1 };
_ = handle.exit_status_request(channel, code);
tokio::time::sleep(std::time::Duration::from_millis(250)).await;
_ = handle.eof(channel);
_ = handle.close(channel);
}
});
}
}
/// Start a duplex git session for `upload-pack` / `upload-archive`: stdin is fed
/// from `rx` and stdout is streamed back as it is produced. The client keeps the
/// channel open while it waits for the packfile, so the response cannot be
/// deferred until EOF the way the push path does.
pub(crate) fn spawn_streaming_git(
channel: russh::ChannelId,
state: ChannelState,
handle: russh::server::Handle,
rx: tokio::sync::mpsc::UnboundedReceiver<Vec<u8>>,
) -> tokio::task::AbortHandle {
let op = state.git_op.clone();
let service = state.git_op.as_ref().map(|o| o.service_arg().to_string());
let timeout_secs = state.rpc_timeout_secs;
let task = tokio::spawn(async move {
let Some(op) = op else {
return;
};
let result = streaming_git(&op, &handle, channel, rx, timeout_secs).await;
match &result {
Ok(()) => tracing::debug!(?channel, "ssh streaming git finished"),
Err(e) => tracing::warn!(?channel, error = %e, "ssh streaming git failed"),
}
let code = if result.is_ok() { 0 } else { 1 };
_ = handle.exit_status_request(channel, code).await;
// Give the client a moment to drain the final bytes before the
// channel is torn down (the same race the push path guards against).
tokio::time::sleep(std::time::Duration::from_millis(250)).await;
_ = handle.eof(channel).await;
_ = handle.close(channel).await;
});
let abort = task.abort_handle();
if let Some(name) = service {
tracing::debug!(?channel, service = %name, "ssh streaming git started");
}
abort
}
/// Drive one duplex git process: pump `rx` into stdin, forward stdout to the
/// SSH channel, and collect stderr for diagnostics.
async fn streaming_git(
op: &crate::ssh::handler::git_op::GitOperation,
handle: &russh::server::Handle,
channel: russh::ChannelId,
mut rx: tokio::sync::mpsc::UnboundedReceiver<Vec<u8>>,
timeout_secs: u64,
) -> Result<(), SshError> {
// Stateful, like a real sshd: advertise refs, then answer the request
// arriving on stdin. Must not be gated on stdin EOF.
let mut cmd = tokio::process::Command::new("git");
cmd.current_dir(op.bare()).arg(op.service_arg());
cmd.arg(op.bare())
.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);
cmd.kill_on_drop(true);
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 mut 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()))?;
tracing::debug!(
service = op.service_arg(),
bare = %op.bare().display(),
"spawned git process for duplex ssh operation"
);
let writer = tokio::spawn(async move {
while let Some(chunk) = rx.recv().await {
if stdin.write_all(&chunk).await.is_err() {
break;
}
}
let _ = stdin.shutdown().await;
});
let reader = {
let handle = handle.clone();
tokio::spawn(async move {
let mut buf = vec![0u8; 64 * 1024];
loop {
match stdout.read(&mut buf).await {
Ok(0) => break,
Ok(n) => {
if handle
.data(channel, Bytes::copy_from_slice(&buf[..n]))
.await
.is_err()
{
break;
}
}
Err(e) => {
tracing::warn!(?channel, error = %e, "ssh: git stdout read failed");
break;
}
}
}
})
};
let stderr_task = tokio::spawn(async move {
let mut buf = Vec::new();
let _ = stderr
.take(MAX_GIT_OUTPUT_BUF as u64 + 1)
.read_to_end(&mut buf)
.await;
buf
});
let status = match tokio::time::timeout(
std::time::Duration::from_secs(timeout_secs),
child.wait(),
)
.await
{
Ok(Ok(s)) => s,
_ => {
let _ = child.start_kill();
let _ = child.wait().await;
return Err(SshError::Internal(format!(
"git {} timed out",
op.service_arg()
)));
}
};
// The writer only stops on client half-close, so awaiting it here would
// hang once the process has already finished.
let _ = reader.await;
writer.abort();
let _ = writer.await;
let stderr_bytes = stderr_task.await.unwrap_or_default();
tracing::debug!(
service = op.service_arg(),
exit = ?status.code(),
stderr_bytes = stderr_bytes.len(),
"git process for duplex ssh operation finished"
);
if !stderr_bytes.is_empty() && !status.success() {
let _ = handle
.extended_data(channel, 1, Bytes::from(stderr_bytes.clone()))
.await;
}
if status.success() {
Ok(())
} else {
Err(SshError::Internal(format!(
"git {} failed: exit {:?}",
op.service_arg(),
status.code()
)))
}
}
async fn run_git(
op: Option<&crate::ssh::handler::git_op::GitOperation>,
input: &[u8],
handle: &russh::server::Handle,
channel: russh::ChannelId,
timeout_secs: u64,
strip_nak: bool,
) -> 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");
}
// Same as the advertise path: receive-pack requires an explicit
// <git-dir> argument, GIT_DIR alone is not enough.
cmd.arg(op.bare());
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);
cmd.kill_on_drop(true);
let mut child = cmd.spawn().map_err(SshError::Io)?;
tracing::debug!(
service = op.service_arg(),
bare = %op.bare().display(),
input_bytes = input.len(),
timeout_secs,
"spawned git process for ssh operation"
);
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()
&& 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)?;
let mut stdout_buf = stdout_buf;
if strip_nak {
// We already answered the wants section with a synthetic NAK, so
// remove receive-pack's own NAK+flush to keep the client's
// negotiation read aligned with the report that follows.
const NAK: &[u8] = b"0008NAK\n";
if stdout_buf.starts_with(NAK) {
let mut off = NAK.len();
if stdout_buf.get(off..off + 4) == Some(b"0000") {
off += 4;
}
stdout_buf.drain(..off);
}
}
tracing::debug!(
service = op.service_arg(),
exit = ?status.code(),
stdout_bytes = stdout_buf.len(),
stderr_bytes = stderr_buf.len(),
"git process for ssh operation finished"
);
if !stdout_buf.is_empty()
&& let Err(e) = handle.data(channel, Bytes::from(stdout_buf.clone())).await
{
tracing::warn!(
?e,
bytes = stdout_buf.len(),
"ssh: failed to send git stdout to client"
);
}
if !stderr_buf.is_empty()
&& !status.success()
&& let Err(e) = handle
.extended_data(channel, 1, Bytes::from(stderr_buf.clone()))
.await
{
tracing::warn!(?e, "ssh: failed to send git stderr to client");
}
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 = Arc::new(CacheClient::new(service.smart.git.state.redis.clone()));
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))
}