69f12aa
use std::sync::Arc;
use axum::body::Body;
use axum::http::{HeaderMap, StatusCode, header};
use axum::response::{IntoResponse, Response};
use bytes::Bytes;
use flate2::read::GzDecoder;
use futures::StreamExt as _;
use std::io::Read;
use uuid::Uuid;
use crate::share::access::{BasicAuth, allow_repo_read, require_writer};
use crate::smart::error::{SmartError, SmartResult};
use crate::smart::info_refs as info_refs_module;
use crate::smart::pre_receive::{self, parse_receive_pack_refs};
use crate::smart::rpc::{self, GitService};
use crate::smart::service::SmartService;
use crate::sync::post_receive::enqueue_post_receive_push_with_ci_logged;
const MAX_RPC_BODY: usize = 512 * 1024 * 1024;
#[tracing::instrument(name = "git.info_refs", skip_all, fields(space, repo))]
pub async fn info_refs(
service: &SmartService,
space: &str,
repo: &str,
_headers: &HeaderMap,
auth: &BasicAuth,
service_kind: Option<&str>,
) -> SmartResult<Response> {
let (ctx, _actor_id) = match service_kind {
Some(name) => {
let git_service = GitService::from_name(name)?;
let want_write = matches!(git_service, GitService::ReceivePack);
let ctx = service.resolve(space, repo).await?;
let actor_id = authorize(service, &ctx, auth, want_write).await?;
(ctx, actor_id)
}
None => {
let ctx = service.resolve(space, repo).await?;
allow_repo_read(&service.git, auth, &ctx).await?;
(ctx, None)
}
};
let bare = service.bare_path(&ctx);
tracing::debug!(
space,
repo,
service = service_kind.unwrap_or("dumb"),
bare = %bare.display(),
"info/refs authorized, generating advertisement"
);
match service_kind {
Some(name) => {
let git_service = GitService::from_name(name)?;
let timeout = service.git.state.config.git.git_rpc_timeout_secs;
info_refs_module::smart_advertisement(&bare, git_service, timeout).await
}
None => info_refs_module::dumb_info_refs(&bare).await,
}
}
#[allow(clippy::too_many_arguments)]
#[tracing::instrument(name = "git.rpc", skip_all, fields(space, repo, op = ?git_service))]
pub async fn rpc(
service: &SmartService,
space: &str,
repo: &str,
headers: &HeaderMap,
body: Body,
auth: &BasicAuth,
git_service: GitService,
) -> SmartResult<Response> {
let want_write = matches!(git_service, GitService::ReceivePack);
let ctx = service.resolve(space, repo).await?;
let actor_id = authorize(service, &ctx, auth, want_write).await?;
let bare = service.bare_path(&ctx);
let space_id = service.owning_space_id(&ctx).await?;
tracing::info!(
space,
repo,
service = git_service.name(),
actor_id = ?actor_id,
repo_id = %ctx.model.id,
"git rpc authorized, processing request"
);
let run = async {
let input = decode_request_body(headers, body, MAX_RPC_BODY).await?;
let is_receive = git_service == GitService::ReceivePack;
tracing::debug!(
service = git_service.name(),
body_bytes = input.len(),
"rpc request body decoded"
);
if is_receive && actor_id.is_some() {
let updates = parse_receive_pack_refs(&input)?;
tracing::debug!(
repo_id = %ctx.model.id,
ref_count = updates.len(),
"parsed receive-pack ref updates"
);
if !updates.is_empty()
&& let Some(actor_uid) = actor_id
{
let baby = service.git.baby(&ctx);
let verdict = pre_receive::validate_pre_receive(
service.git.db_read(),
&baby,
ctx.model.id,
actor_uid,
&updates,
)
.await?;
verdict.accepted_only()?;
tracing::debug!(repo_id = %ctx.model.id, "pre-receive validation passed");
}
if let Err(e) = crate::sync::post_receive::prepare_post_receive(&bare) {
tracing::warn!(?e, "post-receive hook setup failed");
}
}
let timeout = service.git.state.config.git.git_rpc_timeout_secs;
let stream = rpc::run_rpc_stream(&bare, git_service, false, input, timeout).await?;
let post = PostReceiveStream {
is_receive,
queue: service.queue.clone(),
settings: service.sync_settings.clone(),
ci_topic: service.ci_event_topic().map(|t| t.to_string()),
space_id,
repo_id: ctx.model.id,
actor_id,
bare: bare.clone(),
};
let stream = stream.map(|chunk| chunk.map_err(|e| axum::Error::new(e.to_string())));
let stream = body_stream_with_post_receive(stream, post);
Ok::<_, SmartError>(
(
StatusCode::OK,
[
(header::CONTENT_TYPE, git_service.result_content_type()),
(header::CACHE_CONTROL, "no-cache"),
],
Body::from_stream(stream),
)
.into_response(),
)
};
match git_service {
GitService::ReceivePack => {
let key = format!("push:{}:{}", space_id, ctx.model.id);
service.push_coordinator().run(&key, run).await
}
GitService::UploadPack => run.await,
}
}
async fn authorize(
service: &SmartService,
ctx: &crate::context::facade::GitContext,
auth: &BasicAuth,
want_write: bool,
) -> SmartResult<Option<i64>> {
let svc = &service.git;
if want_write {
let uid = require_writer(svc, auth, ctx).await?;
Ok(Some(uid))
} else {
let uid = allow_repo_read(svc, auth, ctx).await?;
Ok(uid)
}
}
/// Everything the post-receive hook needs once the response body has been
/// streamed through. A named struct rather than a nine-element tuple: the fields
/// are all `Uuid`-shaped, so positional access silently swaps tenant and repo.
pub struct PostReceiveStream {
pub is_receive: bool,
pub queue: Option<Arc<crate::sync::queue::SyncQueue>>,
pub settings: config::git::SyncSettings,
pub ci_topic: Option<String>,
pub space_id: Uuid,
pub repo_id: Uuid,
pub actor_id: Option<i64>,
pub bare: std::path::PathBuf,
}
fn body_stream_with_post_receive(
stream: impl futures::Stream<Item = Result<Bytes, axum::Error>> + Send + 'static,
post: PostReceiveStream,
) -> impl futures::Stream<Item = Result<Bytes, axum::Error>> + Send + 'static {
use futures::{StreamExt, stream};
stream.chain(stream::once(async move {
let PostReceiveStream {
is_receive,
queue,
settings,
ci_topic,
space_id,
repo_id,
actor_id,
bare,
} = post;
if is_receive && let Some(queue) = queue {
enqueue_post_receive_push_with_ci_logged(
&queue,
&settings,
ci_topic.as_deref(),
&crate::sync::PushTarget {
space_id,
repo_id,
actor_id,
bare: &bare,
},
)
.await;
}
Ok::<_, axum::Error>(Bytes::new())
}))
}
pub async fn decode_request_body(
headers: &HeaderMap,
body: Body,
limit: usize,
) -> SmartResult<Bytes> {
let bytes = axum::body::to_bytes(body, limit)
.await
.map_err(|e| SmartError::BadRequest(format!("failed to read request body: {e}")))?;
let encoding = headers
.get(header::CONTENT_ENCODING)
.and_then(|v| v.to_str().ok())
.map(|v| v.to_ascii_lowercase());
if encoding.as_deref() == Some("gzip") {
let decoder = GzDecoder::new(bytes.as_ref());
let mut out = Vec::new();
let cap = limit as u64 + 1;
decoder
.take(cap)
.read_to_end(&mut out)
.map_err(|e| SmartError::BadRequest(format!("failed to decompress gzip body: {e}")))?;
if out.len() as u64 > limit as u64 {
return Err(SmartError::BadRequest(format!(
"decompressed body exceeds limit of {limit} bytes"
)));
}
tracing::debug!(
compressed = bytes.len(),
decompressed = out.len(),
"gzip request body decompressed"
);
Ok(Bytes::from(out))
} else {
Ok(bytes)
}
}