From 1a092624f88e40d2e4ba889338cfa417b4695e31 Mon Sep 17 00:00:00 2001 From: "benjamin.747" Date: Wed, 9 Oct 2024 14:40:30 +0800 Subject: [PATCH 1/3] add pack size check in decode --- Cargo.toml | 26 +++++++++++++------------- ceres/src/pack/mod.rs | 26 +++++++++++++++++++++----- ceres/src/protocol/smart.rs | 7 +++---- common/src/config.rs | 2 ++ common/src/errors.rs | 5 +++++ docker/config.toml | 4 ++++ mega/config.toml | 3 +++ mercury/src/internal/pack/decode.rs | 29 ++++++++++++++++++----------- mono/config.toml | 3 +++ mono/src/git_protocol/http.rs | 3 +-- 10 files changed, 73 insertions(+), 35 deletions(-) diff --git a/Cargo.toml b/Cargo.toml index 27e7bef2c..c2d6fa8a9 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -35,22 +35,22 @@ mega = { path = "mega" } mono = { path = "mono" } libra = { path = "libra" } -anyhow = "1.0.87" +anyhow = "1.0.89" serde = "1.0.210" serde_json = "1.0.128" tracing = "0.1.40" tracing-subscriber = "0.3.18" tracing-appender = "0.2" -thiserror = "1.0.63" +thiserror = "1.0.64" rand = "0.8.5" smallvec = "1.13.2" tokio = "1.40.0" tokio-stream = "0.1.16" tokio-test = "0.4.4" -clap = "4.5.17" -async-trait = "0.1.82" -async-stream = "0.3.5" -bytes = "1.7.1" +clap = "4.5.20" +async-trait = "0.1.83" +async-stream = "0.3.6" +bytes = "1.7.2" memchr = "2.7.4" chrono = "0.4.38" sha1 = "0.10.6" @@ -59,27 +59,27 @@ futures-util = "0.3.30" go-defer = "0.1.0" russh = "0.45.0" russh-keys = "0.45.0" -axum = "0.7.6" +axum = "0.7.7" axum-extra = "0.9.4" axum-server = "0.7.1" -tower-http = "0.6.0" +tower-http = "0.6.1" tower = "0.5.1" hex = "0.4.3" sea-orm = "1.0.1" -flate2 = "1.0.30" +flate2 = "1.0.34" bstr = "1.10.0" colored = "2.1.0" idgenerator = "2.0.0" num_cpus = "1.16.0" config = "0.14.0" -shadow-rs = "0.35.0" -reqwest = "0.12.7" +shadow-rs = "0.35.1" +reqwest = "0.12.8" lazy_static = "1.5.0" uuid = "1.10.0" -regex = "1.10.4" +regex = "1.11.0" ed25519-dalek = "2.1.1" ctrlc = "3.4.4" git2 = "0.19.0" -tempfile = "3.10.1" +tempfile = "3.13.0" home = "0.5.9" ring = "0.17.8" diff --git a/ceres/src/pack/mod.rs b/ceres/src/pack/mod.rs index b94afc688..dbf6a51a4 100644 --- a/ceres/src/pack/mod.rs +++ b/ceres/src/pack/mod.rs @@ -14,7 +14,11 @@ use tokio_stream::wrappers::ReceiverStream; use crate::protocol::import_refs::{RefCommand, Refs}; use callisto::raw_blob; -use common::{config::PackConfig, errors::MegaError, utils::ZERO_ID}; +use common::{ + config::PackConfig, + errors::{MegaError, ProtocolError}, + utils::ZERO_ID, +}; use mercury::internal::pack::Pack; use mercury::{ errors::GitError, @@ -79,7 +83,7 @@ pub trait PackHandler: Send + Sync { &self, pack_config: &PackConfig, stream: Pin> + Send>>, - ) -> Result, GitError> { + ) -> Result, ProtocolError> { let (sender, receiver) = std::sync::mpsc::channel(); let p = Pack::new( None, @@ -87,9 +91,21 @@ pub trait PackHandler: Send + Sync { Some(pack_config.pack_decode_cache_path.clone()), pack_config.clean_cache_after_decode, ); - tokio::spawn(async move { - p.decode_stream(stream, sender).await; - }); + let (unpack_handle, convert) = p + .decode_stream( + stream, + 1024 * 1024 * 1024 * pack_config.maximum_pack_size, + sender, + ) + .await; + match convert.await.unwrap() { + Ok(_) => (), + Err(err) => { + // not working in spawn_blocking? + unpack_handle.abort(); + return Err(err); + } + } Ok(receiver) } diff --git a/ceres/src/protocol/smart.rs b/ceres/src/protocol/smart.rs index 035b6c27a..a9ff2fea2 100644 --- a/ceres/src/protocol/smart.rs +++ b/ceres/src/protocol/smart.rs @@ -214,10 +214,9 @@ impl SmartProtocol { //1. unpack progress let receiver = pack_handler .unpack_stream(&self.context.config.pack, data_stream) - .await - .unwrap(); + .await?; - // can not block main thread here. + // do not block main thread here. let ph_clone = pack_handler.clone(); let unpack_result = tokio::task::spawn_blocking(move || { let handle = tokio::runtime::Handle::current(); @@ -232,7 +231,7 @@ impl SmartProtocol { let mut default_exist = pack_handler.check_default_branch().await; //2. update each refs and build report - for mut command in self.command_list.clone() { + for command in &mut self.command_list { if command.ref_type == RefType::Tag { // just update if refs type is tag pack_handler.update_refs(&command).await.unwrap(); diff --git a/common/src/config.rs b/common/src/config.rs index be1833c4a..61b26b131 100644 --- a/common/src/config.rs +++ b/common/src/config.rs @@ -227,6 +227,7 @@ pub struct PackConfig { pub pack_decode_cache_path: PathBuf, pub clean_cache_after_decode: bool, pub channel_message_size: usize, + pub maximum_pack_size: usize, } impl Default for PackConfig { @@ -236,6 +237,7 @@ impl Default for PackConfig { pack_decode_cache_path: PathBuf::from("/tmp/.mega/cache"), clean_cache_after_decode: true, channel_message_size: 1_000_000, + maximum_pack_size: 1, } } } diff --git a/common/src/errors.rs b/common/src/errors.rs index be3366af8..7bb306c7c 100644 --- a/common/src/errors.rs +++ b/common/src/errors.rs @@ -89,6 +89,8 @@ pub enum ProtocolError { Deny(String), #[error("Repository not found: {0}")] NotFound(String), + #[error("PackFile too large: {0}")] + TooLarge(String), #[error("Invalid Input: {0}")] InvalidInput(String), #[error("HTTP Push Has Been Disabled")] @@ -102,6 +104,9 @@ impl IntoResponse for ProtocolError { // This error is caused by bad user input so don't log it (StatusCode::UNAUTHORIZED, err) } + ProtocolError::TooLarge(err) => { + (StatusCode::PAYLOAD_TOO_LARGE, err) + } ProtocolError::NotFound(err) => { // Because `TraceLayer` wraps each request in a span that contains the request // method, uri, etc we don't need to include those details here diff --git a/docker/config.toml b/docker/config.toml index f9483ee5f..1469cd59f 100644 --- a/docker/config.toml +++ b/docker/config.toml @@ -67,6 +67,10 @@ clean_cache_after_decode = true # The maximum meesage size in channel buffer while decode channel_message_size = 1_000_000 +# Maximum pack size, unit GB, enforces to use LFS off the limit +maximum_pack_size = 1 + + [lfs] # LFS Server url url = "https://git.gitmono.com" diff --git a/mega/config.toml b/mega/config.toml index 303643207..87d8b1d2c 100644 --- a/mega/config.toml +++ b/mega/config.toml @@ -67,6 +67,9 @@ clean_cache_after_decode = true # The maximum meesage size in channel buffer while decode channel_message_size = 1_000_000 +# Maximum pack size, unit GB, enforces to use LFS off the limit +maximum_pack_size = 1 + [lfs] # LFS Server url url = "http://localhost:8000" diff --git a/mercury/src/internal/pack/decode.rs b/mercury/src/internal/pack/decode.rs index ac723c388..8c7de6ac3 100644 --- a/mercury/src/internal/pack/decode.rs +++ b/mercury/src/internal/pack/decode.rs @@ -8,6 +8,7 @@ use std::time::Instant; use axum::Error; use bytes::Bytes; +use common::errors::ProtocolError; use flate2::bufread::ZlibDecoder; use futures_util::{Stream, StreamExt}; use threadpool::ThreadPool; @@ -476,26 +477,34 @@ impl Pack { /// Decode `Pack` with inputting a `Stream` of `Bytes`, and send the `Entry` while decoding. pub async fn decode_stream(mut self, mut stream: impl Stream> + Unpin + Send + 'static, + pack_limit: usize, sender: Sender) - -> Self + -> (tokio::task::JoinHandle, tokio::task::JoinHandle>) { let (tx, rx): (Sender>, Receiver>) = mpsc::channel(); let mut reader = ChannelReader::new(rx); - tokio::spawn(async move { + let mut total_size = 0; + let convert_handle = tokio::spawn(async move { // use Channel to connect `async` & `sync` - while let Some(chunk) = stream.next().await { + Ok(while let Some(chunk) = stream.next().await { let data = chunk.unwrap().to_vec(); + total_size += data.len(); + if total_size > pack_limit { + eprintln!("Body size exceeded 1 GB limit. Terminating connection."); + return Err(ProtocolError::TooLarge(total_size.to_string())) + } tx.send(data).unwrap(); - } + }) }); // CPU-bound task, so use spawn_blocking // DO NOT use thread::spawn, because it will block tokio runtime (if single-threaded runtime, like in tests) - tokio::task::spawn_blocking(move || { + let unpack_handle = tokio::task::spawn_blocking(move || { self.decode(&mut reader, move |entry, _| { if sender.send(entry).is_ok() {} }).unwrap(); self - }).await.unwrap() + }); + (unpack_handle, convert_handle) } /// CacheObjects + Index size of Caches @@ -735,12 +744,10 @@ mod tests { let stream = ReaderStream::new(f).map_err(|e| { axum::Error::new(e) }); - let p = Pack::new(Some(20), Some(1024*1024*1024*2), Some(tmp.clone()), true); + let p = Pack::new(Some(20), Some(1024*1024*1024*4), Some(tmp.clone()), true); let (tx, rx) = std::sync::mpsc::channel(); - let handle = tokio::spawn(async move { - p.decode_stream(stream, tx).await - }); + let (pack, _ ) = p.decode_stream(stream, 1024 * 1024 * 1024, tx).await; let count = Arc::new(AtomicUsize::new(0)); let count_c = count.clone(); @@ -753,7 +760,7 @@ mod tests { tracing::info!("Received: {}", cnt); count_c.store(cnt, Ordering::Relaxed); }).await.unwrap(); - let p = handle.await.unwrap(); + let p = pack.await.unwrap(); assert_eq!(count.load(Ordering::Relaxed), p.number); } diff --git a/mono/config.toml b/mono/config.toml index 176d13178..4164f0223 100644 --- a/mono/config.toml +++ b/mono/config.toml @@ -67,6 +67,9 @@ clean_cache_after_decode = true # The maximum meesage size in channel buffer while decode channel_message_size = 1_000_000 +# Maximum pack size, unit GB, enforces to use LFS off the limit +maximum_pack_size = 1 + [lfs] # LFS Server url url = "http://localhost:8000" diff --git a/mono/src/git_protocol/http.rs b/mono/src/git_protocol/http.rs index 967c2fd95..1cd092e62 100644 --- a/mono/src/git_protocol/http.rs +++ b/mono/src/git_protocol/http.rs @@ -140,8 +140,7 @@ pub async fn git_receive_pack( let remaining_stream = stream::once(async { Ok(remaining_bytes) }).chain(data_stream); report_status = pack_protocol .git_receive_pack_stream(Box::pin(remaining_stream)) - .await - .unwrap(); + .await?; break; } } From 71907af74a4ba02836c4aac3db7616d50916bed9 Mon Sep 17 00:00:00 2001 From: "benjamin.747" Date: Wed, 9 Oct 2024 14:48:43 +0800 Subject: [PATCH 2/3] [ui] disable merge if not login in --- lunar/src-tauri/Cargo.toml | 9 ++++----- lunar/src-tauri/src/main.rs | 2 +- moon/src/app/(dashboard)/mr/[id]/page.tsx | 9 +++++++++ moon/src/app/api/auth/route.ts | 11 +++++++++++ moon/src/app/api/mr/[id]/merge/route.ts | 1 - moon/src/app/lib/dal.ts | 5 +++++ 6 files changed, 30 insertions(+), 7 deletions(-) create mode 100644 moon/src/app/api/auth/route.ts diff --git a/lunar/src-tauri/Cargo.toml b/lunar/src-tauri/Cargo.toml index b8fe42182..33995de7a 100644 --- a/lunar/src-tauri/Cargo.toml +++ b/lunar/src-tauri/Cargo.toml @@ -11,17 +11,16 @@ rust-version = "1.60" # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html [build-dependencies] -tauri-build = { version = "2", features = [] } +tauri-build = { version = "2.0.1", features = [] } [dependencies] serde_json = { workspace = true } serde = { workspace = true, features = ["derive"] } -tauri = { version = "2", features = [ -] } +tauri = { version = "2.0.2", features = [] } tokio = { workspace = true } home = { workspace = true } -tauri-plugin-fs = "2" -tauri-plugin-shell = "2" +tauri-plugin-fs = "2.0" +tauri-plugin-shell = "2.0" [dev-dependencies] tokio = { workspace = true, features = ["macros"] } diff --git a/lunar/src-tauri/src/main.rs b/lunar/src-tauri/src/main.rs index 2c137df47..bbbeb8946 100644 --- a/lunar/src-tauri/src/main.rs +++ b/lunar/src-tauri/src/main.rs @@ -89,7 +89,7 @@ fn start_mega_service( CommandEvent::Stdout(line) => { // line to string // print!("{}", line); - println!("Sidecar stdout: {}", String::from_utf8_lossy(&line)); + print!("Sidecar stdout: {}", String::from_utf8_lossy(&line)); } CommandEvent::Stderr(line) => { eprint!("Sidecar stderr: {}", String::from_utf8_lossy(&line)); diff --git a/moon/src/app/(dashboard)/mr/[id]/page.tsx b/moon/src/app/(dashboard)/mr/[id]/page.tsx index 7fc1f6262..253619199 100644 --- a/moon/src/app/(dashboard)/mr/[id]/page.tsx +++ b/moon/src/app/(dashboard)/mr/[id]/page.tsx @@ -21,6 +21,7 @@ interface Conversation { export default function MRDetailPage({ params }: { params: { id: string } }) { const [editorState, setEditorState] = useState(""); + const [login, setLogin] = useState(false); const [mrDetail, setMrDetail] = useState( { status: "", @@ -31,6 +32,12 @@ export default function MRDetailPage({ params }: { params: { id: string } }) { const [filedata, setFileData] = useState([]); const [loadings, setLoadings] = useState([]); + + const checkLogin = async () => { + const res = await fetch(`/api/auth`); + setLogin(res.ok); + }; + const fetchDetail = async () => { const detail = await fetch(`/api/mr/${params.id}/detail`); const detail_json = await detail.json(); @@ -51,6 +58,7 @@ export default function MRDetailPage({ params }: { params: { id: string } }) { useEffect(() => { fetchDetail() fetchFileList(); + checkLogin(); }, [params.id]); const set_to_loading = (index: number) => { @@ -149,6 +157,7 @@ export default function MRDetailPage({ params }: { params: { id: string } }) { diff --git a/moon/src/app/api/auth/route.ts b/moon/src/app/api/auth/route.ts new file mode 100644 index 000000000..34e87bf15 --- /dev/null +++ b/moon/src/app/api/auth/route.ts @@ -0,0 +1,11 @@ +import { isLoginIn } from '@/app/lib/dal'; + +export async function GET(request: Request, { params }: { params: { id: string } }) { + + const login = isLoginIn(); + + if (!login) { + return new Response(null, { status: 401 }) + } + return new Response(null, { status: 200 }) +}; \ No newline at end of file diff --git a/moon/src/app/api/mr/[id]/merge/route.ts b/moon/src/app/api/mr/[id]/merge/route.ts index 9b86e0556..532b1aaf4 100644 --- a/moon/src/app/api/mr/[id]/merge/route.ts +++ b/moon/src/app/api/mr/[id]/merge/route.ts @@ -6,7 +6,6 @@ export const dynamic = 'force-dynamic' // defaults to auto export async function POST(request: Request, { params }: { params: { id: string } }) { const session = await verifySession() - console.log("ssss", session); // Check if the user is authenticated if (!session) { // User is not authenticated diff --git a/moon/src/app/lib/dal.ts b/moon/src/app/lib/dal.ts index 1cf94a610..6aeb7f62d 100644 --- a/moon/src/app/lib/dal.ts +++ b/moon/src/app/lib/dal.ts @@ -12,3 +12,8 @@ export const verifySession = cache(async () => { return { session } }) + +export const isLoginIn = () => { + const session = cookies().get('SESSION')?.value + return session != null +} From 2a1f1d54a305ebac8124ae73d4a027e16e9fd6f7 Mon Sep 17 00:00:00 2001 From: "benjamin.747" Date: Wed, 9 Oct 2024 15:51:44 +0800 Subject: [PATCH 3/3] fix clippy err --- ceres/src/protocol/smart.rs | 4 ++-- lunar/src-tauri/build.rs | 4 ++-- mercury/src/internal/pack/decode.rs | 7 ++++--- 3 files changed, 8 insertions(+), 7 deletions(-) diff --git a/ceres/src/protocol/smart.rs b/ceres/src/protocol/smart.rs index a9ff2fea2..4063ef193 100644 --- a/ceres/src/protocol/smart.rs +++ b/ceres/src/protocol/smart.rs @@ -234,7 +234,7 @@ impl SmartProtocol { for command in &mut self.command_list { if command.ref_type == RefType::Tag { // just update if refs type is tag - pack_handler.update_refs(&command).await.unwrap(); + pack_handler.update_refs(command).await.unwrap(); } else { // Updates can be unsuccessful for a number of reasons. // a.The reference can have changed since the reference discovery phase was originally sent, meaning someone pushed in the meantime. @@ -246,7 +246,7 @@ impl SmartProtocol { command.default_branch = true; default_exist = true; } - pack_handler.update_refs(&command).await.unwrap(); + pack_handler.update_refs(command).await.unwrap(); } Err(ref err) => { command.failed(err.to_string()); diff --git a/lunar/src-tauri/build.rs b/lunar/src-tauri/build.rs index 11bc6c129..16172e51d 100644 --- a/lunar/src-tauri/build.rs +++ b/lunar/src-tauri/build.rs @@ -39,10 +39,10 @@ fn main() { }; std::fs::copy(format!("../../target/{}/mega{}", debug_path, extension), sidecar_path) - .expect("Run Cargo build for mega first"); + .expect("Run cargo build to build mega bin for Lunar first"); std::fs::copy(format!("../../target/{}/libra{}", debug_path, extension), libra_path) - .expect("Run Cargo build for libra first"); + .expect("Run cargo build to build libra bin for Lunar first"); // Copy libpipy due to target os #[cfg(target_os = "macos")] diff --git a/mercury/src/internal/pack/decode.rs b/mercury/src/internal/pack/decode.rs index 8c7de6ac3..cbed66fca 100644 --- a/mercury/src/internal/pack/decode.rs +++ b/mercury/src/internal/pack/decode.rs @@ -486,15 +486,16 @@ impl Pack { let mut total_size = 0; let convert_handle = tokio::spawn(async move { // use Channel to connect `async` & `sync` - Ok(while let Some(chunk) = stream.next().await { + while let Some(chunk) = stream.next().await { let data = chunk.unwrap().to_vec(); total_size += data.len(); if total_size > pack_limit { - eprintln!("Body size exceeded 1 GB limit. Terminating connection."); + eprintln!("Body size exceeded limit. Terminating connection."); return Err(ProtocolError::TooLarge(total_size.to_string())) } tx.send(data).unwrap(); - }) + } + Ok(()) }); // CPU-bound task, so use spawn_blocking // DO NOT use thread::spawn, because it will block tokio runtime (if single-threaded runtime, like in tests)