feat: per-connection backend authentication (federated backends, WIP)

Cargo.toml

@@ -18,6 +18,10 @@ path = "tests/routing.rs"
 name = "pagination"
 path = "tests/pagination.rs"
 
+[[test]]
+name = "backend_auth"
+path = "tests/backend_auth.rs"
+
 [dependencies]
 # Multistore
 multistore = { version = "0.4.0", features = ["azure"] }
@@ -55,3 +59,15 @@ web-sys = { version = "0.3", features = [
 ] }
 worker = { version = "=0.7.4", features = ["http"] }
 worker-macros = { version = "=0.7.4", features = ["http"] }
+
+# Track multistore ahead of a crates.io release. Pinned to an exact `rev` (not a
+# branch) so builds are reproducible and `cargo update` can't float to a newer
+# commit. The rev is on the `feat/shareable-credential-cache` branch, which adds
+# `Clone` to `OidcCredentialProvider` so the worker can keep one warm credential
+# cache across requests. Drop this patch and bump the versions above once it ships.
+[patch.crates-io]
+multistore = { git = "https://github.com/developmentseed/multistore", rev = "81b24ec1afad4c947a972b8c7f834806d7c0205e" }
+multistore-oidc-provider = { git = "https://github.com/developmentseed/multistore", rev = "81b24ec1afad4c947a972b8c7f834806d7c0205e" }
+multistore-path-mapping = { git = "https://github.com/developmentseed/multistore", rev = "81b24ec1afad4c947a972b8c7f834806d7c0205e" }
+multistore-sts = { git = "https://github.com/developmentseed/multistore", rev = "81b24ec1afad4c947a972b8c7f834806d7c0205e" }
+multistore-cf-workers = { git = "https://github.com/developmentseed/multistore", rev = "81b24ec1afad4c947a972b8c7f834806d7c0205e" }

src/backend_auth.rs

@@ -0,0 +1,123 @@
+//! Per-connection backend authentication: the model the Source API reports for a
+//! data connection, and its translation into multistore `backend_options`.
+//!
+//! Kept in its own module — free of wasm-only deps — so it can be unit-tested on
+//! native targets despite the crate's `[lib] test = false`. See
+//! `tests/backend_auth.rs`.
+
+use multistore::error::ProxyError;
+use serde::Deserialize;
+use std::collections::HashMap;
+
+/// `aud` claim for the proxy's AWS `AssumeRoleWithWebIdentity` assertions. AWS's
+/// fixed web-identity convention — the value the customer registers their IAM
+/// OIDC provider with and conditions the role trust policy on — so it is constant
+/// across connections. Applied at the OIDC backend-auth provider (see `lib.rs`).
+pub(crate) const AWS_STS_AUDIENCE: &str = "sts.amazonaws.com";
+
+/// Per-connection backend authentication, as reported by the Source API
+/// (a sibling of `details` on the connection).
+///
+/// Internally tagged on `type`; defaults to [`Unsigned`](BackendAuth::Unsigned)
+/// when the field is omitted, so existing connections keep issuing unsigned
+/// requests until a role is configured. Unknown `type`s (e.g. the app-side
+/// GCP/Azure workload-identity variants) deserialize to
+/// [`Unsupported`](BackendAuth::Unsupported) instead of failing the request.
+///
+/// The AWS variant carries only `role_arn`; the audience is the fixed constant
+/// [`AWS_STS_AUDIENCE`] set on the OIDC backend-auth provider, and session
+/// duration / subject scope may be added later.
+#[derive(Debug, Clone, Default, Deserialize, PartialEq, Eq)]
+#[serde(tag = "type", rename_all = "snake_case")]
+pub enum BackendAuth {
+    /// Public bucket — issue unsigned requests, no backend credentials.
+    #[default]
+    Unsigned,
+    /// Federate the proxy's OIDC identity into a customer-owned AWS role via
+    /// `AssumeRoleWithWebIdentity`, signing backend requests with the resulting
+    /// temporary credentials. (S3 only for now.)
+    S3WebIdentityRole {
+        /// ARN of the IAM role the proxy assumes for this connection.
+        role_arn: String,
+    },
+    /// An authentication type this proxy build does not implement — e.g. the
+    /// Source API's `gcp_workload_identity` / `azure_workload_identity` variants,
+    /// scaffolded app-side but without proxy/multistore support yet. Captured via
+    /// `#[serde(other)]` so an unknown `type` deserializes gracefully; treated as
+    /// unsupported (served unsigned, with a warning).
+    #[serde(other)]
+    Unsupported,
+}
+
+impl BackendAuth {
+    /// Short, stable label for logs/spans (no secrets — the role ARN is not
+    /// included).
+    pub(crate) fn kind(&self) -> &'static str {
+        match self {
+            BackendAuth::Unsigned => "unsigned",
+            BackendAuth::S3WebIdentityRole { .. } => "s3_web_identity_role",
+            BackendAuth::Unsupported => "unsupported",
+        }
+    }
+}
+
+/// Lenient `deserialize_with` for a connection's `authentication` field.
+///
+/// A *present* value that doesn't parse as a known [`BackendAuth`] — unknown
+/// `type`, missing `role_arn`, wrong shape — becomes [`Unsupported`], and `null`
+/// becomes [`Unsigned`]. This keeps a single malformed `authentication` from
+/// failing deserialization of the *entire* data-connection list, which the proxy
+/// parses in one `serde_json::from_str`. An *absent* field is handled by
+/// `#[serde(default)]` and never reaches this function.
+pub(crate) fn deserialize_lenient<'de, D>(deserializer: D) -> Result<BackendAuth, D::Error>
+where
+    D: serde::Deserializer<'de>,
+{
+    let value = serde_json::Value::deserialize(deserializer)?;
+    if value.is_null() {
+        return Ok(BackendAuth::Unsigned);
+    }
+    Ok(serde_json::from_value(value).unwrap_or(BackendAuth::Unsupported))
+}
+
+/// Translate a connection's [`BackendAuth`] into multistore `backend_options`.
+///
+/// - [`Unsigned`](BackendAuth::Unsigned) sets `skip_signature` so the proxy
+///   issues an unsigned request to a public bucket.
+/// - [`S3WebIdentityRole`](BackendAuth::S3WebIdentityRole) hands the role ARN and
+///   a per-connection subject (`scv1:conn:{id}`) to multistore's OIDC backend-auth
+///   middleware (wired in `lib.rs`), which mints the assertion (with the fixed AWS
+///   audience set on the provider), exchanges it at AWS STS, and injects the
+///   temporary credentials — clearing `skip_signature` so the request is signed.
+/// - [`Unsupported`](BackendAuth::Unsupported) can't be fulfilled, so it **fails
+///   closed** with [`ProxyError::BackendAuthError`] rather than silently serving
+///   unsigned.
+pub(crate) fn apply_backend_auth(
+    auth: &BackendAuth,
+    connection_id: &str,
+    options: &mut HashMap<String, String>,
+) -> Result<(), ProxyError> {
+    match auth {
+        BackendAuth::Unsigned => {
+            options.insert("skip_signature".to_string(), "true".to_string());
+        }
+        BackendAuth::S3WebIdentityRole { role_arn } => {
+            options.insert("auth_type".to_string(), "oidc".to_string());
+            options.insert("oidc_role_arn".to_string(), role_arn.clone());
+            options.insert(
+                "oidc_subject".to_string(),
+                format!("scv1:conn:{connection_id}"),
+            );
+        }
+        // Fail closed: a scheme we can't fulfill (the app-side GCP/Azure
+        // workload-identity variants, or a malformed `authentication`) must not
+        // fall back to unsigned — that could expose an anonymously-readable
+        // backend. Deny so the misconfiguration surfaces explicitly.
+        BackendAuth::Unsupported => {
+            return Err(ProxyError::BackendAuthError(format!(
+                "connection {connection_id}: unsupported backend authentication type"
+            )));
+        }
+    }
+    Ok(())
+}

