add pack size check in decode

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"

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,17 +83,29 @@ pub trait PackHandler: Send + Sync {
         &self,
         pack_config: &PackConfig,
         stream: Pin<Box<dyn Stream<Item = Result<Bytes, axum::Error>> + Send>>,
-    ) -> Result<Receiver<Entry>, GitError> {
+    ) -> Result<Receiver<Entry>, ProtocolError> {
         let (sender, receiver) = std::sync::mpsc::channel();
         let p = Pack::new(
             None,
             Some(1024 * 1024 * 1024 * pack_config.pack_decode_mem_size),
             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)
     }
 

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,10 +231,10 @@ 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();
+                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.
@@ -247,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());

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

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

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"

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"] }

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

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

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"

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,35 @@ 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<Item = Result<Bytes, Error>> + Unpin + Send + 'static,
+                               pack_limit: usize,
                                sender: Sender<Entry>)
-        -> Self
+        -> (tokio::task::JoinHandle<Pack>, tokio::task::JoinHandle<Result<(), ProtocolError>>)
     {
         let (tx, rx): (Sender<Vec<u8>>, Receiver<Vec<u8>>) = 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 {
                 let data = chunk.unwrap().to_vec();
+                total_size += data.len();
+                if total_size > pack_limit {
+                    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)
-        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 +745,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 +761,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);
     }
 

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"

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

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<MRDetail>(
         {
             status: "",
@@ -31,6 +32,12 @@ export default function MRDetailPage({ params }: { params: { id: string } }) {
     const [filedata, setFileData] = useState([]);
     const [loadings, setLoadings] = useState<boolean[]>([]);
 
+
+    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 } }) {
                 <Button
                     loading={loadings[1]}
                     onClick={() => approve_mr()}
+                    disabled={!login}
                 >
                     Merge MR
                 </Button>

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

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