check repo exist in protocol

ceres/Cargo.toml

@@ -20,7 +20,6 @@ anyhow = { workspace = true }
 tokio = { workspace = true, features = ["net"] }
 tokio-stream = { workspace = true }
 axum = { workspace = true }
-async-stream = { workspace = true }
 tracing = { workspace = true }
 serde = { workspace = true }
 serde_json = { workspace = true }

ceres/src/lib.rs

@@ -1,5 +1,4 @@
 pub mod api_service;
-pub mod http;
 pub mod lfs;
 pub mod pack;
 pub mod protocol;

ceres/src/protocol/mod.rs

@@ -2,7 +2,10 @@ use core::fmt;
 use std::{path::PathBuf, str::FromStr, sync::Arc};
 
 use callisto::db_enums::RefType;
-use common::{errors::MegaError, utils::ZERO_ID};
+use common::{
+    errors::{MegaError, ProtocolError},
+    utils::ZERO_ID,
+};
 use jupiter::context::Context;
 use venus::{import_repo::import_refs::RefCommand, import_repo::repo::Repo};
 
@@ -16,8 +19,7 @@ pub struct SmartProtocol {
     pub capabilities: Vec<Capability>,
     pub path: PathBuf,
     pub command_list: Vec<RefCommand>,
-    // only needed in ssh protocal
-    pub service_type: ServiceType,
+    pub service_type: Option<ServiceType>,
     pub context: Context,
 }
 
@@ -125,7 +127,7 @@ impl SmartProtocol {
             capabilities: Vec::new(),
             path,
             command_list: Vec::new(),
-            service_type: ServiceType::ReceivePack,
+            service_type: None,
             context,
         }
     }
@@ -137,29 +139,35 @@ impl SmartProtocol {
             capabilities: Vec::new(),
             path: PathBuf::new(),
             command_list: Vec::new(),
-            service_type: ServiceType::ReceivePack,
+            service_type: None,
             context,
         }
     }
 
