563bb99
use axum::body::Body;
use axum::http::{HeaderMap, Method, StatusCode, header};
use axum::response::{IntoResponse, Response};
use futures::StreamExt as _;
use serde_json::json;
use tokio_util::io::StreamReader;
use crate::lfs::types::{
LfsBatchRequest, LfsLockCreateRequest, LfsOperation, LfsVerifyLocksRequest,
};
use crate::lfs::{lock, object};
use crate::share::access::{BasicAuth, allow_repo_read, require_writer};
use crate::smart::error::{SmartError, SmartResult};
use crate::smart::service::SmartService;
const MAX_LFS_BODY: usize = 64 * 1024 * 1024;
#[allow(clippy::too_many_arguments)]
pub async fn dispatch(
service: &SmartService,
space: &str,
repo: &str,
headers: &HeaderMap,
path: &str,
method: Method,
body: Body,
auth: &BasicAuth,
) -> SmartResult<Response> {
tracing::debug!(space, repo, path, method = %method, "lfs request received");
match path {
"info/lfs/objects/batch" if method == Method::POST => {
let bytes = read_body(headers, body, MAX_LFS_BODY).await?;
let req: LfsBatchRequest = serde_json::from_slice(&bytes)
.map_err(|e| SmartError::BadRequest(format!("invalid batch request: {e}")))?;
batch(service, space, repo, req, auth).await
}
"info/lfs/verify" if method == Method::POST => {
let bytes = read_body(headers, body, MAX_LFS_BODY).await?;
verify(service, space, repo, bytes, auth).await
}
p if p.starts_with("info/lfs/objects/") => {
let oid = &p["info/lfs/objects/".len()..];
match method {
Method::PUT => {
upload(service, space, repo, oid.to_string(), headers, body, auth).await
}
Method::GET => download(service, space, repo, oid, auth).await,
_ => Ok(SmartError::BadRequest("method not allowed".into()).into_response()),
}
}
"info/lfs/locks" if method == Method::GET => list_locks(service, space, repo, auth).await,
"info/lfs/locks" if method == Method::POST => {
create_lock(service, space, repo, body, auth).await
}
"info/lfs/locks/verify" if method == Method::POST => {
verify_locks(service, space, repo, body, auth).await
}
p if p.starts_with("info/lfs/locks/")
&& p.ends_with("/unlock")
&& method == Method::POST =>
{
let id = &p["info/lfs/locks/".len()..p.len() - "/unlock".len()];
unlock(service, space, repo, id.to_string(), auth).await
}
_ => Ok(SmartError::NotFound("lfs endpoint not found".into()).into_response()),
}
}
async fn batch(
service: &SmartService,
space: &str,
repo: &str,
req: LfsBatchRequest,
auth: &BasicAuth,
) -> SmartResult<Response> {
let svc = &service.git;
let ctx = service.resolve(space, repo).await?;
match req.operation {
LfsOperation::Upload | LfsOperation::Verify => {
require_writer(svc, auth, &ctx).await?;
}
LfsOperation::Download => {
allow_repo_read(svc, auth, &ctx).await?;
}
}
tracing::debug!(
space,
repo,
operation = ?req.operation,
objects = req.objects.len(),
"lfs batch authorized"
);
let resp = object::batch(svc, space.to_string(), repo.to_string(), req).await?;
tracing::debug!(space, repo, "lfs batch processed");
Ok((
StatusCode::OK,
[(header::CONTENT_TYPE, "application/vnd.git-lfs+json")],
serde_json::to_vec(&resp).map_err(|e| SmartError::Internal(format!("serialize: {e}")))?,
)
.into_response())
}
async fn verify(
service: &SmartService,
space: &str,
repo: &str,
bytes: bytes::Bytes,
auth: &BasicAuth,
) -> SmartResult<Response> {
let svc = &service.git;
let req: crate::lfs::types::LfsVerifyRequest = serde_json::from_slice(&bytes)
.map_err(|e| SmartError::BadRequest(format!("invalid verify request: {e}")))?;
let ctx = service.resolve(space, repo).await?;
require_writer(svc, auth, &ctx).await?;
object::verify(svc, space.to_string(), repo.to_string(), req).await?;
Ok(StatusCode::OK.into_response())
}
#[allow(clippy::too_many_arguments)]
async fn upload(
service: &SmartService,
space: &str,
repo: &str,
oid: String,
headers: &HeaderMap,
body: Body,
auth: &BasicAuth,
) -> SmartResult<Response> {
let svc = &service.git;
let ctx = service.resolve(space, repo).await?;
let uid = require_writer(svc, auth, &ctx).await?;
let canonical = crate::lfs::canonical_oid(&oid)?;
let size = headers
.get(header::CONTENT_LENGTH)
.and_then(|v| v.to_str().ok())
.and_then(|s| s.parse::<i64>().ok())
.unwrap_or(0);
let content_type = headers
.get(header::CONTENT_TYPE)
.and_then(|v| v.to_str().ok())
.map(str::to_string);
let stream = body
.into_data_stream()
.map(|chunk| chunk.map_err(|e| std::io::Error::other(e.to_string())));
let reader = StreamReader::new(stream);
tracing::debug!(space, repo, oid = %canonical, size, "lfs object upload started");
object::upload_stream(
svc,
space.to_string(),
repo.to_string(),
uid,
canonical,
size,
content_type,
reader,
)
.await?;
tracing::debug!(space, repo, "lfs object upload completed");
Ok(StatusCode::OK.into_response())
}
async fn download(
service: &SmartService,
space: &str,
repo: &str,
oid: &str,
auth: &BasicAuth,
) -> SmartResult<Response> {
let svc = &service.git;
let ctx = service.resolve(space, repo).await?;
allow_repo_read(svc, auth, &ctx).await?;
let canonical = crate::lfs::canonical_oid(oid)?;
let key = format!("lfs/{}/{}", ctx.model.id.simple(), canonical);
tracing::debug!(space, repo, oid = %canonical, "lfs object download started");
let stream = svc
.state
.storage
.client()
.get_stream(&key)
.await
.map_err(|e| SmartError::Internal(format!("lfs get: {e}")))?;
let body = Body::from_stream(
stream.map(|chunk| chunk.map_err(|e| -> axum::Error { axum::Error::new(e.to_string()) })),
);
Ok((
StatusCode::OK,
[(header::CONTENT_TYPE, "application/octet-stream")],
body,
)
.into_response())
}
async fn list_locks(
service: &SmartService,
space: &str,
repo: &str,
auth: &BasicAuth,
) -> SmartResult<Response> {
let svc = &service.git;
let ctx = service.resolve(space, repo).await?;
allow_repo_read(svc, auth, &ctx).await?;
let resp = lock::list(svc, space.to_string(), repo.to_string(), None, 100).await?;
Ok((
StatusCode::OK,
[(header::CONTENT_TYPE, "application/vnd.git-lfs+json")],
serde_json::to_vec(&resp).map_err(|e| SmartError::Internal(format!("serialize: {e}")))?,
)
.into_response())
}
async fn create_lock(
service: &SmartService,
space: &str,
repo: &str,
body: Body,
auth: &BasicAuth,
) -> SmartResult<Response> {
let svc = &service.git;
let ctx = service.resolve(space, repo).await?;
let uid = require_writer(svc, auth, &ctx).await?;
let bytes = axum::body::to_bytes(body, MAX_LFS_BODY)
.await
.map_err(|e| SmartError::BadRequest(format!("read body: {e}")))?;
let req: LfsLockCreateRequest = serde_json::from_slice(&bytes)
.map_err(|e| SmartError::BadRequest(format!("invalid lock request: {e}")))?;
let lock = crate::lfs::create_lock(
svc,
space.to_string(),
repo.to_string(),
uid,
req.path,
None,
)
.await?;
tracing::debug!(space, repo, lock_id = %lock.id, "lfs lock created");
Ok((
StatusCode::CREATED,
[(header::CONTENT_TYPE, "application/vnd.git-lfs+json")],
serde_json::to_vec(&json!({ "lock": lock }))
.map_err(|e| SmartError::Internal(format!("serialize: {e}")))?,
)
.into_response())
}
async fn verify_locks(
service: &SmartService,
space: &str,
repo: &str,
body: Body,
auth: &BasicAuth,
) -> SmartResult<Response> {
let svc = &service.git;
let ctx = service.resolve(space, repo).await?;
let uid = allow_repo_read(svc, auth, &ctx).await?;
let uid = uid.ok_or(SmartError::Unauthorized)?;
let bytes = axum::body::to_bytes(body, MAX_LFS_BODY)
.await
.map_err(|e| SmartError::BadRequest(format!("read body: {e}")))?;
let req: LfsVerifyLocksRequest = serde_json::from_slice(&bytes)
.map_err(|e| SmartError::BadRequest(format!("invalid request: {e}")))?;
let resp = lock::verify_locks(
svc,
space.to_string(),
repo.to_string(),
uid,
req.limit.unwrap_or(100) as u64,
)
.await?;
Ok((
StatusCode::OK,
[(header::CONTENT_TYPE, "application/vnd.git-lfs+json")],
serde_json::to_vec(&resp).map_err(|e| SmartError::Internal(format!("serialize: {e}")))?,
)
.into_response())
}
async fn unlock(
service: &SmartService,
space: &str,
repo: &str,
lock_id: String,
auth: &BasicAuth,
) -> SmartResult<Response> {
let svc = &service.git;
let ctx = service.resolve(space, repo).await?;
let uid = require_writer(svc, auth, &ctx).await?;
tracing::debug!(space, repo, lock_id = %lock_id, "lfs unlock requested");
crate::lfs::unlock_by_id(
svc,
space.to_string(),
repo.to_string(),
uid,
lock_id,
false,
)
.await?;
Ok((
StatusCode::OK,
[(header::CONTENT_TYPE, "application/vnd.git-lfs+json")],
serde_json::to_vec(&json!({ "message": "unlock successful" }))
.map_err(|e| SmartError::Internal(format!("serialize: {e}")))?,
)
.into_response())
}
async fn read_body(_headers: &HeaderMap, body: Body, limit: usize) -> SmartResult<bytes::Bytes> {
axum::body::to_bytes(body, limit)
.await
.map_err(|e| SmartError::BadRequest(format!("failed to read body: {e}")))
}