d977f67
zhenyi
refactor(dvcs): refresh dvcs/git, dvcs/gitsmart, dvcs/gitssh, dvcs/gitsync, dvcs/gitauth
d977f67
use bytes::Bytes;
use gitsmart::pre_receive::{self, parse_receive_pack_refs};
use russh::ChannelId;
use russh::server::Session;
use crate::git_service::apply_hardened_env;
use crate::handler::GitSshHandler;
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 channel_id = state.channel_id;
let actor_id = state.actor_id;
let push_coordinator = self.service.push_coordinator();
let queue = self.service.queue.clone();
let sync_settings = self.service.sync_settings.clone();
let handle = session.handle();
let db = self.service.auth.db.clone();
tokio::spawn(async move {
if overflowed {
tracing::warn!(
channel = ?channel,
"refusing to spawn git for ssh channel that exceeded buffer cap"
);
let _ = handle.exit_status_request(channel, 1).await;
let _ = handle.eof(channel).await;
let _ = handle.close(channel).await;
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::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 db_conn = db.reader();
match pre_receive::validate_pre_receive(&db_conn, repo_id, actor_uid, &updates)
.await
{
Ok(verdict) => {
if let Err(e) = verdict.accepted_only() {
let msg = format!("remote: {}\n", e);
let _ = handle.data(channel, Bytes::from(msg.into_bytes())).await;
let _ = handle.exit_status_request(channel, 1).await;
let _ = handle.eof(channel).await;
let _ = handle.close(channel).await;
return;
}
}
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;
let _ = handle.exit_status_request(channel, 1).await;
let _ = handle.eof(channel).await;
let _ = handle.close(channel).await;
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;
let _ = handle.exit_status_request(channel, 1).await;
let _ = handle.eof(channel).await;
let _ = handle.close(channel).await;
return;
}
}
}
if is_write
&& let Some(bare) = op.as_ref().map(|o| o.bare().to_path_buf())
&& let Err(e) = gitsync::prepare_post_receive(&bare)
{
tracing::warn!(?e, "post-receive hook setup failed");
}
let push_key = op
.as_ref()
.map(|o| format!("git_op:{}", o.bare().display()))
.unwrap_or_default();
let run = async { run_git(op.as_ref(), &buffer, &handle, channel).await };
let result = if is_write {
push_coordinator
.run(&push_key, async {
run.await
.map_err(|e| gitsmart::SmartError::Internal(e.to_string()))
})
.await
.map_err(crate::error::SshError::from)
} else {
run.await
};
if result.is_ok()
&& is_write
&& let (Some(repo_id), Some(channel_id), Some(bare)) = (
repo_id,
channel_id,
op.as_ref().map(|o| o.bare().to_path_buf()),
)
&& let Some(queue) = &queue
{
match gitsync::enqueue_post_receive_push(
queue,
&sync_settings,
channel_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 = match &result {
Ok(_) => 0,
Err(_) => 1,
};
let _ = handle.exit_status_request(channel, code).await;
tokio::time::sleep(std::time::Duration::from_millis(20)).await;
let _ = handle.eof(channel).await;
let _ = handle.close(channel).await;
});
Ok(())
}
pub async fn shell_request_impl(
&mut self,
channel: ChannelId,
session: &mut Session,
) -> Result<(), russh::Error> {
let msg = Bytes::from_static(
b"Welcome to GitSsh. This is a git-only server; shell access is disabled.\r\n",
);
let _ = session.data(channel, msg);
let _ = session.exit_status_request(channel, 0);
let _ = session.eof(channel);
let _ = session.close(channel);
let _ = session.flush();
Ok(())
}
}
async fn run_git(
op: Option<&crate::handler::git_op::GitOperation>,
input: &[u8],
handle: &russh::server::Handle,
channel: russh::ChannelId,
) -> Result<(), crate::error::SshError> {
let Some(op) = op else {
return Ok(());
};
let _timer = crate::metrics::SubprocessTimer::start();
let result: Result<(), crate::error::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(crate::error::SshError::Io)?;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
if let Some(mut stdin) = child.stdin.take() {
stdin
.write_all(input)
.await
.map_err(crate::error::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(crate::error::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(crate::error::SshError::Io)?;
buf
} else {
Vec::new()
};
let status = child.wait().await.map_err(crate::error::SshError::Io)?;
if stdout.len() > MAX_GIT_OUTPUT_BUF || stderr.len() > MAX_GIT_OUTPUT_BUF {
return Err(crate::error::SshError::Internal(format!(
"git {} output exceeded {} bytes",
op.service_arg(),
MAX_GIT_OUTPUT_BUF
)));
}
if !stdout.is_empty() {
let _ = handle.data(channel, Bytes::from(stdout)).await;
}
if !stderr.is_empty() && !status.success() {
let _ = handle
.extended_data(channel, 1, Bytes::from(stderr))
.await;
}
if status.success() {
Ok(())
} else {
Err(crate::error::SshError::Internal(format!(
"git {} failed: exit {:?}",
op.service_arg(),
status.code()
)))
}
}
.await;
crate::metrics::record_subprocess(op.service_arg(), result.is_ok());
result
}