fadf6a5
use bytes::Bytes;
use russh::ChannelId;
use russh::server::Session;
use crate::handler::GitSshHandler;
impl GitSshHandler {
pub async fn data_impl(&mut self, channel: ChannelId, data: &[u8]) -> Result<(), russh::Error> {
let state = self.channels.entry(channel).or_default();
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(state) = self.channels.get_mut(&channel) else {
let _ = session.close(channel);
let _ = session.flush();
return Ok(());
};
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();
tokio::spawn(async move {
let is_write = op.as_ref().is_some_and(|o| o.is_write());
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())
.arg("--stateless-rpc")
.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());
let mut child = cmd.spawn().map_err(crate::error::SshError::Io)?;
use tokio::io::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 output = child
.wait_with_output()
.await
.map_err(crate::error::SshError::Io)?;
if !output.stdout.is_empty() {
let _ = handle.data(channel, Bytes::from(output.stdout)).await;
}
if !output.stderr.is_empty() && !output.status.success() {
let _ = handle
.extended_data(channel, 1, Bytes::from(output.stderr))
.await;
}
if output.status.success() {
Ok(())
} else {
Err(crate::error::SshError::Internal(format!(
"git {} failed: exit {:?}",
op.service_arg(),
output.status.code()
)))
}
}
.await;
crate::metrics::record_subprocess(op.service_arg(), result.is_ok());
result
}