feat(lib/git): add smart HTTP protocol handlers (info_refs, push, rpc, lfs, pktline, dispatch, etc.)
ZhenYi committed
Added lib/git/src/smart/dispatch.rs lib/git/src/smart/dispatch.rs +183 −0 View lib/git/src/smart/dispatch.rs
@@ -0,0 +1,183 @@
1use std::path::PathBuf;
2
3use axum::body::Body;
4use axum::http::{HeaderMap, StatusCode, header};
5use axum::response::{IntoResponse, Response};
6use bytes::Bytes;
7use flate2::read::GzDecoder;
8use std::io::Read;
9use uuid::Uuid;
10
11use crate::context::facade::GitContext;
12use crate::share::access::{BasicAuth, allow_repo_read, require_writer};
13use crate::smart::error::{SmartError, SmartResult};
14use crate::smart::info_refs as info_refs_module;
15use crate::smart::pre_receive::{self, parse_receive_pack_refs};
16use crate::smart::rpc::{self, GitService};
17use crate::smart::service::SmartService;
18use crate::sync::post_receive::enqueue_post_receive_push;
19
20const MAX_RPC_BODY: usize = 512 * 1024 * 1024;
21
22pub async fn info_refs(
23 service: &SmartService,
24 space: &str,
25 repo: &str,
26 _headers: &HeaderMap,
27 auth: &BasicAuth,
28 service_kind: Option<&str>,
29) -> SmartResult<Response> {
30 let (ctx, _actor_id) = match service_kind {
31 Some(name) => {
32 let git_service = GitService::from_name(name)?;
33 let want_write = matches!(git_service, GitService::ReceivePack);
34 let ctx = service.resolve(space, repo).await?;
35 let actor_id = authorize(service, &ctx, auth, want_write).await?;
36 (ctx, actor_id)
37 }
38 None => {
39 let ctx = service.resolve(space, repo).await?;
40 allow_repo_read(&service.git, auth, &ctx).await?;
41 (ctx, None)
42 }
43 };
44 let bare = service.bare_path(&ctx);
45 match service_kind {
46 Some(name) => {
47 let git_service = GitService::from_name(name)?;
48 info_refs_module::smart_advertisement(&bare, git_service).await
49 }
50 None => info_refs_module::dumb_info_refs(&bare).await,
51 }
52}
53
54#[allow(clippy::too_many_arguments)]
55pub async fn rpc(
56 service: &SmartService,
57 space: &str,
58 repo: &str,
59 headers: &HeaderMap,
60 body: Body,
61 auth: &BasicAuth,
62 git_service: GitService,
63) -> SmartResult<Response> {
64 let want_write = matches!(git_service, GitService::ReceivePack);
65 let ctx = service.resolve(space, repo).await?;
66 let actor_id = authorize(service, &ctx, auth, want_write).await?;
67 let bare = service.bare_path(&ctx);
68 let space_id = service.owning_space_id(&ctx).await?;
69
70 let run = async {
71 let input = decode_request_body(headers, body, MAX_RPC_BODY).await?;
72 let is_receive = git_service == GitService::ReceivePack;
73 if is_receive && actor_id.is_some() {
74 let updates = parse_receive_pack_refs(&input)?;
75 if !updates.is_empty()
76 && let Some(actor_uid) = actor_id {
77 let baby = service.git.baby(&ctx);
78 let verdict = pre_receive::validate_pre_receive(
79 service.git.db_read(),
80 &baby,
81 ctx.model.id,
82 actor_uid,
83 &updates,
84 )
85 .await?;
86 verdict.accepted_only()?;
87 }
88 if let Err(e) = crate::sync::post_receive::prepare_post_receive(&bare) {
89 tracing::warn!(?e, "post-receive hook setup failed");
90 }
91 }
92 let output = rpc::run_rpc_bytes(&bare, git_service, false, input).await?;
93 if is_receive
94 && let Some(queue) = &service.queue
95 {
96 match enqueue_post_receive_push(
97 queue,
98 &service.sync_settings,
99 space_id,
100 ctx.model.id,
101 actor_id,
102 &bare,
103 )
104 .await
105 {
106 Ok(n) if n > 0 => {
107 tracing::debug!(count = n, "enqueued post-receive ref updates")
108 }
109 Ok(_) => tracing::debug!("no refs to enqueue"),
110 Err(e) => tracing::warn!(?e, "post-receive enqueue failed"),
111 }
112 }
113 Ok::<_, SmartError>(
114 (
115 StatusCode::OK,
116 [
117 (header::CONTENT_TYPE, git_service.result_content_type()),
118 (header::CACHE_CONTROL, "no-cache"),
119 ],
120 output,
121 )
122 .into_response(),
123 )
124 };
125
126 match git_service {
127 GitService::ReceivePack => {
128 let key = format!("push:{}:{}", space_id, ctx.model.id);
129 service.push_coordinator().run(&key, run).await
130 }
131 GitService::UploadPack => run.await,
132 }
133}
134
135async fn authorize(
136 service: &SmartService,
137 ctx: &crate::context::facade::GitContext,
138 auth: &BasicAuth,
139 want_write: bool,
140) -> SmartResult<Option<Uuid>> {
141 let svc = &service.git;
142 if want_write {
143 let uid = require_writer(svc, auth, ctx).await?;
144 Ok(Some(uid))
145 } else {
146 let uid = allow_repo_read(svc, auth, ctx).await?;
147 Ok(uid)
148 }
149}
150
151pub async fn decode_request_body(
152 headers: &HeaderMap,
153 body: Body,
154 limit: usize,
155) -> SmartResult<Bytes> {
156 let bytes = axum::body::to_bytes(body, limit)
157 .await
158 .map_err(|e| SmartError::BadRequest(format!("failed to read request body: {e}")))?;
159 let encoding = headers
160 .get(header::CONTENT_ENCODING)
161 .and_then(|v| v.to_str().ok())
162 .map(|v| v.to_ascii_lowercase());
163 if encoding.as_deref() == Some("gzip") {
164 let decoder = GzDecoder::new(bytes.as_ref());
165 let mut out = Vec::new();
166 let cap = limit as u64 + 1;
167 decoder
168 .take(cap)
169 .read_to_end(&mut out)
170 .map_err(|e| SmartError::BadRequest(format!("failed to decompress gzip body: {e}")))?;
171 if out.len() as u64 > limit as u64 {
172 return Err(SmartError::BadRequest(format!(
173 "decompressed body exceeds limit of {limit} bytes"
174 )));
175 }
176 Ok(Bytes::from(out))
177 } else {
178 Ok(bytes)
179 }
180}
181
182#[allow(dead_code)]
183fn _unused(_ctx: &GitContext, _p: &PathBuf) {}
Added lib/git/src/smart/error.rs lib/git/src/smart/error.rs +78 −0 View lib/git/src/smart/error.rs
@@ -0,0 +1,78 @@
1use axum::http::StatusCode;
2use axum::response::{IntoResponse, Response};
3
4#[derive(Debug, thiserror::Error)]
5pub enum SmartError {
6 #[error("{0}")]
7 BadRequest(String),
8 #[error("unauthorized")]
9 Unauthorized,
10 #[error("{0}")]
11 Forbidden(String),
12 #[error("{0}")]
13 NotFound(String),
14 #[error("{0}")]
15 Conflict(String),
16 #[error("{0}")]
17 Internal(String),
18 #[error("io error: {0}")]
19 Io(#[from] std::io::Error),
20 #[error("database error: {0}")]
21 Db(#[from] sea_orm::DbErr),
22 #[error("redis error: {0}")]
23 Redis(#[from] redis::RedisError),
24}
25
26pub type SmartResult<T> = Result<T, SmartError>;
27
28impl SmartError {
29 pub fn status(&self) -> StatusCode {
30 match self {
31 Self::BadRequest(_) => StatusCode::BAD_REQUEST,
32 Self::Unauthorized => StatusCode::UNAUTHORIZED,
33 Self::Forbidden(_) => StatusCode::FORBIDDEN,
34 Self::NotFound(_) => StatusCode::NOT_FOUND,
35 Self::Conflict(_) => StatusCode::CONFLICT,
36 _ => StatusCode::INTERNAL_SERVER_ERROR,
37 }
38 }
39
40 pub fn message(&self) -> String {
41 match self {
42 Self::BadRequest(m)
43 | Self::Forbidden(m)
44 | Self::NotFound(m)
45 | Self::Conflict(m)
46 | Self::Internal(m) => m.clone(),
47 other => other.to_string(),
48 }
49 }
50}
51
52impl From<crate::share::errors::GitError> for SmartError {
53 fn from(err: crate::share::errors::GitError) -> Self {
54 use crate::share::errors::GitError;
55 match err {
56 GitError::Unauthorized => Self::Unauthorized,
57 GitError::NotFound => Self::NotFound("resource not found".into()),
58 GitError::Forbidden => Self::Forbidden("forbidden".into()),
59 GitError::InvalidArgument(m) => Self::BadRequest(m),
60 GitError::Conflict(m) => Self::Conflict(m),
61 other => Self::Internal(other.to_string()),
62 }
63 }
64}
65
66impl IntoResponse for SmartError {
67 fn into_response(self) -> Response {
68 let status = self.status();
69 let mut resp = (status, self.message()).into_response();
70 if status == StatusCode::UNAUTHORIZED {
71 resp.headers_mut().insert(
72 axum::http::header::WWW_AUTHENTICATE,
73 axum::http::HeaderValue::from_static("Basic realm=\"gitdata\""),
74 );
75 }
76 resp
77 }
78}
No newline at end of file
Added lib/git/src/smart/handler.rs lib/git/src/smart/handler.rs +147 −0 View lib/git/src/smart/handler.rs
@@ -0,0 +1,147 @@
1use std::collections::HashMap;
2
3use axum::Router;
4use axum::body::Body;
5use axum::extract::{Path, Query, Request, State};
6use axum::http::{HeaderMap, Method};
7use axum::response::{IntoResponse, Response};
8
9use crate::share::access::{BasicAuth, allow_repo_read};
10use crate::smart::error::{SmartError, SmartResult};
11use crate::smart::lfs;
12use crate::smart::object;
13use crate::smart::rpc::GitService;
14use crate::smart::service::SmartService;
15use crate::smart::dispatch;
16
17pub fn router(service: SmartService) -> Router {
18 Router::new()
19 .route(
20 "/{space}/{repo}/{*path}",
21 axum::routing::get(dispatch)
22 .post(dispatch)
23 .put(dispatch)
24 .head(dispatch),
25 )
26 .with_state(service)
27}
28
29async fn dispatch(
30 State(service): State<SmartService>,
31 Path((space, repo, path)): Path<(String, String, String)>,
32 Query(query): Query<HashMap<String, String>>,
33 headers: HeaderMap,
34 auth: BasicAuth,
35 req: Request<Body>,
36) -> Response {
37 let method = req.method().clone();
38 let body = req.into_body();
39
40 let repo_name = match repo.strip_suffix(".git") {
41 Some(name) if !name.is_empty() => name.to_string(),
42 _ => return SmartError::BadRequest("expected '<repo>.git' in URL".into()).into_response(),
43 };
44
45 let path_lower = path.to_lowercase();
46 let result: SmartResult<Response> = match (path_lower.as_str(), method.clone()) {
47 ("info/refs", m) if m == Method::GET || m == Method::HEAD => {
48 let service_kind = query.get("service").cloned();
49 match dispatch::info_refs(
50 &service,
51 &space,
52 &repo_name,
53 &headers,
54 &auth,
55 service_kind.as_deref(),
56 )
57 .await
58 {
59 Ok(mut resp) => {
60 if m == Method::HEAD {
61 *resp.body_mut() = Body::empty();
62 }
63 Ok(resp)
64 }
65 Err(e) => Err(e),
66 }
67 }
68 ("git-upload-pack", Method::POST) => {
69 dispatch::rpc(
70 &service,
71 &space,
72 &repo_name,
73 &headers,
74 body,
75 &auth,
76 GitService::UploadPack,
77 )
78 .await
79 }
80 ("git-receive-pack", Method::POST) => {
81 dispatch::rpc(
82 &service,
83 &space,
84 &repo_name,
85 &headers,
86 body,
87 &auth,
88 GitService::ReceivePack,
89 )
90 .await
91 }
92 (p, _) if p.starts_with("info/lfs/") => {
93 lfs::dispatch(&service, &space, &repo_name, &headers, p, method.clone(), body, &auth).await
94 }
95 (_, Method::GET) => {
96 let resp = match service.resolve(&space, &repo_name).await {
97 Ok(ctx) => match allow_repo_read(&service.git, &auth, &ctx).await {
98 Ok(_) => {
99 if !is_served_path(&path) {
100 SmartError::Forbidden("path not served".into()).into_response()
101 } else {
102 let bare = service.bare_path(&ctx);
103 match object::serve(&bare, &path).await {
104 Ok(resp) => resp,
105 Err(e) => e.into_response(),
106 }
107 }
108 }
109 Err(e) => e.into_response(),
110 },
111 Err(e) => e.into_response(),
112 };
113 Ok(resp)
114 },
115 _ => Ok(SmartError::BadRequest("method not allowed".into()).into_response()),
116 };
117 result.unwrap_or_else(|e| e.into_response())
118}
119
120/// Whitelist of paths the GET fallback may serve. The dumb HTTP protocol
121/// fetches `info/refs`, `HEAD`, `objects/info/packs`, `objects/pack/*`
122/// and `objects/<xx>/<rest>`; everything else (notably `config`,
123/// `hooks/*`, `packed-refs`, `info/exclude`) is internal and must not be
124/// exposed anonymously.
125fn is_served_path(path: &str) -> bool {
126 if path == "info/refs" || path == "HEAD" {
127 return true;
128 }
129 if let Some(rest) = path.strip_prefix("objects/") {
130 if rest.starts_with("pack/") {
131 return true;
132 }
133 if rest == "info/packs" || rest.starts_with("info/") {
134 return true;
135 }
136 // objects/<2-char dir>/<38 hex> = loose object fetch
137 if rest.len() >= 3 && rest.as_bytes()[2] == b'/' && rest.is_ascii() {
138 let prefix = &rest[..2];
139 let after = &rest[3..];
140 if prefix.chars().all(|c| c.is_ascii_hexdigit()) && !after.is_empty() {
141 return true;
142 }
143 }
144 return false;
145 }
146 false
147}
Added lib/git/src/smart/info_refs.rs lib/git/src/smart/info_refs.rs +42 −0 View lib/git/src/smart/info_refs.rs
@@ -0,0 +1,42 @@
1use axum::http::{StatusCode, header};
2use axum::response::{IntoResponse, Response};
3use bytes::Bytes;
4
5use crate::smart::error::{SmartError, SmartResult};
6use crate::smart::pktline;
7use crate::smart::rpc::{self, GitService};
8
9pub async fn smart_advertisement(
10 bare: &std::path::Path,
11 service: GitService,
12) -> SmartResult<Response> {
13 let output = rpc::run_rpc_bytes(bare, service, true, Bytes::new()).await?;
14 let mut body = pktline::encode_header(service.dashed_name());
15 body.extend_from_slice(&output);
16
17 Ok((
18 StatusCode::OK,
19 [
20 (header::CONTENT_TYPE, service.advertisement_content_type()),
21 (header::CACHE_CONTROL, "no-cache"),
22 ],
23 body,
24 )
25 .into_response())
26}
27
28pub async fn dumb_info_refs(bare: &std::path::Path) -> SmartResult<Response> {
29 let file = bare.join("info").join("refs");
30 match tokio::fs::read(&file).await {
31 Ok(bytes) => Ok((
32 StatusCode::OK,
33 [(header::CONTENT_TYPE, "text/plain; charset=utf-8")],
34 bytes,
35 )
36 .into_response()),
37 Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
38 Err(SmartError::NotFound("info/refs not found".into()))
39 }
40 Err(e) => Err(SmartError::Io(e)),
41 }
42}
No newline at end of file
Added lib/git/src/smart/lfs.rs lib/git/src/smart/lfs.rs +278 −0 View lib/git/src/smart/lfs.rs
@@ -0,0 +1,278 @@
1use axum::body::Body;
2use axum::http::{HeaderMap, Method, StatusCode, header};
3use axum::response::{IntoResponse, Response};
4use futures::StreamExt as _;
5use serde_json::json;
6use tokio_util::io::StreamReader;
7
8use crate::lfs::types::{
9 LfsBatchRequest, LfsLockCreateRequest, LfsOperation, LfsVerifyLocksRequest,
10};
11use crate::lfs::{lock, object};
12use crate::share::access::{BasicAuth, allow_repo_read, require_writer};
13use crate::smart::error::{SmartError, SmartResult};
14use crate::smart::service::SmartService;
15
16const MAX_LFS_BODY: usize = 64 * 1024 * 1024;
17
18#[allow(clippy::too_many_arguments)]
19pub async fn dispatch(
20 service: &SmartService,
21 space: &str,
22 repo: &str,
23 headers: &HeaderMap,
24 path: &str,
25 method: Method,
26 body: Body,
27 auth: &BasicAuth,
28) -> SmartResult<Response> {
29 match path {
30 "info/lfs/objects/batch" if method == Method::POST => {
31 let bytes = read_body(headers, body, MAX_LFS_BODY).await?;
32 let req: LfsBatchRequest = serde_json::from_slice(&bytes)
33 .map_err(|e| SmartError::BadRequest(format!("invalid batch request: {e}")))?;
34 batch(service, space, repo, req, auth).await
35 }
36 "info/lfs/verify" if method == Method::POST => {
37 let bytes = read_body(headers, body, MAX_LFS_BODY).await?;
38 verify(service, space, repo, bytes, auth).await
39 }
40 p if p.starts_with("info/lfs/objects/") => {
41 let oid = &p["info/lfs/objects/".len()..];
42 match method {
43 Method::PUT => upload(service, space, repo, oid.to_string(), headers, body, auth).await,
44 Method::GET => download(service, space, repo, oid, auth).await,
45 _ => Ok(SmartError::BadRequest("method not allowed".into()).into_response()),
46 }
47 }
48 "info/lfs/locks" if method == Method::GET => list_locks(service, space, repo, auth).await,
49 "info/lfs/locks" if method == Method::POST => {
50 create_lock(service, space, repo, body, auth).await
51 }
52 "info/lfs/locks/verify" if method == Method::POST => {
53 verify_locks(service, space, repo, body, auth).await
54 }
55 p if p.starts_with("info/lfs/locks/") && p.ends_with("/unlock") && method == Method::POST => {
56 let id = &p["info/lfs/locks/".len()..p.len() - "/unlock".len()];
57 unlock(service, space, repo, id.to_string(), auth).await
58 }
59 _ => Ok(SmartError::NotFound("lfs endpoint not found".into()).into_response()),
60 }
61}
62
63async fn batch(
64 service: &SmartService,
65 space: &str,
66 repo: &str,
67 req: LfsBatchRequest,
68 auth: &BasicAuth,
69) -> SmartResult<Response> {
70 let svc = &service.git;
71 let ctx = service.resolve(space, repo).await?;
72 match req.operation {
73 LfsOperation::Upload | LfsOperation::Verify => {
74 require_writer(svc, auth, &ctx).await?;
75 }
76 LfsOperation::Download => {
77 allow_repo_read(svc, auth, &ctx).await?;
78 }
79 }
80 let resp = object::batch(svc, space.to_string(), repo.to_string(), req).await?;
81 Ok((
82 StatusCode::OK,
83 [(header::CONTENT_TYPE, "application/vnd.git-lfs+json")],
84 serde_json::to_vec(&resp).map_err(|e| SmartError::Internal(format!("serialize: {e}")))?,
85 )
86 .into_response())
87}
88
89async fn verify(
90 service: &SmartService,
91 space: &str,
92 repo: &str,
93 bytes: bytes::Bytes,
94 auth: &BasicAuth,
95) -> SmartResult<Response> {
96 let svc = &service.git;
97 let req: crate::lfs::types::LfsVerifyRequest = serde_json::from_slice(&bytes)
98 .map_err(|e| SmartError::BadRequest(format!("invalid verify request: {e}")))?;
99 let ctx = service.resolve(space, repo).await?;
100 require_writer(svc, auth, &ctx).await?;
101 object::verify(svc, space.to_string(), repo.to_string(), req).await?;
102 Ok(StatusCode::OK.into_response())
103}
104
105#[allow(clippy::too_many_arguments)]
106async fn upload(
107 service: &SmartService,
108 space: &str,
109 repo: &str,
110 oid: String,
111 headers: &HeaderMap,
112 body: Body,
113 auth: &BasicAuth,
114) -> SmartResult<Response> {
115 let svc = &service.git;
116 let ctx = service.resolve(space, repo).await?;
117 let uid = require_writer(svc, auth, &ctx).await?;
118 let canonical = crate::lfs::canonical_oid(&oid)?;
119 let size = headers
120 .get(header::CONTENT_LENGTH)
121 .and_then(|v| v.to_str().ok())
122 .and_then(|s| s.parse::<i64>().ok())
123 .unwrap_or(0);
124 let content_type = headers
125 .get(header::CONTENT_TYPE)
126 .and_then(|v| v.to_str().ok())
127 .map(str::to_string);
128 let stream = body
129 .into_data_stream()
130 .map(|chunk| chunk.map_err(|e| std::io::Error::other(e.to_string())));
131 let reader = StreamReader::new(stream);
132 object::upload_stream(
133 svc,
134 space.to_string(),
135 repo.to_string(),
136 uid,
137 canonical,
138 size,
139 content_type,
140 reader,
141 )
142 .await?;
143 Ok(StatusCode::OK.into_response())
144}
145
146async fn download(
147 service: &SmartService,
148 space: &str,
149 repo: &str,
150 oid: &str,
151 auth: &BasicAuth,
152) -> SmartResult<Response> {
153 let svc = &service.git;
154 let ctx = service.resolve(space, repo).await?;
155 allow_repo_read(svc, auth, &ctx).await?;
156 let canonical = crate::lfs::canonical_oid(oid)?;
157 let key = format!("lfs/{}/{}", ctx.model.id.simple(), canonical);
158 let stream = svc
159 .state
160 .storage
161 .client()
162 .get_stream(&key)
163 .await
164 .map_err(|e| SmartError::Internal(format!("lfs get: {e}")))?;
165 let body = Body::from_stream(
166 stream.map(|chunk| chunk.map_err(|e| -> axum::Error { axum::Error::new(e.to_string()) })),
167 );
168 Ok((
169 StatusCode::OK,
170 [(header::CONTENT_TYPE, "application/octet-stream")],
171 body,
172 )
173 .into_response())
174}
175
176async fn list_locks(
177 service: &SmartService,
178 space: &str,
179 repo: &str,
180 auth: &BasicAuth,
181) -> SmartResult<Response> {
182 let svc = &service.git;
183 let ctx = service.resolve(space, repo).await?;
184 allow_repo_read(svc, auth, &ctx).await?;
185 let resp = lock::list(svc, space.to_string(), repo.to_string(), None, 100).await?;
186 Ok((
187 StatusCode::OK,
188 [(header::CONTENT_TYPE, "application/vnd.git-lfs+json")],
189 serde_json::to_vec(&resp).map_err(|e| SmartError::Internal(format!("serialize: {e}")))?,
190 )
191 .into_response())
192}
193
194async fn create_lock(
195 service: &SmartService,
196 space: &str,
197 repo: &str,
198 body: Body,
199 auth: &BasicAuth,
200) -> SmartResult<Response> {
201 let svc = &service.git;
202 let ctx = service.resolve(space, repo).await?;
203 let uid = require_writer(svc, auth, &ctx).await?;
204 let bytes = axum::body::to_bytes(body, MAX_LFS_BODY)
205 .await
206 .map_err(|e| SmartError::BadRequest(format!("read body: {e}")))?;
207 let req: LfsLockCreateRequest = serde_json::from_slice(&bytes)
208 .map_err(|e| SmartError::BadRequest(format!("invalid lock request: {e}")))?;
209 let lock = crate::lfs::create_lock(svc, space.to_string(), repo.to_string(), uid, req.path, None).await?;
210 Ok((
211 StatusCode::CREATED,
212 [(header::CONTENT_TYPE, "application/vnd.git-lfs+json")],
213 serde_json::to_vec(&json!({ "lock": lock })).map_err(|e| SmartError::Internal(format!("serialize: {e}")))?,
214 )
215 .into_response())
216}
217
218async fn verify_locks(
219 service: &SmartService,
220 space: &str,
221 repo: &str,
222 body: Body,
223 auth: &BasicAuth,
224) -> SmartResult<Response> {
225 let svc = &service.git;
226 let ctx = service.resolve(space, repo).await?;
227 let uid = allow_repo_read(svc, auth, &ctx).await?;
228 let uid = uid.ok_or(SmartError::Unauthorized)?;
229 let bytes = axum::body::to_bytes(body, MAX_LFS_BODY)
230 .await
231 .map_err(|e| SmartError::BadRequest(format!("read body: {e}")))?;
232 let req: LfsVerifyLocksRequest = serde_json::from_slice(&bytes)
233 .map_err(|e| SmartError::BadRequest(format!("invalid request: {e}")))?;
234 let resp = lock::verify_locks(
235 svc,
236 space.to_string(),
237 repo.to_string(),
238 uid,
239 req.limit.unwrap_or(100) as u64,
240 )
241 .await?;
242 Ok((
243 StatusCode::OK,
244 [(header::CONTENT_TYPE, "application/vnd.git-lfs+json")],
245 serde_json::to_vec(&resp).map_err(|e| SmartError::Internal(format!("serialize: {e}")))?,
246 )
247 .into_response())
248}
249
250async fn unlock(
251 service: &SmartService,
252 space: &str,
253 repo: &str,
254 lock_id: String,
255 auth: &BasicAuth,
256) -> SmartResult<Response> {
257 let svc = &service.git;
258 let ctx = service.resolve(space, repo).await?;
259 let uid = require_writer(svc, auth, &ctx).await?;
260 crate::lfs::unlock_by_id(svc, space.to_string(), repo.to_string(), uid, lock_id, false).await?;
261 Ok((
262 StatusCode::OK,
263 [(header::CONTENT_TYPE, "application/vnd.git-lfs+json")],
264 serde_json::to_vec(&json!({ "message": "unlock successful" }))
265 .map_err(|e| SmartError::Internal(format!("serialize: {e}")))?,
266 )
267 .into_response())
268}
269
270async fn read_body(
271 _headers: &HeaderMap,
272 body: Body,
273 limit: usize,
274) -> SmartResult<bytes::Bytes> {
275 axum::body::to_bytes(body, limit)
276 .await
277 .map_err(|e| SmartError::BadRequest(format!("failed to read body: {e}")))
278}
No newline at end of file
Modified lib/git/src/smart/mod.rs lib/git/src/smart/mod.rs +18 −0 View lib/git/src/smart/mod.rs
@@ -1 +1,19 @@
1pub mod dispatch;
2pub mod error;
3pub mod handler;
4pub mod info_refs;
5pub mod lfs;
6pub mod object;
7pub mod pktline;
8pub mod pre_receive;
9pub mod push;
10pub mod rpc;
11pub mod service;
112
13pub use dispatch::{info_refs, rpc};
14pub use error::{SmartError, SmartResult};
15pub use handler::router;
16pub use pre_receive::{PreReceiveVerdict, RefUpdate, RefVerdict};
17pub use push::{NoopPushBackend, PushCoordinator, PushOperations, RedisPushBackend};
18pub use rpc::GitService;
19pub use service::SmartService;
No newline at end of file
Added lib/git/src/smart/object.rs lib/git/src/smart/object.rs +57 −0 View lib/git/src/smart/object.rs
@@ -0,0 +1,57 @@
1use std::path::{Component, Path, PathBuf};
2
3use axum::http::{StatusCode, header};
4use axum::response::{IntoResponse, Response};
5
6use crate::smart::error::{SmartError, SmartResult};
7
8pub async fn serve(
9 bare: &Path,
10 request_path: &str,
11) -> SmartResult<Response> {
12 let file = resolve_within_root(bare, request_path)?;
13 match tokio::fs::read(&file).await {
14 Ok(bytes) => Ok((
15 StatusCode::OK,
16 [(header::CONTENT_TYPE, guess_content_type(request_path))],
17 bytes,
18 )
19 .into_response()),
20 Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
21 Err(SmartError::NotFound(format!("file not found: {request_path}")))
22 }
23 Err(e) => Err(SmartError::Io(e)),
24 }
25}
26
27fn resolve_within_root(root: &Path, request_path: &str) -> SmartResult<PathBuf> {
28 let mut base = root.to_path_buf();
29 for component in Path::new(request_path).components() {
30 match component {
31 Component::Normal(part) => base.push(part),
32 Component::CurDir => {}
33 Component::ParentDir | Component::RootDir | Component::Prefix(_) => {
34 return Err(SmartError::Forbidden("path traversal denied".into()));
35 }
36 }
37 }
38 let canonical = base
39 .canonicalize()
40 .map_err(|_| SmartError::NotFound("file not found".into()))?;
41 let root_canonical = root
42 .canonicalize()
43 .map_err(|_| SmartError::NotFound("repository not found".into()))?;
44 if canonical.starts_with(&root_canonical) {
45 Ok(canonical)
46 } else {
47 Err(SmartError::Forbidden("path escapes repository root".into()))
48 }
49}
50
51fn guess_content_type(path: &str) -> &'static str {
52 if path.ends_with("/info/refs") || path.ends_with("info/refs") {
53 "text/plain; charset=utf-8"
54 } else {
55 "application/octet-stream"
56 }
57}
No newline at end of file
Added lib/git/src/smart/pktline.rs lib/git/src/smart/pktline.rs +50 −0 View lib/git/src/smart/pktline.rs
@@ -0,0 +1,50 @@
1use crate::smart::error::{SmartError, SmartResult};
2
3const MAX_PKT_SIZE: usize = 65_520;
4
5pub fn encode_line(line: &[u8]) -> Vec<u8> {
6 let len = line.len() + 4;
7 assert!(len <= MAX_PKT_SIZE, "pkt-line too long: {len}");
8 let mut out = Vec::with_capacity(4 + line.len());
9 out.extend_from_slice(format!("{len:04x}").as_bytes());
10 out.extend_from_slice(line);
11 out
12}
13
14pub fn flush_pkt() -> Vec<u8> {
15 b"0000".to_vec()
16}
17
18pub fn encode_header(service: &str) -> Vec<u8> {
19 let payload = format!("# service={service}\n");
20 let mut out = encode_line(payload.as_bytes());
21 out.extend_from_slice(&flush_pkt());
22 out
23}
24
25pub fn decode_stream(input: &[u8]) -> SmartResult<Vec<Vec<u8>>> {
26 let mut lines = Vec::new();
27 let mut pos = 0usize;
28 while pos + 4 <= input.len() {
29 let header = &input[pos..pos + 4];
30 let len = usize::from_str_radix(
31 std::str::from_utf8(header).map_err(|_| SmartError::BadRequest("invalid pkt-line length header".into()))?,
32 16,
33 )
34 .map_err(|_| SmartError::BadRequest("invalid pkt-line length header".into()))?;
35 pos += 4;
36 if len == 0 {
37 break;
38 }
39 if !(4..=MAX_PKT_SIZE).contains(&len) {
40 return Err(SmartError::BadRequest(format!("invalid pkt-line length: {len}")));
41 }
42 let payload_end = pos + len - 4;
43 if payload_end > input.len() {
44 return Err(SmartError::BadRequest("truncated pkt-line".into()));
45 }
46 lines.push(input[pos..payload_end].to_vec());
47 pos = payload_end;
48 }
49 Ok(lines)
50}
No newline at end of file
Added lib/git/src/smart/pre_receive.rs lib/git/src/smart/pre_receive.rs +254 −0 View lib/git/src/smart/pre_receive.rs
@@ -0,0 +1,254 @@
1use bytes::Bytes;
2use globset::{Glob, GlobMatcher};
3use sea_orm::{ColumnTrait, EntityTrait, QueryFilter};
4use uuid::Uuid;
5
6use crate::smart::error::{SmartError, SmartResult};
7use crate::smart::pktline;
8
9#[derive(Debug, Clone, PartialEq, Eq)]
10pub struct RefUpdate {
11 pub old_oid: String,
12 pub new_oid: String,
13 pub refname: String,
14}
15
16#[derive(Debug, Clone)]
17pub enum RefVerdict {
18 Accept,
19 Reject(String),
20}
21
22#[derive(Debug, Clone)]
23pub struct PreReceiveVerdict {
24 pub per_ref: Vec<(RefUpdate, RefVerdict)>,
25}
26
27impl PreReceiveVerdict {
28 pub fn accepted_only(self) -> SmartResult<Vec<RefUpdate>> {
29 let mut accepted = Vec::with_capacity(self.per_ref.len());
30 for (r, v) in self.per_ref {
31 match v {
32 RefVerdict::Accept => accepted.push(r),
33 RefVerdict::Reject(msg) => return Err(SmartError::Forbidden(msg)),
34 }
35 }
36 Ok(accepted)
37 }
38}
39
40pub async fn validate_pre_receive(
41 db: &sea_orm::DatabaseConnection,
42 baby: &gitbaby::GitBaby,
43 repo_id: Uuid,
44 actor_user_id: Uuid,
45 refs: &[RefUpdate],
46) -> SmartResult<PreReceiveVerdict> {
47 let repo = models::repo::repos::Entity::find_by_id(repo_id)
48 .one(db)
49 .await?
50 .ok_or_else(|| SmartError::NotFound(format!("repo {repo_id}")))?;
51
52 let branch_policies = models::repo::repo_branch_protection_policy::Entity::find()
53 .filter(models::repo::repo_branch_protection_policy::Column::RepoId.eq(repo.id))
54 .all(db)
55 .await?;
56 let tag_policies = models::repo::repo_tag_protection_policy::Entity::find()
57 .filter(models::repo::repo_tag_protection_policy::Column::RepoId.eq(repo.id))
58 .all(db)
59 .await?;
60
61 let actor_collab = models::repo::repo_collaborations::Entity::find()
62 .filter(models::repo::repo_collaborations::Column::RepoId.eq(repo.id))
63 .filter(models::repo::repo_collaborations::Column::UserId.eq(actor_user_id))
64 .one(db)
65 .await?;
66 let actor_is_admin = matches!(
67 actor_collab.map(|c| c.mode),
68 Some(models::repo::repo_collaborations::CollaborationMode::Admin)
69 );
70
71 let oid_len = oid_len_of(refs);
72 let zero_oid = "0".repeat(oid_len);
73
74 let mut out = Vec::with_capacity(refs.len());
75 for r in refs {
76 let verdict = judge(
77 baby,
78 r,
79 &branch_policies,
80 &tag_policies,
81 actor_user_id,
82 actor_is_admin,
83 &zero_oid,
84 )
85 .await?;
86 out.push((r.clone(), verdict));
87 }
88 Ok(PreReceiveVerdict { per_ref: out })
89}
90
91async fn judge(
92 baby: &gitbaby::GitBaby,
93 r: &RefUpdate,
94 branch_policies: &[models::repo::repo_branch_protection_policy::Model],
95 tag_policies: &[models::repo::repo_tag_protection_policy::Model],
96 actor_user_id: Uuid,
97 actor_is_admin: bool,
98 zero_oid: &str,
99) -> SmartResult<RefVerdict> {
100 if let Some(branch) = r.refname.strip_prefix("refs/heads/") {
101 return judge_branch(baby, branch, r, branch_policies, actor_is_admin, zero_oid).await;
102 }
103 if let Some(tag) = r.refname.strip_prefix("refs/tags/") {
104 return Ok(judge_tag(tag, tag_policies, actor_user_id));
105 }
106 Ok(RefVerdict::Accept)
107}
108
109async fn judge_branch(
110 baby: &gitbaby::GitBaby,
111 branch: &str,
112 r: &RefUpdate,
113 branch_policies: &[models::repo::repo_branch_protection_policy::Model],
114 actor_is_admin: bool,
115 zero_oid: &str,
116) -> SmartResult<RefVerdict> {
117 let matchers: Vec<(&models::repo::repo_branch_protection_policy::Model, GlobMatcher)> =
118 match branch_policies
119 .iter()
120 .filter_map(|m| Glob::new(&m.branch_pattern).ok().map(|g| (m, g.compile_matcher())))
121 .find(|(_, matcher)| matcher.is_match(branch))
122 {
123 Some(found) => vec![found],
124 None => Vec::new(),
125 };
126 if matchers.is_empty() {
127 return Ok(RefVerdict::Accept);
128 }
129 let policy = matchers[0].0;
130 let is_delete = r.new_oid == zero_oid;
131 let is_force = r.old_oid != zero_oid
132 && r.new_oid != zero_oid
133 && !is_fast_forward(baby, &r.old_oid, &r.new_oid).await?;
134
135 if is_delete {
136 return Ok(RefVerdict::Reject(format!(
137 "deleting protected branch `{branch}` is not allowed"
138 )));
139 }
140
141 if !policy.enable_push {
142 return Ok(RefVerdict::Reject(format!(
143 "pushes to protected branch `{branch}` are not allowed"
144 )));
145 }
146
147 if is_force && !policy.enable_force_push && !actor_is_admin {
148 return Ok(RefVerdict::Reject(format!(
149 "force-push to protected branch `{branch}` is not allowed"
150 )));
151 }
152
153 Ok(RefVerdict::Accept)
154}
155
156fn judge_tag(
157 tag: &str,
158 tag_policies: &[models::repo::repo_tag_protection_policy::Model],
159 _actor_user_id: Uuid,
160) -> RefVerdict {
161 for p in tag_policies {
162 let Ok(glob) = Glob::new(&p.name_pattern) else {
163 continue;
164 };
165 if !glob.compile_matcher().is_match(tag) {
166 continue;
167 }
168 return RefVerdict::Reject(format!(
169 "tag `{tag}` is protected; only the policy owner(s) can modify it"
170 ));
171 }
172 RefVerdict::Accept
173}
174
175/// Real fast-forward check via `git merge-base --is-ancestor old new`.
176/// `old == new` (no-op push) and `old == zero_oid` (new branch) are not
177/// ancestors checks — they short-circuit to true without spawning git.
178async fn is_fast_forward(
179 baby: &gitbaby::GitBaby,
180 old: &str,
181 new: &str,
182) -> SmartResult<bool> {
183 if old == new {
184 return Ok(true);
185 }
186 if old.chars().all(|c| c == '0') || new.chars().all(|c| c == '0') {
187 // zero_oid cases (delete / new branch) are handled upstream;
188 // treat as non-ff so the caller still rejects force push.
189 return Ok(false);
190 }
191 baby.is_ancestor(old, new)
192 .await
193 .map_err(|e| SmartError::Internal(format!("is_ancestor failed: {e}")))
194}
195
196fn oid_len_of(refs: &[RefUpdate]) -> usize {
197 refs.iter()
198 .find_map(|r| {
199 let max = r.old_oid.len().max(r.new_oid.len());
200 (max == 40 || max == 64).then_some(max)
201 })
202 .unwrap_or(40)
203}
204
205pub fn parse_receive_pack_refs(input: &[u8]) -> SmartResult<Vec<RefUpdate>> {
206 let lines = pktline::decode_stream(input).map_err(|e| {
207 SmartError::BadRequest(format!("invalid receive-pack request: {e}"))
208 })?;
209 let mut out = Vec::new();
210 for mut line in lines {
211 if let Some(nul) = line.iter().position(|b| *b == 0) {
212 line.truncate(nul);
213 }
214 if line.is_empty() {
215 continue;
216 }
217 let s = std::str::from_utf8(&line)
218 .map_err(|_| SmartError::BadRequest("receive-pack ref line is not valid UTF-8".to_string()))?;
219 let mut parts = s.splitn(3, ' ');
220 let old = parts.next().unwrap_or_default();
221 let new = parts.next().unwrap_or_default();
222 let refname = parts.next().unwrap_or_default();
223 if old.is_empty() || new.is_empty() || refname.is_empty() {
224 return Err(SmartError::BadRequest(format!(
225 "malformed receive-pack ref line: {s}"
226 )));
227 }
228 let oid_len = old.len();
229 if oid_len != 40 && oid_len != 64 {
230 return Err(SmartError::BadRequest(format!(
231 "invalid oid length {oid_len} in receive-pack ref line: {s}"
232 )));
233 }
234 if new.len() != oid_len
235 || !old.chars().all(|c| c.is_ascii_hexdigit())
236 || !new.chars().all(|c| c.is_ascii_hexdigit())
237 {
238 return Err(SmartError::BadRequest(format!(
239 "invalid oid in receive-pack ref line: {s}"
240 )));
241 }
242 out.push(RefUpdate {
243 old_oid: old.to_string(),
244 new_oid: new.to_string(),
245 refname: refname.to_string(),
246 });
247 }
248 Ok(out)
249}
250
251#[allow(dead_code)]
252pub fn update_pkts(input: &Bytes) -> SmartResult<Vec<RefUpdate>> {
253 parse_receive_pack_refs(input)
254}
No newline at end of file
Added lib/git/src/smart/push.rs lib/git/src/smart/push.rs +266 −0 View lib/git/src/smart/push.rs
@@ -0,0 +1,266 @@
1use std::sync::Arc;
2use std::time::Duration;
3
4use async_trait::async_trait;
5use export::{AppRedis, RedisConnection};
6use uuid::Uuid;
7
8use crate::smart::error::{SmartError, SmartResult};
9
10#[async_trait]
11pub trait PushOperations: Send + Sync {
12 async fn acquire(&self, key: &str, member: &[u8], ttl_secs: u64) -> SmartResult<bool>;
13 async fn renew(&self, key: &str, token: &[u8], ttl_secs: u64) -> SmartResult<bool>;
14 async fn release(&self, key: &str, token: &[u8]) -> SmartResult<()>;
15 async fn enqueue(&self, key: &str, member: &[u8]) -> SmartResult<()>;
16 async fn dequeue(&self, key: &str, member: &[u8]) -> SmartResult<()>;
17}
18
19macro_rules! dispatch {
20 ($conn:expr, |$c:ident| $body:expr) => {
21 match &mut $conn {
22 RedisConnection::Standalone($c) => $body,
23 RedisConnection::Cluster($c) => $body,
24 }
25 };
26}
27
28const RENEW_SCRIPT: &str =
29 "if redis.call('GET', KEYS[1]) == ARGV[1] then return redis.call('SET', KEYS[1], ARGV[1], 'EX', ARGV[2]) else return 0 end";
30const RELEASE_SCRIPT: &str =
31 "if redis.call('GET', KEYS[1]) == ARGV[1] then return redis.call('DEL', KEYS[1]) else return 0 end";
32
33pub struct RedisPushBackend {
34 pub kv: Arc<AppRedis>,
35}
36
37impl RedisPushBackend {
38 async fn eval_bool(&self, script: &str, key: &str, argv: Vec<Vec<u8>>) -> SmartResult<bool> {
39 let mut conn = self.kv.connection_clone();
40 let mut cmd = redis::cmd("EVAL");
41 cmd.arg(script).arg(1).arg(key);
42 for a in argv {
43 cmd.arg(a);
44 }
45 let n: i64 = dispatch!(conn, |c| cmd.query_async(c).await)?;
46 Ok(n != 0)
47 }
48}
49
50#[async_trait]
51impl PushOperations for RedisPushBackend {
52 async fn acquire(&self, key: &str, member: &[u8], ttl_secs: u64) -> SmartResult<bool> {
53 let mut conn = self.kv.connection_clone();
54 let resp: Option<String> = dispatch!(conn, |c| {
55 redis::cmd("SET")
56 .arg(key)
57 .arg(member)
58 .arg("NX")
59 .arg("EX")
60 .arg(ttl_secs)
61 .query_async(c)
62 .await
63 })?;
64 Ok(resp.is_some())
65 }
66
67 async fn renew(&self, key: &str, token: &[u8], ttl_secs: u64) -> SmartResult<bool> {
68 self.eval_bool(RENEW_SCRIPT, key, vec![token.to_vec(), ttl_secs.to_string().into_bytes()])
69 .await
70 }
71
72 async fn release(&self, key: &str, token: &[u8]) -> SmartResult<()> {
73 self.eval_bool(RELEASE_SCRIPT, key, vec![token.to_vec()]).await?;
74 Ok(())
75 }
76
77 async fn enqueue(&self, key: &str, member: &[u8]) -> SmartResult<()> {
78 let mut conn = self.kv.connection_clone();
79 let _: i64 = dispatch!(conn, |c| {
80 redis::cmd("RPUSH")
81 .arg(key)
82 .arg(member)
83 .query_async(c)
84 .await
85 })?;
86 Ok(())
87 }
88
89 async fn dequeue(&self, key: &str, member: &[u8]) -> SmartResult<()> {
90 let mut conn = self.kv.connection_clone();
91 let _: i64 = dispatch!(conn, |c| {
92 redis::cmd("LREM")
93 .arg(key)
94 .arg(1)
95 .arg(member)
96 .query_async(c)
97 .await
98 })?;
99 Ok(())
100 }
101}
102
103pub struct NoopPushBackend;
104
105#[async_trait]
106impl PushOperations for NoopPushBackend {
107 async fn acquire(
108 &self,
109 _key: &str,
110 _member: &[u8],
111 _ttl_secs: u64,
112 ) -> SmartResult<bool> {
113 Err(SmartError::Internal(
114 "push coordination disabled: redis not configured".into(),
115 ))
116 }
117
118 async fn renew(&self, _key: &str, _token: &[u8], _ttl_secs: u64) -> SmartResult<bool> {
119 Err(SmartError::Internal(
120 "push coordination disabled: redis not configured".into(),
121 ))
122 }
123
124 async fn release(&self, _key: &str, _token: &[u8]) -> SmartResult<()> {
125 Ok(())
126 }
127
128 async fn enqueue(&self, _key: &str, _member: &[u8]) -> SmartResult<()> {
129 Err(SmartError::Internal(
130 "push coordination disabled: redis not configured".into(),
131 ))
132 }
133
134 async fn dequeue(&self, _key: &str, _member: &[u8]) -> SmartResult<()> {
135 Err(SmartError::Internal(
136 "push coordination disabled: redis not configured".into(),
137 ))
138 }
139}
140
141#[derive(Clone)]
142pub struct PushCoordinator {
143 backend: Arc<dyn PushOperations>,
144 lock_ttl_secs: u64,
145 renew_interval_secs: u64,
146 poll_interval: Duration,
147}
148
149const DEFAULT_LOCK_TTL_SECS: u64 = 120;
150const DEFAULT_RENEW_INTERVAL_SECS: u64 = 40;
151const DEFAULT_POLL_INTERVAL: Duration = Duration::from_millis(200);
152
153impl PushCoordinator {
154 pub fn new(backend: Arc<dyn PushOperations>) -> Self {
155 Self {
156 backend,
157 lock_ttl_secs: DEFAULT_LOCK_TTL_SECS,
158 renew_interval_secs: DEFAULT_RENEW_INTERVAL_SECS,
159 poll_interval: DEFAULT_POLL_INTERVAL,
160 }
161 }
162
163 pub async fn run<T>(
164 &self,
165 key: &str,
166 f: impl std::future::Future<Output = SmartResult<T>>,
167 ) -> SmartResult<T> {
168 let token = Uuid::now_v7().to_string();
169 let result: SmartResult<T> = async {
170 if self
171 .backend
172 .acquire(key, token.as_bytes(), self.lock_ttl_secs)
173 .await?
174 {
175 return self.run_locked(key, &token, f).await;
176 }
177
178 self.backend.enqueue(key, token.as_bytes()).await?;
179 loop {
180 tokio::time::sleep(self.poll_interval).await;
181 if self
182 .backend
183 .acquire(key, token.as_bytes(), self.lock_ttl_secs)
184 .await?
185 {
186 self.backend.dequeue(key, token.as_bytes()).await?;
187 return self.run_locked(key, &token, f).await;
188 }
189 }
190 }
191 .await;
192 result
193 }
194
195 async fn run_locked<T>(
196 &self,
197 key: &str,
198 token: &str,
199 f: impl std::future::Future<Output = SmartResult<T>>,
200 ) -> SmartResult<T> {
201 let heartbeat = Heartbeat::spawn(
202 self.backend.clone(),
203 key.to_string(),
204 token.to_string(),
205 self.lock_ttl_secs,
206 self.renew_interval_secs,
207 );
208 let result = f.await;
209 heartbeat.stop().await;
210 self.backend.release(key, token.as_bytes()).await?;
211 result
212 }
213}
214
215struct Heartbeat {
216 handle: Option<tokio::task::JoinHandle<()>>,
217 stop: Arc<tokio::sync::Notify>,
218}
219
220impl Drop for Heartbeat {
221 fn drop(&mut self) {
222 self.stop.notify_one();
223 if let Some(handle) = &self.handle
224 && !handle.is_finished()
225 {
226 handle.abort();
227 }
228 }
229}
230
231impl Heartbeat {
232 fn spawn(
233 backend: Arc<dyn PushOperations>,
234 key: String,
235 token: String,
236 ttl: u64,
237 interval: u64,
238 ) -> Self {
239 let stop = Arc::new(tokio::sync::Notify::new());
240 let stop_clone = stop.clone();
241 let handle = tokio::spawn(async move {
242 loop {
243 tokio::select! {
244 _ = stop_clone.notified() => break,
245 _ = tokio::time::sleep(Duration::from_secs(interval)) => {
246 match backend.renew(&key, token.as_bytes(), ttl).await {
247 Ok(false) | Err(_) => break,
248 Ok(true) => {}
249 }
250 }
251 }
252 }
253 });
254 Self {
255 handle: Some(handle),
256 stop,
257 }
258 }
259
260 async fn stop(mut self) {
261 self.stop.notify_one();
262 if let Some(handle) = self.handle.take() {
263 let _ = handle.await;
264 }
265 }
266}
No newline at end of file
Added lib/git/src/smart/rpc.rs lib/git/src/smart/rpc.rs +152 −0 View lib/git/src/smart/rpc.rs
@@ -0,0 +1,152 @@
1use std::path::Path;
2use std::process::Stdio;
3use std::sync::Arc;
4
5use bytes::Bytes;
6use tokio::io::{AsyncReadExt, AsyncWriteExt};
7use tokio::process::Command;
8use tokio::sync::Mutex;
9use tokio::task::JoinHandle;
10
11use crate::smart::error::{SmartError, SmartResult};
12
13pub type RpcStream = futures::stream::BoxStream<'static, Result<Bytes, std::io::Error>>;
14
15#[derive(Debug, Clone, Copy, PartialEq, Eq)]
16pub enum GitService {
17 UploadPack,
18 ReceivePack,
19}
20
21impl GitService {
22 pub fn from_name(name: &str) -> SmartResult<Self> {
23 match name {
24 "git-upload-pack" => Ok(Self::UploadPack),
25 "git-receive-pack" => Ok(Self::ReceivePack),
26 _ => Err(SmartError::BadRequest(format!(
27 "unsupported git service: {name}"
28 ))),
29 }
30 }
31
32 pub fn name(self) -> &'static str {
33 match self {
34 Self::UploadPack => "upload-pack",
35 Self::ReceivePack => "receive-pack",
36 }
37 }
38
39 pub fn dashed_name(self) -> &'static str {
40 match self {
41 Self::UploadPack => "git-upload-pack",
42 Self::ReceivePack => "git-receive-pack",
43 }
44 }
45
46 pub fn advertisement_content_type(self) -> &'static str {
47 match self {
48 Self::UploadPack => "application/x-git-upload-pack-advertisement",
49 Self::ReceivePack => "application/x-git-receive-pack-advertisement",
50 }
51 }
52
53 pub fn result_content_type(self) -> &'static str {
54 match self {
55 Self::UploadPack => "application/x-git-upload-pack-result",
56 Self::ReceivePack => "application/x-git-receive-pack-result",
57 }
58 }
59}
60
61pub async fn run_rpc_bytes(
62 bare_path: &Path,
63 service: GitService,
64 advertise: bool,
65 input: Bytes,
66) -> SmartResult<Bytes> {
67 let mut cmd = Command::new("git");
68 cmd.arg(service.name())
69 .arg("--stateless-rpc");
70 if advertise {
71 cmd.arg("--advertise-refs");
72 }
73 cmd.arg(bare_path)
74 .env("GIT_DIR", bare_path)
75 .env("GIT_PROTOCOL", "version=2")
76 .env("GIT_TERMINAL_PROMPT", "0")
77 .env("GIT_CONFIG_NOSYSTEM", "1")
78 .env("GIT_NO_REPLACE_OBJECTS", "1")
79 .stdin(Stdio::piped())
80 .stdout(Stdio::piped())
81 .stderr(Stdio::piped());
82
83 let service_name = service.name();
84 let mut child = cmd
85 .spawn()
86 .map_err(|e| SmartError::Internal(format!("failed to spawn git {service_name}: {e}")))?;
87
88 let mut stdin = child.stdin.take().expect("stdin piped");
89 let mut stdout = child.stdout.take().expect("stdout piped");
90 let mut stderr = child.stderr.take().expect("stderr piped");
91
92 let stdin_task = tokio::spawn(async move {
93 if stdin.write_all(&input).await.is_err() {
94 return Err(std::io::Error::other("stdin write failed"));
95 }
96 let _ = stdin.shutdown().await;
97 Ok(())
98 });
99
100 let child_guard = Arc::new(Mutex::new(Some(child)));
101 let stderr_buf = Arc::new(Mutex::new(Vec::<u8>::new()));
102
103 let stderr_writer = stderr_buf.clone();
104 let stderr_task: JoinHandle<()> = tokio::spawn(async move {
105 let mut chunk = vec![0u8; 4096];
106 loop {
107 match stderr.read(&mut chunk).await {
108 Ok(0) | Err(_) => break,
109 Ok(n) => stderr_writer.lock().await.extend_from_slice(&chunk[..n]),
110 }
111 }
112 });
113
114 let mut out = Vec::new();
115 let mut buf = vec![0u8; 8192];
116 loop {
117 match stdout.read(&mut buf).await {
118 Ok(0) => break,
119 Ok(n) => out.extend_from_slice(&buf[..n]),
120 Err(e) => {
121 let _ = stderr_task.await;
122 let msg = format!(
123 "git {service_name} stdout read failed: {e}; stderr: {}",
124 String::from_utf8_lossy(&stderr_buf.lock().await)
125 );
126 return Err(SmartError::Internal(msg));
127 }
128 }
129 }
130
131 let _ = stderr_task.await;
132 let mut guard = child_guard.lock().await;
133 let Some(mut child) = guard.take() else {
134 return Err(SmartError::Internal(format!(
135 "git {service_name} was killed"
136 )));
137 };
138 let status = child.wait().await.map_err(|e| {
139 SmartError::Internal(format!("waiting for git {service_name} failed: {e}"))
140 })?;
141
142 let _ = stdin_task.await;
143
144 if status.success() {
145 Ok(Bytes::from(out))
146 } else {
147 Err(SmartError::Internal(format!(
148 "git {service_name} exited with {status}: {}",
149 String::from_utf8_lossy(&stderr_buf.lock().await)
150 )))
151 }
152}
No newline at end of file
Added lib/git/src/smart/service.rs lib/git/src/smart/service.rs +103 −0 View lib/git/src/smart/service.rs
@@ -0,0 +1,103 @@
1use std::sync::Arc;
2
3use export::AppRedis;
4use sea_orm::{ColumnTrait, EntityTrait, QueryFilter};
5use uuid::Uuid;
6
7use crate::context::facade::GitContext;
8use crate::service::GitService;
9use crate::smart::error::{SmartError, SmartResult};
10use crate::smart::push::{NoopPushBackend, PushCoordinator, PushOperations, RedisPushBackend};
11use crate::sync::queue::SyncQueue;
12use config::git::SyncSettings;
13
14#[derive(Clone)]
15pub struct SmartService {
16 pub git: GitService,
17 pub kv: Option<Arc<AppRedis>>,
18 pub queue: Option<Arc<SyncQueue>>,
19 pub sync_settings: SyncSettings,
20}
21
22impl SmartService {
23 pub fn new(git: GitService) -> Self {
24 Self {
25 git,
26 kv: None,
27 queue: None,
28 sync_settings: SyncSettings::default(),
29 }
30 }
31
32 pub fn with_kv(mut self, kv: Option<Arc<AppRedis>>) -> Self {
33 self.kv = kv;
34 self
35 }
36
37 pub fn with_queue(mut self, queue: Option<Arc<SyncQueue>>) -> Self {
38 self.queue = queue;
39 self
40 }
41
42 pub fn with_sync_settings(mut self, settings: SyncSettings) -> Self {
43 self.sync_settings = settings;
44 self
45 }
46
47 pub fn push_coordinator(&self) -> PushCoordinator {
48 let backend: Arc<dyn PushOperations> = match &self.kv {
49 Some(kv) => Arc::new(RedisPushBackend { kv: kv.clone() }),
50 None => Arc::new(NoopPushBackend),
51 };
52 PushCoordinator::new(backend)
53 }
54
55 pub fn bare_path(&self, ctx: &GitContext) -> std::path::PathBuf {
56 ctx.repo_dir()
57 }
58
59 pub async fn resolve(&self, space: &str, repo: &str) -> SmartResult<GitContext> {
60 self.git
61 .context(space.to_string(), repo.to_string())
62 .await
63 .map_err(|e| match e {
64 crate::share::errors::GitError::NotFound => SmartError::NotFound(format!("repo {space}/{repo}")),
65 other => SmartError::from(other),
66 })
67 }
68
69 pub async fn owning_space_id(&self, ctx: &GitContext) -> SmartResult<Uuid> {
70 let space_name = self.space_for(ctx).await?;
71 let Some(space) = models::space::spaces::Entity::find()
72 .filter(models::space::spaces::Column::LowerName.eq(space_name.to_lowercase()))
73 .one(self.git.db_read())
74 .await?
75 else {
76 return Err(SmartError::NotFound(format!("space {space_name}")));
77 };
78 Ok(space.id)
79 }
80
81 async fn space_for(&self, ctx: &GitContext) -> SmartResult<String> {
82 let mut dir_id = ctx.model.dir_id;
83 loop {
84 let dir = models::dir::dirs::Entity::find_by_id(dir_id)
85 .one(self.git.db_read())
86 .await?;
87 let dir = dir.ok_or_else(|| SmartError::NotFound(format!("dir {dir_id}")))?;
88 if let Some(space_id) = dir.space_id {
89 let space = models::space::spaces::Entity::find_by_id(space_id)
90 .one(self.git.db_read())
91 .await?
92 .ok_or_else(|| SmartError::NotFound(format!("space {space_id}")))?;
93 return Ok(space.lower_name);
94 }
95 match dir.parent_dir_id {
96 Some(parent) => dir_id = parent,
97 None => return Err(SmartError::NotFound("space".into())),
98 }
99 }
100 }
101}
102
103pub use crate::smart::error::SmartError as SmartServiceError;
No newline at end of file