Inbound migrations work

src/api/repo/blob.rs

@@ -1,3 +1,4 @@
+use crate::auth::{ServiceTokenVerifier, is_service_token};
 use crate::state::AppState;
 use axum::body::Bytes;
 use axum::{
@@ -13,22 +14,16 @@ use serde::{Deserialize, Serialize};
 use serde_json::json;
 use sha2::{Digest, Sha256};
 use std::str::FromStr;
-use tracing::error;
+use tracing::{debug, error};
 
 const MAX_BLOB_SIZE: usize = 1_000_000;
+const MAX_VIDEO_BLOB_SIZE: usize = 100_000_000;
 
 pub async fn upload_blob(
     State(state): State<AppState>,
     headers: axum::http::HeaderMap,
     body: Bytes,
 ) -> Response {
-    if body.len() > MAX_BLOB_SIZE {
-        return (
-            StatusCode::PAYLOAD_TOO_LARGE,
-            Json(json!({"error": "BlobTooLarge", "message": format!("Blob size {} exceeds maximum of {} bytes", body.len(), MAX_BLOB_SIZE)})),
-        )
-            .into_response();
-    }
     let token = match crate::auth::extract_bearer_token_from_header(
         headers.get("Authorization").and_then(|h| h.to_str().ok()),
     ) {
@@ -41,17 +36,66 @@ pub async fn upload_blob(
                 .into_response();
         }
     };
-    let auth_user = match crate::auth::validate_bearer_token(&state.db, &token).await {
-        Ok(user) => user,
-        Err(_) => {
-            return (
-                StatusCode::UNAUTHORIZED,
-                Json(json!({"error": "AuthenticationFailed"})),
-            )
-                .into_response();
+
+    let is_service_auth = is_service_token(&token);
+
+    let (did, is_migration) = if is_service_auth {
+        debug!("Verifying service token for blob upload");
+        let verifier = ServiceTokenVerifier::new();
+        match verifier
+            .verify_service_token(&token, Some("com.atproto.repo.uploadBlob"))
+            .await
+        {
+            Ok(claims) => {
+                debug!("Service token verified for DID: {}", claims.iss);
+                (claims.iss, false)
+            }
+            Err(e) => {
+                error!("Service token verification failed: {:?}", e);
+                return (
+                    StatusCode::UNAUTHORIZED,
+                    Json(json!({"error": "AuthenticationFailed", "message": format!("Service token verification failed: {}", e)})),
+                )
+                    .into_response();
+            }
+        }
+    } else {
+        match crate::auth::validate_bearer_token_allow_deactivated(&state.db, &token).await {
+            Ok(user) => {
+                let deactivated = sqlx::query_scalar!(
+                    "SELECT deactivated_at FROM users WHERE did = $1",
+                    user.did
+                )
+                .fetch_optional(&state.db)
+                .await
+                .ok()
+                .flatten()
+                .flatten();
+                (user.did, deactivated.is_some())
+            }
+            Err(_) => {
+                return (
+                    StatusCode::UNAUTHORIZED,
+                    Json(json!({"error": "AuthenticationFailed"})),
+                )
+                    .into_response();
+            }
         }
     };
-    let did = auth_user.did;
+
+    let max_size = if is_service_auth || is_migration {
+        MAX_VIDEO_BLOB_SIZE
+    } else {
+        MAX_BLOB_SIZE
+    };
+
+    if body.len() > max_size {
+        return (
+            StatusCode::PAYLOAD_TOO_LARGE,
+            Json(json!({"error": "BlobTooLarge", "message": format!("Blob size {} exceeds maximum of {} bytes", body.len(), max_size)})),
+        )
+            .into_response();
+    }
     let mime_type = headers
         .get("content-type")
         .and_then(|h| h.to_str().ok())

src/api/repo/import.rs

@@ -53,7 +53,7 @@ pub async fn import_repo(
         Some(t) => t,
         None => return ApiError::AuthenticationRequired.into_response(),
     };
-    let auth_user = match crate::auth::validate_bearer_token(&state.db, &token).await {
+    let auth_user = match crate::auth::validate_bearer_token_allow_deactivated(&state.db, &token).await {
         Ok(user) => user,
         Err(e) => return ApiError::from(e).into_response(),
     };
@@ -82,16 +82,6 @@ pub async fn import_repo(
                 .into_response();
         }
     };
-    if user.deactivated_at.is_some() {
-        return (
-            StatusCode::FORBIDDEN,
-            Json(json!({
-                "error": "AccountDeactivated",
-                "message": "Account is deactivated"
-            })),
-        )
-            .into_response();
-    }
     if user.takedown_ref.is_some() {
         return (
             StatusCode::FORBIDDEN,
@@ -185,7 +175,58 @@ pub async fn import_repo(
     let skip_verification = std::env::var("SKIP_IMPORT_VERIFICATION")
         .map(|v| v == "true" || v == "1")
         .unwrap_or(false);
-    if !skip_verification {
+    let is_migration = user.deactivated_at.is_some();
+    if skip_verification {
+        warn!("Skipping all CAR verification for import (SKIP_IMPORT_VERIFICATION=true)");
+    } else if is_migration {
+        debug!("Verifying CAR file structure for migration (skipping signature verification)");
+        let verifier = CarVerifier::new();
+        match verifier.verify_car_structure_only(did, &root, &blocks) {
+            Ok(verified) => {
+                debug!(
+                    "CAR structure verification successful: rev={}, data_cid={}",
+                    verified.rev, verified.data_cid
+                );
+            }
+            Err(crate::sync::verify::VerifyError::DidMismatch {
+                commit_did,
+                expected_did,
+            }) => {
+                return (
+                    StatusCode::FORBIDDEN,
+                    Json(json!({
+                        "error": "InvalidRequest",
+                        "message": format!(
+                            "CAR file is for DID {} but you are authenticated as {}",
+                            commit_did, expected_did
+                        )
+                    })),
+                )
+                    .into_response();
+            }
+            Err(crate::sync::verify::VerifyError::MstValidationFailed(msg)) => {
+                return (
+                    StatusCode::BAD_REQUEST,
+                    Json(json!({
+                        "error": "InvalidRequest",
+                        "message": format!("MST validation failed: {}", msg)
+                    })),
+                )
+                    .into_response();
+            }
+            Err(e) => {
+                error!("CAR structure verification error: {:?}", e);
+                return (
+                    StatusCode::BAD_REQUEST,
+                    Json(json!({
+                        "error": "InvalidRequest",
+                        "message": format!("CAR verification failed: {}", e)
+                    })),
+                )
+                    .into_response();
+            }
+        }
+    } else {
         debug!("Verifying CAR file signature and structure for DID {}", did);
         let verifier = CarVerifier::new();
         match verifier.verify_car(did, &root, &blocks).await {
@@ -264,8 +305,6 @@ pub async fn import_repo(
                     .into_response();
             }
         }
-    } else {
-        warn!("Skipping CAR signature verification for import (SKIP_IMPORT_VERIFICATION=true)");
     }
     let max_blocks: usize = std::env::var("MAX_IMPORT_BLOCKS")
         .ok()