194c44c
use bytes::Bytes;
use russh::ChannelId;
use russh::server::Session;
use uuid::Uuid;
use crate::share::access::authorize_repo_access;
use crate::share::errors::GitError;
use crate::smart::error::SmartError;
use crate::ssh::error::SshError;
use crate::ssh::git_service::{
GitService, build_git_advertise_command, parse_git_command, parse_repo_path, sanitize_for_log,
};
use crate::ssh::handler::GitSshHandler;
use crate::ssh::handler::git_op::GitOperation;
impl GitSshHandler {
pub async fn exec_request_impl(
&mut self,
channel: ChannelId,
data: &[u8],
session: &mut Session,
) -> Result<(), russh::Error> {
let cmd = match std::str::from_utf8(data) {
Ok(c) => c,
Err(_) => {
invalid_command("invalid command encoding", session, channel).await;
return Ok(());
}
};
let cmd = cmd.trim();
tracing::info!(
"ssh exec from {:?}: {}",
self.client_addr,
sanitize_for_log(cmd)
);
let Some((service, raw_path)) = parse_git_command(cmd) else {
invalid_command("invalid git command", session, channel).await;
return Ok(());
};
let Some((space_name, repo_name)) = parse_repo_path(raw_path) else {
tracing::debug!(path = %sanitize_for_log(raw_path), "ssh exec: invalid repository path");
invalid_command("invalid repository path", session, channel).await;
return Ok(());
};
tracing::debug!(
service = service.as_subcommand(),
space = space_name,
repo = repo_name,
"ssh exec: parsed git command"
);
let actor_id = self.operator.as_ref().map(|u| u.uid);
let required = if service.is_write() { "write" } else { "read" };
if let Err(msg) = self
.authorize_repo(space_name, repo_name, service.is_write())
.await
{
tracing::warn!("ssh {service:?} {space_name}/{repo_name} denied: {msg}");
invalid_command(&msg, session, channel).await;
return Ok(());
}
let ctx = match self.service.smart.resolve(space_name, repo_name).await {
Ok(c) => c,
Err(e) => {
tracing::warn!(space = space_name, repo = repo_name, error = %e, "ssh exec: repo resolve failed");
let msg = format!("fatal: {e}");
invalid_command(&msg, session, channel).await;
return Ok(());
}
};
tracing::debug!(
space = space_name,
repo = repo_name,
repo_id = %ctx.model.id,
"ssh exec: repo access authorized and resolved"
);
let space_id = match self.space_id_for(&ctx.model.dir_id).await {
Ok(id) => id,
Err(error) => {
tracing::warn!(%error, "space lookup failed, falling back to dir id");
ctx.model.dir_id
}
};
let bare = ctx.repo_dir();
if !bare.is_dir() {
tracing::warn!(bare = %bare.display(), "ssh exec: bare repository directory missing");
let msg = format!("fatal: repository '{space_name}/{repo_name}' does not exist");
invalid_command(&msg, session, channel).await;
return Ok(());
}
// Push is driven from a fully buffered request, so receive-pack only
// runs after EOF and its advertisement is produced up front here.
// Upload services are duplex and emit the advertisement themselves.
let mut adv_len = 0usize;
if service.is_write() {
let adv = match run_advertise(
&bare,
service,
self.service.smart.git.state.config.git.git_rpc_timeout_secs,
)
.await
{
Ok(adv) => adv,
Err(e) => {
tracing::warn!(
service = service.as_subcommand(),
bare = %bare.display(),
error = %e,
"ssh exec: ref advertisement failed"
);
let msg = format!("fatal: unable to read repository advertisement: {e}");
invalid_command(&msg, session, channel).await;
return Ok(());
}
};
adv_len = adv.len();
if !adv.is_empty() {
let _ = session.data(channel, Bytes::from(adv));
}
}
let _ = session.channel_success(channel);
let _ = session.flush();
tracing::debug!(
service = service.as_subcommand(),
repo_id = %ctx.model.id,
adv_bytes = adv_len,
"ssh exec: channel armed"
);
let op = match service {
GitService::UploadPack => GitOperation::UploadPack { bare: bare.clone() },
GitService::UploadArchive => GitOperation::UploadArchive { bare: bare.clone() },
GitService::ReceivePack => GitOperation::ReceivePack { bare: bare.clone() },
};
// Fetch/clone/archive is duplex: waiting for EOF before starting git,
// as the push path does, would deadlock against a client that is
// waiting for the packfile.
let streaming = !op.is_write();
{
let state = self.channels.entry(channel).or_default();
state.git_op = Some(op);
state.repo_id = Some(ctx.model.id);
state.space_id = Some(space_id);
state.actor_id = actor_id;
}
if streaming {
let state = self.channels.remove(&channel).expect("channel state");
let mut state = state;
state.rpc_timeout_secs = self.service.smart.git.state.config.git.git_rpc_timeout_secs;
let (tx, rx) = tokio::sync::mpsc::unbounded_channel::<Vec<u8>>();
let abort = crate::ssh::handler::channel::spawn_streaming_git(
channel,
state,
session.handle(),
rx,
);
let state = self.channels.entry(channel).or_default();
state.dispatched = true;
state.stdin_tx = Some(tx);
state.stream_abort = Some(abort);
}
let _ = required;
Ok(())
}
async fn authorize_repo(
&self,
space: &str,
repo: &str,
want_write: bool,
) -> Result<(), String> {
let svc = &self.service.smart.git;
let ctx = svc
.context(space.to_string(), repo.to_string())
.await
.map_err(|e| format!("{e}"))?;
let Some(operator) = &self.operator else {
return Err("authentication required".into());
};
authorize_repo_access(svc, operator.uid, &ctx, want_write, false)
.await
.map_err(|e| match e {
GitError::Forbidden => "permission denied".to_string(),
other => format!("{other}"),
})
}
async fn space_id_for(&self, dir_id: &Uuid) -> Result<Uuid, SmartError> {
crate::share::access::space_for_dir(self.service.smart.git.db_read(), *dir_id)
.await
.map_err(SmartError::from)?
.map(|s| s.id)
.ok_or_else(|| SmartError::NotFound("space".into()))
}
pub async fn shell_request_impl(
&mut self,
channel: ChannelId,
session: &mut Session,
) -> Result<(), russh::Error> {
tracing::debug!(client = ?self.client_addr, "interactive shell request denied (git-only server)");
let _ = session.data(
channel,
Bytes::from_static(
b"Welcome to GitData SSH. This is a git-only server; shell access is disabled.\r\n",
),
);
let _ = session.exit_status_request(channel, 0);
let _ = session.eof(channel);
let _ = session.close(channel);
let _ = session.flush();
Ok(())
}
}
async fn invalid_command(msg: &str, session: &mut Session, channel: ChannelId) {
let _ = session.data(channel, Bytes::from(format!("{msg}\n").into_bytes()));
let _ = session.channel_failure(channel);
let _ = session.exit_status_request(channel, 1);
let _ = session.eof(channel);
let _ = session.close(channel);
let _ = session.flush();
}
async fn run_advertise(
bare: &std::path::Path,
service: GitService,
timeout_secs: u64,
) -> Result<Vec<u8>, SshError> {
let mut cmd = build_git_advertise_command(service, bare);
cmd.stdout(std::process::Stdio::piped())
.stderr(std::process::Stdio::piped());
let mut child = cmd.spawn().map_err(SshError::Io)?;
let stderr_task = child.stderr.take().map(|stderr| {
tokio::spawn(async move {
use tokio::io::AsyncReadExt;
let s = stderr;
let mut buf = Vec::new();
let _ = s
.take((64 * 1024 * 1024) as u64 + 1)
.read_to_end(&mut buf)
.await;
buf
})
});
let stdout = child
.stdout
.take()
.ok_or_else(|| SshError::Internal("git stdout missing".to_string()))?;
let mut out = Vec::new();
use tokio::io::AsyncReadExt;
stdout
.take((64 * 1024 * 1024) as u64 + 1)
.read_to_end(&mut out)
.await
.map_err(SshError::Io)?;
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 {} --advertise-refs timed out",
service.as_subcommand()
)));
}
};
let stderr_bytes = match stderr_task {
Some(t) => t
.await
.map_err(|e| SshError::Internal(format!("stderr task join failed: {e}")))?,
None => Vec::new(),
};
if !status.success() {
return Err(SshError::Internal(format!(
"git {} --advertise-refs failed: {}",
service.as_subcommand(),
String::from_utf8_lossy(&stderr_bytes)
)));
}
Ok(out)
}