fix: harden backend resource handles

crates/worker-runtime/src/resource.rs

@@ -254,3 +254,83 @@ pub fn validate_resource_handle_text(label: &str, value: &str) -> Result<(), Str
     }
     Ok(())
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::profile_archive::ProfileSourceGraphSummary;
+    use std::collections::BTreeMap;
+
+    fn graph() -> ProfileSourceGraphSummary {
+        ProfileSourceGraphSummary {
+            source_count: 1,
+            total_source_bytes: 4,
+            entrypoints: BTreeMap::from([(
+                "default".to_string(),
+                "profiles/default.dcdl".to_string(),
+            )]),
+            import_count: 0,
+        }
+    }
+
+    fn handle_for(bytes: &[u8]) -> BackendResourceHandle {
+        BackendResourceHandle {
+            kind: BackendResourceKind::ProfileSourceArchive,
+            workspace_id: "workspace-test".to_string(),
+            scope_id: Some("workspace-profile-source".to_string()),
+            runtime_id: Some("runtime-test".to_string()),
+            worker_id: Some("worker-test".to_string()),
+            resource_id: "profile-source-archive:test".to_string(),
+            digest: sha256_hex(bytes),
+            operation: BackendResourceOperation::FetchArchive,
+            expires_at_unix_seconds: 4_102_444_800,
+            nonce: "nonce-test".to_string(),
+            revision: sha256_hex(bytes),
+            generation: Some(1),
+            max_bytes: 1024,
+            content_type: PROFILE_SOURCE_ARCHIVE_CONTENT_TYPE.to_string(),
+            redaction: ResourceRedactionPolicy::RuntimeInternalOnly,
+            audit_correlation_id: "audit-test".to_string(),
+            profile_source_graph: Some(graph()),
+        }
+    }
+
+    #[test]
+    fn response_verification_detects_digest_mismatch() {
+        let bytes = b"archive-bytes";
+        let handle = handle_for(bytes);
+        let error = profile_source_archive_from_response(
+            &handle,
+            BackendResourceFetchResponse {
+                kind: BackendResourceKind::ProfileSourceArchive,
+                resource_id: handle.resource_id.clone(),
+                digest: handle.digest.clone(),
+                content_type: PROFILE_SOURCE_ARCHIVE_CONTENT_TYPE.to_string(),
+                bytes: b"tampered-bytes".to_vec(),
+                audit_correlation_id: handle.audit_correlation_id.clone(),
+            },
+        )
+        .unwrap_err();
+        assert!(matches!(error, BackendResourceError::DigestMismatch { .. }));
+    }
+
+    #[test]
+    fn response_verification_rejects_oversized_bytes() {
+        let bytes = b"archive-bytes";
+        let mut handle = handle_for(bytes);
+        handle.max_bytes = 2;
+        let error = profile_source_archive_from_response(
+            &handle,
+            BackendResourceFetchResponse {
+                kind: BackendResourceKind::ProfileSourceArchive,
+                resource_id: handle.resource_id.clone(),
+                digest: handle.digest.clone(),
+                content_type: PROFILE_SOURCE_ARCHIVE_CONTENT_TYPE.to_string(),
+                bytes: bytes.to_vec(),
+                audit_correlation_id: handle.audit_correlation_id.clone(),
+            },
+        )
+        .unwrap_err();
+        assert!(matches!(error, BackendResourceError::Oversized { .. }));
+    }
+}

crates/worker-runtime/src/worker_backend.rs

@@ -184,11 +184,6 @@ impl ProfileRuntimeWorkerFactory {
                     bundle.metadata.id
                 )
             })?;