-    pub async fn pack_handler(&self) -> Arc<dyn PackHandler> {
+    pub async fn pack_handler(&self) -> Result<Arc<dyn PackHandler>, ProtocolError> {
         let import_dir = self.context.config.monorepo.import_dir.clone();
         if self.path.starts_with(import_dir.clone()) && self.path != import_dir {
             let storage = self.context.services.git_db_storage.clone();
-
             let path_str = self.path.to_str().unwrap();
             let model = storage.find_git_repo_exact_match(path_str).await.unwrap();
             let repo = if let Some(repo) = model {
                 repo.into()
             } else {
-                let repo = Repo::new(self.path.clone(), false);
-                storage.save_git_repo(repo.clone()).await.unwrap();
-                repo
+                match self.service_type.unwrap() {
+                    ServiceType::UploadPack => {
+                        return Err(ProtocolError::NotFound("Repository not found.".to_owned()))
+                    }
+                    ServiceType::ReceivePack => {
+                        let repo = Repo::new(self.path.clone(), false);
+                        storage.save_git_repo(repo.clone()).await.unwrap();
+                        repo
+                    }
+                }
             };
-            Arc::new(ImportRepo {
+            Ok(Arc::new(ImportRepo {
                 context: self.context.clone(),
                 repo,
-            })
+            }))
         } else {
             let mut res = MonoRepo {
                 context: self.context.clone(),
@@ -175,7 +183,7 @@ impl SmartProtocol {
                 res.from_hash = Some(command.old_id.clone());
                 res.to_hash = Some(command.new_id.clone());
             }
-            Arc::new(res)
+            Ok(Arc::new(res))
         }
     }
 }

ceres/src/protocol/smart.rs

@@ -6,6 +6,7 @@ use futures::Stream;
 use tokio_stream::wrappers::ReceiverStream;
 
 use callisto::db_enums::RefType;
+use common::errors::ProtocolError;
 
 use crate::protocol::ZERO_ID;
 use crate::protocol::{
@@ -57,10 +58,10 @@ impl SmartProtocol {
     /// Tracing information is logged regarding the response packet line stream.
     ///
     /// Finally, the constructed packet line stream is returned.
-    pub async fn git_info_refs(&self) -> BytesMut {
-        let pack_handler = self.pack_handler().await;
+    pub async fn git_info_refs(&self) -> Result<BytesMut, ProtocolError> {
+        let pack_handler = self.pack_handler().await?;
 
-        let service_type = self.service_type;
+        let service_type = self.service_type.unwrap();
 
         // The stream MUST include capability declarations behind a NUL on the first ref.
         let (head_hash, git_refs) = pack_handler.head_hash().await;
@@ -82,14 +83,14 @@ impl SmartProtocol {
         }
         let pkt_line_stream = self.build_smart_reply(&ref_list, service_type.to_string());
         tracing::debug!("git_info_refs response: {:?}", pkt_line_stream);
-        pkt_line_stream
+        Ok(pkt_line_stream)
     }
 
     pub async fn git_upload_pack(
         &mut self,
         upload_request: &mut Bytes,
-    ) -> Result<(ReceiverStream<Vec<u8>>, BytesMut)> {
-        let pack_handler = self.pack_handler().await;
+    ) -> Result<(ReceiverStream<Vec<u8>>, BytesMut), ProtocolError> {
+        let pack_handler = self.pack_handler().await?;
 
         let mut want: Vec<String> = Vec::new();
         let mut have: Vec<String> = Vec::new();
@@ -207,10 +208,10 @@ impl SmartProtocol {
     pub async fn git_receive_pack_stream(
         &mut self,
         data_stream: Pin<Box<dyn Stream<Item = Result<Bytes, axum::Error>> + Send>>,
-    ) -> Result<Bytes> {
+    ) -> Result<Bytes, ProtocolError> {
         // After receiving the pack data from the sender, the receiver sends a report
         let mut report_status = BytesMut::new();
-        let pack_handler = self.pack_handler().await;
+        let pack_handler = self.pack_handler().await?;
         //1. unpack progress
         let receiver = pack_handler
             .unpack_stream(&self.context.config.pack, data_stream)
@@ -223,7 +224,8 @@ impl SmartProtocol {
             let handle = tokio::runtime::Handle::current();
             handle.block_on(async { ph_clone.handle_receiver(receiver).await })
         })
-        .await.unwrap();
+        .await
+        .unwrap();
 
         // write "unpack ok\n to report"
         add_pkt_line_string(&mut report_status, "unpack ok\n".to_owned());

common/src/errors.rs

@@ -75,5 +75,16 @@ pub enum GitLFSError {
     GeneralError(String),
 }
 
+
+#[derive(Debug, Error)]
+pub enum ProtocolError {
+    #[error("{0}")]
+    IO(#[from] std::io::Error),
+    #[error("Authentication failed: {0}")]
+    Deny(String),
+    #[error("Repository not found: {0}")]
+    NotFound(String),
+}
+
 #[cfg(test)]
 mod tests {}

gateway/Cargo.toml

@@ -39,6 +39,8 @@ tower-http = { workspace = true, features = [
 ] }
 axum-extra = { workspace = true, features = ["typed-header"]}
 tokio = { workspace = true, features = ["net"] }
+tokio-stream = { workspace = true }
+async-stream = { workspace = true }
 reqwest = { workspace = true, features = ["json"] }
 uuid = { workspace = true, features = ["v4"] }
 regex = "1.10.4"

gateway/src/api/oauth/github.rs

@@ -22,8 +22,8 @@ const GITHUB_API_ENDPOINT: &str = "https://api.github.com";
 impl OauthHandler for GithubOauthService {
     fn authorize_url(&self, params: &AuthorizeParams, state: &str) -> String {
         let auth_url = format!(
-            "https://github.com/login/oauth/authorize?client_id={}&redirect_uri={}&state={}",
-            self.client_id, params.redirect_uri, state
+            "{}/login/oauth/authorize?client_id={}&redirect_uri={}&state={}",
+            GITHUB_ENDPOINT, self.client_id, params.redirect_uri, state
         );
         auth_url
     }

ceres/src/http/handler.rs → gateway/src/git_protocol/http.rs

@@ -1,18 +1,17 @@
-use std::collections::HashMap;
 use std::convert::Infallible;
 
 use anyhow::Result;
 use axum::body::Body;
-use axum::http::response::Builder;
-use axum::http::{Request, Response, StatusCode};
+use axum::http::{HeaderValue, Request, Response, StatusCode};
 use bytes::{Bytes, BytesMut};
+use common::errors::ProtocolError;
 use futures::{stream, TryStreamExt};
 use tokio::io::AsyncReadExt;
 use tokio_stream::StreamExt;
 
 use common::model::GetParams;
 
-use crate::protocol::{smart, ServiceType, SmartProtocol};
+use ceres::protocol::{smart, ServiceType, SmartProtocol};
 
 // # Discovering Reference
 // HTTP clients that support the "smart" protocol (or both the "smart" and "dumb" protocols) MUST
@@ -25,11 +24,18 @@ pub async fn git_info_refs(
     mut pack_protocol: SmartProtocol,
 ) -> Result<Response<Body>, (StatusCode, String)> {
     let service_name = params.service.unwrap();
-    pack_protocol.service_type = service_name.parse::<ServiceType>().unwrap();
-    let resp = build_res_header(format!("application/x-{}-advertisement", service_name));
-    let pkt_line_stream = pack_protocol.git_info_refs().await;
-    let body = Body::from(pkt_line_stream.freeze());
-    Ok(resp.body(body).unwrap())
+    pack_protocol.service_type = Some(service_name.parse::<ServiceType>().unwrap());
+
+    let response = match pack_protocol.git_info_refs().await {
+        Ok(pkt_line_stream) => Response::builder()
+            .body(Body::from(pkt_line_stream.freeze()))
+            .unwrap(),
+        Err(err) => return Ok(handler_err(err)),
+    };
+
+    let content_type = format!("application/x-{}-advertisement", service_name);
+    let response = add_default_header(content_type, response);
+    Ok(response)
 }
 
 /// # Handles a Git upload pack request and prepares the response.
@@ -66,37 +72,44 @@ pub async fn git_upload_pack(
         .await
         .unwrap();
     tracing::debug!("bytes from client: {:?}", upload_request);
-    let (mut send_pack_data, protocol_buf) = pack_protocol
+    let response = match pack_protocol
         .git_upload_pack(&mut upload_request.freeze())
         .await
-        .unwrap();
-
-    let resp = build_res_header("application/x-git-upload-pack-result".to_owned());
-
-    let body_stream = async_stream::stream! {
-        tracing::info!("send ack/nak message buf: {:?}", &protocol_buf);
-        yield Ok::<_, Infallible>(Bytes::copy_from_slice(&protocol_buf));
-        // send packdata with sideband64k
-        while let Some(chunk) = send_pack_data.next().await {
-            let mut reader = chunk.as_slice();
-            loop {
-                let mut temp = BytesMut::new();
-                temp.reserve(65500);
-                let length = reader.read_buf(&mut temp).await.unwrap();
-                if length == 0 {
-                    break;
+    {
+        Ok((mut send_pack_data, protocol_buf)) => {
+            let body_stream = async_stream::stream! {
+                tracing::info!("send ack/nak message buf: {:?}", &protocol_buf);
+                yield Ok::<_, Infallible>(Bytes::copy_from_slice(&protocol_buf));
+                // send packdata with sideband64k
+                while let Some(chunk) = send_pack_data.next().await {
+                    let mut reader = chunk.as_slice();
+                    loop {
+                        let mut temp = BytesMut::new();
+                        temp.reserve(65500);
+                        let length = reader.read_buf(&mut temp).await.unwrap();
+                        if length == 0 {
+                            break;
+                        }
+                        let bytes_out = pack_protocol.build_side_band_format(temp, length);
+                        // tracing::info!("send pack file: length: {:?}", bytes_out.len());
+                        yield Ok::<_, Infallible>(bytes_out.freeze());
+                    }
                 }
-                let bytes_out = pack_protocol.build_side_band_format(temp, length);
-                // tracing::info!("send pack file: length: {:?}", bytes_out.len());
-                yield Ok::<_, Infallible>(bytes_out.freeze());
-            }
+                let bytes_out = Bytes::from_static(smart::PKT_LINE_END_MARKER);
+                tracing::info!("send back pkt-flush line '0000', actually: {:?}", bytes_out);
+                yield Ok::<_, Infallible>(bytes_out);
+            };
+            Response::builder()
+                .body(Body::from_stream(body_stream))
+                .unwrap()
         }
-        let bytes_out = Bytes::from_static(smart::PKT_LINE_END_MARKER);
-        tracing::info!("send back pkt-flush line '0000', actually: {:?}", bytes_out);
-        yield Ok::<_, Infallible>(bytes_out);
+        Err(err) => return Ok(handler_err(err)),
     };
-    let resp = resp.body(Body::from_stream(body_stream)).unwrap();
-    Ok(resp)
+    let response = add_default_header(
+        String::from("application/x-git-upload-pack-result"),
+        response,
+    );
+    Ok(response)
 }
 
 /// Handles the Git receive-pack protocol for receiving and processing data from a client.
@@ -139,9 +152,12 @@ pub async fn git_receive_pack(
         }
     }
     tracing::info!("report status:{:?}", report_status);
-    let resp = build_res_header("application/x-git-receive-pack-result".to_owned());
-    let resp = resp.body(Body::from(report_status)).unwrap();
-    Ok(resp)
+    let response = Response::builder().body(Body::from(report_status)).unwrap();
+    let response = add_default_header(
+        String::from("application/x-git-receive-pack-result"),
+        response,
+    );
+    Ok(response)
 }
 
 // Function to find the subsequence in a slice
@@ -152,19 +168,34 @@ fn search_subsequence(chunk: &[u8], search: &[u8]) -> Option<usize> {
 /// # Build Response headers for Smart Server.
 /// Clients MUST NOT reuse or revalidate a cached response.
 /// Servers MUST include sufficient Cache-Control headers to prevent caching of the response.
-pub fn build_res_header(content_type: String) -> Builder {
-    let mut headers = HashMap::new();
-    headers.insert("Content-Type".to_string(), content_type);
-    headers.insert(
-        "Cache-Control".to_string(),
-        "no-cache, max-age=0, must-revalidate".to_string(),
+fn add_default_header<T>(content_type: String, mut response: Response<T>) -> Response<T> {
+    response.headers_mut().insert(
+        "Content-Type",
+        HeaderValue::from_str(&content_type).unwrap(),
+    );
+    response.headers_mut().insert(
+        "Cache-Control",
+        HeaderValue::from_static("no-cache, max-age=0, must-revalidate"),
     );
-    let mut resp = Response::builder();
+    response
+}
 
-    for (key, val) in headers {
-        resp = resp.header(&key, val);
+fn handler_err(err: ProtocolError) -> Response<Body> {
+    match err {
+        ProtocolError::NotFound(err) => Response::builder()
+            .status(StatusCode::NOT_FOUND)
+            .body(Body::from(err.to_string()))
+            .unwrap(),
+        ProtocolError::Deny(err) => Response::builder()
+            .status(StatusCode::UNAUTHORIZED)
+            .body(Body::from(err.to_string()))
+            .unwrap(),
+        _ => Response::builder()
+            .status(StatusCode::INTERNAL_SERVER_ERROR)
+            .body(Body::empty())
+            .unwrap(),
     }
-    resp
 }
+
 #[cfg(test)]
 mod tests {}

gateway/src/git_protocol/mod.rs

@@ -1 +1,2 @@
-pub mod ssh;
+pub mod ssh;
+pub mod http;