refactor(lib/git): harden smart/ssh/sync layers, add gitcmd runner and timeouts

Modified lib/config/src/git.rs +6 −0 View lib/config/src/git.rs

@@ -7,6 +7,8 @@

77 pub repo_root: String,

88 pub objects_root: String,

99 pub max_blob_size: u64,

10 pub git_command_timeout_secs: u64,

11 pub git_rpc_timeout_secs: u64,

1012 pub sync: SyncSettings,

1113 pub ssh: GitSshConfig,

1214}

@@ -17,6 +19,8 @@

1719 repo_root: "/opt/gitdata/repos".to_string(),

1820 objects_root: "/opt/gitdata/objects".to_string(),

1921 max_blob_size: 100 * 1024 * 1024,

22 git_command_timeout_secs: 300,

23 git_rpc_timeout_secs: 3600,

2024 sync: SyncSettings::default(),

2125 ssh: GitSshConfig::default(),

2226 }

@@ -65,6 +69,7 @@

6569 pub max_languages: usize,

6670 pub language_timeout_secs: u64,

6771 pub brpop_timeout_secs: u64,

72 pub git_command_timeout_secs: u64,

6873}

6974

7075impl Default for SyncSettings {

@@ -79,6 +84,7 @@

7984 max_languages: 20,

8085 language_timeout_secs: 30,

8186 brpop_timeout_secs: 5,

87 git_command_timeout_secs: 120,

8288 }

8389 }

8490}

No newline at end of file

Modified lib/git/src/graphql/query/repo.rs +25 −15 View lib/git/src/graphql/query/repo.rs

@@ -58,19 +58,21 @@

5858

