use crate::error::GitError; use crate::http::HttpAppState; use crate::http::auth::authorize_repo_access; use crate::http::auth::verify_access_token; use crate::http::handler::is_valid_lfs_oid; use crate::http::lfs::{BatchRequest, CreateLockRequest, LfsHandler}; use crate::http::utils::{extract_basic_credentials, get_repo_model}; use crate::ssh::authz::SshAuthService; use crate::ssh::push_queue::{ PushQueueEvent, PushQueueLease, PushQueueWaitError, wait_for_push_queue_slot, }; use actix_web::{Error, HttpRequest, HttpResponse, web}; use argon2::Argon2; use argon2::password_hash::{PasswordHash, PasswordVerifier}; use models::repos::repo; use models::users::{user, user_token}; use sea_orm::prelude::*; use std::path::PathBuf; fn base_url(req: &HttpRequest) -> String { let conn_info = req.connection_info(); format!("{}://{}", conn_info.scheme(), conn_info.host()) } fn bearer_token(req: &HttpRequest) -> Result { let auth_header = req .headers() .get("authorization") .ok_or_else(|| actix_web::error::ErrorUnauthorized("Missing authorization header"))? .to_str() .map_err(|_| actix_web::error::ErrorUnauthorized("Invalid authorization header"))?; if let Some(token) = auth_header.strip_prefix("Bearer ") { Ok(token.to_string()) } else { Err(actix_web::error::ErrorUnauthorized( "Invalid authorization format", )) } } /// Derive the acting user from the authenticated bearer token. async fn user_uid(req: &HttpRequest, db: &db::database::AppDatabase) -> Result { if let Ok((username, access_key)) = extract_basic_credentials(req) { return verify_access_token(db, &username, &access_key) .await .map(|user| user.uid); } let token = bearer_token(req)?; find_user_by_bearer_token(&token, db).await } /// Store LFS batch-generated token in Redis with TTL. /// Key format: `lfs:token:{token}` → `{repo_id}:{user_uid}:{operation}` pub async fn store_lfs_token( cache: &db::cache::AppCache, token: &str, repo_id: uuid::Uuid, user_uid: uuid::Uuid, operation: &str, ) { if let Ok(mut conn) = cache.conn().await { use redis::AsyncCommands; let value = format!("{}:{}:{}", repo_id, user_uid, operation); let _: () = conn .set_ex(format!("lfs:token:{}", token), value, 3600_u64) .await .map_err(|e| tracing::warn!(error = %e, "failed to store lfs token")) .unwrap_or(()); } } /// Validate a bearer token for LFS upload/download. /// Checks two sources: /// 1. LFS batch-generated token (stored in Redis under `lfs:token:{token}`) /// 2. Regular user access token (stored in user_token table) /// /// Returns (user_uid, repo_id, operation) for batch tokens, or /// (user_uid, None, None) for regular access tokens. async fn validate_lfs_token( token: &str, cache: &db::cache::AppCache, db: &db::database::AppDatabase, expected_repo_id: uuid::Uuid, expected_operation: &str, ) -> Result { // First: check if it's a LFS batch token in Redis if let Ok(mut conn) = cache.conn().await { use redis::AsyncCommands; let stored: Option = conn .get::(format!("lfs:token:{}", token)) .await .ok(); if let Some(value) = stored { let parts: Vec<&str> = value.split(':').collect(); if parts.len() == 3 { let repo_id = uuid::Uuid::parse_str(parts[0]) .map_err(|_| actix_web::error::ErrorUnauthorized("Invalid batch token"))?; let user_uid = uuid::Uuid::parse_str(parts[1]) .map_err(|_| actix_web::error::ErrorUnauthorized("Invalid batch token"))?; let operation = parts[2]; if repo_id != expected_repo_id { return Err(actix_web::error::ErrorUnauthorized( "Token not valid for this repo", )); } if operation != expected_operation { return Err(actix_web::error::ErrorUnauthorized( "Token not valid for this operation", )); } // Consume the token (one-time use) let _: Result<(), redis::RedisError> = conn.del(format!("lfs:token:{}", token)).await; return Ok(user_uid); } } } // Second: check if it's a regular user access token. find_user_by_bearer_token(token, db).await } async fn find_user_by_bearer_token( token: &str, db: &db::database::AppDatabase, ) -> Result { let tokens = user_token::Entity::find() .filter(user_token::Column::IsRevoked.eq(false)) .all(db.reader()) .await .map_err(|_| actix_web::error::ErrorUnauthorized("Authentication failed"))?; for token_model in tokens { if token_model .expires_at .map(|expires_at| expires_at < chrono::Utc::now()) .unwrap_or(false) { continue; } let Ok(hash) = PasswordHash::new(&token_model.token_hash) else { tracing::warn!( token_id = token_model.id, "invalid stored bearer token hash" ); continue; }; if Argon2::default() .verify_password(token.as_bytes(), &hash) .is_ok() { return Ok(token_model.user); } } Err(actix_web::error::ErrorUnauthorized("Invalid token")) } async fn authorize_user_repo_access( db: &db::database::AppDatabase, user_uid: uuid::Uuid, repo: &repo::Model, is_write: bool, ) -> Result<(), Error> { let user = user::Entity::find() .filter(user::Column::Uid.eq(user_uid)) .one(db.reader()) .await .map_err(|_| actix_web::error::ErrorUnauthorized("Authentication failed"))? .ok_or_else(|| actix_web::error::ErrorUnauthorized("Invalid token user"))?; let authz = SshAuthService::new(db.clone()); if authz.check_repo_permission(&user, repo, is_write).await { Ok(()) } else { Err(actix_web::error::ErrorForbidden( "No permission for repository", )) } } async fn acquire_lfs_write_queue( state: &HttpAppState, repo: &repo::Model, operation: &'static str, ) -> Result { match wait_for_push_queue_slot(state.sync.clone(), repo.id, |event, request_id| { let request_id = request_id.to_string(); match event { PushQueueEvent::Waiting(position) => { tracing::info!( repo = %repo.repo_name, repo_id = %repo.id, request_id = %request_id, operation = operation, position = position.position, total = position.total, "lfs_write_queue_waiting" ); } PushQueueEvent::Acquired => { tracing::info!( repo = %repo.repo_name, repo_id = %repo.id, request_id = %request_id, operation = operation, "lfs_write_queue_acquired" ); } } }) .await { Ok(lease) => Ok(lease), Err(PushQueueWaitError::Join(e)) => { tracing::error!( error = %e, repo = %repo.repo_name, repo_id = %repo.id, operation = operation, "lfs_write_queue_join_failed" ); Err(actix_web::error::ErrorServiceUnavailable( "GitData: LFS write queue is temporarily unavailable. Please retry later.", )) } Err(PushQueueWaitError::Lock(e)) => { tracing::error!( error = %e, repo = %repo.repo_name, repo_id = %repo.id, operation = operation, "lfs_write_queue_lock_failed" ); Err(actix_web::error::ErrorServiceUnavailable( "GitData: LFS write queue lock failed. Please retry later.", )) } Err(PushQueueWaitError::Timeout) => { tracing::warn!( repo = %repo.repo_name, repo_id = %repo.id, operation = operation, "lfs_write_queue_timeout" ); Err(actix_web::error::ErrorServiceUnavailable( "GitData: LFS write queue timed out. Please retry in a moment.", )) } } } pub async fn lfs_batch( req: HttpRequest, path: web::Path<(String, String)>, body: web::Json, state: web::Data, ) -> Result { let (namespace, repo_name) = path.into_inner(); let batch_req = body.into_inner(); let is_write = batch_req.operation == "upload"; let repo = get_repo_model(&namespace, &repo_name, &state.db).await?; // Auth check: private repos always require auth; upload always requires auth if repo.is_private || is_write { let uid = user_uid(&req, &state.db).await?; authorize_repo_access(&req, &state.db, &repo, is_write).await?; let handler = LfsHandler::new(PathBuf::from(&repo.storage_path), repo, state.db.clone()); let response = handler .batch_with_auth(batch_req, &base_url(&req), uid, &state.cache) .await .map_err(|_| actix_web::error::ErrorInternalServerError("LFS batch failed"))?; Ok(HttpResponse::Ok() .content_type("application/vnd.git-lfs+json") .json(response)) } else { // Public repo + download: allow anonymous let handler = LfsHandler::new(PathBuf::from(&repo.storage_path), repo, state.db.clone()); let response = handler .batch(batch_req, &base_url(&req)) .await .map_err(|_| actix_web::error::ErrorInternalServerError("LFS batch failed"))?; Ok(HttpResponse::Ok() .content_type("application/vnd.git-lfs+json") .json(response)) } } pub async fn lfs_upload( req: HttpRequest, path: web::Path<(String, String, String)>, payload: web::Payload, state: web::Data, ) -> Result { let (namespace, repo_name, oid) = path.into_inner(); if !is_valid_lfs_oid(&oid) { return Err(actix_web::error::ErrorBadRequest("Invalid OID format")); } let repo = get_repo_model(&namespace, &repo_name, &state.db).await?; let token = bearer_token(&req)?; // Validate token (batch token or user access token) with write permission let uid = validate_lfs_token(&token, &state.cache, &state.db, repo.id, "upload").await?; authorize_user_repo_access(&state.db, uid, &repo, true).await?; let handler = LfsHandler::new(PathBuf::from(&repo.storage_path), repo, state.db.clone()); let mut queue_lease = acquire_lfs_write_queue(&state, &handler.model, "upload").await?; let result = match handler.upload_object(&oid, payload).await { Ok(response) => Ok(response), Err(GitError::InvalidOid(_)) => Err(actix_web::error::ErrorBadRequest("Invalid OID")), Err(GitError::AuthFailed(_)) => Err(actix_web::error::ErrorUnauthorized("Unauthorized")), Err(_e) => Err(actix_web::error::ErrorInternalServerError("Upload failed")), }; queue_lease.release().await; result } pub async fn lfs_download( req: HttpRequest, path: web::Path<(String, String, String)>, state: web::Data, ) -> Result { let (namespace, repo_name, oid) = path.into_inner(); if !is_valid_lfs_oid(&oid) { return Err(actix_web::error::ErrorBadRequest("Invalid OID format")); } let repo = get_repo_model(&namespace, &repo_name, &state.db).await?; // Auth check: private repos require auth; public repos allow anonymous if repo.is_private { let token = bearer_token(&req)?; let uid = validate_lfs_token(&token, &state.cache, &state.db, repo.id, "download").await?; authorize_user_repo_access(&state.db, uid, &repo, false).await?; } let handler = LfsHandler::new(PathBuf::from(&repo.storage_path), repo, state.db.clone()); match handler.download_object(&oid).await { Ok(response) => Ok(response), Err(GitError::NotFound(_)) => Err(actix_web::error::ErrorNotFound("Object not found")), Err(GitError::AuthFailed(_)) => Err(actix_web::error::ErrorUnauthorized("Unauthorized")), Err(_e) => Err(actix_web::error::ErrorInternalServerError( "Download failed", )), } } pub async fn lfs_lock_create( req: HttpRequest, path: web::Path<(String, String)>, body: web::Json, state: web::Data, ) -> Result { let (namespace, repo_name) = path.into_inner(); let repo = get_repo_model(&namespace, &repo_name, &state.db).await?; let uid = user_uid(&req, &state.db).await?; authorize_repo_access(&req, &state.db, &repo, true).await?; let handler = LfsHandler::new(PathBuf::from(&repo.storage_path), repo, state.db.clone()); let mut queue_lease = acquire_lfs_write_queue(&state, &handler.model, "lock_create").await?; let result = match handler.lock_object(&body.oid, uid).await { Ok(lock) => Ok(HttpResponse::Created().json(lock)), Err(GitError::Locked(msg)) => Ok(HttpResponse::Conflict().body(msg)), Err(_e) => Err(actix_web::error::ErrorInternalServerError("Lock failed")), }; queue_lease.release().await; result } pub async fn lfs_lock_list( req: HttpRequest, path: web::Path<(String, String)>, query: web::Query>, state: web::Data, ) -> Result { let (namespace, repo_name) = path.into_inner(); let repo = get_repo_model(&namespace, &repo_name, &state.db).await?; // Auth check: private repos require auth for lock listing if repo.is_private { let uid = user_uid(&req, &state.db).await?; authorize_user_repo_access(&state.db, uid, &repo, false).await?; } let maybe_oid = query.get("oid").map(|s| s.as_str()); let handler = LfsHandler::new(PathBuf::from(&repo.storage_path), repo, state.db.clone()); match handler.list_locks(maybe_oid).await { Ok(list) => Ok(HttpResponse::Ok().json(list)), Err(_e) => Err(actix_web::error::ErrorInternalServerError( "Lock list failed", )), } } pub async fn lfs_lock_get( req: HttpRequest, path: web::Path<(String, String, String)>, state: web::Data, ) -> Result { let (namespace, repo_name, lock_path) = path.into_inner(); let repo = get_repo_model(&namespace, &repo_name, &state.db).await?; // Auth check: private repos require auth for lock viewing if repo.is_private { let uid = user_uid(&req, &state.db).await?; authorize_user_repo_access(&state.db, uid, &repo, false).await?; } let handler = LfsHandler::new(PathBuf::from(&repo.storage_path), repo, state.db.clone()); match handler.get_lock(&lock_path).await { Ok(lock) => Ok(HttpResponse::Ok().json(lock)), Err(GitError::NotFound(_)) => Err(actix_web::error::ErrorNotFound("Lock not found")), Err(_e) => Err(actix_web::error::ErrorInternalServerError( "Lock get failed", )), } } pub async fn lfs_lock_delete( req: HttpRequest, path: web::Path<(String, String, String)>, state: web::Data, ) -> Result { let (namespace, repo_name, lock_id) = path.into_inner(); let repo = get_repo_model(&namespace, &repo_name, &state.db).await?; let uid = user_uid(&req, &state.db).await?; authorize_repo_access(&req, &state.db, &repo, true).await?; let handler = LfsHandler::new(PathBuf::from(&repo.storage_path), repo, state.db.clone()); let mut queue_lease = acquire_lfs_write_queue(&state, &handler.model, "lock_delete").await?; let result = match handler.unlock_object(&lock_id, uid).await { Ok(()) => Ok(HttpResponse::NoContent().finish()), Err(GitError::PermissionDenied(_)) => Err(actix_web::error::ErrorForbidden("Not allowed")), Err(GitError::NotFound(_)) => Err(actix_web::error::ErrorNotFound("Lock not found")), Err(_e) => Err(actix_web::error::ErrorInternalServerError( "Lock delete failed", )), }; queue_lease.release().await; result }