-        if let Some(cached) = self.profile_archive_cache.get(&handle.digest) {
-            return cached
-                .verify()
-                .map_err(|err| format!("failed to verify cached profile source archive: {err}"));
-        }
         let client = self.resource_client.as_ref().ok_or_else(|| {
             format!(
                 "config bundle {} requires a Backend resource client for profile source archive fetch",
@@ -204,10 +199,15 @@ impl ProfileRuntimeWorkerFactory {
             .fetch_resource(fetch_request)
             .await
             .map_err(format_backend_resource_error)?;
-        let archive = profile_source_archive_from_response(&handle, response)
+        let fetched_archive = profile_source_archive_from_response(&handle, response)
             .map_err(format_backend_resource_error)?;
-        self.profile_archive_cache.insert(archive.clone());
+        if let Some(cached) = self.profile_archive_cache.get(&handle.digest) {
-        archive
+            return cached
+                .verify()
+                .map_err(|err| format!("failed to verify cached profile source archive: {err}"));
+        }
+        self.profile_archive_cache.insert(fetched_archive.clone());
+        fetched_archive
             .verify()
             .map_err(|err| format!("failed to verify fetched profile source archive: {err}"))
     }
@@ -701,6 +701,7 @@ where
 #[cfg(test)]
 mod tests {
     use super::*;
+    use std::collections::{BTreeMap, VecDeque};
     use std::fs;
     use std::pin::Pin;
     use std::process::Command;
@@ -711,6 +712,7 @@ mod tests {
         ConfigBundleRef, CreateWorkerRequest, DirtyStatePolicy, MaterializerKind, ProfileSelector,
         RepositorySelector, WorkingDirectoryRepository, WorkingDirectoryRequest,
     };
+    use crate::execution::WorkerExecutionContext;
     use crate::identity::RuntimeId;
     use crate::management::RuntimeOptions;
     use crate::observation::{TranscriptQuery, TranscriptRole};
@@ -854,6 +856,125 @@ mod tests {
         .with_computed_digest()
     }
 
+    fn sample_profile_archive() -> crate::profile_archive::ProfileSourceArchive {
+        let entrypoints =
+            BTreeMap::from([("default".to_string(), "profiles/default.dcdl".to_string())]);
+        let sources = BTreeMap::from([(
+            "profiles/default.dcdl".to_string(),
+            r#"{
+                slug = "default";
+                description = "Default";
+                scope = "workspace_read";
+            }"#
+            .to_string(),
+        )]);
+        crate::profile_archive::ProfileSourceArchive::build(
+            crate::profile_archive::ProfileSourceArchiveInput {
+                id: "profile-source-archive:test".to_string(),
+                entrypoints,
+                imports: BTreeMap::new(),
+                sources,
+            },
+        )
+        .unwrap()
+    }
+
+    fn handle_for_archive(
+        archive: &crate::profile_archive::ProfileSourceArchive,
+    ) -> crate::resource::BackendResourceHandle {
+        crate::resource::BackendResourceHandle {
+            kind: crate::resource::BackendResourceKind::ProfileSourceArchive,
+            workspace_id: "workspace-test".to_string(),
+            scope_id: Some("workspace-profile-source".to_string()),
+            runtime_id: Some("runtime-test".to_string()),
+            worker_id: Some("worker-test".to_string()),
+            resource_id: archive.reference.id.clone(),
+            digest: archive.reference.digest.clone(),
+            operation: crate::resource::BackendResourceOperation::FetchArchive,
+            expires_at_unix_seconds: 4_102_444_800,
+            nonce: "nonce-test".to_string(),
+            revision: archive.reference.digest.clone(),
+            generation: Some(1),
+            max_bytes: crate::resource::DEFAULT_PROFILE_SOURCE_ARCHIVE_MAX_BYTES,
+            content_type: crate::resource::PROFILE_SOURCE_ARCHIVE_CONTENT_TYPE.to_string(),
+            redaction: crate::resource::ResourceRedactionPolicy::RuntimeInternalOnly,
+            audit_correlation_id: "audit-test".to_string(),
+            profile_source_graph: Some(archive.reference.source_graph.clone()),
+        }
+    }
+
+    fn response_for_archive(
+        handle: &crate::resource::BackendResourceHandle,
+        archive: &crate::profile_archive::ProfileSourceArchive,
+    ) -> crate::resource::BackendResourceFetchResponse {
+        crate::resource::BackendResourceFetchResponse {
+            kind: crate::resource::BackendResourceKind::ProfileSourceArchive,
+            resource_id: handle.resource_id.clone(),
+            digest: handle.digest.clone(),
+            content_type: crate::resource::PROFILE_SOURCE_ARCHIVE_CONTENT_TYPE.to_string(),
+            bytes: archive.content.clone(),
+            audit_correlation_id: handle.audit_correlation_id.clone(),
+        }
+    }
+
+    #[derive(Clone)]
+    struct SequencedResourceClient {
+        responses: Arc<
+            Mutex<
+                VecDeque<
+                    Result<
+                        crate::resource::BackendResourceFetchResponse,
+                        crate::resource::BackendResourceError,
+                    >,
+                >,
+            >,
+        >,
+        call_count: Arc<AtomicUsize>,
+    }
+
+    #[async_trait]
+    impl crate::resource::BackendResourceClient for SequencedResourceClient {
+        async fn fetch_resource(
+            &self,
+            _request: crate::resource::BackendResourceFetchRequest,
+        ) -> Result<
+            crate::resource::BackendResourceFetchResponse,
+            crate::resource::BackendResourceError,
+        > {
+            self.call_count.fetch_add(1, Ordering::SeqCst);
+            self.responses
+                .lock()
+                .unwrap()
+                .pop_front()
+                .expect("missing sequenced resource response")
+        }
+    }
+
+    fn spawn_request_with_bundle(
+        bundle: crate::config_bundle::ConfigBundle,
+    ) -> WorkerExecutionSpawnRequest {
+        let worker_ref = crate::identity::WorkerRef::new(
+            RuntimeId::new("runtime-test").unwrap(),
+            crate::identity::WorkerId::new("worker-test").unwrap(),
+        );
+        WorkerExecutionSpawnRequest {
+            worker_ref: worker_ref.clone(),
+            request: CreateWorkerRequest {
+                profile: ProfileSelector::RuntimeDefault,
+                config_bundle: ConfigBundleRef {
+                    id: bundle.metadata.id.clone(),
+                    digest: bundle.metadata.digest.clone(),
+                },
+                initial_input: None,
+                working_directory_request: None,
+                working_directory: None,
+            },
+            context: WorkerExecutionContext::new(worker_ref, Arc::new(|_, _| panic!("unused"))),
+            working_directory: None,
+            config_bundle: Some(bundle),
+        }
+    }
+
     fn create_request(_name: &str) -> CreateWorkerRequest {
         let bundle = test_bundle();
         CreateWorkerRequest {
@@ -906,6 +1027,37 @@ mod tests {
         }
     }
 
+    #[tokio::test]
+    async fn cached_profile_archive_still_requires_backend_authorization() {
+        let archive = sample_profile_archive();
+        let handle = handle_for_archive(&archive);
+        let call_count = Arc::new(AtomicUsize::new(0));
+        let client = SequencedResourceClient {
+            responses: Arc::new(Mutex::new(VecDeque::from([
+                Ok(response_for_archive(&handle, &archive)),
+                Err(crate::resource::BackendResourceError::Expired),
+            ]))),
+            call_count: call_count.clone(),
+        };
+        let factory = ProfileRuntimeWorkerFactory::new(tempfile::tempdir().unwrap().path())
+            .with_resource_client(Arc::new(client));
+        let mut bundle = test_bundle();
+        bundle.profile_source_archive_handle = Some(handle);
+        bundle = bundle.with_computed_digest();
+
+        factory
+            .resolve_profile_source_archive(&bundle, &spawn_request_with_bundle(bundle.clone()))
+            .await
+            .expect("first fetch should authorize and cache archive");
+        let err = factory
+            .resolve_profile_source_archive(&bundle, &spawn_request_with_bundle(bundle.clone()))
+            .await
+            .unwrap_err();
+
+        assert!(err.contains("expired"), "unexpected error: {err}");
+        assert_eq!(call_count.load(Ordering::SeqCst), 2);
+    }
+
     #[test]
     fn builtin_profile_selector_is_not_double_prefixed() {
         assert_eq!(

crates/workspace-server/src/resource_broker.rs

@@ -19,7 +19,6 @@ pub struct BackendResourceBroker {
 
 #[derive(Clone)]
 struct StoredResource {
-    workspace_id: String,
     runtime_id: Option<String>,
     worker_id: Option<String>,
     handle: BackendResourceHandle,
@@ -58,7 +57,6 @@ impl BackendResourceBroker {
             profile_source_graph: Some(archive.reference.source_graph.clone()),
         };
         let stored = StoredResource {
-            workspace_id,
             runtime_id: runtime_id.map(|id| id.as_str().to_string()),
             worker_id: worker_id.map(|id| id.as_str().to_string()),
             handle: handle.clone(),
@@ -75,9 +73,6 @@ impl BackendResourceBroker {
         request: BackendResourceFetchRequest,
     ) -> Result<BackendResourceFetchResponse, BackendResourceError> {
         verify_handle_shape(&request.handle)?;
-        if request.handle.expires_at_unix_seconds < Utc::now().timestamp() {
-            return Err(BackendResourceError::Expired);
-        }
         let stored = self
             .resources
             .lock()
@@ -87,15 +82,20 @@ impl BackendResourceBroker {
             .get(&request.handle.nonce)
             .cloned()
             .ok_or(BackendResourceError::MissingResource)?;
-        if stored.workspace_id != request.handle.workspace_id
+        verify_handle_shape(&stored.handle)?;
-            || stored.runtime_id != request.handle.runtime_id
+        if stored.handle.expires_at_unix_seconds < Utc::now().timestamp() {
-            || stored.worker_id != request.handle.worker_id
+            return Err(BackendResourceError::Expired);
-            || stored.handle.resource_id != request.handle.resource_id
+        }
-            || stored.handle.digest != request.handle.digest
+        let actual_bytes = stored.archive.content.len() as u64;
-            || stored.handle.revision != request.handle.revision
+        if actual_bytes > stored.handle.max_bytes {
-        {
+            return Err(BackendResourceError::Oversized {
+                max_bytes: stored.handle.max_bytes,
+                actual_bytes,
+            });
+        }
+        if request.handle != stored.handle {
             return Err(BackendResourceError::Unauthorized {
-                message: "resource handle metadata does not match broker record".to_string(),
+                message: "resource handle does not match broker-issued handle".to_string(),
             });
         }
         if let Some(expected_runtime_id) = stored.runtime_id.as_deref() {
@@ -112,13 +112,6 @@ impl BackendResourceBroker {
                 });
             }
         }
-        let actual_bytes = stored.archive.content.len() as u64;
-        if actual_bytes > request.handle.max_bytes {
-            return Err(BackendResourceError::Oversized {
-                max_bytes: request.handle.max_bytes,
-                actual_bytes,
-            });
-        }
         Ok(BackendResourceFetchResponse {
             kind: BackendResourceKind::ProfileSourceArchive,
             resource_id: stored.archive.reference.id,
@@ -162,7 +155,7 @@ fn verify_handle_shape(handle: &BackendResourceHandle) -> Result<(), BackendReso
 mod tests {
     use super::*;
     use std::collections::BTreeMap;
-    use worker_runtime::identity::RuntimeId;
+    use worker_runtime::identity::{RuntimeId, WorkerId};
     use worker_runtime::profile_archive::{
         ProfileSourceArchive, ProfileSourceArchiveRef, ProfileSourceGraphSummary, sha256_hex,
     };
@@ -187,6 +180,39 @@ mod tests {
         }
     }
 
+    fn archive_with_len(len: usize) -> ProfileSourceArchive {
+        let content = vec![b'x'; len];
+        let mut entrypoints = BTreeMap::new();
+        entrypoints.insert("default".to_string(), "profiles/default.dcdl".to_string());
+        ProfileSourceArchive {
+            reference: ProfileSourceArchiveRef {
+                id: format!("profile-source-archive:test-{len}"),
+                digest: sha256_hex(&content),
+                size_bytes: content.len() as u64,
+                source_graph: ProfileSourceGraphSummary {
+                    entrypoints,
+                    source_count: 1,
+                    import_count: 0,
+                    total_source_bytes: content.len() as u64,
+                },
+            },
+            content,
+        }
+    }
+
+    fn request(
+        handle: BackendResourceHandle,
+        runtime_id: &RuntimeId,
+        worker_id: Option<&WorkerId>,
+    ) -> BackendResourceFetchRequest {
+        BackendResourceFetchRequest {
+            audit_correlation_id: handle.audit_correlation_id.clone(),
+            handle,
+            runtime_id: runtime_id.as_str().to_string(),
+            worker_id: worker_id.map(|id| id.as_str().to_string()),
+        }
+    }
+
     #[test]
     fn broker_issues_and_verifies_profile_source_archive_handles() {
         let broker = BackendResourceBroker::default();
@@ -212,20 +238,99 @@ mod tests {
     #[test]
     fn broker_rejects_runtime_mismatch() {
         let broker = BackendResourceBroker::default();
+        let runtime_a = RuntimeId::new("runtime-a").unwrap();
         let handle = broker.issue_profile_source_archive_handle(
             "workspace-test",
-            Some(&RuntimeId::new("runtime-a").unwrap()),
+            Some(&runtime_a),
             None,
             archive(),
         );
         let err = broker
-            .fetch_profile_source_archive(BackendResourceFetchRequest {
+            .fetch_profile_source_archive(request(
-                handle: handle.clone(),
+                handle,
-                runtime_id: RuntimeId::new("runtime-b").unwrap().as_str().to_string(),
+                &RuntimeId::new("runtime-b").unwrap(),
-                worker_id: None,
+                None,
-                audit_correlation_id: handle.audit_correlation_id.clone(),
+            ))
-            })
             .unwrap_err();
         assert!(matches!(err, BackendResourceError::Unauthorized { .. }));
     }
+
+    #[test]
+    fn broker_rejects_worker_mismatch() {
+        let broker = BackendResourceBroker::default();
+        let runtime_id = RuntimeId::new("runtime-test").unwrap();
+        let worker_a = WorkerId::new("worker-a").unwrap();
+        let worker_b = WorkerId::new("worker-b").unwrap();
+        let handle = broker.issue_profile_source_archive_handle(
+            "workspace-test",
+            Some(&runtime_id),
+            Some(&worker_a),
+            archive(),
+        );
+        let err = broker
+            .fetch_profile_source_archive(request(handle, &runtime_id, Some(&worker_b)))
+            .unwrap_err();
+        assert!(matches!(err, BackendResourceError::Unauthorized { .. }));
+    }
+
+    #[test]
+    fn broker_rejects_expiry_extension_from_request_handle() {
+        let broker = BackendResourceBroker::default();
+        let runtime_id = RuntimeId::new("runtime-test").unwrap();
+        let handle = broker.issue_profile_source_archive_handle(
+            "workspace-test",
+            Some(&runtime_id),
+            None,
+            archive(),
+        );
+        broker
+            .resources
+            .lock()
+            .unwrap()
+            .get_mut(&handle.nonce)
+            .unwrap()
+            .handle
+            .expires_at_unix_seconds = 1;
+        let mut extended = handle;
+        extended.expires_at_unix_seconds = 4_102_444_800;
+        let err = broker
+            .fetch_profile_source_archive(request(extended, &runtime_id, None))
+            .unwrap_err();
+        assert!(matches!(err, BackendResourceError::Expired));
+    }
+
+    #[test]
+    fn broker_rejects_policy_tampered_request_handle() {
+        let broker = BackendResourceBroker::default();
+        let runtime_id = RuntimeId::new("runtime-test").unwrap();
+        let mut handle = broker.issue_profile_source_archive_handle(
+            "workspace-test",
+            Some(&runtime_id),
+            None,
+            archive(),
+        );
+        handle.scope_id = Some("tampered-scope".to_string());
+        let err = broker
+            .fetch_profile_source_archive(request(handle, &runtime_id, None))
+            .unwrap_err();
+        assert!(matches!(err, BackendResourceError::Unauthorized { .. }));
+    }
+
+    #[test]
+    fn broker_uses_stored_max_bytes_when_request_handle_is_tampered() {
+        let broker = BackendResourceBroker::default();
+        let runtime_id = RuntimeId::new("runtime-test").unwrap();
+        let archive = archive_with_len((DEFAULT_PROFILE_SOURCE_ARCHIVE_MAX_BYTES + 1) as usize);
+        let mut handle = broker.issue_profile_source_archive_handle(
+            "workspace-test",
+            Some(&runtime_id),
+            None,
+            archive,
+        );
+        handle.max_bytes = DEFAULT_PROFILE_SOURCE_ARCHIVE_MAX_BYTES + 1024;
+        let err = broker
+            .fetch_profile_source_archive(request(handle, &runtime_id, None))
+            .unwrap_err();
+        assert!(matches!(err, BackendResourceError::Oversized { .. }));
+    }
 }

crates/workspace-server/src/server.rs

@@ -3269,6 +3269,7 @@ mod tests {
     use tokio_tungstenite::connect_async;
     use tokio_tungstenite::tungstenite::Message;
     use tower::ServiceExt;
+    use worker_runtime::resource::BackendResourceClient;
 
     use crate::hosts::{
         TicketWorkerRole, WorkerInputKind, WorkerOperationState, WorkerSpawnAcceptanceRequirement,
@@ -3441,16 +3442,161 @@ mod tests {
         }
     }
 
-    async fn test_app(workspace_root: impl Into<PathBuf>) -> Router {
+    async fn test_api(workspace_root: impl Into<PathBuf>) -> WorkspaceApi {
         let store = SqliteWorkspaceStore::in_memory().unwrap();
-        let api = WorkspaceApi::new_with_execution_backend(
+        WorkspaceApi::new_with_execution_backend(
             test_server_config(workspace_root),
             Arc::new(store),
             Arc::new(DeterministicExecutionBackend::default()),
         )
         .await
-        .unwrap();
+        .unwrap()
-        build_router(api)
+    }
+
+    async fn test_app(workspace_root: impl Into<PathBuf>) -> Router {
+        build_router(test_api(workspace_root).await)
+    }
+
+    fn test_profile_archive() -> worker_runtime::profile_archive::ProfileSourceArchive {
+        use worker_runtime::profile_archive::{ProfileSourceArchive, ProfileSourceArchiveInput};
+        ProfileSourceArchive::build(ProfileSourceArchiveInput {
+            id: "profile-source-archive:server-test".to_string(),
+            entrypoints: std::collections::BTreeMap::from([(
+                "default".to_string(),
+                "profiles/default.dcdl".to_string(),
+            )]),
+            imports: std::collections::BTreeMap::new(),
+            sources: std::collections::BTreeMap::from([(
+                "profiles/default.dcdl".to_string(),
+                r#"{
+                    slug = "default";
+                    description = "Default";
+                    scope = "workspace_read";
+                }"#
+                .to_string(),
+            )]),
+        })
+        .unwrap()
+    }
+
+    fn missing_resource_handle() -> worker_runtime::resource::BackendResourceHandle {
+        worker_runtime::resource::BackendResourceHandle {
+            kind: worker_runtime::resource::BackendResourceKind::ProfileSourceArchive,
+            workspace_id: "workspace-test".to_string(),
+            scope_id: Some("workspace-profile-source".to_string()),
+            runtime_id: Some("runtime-test".to_string()),
+            worker_id: None,
+            resource_id: "profile-source-archive:missing".to_string(),
+            digest: "sha256:0000000000000000000000000000000000000000000000000000000000000000"
+                .to_string(),
+            operation: worker_runtime::resource::BackendResourceOperation::FetchArchive,
+            expires_at_unix_seconds: 4_102_444_800,
+            nonce: "missing-nonce".to_string(),
+            revision: "missing-revision".to_string(),
+            generation: None,
+            max_bytes: worker_runtime::resource::DEFAULT_PROFILE_SOURCE_ARCHIVE_MAX_BYTES,
+            content_type: worker_runtime::resource::PROFILE_SOURCE_ARCHIVE_CONTENT_TYPE.to_string(),
+            redaction: worker_runtime::resource::ResourceRedactionPolicy::RuntimeInternalOnly,
+            audit_correlation_id: "audit-missing".to_string(),
+            profile_source_graph: Some(
+                worker_runtime::profile_archive::ProfileSourceGraphSummary {
+                    entrypoints: std::collections::BTreeMap::from([(
+                        "default".to_string(),
+                        "profiles/default.dcdl".to_string(),
+                    )]),
+                    source_count: 1,
+                    import_count: 0,
+                    total_source_bytes: 0,
+                },
+            ),
+        }
+    }
+
+    #[tokio::test]
+    async fn internal_resource_fetch_rest_returns_typed_missing_resource() {
+        let workspace = tempfile::tempdir().unwrap();
+        init_clean_git_workspace(workspace.path());
+        let app = test_app(workspace.path()).await;
+        let handle = missing_resource_handle();
+        let response = app
+            .oneshot(
+                Request::post("/internal/runtime/resources/fetch")
+                    .header("content-type", "application/json")
+                    .body(Body::from(
+                        serde_json::to_vec(
+                            &worker_runtime::resource::BackendResourceFetchRequest {
+                                audit_correlation_id: handle.audit_correlation_id.clone(),
+                                runtime_id: "runtime-test".to_string(),
+                                worker_id: None,
+                                handle,
+                            },
+                        )
+                        .unwrap(),
+                    ))
+                    .unwrap(),
+            )
+            .await
+            .unwrap();
+        assert_eq!(response.status(), StatusCode::NOT_FOUND);
+        let bytes = to_bytes(response.into_body(), usize::MAX).await.unwrap();
+        let error: worker_runtime::resource::BackendResourceError =
+            serde_json::from_slice(&bytes).unwrap();
+        assert!(matches!(
+            error,
+            worker_runtime::resource::BackendResourceError::MissingResource
+        ));
+    }
+
+    #[tokio::test]
+    async fn remote_http_resource_fetch_uses_backend_resource_contract() {
+        let workspace = tempfile::tempdir().unwrap();
+        init_clean_git_workspace(workspace.path());
+        let api = test_api(workspace.path()).await;
+        let broker = api.resource_broker.clone();
+        let archive = test_profile_archive();
+        let runtime_id = worker_runtime::identity::RuntimeId::new("runtime-test").unwrap();
+        let handle = broker.issue_profile_source_archive_handle(
+            "workspace-test",
+            Some(&runtime_id),
+            None,
+            archive,
+        );
+        let app = build_router(api);
+        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let addr = listener.local_addr().unwrap();
+        let server = tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
+        let client = worker_runtime::resource::HttpBackendResourceClient::new(
+            format!("http://{addr}/internal/runtime/resources/fetch"),
+            None,
+        );
+
+        let response = client
+            .fetch_resource(worker_runtime::resource::BackendResourceFetchRequest {
+                audit_correlation_id: handle.audit_correlation_id.clone(),
+                runtime_id: runtime_id.as_str().to_string(),
+                worker_id: None,
+                handle: handle.clone(),
+            })
+            .await
+            .expect("remote HTTP resource fetch succeeds");
+        assert_eq!(response.digest, handle.digest);
+
+        let mut tampered = handle;
+        tampered.scope_id = Some("tampered".to_string());
+        let error = client
+            .fetch_resource(worker_runtime::resource::BackendResourceFetchRequest {
+                audit_correlation_id: tampered.audit_correlation_id.clone(),
+                runtime_id: runtime_id.as_str().to_string(),
+                worker_id: None,
+                handle: tampered,
+            })
+            .await
+            .unwrap_err();
+        assert!(matches!(
+            error,
+            worker_runtime::resource::BackendResourceError::Unauthorized { .. }
+        ));
+        server.abort();
     }
 
     fn runtime_test_bundle() -> worker_runtime::config_bundle::ConfigBundle {