5959 async fn repo_stats(&self) -> Result<RepoStatsObj, async_graphql::Error> {

6060 let paths = self.inner.ctx.repo_dir();

61 let out = tokio::task::spawn_blocking({

62 let paths = paths.clone();

63 move || {

64 use std::process::Command;

65 Command::new("git")

66 .arg("-C")

67 .arg(&paths)

68 .args(["count-objects", "-v"])

69 .output()

70 }

71 })

61 let timeout = self

62 .inner

63 .svc

64 .state

65 .config

66 .git

67 .git_command_timeout_secs;

68 let out = crate::service::gitcmd::run_git_soft(

69 "graphql::repo_stats",

70 &paths,

71 vec!["count-objects".to_string(), "-v".to_string()],

72 &[],

73 timeout,

74 )

7275 .await

73 .map_err(map_git_baby)?

7476 .map_err(map_git_baby)?;

7577 if !out.status.success() {

7678 return Err(async_graphql::Error::new(

@@ -133,9 +135,17 @@

133135

134136 async fn blob(&self, oid: String) -> Result<BlobObj, async_graphql::Error> {

135137 let id = parse_oid(&oid)?;

136 let baby = self.baby();

137 let blob = baby

138 .get_blob(id, gitbaby::blob::types::GetBlobOptions { limit: None })

138 let cap = self

139 .inner

140 .svc

141 .state

142 .config

143 .git

144 .max_blob_size

145 .min(i64::MAX as u64) as i64;

146 let blob = self

147 .baby()

148 .get_blob(id, gitbaby::blob::types::GetBlobOptions { limit: Some(cap) })

139149 .await

140150 .map_err(map_git_baby)?;

141151 Ok(BlobObj::new(blob.info.oid, &blob.data, blob.info.size))

Modified lib/git/src/graphql/query/root.rs +14 −1 View lib/git/src/graphql/query/root.rs

@@ -1,7 +1,10 @@

1use async_graphql::{Context, EmptyMutation, EmptySubscription, Object, Schema};

1use async_graphql::{Context, EmptyMutation, EmptySubscription, ErrorExtensions, Object, Schema};

22use std::sync::Arc;

3use uuid::Uuid;

34

45use crate::service::GitService;

6use crate::share::access::authorize_repo_access;

7use crate::share::errors::GitError;

58

69use super::context::{RepoCtx, map_git};

710use super::repo::Repo;

@@ -27,6 +30,16 @@

2730 .context(space_name.clone(), repo_name.clone())

2831 .await

2932 .map_err(map_git)?;

33 if ctx_data.model.is_private {

34 let uid = ctx

35 .data::<Option<Uuid>>()

36 .ok()

37 .and_then(|o| *o)

38 .ok_or_else(|| GitError::Unauthorized.extend())?;

39 authorize_repo_access(&svc, uid, &ctx_data, false, false)

40 .await

41 .map_err(map_git)?;

42 }

3043 let baby = svc.baby(&ctx_data);

3144 Ok(Repo {

3245 inner: RepoCtx {

Modified lib/git/src/handler/mod.rs +3 −1 View lib/git/src/handler/mod.rs

@@ -25,4 +25,6 @@

2525pub mod tags;

2626pub mod trees;

2727

28pub use share::{RepoPath, RepoPathRev, ensure_caller, require_writer, resolve_reader};

No newline at end of file

28pub use share::{

29 RepoPath, RepoPathRev, ensure_caller, require_repo_admin, require_writer, resolve_reader,

30};

No newline at end of file

Modified lib/git/src/handler/share.rs +17 −0 View lib/git/src/handler/share.rs

@@ -92,4 +92,21 @@

9292 .await

9393 .map_err(super::super::errors::GitApiError::from)?;

9494 Ok((ctx, uid))

95}

96

97pub async fn require_repo_admin(

98 service: &GitService,

99 session: &Session,

100 space_name: String,

101 repo_name: String,

102) -> Result<(crate::context::facade::GitContext, Uuid), ApiError> {

103 let ctx = service

104 .context(space_name, repo_name)

105 .await

106 .map_err(super::super::errors::GitApiError::from)?;

107 let uid = session.uid().ok_or(ApiError::unauthorized())?;

108 authorize_repo_access(service, uid, &ctx, true, true)

109 .await

110 .map_err(super::super::errors::GitApiError::from)?;

111 Ok((ctx, uid))

95112}

No newline at end of file

Modified lib/git/src/lfs/handle.rs +3 −6 View lib/git/src/lfs/handle.rs

@@ -356,12 +356,6 @@

356356 }

357357 }

358358

359 fn forbidden(message: &str) -> Self {

360 Self {

361 response: status_json(StatusCode::FORBIDDEN, message),

362 }

363 }

364

365359 fn bad_request(message: &str) -> Self {

366360 Self {

367361 response: status_json(StatusCode::BAD_REQUEST, message),

@@ -381,6 +375,9 @@

381375 .unwrap_or(StatusCode::INTERNAL_SERVER_ERROR);

382376 let message = match &err {

383377 GitError::Unauthorized => "authentication required".to_string(),

378 GitError::Cache(_) | GitError::Git(_) | GitError::Internal(_) => {

379 "internal server error".to_string()

380 }

384381 _ => err.to_string(),

385382 };

386383 if status == StatusCode::UNAUTHORIZED {

Modified lib/git/src/lfs/object.rs +40 −32 View lib/git/src/lfs/object.rs

@@ -21,6 +21,7 @@

2121

2222pub const UPLOAD_TTL_SECS: u32 = 3600;

2323pub const DOWNLOAD_TTL_SECS: u32 = 3600;

24pub const MAX_BATCH_OBJECTS: usize = 200;

2425

2526

2627pub async fn batch(

@@ -29,6 +30,11 @@

2930 repo_name: String,

3031 req: LfsBatchRequest,

3132) -> Result<LfsBatchResponse, GitError> {

33 if req.objects.len() > MAX_BATCH_OBJECTS {

34 return Err(GitError::InvalidArgument(format!(

35 "batch request exceeds {MAX_BATCH_OBJECTS} objects"

36 )));

37 }

3238 let ctx = svc.context(space_name.clone(), repo_name.clone()).await?;

3339 if !ctx.model.enable_lfs {

3440 return Err(GitError::NotFound);

@@ -164,23 +170,16 @@

164170 }

165171 let oid = canonical_oid(&oid)?;

166172 validate_size(size, svc.state.config.git.max_blob_size)?;

167 if let Some(cap) = ctx.model.storage_limit_size

168 && ctx.model.lfs_size + size > cap

169 {

170 return Err(GitError::InvalidArgument(

171 "lfs storage limit exceeded".to_string(),

172 ));

173 }

174173 let storage: &AppStorage = &svc.state.storage;

175174 let key = storage_key(&ctx.model.id, &oid);

176 let mut reader = HashingReader::new(reader);

177

178 if !storage

175 let existed = storage

179176 .client()

180177 .exists(&key)

181178 .await

182 .map_err(|e| GitError::Internal(format!("storage exists: {e}")))?

183 {

179 .map_err(|e| GitError::Internal(format!("storage exists: {e}")))?;

180

181 if !existed {

182 let mut reader = HashingReader::new(tokio::io::AsyncReadExt::take(reader, size as u64 + 1));

184183 let sent = storage

185184 .client()

186185 .put_stream(

@@ -224,18 +223,34 @@

224223 other => return Err(other),

225224 },

226225 }

227 repos::Entity::update_many()

228 .col_expr(

229 repos::Column::LfsSize,

230 Expr::col(repos::Column::LfsSize).add(size),

231 )

232 .col_expr(

233 repos::Column::UpdatedAt,

234 Expr::val(OffsetDateTime::now_utc()),

235 )

236 .filter(repos::Column::Id.eq(ctx.model.id))

237 .exec(svc.db_write())

238 .await?;

226

227 if !existed {

228 let update = repos::Entity::update_many()

229 .col_expr(

230 repos::Column::LfsSize,

231 Expr::col(repos::Column::LfsSize).add(size),

232 )

233 .col_expr(

234 repos::Column::UpdatedAt,

235 Expr::val(OffsetDateTime::now_utc()),

236 )

237 .filter(repos::Column::Id.eq(ctx.model.id));

238 let res = if let Some(cap) = ctx.model.storage_limit_size {

239 let limit = cap.saturating_sub(size);

240 update

241 .filter(Expr::col(repos::Column::LfsSize).lte(limit))

242 .exec(svc.db_write())

243 .await?

244 } else {

245 update.exec(svc.db_write()).await?

246 };

247 if res.rows_affected == 0 {

248 let _ = storage.client().delete(&key).await;

249 return Err(GitError::InvalidArgument(

250 "lfs storage limit exceeded".to_string(),

251 ));

252 }

253 }

239254 Ok(true)

240255}

241256

@@ -424,7 +439,6 @@

424439 inner: R,

425440 hasher: Sha256,

426441 count: u64,

427 finished: bool,

428442}

429443

430444impl<R> HashingReader<R> {

@@ -433,14 +447,10 @@

433447 inner,

434448 hasher: Sha256::new(),

435449 count: 0,

436 finished: false,

437450 }

438451 }

439452

440 /// Returns `(sha256_hex, bytes_read)`; panics if the stream was not fully

441 /// consumed (which the storage layer guarantees on success).

442453 fn finish(self) -> (String, u64) {

443 assert!(self.finished, "reader not fully consumed");

444454 let hex = self

445455 .hasher

446456 .finalize()

@@ -461,9 +471,7 @@

461471 let poll = Pin::new(&mut self.inner).poll_read(cx, buf);

462472 if poll.is_ready() {

463473 let written = &buf.filled()[before..];

464 if written.is_empty() && buf.remaining() > 0 {

465 self.finished = true;

466 } else {

474 if !(written.is_empty() && buf.remaining() > 0) {

467475 self.hasher.update(written);

468476 self.count += written.len() as u64;

469477 }

Modified lib/git/src/service/cache.rs +1 −1 View lib/git/src/service/cache.rs

@@ -7,7 +7,7 @@

77

88pub const REV_TTL: Duration = Duration::from_secs(60);

99

10fn sanitize_key(k: &str) -> String {

10pub(crate) fn sanitize_key(k: &str) -> String {

1111 let mut out = String::with_capacity(k.len());

1212 for c in k.chars() {

1313 if c.is_ascii_alphanumeric() || matches!(c, '-' | '_' | '.' | ':') {

Modified lib/git/src/service/context.rs +15 −21 View lib/git/src/service/context.rs

@@ -2,13 +2,13 @@

22use std::sync::Arc;

33

44use cache::CacheClient;

5use models::dir::dirs;

65use models::repo::repos;

76use models::space::spaces;

87use sea_orm::{ColumnTrait, EntityTrait, QueryFilter};

98

109use crate::context::facade::GitContext;

1110use crate::service::GitService;

11use crate::share::access::space_for_dir;

1212use crate::share::errors::GitError;

1313

1414impl GitService {

@@ -32,29 +32,23 @@

3232 if repo_lower.is_empty() {

3333 return Err(GitError::NotFound);

3434 }

35 let repo = repos::Entity::find()

35 let candidates = repos::Entity::find()

3636 .filter(repos::Column::LowerName.eq(&repo_lower))

37 .one(self.db_read())

38 .await?

39 .ok_or(GitError::NotFound)?;

40

41 let mut dir_id = repo.dir_id;

42 let owned_by_space = loop {

43 let dir = dirs::Entity::find_by_id(dir_id)

44 .one(self.db_read())

45 .await?

46 .ok_or(GitError::NotFound)?;

47 if dir.space_id == Some(space.id) {

48 break true;

49 }

50 match dir.parent_dir_id {

51 Some(parent) => dir_id = parent,

52 None => break false,

37 .all(self.db_read())

38 .await?;

39 let repo = {

40 let mut found = None;

41 for candidate in candidates {

42 if space_for_dir(self.db_read(), candidate.dir_id)

43 .await?

44 .is_some_and(|s| s.id == space.id)

45 {

46 found = Some(candidate);

47 break;

48 }

5349 }

50 found.ok_or(GitError::NotFound)?

5451 };

55 if !owned_by_space {

56 return Err(GitError::NotFound);

57 }

5852

5953 let cache = Arc::new(CacheClient::new(&self.state.config.cache).await?);

6054 let git = &self.state.config.git;

Added lib/git/src/service/gitcmd.rs +118 −0 View lib/git/src/service/gitcmd.rs

@@ -0,0 +1,118 @@

1use std::path::Path;

2

3use gitbaby::command::cmd::{Cmd, CmdOutput};

4use gitbaby::command::env::Env;

5

6use crate::share::errors::GitError;

7

8fn build_cmd(

9 label: &str,

10 dir: &Path,

11 args: Vec<String>,

12 env: &[(String, String)],

13 timeout_secs: u64,

14) -> Cmd {

15 let mut cmd = Cmd::new(

16 label.to_string(),

17 "git",

18 args,

19 Vec::new(),

20 dir.to_path_buf(),

21 Env::new(),

22 Some(timeout_secs),

23 );

24 cmd.env.set("GIT_TERMINAL_PROMPT", "0");

25 for (k, v) in env {

26 cmd.env.set(k.clone(), v.clone());

27 }

28 cmd

29}

30

31pub async fn run_git(

32 label: impl Into<String>,

33 dir: &Path,

34 args: Vec<String>,

35 env: &[(String, String)],

36 timeout_secs: u64,

37) -> Result<CmdOutput, GitError> {

38 let label = label.into();

39 let mut cmd = build_cmd(&label, dir, args, env, timeout_secs);

40 cmd.run()

41 .await

42 .map_err(|e| GitError::Internal(format!("{label} failed: {e}")))

43}

44

45pub async fn run_git_soft(

46 label: impl Into<String>,

47 dir: &Path,

48 args: Vec<String>,

49 env: &[(String, String)],

50 timeout_secs: u64,

51) -> Result<CmdOutput, GitError> {

52 let label = label.into();

53 let mut cmd = build_cmd(&label, dir, args, env, timeout_secs);

54 cmd.run_soft()

55 .await

56 .map_err(|e| GitError::Internal(format!("{label} failed: {e}")))

57}

58

59pub async fn run_git_capped(

60 label: impl Into<String>,

61 dir: &Path,

62 args: Vec<String>,

63 env: &[(String, String)],

64 timeout_secs: u64,

65 output_cap: usize,

66) -> Result<CmdOutput, GitError> {

67 let label = label.into();

68 let mut cmd = build_cmd(&label, dir, args, env, timeout_secs);

69 cmd.output_cap = output_cap;

70 cmd.run()

71 .await

72 .map_err(|e| GitError::Internal(format!("{label} failed: {e}")))

73}

74

75pub async fn run_exec_cmd(

76 label: impl Into<String>,

77 prog: &str,

78 dir: &Path,

79 args: &[String],

80 timeout_secs: u64,

81 output_cap: usize,

82) -> Result<CmdOutput, GitError> {

83 let label = label.into();

84 let mut cmd = Cmd::new(

85 label.clone(),

86 prog.to_string(),

87 args.to_vec(),

88 Vec::new(),

89 dir.to_path_buf(),

90 Env::new(),

91 Some(timeout_secs),

92 );

93 cmd.output_cap = output_cap;

94 cmd.run()

95 .await

96 .map_err(|e| GitError::Internal(format!("{label} failed: {e}")))

97}

98

99pub async fn run_git_with_stdin(

100 label: impl Into<String>,

101 dir: &Path,

102 args: Vec<String>,

103 env: &[(String, String)],

104 stdin_data: &[u8],

105 timeout_secs: u64,

106) -> Result<CmdOutput, GitError> {

107 let label = label.into();

108 let mut cmd = build_cmd(&label, dir, args, env, timeout_secs);

109 cmd.spawn()

110 .await

111 .map_err(|e| GitError::Internal(format!("{label} spawn failed: {e}")))?;

112 cmd.feed(stdin_data)

113 .await

114 .map_err(|e| GitError::Internal(format!("{label} stdin failed: {e}")))?;

115 cmd.run()

116 .await

117 .map_err(|e| GitError::Internal(format!("{label} failed: {e}")))

118}

No newline at end of file

Modified lib/git/src/service/mod.rs +1 −0 View lib/git/src/service/mod.rs

@@ -40,6 +40,7 @@

4040pub mod content;

4141pub mod context;

4242pub mod diffs;

43pub mod gitcmd;

4344pub mod hooks;

4445pub mod lfs;

4546pub mod logs;

Modified lib/git/src/service/signatures.rs +23 −9 View lib/git/src/service/signatures.rs

@@ -1,11 +1,10 @@

1use std::process::Command;

2

31use serde::Serialize;

42

53use crate::service::GitService;

4use crate::service::gitcmd::run_git_soft;

65use crate::share::errors::{GitError, map_baby};

76

8#[derive(Debug, Clone, Serialize)]

7#[derive(Debug, Clone, Serialize, utoipa::ToSchema)]

98pub struct VerifyResultDto {

109 pub signed: bool,

1110 pub valid: bool,

@@ -21,13 +20,28 @@

2120 repo_name: String,

2221 sha: String,

2322 ) -> Result<VerifyResultDto, GitError> {

23 let sha_trim = sha.trim();

24 if !(sha_trim.len() == 40 || sha_trim.len() == 64)

25 || !sha_trim.bytes().all(|b| b.is_ascii_hexdigit())

26 {

27 return Err(GitError::InvalidArgument(format!(

28 "invalid commit sha `{sha}`"

29 )));

30 }

2431 let ctx = self.context(space_name, repo_name).await?;

25 let out = Command::new("git")

26 .arg("-C")

27 .arg(ctx.repo_dir())

28 .args(["verify-commit", "--raw", &sha])

29 .output()

30 .map_err(|e| GitError::Internal(format!("verify-commit: {e}")))?;

32 let timeout = self.state.config.git.git_command_timeout_secs;

33 let out = run_git_soft(

34 "signatures::verify_commit",

35 &ctx.repo_dir(),

36 vec![

37 "verify-commit".to_string(),

38 "--raw".to_string(),

39 sha_trim.to_string(),

40 ],

41 &[],

42 timeout,

43 )

44 .await?;

3145 let stdout = String::from_utf8_lossy(&out.stdout);

3246 let mut valid = false;

3347 let mut signer: Option<String> = None;

Modified lib/git/src/share/access.rs +23 −15 View lib/git/src/share/access.rs

@@ -197,29 +197,37 @@

197197 db: &sea_orm::DatabaseConnection,

198198 repo: &models::repo::repos::Model,

199199) -> Result<models::space::spaces::Model, GitError> {

200 let mut dir_id = repo.dir_id;

201 let mut space_id: Option<Uuid> = None;

202 for _ in 0..32 {

200 space_for_dir(db, repo.dir_id)

201 .await?

202 .ok_or_else(|| GitError::Internal(format!("space for repo {} not found", repo.id)))

203}

204

205pub const MAX_DIR_DEPTH: usize = 64;

206

207pub async fn space_for_dir(

208 db: &sea_orm::DatabaseConnection,

209 mut dir_id: Uuid,

210) -> Result<Option<models::space::spaces::Model>, GitError> {

211 for _ in 0..MAX_DIR_DEPTH {

203212 let row = models::dir::dirs::Entity::find_by_id(dir_id)

204213 .one(db)

205214 .await?

206215 .ok_or_else(|| GitError::Internal(format!("dir {dir_id} not found")))?;

207 if row.space_id.is_some() {

208 space_id = row.space_id;

209 break;

216 if let Some(space_id) = row.space_id {

217 let space = models::space::spaces::Entity::find_by_id(space_id)

218 .one(db)

219 .await?

220 .ok_or_else(|| GitError::Internal(format!("space {space_id} not found")))?;

221 return Ok(Some(space));

210222 }

211223 match row.parent_dir_id {

212 Some(p) => dir_id = p,

213 None => break,

224 Some(parent) => dir_id = parent,

225 None => return Ok(None),

214226 }

215227 }

216 let sid = space_id.ok_or_else(|| {

217 GitError::Internal(format!("space for repo {} not found", repo.id))

218 })?;

219 models::space::spaces::Entity::find_by_id(sid)

220 .one(db)

221 .await?

222 .ok_or_else(|| GitError::Internal(format!("space {sid} not found")))

228 Err(GitError::Internal(format!(

229 "dir tree deeper than {MAX_DIR_DEPTH} levels"

230 )))

223231}

224232

225233fn sha256_hex(data: &[u8]) -> String {

Modified lib/git/src/share/errors.rs +8 −1 View lib/git/src/share/errors.rs

@@ -67,7 +67,14 @@

6767 Self::InvalidArgument(_) => StatusCode::BAD_REQUEST,

6868 Self::Cache(_) | Self::Git(_) | Self::Internal(_) => StatusCode::INTERNAL_SERVER_ERROR,

6969 };

70 let mut resp = (status, self.to_string()).into_response();

70 let body = match &self {

71 Self::Cache(_) | Self::Git(_) | Self::Internal(_) => {

72 tracing::warn!(error = %self, "git internal error");

73 "internal server error".to_string()

74 }

75 other => other.to_string(),

76 };

77 let mut resp = (status, body).into_response();

7178 if status == StatusCode::UNAUTHORIZED {

7279 resp.headers_mut().insert(

7380 axum::http::header::WWW_AUTHENTICATE,

Modified lib/git/src/share/validate.rs +60 −11 View lib/git/src/share/validate.rs

@@ -133,31 +133,74 @@

133133 "remote url userinfo with password is not allowed".to_string(),

134134 ));

135135 }

136 let mut host = hostport

136 let host = hostport

137137 .rsplit('@')

138138 .next()

139139 .unwrap_or_default()

140140 .trim()

141 .trim_matches(['[', ']']);

142 if let Some(idx) = host.rfind(':') {

141 .to_string();

142 let host = strip_brackets_and_port(&host);

143 let host = host.to_ascii_lowercase();

144 if host.is_empty() || is_forbidden_host(&host) {

145 return Err(GitError::InvalidArgument(format!(

146 "remote url host `{host}` is not allowed"

147 )));

148 }

149 Ok(())

150}

151

152fn strip_brackets_and_port(host: &str) -> String {

153 if host.starts_with('[')

154 && let Some(end) = host.find(']')

155 {

156 let inner = &host[1..end];

157 let rest = &host[end + 1..];

158 if !inner.is_empty() && (rest.is_empty() || rest.starts_with(':')) {

159 return inner.to_string();

160 }

161 }

162 if host.matches(':').count() == 1

163 && let Some(idx) = host.rfind(':')

164 {

143165 let (h, p) = host.split_at(idx);

144166 let port = &p[1..];

145 if host.matches(':').count() == 1

146 && !port.is_empty()

147 && port.chars().all(|c| c.is_ascii_digit())

148 {

149 host = h;

167 if !port.is_empty() && port.chars().all(|c| c.is_ascii_digit()) {

168 return h.to_string();

150169 }

151170 }

152 let host = host.to_ascii_lowercase();

153 if host.is_empty() || is_forbidden_host(&host) {

171 host.to_string()

172}

173

174pub fn validate_remote_name(name: &str) -> Result<(), GitError> {

175 let n = name.trim();

176 if n.is_empty() || n.len() > 255 || n.starts_with('-') {

177 return Err(GitError::InvalidArgument(format!(

178 "invalid remote name `{n}`"

179 )));

180 }

181 if n.chars().any(|c| c.is_whitespace() || c.is_control()) {

154182 return Err(GitError::InvalidArgument(format!(

155 "remote url host `{host}` is not allowed"

183 "invalid remote name `{n}`"

156184 )));

157185 }

158186 Ok(())

159187}

160188

189pub fn validate_refspec(refspec: &str) -> Result<(), GitError> {

190 let r = refspec.trim();

191 if r.is_empty() || r.len() > 1024 || r.starts_with('-') {

192 return Err(GitError::InvalidArgument(format!(

193 "invalid refspec `{r}`"

194 )));

195 }

196 if r.chars().any(|c| c.is_control()) {

197 return Err(GitError::InvalidArgument(format!(

198 "invalid refspec `{r}`"

199 )));

200 }

201 Ok(())

202}

203

161204fn is_forbidden_host(host: &str) -> bool {

162205 if host == "localhost" || host.ends_with(".localhost") {

163206 return true;

@@ -165,6 +208,11 @@

165208 if host.ends_with(".local") || host.ends_with(".internal") {

166209 return true;

167210 }

211 if let Some(mapped) = host.strip_prefix("::ffff:") {

212 if parse_ipv4(mapped).is_some() {

213 return true;

214 }

215 }

168216 if let Some(segments) = parse_ipv4(host) {

169217 if segments[0] == 127

170218 || segments[0] == 10

@@ -176,6 +224,7 @@

176224 return true;

177225 }

178226 } else if host == "::1"

227 || host == "::"

179228 || host.starts_with("fe80:")

180229 || host.starts_with("fc")

181230 || host.starts_with("fd")

Modified lib/git/src/smart/dispatch.rs +57 −23 View lib/git/src/smart/dispatch.rs

@@ -1,10 +1,12 @@

11use std::path::PathBuf;

2use std::sync::Arc;

23

34use axum::body::Body;

45use axum::http::{HeaderMap, StatusCode, header};

56use axum::response::{IntoResponse, Response};

67use bytes::Bytes;

78use flate2::read::GzDecoder;

9use futures::StreamExt as _;

810use std::io::Read;

911use uuid::Uuid;

1012

@@ -45,7 +47,8 @@

4547 match service_kind {

4648 Some(name) => {

4749 let git_service = GitService::from_name(name)?;

48 info_refs_module::smart_advertisement(&bare, git_service).await

50 let timeout = service.git.state.config.git.git_rpc_timeout_secs;

51 info_refs_module::smart_advertisement(&bare, git_service, timeout).await

4952 }

5053 None => info_refs_module::dumb_info_refs(&bare).await,

5154 }

@@ -89,27 +92,19 @@

8992 tracing::warn!(?e, "post-receive hook setup failed");

9093 }

9194 }

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 }

95 let timeout = service.git.state.config.git.git_rpc_timeout_secs;

96 let stream = rpc::run_rpc_stream(&bare, git_service, false, input, timeout).await?;

97 let post = (

98 is_receive,

99 service.queue.clone(),

100 service.sync_settings.clone(),

101 space_id,

102 ctx.model.id,

103 actor_id,

104 bare.clone(),

105 );

106 let stream = stream.map(|chunk| chunk.map_err(|e| axum::Error::new(e.to_string())));

107 let stream = body_stream_with_post_receive(stream, post);

113108 Ok::<_, SmartError>(

114109 (

115110 StatusCode::OK,

@@ -117,7 +112,7 @@

117112 (header::CONTENT_TYPE, git_service.result_content_type()),

118113 (header::CACHE_CONTROL, "no-cache"),

119114 ],

120 output,

115 Body::from_stream(stream),

121116 )

122117 .into_response(),

123118 )

@@ -148,6 +143,45 @@

148143 }

149144}

150145

146fn body_stream_with_post_receive(

147 stream: impl futures::Stream<Item = Result<Bytes, axum::Error>> + Send + 'static,

148 post: (

149 bool,

150 Option<Arc<crate::sync::queue::SyncQueue>>,

151 config::git::SyncSettings,

152 Uuid,

153 Uuid,

154 Option<Uuid>,

155 std::path::PathBuf,

156 ),

157) -> impl futures::Stream<Item = Result<Bytes, axum::Error>> + Send + 'static {

158 use futures::{StreamExt, stream};

159 stream.chain(stream::once(async move {

160 let (is_receive, queue, settings, space_id, repo_id, actor_id, bare) = post;

161 if is_receive

162 && let Some(queue) = queue

163 {

164 match enqueue_post_receive_push(

165 &queue,

166 &settings,

167 space_id,

168 repo_id,

169 actor_id,

170 &bare,

171 )

172 .await

173 {

174 Ok(n) if n > 0 => {

175 tracing::debug!(count = n, "enqueued post-receive ref updates")

176 }

177 Ok(_) => tracing::debug!("no refs to enqueue"),

178 Err(e) => tracing::warn!(?e, "post-receive enqueue failed"),

179 }

180 }

181 Ok::<_, axum::Error>(Bytes::new())

182 }))

183}

184

151185pub async fn decode_request_body(

152186 headers: &HeaderMap,

153187 body: Body,

Modified lib/git/src/smart/error.rs +5 −2 View lib/git/src/smart/error.rs

@@ -42,8 +42,11 @@

4242 Self::BadRequest(m)

4343 | Self::Forbidden(m)

4444 | Self::NotFound(m)

45 | Self::Conflict(m)

46 | Self::Internal(m) => m.clone(),

45 | Self::Conflict(m) => m.clone(),

46 Self::Internal(m) => {

47 tracing::warn!(error = %m, "smart http internal error");

48 "internal server error".to_string()

49 }

4750 other => other.to_string(),

4851 }

4952 }

Modified lib/git/src/smart/info_refs.rs +3 −2 View lib/git/src/smart/info_refs.rs

@@ -9,9 +9,10 @@

99pub async fn smart_advertisement(

1010 bare: &std::path::Path,

1111 service: GitService,

12 timeout_secs: u64,

1213) -> 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());

14 let output = rpc::run_rpc_bytes(bare, service, true, Bytes::new(), timeout_secs).await?;

15 let mut body = pktline::encode_header(service.dashed_name())?;

1516 body.extend_from_slice(&output);

1617

1718 Ok((

Modified lib/git/src/smart/pktline.rs +10 −6 View lib/git/src/smart/pktline.rs

@@ -2,24 +2,28 @@

22

33const MAX_PKT_SIZE: usize = 65_520;

44

5pub fn encode_line(line: &[u8]) -> Vec<u8> {

5pub fn encode_line(line: &[u8]) -> SmartResult<Vec<u8>> {

66 let len = line.len() + 4;

7 assert!(len <= MAX_PKT_SIZE, "pkt-line too long: {len}");

7 if len > MAX_PKT_SIZE {

8 return Err(SmartError::BadRequest(format!(

9 "pkt-line too long: {len}"

10 )));

11 }

812 let mut out = Vec::with_capacity(4 + line.len());

913 out.extend_from_slice(format!("{len:04x}").as_bytes());

1014 out.extend_from_slice(line);

11 out

15 Ok(out)

1216}

1317

1418pub fn flush_pkt() -> Vec<u8> {

1519 b"0000".to_vec()

1620}

1721

18pub fn encode_header(service: &str) -> Vec<u8> {

22pub fn encode_header(service: &str) -> SmartResult<Vec<u8>> {

1923 let payload = format!("# service={service}\n");

20 let mut out = encode_line(payload.as_bytes());

24 let mut out = encode_line(payload.as_bytes())?;

2125 out.extend_from_slice(&flush_pkt());

22 out

26 Ok(out)

2327}

2428

2529pub fn decode_stream(input: &[u8]) -> SmartResult<Vec<Vec<u8>>> {

Modified lib/git/src/smart/push.rs +7 −12 View lib/git/src/smart/push.rs

@@ -102,6 +102,9 @@

102102

103103pub struct NoopPushBackend;

104104

105const NOPUSH_BACKEND_ERROR: &str =

106 "push requires redis: SmartService kv is not configured (NoopPushBackend)";

107

105108#[async_trait]

106109impl PushOperations for NoopPushBackend {

107110 async fn acquire(

@@ -110,15 +113,11 @@

110113 _member: &[u8],

111114 _ttl_secs: u64,

112115 ) -> SmartResult<bool> {

113 Err(SmartError::Internal(

114 "push coordination disabled: redis not configured".into(),

115 ))

116 Err(SmartError::Internal(NOPUSH_BACKEND_ERROR.to_string()))

116117 }

117118

118119 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 ))

120 Err(SmartError::Internal(NOPUSH_BACKEND_ERROR.to_string()))

122121 }

123122

124123 async fn release(&self, _key: &str, _token: &[u8]) -> SmartResult<()> {

@@ -126,15 +125,11 @@

126125 }

127126

128127 async fn enqueue(&self, _key: &str, _member: &[u8]) -> SmartResult<()> {

129 Err(SmartError::Internal(

130 "push coordination disabled: redis not configured".into(),

131 ))

128 Err(SmartError::Internal(NOPUSH_BACKEND_ERROR.to_string()))

132129 }

133130

134131 async fn dequeue(&self, _key: &str, _member: &[u8]) -> SmartResult<()> {

135 Err(SmartError::Internal(

136 "push coordination disabled: redis not configured".into(),

137 ))

132 Err(SmartError::Internal(NOPUSH_BACKEND_ERROR.to_string()))

138133 }

139134}

140135

Modified lib/git/src/smart/rpc.rs +125 −65 View lib/git/src/smart/rpc.rs

@@ -1,17 +1,21 @@

11use std::path::Path;

22use std::process::Stdio;

33use std::sync::Arc;

4use std::time::Duration;

45

56use bytes::Bytes;

7use futures::StreamExt;

68use tokio::io::{AsyncReadExt, AsyncWriteExt};

79use tokio::process::Command;

8use tokio::sync::Mutex;

9use tokio::task::JoinHandle;

10use tokio::sync::{Mutex, oneshot};

11use tokio_util::io::ReaderStream;

1012

1113use crate::smart::error::{SmartError, SmartResult};

1214

1315pub type RpcStream = futures::stream::BoxStream<'static, Result<Bytes, std::io::Error>>;

1416

17const MAX_STDERR_BYTES: usize = 64 * 1024 * 1024;

18

1519#[derive(Debug, Clone, Copy, PartialEq, Eq)]

1620pub enum GitService {

1721 UploadPack,

@@ -58,12 +62,7 @@

5862 }

5963}

6064

61pub async fn run_rpc_bytes(

62 bare_path: &Path,

63 service: GitService,

64 advertise: bool,

65 input: Bytes,

66) -> SmartResult<Bytes> {

65fn build_command(bare_path: &Path, service: GitService, advertise: bool) -> Command {

6766 let mut cmd = Command::new("git");

6867 cmd.arg(service.name())

6968 .arg("--stateless-rpc");

@@ -79,74 +78,135 @@

7978 .stdin(Stdio::piped())

8079 .stdout(Stdio::piped())

8180 .stderr(Stdio::piped());

81 cmd

82}

8283

83 let service_name = service.name();

84#[allow(clippy::type_complexity)]

85fn spawn_with_supervisor(

86 bare_path: &Path,

87 service: GitService,

88 advertise: bool,

89 input: Bytes,

90 timeout_secs: u64,

91) -> SmartResult<(

92 RpcStream,

93 Option<tokio::sync::oneshot::Receiver<SmartResult<()>>>,

94)> {

95 let mut cmd = build_command(bare_path, service, advertise);

8496 let mut child = cmd

8597 .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 });

98 .map_err(|e| SmartError::Internal(format!("failed to spawn git {}: {e}", service.name())))?;

99 let mut stdin = child

100 .stdin

101 .take()

102 .ok_or_else(|| SmartError::Internal("stdin not piped".to_string()))?;

103 let stdout = child

104 .stdout

105 .take()

106 .ok_or_else(|| SmartError::Internal("stdout not piped".to_string()))?;

107 let mut stderr = child

108 .stderr

109 .take()

110 .ok_or_else(|| SmartError::Internal("stderr not piped".to_string()))?;

99111

100 let child_guard = Arc::new(Mutex::new(Some(child)));

101 let stderr_buf = Arc::new(Mutex::new(Vec::<u8>::new()));

112 let child_arc = Arc::new(Mutex::new(Some(child)));

102113

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]),

114 {

115 let child_arc = child_arc.clone();

116 tokio::spawn(async move {

117 if stdin.write_all(&input).await.is_err() {

118 if let Some(mut ch) = child_arc.lock().await.take() {

119 let _ = ch.start_kill();

120 let _ = ch.wait().await;

121 }

110122 }

111 }

112 });

123 let _ = stdin.shutdown().await;

124 });

125 }

113126

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 let (tx, rx) = oneshot::channel();

128 {

129 let child_arc = child_arc.clone();

130 tokio::spawn(async move {

131 let status = {

132 let mut guard = child_arc.lock().await;

133 match guard.as_mut() {

134 None => None,

135 Some(ch) => {

136 match tokio::time::timeout(Duration::from_secs(timeout_secs), ch.wait())

137 .await

138 {

139 Ok(Ok(s)) => Some(s),

140 _ => {

141 let _ = ch.start_kill();

142 let _ = ch.wait().await;

143 None

144 }

145 }

146 }

147 }

148 };

149 let mut buf = vec![0u8; 8192];

150 let mut total = 0usize;

151 let mut text = Vec::new();

152 loop {

153 match stderr.read(&mut buf).await {

154 Ok(0) | Err(_) => break,

155 Ok(n) => {

156 total += n;

157 if total <= MAX_STDERR_BYTES {

158 text.extend_from_slice(&buf[..n]);

159 }

160 }

161 }

127162 }

128 }

163 let verdict = match status {

164 Some(s) if s.success() => Ok(()),

165 Some(s) => Err(SmartError::Internal(format!(

166 "git {} exited with {s}: {}",

167 service.name(),

168 String::from_utf8_lossy(text.as_slice()).trim()

169 ))),

170 None => Err(SmartError::Internal(format!(

171 "git {} timed out after {timeout_secs}s",

172 service.name()

173 ))),

174 };

175 let _ = tx.send(verdict);

176 });

129177 }

130178

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 })?;

179 let stream: RpcStream = Box::pin(ReaderStream::new(stdout));

180 Ok((stream, Some(rx)))

181}

141182

142 let _ = stdin_task.await;

183pub async fn run_rpc_stream(

184 bare_path: &Path,

185 service: GitService,

186 advertise: bool,

187 input: Bytes,

188 timeout_secs: u64,

189) -> SmartResult<RpcStream> {

190 let (stream, _rx) = spawn_with_supervisor(bare_path, service, advertise, input, timeout_secs)?;

191 Ok(stream)

192}

143193

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 )))

194pub async fn run_rpc_bytes(

195 bare_path: &Path,

196 service: GitService,

197 advertise: bool,

198 input: Bytes,

199 timeout_secs: u64,

200) -> SmartResult<Bytes> {

201 let (mut stream, rx) = spawn_with_supervisor(bare_path, service, advertise, input, timeout_secs)?;

202 let mut out: Vec<u8> = Vec::new();

203 while let Some(chunk) = stream.next().await {

204 out.extend_from_slice(&chunk.map_err(|e| {

205 SmartError::Internal(format!("git {} stdout read failed: {e}", service.name()))

206 })?);

151207 }

208 if let Some(rx) = rx {

209 let _ = rx.await;

210 }

211 Ok(Bytes::from(out))

152212}

No newline at end of file

Modified lib/git/src/smart/service.rs +4 −18 View lib/git/src/smart/service.rs

@@ -79,24 +79,10 @@

7979 }

