Skip to content
zhenyi

gitdata-core

0
Clone

HTTPS

SSH

SSH needs a key in your settings.

Download ZIP

refactor(dvcs): refresh dvcs/git, dvcs/gitsmart, dvcs/gitssh, dvcs/gitsync, dvcs/gitauth

zhenyi committed
parent 9a98508 commit d977f67 Browse files

36 files changed

+711 −261

File tree 36
Modified dvcs/git/archive/usecase.rs View dvcs/git/archive/usecase.rs
@@ -47,8 +47,8 @@
4747 let storage = self.storage.clone();
4848 let cache_key = format!("{}:{}:{}", repo, rev, ext);
4949
50 if let Some(hit) = self.cache.object_pool_get_sync::<ArchiveResult>(&cache_key) {
51 tracing::info!(rev, format = ?format, size = hit.size, "archive object-pool hit");
50 if let Some(hit) = self.cache.get_sync::<ArchiveResult>(&cache_key) {
51 tracing::info!(rev, format = ?format, size = hit.size, "archive cache hit");
5252 return Ok(hit);
5353 }
5454
Modified dvcs/git/commit/types.rs View dvcs/git/commit/types.rs
@@ -36,7 +36,16 @@
3636 Signature {
3737 name: sig.name.to_string(),
3838 email: sig.email.to_string(),
39 when: std::time::UNIX_EPOCH + std::time::Duration::from_secs(sig.time.seconds as u64),
39 when: {
40 let seconds = sig.time.seconds;
41 if seconds >= 0 {
42 std::time::UNIX_EPOCH + std::time::Duration::from_secs(seconds as u64)
43 } else {
44 std::time::UNIX_EPOCH
45 .checked_sub(std::time::Duration::from_secs(seconds.unsigned_abs()))
46 .unwrap_or(std::time::UNIX_EPOCH)
47 }
48 },
4049 offset_seconds: sig.time.offset,
4150 }
4251 }
Modified dvcs/git/conflict/usecase.rs View dvcs/git/conflict/usecase.rs
@@ -5,6 +5,7 @@
55use crate::GitRepository;
66use crate::conflict::impls::ConflictOps;
77use crate::conflict::types::{ConflictDetail, ConflictFile, MergeStatus};
8use crate::diff::types::{DiffFile, DiffFileType};
89
910fn resolve_tree_id(repo: &gix::Repository, rev: &str) -> Result<gix::hash::ObjectId, crate::Error> {
1011 let id = repo
@@ -93,6 +94,48 @@
9394 out
9495}
9596
97fn threeway_conflicts(
98 repo: &gix::Repository,
99 ancestor_tree_id: gix::hash::ObjectId,
100 ours_tree_id: gix::hash::ObjectId,
101 theirs_tree_id: gix::hash::ObjectId,
102) -> Result<Vec<ConflictFile>, crate::Error> {
103 let ancestor_tree = repo
104 .find_tree(ancestor_tree_id)
105 .map_err(|e| crate::Error::Find(e.to_string()))?;
106 let ours_tree = repo
107 .find_tree(ours_tree_id)
108 .map_err(|e| crate::Error::Find(e.to_string()))?;
109 let theirs_tree = repo
110 .find_tree(theirs_tree_id)
111 .map_err(|e| crate::Error::Find(e.to_string()))?;
112
113 let ancestor_to_ours = repo
114 .diff_tree_to_tree(Some(&ancestor_tree), Some(&ours_tree), None)
115 .map_err(|e| crate::Error::Find(e.to_string()))?;
116 let ancestor_to_theirs = repo
117 .diff_tree_to_tree(Some(&ancestor_tree), Some(&theirs_tree), None)
118 .map_err(|e| crate::Error::Find(e.to_string()))?;
119
120 let ours_map = changes_to_oid_map(&ancestor_to_ours);
121 let theirs_map = changes_to_oid_map(&ancestor_to_theirs);
122
123 let mut conflicts: Vec<ConflictFile> = Vec::new();
124 for (path, their_oid) in &theirs_map {
125 if let Some(our_oid) = ours_map.get(path)
126 && our_oid != their_oid
127 {
128 conflicts.push(ConflictFile {
129 path: path.clone(),
130 our_oid: *our_oid,
131 their_oid: *their_oid,
132 base_oid: None,
133 });
134 }
135 }
136 Ok(conflicts)
137}
138
96139impl ConflictOps for GitRepository {
97140 fn conflict_check(&self, base: &str, head: &str) -> Result<MergeStatus, crate::Error> {
98141 let tip = self.tip_id()?;
@@ -117,11 +160,9 @@
117160 .map_err(|e| crate::Error::Find(e.to_string()))?
118161 .detach();
119162 let base_tree_id = resolve_tree_id(&repo, base)?;
120 let fast_forwardable = repo
121 .merge_base(b, h)
122 .map(|mb| mb.detach() == b)
123 .unwrap_or(false);
124 let conflicts = if base_tree_id == head_tree_id {
163 let mb = repo.merge_base(b, h).ok();
164 let fast_forwardable = mb.as_ref().is_some_and(|m| m.detach() == b);
165 let conflicts = if fast_forwardable || base_tree_id == head_tree_id {
125166 Vec::new()
126167 } else {
127168 let base_tree = repo
@@ -130,10 +171,24 @@
130171 let head_tree = repo
131172 .find_tree(head_tree_id)
132173 .map_err(|e| crate::Error::Find(e.to_string()))?;
133 let changes = repo
134 .diff_tree_to_tree(Some(&base_tree), Some(&head_tree), None)
135 .map_err(|e| crate::Error::Find(e.to_string()))?;
136 changes_to_conflicts(&changes)
174 match mb {
175 Some(ancestor_id) => {
176 let ancestor_commit = repo
177 .find_commit(ancestor_id.detach())
178 .map_err(|e| crate::Error::Find(e.to_string()))?;
179 let ancestor_tree_id = ancestor_commit
180 .tree_id()
181 .map_err(|e| crate::Error::Find(e.to_string()))?
182 .detach();
183 threeway_conflicts(&repo, ancestor_tree_id, base_tree_id, head_tree_id)?
184 }
185 None => {
186 let changes = repo
187 .diff_tree_to_tree(Some(&base_tree), Some(&head_tree), None)
188 .map_err(|e| crate::Error::Find(e.to_string()))?;
189 changes_to_conflicts(&changes)
190 }
191 }
137192 };
138193 let v = MergeStatus {
139194 mergeable: conflicts.is_empty(),
@@ -245,10 +300,44 @@
245300 return Ok(hit);
246301 }
247302 let status = self.conflict_check(base, head)?;
248 let _conflict = status.conflicts.into_iter().find(|c| c.path == path);
303 let mut diff = Vec::new();
304 if let Some(c) = status.conflicts.into_iter().find(|c| c.path == path) {
305 let (file_type, is_created, is_deleted) = match (c.our_oid, c.their_oid) {
306 (Some(_), Some(_)) => (DiffFileType::Change, false, false),
307 (None, Some(_)) => (DiffFileType::Add, true, false),
308 (Some(_), None) => (DiffFileType::Del, false, true),
309 (None, None) => (DiffFileType::Change, false, false),
310 };
311 diff.push(DiffFile {
312 name: path.to_string(),
313 old_name: String::new(),
314 addition: 1,
315 deletion: 1,
316 r#type: file_type,
317 entry_mode: String::new(),
318 old_entry_mode: String::new(),
319 is_created,
320 is_deleted,
321 is_bin: false,
322 is_lfs_file: false,
323 is_renamed: false,
324 is_submodule: false,
325 is_generated: false,
326 is_vendored: false,
327 is_incomplete: false,
328 is_incomplete_line_too_long: false,
329 sections: Vec::new(),
330 left_blob_size: 0,
331 right_blob_size: 0,
332 left_blob_mime_type: String::new(),
333 right_blob_mime_type: String::new(),
334 is_blob_type_image: false,
335 is_blob_type_csv: false,
336 });
337 }
249338 let v = ConflictDetail {
250339 path: path.to_string(),
251 diff: Vec::new(),
340 diff,
252341 };
253342 self.cache_put_rev("conflict", "detail", &key, tip, &v);
254343 Ok(v)
Modified dvcs/git/diff/usecase.rs View dvcs/git/diff/usecase.rs
@@ -144,7 +144,7 @@
144144 .map(|r| {
145145 let (tp, name, old_name) = match r.status.as_str() {
146146 "added" => (DiffFileType::Add, r.head_path.clone(), String::new()),
147 "deleted" => (DiffFileType::Del, r.head_path.clone(), r.base_path.clone()),
147 "deleted" => (DiffFileType::Del, r.base_path.clone(), String::new()),
148148 "renamed" => (
149149 DiffFileType::Rename,
150150 r.head_path.clone(),
@@ -205,10 +205,30 @@
205205 let base_tree_id = resolve_tree_id(&repo, base)?;
206206 let head_tree_id = resolve_tree_id(&repo, head)?;
207207 let records = collect_changes(&repo, base_tree_id, head_tree_id)?;
208 let out = crate::gitcli::run(
209 &repo.path(),
210 &[
211 "diff",
212 "--numstat",
213 "--no-renames",
214 &base_tree_id.to_string(),
215 &head_tree_id.to_string(),
216 ],
217 )?;
218 let mut total_addition: i32 = 0;
219 let mut total_deletion: i32 = 0;
220 for line in out.lines() {
221 let mut cols = line.splitn(3, '\t');
222 let (Some(a), Some(d)) = (cols.next(), cols.next()) else {
223 continue;
224 };
225 total_addition += a.parse::<i32>().unwrap_or(0);
226 total_deletion += d.parse::<i32>().unwrap_or(0);
227 }
208228 let v = DiffShortStat {
209229 num_files: records.len() as i32,
210 total_addition: 0,
211 total_deletion: 0,
230 total_addition,
231 total_deletion,
212232 };
213233 self.cache_put_rev("diff", "short", &key, tip, &v);
214234 Ok(v)
@@ -228,7 +248,10 @@
228248 let base_tree_id = resolve_tree_id(&repo, base)?;
229249 let head_tree_id = resolve_tree_id(&repo, head)?;
230250 let records = collect_changes(&repo, base_tree_id, head_tree_id)?;
231 let v: Vec<String> = records.into_iter().map(|r| r.head_path).collect();
251 let v: Vec<String> = records
252 .into_iter()
253 .map(|r| if r.head_path.is_empty() { r.base_path } else { r.head_path })
254 .collect();
232255 self.cache_put_rev("diff", "files", &key, tip, &v);
233256 Ok(v)
234257 }
Modified dvcs/git/gc/usecase.rs View dvcs/git/gc/usecase.rs
@@ -56,7 +56,10 @@
5656 Ok(obj) => {
5757 let kind = obj.kind;
5858 let data = obj.data.clone();
59 let computed = gix::objs::Write::write_buf(&repo, kind, &data)
59 // Pure in-memory checksum verification: computing the
60 // object id must not write anything back to the odb
61 // (Write::write_buf would hydrate loose objects).
62 let computed = gix::objs::compute_hash(repo.object_hash(), kind, &data)
6063 .map_err(|e| crate::Error::Odb(e.to_string()))?;
6164 if computed != id {
6265 errors.push(format!("object {id} failed checksum verification"));
Modified dvcs/git/graphql/query.rs View dvcs/git/graphql/query.rs
@@ -154,7 +154,8 @@
154154 let opts = opts.unwrap_or_default();
155155 let first = clamp_first(opts.first);
156156 if opts.rev.is_none() || opts.rev.as_deref() == Some("HEAD") {
157 let all = crate::commit::impls::CommitOps::commit_list(&self.repo, first as u64, 0)?;
157 let skip = parse_cursor(opts.after.clone())? as u64;
158 let all = crate::commit::impls::CommitOps::commit_list(&self.repo, first as u64, skip)?;
158159 return Ok(all.commits.iter().map(CommitObj::from_commit).collect());
159160 }
160161 let log_opts = crate::log::types::LogOptions {
Modified dvcs/git/graphql/types.rs View dvcs/git/graphql/types.rs
@@ -602,14 +602,9 @@
602602 .and_then(|raw| std::str::from_utf8(raw).ok().and_then(parse_signature_str));
603603 drop(tag);
604604 drop(gix_repo);
605 let target_type = crate::object::impls::ObjectOps::object_type_of(repo, &peeled)
606 .map(GitObjectType::from)
607 .unwrap_or(GitObjectType::Commit);
608 let tag_type = if target_type == GitObjectType::Commit {
609 tag_t::TagType::Lightweight
610 } else {
611 tag_t::TagType::Annotated
612 };
605 // 能 decode 出独立 Tag 对象（含 message/tagger）即为 annotated tag；
606 // lightweight tag 只是指向 commit 的引用，不会进入本分支
607 let tag_type = tag_t::TagType::Annotated;
613608 let tag = tag_t::Tag {
614609 name,
615610 id: oid,
Modified dvcs/git/log/usecase.rs View dvcs/git/log/usecase.rs
@@ -31,7 +31,7 @@
3131 .map_err(|e| crate::Error::RevParse(e.to_string()))?;
3232 let mut commits = Vec::new();
3333 let mut skipped = 0u64;
34 let mut last_id = None;
34 let mut truncated = false;
3535 for info in walk.by_ref() {
3636 let info = info.map_err(|e| crate::Error::Find(e.to_string()))?;
3737 if skipped < skip {
@@ -39,17 +39,17 @@
3939 continue;
4040 }
4141 if commits.len() as u64 >= limit {
42 last_id = Some(info.id().to_string());
42 truncated = true;
4343 break;
4444 }
4545 let id = info.id().detach();
46 last_id = Some(id.to_string());
4746 commits.push(decode_commit(&repo, id)?);
4847 }
48 let cursor = (skip + commits.len() as u64).to_string();
4949 let v = LogList {
5050 commits,
51 cursor: last_id,
52 has_more: false,
51 cursor: Some(cursor),
52 has_more: truncated,
5353 };
5454 self.cache_put_rev("log", "list", &key, tip, &v);
5555 Ok(v)
Modified dvcs/git/object/cache.rs View dvcs/git/object/cache.rs
@@ -148,7 +148,18 @@
148148 F: std::future::Future<Output = T>,
149149{
150150 match tokio::runtime::Handle::try_current() {
151 Ok(handle) => handle.block_on(f()),
151 Ok(handle) => match handle.runtime_flavor() {
152 tokio::runtime::RuntimeFlavor::MultiThread => {
153 // 已在 multi-thread runtime 的异步上下文（如 GraphQL resolver）内：
154 // 直接 handle.block_on 会 panic，改用 block_in_place 迁出 worker 线程执行
155 tokio::task::block_in_place(|| handle.block_on(f()))
156 }
157 _ => tokio::runtime::Builder::new_current_thread()
158 .enable_all()
159 .build()
160 .expect("cache runtime")
161 .block_on(f()),
162 },
152163 Err(_) => tokio::runtime::Builder::new_current_thread()
153164 .enable_all()
154165 .build()
Modified dvcs/git/pack/usecase.rs View dvcs/git/pack/usecase.rs
@@ -31,7 +31,18 @@
3131 .stdin_bytes(data)
3232 .read()
3333 .map_err(|e| crate::Error::Other(format!("git index-pack failed: {e}")))?;
34 let object_count = out.lines().count() as u64;
34 // stdout 仅含一行 pack 路径（"pack\t<name>.pack"），对象数需从生成的 .idx 读取
35 let object_count = out
36 .lines()
37 .find_map(|l| l.strip_prefix("pack\t"))
38 .map(|s| s.trim().to_string())
39 .and_then(|name| {
40 let idx_path = path.join("objects/pack").join(format!("{name}.idx"));
41 gix::odb::pack::index::File::at(&idx_path, repo.object_hash())
42 .map(|f| f.num_objects() as u64)
43 .ok()
44 })
45 .unwrap_or(0);
3546 tracing::info!(
3647 object_count,
3748 size = data.len(),
@@ -45,6 +56,14 @@
4556 }
4657
4758 fn pack_index(&self, pack_id: &str) -> Result<IndexResult, crate::Error> {
59 // pack_id 必须为 40/64 位 hex，防止路径穿越（../../ 将任意文件喂给 index-pack）
60 if !(pack_id.len() == 40 || pack_id.len() == 64)
61 || !pack_id.bytes().all(|b| b.is_ascii_hexdigit())
62 {
63 return Err(crate::Error::InvalidRevSpec(format!(
64 "invalid pack id: {pack_id}"
65 )));
66 }
4867 let repo = self.repo()?;
4968 let path = repo.path().to_path_buf();
5069 let pack_dir = path.join("objects/pack");
@@ -54,7 +73,18 @@
5473 }
5574 let pack_file = pack_file.to_string_lossy().to_string();
5675 let out = crate::gitcli::run(&path, &["index-pack", &pack_file])?;
57 let object_count = out.lines().count() as u64;
76 // stdout 仅含 pack 路径（"pack\t<name>.pack"），对象数从同目录 .idx 读取
77 let object_count = out
78 .lines()
79 .find_map(|l| l.strip_prefix("pack\t"))
80 .map(|s| s.trim().to_string())
81 .and_then(|name| {
82 let idx_path = path.join("objects/pack").join(format!("{name}.idx"));
83 gix::odb::pack::index::File::at(&idx_path, repo.object_hash())
84 .map(|f| f.num_objects() as u64)
85 .ok()
86 })
87 .unwrap_or(0);
5888 tracing::info!(pack_id, object_count, "pack indexed");
5989 Ok(IndexResult {
6090 pack_id: pack_id.into(),
Modified dvcs/git/patch/usecase.rs View dvcs/git/patch/usecase.rs
@@ -80,6 +80,14 @@
8080 fn patch_apply(&self, patch: &str, base: &str) -> Result<ApplyResult, crate::Error> {
8181 let repo = self.repo()?;
8282 let repo_path = repo.path().to_path_buf();
83 // base 先解析为 object id，防止把用户输入原样拼进 git fetch 参数
84 // （如 --upload-pack=<程序> 选项注入可导致任意命令执行）
85 let base_id = repo
86 .rev_parse_single(base.as_bytes())
87 .map_err(|e| crate::Error::RevParse(e.to_string()))?
88 .detach();
89 let base_hex = base_id.to_hex().to_string();
90 crate::validate::validate_oid(&base_hex, base_id.kind().len_in_hex() as u64)?;
8391 let tmp = std::env::temp_dir().join(format!(
8492 "git-patch-{}-{}",
8593 std::process::id(),
@@ -102,7 +110,7 @@
102110 }
103111 let fetch = duct::cmd(
104112 "git",
105 &["fetch", "-q", repo_path.to_str().unwrap_or(""), base],
113 &["fetch", "-q", repo_path.to_str().unwrap_or(""), &base_hex],
106114 )
107115 .dir(&tmp)
108116 .run()
Modified dvcs/git/refs/usecase.rs View dvcs/git/refs/usecase.rs
@@ -60,9 +60,10 @@
6060 let mut out = Vec::new();
6161 for r in iter {
6262 let r = r.map_err(|e| crate::Error::Ref(e.to_string()))?;
63 let oid = r.target().try_id().map(|o| o.to_owned()).ok_or_else(|| {
64 crate::Error::Ref(format!("non-object ref {}", r.name().as_bstr()))
65 })?;
63 // 符号引用（如 refs/remotes/origin/HEAD）无对象目标，跳过而非整表报错
64 let Some(oid) = r.target().try_id().map(|o| o.to_owned()) else {
65 continue;
66 };
6667 out.push(ref_to_info(&r.name().as_bstr().to_string(), oid));
6768 }
6869 self.cache_put_rev("refs", "list", "all", tip, &out);
@@ -85,9 +86,10 @@
8586 let mut out = Vec::new();
8687 for r in iter {
8788 let r = r.map_err(|e| crate::Error::Ref(e.to_string()))?;
88 let oid = r.target().try_id().map(|o| o.to_owned()).ok_or_else(|| {
89 crate::Error::Ref(format!("non-object ref {}", r.name().as_bstr()))
90 })?;
89 // 符号引用无对象目标，跳过而非整表报错
90 let Some(oid) = r.target().try_id().map(|o| o.to_owned()) else {
91 continue;
92 };
9193 out.push(ref_to_info(&r.name().as_bstr().to_string(), oid));
9294 }
9395 self.cache_put_rev("refs", "list_filtered", pattern, tip, &out);
@@ -245,9 +247,10 @@
245247 let mut skipped = 0u64;
246248 for r in iter {
247249 let r = r.map_err(|e| crate::Error::Ref(e.to_string()))?;
248 let oid = r.target().try_id().map(|o| o.to_owned()).ok_or_else(|| {
249 crate::Error::Ref(format!("non-object ref {}", r.name().as_bstr()))
250 })?;
250 // 符号引用无对象目标，跳过而非整表报错
251 let Some(oid) = r.target().try_id().map(|o| o.to_owned()) else {
252 continue;
253 };
251254 if skipped < skip {
252255 skipped += 1;
253256 continue;
Modified dvcs/git/remote/usecase.rs View dvcs/git/remote/usecase.rs
@@ -70,7 +70,9 @@
7070 validate::allowed_remote_scheme(&config.url)?;
7171 let repo = self.repo()?;
7272 let path = repo.path().to_path_buf();
73 let rs = refspec.unwrap_or("+refs/heads/*:refs/heads/*");
73 // Default refspec without the "+" force prefix: a plain push must
74 // never force-overwrite the remote without an explicit opt-in.
75 let rs = refspec.unwrap_or("refs/heads/*:refs/heads/*");
7476 crate::gitcli::run_remote(&path, &["push", "--quiet", &config.url, rs], &config.auth)?;
7577 tracing::info!(url = %config.url, "remote push completed");
7678 Ok(PushResult {
Modified dvcs/git/signature/usecase.rs View dvcs/git/signature/usecase.rs
@@ -110,7 +110,15 @@
110110 };
111111 let mut buf = Vec::new();
112112 use gix::objs::WriteTo;
113 tag_ref
113 let signed_tag = gix::objs::Tag {
114 target: tag_ref.target(),
115 target_kind: tag_ref.target_kind,
116 name: tag_ref.name.into(),
117 tagger: tag_ref.tagger().ok().flatten().and_then(|s| s.to_owned().ok()),
118 message: tag_ref.message.into(),
119 pgp_signature: None,
120 };
121 signed_tag
114122 .write_to(&mut buf)
115123 .map_err(|e| crate::Error::Odb(e.to_string()))?;
116124 let valid = verify_pgp(sig.as_ref(), &buf)?;
Modified dvcs/git/tag/usecase.rs View dvcs/git/tag/usecase.rs
@@ -11,12 +11,11 @@
1111use crate::validate;
1212
1313fn resolve_commit(repo: &gix::Repository, rev: &str) -> Result<ObjectId, crate::Error> {
14 let mut name = rev.to_string();
14 let name = rev.to_string();
1515 if name.starts_with("refs/") {
1616 validate::validate_qualified_ref(&name)?;
1717 } else {
1818 validate::validate_short_name("tag", &name)?;
19 name = format!("refs/tags/{name}");
2019 }
2120 let id = repo
2221 .rev_parse_single(name.as_bytes())
Modified dvcs/git/tree/usecase.rs View dvcs/git/tree/usecase.rs
@@ -31,6 +31,48 @@
3131 }
3232}
3333
34fn unquote_path(s: &str) -> String {
35 if !s.starts_with('"') || !s.ends_with('"') || s.len() < 2 {
36 return s.to_string();
37 }
38 let inner = &s[1..s.len() - 1];
39 let mut out = String::with_capacity(inner.len());
40 let mut chars = inner.chars();
41 while let Some(c) = chars.next() {
42 if c != '\\' {
43 out.push(c);
44 continue;
45 }
46 match chars.next() {
47 Some('n') => out.push('\n'),
48 Some('t') => out.push('\t'),
49 Some('r') => out.push('\r'),
50 Some('"') => out.push('"'),
51 Some('\\') => out.push('\\'),
52 Some(o @ '0'..='7') => {
53 let mut v = o.to_digit(8).unwrap_or(0);
54 let mut n = 1;
55 while n < 3 {
56 match chars.next() {
57 Some(d @ '0'..='7') => {
58 v = v * 8 + d.to_digit(8).unwrap_or(0);
59 n += 1;
60 }
61 _ => break,
62 }
63 }
64 out.push(char::from_u32(v).unwrap_or('\u{FFFD}'));
65 }
66 Some(other) => {
67 out.push('\\');
68 out.push(other);
69 }
70 None => out.push('\\'),
71 }
72 }
73 out
74}
75
3476impl TreeOps for GitRepository {
3577 fn tree_read(&self, id: &gix::hash::ObjectId) -> Result<Tree, crate::Error> {
3678 if let Some(hit) = self.cache_get_oid::<Tree>("tree", id) {
@@ -224,7 +266,7 @@
224266 }
225267 _ => {
226268 if let (Some(ts), Some(id)) = (cur_time, cur_hash) {
227 last.entry(line.to_string()).or_insert((ts, id));
269 last.entry(unquote_path(line)).or_insert((ts, id));
228270 }
229271 }
230272 }
Modified dvcs/git/validate.rs View dvcs/git/validate.rs
@@ -20,6 +20,11 @@
2020 "invalid {kind} name: {name}"
2121 )));
2222 }
23 if name.starts_with('-') {
24 return Err(crate::Error::InvalidRefName(format!(
25 "invalid {kind} name: {name}"
26 )));
27 }
2328 if name.starts_with("refs/") {
2429 return Err(crate::Error::InvalidRefName(format!(
2530 "expected short {kind} name, got qualified: {name}"
@@ -58,11 +63,45 @@
5863 .map(|(s, _)| s.to_ascii_lowercase())
5964 .unwrap_or_default();
6065 match scheme.as_str() {
61 "http" | "https" | "ssh" => Ok(()),
66 "http" | "https" | "ssh" => {}
6267 _ => Err(crate::Error::InvalidRevSpec(format!(
6368 "remote url scheme not allowed: {url}"
64 ))),
69 )))?,
6570 }
71 let host = url
72 .split_once("://")
73 .map(|(_, rest)| rest)
74 .unwrap_or_default()
75 .rsplit('@')
76 .next()
77 .unwrap_or_default()
78 .split('/')
79 .next()
80 .unwrap_or_default();
81 if is_private_host(host) {
82 return Err(crate::Error::InvalidRevSpec(format!(
83 "remote url host not allowed: {url}"
84 )));
85 }
86 Ok(())
87}
88
89fn is_private_host(host: &str) -> bool {
90 let host = match host.strip_prefix('[') {
91 Some(rest) => rest.split(']').next().unwrap_or(rest),
92 None => host.split(':').next().unwrap_or(host),
93 };
94 if host.is_empty() || host.eq_ignore_ascii_case("localhost") || host.eq_ignore_ascii_case("::1") {
95 return true;
96 }
97 if let Ok(ip) = host.parse::<std::net::Ipv4Addr>() {
98 return ip.is_loopback() || ip.is_private() || ip.is_link_local() || ip.is_unspecified() || ip.is_broadcast();
99 }
100 if let Ok(ip) = host.parse::<std::net::Ipv6Addr>() {
101 return ip.is_loopback() || ip.is_unspecified();
102 }
103 let lower = host.to_ascii_lowercase();
104 lower.ends_with(".local") || lower.ends_with(".lan") || lower.ends_with(".internal")
66105}
67106
68107pub const MAX_OBJECT_SIZE: u64 = 100 * 1024 * 1024;
Modified dvcs/gitauth/permission.rs View dvcs/gitauth/permission.rs
@@ -28,12 +28,10 @@
2828 }
2929 None => (None, None),
3030 };
31 let channel_role = None;
32
3331 repo::permission::require(
3432 actor,
3533 required,
36 channel_role,
34 None,
3735 space_role,
3836 collaborator,
3937 !resolved.repo.is_private,
Modified dvcs/gitauth/service.rs View dvcs/gitauth/service.rs
@@ -139,6 +139,17 @@
139139 }
140140 }
141141
142 fn constant_time_eq(a: &str, b: &str) -> bool {
143 if a.len() != b.len() {
144 return false;
145 }
146 let mut diff: u8 = 0;
147 for (x, y) in a.bytes().zip(b.bytes()) {
148 diff |= x ^ y;
149 }
150 diff == 0
151 }
152
142153 async fn verify_access_key(
143154 &self,
144155 user: &users::Model,
@@ -149,9 +160,10 @@
149160 .filter(user_access_keys::Column::UserId.eq(user.uid))
150161 .all(&db)
151162 .await?;
152 Ok(keys.into_iter().find(|key| {
153 auth::features::access_token::hash(presented, &key.token_salt) == key.token_hash
154 }))
163Ok(keys.into_iter().find(|key| {
164 let expected = auth::features::access_token::hash(presented, &key.token_salt);
165 Self::constant_time_eq(&expected, &key.token_hash)
166 }))
155167 }
156168}
157169
Modified dvcs/gitsmart/handler/lfs.rs View dvcs/gitsmart/handler/lfs.rs
@@ -14,7 +14,7 @@
1414use crate::handler::{authenticate, authorize, decode_request_body};
1515use crate::service::{ResolvedRepo, SmartService};
1616
17const MAX_LFS_BODY: usize = 8 * 1024 * 1024 * 1024;
17const MAX_LFS_BODY: usize = 64 * 1024 * 1024;
1818const STREAM_BUF_LEN: usize = 64 * 1024;
1919
2020pub async fn dispatch(
@@ -29,9 +29,15 @@
2929 let actor = authenticate(state, headers).await?;
3030 match path {
3131 "info/lfs/objects/batch" if method == Method::POST => {
32 let resolved =
33 authorize(state, space, repo, &actor, repo::actor::RepoAccess::Write).await?;
34 batch(state, &resolved, headers, body).await
32 let bytes = decode_request_body(headers, body, MAX_LFS_BODY).await?;
33 let req: LfsBatchRequest = serde_json::from_slice(&bytes)
34 .map_err(|e| SmartError::BadRequest(format!("invalid batch request: {e}")))?;
35 let required = match req.operation {
36 LfsOperation::Upload => repo::actor::RepoAccess::Write,
37 LfsOperation::Download | LfsOperation::Verify => repo::actor::RepoAccess::Read,
38 };
39 let resolved = authorize(state, space, repo, &actor, required).await?;
40 batch(state, &resolved, req).await
3541 }
3642 "info/lfs/verify" if method == Method::POST => {
3743 let resolved =
@@ -66,12 +72,8 @@
6672async fn batch(
6773 state: &SmartService,
6874 resolved: &ResolvedRepo,
69 headers: &HeaderMap,
70 body: Body,
75 req: LfsBatchRequest,
7176) -> SmartResult<Response> {
72 let bytes = decode_request_body(headers, body, MAX_LFS_BODY).await?;
73 let req: LfsBatchRequest = serde_json::from_slice(&bytes)
74 .map_err(|e| SmartError::BadRequest(format!("invalid batch request: {e}")))?;
7577 let repo = state.git_repository(resolved)?;
7678 let lfs = repo.lfs;
7779 let base = format!(
Modified dvcs/gitsmart/handler/mod.rs View dvcs/gitsmart/handler/mod.rs
@@ -54,12 +54,16 @@
5454 let mut endpoint: &'static str = "other";
5555 let result: SmartResult<Response> = async {
5656 let path_lower = path.to_lowercase();
57 let is_get = method == Method::GET;
57 let is_get = method == Method::GET || method == Method::HEAD;
5858 match (path_lower.as_str(), is_get) {
5959 ("info/refs", true) => {
6060 endpoint = "info_refs";
6161 let service = query.get("service").map(String::as_str);
62 smart::info_refs(&state, &space, &repo_name, &headers, service).await
62 let mut resp = smart::info_refs(&state, &space, &repo_name, &headers, service).await?;
63 if method == Method::HEAD {
64 *resp.body_mut() = Body::empty();
65 }
66 Ok(resp)
6367 }
6468 ("git-upload-pack", false) => {
6569 endpoint = "upload_pack";
@@ -147,21 +151,32 @@
147151}
148152
149153fn url_decode(s: &str) -> String {
150 let mut out = String::with_capacity(s.len());
154 let mut out: Vec<u8> = Vec::with_capacity(s.len());
151155 let bytes = s.as_bytes();
152156 let mut i = 0;
153157 while i < bytes.len() {
154158 if bytes[i] == b'%' && i + 2 < bytes.len() {
155 if let Ok(byte) = u8::from_str_radix(&s[i + 1..i + 3], 16) {
156 out.push(byte as char);
159 let hi = hex_value(bytes[i + 1]);
160 let lo = hex_value(bytes[i + 2]);
161 if let (Some(hi), Some(lo)) = (hi, lo) {
162 out.push((hi << 4) | lo);
157163 i += 3;
158164 continue;
159165 }
160166 }
161 out.push(bytes[i] as char);
167 out.push(bytes[i]);
162168 i += 1;
163169 }
164 out
170 String::from_utf8_lossy(&out).into_owned()
171}
172
173fn hex_value(b: u8) -> Option<u8> {
174 match b {
175 b'0'..=b'9' => Some(b - b'0'),
176 b'a'..=b'f' => Some(b - b'a' + 10),
177 b'A'..=b'F' => Some(b - b'A' + 10),
178 _ => None,
179 }
165180}
166181
167182pub(crate) async fn authenticate(
Modified dvcs/gitsmart/handler/smart.rs View dvcs/gitsmart/handler/smart.rs
@@ -11,7 +11,7 @@
1111use crate::protocol::{GitService, pktline, rpc};
1212use crate::service::{ResolvedRepo, SmartService};
1313
14const MAX_RPC_BODY: usize = 1 << 30;
14const MAX_RPC_BODY: usize = 512 * 1024 * 1024;
1515
1616pub async fn info_refs(
1717 state: &SmartService,
@@ -84,7 +84,7 @@
8484 let is_receive = service == GitService::ReceivePack;
8585 if is_receive {
8686 if let Some(actor_uid) = actor.user_id {
87 let updates = parse_receive_pack_refs(&input);
87 let updates = parse_receive_pack_refs(&input)?;
8888 if !updates.is_empty() {
8989 let db = state.auth.db.reader();
9090 let verdict = pre_receive::validate_pre_receive(
Modified dvcs/gitsmart/protocol/pktline.rs View dvcs/gitsmart/protocol/pktline.rs
@@ -6,7 +6,7 @@
66
77pub fn encode_line(line: &[u8]) -> Vec<u8> {
88 let len = line.len() + 4;
9 debug_assert!(len <= MAX_PKT_SIZE, "pkt-line too long: {len}");
9 assert!(len <= MAX_PKT_SIZE, "pkt-line too long: {len}");
1010 let mut out = Vec::with_capacity(4 + line.len());
1111 write!(out, "{len:04x}").expect("write hex len");
1212 out.extend_from_slice(line);
Modified dvcs/gitsmart/protocol/rpc.rs View dvcs/gitsmart/protocol/rpc.rs
@@ -6,7 +6,7 @@
66use bytes::Bytes;
77use futures::StreamExt;
88use tokio::io::{AsyncReadExt, AsyncWriteExt};
9use tokio::process::{ChildStdout, Command};
9use tokio::process::{Child, ChildStdout, Command};
1010use tokio::task::JoinHandle;
1111
1212use crate::error::{SmartError, SmartResult};
@@ -21,9 +21,22 @@
2121 reader: ChildStdout,
2222 reaper: Option<JoinHandle<()>>,
2323 shared: SharedResult,
24 stdin_task: Option<JoinHandle<io::Result<()>>>,
25 child: Arc<tokio::sync::Mutex<Option<Child>>>,
2426 done: bool,
2527}
2628
29impl Drop for StreamState {
30 fn drop(&mut self) {
31 if let Some(task) = self.stdin_task.take() {
32 task.abort();
33 }
34 if let Some(child) = self.child.blocking_lock().as_mut() {
35 let _ = child.start_kill();
36 }
37 }
38}
39
2740pub async fn run_rpc(
2841 bare_path: &Path,
2942 service: GitService,
@@ -50,20 +63,26 @@
5063 .map_err(|e| SmartError::Internal(format!("failed to spawn git {service_name}: {e}")))?;
5164
5265 let mut stdin = child.stdin.take().expect("stdin piped");
66 let stdout = child.stdout.take().expect("stdout piped");
67 let mut stderr = child.stderr.take().expect("stderr piped");
5368 let stdin_task = tokio::spawn(async move {
5469 let mut input = input;
5570 while let Some(chunk) = input.next().await {
56 let chunk = chunk?;
57 stdin.write_all(&chunk).await?;
71 let chunk = match chunk {
72 Ok(c) => c,
73 Err(_) => break,
74 };
75 if stdin.write_all(&chunk).await.is_err() {
76 break;
77 }
5878 }
59 stdin.shutdown().await?;
79 let _ = stdin.shutdown().await;
6080 io::Result::Ok(())
6181 });
62
63 let stdout = child.stdout.take().expect("stdout piped");
64 let mut stderr = child.stderr.take().expect("stderr piped");
82 let child_guard = Arc::new(tokio::sync::Mutex::new(Some(child)));
6583 let shared: SharedResult = Arc::new(tokio::sync::Mutex::new(None));
6684 let shared_clone = Arc::clone(&shared);
85 let child_clone = Arc::clone(&child_guard);
6786 let reaper = tokio::spawn(async move {
6887 let mut stderr_buf = Vec::new();
6988 let mut chunk = vec![0u8; 4096];
@@ -73,13 +92,16 @@
7392 Ok(n) => stderr_buf.extend_from_slice(&chunk[..n]),
7493 }
7594 }
76 let result = match child.wait().await {
77 Ok(status) if status.success() => Ok(()),
78 Ok(status) => Err(format!(
79 "git {service_name} exited with {status}: {}",
80 String::from_utf8_lossy(&stderr_buf)
81 )),
82 Err(e) => Err(format!("waiting for git {service_name} failed: {e}")),
95let result = match child_clone.lock().await.take() {
96 Some(mut child) => match child.wait().await {
97 Ok(status) if status.success() => Ok(()),
98 Ok(status) => Err(format!(
99 "git {service_name} exited with {status}: {}",
100 String::from_utf8_lossy(&stderr_buf)
101 )),
102 Err(e) => Err(format!("waiting for git {service_name} failed: {e}")),
103 },
104 None => Err(format!("git {service_name} was killed")),
83105 };
84106 *shared_clone.lock().await = Some(result);
85107 });
@@ -88,6 +110,8 @@
88110 reader: stdout,
89111 reaper: Some(reaper),
90112 shared,
113 stdin_task: Some(stdin_task),
114 child: child_guard,
91115 done: false,
92116 };
93117
@@ -116,7 +140,6 @@
116140 }
117141 });
118142
119 let _ = stdin_task;
120143 Ok(Box::pin(stream))
121144}
122145
Modified dvcs/gitsmart/pre_receive.rs View dvcs/gitsmart/pre_receive.rs
@@ -21,14 +21,14 @@
2121
2222impl PreReceiveVerdict {
2323 pub fn accepted_only(self) -> SmartResult<Vec<RefUpdate>> {
24 let mut iter = self.per_ref.into_iter();
25 if let Some((_, RefVerdict::Reject(msg))) = iter
26 .by_ref()
27 .find(|(_, v)| matches!(v, RefVerdict::Reject(_)))
28 {
29 return Err(SmartError::Forbidden(msg));
24 let mut accepted = Vec::with_capacity(self.per_ref.len());
25 for (r, v) in self.per_ref {
26 match v {
27 RefVerdict::Accept => accepted.push(r),
28 RefVerdict::Reject(msg) => return Err(SmartError::Forbidden(msg)),
29 }
3030 }
31 Ok(iter.map(|(r, _)| r).collect())
31 Ok(accepted)
3232 }
3333}
3434
@@ -66,7 +66,8 @@
6666 Some(collab_entity::CollaborationMode::Admin)
6767 );
6868
69 let zero_oid = "0".repeat(40);
69 let oid_len = oid_len_of(refs);
70 let zero_oid = zero_oid(oid_len);
7071
7172 let mut out = Vec::with_capacity(refs.len());
7273 for r in refs {
@@ -174,44 +175,64 @@
174175 RefVerdict::Accept
175176}
176177
178fn zero_oid(len: usize) -> String {
179 "0".repeat(len)
180}
181
182/// pre-receive 校验阶段对象尚未写入 bare 仓库（校验发生在 run_rpc_bytes 之前），
183/// 无法用 merge-base 判定快进；仅在可静态确认（无变化更新）时为 true，
184/// 其余情况视为潜在 force-push 交由策略 fail-closed 处理。
177185fn is_fast_forward(old: &str, new: &str) -> bool {
178 old == new || old == "0".repeat(40) || new == "0".repeat(40)
186 old == new
179187}
180188
181pub fn parse_receive_pack_refs(input: &[u8]) -> Vec<RefUpdate> {
182 let lines = match crate::protocol::pktline::decode_stream(input) {
183 Ok(lines) => lines,
184 Err(_) => return Vec::new(),
185 };
189fn oid_len_of(refs: &[RefUpdate]) -> usize {
190 refs.iter()
191 .find_map(|r| {
192 let max = r.old_oid.len().max(r.new_oid.len());
193 (max == 40 || max == 64).then_some(max)
194 })
195 .unwrap_or(40)
196}
197
198pub fn parse_receive_pack_refs(input: &[u8]) -> SmartResult<Vec<RefUpdate>> {
199 let lines = crate::protocol::pktline::decode_stream(input).map_err(|e| {
200 SmartError::BadRequest(format!("invalid receive-pack request: {e}"))
201 })?;
186202 let mut out = Vec::new();
187203 for mut line in lines {
188204 if let Some(nul) = line.iter().position(|b| *b == 0) {
189205 line.truncate(nul);
190206 }
191 let s = match std::str::from_utf8(&line) {
192 Ok(s) => s,
193 Err(_) => continue,
194 };
195 let mut parts = s.splitn(3, ' ');
196 let old = match parts.next() {
197 Some(p) if !p.is_empty() => p,
198 _ => continue,
199 };
200 let new = match parts.next() {
201 Some(p) if !p.is_empty() => p,
202 _ => continue,
203 };
204 let refname = match parts.next() {
205 Some(p) if !p.is_empty() => p,
206 _ => continue,
207 };
208 if old.len() != 40 || new.len() != 40 {
207 if line.is_empty() {
209208 continue;
210209 }
211 if !old.chars().all(|c| c.is_ascii_hexdigit())
210 let s = std::str::from_utf8(&line).map_err(|_| {
211 SmartError::BadRequest("receive-pack ref line is not valid UTF-8".to_string())
212 })?;
213 let mut parts = s.splitn(3, ' ');
214 let old = parts.next().unwrap_or_default();
215 let new = parts.next().unwrap_or_default();
216 let refname = parts.next().unwrap_or_default();
217 if old.is_empty() || new.is_empty() || refname.is_empty() {
218 return Err(SmartError::BadRequest(format!(
219 "malformed receive-pack ref line: {s}"
220 )));
221 }
222 let oid_len = old.len();
223 if oid_len != 40 && oid_len != 64 {
224 return Err(SmartError::BadRequest(format!(
225 "invalid oid length {} in receive-pack ref line: {s}",
226 oid_len
227 )));
228 }
229 if new.len() != oid_len
230 || !old.chars().all(|c| c.is_ascii_hexdigit())
212231 || !new.chars().all(|c| c.is_ascii_hexdigit())
213232 {
214 continue;
233 return Err(SmartError::BadRequest(format!(
234 "invalid oid in receive-pack ref line: {s}"
235 )));
215236 }
216237 out.push(RefUpdate {
217238 old_oid: old.to_string(),
@@ -219,5 +240,5 @@
219240 refname: refname.to_string(),
220241 });
221242 }
222 out
243 Ok(out)
223244}
Modified dvcs/gitsmart/push.rs View dvcs/gitsmart/push.rs
@@ -64,11 +64,15 @@
6464 _member: &[u8],
6565 _ttl_secs: u64,
6666 ) -> Result<bool, SmartError> {
67 Ok(true)
67 Err(SmartError::Internal(
68 "push coordination disabled: redis not configured".into(),
69 ))
6870 }
6971
7072 async fn renew(&self, _key: &str, _token: &[u8], _ttl_secs: u64) -> Result<bool, SmartError> {
71 Ok(true)
73 Err(SmartError::Internal(
74 "push coordination disabled: redis not configured".into(),
75 ))
7276 }
7377
7478 async fn release(&self, _key: &str, _token: &[u8]) -> Result<(), SmartError> {
@@ -76,11 +80,15 @@
7680 }
7781
7882 async fn enqueue(&self, _key: &str, _member: &[u8]) -> Result<(), SmartError> {
79 Ok(())
83 Err(SmartError::Internal(
84 "push coordination disabled: redis not configured".into(),
85 ))
8086 }
8187
8288 async fn dequeue(&self, _key: &str, _member: &[u8]) -> Result<(), SmartError> {
83 Ok(())
89 Err(SmartError::Internal(
90 "push coordination disabled: redis not configured".into(),
91 ))
8492 }
8593}
8694
@@ -162,10 +170,23 @@
162170}
163171
164172struct Heartbeat {
165 handle: tokio::task::JoinHandle<()>,
173 handle: Option<tokio::task::JoinHandle<()>>,
166174 stop: Arc<tokio::sync::Notify>,
167175}
168176
177impl Drop for Heartbeat {
178 /// panic/unwind 逃逸 `run_locked` 时也必须停止心跳任务，
179 /// 否则该任务的续租循环会永久持有 push 锁，后续所有 push 卡死。
180 fn drop(&mut self) {
181 self.stop.notify_one();
182 if let Some(handle) = &self.handle {
183 if !handle.is_finished() {
184 handle.abort();
185 }
186 }
187 }
188}
189
169190impl Heartbeat {
170191 fn spawn(
171192 backend: Arc<dyn PushOperations>,
@@ -181,22 +202,25 @@
181202 tokio::select! {
182203 _ = stop_clone.notified() => break,
183204 _ = tokio::time::sleep(Duration::from_secs(interval)) => {
184 if backend
185 .renew(&key, token.as_bytes(), ttl)
186 .await
187 .is_err()
188 {
189 break;
205 match backend.renew(&key, token.as_bytes(), ttl).await {
206 // 锁已被他人夺走（TTL 过期/Redis 抖动），再续租无意义
207 Ok(false) | Err(_) => break,
208 Ok(true) => {}
190209 }
191210 }
192211 }
193212 }
194213 });
195 Self { handle, stop }
214 Self {
215 handle: Some(handle),
216 stop,
217 }
196218 }
197219
198 async fn stop(self) {
220 async fn stop(mut self) {
199221 self.stop.notify_one();
200 let _ = self.handle.await;
222 if let Some(handle) = self.handle.take() {
223 let _ = handle.await;
224 }
201225 }
202226}
Modified dvcs/gitssh/handler/channel.rs View dvcs/gitssh/handler/channel.rs
@@ -3,13 +3,23 @@
33use russh::ChannelId;
44use russh::server::Session;
55
6use crate::git_service::apply_hardened_env;
67use crate::handler::GitSshHandler;
78
8pub const MAX_SSH_CHANNEL_BUF: usize = 1 << 30;
9pub const MAX_SSH_CHANNEL_BUF: usize = 64 * 1024 * 1024;
10pub const MAX_GIT_OUTPUT_BUF: usize = 256 * 1024 * 1024;
911
1012impl GitSshHandler {
11 pub async fn data_impl(&mut self, channel: ChannelId, data: &[u8]) -> Result<(), russh::Error> {
13 pub async fn data_impl(
14 &mut self,
15 channel: ChannelId,
16 data: &[u8],
17 session: &mut Session,
18 ) -> Result<(), russh::Error> {
1219 let state = self.channels.entry(channel).or_default();
20 if state.buffer_overflow {
21 return Ok(());
22 }
1323 let projected = state.buffer.len().saturating_add(data.len());
1424 if projected > MAX_SSH_CHANNEL_BUF {
1525 tracing::warn!(
@@ -17,10 +27,13 @@
1727 buffered = state.buffer.len(),
1828 incoming = data.len(),
1929 limit = MAX_SSH_CHANNEL_BUF,
20 "ssh channel buffer overflow; dropping channel"
30 "ssh channel buffer overflow; closing channel"
2131 );
2232 state.buffer_overflow = true;
23 return Err(russh::Error::Disconnect);
33 state.buffer.clear();
34 let _ = session.close(channel);
35 let _ = session.flush();
36 return Ok(());
2437 }
2538 state.buffer.extend_from_slice(data);
2639 Ok(())
@@ -35,7 +48,7 @@
3548 channel: ChannelId,
3649 session: &mut Session,
3750 ) -> Result<(), russh::Error> {
38 let Some(state) = self.channels.get_mut(&channel) else {
51 let Some(mut state) = self.channels.remove(&channel) else {
3952 let _ = session.close(channel);
4053 let _ = session.flush();
4154 return Ok(());
@@ -78,41 +91,61 @@
7891 );
7992
8093 if is_receive && let (Some(repo_id), Some(actor_uid)) = (repo_id, actor_id) {
81 let updates = parse_receive_pack_refs(&buffer);
82 if !updates.is_empty() {
83 let db_conn = db.reader();
84 match pre_receive::validate_pre_receive(&db_conn, repo_id, actor_uid, &updates)
85 .await
86 {
87 Ok(verdict) => {
88 if let Err(e) = verdict.accepted_only() {
89 let msg = format!("remote: {}\n", e);
90 let _ = handle.data(channel, Bytes::from(msg.into_bytes())).await;
94 match parse_receive_pack_refs(&buffer) {
95 Ok(updates) if !updates.is_empty() => {
96 let db_conn = db.reader();
97 match pre_receive::validate_pre_receive(&db_conn, repo_id, actor_uid, &updates)
98 .await
99 {
100 Ok(verdict) => {
101 if let Err(e) = verdict.accepted_only() {
102 let msg = format!("remote: {}\n", e);
103 let _ = handle.data(channel, Bytes::from(msg.into_bytes())).await;
104 let _ = handle.exit_status_request(channel, 1).await;
105 let _ = handle.eof(channel).await;
106 let _ = handle.close(channel).await;
107 return;
108 }
109 }
110 Err(e) => {
111 tracing::error!(
112 ?e, repo_id = %repo_id,
113 "pre-receive validation failed; refusing push"
114 );
115 let _ = handle
116 .data(
117 channel,
118 Bytes::from_static(
119 b"remote: internal error during pre-receive validation\n",
120 ),
121 )
122 .await;
91123 let _ = handle.exit_status_request(channel, 1).await;
92124 let _ = handle.eof(channel).await;
93125 let _ = handle.close(channel).await;
94126 return;
95127 }
96128 }
97 Err(e) => {
98 tracing::error!(
99 ?e, repo_id = %repo_id,
100 "pre-receive validation failed; refusing push"
101 );
102 let _ = handle
103 .data(
104 channel,
105 Bytes::from_static(
106 b"remote: internal error during pre-receive validation\n",
107 ),
108 )
109 .await;
110 let _ = handle.exit_status_request(channel, 1).await;
111 let _ = handle.eof(channel).await;
112 let _ = handle.close(channel).await;
113 return;
114 }
115129 }
130 Ok(_) => {}
131 Err(e) => {
132 tracing::error!(
133 ?e, repo_id = %repo_id,
134 "receive-pack ref parse failed; refusing push"
135 );
136 let _ = handle
137 .data(
138 channel,
139 Bytes::from_static(
140 b"remote: invalid receive-pack ref update\n",
141 ),
142 )
143 .await;
144 let _ = handle.exit_status_request(channel, 1).await;
145 let _ = handle.eof(channel).await;
146 let _ = handle.close(channel).await;
147 return;
148 }
116149 }
117150 }
118151
@@ -216,9 +249,10 @@
216249 .stdin(std::process::Stdio::piped())
217250 .stdout(std::process::Stdio::piped())
218251 .stderr(std::process::Stdio::piped());
252 apply_hardened_env(&mut cmd);
219253
220254 let mut child = cmd.spawn().map_err(crate::error::SshError::Io)?;
221 use tokio::io::AsyncWriteExt;
255 use tokio::io::{AsyncReadExt, AsyncWriteExt};
222256 if let Some(mut stdin) = child.stdin.take() {
223257 stdin
224258 .write_all(input)
@@ -226,25 +260,44 @@
226260 .map_err(crate::error::SshError::Io)?;
227261 let _ = stdin.shutdown().await;
228262 }
229 let output = child
230 .wait_with_output()
231 .await
232 .map_err(crate::error::SshError::Io)?;
233 if !output.stdout.is_empty() {
234 let _ = handle.data(channel, Bytes::from(output.stdout)).await;
263 let stdout_cap = MAX_GIT_OUTPUT_BUF as u64 + 1;
264 let stdout = if let Some(out) = child.stdout.take() {
265 let mut buf = Vec::new();
266 out.take(stdout_cap).read_to_end(&mut buf).await.map_err(crate::error::SshError::Io)?;
267 buf
268 } else {
269 Vec::new()
270 };
271 let stderr = if let Some(err) = child.stderr.take() {
272 let mut buf = Vec::new();
273 err.take(stdout_cap).read_to_end(&mut buf).await.map_err(crate::error::SshError::Io)?;
274 buf
275 } else {
276 Vec::new()
277 };
278 let status = child.wait().await.map_err(crate::error::SshError::Io)?;
279 if stdout.len() > MAX_GIT_OUTPUT_BUF || stderr.len() > MAX_GIT_OUTPUT_BUF {
280 return Err(crate::error::SshError::Internal(format!(
281 "git {} output exceeded {} bytes",
282 op.service_arg(),
283 MAX_GIT_OUTPUT_BUF
284 )));
285 }
286 if !stdout.is_empty() {
287 let _ = handle.data(channel, Bytes::from(stdout)).await;
235288 }
236 if !output.stderr.is_empty() && !output.status.success() {
289 if !stderr.is_empty() && !status.success() {
237290 let _ = handle
238 .extended_data(channel, 1, Bytes::from(output.stderr))
291 .extended_data(channel, 1, Bytes::from(stderr))
239292 .await;
240293 }
241 if output.status.success() {
294 if status.success() {
242295 Ok(())
243296 } else {
244297 Err(crate::error::SshError::Internal(format!(
245298 "git {} failed: exit {:?}",
246299 op.service_arg(),
247 output.status.code()
300 status.code()
248301 )))
249302 }
250303 }
Modified dvcs/gitssh/handler/command.rs View dvcs/gitssh/handler/command.rs
@@ -6,7 +6,7 @@
66use repo::actor::{ActorContext, RepoAccess};
77
88use crate::error::SshError;
9use crate::git_service::{GitService, parse_git_command, parse_repo_path, sanitize_for_log};
9use crate::git_service::{GitService, build_git_advertise_command, parse_git_command, parse_repo_path, sanitize_for_log};
1010use crate::handler::GitSshHandler;
1111use crate::handler::git_op::GitOperation;
1212
@@ -18,9 +18,14 @@
1818 session: &mut Session,
1919 ) -> Result<(), russh::Error> {
2020 let mut service_label: &'static str = "unknown";
21 let result: Result<(), russh::Error> = async {
22 let cmd = std::str::from_utf8(data)
23 .map_err(|_| invalid_command_error("Invalid command encoding", session, channel))?;
21 let outcome: Result<(), ()> = async {
22 let cmd = match std::str::from_utf8(data) {
23 Ok(c) => c,
24 Err(_) => {
25 invalid_command_error("Invalid command encoding", session, channel);
26 return Err(());
27 }
28 };
2429 let cmd = cmd.trim();
2530 tracing::info!(
2631 "ssh exec from {:?}: {}",
@@ -29,18 +34,12 @@
2934 );
3035
3136 let Some((service, raw_path)) = parse_git_command(cmd) else {
32 return Err(invalid_command_error(
33 "Invalid git command",
34 session,
35 channel,
36 ));
37 invalid_command_error("Invalid git command", session, channel);
38 return Err(());
3739 };
3840 let Some((space, repo)) = parse_repo_path(raw_path) else {
39 return Err(invalid_command_error(
40 "Invalid repository path",
41 session,
42 channel,
43 ));
41 invalid_command_error("Invalid repository path", session, channel);
42 return Err(());
4443 };
4544 service_label = service.as_subcommand();
4645
@@ -70,22 +69,31 @@
7069 Err(e) => {
7170 let msg = auth_error_message(&e);
7271 tracing::warn!("ssh {service:?} {space}/{repo} denied: {msg}");
73 return Err(invalid_command_error(msg.as_str(), session, channel));
72 invalid_command_error(msg.as_str(), session, channel);
73 return Err(());
7474 }
7575 };
7676
7777 let bare = self.service.bare_path(&resolved);
7878 if !bare.is_dir() {
7979 let msg = format!("fatal: repository '{space}/{repo}' does not exist");
80 return Err(invalid_command_error(msg.as_str(), session, channel));
80 invalid_command_error(msg.as_str(), session, channel);
81 return Err(());
8182 }
8283
83 let adv = run_advertise(&bare, service).await.unwrap_or_default();
84 if !adv.is_empty() {
85 let _ = session.data(channel, Bytes::from(adv));
86 }
87 let _ = session.channel_success(channel);
88 let _ = session.flush();
84let adv = match run_advertise(&bare, service).await {
85 Ok(adv) => adv,
86 Err(e) => {
87 let msg = format!("fatal: unable to read repository advertisement: {e}");
88 invalid_command_error(msg.as_str(), session, channel);
89 return Err(());
90 }
91};
92if !adv.is_empty() {
93 let _ = session.data(channel, Bytes::from(adv));
94}
95let _ = session.channel_success(channel);
96let _ = session.flush();
8997
9098 let op = match service {
9199 GitService::UploadPack => GitOperation::UploadPack { bare: bare.clone() },
@@ -101,20 +109,19 @@
101109 Ok(())
102110 }
103111 .await;
104 crate::metrics::record_exec(service_label, result.is_ok());
105 result
112 crate::metrics::record_exec(service_label, outcome.is_ok());
113 Ok(())
106114 }
107115}
108116
109fn invalid_command_error(msg: &str, session: &mut Session, channel: ChannelId) -> russh::Error {
117/// Notifies the client that its exec request was rejected and closes only that
118/// channel. Thus a bad command or denied access does not tear down the whole
119/// SSH session or other concurrent channels on the same connection.
120fn invalid_command_error(_msg: &str, session: &mut Session, channel: ChannelId) {
110121 let _ = session.channel_failure(channel);
111122 let _ = session.exit_status_request(channel, 1);
112123 let _ = session.close(channel);
113124 let _ = session.flush();
114 russh::Error::IO(std::io::Error::new(
115 std::io::ErrorKind::InvalidData,
116 msg.to_string(),
117 ))
118125}
119126
120127fn auth_error_message(e: &gitauth::GitAuthError) -> String {
@@ -132,13 +139,8 @@
132139 bare: &std::path::PathBuf,
133140 service: GitService,
134141) -> Result<Vec<u8>, SshError> {
135 let mut cmd = tokio::process::Command::new("git");
136 cmd.current_dir(bare)
137 .arg(service.as_subcommand())
138 .arg("--advertise-refs")
139 .env("GIT_DIR", bare)
140 .env("GIT_TERMINAL_PROMPT", "0")
141 .stdout(std::process::Stdio::piped())
142 let mut cmd = build_git_advertise_command(service, bare);
143 cmd.stdout(std::process::Stdio::piped())
142144 .stderr(std::process::Stdio::piped());
143145 let mut child = cmd.spawn().map_err(SshError::Io)?;
144146 let stderr_task = if let Some(stderr) = child.stderr.take() {
Modified dvcs/gitssh/handler/mod.rs View dvcs/gitssh/handler/mod.rs
@@ -132,9 +132,9 @@
132132 &mut self,
133133 channel: ChannelId,
134134 data: &[u8],
135 _session: &mut Session,
135 session: &mut Session,
136136 ) -> Result<(), Self::Error> {
137 self.data_impl(channel, data).await
137 self.data_impl(channel, data, session).await
138138 }
139139
140140 async fn exec_request(
Modified dvcs/gitssh/error.rs View dvcs/gitssh/error.rs
@@ -28,8 +28,8 @@
2828 fn from(err: gitsmart::SmartError) -> Self {
2929 match err {
3030 gitsmart::SmartError::Unauthorized(m)
31 | gitsmart::SmartError::BadRequest(m)
32 | gitsmart::SmartError::Conflict(m) => Self::AuthRejected(m),
31 | gitsmart::SmartError::BadRequest(m) => Self::AuthRejected(m),
32 gitsmart::SmartError::Conflict(m) => Self::Internal(m),
3333 gitsmart::SmartError::Forbidden(m) => Self::PermissionDenied(m),
3434 gitsmart::SmartError::NotFound(m) => Self::RepoNotFound(m),
3535 gitsmart::SmartError::Internal(m) => Self::Internal(m),
Modified dvcs/gitssh/git_service.rs View dvcs/gitssh/git_service.rs
@@ -89,15 +89,21 @@
8989 out
9090}
9191
92pub fn build_git_command(service: GitService, bare: &PathBuf) -> tokio::process::Command {
92pub fn build_git_advertise_command(service: GitService, bare: &PathBuf) -> tokio::process::Command {
9393 let mut cmd = tokio::process::Command::new("git");
9494 cmd.current_dir(bare)
9595 .arg(service.as_subcommand())
96 .arg(".")
97 .env("GIT_CONFIG_NOSYSTEM", "1")
96 .arg("--advertise-refs")
97 .env("GIT_DIR", bare)
98 .env("GIT_TERMINAL_PROMPT", "0");
99 apply_hardened_env(&mut cmd);
100 cmd
101}
102
103pub fn apply_hardened_env(cmd: &mut tokio::process::Command) {
104 cmd.env("GIT_CONFIG_NOSYSTEM", "1")
98105 .env("GIT_NO_REPLACE_OBJECTS", "1");
99106 #[cfg(unix)]
100107 cmd.env("GIT_CONFIG_GLOBAL", "/dev/null")
101108 .env("GIT_CONFIG_SYSTEM", "/dev/null");
102 cmd
103109}
Modified dvcs/gitsync/post_receive.rs View dvcs/gitsync/post_receive.rs
@@ -25,6 +25,10 @@
2525pub fn ensure_post_receive_hook(bare: &Path) -> io::Result<()> {
2626 let path = hook_path(bare);
2727 if path.exists() {
28 tracing::error!(
29 path = %path.display(),
30 "existing post-receive hook will not be overwritten; post-receive sync may not capture pushes"
31 );
2832 return Ok(());
2933 }
3034 let hooks_dir = path.parent().expect("hook path always has a parent");
Modified dvcs/gitsync/stats.rs View dvcs/gitsync/stats.rs
@@ -1,6 +1,6 @@
11use db::repo::repos;
22use git::gc::impls::GcOps;
3use sea_orm::{ActiveModelTrait, IntoActiveModel, Set};
3use sea_orm::{ActiveModelTrait, ColumnTrait, EntityTrait, IntoActiveModel, QueryFilter, Set};
44
55use crate::error::{SyncError, SyncResult};
66
@@ -19,9 +19,18 @@
1919
2020 let mut am = repo.clone().into_active_model();
2121 am.git_size = Set(stats.git_size as i64);
22 am.nums_commits = Set(repo.nums_commits + new_commits as i32);
2322 am.nums_branches = Set(branch_count as i32);
2423 am.nums_tags = Set(tag_count as i32);
24
25 // 事务内重读最新 commit 计数，避免用加锁前加载的陈旧基线覆盖并发写入（lost update）
26 let fresh = repos::Entity::find_by_id(repo.id)
27 .filter(repos::Column::Id.eq(repo.id))
28 .one(txn)
29 .await
30 .map_err(SyncError::Db)?
31 .ok_or_else(|| SyncError::NotFound(format!("repo {}", repo.id)))?;
32 am.nums_commits = Set(fresh.nums_commits + new_commits as i32);
33
2534 am.update(txn).await.map_err(SyncError::Db)?;
2635
2736 Ok(())
Modified dvcs/gitsync/sync.rs View dvcs/gitsync/sync.rs
@@ -16,6 +16,19 @@
1616use crate::{branches, languages, stats, tags};
1717use config::git::SyncSettings;
1818
19/// panic/unwind 时也必须停止心跳任务，否则续租循环会永久持有 sync 锁
20struct HeartbeatGuard {
21 handle: tokio::task::JoinHandle<()>,
22}
23
24impl Drop for HeartbeatGuard {
25 fn drop(&mut self) {
26 if !self.handle.is_finished() {
27 self.handle.abort();
28 }
29 }
30}
31
1932#[derive(Clone)]
2033pub struct RepoSync {
2134 db: Db,
@@ -64,7 +77,7 @@
6477 };
6578 let lock = Arc::new(lock);
6679
67 let (tx, mut rx) = watch::channel(());
80 let (tx, mut rx) = watch::channel(false);
6881 let heartbeat_lock = lock.clone();
6982 let heartbeat = tokio::spawn(async move {
7083 let interval = (ttl / 3).max(1);
@@ -73,25 +86,27 @@
7386 match heartbeat_lock.refresh(ttl).await {
7487 Ok(true) => {}
7588 Ok(false) | Err(_) => {
76 let _ = tx.send(());
89 // watch::channel 只在值变化时通知；发送变化值而非 ()
90 let _ = tx.send(true);
7791 break;
7892 }
7993 }
8094 }
8195 });
96 let heartbeat = HeartbeatGuard { handle: heartbeat };
8297
8398 let result = tokio::select! {
8499 r = self.sync_work() => r,
85100 _ = rx.changed() => Err(SyncError::Locked),
86101 };
87
88 heartbeat.abort();
89102
90103 let locked = result.as_ref().err().is_some_and(SyncError::is_locked);
91104 if !locked {
92105 lock.release().await?;
93106 }
94107
108 drop(heartbeat);
109
95110 result
96111 }
97112
Modified dvcs/gitsync/webhook.rs View dvcs/gitsync/webhook.rs
@@ -13,7 +13,7 @@
1313#[async_trait::async_trait]
1414impl WebhookTx for NoopWebhookTx {
1515 async fn dispatch(&self, task: &PostReceiveTask) -> Result<(), Box<dyn Error + Send + Sync>> {
16 tracing::debug!(repo_id = %task.repo_id, refs = task.refs.len(), "webhook dispatch placeholder (not implemented)");
16 tracing::warn!(repo_id = %task.repo_id, refs = task.refs.len(), "webhook dispatch not configured; message dropped (NoopWebhookTx)");
1717 Ok(())
1818 }
1919}
Modified dvcs/gitsync/worker.rs View dvcs/gitsync/worker.rs
@@ -178,53 +178,57 @@
178178 }
179179 Err(e) if e.is_locked() => {
180180 metrics::counter!("gitsync.tasks_total", "result" => "locked").increment(1);
181 requeue_on_lock(queue, kv, settings, &task).await;
182 Ok(())
181 requeue_or_dlq(queue, kv, settings, &task)
182 .await
183 .map_err(|e| queue_err_str(&e))
183184 }
184185 Err(e) => {
185186 tracing::error!(repo_id = %task.repo_id, error = %e, "sync task failed");
186187 metrics::counter!("gitsync.tasks_total", "result" => "failure").increment(1);
187 Err(queue_err_str(&e))
188 // 重试循环只能由队列层驱动（返回 Err），此处将任务重新入队或送 DLQ；
189 // 若连入队都失败，返回 Err 让 consumer 层重试，避免静默丢任务
190 requeue_or_dlq(queue, kv, settings, &task)
191 .await
192 .map_err(|e| queue_err_str(&e))
188193 }
189194 }
190195}
191196
192async fn requeue_on_lock(
197/// 将失败任务重新入队；超过重试上限则送入 DLQ。
198/// 此前仅在日志中记录失败（任务被 ack 后永久丢失），现改为显式返回错误。
199async fn requeue_or_dlq(
193200 queue: &Queue,
194201 kv: &Arc<RedisCache>,
195202 settings: &SyncSettings,
196203 task: &PostReceiveTask,
197) {
204) -> SyncResult<()> {
198205 let key = format!("{ATTEMPT_KEY_PREFIX}{}", task.id);
199 let attempts = match kv.incr_with_ttl(&key, ATTEMPT_TTL_SECS).await {
200 Ok(n) => n,
201 Err(e) => {
202 tracing::warn!(error = %e, "failed to record lock retry attempt");
203 1
204 }
205 };
206 let attempts = kv.incr_with_ttl(&key, ATTEMPT_TTL_SECS).await?;
206207
207208 if attempts > u64::from(settings.max_lock_retries) {
208209 let _ = kv.delete(&key).await;
209210 match crate::event::enqueue_post_receive_dlq(queue, settings, task).await {
210211 Ok(()) => {
211212 metrics::counter!("gitsync.tasks_total", "result" => "dlq").increment(1);
212 tracing::warn!(repo_id = %task.repo_id, attempts, "task exceeded lock retry limit; sent to DLQ");
213 tracing::warn!(repo_id = %task.repo_id, attempts, "task exceeded retry limit; sent to DLQ");
214 Ok(())
213215 }
214216 Err(e) => {
215 tracing::error!(repo_id = %task.repo_id, error = %e, "failed to route task to DLQ")
217 metrics::counter!("gitsync.tasks_total", "result" => "dlq_failed").increment(1);
218 Err(SyncError::Queue(e))
216219 }
217220 }
218 return;
219 }
220
221 tokio::time::sleep(std::time::Duration::from_secs(5)).await;
222 match crate::event::enqueue_post_receive(queue, settings, task).await {
223 Ok(()) => {
224 tracing::debug!(repo_id = %task.repo_id, attempts, "re-queued lock-contended sync task")
225 }
226 Err(e) => {
227 tracing::error!(repo_id = %task.repo_id, error = %e, "failed to re-queue sync task")
221 } else {
222 tokio::time::sleep(std::time::Duration::from_secs(5)).await;
223 match crate::event::enqueue_post_receive(queue, settings, task).await {
224 Ok(()) => {
225 tracing::debug!(repo_id = %task.repo_id, attempts, "re-queued retry task");
226 Ok(())
227 }
228 Err(e) => {
229 metrics::counter!("gitsync.tasks_total", "result" => "requeue_failed").increment(1);
230 Err(SyncError::Queue(e))
231 }
228232 }
229233 }
230234}