src/lib.rs

@@ -1,5 +1,6 @@
 mod analytics;
 mod auth;
+mod backend_auth;
 mod cache;
 mod config;
 mod handlers;
@@ -18,7 +19,9 @@ use multistore_cf_workers::{
     collect_js_body, GatewayResponseExt, NoopCredentialRegistry, RequestParts, WorkerBackend,
     WorkerSubscriber,
 };
+use multistore_oidc_provider::backend_auth::{AwsBackendAuth, MaybeOidcAuth};
 use multistore_oidc_provider::route_handler::OidcRouterExt;
+use multistore_oidc_provider::{HttpExchange, OidcCredentialProvider, OidcProviderError};
 use multistore_path_mapping::{MappedRegistry, PathMapping};
 use multistore_sts::jwks::JwksCache;
 use multistore_sts::route_handler::StsRouterExt;
@@ -51,6 +54,45 @@ fn jwks_cache() -> JwksCache {
         .clone()
 }
 
+/// [`HttpExchange`] for outbound STS calls, backed by the shared reqwest client
+/// (reqwest wraps `web_sys::fetch` on wasm). This is what lets the OIDC
+/// backend-auth middleware POST `AssumeRoleWithWebIdentity` to AWS STS.
+#[derive(Clone)]
+struct FetchHttpExchange {
+    client: reqwest::Client,
+}
+
+impl HttpExchange for FetchHttpExchange {
+    async fn post_form(
+        &self,
+        url: &str,
+        form: &[(&str, &str)],
+    ) -> std::result::Result<String, OidcProviderError> {
+        let resp = self
+            .client
+            .post(url)
+            .form(form)
+            .send()
+            .await
+            .map_err(|e| OidcProviderError::HttpError(e.to_string()))?;
+        // Intentionally NOT checking the HTTP status / calling
+        // `error_for_status()`: AWS STS returns its `<ErrorResponse>` XML in the
+        // body on 4xx/5xx, and multistore's `parse_response` reads the error
+        // (code + message) out of that body. Discarding it on a non-2xx would
+        // lose the diagnostic and the precise ProxyError mapping.
+        resp.text()
+            .await
+            .map_err(|e| OidcProviderError::HttpError(e.to_string()))
+    }
+}
+
+/// Isolate-shared OIDC credential provider for backend federation. The gateway
+/// (and its middleware) are rebuilt per request, but the provider — and its
+/// credential cache — must persist so the proxy doesn't re-mint a JWT and re-run
+/// `AssumeRoleWithWebIdentity` on every request to the same role. Initialized
+/// from the first request's signing config, which is constant for the isolate.
+static OIDC_PROVIDER: OnceLock<OidcCredentialProvider<FetchHttpExchange>> = OnceLock::new();
+
 #[event(fetch)]
 async fn fetch(req: web_sys::Request, env: Env, ctx: Context) -> Result<web_sys::Response> {
     console_error_panic_hook::set_once();
@@ -144,12 +186,34 @@ async fn fetch(req: web_sys::Request, env: Env, ctx: Context) -> Result<web_sys:
             AccountListHandler::new(registry.clone(), &mapping),
         );
 