8080

8181 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 }

82 let space = crate::share::access::space_for_dir(self.git.db_read(), ctx.model.dir_id)

83 .await?

84 .ok_or_else(|| SmartError::NotFound("space".into()))?;

85 Ok(space.lower_name)

10086 }

10187}

10288

Modified lib/git/src/ssh/handler/channel.rs +85 −26 View lib/git/src/ssh/handler/channel.rs

@@ -75,6 +75,7 @@

7575

7676 let handle = session.handle();

7777 let db = self.service.smart.git.state.database.clone();

78 let rpc_timeout = self.service.smart.git.state.config.git.git_rpc_timeout_secs;

7879 tokio::spawn(async move {

7980 if overflowed {

8081 _ = handle.exit_status_request(channel, 1);

@@ -187,7 +188,7 @@

187188 let push_key = format!("git_op:{}", bare.display());

188189

189190 let run = async {

190 run_git(op.as_ref(), &buffer, &handle, channel).await

191 run_git(op.as_ref(), &buffer, &handle, channel, rpc_timeout).await

191192 };

192193 let result = push_coordinator

193194 .run(&push_key, async {

@@ -221,7 +222,7 @@

221222 _ = handle.eof(channel);

222223 _ = handle.close(channel);

223224 } else {

224 let result = run_git(op.as_ref(), &buffer, &handle, channel).await;

225 let result = run_git(op.as_ref(), &buffer, &handle, channel, rpc_timeout).await;

225226 let code = if result.is_ok() { 0 } else { 1 };

226227 _ = handle.exit_status_request(channel, code);

227228 tokio::time::sleep(std::time::Duration::from_millis(20)).await;

@@ -239,6 +240,7 @@

239240 input: &[u8],

240241 handle: &russh::server::Handle,

241242 channel: russh::ChannelId,

243 timeout_secs: u64,

242244) -> Result<(), SshError> {

243245 let Some(op) = op else {

244246 return Ok(());

@@ -257,41 +259,98 @@

257259 apply_hardened_env(&mut cmd);

258260

259261 let mut child = cmd.spawn().map_err(SshError::Io)?;

260 if let Some(mut stdin) = child.stdin.take() {

261 stdin

262 .write_all(input)

263 .await

264 .map_err(SshError::Io)?;

265 let _ = stdin.shutdown().await;

262 let mut stdin = child

263 .stdin

264 .take()

265 .ok_or_else(|| SshError::Internal("git stdin missing".to_string()))?;

266 let stdout = child

267 .stdout

268 .take()

269 .ok_or_else(|| SshError::Internal("git stdout missing".to_string()))?;

270 let stderr = child

271 .stderr

272 .take()

273 .ok_or_else(|| SshError::Internal("git stderr missing".to_string()))?;

274 let child_arc = Arc::new(tokio::sync::Mutex::new(Some(child)));

275

276 {

277 let child_arc = child_arc.clone();

278 let input = input.to_vec();

279 tokio::spawn(async move {

280 if stdin.write_all(&input).await.is_err() {

281 if let Some(mut ch) = child_arc.lock().await.take() {

282 let _ = ch.start_kill();

283 let _ = ch.wait().await;

284 }

285 }

286 let _ = stdin.shutdown().await;

287 });

266288 }

267 let stdout_cap = MAX_GIT_OUTPUT_BUF as u64 + 1;

268 let stdout = if let Some(out) = child.stdout.take() {

289

290 let stdout_task = tokio::spawn(async move {

291 let cap = MAX_GIT_OUTPUT_BUF as u64 + 1;

269292 let mut buf = Vec::new();

270 out.take(stdout_cap).read_to_end(&mut buf).await.map_err(SshError::Io)?;

271 buf

272 } else {

273 Vec::new()

274 };

275 let stderr = if let Some(err) = child.stderr.take() {

293 let res = stdout.take(cap).read_to_end(&mut buf).await;

294 (buf, res)

295 });

296 let stderr_task = tokio::spawn(async move {

297 let cap = MAX_GIT_OUTPUT_BUF as u64 + 1;

276298 let mut buf = Vec::new();

277 err.take(stdout_cap).read_to_end(&mut buf).await.map_err(SshError::Io)?;

278 buf

279 } else {

280 Vec::new()

299 let res = stderr.take(cap).read_to_end(&mut buf).await;

300 (buf, res)

301 });

302

303 let wait_task = {

304 let child_arc = child_arc.clone();

305 async move {

306 let mut guard = child_arc.lock().await;

307 match guard.as_mut() {

308 None => None,

309 Some(ch) => {

310 match tokio::time::timeout(

311 std::time::Duration::from_secs(timeout_secs),

312 ch.wait(),

313 )

314 .await

315 {

316 Ok(Ok(s)) => Some(s),

317 _ => {

318 let _ = ch.start_kill();

319 let _ = ch.wait().await;

320 None

321 }

322 }

323 }

324 }

325 }

281326 };

282 let status = child.wait().await.map_err(SshError::Io)?;

283 if stdout.len() > MAX_GIT_OUTPUT_BUF || stderr.len() > MAX_GIT_OUTPUT_BUF {

327

328 let (wait_out, stdout_res, stderr_res) =

329 tokio::join!(wait_task, stdout_task, stderr_task);

330

331 let (stdout_buf, stdout_read) = stdout_res

332 .map_err(|e| SshError::Internal(format!("stdout task: {e}")))?;

333 let (stderr_buf, _stderr_read) = stderr_res

334 .map_err(|e| SshError::Internal(format!("stderr task: {e}")))?;

335 let status = wait_out.ok_or_else(|| {

336 SshError::Internal(format!("git {} timed out", op.service_arg()))

337 })?;

338

339 if stdout_buf.len() > MAX_GIT_OUTPUT_BUF || stderr_buf.len() > MAX_GIT_OUTPUT_BUF {

284340 return Err(SshError::Internal(format!(

285341 "git {} output exceeded {} bytes",

286342 op.service_arg(),

287343 MAX_GIT_OUTPUT_BUF

288344 )));

289345 }

290 if !stdout.is_empty() {

291 _ = handle.data(channel, Bytes::from(stdout)).await;

346 stdout_read.map_err(SshError::Io)?;

347 if !stdout_buf.is_empty() {

348 let _ = handle.data(channel, Bytes::from(stdout_buf)).await;

292349 }

293 if !stderr.is_empty() && !status.success() {

294 _ = handle.extended_data(channel, 1, Bytes::from(stderr)).await;

350 if !stderr_buf.is_empty() && !status.success() {

351 let _ = handle

352 .extended_data(channel, 1, Bytes::from(stderr_buf))

353 .await;

295354 }

296355 if status.success() {

297356 Ok(())

Modified lib/git/src/ssh/handler/command.rs +52 −28 View lib/git/src/ssh/handler/command.rs

@@ -1,7 +1,7 @@

11use bytes::Bytes;

22use russh::ChannelId;

33use russh::server::Session;

4use sea_orm::EntityTrait;

4

55use uuid::Uuid;

66

77use crate::share::access::authorize_repo_access;

@@ -87,7 +87,19 @@

8787 return Ok(());

8888 }

8989

90 let adv = match run_advertise(&bare, service).await {

90 let adv = match run_advertise(

91 &bare,

92 service,

93 self.service

94 .smart

95 .git

96 .state

97 .config

98 .git

99 .git_rpc_timeout_secs,

100 )

101 .await

102 {

91103 Ok(adv) => adv,

92104 Err(e) => {

93105 let msg = format!("fatal: unable to read repository advertisement: {e}");

@@ -138,22 +150,11 @@

138150 }

139151

140152 async fn space_id_for(&self, dir_id: &Uuid) -> Result<Uuid, SmartError> {

141 let svc = &self.service.smart.git;

142 let mut current = *dir_id;

143 loop {

144 let dir = models::dir::dirs::Entity::find_by_id(current)

145 .one(svc.db_read())

146 .await

147 .map_err(SmartError::Db)?;

148 let dir = dir.ok_or_else(|| SmartError::NotFound(format!("dir {current}")))?;

149 if let Some(space_id) = dir.space_id {

150 return Ok(space_id);

151 }

152 match dir.parent_dir_id {

153 Some(parent) => current = parent,

154 None => return Err(SmartError::NotFound("space".into())),

155 }

156 }

153 crate::share::access::space_for_dir(self.service.smart.git.db_read(), *dir_id)

154 .await

155 .map_err(SmartError::from)?

156 .map(|s| s.id)

157 .ok_or_else(|| SmartError::NotFound("space".into()))

157158 }

158159

159160 pub async fn shell_request_impl(

@@ -184,30 +185,53 @@

184185async fn run_advertise(

185186 bare: &std::path::Path,

186187 service: GitService,

188 timeout_secs: u64,

187189) -> Result<Vec<u8>, SshError> {

188190 let mut cmd = build_git_advertise_command(service, bare);

189191 cmd.stdout(std::process::Stdio::piped())

190192 .stderr(std::process::Stdio::piped());

191193 let mut child = cmd.spawn().map_err(SshError::Io)?;

192 let stderr_task = child.stderr.take().map(|stderr| {

193 tokio::spawn(async move {

194 let stderr_task = match child.stderr.take() {

195 Some(stderr) => Some(tokio::spawn(async move {

194196 use tokio::io::AsyncReadExt;

195 let mut s = stderr;

197 let s = stderr;

196198 let mut buf = Vec::new();

197 let _ = s.read_to_end(&mut buf).await;

199 let _ = s.take((64 * 1024 * 1024) as u64 + 1)

200 .read_to_end(&mut buf)

201 .await;

198202 buf

199 })

200 });

201 let mut stdout = child

203 })),

204 None => None,

205 };

206 let stdout = child

202207 .stdout

203208 .take()

204209 .ok_or_else(|| SshError::Internal("git stdout missing".to_string()))?;

205210 let mut out = Vec::new();

206211 use tokio::io::AsyncReadExt;

207 stdout.read_to_end(&mut out).await.map_err(SshError::Io)?;

208 let status = child.wait().await.map_err(SshError::Io)?;

212 stdout

213 .take((64 * 1024 * 1024) as u64 + 1)

214 .read_to_end(&mut out)

215 .await

216 .map_err(SshError::Io)?;

217 let status =

218 match tokio::time::timeout(std::time::Duration::from_secs(timeout_secs), child.wait())

219 .await

220 {

221 Ok(Ok(s)) => s,

222 _ => {

223 let _ = child.start_kill();

224 let _ = child.wait().await;

225 return Err(SshError::Internal(format!(

226 "git {} --advertise-refs timed out",

227 service.as_subcommand()

228 )));

229 }

230 };

209231 let stderr_bytes = match stderr_task {

210 Some(t) => t.await.unwrap_or_default(),

232 Some(t) => t.await.map_err(|e| {

233 SshError::Internal(format!("stderr task join failed: {e}"))

234 })?,

211235 None => Vec::new(),

212236 };

213237 if !status.success() {

Modified lib/git/src/sync/fsck.rs +36 −44 View lib/git/src/sync/fsck.rs

@@ -1,58 +1,50 @@

11use std::path::Path;

2use std::process::Command;

32

4use crate::sync::error::{SyncError, SyncResult};

3use crate::service::gitcmd::{run_git, run_git_soft};

54

6pub async fn run_gc(repo_dir: &Path) {

5pub async fn run_gc(repo_dir: &Path, timeout_secs: u64) {

76 let dir_str = repo_dir.display().to_string();

8 let dir = repo_dir.to_path_buf();

9 let result = tokio::task::spawn_blocking(move || {

10 let _ = run_git(&dir, &["gc", "--prune=now", "--quiet"]);

11 })

7 match run_git(

8 "sync::gc",

9 repo_dir,

10 vec![

11 "gc".to_string(),

12 "--prune=now".to_string(),

13 "--quiet".to_string(),

14 ],

15 &[],

16 timeout_secs,

17 )

1218 .await

13 .map_err(|e| SyncError::Internal(format!("join error: {e}")));

14

15 if let Err(e) = result {

16 tracing::warn!(repo_dir = %dir_str, error = %e, "git gc failed");

19 {

20 Ok(_) => {}

21 Err(e) => tracing::warn!(repo_dir = %dir_str, error = %e, "git gc failed"),

1722 }

1823}

1924

20pub async fn run_fsck(repo_dir: &Path) {

25pub async fn run_fsck(repo_dir: &Path, timeout_secs: u64) {

2126 let dir_str = repo_dir.display().to_string();

22 let dir = repo_dir.to_path_buf();

23 let result = tokio::task::spawn_blocking(move || run_git(&dir, &["fsck", "--no-progress"]))

24 .await

25 .map_err(|e| SyncError::Internal(format!("join error: {e}")));

26

27 match result {

28 Ok(Ok(stdout)) if stdout.contains("error") || stdout.contains("broken") => {

29 tracing::warn!(

30 repo_dir = %dir_str,

31 output = %stdout.trim(),

32 "fsck reported integrity issues"

33 );

27 match run_git_soft(

28 "sync::fsck",

29 repo_dir,

30 vec!["fsck".to_string(), "--no-progress".to_string()],

31 &[],

32 timeout_secs,

33 )

34 .await

35 {

36 Ok(out) => {

37 let stdout = String::from_utf8_lossy(&out.stdout);

38 if stdout.contains("error") || stdout.contains("broken") {

39 tracing::warn!(

40 repo_dir = %dir_str,

41 output = %stdout.trim(),

42 "fsck reported integrity issues"

43 );

44 }

3445 }

35 Ok(_) => {}

3646 Err(e) => {

3747 tracing::warn!(repo_dir = %dir_str, error = %e, "fsck failed");

3848 }

3949 }

40}

41

42fn run_git(repo_dir: &Path, args: &[&str]) -> SyncResult<String> {

43 let output = Command::new("git")

44 .arg("-C")

45 .arg(repo_dir)

46 .args(args)

47 .output()

48 .map_err(|e| SyncError::Internal(format!("spawn git: {e}")))?;

49 if !output.status.success() {

50 return Err(SyncError::Internal(format!(

51 "git {} exited {}: {}",

52 args.join(" "),

53 output.status.code().unwrap_or(-1),

54 String::from_utf8_lossy(&output.stderr).trim()

55 )));

56 }

57 Ok(String::from_utf8_lossy(&output.stdout).into_owned())

58}

50}

No newline at end of file

Modified lib/git/src/sync/post_receive.rs +3 −1 View lib/git/src/sync/post_receive.rs

@@ -33,7 +33,9 @@

3333 );

3434 return Ok(());

3535 }

36 let hooks_dir = path.parent().expect("hook path always has a parent");

36 let Some(hooks_dir) = path.parent() else {

37 return Err(io::Error::other("hook path has no parent"));

38 };

3739 fs::create_dir_all(hooks_dir)?;

3840 let script = format!(

3941 "#!/bin/sh\n# gitdata post-receive capture hook\nwhile read old_oid new_oid refname; do\n printf '%s %s %s\\n' \"$old_oid\" \"$new_oid\" \"$refname\" >> \"$GIT_DIR/{CAPTURE_FILE}\"\ndone\n"

Modified lib/git/src/sync/stats.rs +2 −1 View lib/git/src/sync/stats.rs

@@ -11,11 +11,12 @@

1111 new_commits: usize,

1212 branch_count: usize,

1313 tag_count: usize,

14 command_timeout_secs: u64,

1415) -> SyncResult<()> {

1516 let git_size = dir_size(repo_dir)

1617 .await

1718 .map_err(|e| SyncError::Internal(format!("dir_size: {e}")))? as i64;

18 let total_commits = count_commits(repo_dir)

19 let total_commits = count_commits(repo_dir, command_timeout_secs)

1920 .await

2021 .map_err(|e| SyncError::Internal(format!("count_commits: {e}")))?

2122 as i32;

Modified lib/git/src/sync/sync.rs +4 −2 View lib/git/src/sync/sync.rs

@@ -143,13 +143,15 @@

143143 new_commits.len(),

144144 branch_count,

145145 tag_count,

146 self.settings.git_command_timeout_secs,

146147 )

147148 .await?;

148149

149150 txn.commit().await.map_err(SyncError::Db)?;

150151

151 fsck::run_gc(&self.repo_dir()).await;

152 fsck::run_fsck(&self.repo_dir()).await;

152 let gc_timeout = self.settings.git_command_timeout_secs.max(1800);

153 fsck::run_gc(&self.repo_dir(), gc_timeout).await;

154 fsck::run_fsck(&self.repo_dir(), gc_timeout).await;

153155

154156 Ok(())

155157 }

Modified lib/git/src/errors.rs +1 −4 View lib/git/src/errors.rs

@@ -132,10 +132,7 @@

132132impl IntoResponse for ApiError {

133133 fn into_response(self) -> Response {

134134 let (code, envelope) = self.decompose();

135 let status = match StatusCode::from_u16(code) {

136 Ok(status) => status,

137 Err(_) => StatusCode::INTERNAL_SERVER_ERROR,

138 };

135 let status = StatusCode::from_u16(code).unwrap_or_else(|_| StatusCode::INTERNAL_SERVER_ERROR);

139136 (status, Json(envelope)).into_response()

140137 }

141138}