More local methods to not proxy

Cargo.lock

···
 ]
 
 [[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"
 source = "registry+https://github.com/rust-lang/crates.io-index"
···
 checksum = "79cf5c93f93228cf8efb3ba362535fb11199ac548a09ce117c9b1adc3030d706"
 dependencies = [
  "rustversion",
+]
+
+[[package]]
+name = "infer"
+version = "0.19.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a588916bfdfd92e71cacef98a63d9b1f0d74d6599980d11894290e7ddefffcf7"
+dependencies = [
+ "cfb",
 ]
 
 [[package]]
···
  "hmac",
  "http 1.4.0",
  "image",
+ "infer",
  "ipld-core",
  "iroh-car",
  "jacquard",

Cargo.toml

···
 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"

frontend/src/lib/migration/atproto-client.ts

···
     });
   }
 
+  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<string, string> = {};
+    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,

frontend/src/lib/migration/blob-migration.ts

···
       });
 
       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) {

src/api/error.rs

···
                 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,

src/api/proxy.rs

···
 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 {
···
             .headers()
             .contains_key(http::HeaderName::from(jacquard::xrpc::Header::AtprotoProxy))
         {
-            // If the age assurance override is set and this is an age assurance call then we dont want to proxy even if the client requests it.
-            if !std::env::var("PDS_AGE_ASSURANCE_OVERRIDE").is_err()
-                && (req.uri().path().ends_with("app.bsky.ageassurance.getState")
-                    || req
-                        .uri()
-                        .path()
-                        .ends_with("app.bsky.unspecced.getAgeAssuranceState"))
+            let path = req.uri().path();
+            let method = path.trim_start_matches("/");
+
+            if is_protected_method(method) {
+                return Either::Right(self.inner.call(req));
+            }
+
+            // If the age assurance override is set and this is an age assurance call then we dont want to proxy even if the client requests it
+            if std::env::var("PDS_AGE_ASSURANCE_OVERRIDE").is_ok()
+                && (path.ends_with("app.bsky.ageassurance.getState")
+                    || path.ends_with("app.bsky.unspecced.getAgeAssuranceState"))
             {
                 return Either::Right(self.inner.call(req));
             }

src/api/repo/blob.rs

···
 use serde::{Deserialize, Serialize};
 use serde_json::json;
 use std::pin::Pin;
-use tracing::{debug, error, info};
+use tracing::{debug, error, info, warn};
+
+fn detect_mime_type(data: &[u8], client_hint: &str) -> 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<AppState>,
···
         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)
···
         ))
         .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,

src/storage/mod.rs

···
     async fn put_bytes(&self, key: &str, data: Bytes) -> Result<(), StorageError>;
     async fn get(&self, key: &str) -> Result<Vec<u8>, StorageError>;
     async fn get_bytes(&self, key: &str) -> Result<Bytes, StorageError>;
+    async fn get_head(&self, key: &str, size: usize) -> Result<Bytes, StorageError>;
     async fn delete(&self, key: &str) -> Result<(), StorageError>;
     async fn put_stream(
         &self,
···
             .into_bytes();
 
         crate::metrics::record_s3_operation("get", "success");
+        Ok(data)
+    }
+
+    async fn get_head(&self, key: &str, size: usize) -> Result<Bytes, StorageError> {
+        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)
     }