+    // ── Backend federation middleware ─────────────────────────────
+    // For a connection resolved with auth_type=oidc, mint the proxy's OIDC
+    // assertion, exchange it at AWS STS (AssumeRoleWithWebIdentity) over fetch,
+    // and inject the temporary credentials so the backend request is signed.
+    // A no-op for connections without auth_type=oidc (i.e. unsigned/public).
+    // Reuse the isolate-shared provider so its credential cache stays warm across
+    // requests; `clone()` is cheap and shares that cache.
+    let provider = OIDC_PROVIDER
+        .get_or_init(|| {
+            OidcCredentialProvider::new(
+                config.oidc.signer.clone(),
+                FetchHttpExchange {
+                    client: http_client(),
+                },
+                config.oidc.issuer.clone(),
+                crate::backend_auth::AWS_STS_AUDIENCE.to_string(),
+            )
+        })
+        .clone();
+    let backend_auth = MaybeOidcAuth::Enabled(Box::new(AwsBackendAuth::new(provider)));
+
     let gateway = ProxyGateway::new(
         WorkerBackend,
         MappedRegistry::new(registry, mapping.clone()),
         NoopCredentialRegistry,
         None,
     )
+    .with_middleware(backend_auth)
     .with_router(router)
     .with_debug_errors(max_level >= tracing::Level::DEBUG)
     .with_credential_resolver(config.session_token_key.clone());

src/registry.rs

@@ -7,6 +7,8 @@ use multistore::types::{BucketConfig, ResolvedIdentity, S3Operation};
 use serde::Deserialize;
 use std::collections::HashMap;
 
+use crate::backend_auth::{apply_backend_auth, BackendAuth};
+
 /// Registry that resolves Source Cooperative products to multistore `BucketConfig`s
 /// by calling the Source Cooperative API.
 #[derive(Clone)]
@@ -102,6 +104,7 @@ async fn resolve_product(
         account = %account,
         product = %product,
         backend_type = tracing::field::Empty,
+        auth_type = tracing::field::Empty,
     );
     let _guard = span.enter();
 
@@ -182,10 +185,20 @@ async fn resolve_product(
         _ => {}
     }
 
-    // TODO: For authenticated users, provide real backend credentials so that
-    // write operations can be forwarded to the storage backend. Currently all
-    // requests use anonymous/unsigned access, so writes will fail at the backend.
-    backend_options.insert("skip_signature".to_string(), "true".to_string());
+    // Backend authentication: unsigned (public) by default, or federate the
+    // proxy's OIDC identity into the connection's role.
+    //
+    // The confused-deputy guard is upstream: the subject-scoped Source API
+    // fetches above (get_or_fetch_product / get_or_fetch_data_connections, keyed
+    // on the caller's principal) only return products/connections this caller is
+    // authorized for — so reaching here means the caller is already cleared for
+    // this connection's backend. Federation does not re-authorize.
+    span.record("auth_type", connection.authentication.kind());
+    apply_backend_auth(
+        &connection.authentication,
+        &connection.data_connection_id,
+        &mut backend_options,
+    )?;
 
     // 5. Build prefix: connection.base_prefix + mirror.prefix
     let base_prefix = connection.details.base_prefix.as_deref().unwrap_or("");
@@ -272,6 +285,13 @@ pub struct SourceProductMirrorConfig {
 pub struct DataConnection {
     pub data_connection_id: String,
     pub details: DataConnectionDetails,
+    /// How the proxy authenticates to this connection's backend. A sibling of
+    /// `details`, matching the Source API's `DataConnection` shape. Absent →
+    /// [`BackendAuth::Unsigned`] (public bucket); a present-but-malformed value
+    /// becomes `Unsupported` rather than failing the whole list (see
+    /// [`deserialize_lenient`](crate::backend_auth::deserialize_lenient)).
+    #[serde(default, deserialize_with = "crate::backend_auth::deserialize_lenient")]
+    pub authentication: BackendAuth,
 }
 
 #[derive(Debug, Clone, Deserialize)]