diff --git a/Cargo.lock b/Cargo.lock index b506abe..68b10d0 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1109,6 +1109,17 @@ dependencies = [ "slab", ] +[[package]] +name = "cfb" +version = "0.7.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d38f2da7a0a2c4ccf0065be06397cc26a81f4e528be095826eee9d4adbb8c60f" +dependencies = [ + "byteorder", + "fnv", + "uuid", +] + [[package]] name = "cfg-if" version = "1.0.4" @@ -3005,6 +3016,15 @@ dependencies = [ "rustversion", ] +[[package]] +name = "infer" +version = "0.19.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a588916bfdfd92e71cacef98a63d9b1f0d74d6599980d11894290e7ddefffcf7" +dependencies = [ + "cfb", +] + [[package]] name = "inout" version = "0.1.4" @@ -6323,6 +6343,7 @@ dependencies = [ "hmac", "http 1.4.0", "image", + "infer", "ipld-core", "iroh-car", "jacquard", diff --git a/Cargo.toml b/Cargo.toml index 69bf2fb..facabed 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -22,6 +22,7 @@ governor = "0.10" hex = "0.4" hkdf = "0.12" hmac = "0.12" +infer = "0.19" aes-gcm = "0.10" jacquard = { version = "0.9.5", default-features = false, features = ["api", "api_bluesky", "api_full", "derive", "dns"] } jacquard-axum = "0.9.6" diff --git a/frontend/src/lib/migration/atproto-client.ts b/frontend/src/lib/migration/atproto-client.ts index 4d9f583..b792616 100644 --- a/frontend/src/lib/migration/atproto-client.ts +++ b/frontend/src/lib/migration/atproto-client.ts @@ -227,6 +227,28 @@ export class AtprotoClient { }); } + async getBlobWithContentType( + did: string, + cid: string, + ): Promise<{ data: Uint8Array; contentType: string }> { + const url = `${this.baseUrl}/xrpc/com.atproto.sync.getBlob?did=${encodeURIComponent(did)}&cid=${encodeURIComponent(cid)}`; + const headers: Record = {}; + if (this.accessToken) { + headers["Authorization"] = `Bearer ${this.accessToken}`; + } + const res = await fetch(url, { headers }); + if (!res.ok) { + const err = await res.json().catch(() => ({ + error: "Unknown", + message: res.statusText, + })); + throw new Error(err.message || err.error || res.statusText); + } + const contentType = res.headers.get("content-type") || "application/octet-stream"; + const data = new Uint8Array(await res.arrayBuffer()); + return { data, contentType }; + } + async uploadBlob( data: Uint8Array, mimeType: string, diff --git a/frontend/src/lib/migration/blob-migration.ts b/frontend/src/lib/migration/blob-migration.ts index 54079d5..fc6e1d2 100644 --- a/frontend/src/lib/migration/blob-migration.ts +++ b/frontend/src/lib/migration/blob-migration.ts @@ -87,15 +87,17 @@ export async function migrateBlobs( }); console.log("[blob-migration] Fetching blob", cid, "from source"); - const blobData = await sourceClient.getBlob(userDid, cid); + const { data: blobData, contentType } = await sourceClient.getBlobWithContentType(userDid, cid); console.log( "[blob-migration] Got blob", cid, "size:", blobData.byteLength, + "contentType:", + contentType, ); - await localClient.uploadBlob(blobData, "application/octet-stream"); - console.log("[blob-migration] Uploaded blob", cid); + await localClient.uploadBlob(blobData, contentType); + console.log("[blob-migration] Uploaded blob", cid, "with contentType:", contentType); migrated++; onProgress({ blobsMigrated: migrated }); } catch (e) { diff --git a/src/api/error.rs b/src/api/error.rs index 49c7ca9..9603dc8 100644 --- a/src/api/error.rs +++ b/src/api/error.rs @@ -480,7 +480,7 @@ impl From for ApiError { Self::AuthenticationFailed(None) } crate::auth::extractor::AuthError::TokenExpired => { - Self::AuthenticationFailed(Some("Token has expired".to_string())) + Self::ExpiredToken(Some("Token has expired".to_string())) } crate::auth::extractor::AuthError::AccountDeactivated => Self::AccountDeactivated, crate::auth::extractor::AuthError::AccountTakedown => Self::AccountTakedown, diff --git a/src/api/proxy.rs b/src/api/proxy.rs index 1ea3cab..c566f00 100644 --- a/src/api/proxy.rs +++ b/src/api/proxy.rs @@ -15,22 +15,87 @@ use tower::{Service, util::BoxCloneSyncService}; use tracing::{error, info, warn}; const PROTECTED_METHODS: &[&str] = &[ + "app.bsky.actor.getPreferences", + "app.bsky.actor.putPreferences", + "com.atproto.admin.deleteAccount", + "com.atproto.admin.disableAccountInvites", + "com.atproto.admin.disableInviteCodes", + "com.atproto.admin.enableAccountInvites", + "com.atproto.admin.getAccountInfo", + "com.atproto.admin.getAccountInfos", + "com.atproto.admin.getInviteCodes", + "com.atproto.admin.getSubjectStatus", + "com.atproto.admin.searchAccounts", "com.atproto.admin.sendEmail", + "com.atproto.admin.updateAccountEmail", + "com.atproto.admin.updateAccountHandle", + "com.atproto.admin.updateAccountPassword", + "com.atproto.admin.updateSubjectStatus", + "com.atproto.identity.getRecommendedDidCredentials", "com.atproto.identity.requestPlcOperationSignature", "com.atproto.identity.signPlcOperation", + "com.atproto.identity.submitPlcOperation", "com.atproto.identity.updateHandle", + "com.atproto.repo.applyWrites", + "com.atproto.repo.createRecord", + "com.atproto.repo.deleteRecord", + "com.atproto.repo.importRepo", + "com.atproto.repo.putRecord", + "com.atproto.repo.uploadBlob", "com.atproto.server.activateAccount", + "com.atproto.server.checkAccountStatus", "com.atproto.server.confirmEmail", + "com.atproto.server.confirmSignup", + "com.atproto.server.createAccount", "com.atproto.server.createAppPassword", + "com.atproto.server.createInviteCode", + "com.atproto.server.createInviteCodes", + "com.atproto.server.createSession", + "com.atproto.server.createTotpSecret", "com.atproto.server.deactivateAccount", + "com.atproto.server.deleteAccount", + "com.atproto.server.deletePasskey", + "com.atproto.server.deleteSession", + "com.atproto.server.describeServer", + "com.atproto.server.disableTotp", + "com.atproto.server.enableTotp", + "com.atproto.server.finishPasskeyRegistration", "com.atproto.server.getAccountInviteCodes", + "com.atproto.server.getServiceAuth", "com.atproto.server.getSession", + "com.atproto.server.getTotpStatus", "com.atproto.server.listAppPasswords", + "com.atproto.server.listPasskeys", + "com.atproto.server.refreshSession", + "com.atproto.server.regenerateBackupCodes", "com.atproto.server.requestAccountDelete", "com.atproto.server.requestEmailConfirmation", "com.atproto.server.requestEmailUpdate", + "com.atproto.server.requestPasswordReset", + "com.atproto.server.resendMigrationVerification", + "com.atproto.server.resendVerification", + "com.atproto.server.reserveSigningKey", + "com.atproto.server.resetPassword", "com.atproto.server.revokeAppPassword", + "com.atproto.server.startPasskeyRegistration", "com.atproto.server.updateEmail", + "com.atproto.server.updatePasskey", + "com.atproto.server.verifyMigrationEmail", + "com.atproto.sync.getBlob", + "com.atproto.sync.getBlocks", + "com.atproto.sync.getCheckout", + "com.atproto.sync.getHead", + "com.atproto.sync.getLatestCommit", + "com.atproto.sync.getRecord", + "com.atproto.sync.getRepo", + "com.atproto.sync.getRepoStatus", + "com.atproto.sync.listBlobs", + "com.atproto.sync.listRepos", + "com.atproto.sync.notifyOfUpdate", + "com.atproto.sync.requestCrawl", + "com.atproto.sync.subscribeRepos", + "com.atproto.temp.checkSignupQueue", + "com.atproto.temp.dereferenceScope", ]; fn is_protected_method(method: &str) -> bool { @@ -89,13 +154,17 @@ impl> Service String { + if let Some(kind) = infer::get(data) { + let detected = kind.mime_type().to_string(); + if detected != client_hint { + debug!( + "MIME type detection: client sent '{}', detected '{}'", + client_hint, detected + ); + } + detected + } else { + if client_hint == "*/*" || client_hint.is_empty() { + warn!("Could not detect MIME type and client sent invalid hint: '{}'", client_hint); + "application/octet-stream".to_string() + } else { + client_hint.to_string() + } + } +} pub async fn upload_blob( State(state): State, @@ -91,11 +111,10 @@ pub async fn upload_blob( return ApiError::Forbidden.into_response(); } - let mime_type = headers + let client_mime_hint = headers .get("content-type") .and_then(|h| h.to_str().ok()) - .unwrap_or("application/octet-stream") - .to_string(); + .unwrap_or("application/octet-stream"); let user_query = sqlx::query!("SELECT id FROM users WHERE did = $1", did) .fetch_optional(&state.db) @@ -136,6 +155,14 @@ pub async fn upload_blob( .into_response(); } + let mime_type = match state.blob_store.get_head(&temp_key, 8192).await { + Ok(head_bytes) => detect_mime_type(&head_bytes, client_mime_hint), + Err(e) => { + warn!("Failed to read blob head for MIME detection: {:?}", e); + client_mime_hint.to_string() + } + }; + let multihash = match Multihash::wrap(0x12, &upload_result.sha256_hash) { Ok(mh) => mh, Err(e) => { diff --git a/src/storage/mod.rs b/src/storage/mod.rs index 62a0d0b..4374295 100644 --- a/src/storage/mod.rs +++ b/src/storage/mod.rs @@ -34,6 +34,7 @@ pub trait BlobStorage: Send + Sync { async fn put_bytes(&self, key: &str, data: Bytes) -> Result<(), StorageError>; async fn get(&self, key: &str) -> Result, StorageError>; async fn get_bytes(&self, key: &str) -> Result; + async fn get_head(&self, key: &str, size: usize) -> Result; async fn delete(&self, key: &str) -> Result<(), StorageError>; async fn put_stream( &self, @@ -233,6 +234,35 @@ impl BlobStorage for S3BlobStorage { Ok(data) } + async fn get_head(&self, key: &str, size: usize) -> Result { + let range = format!("bytes=0-{}", size.saturating_sub(1)); + let resp = self + .client + .get_object() + .bucket(&self.bucket) + .key(key) + .range(range) + .send() + .await + .map_err(|e| { + crate::metrics::record_s3_operation("get_head", "error"); + StorageError::S3(e.to_string()) + })?; + + let data = resp + .body + .collect() + .await + .map_err(|e| { + crate::metrics::record_s3_operation("get_head", "error"); + StorageError::S3(e.to_string()) + })? + .into_bytes(); + + crate::metrics::record_s3_operation("get_head", "success"); + Ok(data) + } + async fn delete(&self, key: &str) -> Result<(), StorageError> { let result = self .client