test(authz): extract decide_backend_auth + CI ordering test (#142) (#170)

Closes #142.

## What

The federation wiring for #142 already landed on `main`. The one
remaining gap was a **CI-runnable** proof of the security-critical
ordering invariant: *an unauthorized caller must never reach federation
/ `apply_backend_auth`* (the confused-deputy guard). The only existing
guard,
`tests/test_federation.py::test_restricted_product_denied_to_anonymous`,
is env-gated and **skips in CI** without a live deployed proxy, so CI
never exercised the invariant.

This PR closes that gap by extracting the authorization → federation
decision out of `resolve_product` into a pure, wasm-free function and
unit-testing it in CI.

## How

- **New `authz::decide_backend_auth`** — a pure function that takes the
authorization outcome as plain values, gates the request, and *only on
success* calls `apply_backend_auth`. The confused-deputy guard is the
first line: `authentication: Option<&BackendAuth>` is `None` when the
upstream subject-scoped fetch denied the caller ⇒ `Err(AccessDenied)`
with no federation.
- **Consolidation, not addition** — the scattered write-gate previously
inlined in `resolve_product` (anon denied / read-only / non-signable /
`write`-permission) plus the trailing `apply_backend_auth` call are now
one tested function. `resolve_product` keeps the I/O (subject-scoped
Source API fetches) and routes its result through the pure fn. Net:
registry logic shrinks, the gate has a single source of truth.
- **CI unit tests** in `tests/authz.rs` (the crate is `cdylib` with
`[lib] test = false`; native tests only compile wasm-free modules —
`authz` is one, and already had a test binary). The module is pulled in
via `#[path]`, mirroring `tests/backend_auth.rs`; `backend_auth` is
included alongside it so the `crate::backend_auth` path resolves the
same way it does in the lib build.

New tests (`tests/authz.rs`):
- `unauthorized_outcome_never_federates` — `None` authentication (read
**and** write) ⇒ `AccessDenied` **and** `options` stays empty (no
`oidc_role_arn` / `skip_signature`).
- `authorized_read_unsigned_populates_options`,
`authorized_read_federated_populates_options` — authorized reads
populate options.
- `write_by_anonymous_denied`, `write_to_read_only_denied`,
`write_to_non_signable_denied`, `write_without_write_permission_denied`
— each write denial ⇒ `Err` and empty `options`.
- `authorized_write_populates_options` — authorized write
(case-insensitive `write` permission) populates options.

**Not test theater:** every denial asserts `options` is empty, and the
guard was verified non-vacuous — reverting the `None` guard to a default
makes `unauthorized_outcome_never_federates` fail.

## Acceptance criteria (#142)

- [x] Private S3 product served end-to-end via federated creds →
existing gated
`tests/test_federation.py::test_federated_object_is_served` (live-infra
integration test).
- [x] Public (no-auth) connections still use the anonymous path →
`tests/backend_auth.rs` (`unsigned_sets_skip_signature`) +
`authorized_read_unsigned_populates_options`.
- [x] **Test: an unauthorized caller never triggers `federate_*`** →
**NEW CI unit test** `unauthorized_outcome_never_federates` + the
structural data-dependency guarantee in `resolve_product` (the `?` on
the subject-scoped fetch short-circuits before federation).
- [x] Integration test against a private bucket → existing gated
`tests/test_federation.py::test_restricted_product_denied_to_anonymous`.

## Behavior change (called out, not silent)

Consolidating the gate means the permissions check no longer
short-circuits before the permissions API call. A write to a connection
that can't accept it (read-only / non-signable) now performs the
(cached) permissions fetch before denying, rather than skipping it.
**Same `403` result**, one extra cached API call on that specific deny
path — a rare write-to-misconfigured-connection corner, not the hot read
path.

Relatedly, `build_backend_options` now runs before the write gate, so a
write to a connection with an *unknown provider* surfaces `500` instead
of `403`. Such a connection is already broken for everyone (reads `500`
too), so this is consistent rather than a regression.

## Verification

- `cargo test` — 45 native tests pass (10 authz incl. 8 new, 10
backend_auth, 13 pagination, 12 routing).
- `cargo fmt --all -- --check` — clean.
- `cargo clippy --target wasm32-unknown-unknown -- -D warnings` — clean.
- `cargo check --target wasm32-unknown-unknown` — `resolve_product` and
the lib still compile for the worker target.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>

Author: alukach

src/authz.rs

@@ -1,10 +1,15 @@
-//! Write-action classification. Kept wasm-free so it can be unit-tested
-//! natively (see `tests/authz.rs`), despite the crate's `[lib] test = false`.
-//! The rest of the write gate (read-only / signable / permission checks) is
-//! trivial enough to live inline in the registry.
+//! Authorization for product backends: write-action classification and the
+//! authorization → federation decision ([`decide_backend_auth`]). Kept wasm-free
+//! so both can be unit-tested natively (see `tests/authz.rs`), despite the
+//! crate's `[lib] test = false`.
 
+use std::collections::HashMap;
+
+use multistore::error::ProxyError;
 use multistore::types::Action;
 
+use crate::backend_auth::{apply_backend_auth, BackendAuth};
+
 /// Whether an S3 action mutates the backend. Reads (GET/HEAD/LIST) are served
 /// without a write check; everything else is a write and must be authorized.
 ///
@@ -18,3 +23,56 @@ pub(crate) fn is_write_action(action: Action) -> bool {
         Action::GetObject | Action::HeadObject | Action::ListBucket
     )
 }
+
+/// Authorize a resolved product's request and, only on success, translate the
+/// connection's backend authentication into multistore `backend_options`. This
+/// is the single authorization → federation seam: `resolve_product` performs the
+/// I/O (subject-scoped Source API fetches) and hands the outcome here as plain
+/// values, so the security-critical ordering can be unit-tested off-wasm.
+///
+/// `authentication` is `Some` only when the caller's subject-scoped connection
+/// fetch succeeded. `None` models that upstream lookup having *denied* the
+/// caller, so we deny here without federating — the confused-deputy guard.
+///
+/// A write additionally requires an authenticated caller (`subject_present`) who
+/// holds the product's `write` permission, a connection that is not `read_only`,
+/// and a connection the proxy can actually sign as (an S3 web-identity role).
+///
+/// On every denial, `options` is left untouched: an unauthorized request must
+/// never have backend credentials or `skip_signature` emitted on its behalf.
+/// Only on success does [`apply_backend_auth`] populate `options`.
+#[allow(clippy::too_many_arguments)]
+pub(crate) fn decide_backend_auth(
+    authentication: Option<&BackendAuth>,
+    read_only: bool,
+    is_write: bool,
+    subject_present: bool,
+    permissions: &[String],
+    connection_id: &str,
+    backend_type: &str,
+    options: &mut HashMap<String, String>,
+) -> Result<(), ProxyError> {
+    // Confused-deputy guard: no authorized connection ⇒ never federate.
+    let auth = authentication.ok_or(ProxyError::AccessDenied)?;
+
+    if is_write {
+        // Anonymous callers can never write (and there is no subject to query
+        // permissions with).
+        if !subject_present {
+            return Err(ProxyError::AccessDenied);
+        }
+        // A connection can sign writes only via an S3 web-identity role; an
+        // unsigned/unsupported connection (or one flagged read-only) cannot
+        // accept them regardless of the caller's permissions.
+        let signable = matches!(auth, BackendAuth::S3WebIdentityRole { .. });
+        if read_only || !signable {
+            return Err(ProxyError::AccessDenied);
+        }
+        // The caller must hold the product's `write` permission.
+        if !permissions.iter().any(|p| p.eq_ignore_ascii_case("write")) {
+            return Err(ProxyError::AccessDenied);
+        }
+    }
+
+    apply_backend_auth(auth, connection_id, backend_type, options)
+}

src/source_api/registry.rs

@@ -6,8 +6,7 @@ use multistore::registry::{BucketRegistry, ResolvedBucket};
 use multistore::types::{Action, BucketConfig, ResolvedIdentity, S3Operation};
 use std::collections::HashMap;
 
-use crate::authz::is_write_action;
-use crate::backend_auth::{apply_backend_auth, BackendAuth};
+use crate::authz::{decide_backend_auth, is_write_action};
 
 use super::types::DataConnectionDetails;
 
@@ -165,60 +164,56 @@ async fn resolve_product(
     )
     .await?;
 
-    // Authorize writes. The subject-scoped fetches above already cleared the
-    // caller to *see* this product; a write additionally requires an
-    // authenticated caller who holds the product's `write` permission, a
-    // connection that is not read-only, and a connection the proxy can sign as.
-    if is_write {
-        // Anonymous callers can never write (and there is no subject to query
-        // permissions with).
-        let subject = subject.ok_or(ProxyError::AccessDenied)?;
-        // Connection-level denials need no caller lookup — check them first so a
-        // write the connection can't accept skips the permissions API call. A
-        // connection can sign writes only via an S3 web-identity role.
-        let signable = matches!(
-            connection.authentication,
-            BackendAuth::S3WebIdentityRole { .. }
-        );
-        if connection.read_only || !signable {
-            return Err(ProxyError::AccessDenied);
-        }
-        // The caller must hold the product's `write` permission.
-        let permissions = super::cache::get_or_fetch_permissions(
-            api_base_url,
-            account,
-            product,
-            api_auth,
-            request_id,
-            subject,
-        )
-        .await?;
-        if !permissions.iter().any(|p| p.eq_ignore_ascii_case("write")) {
-            return Err(ProxyError::AccessDenied);
-        }
-    }
-
     // 4. Build BucketConfig
     let (backend_type, mut backend_options) = build_backend_options(&connection.details)?;
 
+    // A write needs the caller's product permissions; fetch them only for an
+    // authenticated write, since reads never consult them and an anonymous write
+    // is denied before they're read. (Earlier revisions also skipped this fetch
+    // for writes the connection itself could not accept — read-only / unsignable —
+    // as a micro-opt. Consolidating the whole gate into `decide_backend_auth`
+    // trades that for a single source of truth, at the cost of one extra API call
+    // on that specific deny path.)
+    let permissions = match subject {
+        Some(subject) if is_write => {
+            super::cache::get_or_fetch_permissions(
+                api_base_url,
+                account,
+                product,
+                api_auth,
+                request_id,
+                subject,
+            )
+            .await?
+        }
+        _ => Vec::new(),
+    };
+
     // Backend authentication: unsigned (public) by default, or federate the
-    // proxy's OIDC identity into the connection's role.
+    // proxy's OIDC identity into the connection's role — but only after the write
+    // gate passes. Both decisions live in `decide_backend_auth`.
     //
     // The confused-deputy guard is upstream: the subject-scoped Source API
     // fetches above (get_or_fetch_product / get_or_fetch_data_connection, keyed
-    // on the caller's principal) only return the product/connection this caller
-    // is authorized for — so reaching here means the caller is already cleared
-    // for this connection's backend. Federation does not re-authorize.
+    // on the caller's principal) only return the product/connection this caller is
+    // authorized for — so reaching here means the caller is already cleared for
+    // this connection's backend, hence we pass `Some(..)`. Federation does not
+    // re-authorize.
     //
     // This ordering is enforced by data dependency, not just statement order:
-    // apply_backend_auth needs `connection`, which only exists once the
-    // subject-scoped fetch succeeds. A 403/404 from that fetch propagates via `?`
-    // before we ever get here, so an unauthorized caller can never reach
-    // federation. Guarded end-to-end by tests/test_federation.py's
-    // test_restricted_product_denied_to_anonymous.
+    // `connection` only exists once the subject-scoped fetch succeeds, and a
+    // 403/404 from that fetch propagates via `?` before we ever get here — so an
+    // unauthorized caller can never reach federation. Guarded in CI by the
+    // `tests/authz.rs` `decide_backend_auth` unit tests (an unauthorized outcome
+    // ⇒ AccessDenied with no options emitted) and end-to-end by
+    // tests/test_federation.py's test_restricted_product_denied_to_anonymous.
     span.record("auth_type", connection.authentication.kind());
-    apply_backend_auth(
-        &connection.authentication,
+    decide_backend_auth(
+        Some(&connection.authentication),
+        connection.read_only,
+        is_write,
+        subject.is_some(),
+        &permissions,
         &connection.data_connection_id,
         &backend_type,
         &mut backend_options,

tests/authz.rs

@@ -1,12 +1,21 @@
 //! Native unit tests for the wasm-free `authz` module, included via `#[path]`
 //! (the lib itself is `cdylib` with `test = false`). Mirrors the pattern in
 //! `tests/backend_auth.rs`.
+//!
+//! `authz` references `crate::backend_auth`, so that module is pulled in here too
+//! (under the test crate root) so the `crate::` path resolves the same way it
+//! does in the lib build.
 
 #[path = "../src/authz.rs"]
 mod authz;
+#[path = "../src/backend_auth.rs"]
+mod backend_auth;
 
-use authz::is_write_action;
+use authz::{decide_backend_auth, is_write_action};
+use backend_auth::BackendAuth;
+use multistore::error::ProxyError;
 use multistore::types::Action;
+use std::collections::HashMap;
 
 #[test]
 fn reads_are_not_writes() {
@@ -28,3 +37,167 @@ fn mutations_are_writes() {
         assert!(is_write_action(action), "{action:?} should be a write");
     }
 }
+
+// ── decide_backend_auth: authorization → federation ordering (#142) ─────────
+//
+// The invariant under test: an unauthorized request must be denied *before* any
+// backend authentication is applied — so a denial returns `AccessDenied` and
+// leaves `options` empty (no `oidc_role_arn` / `skip_signature` leaked). If
+// someone reordered the gate so federation ran before the checks, these tests
+// would catch populated options on a denial.
+
+fn role() -> BackendAuth {
+    BackendAuth::S3WebIdentityRole {
+        role_arn: "arn:aws:iam::1:role/r".into(),
+    }
+}
+
+/// `None` authentication = the upstream subject-scoped fetch denied the caller.
+/// Federation must never happen and `options` must stay empty, for reads or
+/// writes alike.
+#[test]
+fn unauthorized_outcome_never_federates() {
+    for is_write in [false, true] {
+        let mut o = HashMap::new();
+        let result = decide_backend_auth(
+            None,
+            false,
+            is_write,
+            true,
+            &["write".to_string()],
+            "conn-1",
+            "s3",
+            &mut o,
+        );
+        assert!(matches!(result, Err(ProxyError::AccessDenied)));
+        assert!(o.is_empty(), "denied request must not emit backend options");
+    }
+}
+
+#[test]
+fn authorized_read_unsigned_populates_options() {
+    let mut o = HashMap::new();
+    decide_backend_auth(
+        Some(&BackendAuth::Unsigned),
+        false,
+        false,
+        false,
+        &[],
+        "conn-1",
+        "s3",
+        &mut o,
+    )
+    .unwrap();
+    assert_eq!(o.get("skip_signature").map(String::as_str), Some("true"));
+}
+
+#[test]
+fn authorized_read_federated_populates_options() {
+    let mut o = HashMap::new();
+    decide_backend_auth(
+        Some(&role()),
+        false,
+        false,
+        false,
+        &[],
+        "conn-1",
+        "s3",
+        &mut o,
+    )
+    .unwrap();
+    assert_eq!(
+        o.get("oidc_role_arn").map(String::as_str),
+        Some("arn:aws:iam::1:role/r")
+    );
+    assert!(!o.contains_key("skip_signature"));
+}
+
+#[test]
+fn write_by_anonymous_denied() {
+    let mut o = HashMap::new();
+    let result = decide_backend_auth(
+        Some(&role()),
+        false,
+        true,
+        false, // no subject
+        &["write".to_string()],
+        "conn-1",
+        "s3",
+        &mut o,
+    );
+    assert!(matches!(result, Err(ProxyError::AccessDenied)));
+    assert!(o.is_empty());
+}
+
+#[test]
+fn write_to_read_only_denied() {
+    let mut o = HashMap::new();
+    let result = decide_backend_auth(
+        Some(&role()),
+        true, // read_only
+        true,
+        true,
+        &["write".to_string()],
+        "conn-1",
+        "s3",
+        &mut o,
+    );
+    assert!(matches!(result, Err(ProxyError::AccessDenied)));
+    assert!(o.is_empty());
+}
+
+#[test]
+fn write_to_non_signable_denied() {
+    // Unsigned (public) connections can't sign writes, even with the permission.
+    let mut o = HashMap::new();
+    let result = decide_backend_auth(
+        Some(&BackendAuth::Unsigned),
+        false,
+        true,
+        true,
+        &["write".to_string()],
+        "conn-1",
+        "s3",
+        &mut o,
+    );
+    assert!(matches!(result, Err(ProxyError::AccessDenied)));
+    assert!(o.is_empty());
+}
+
+#[test]
+fn write_without_write_permission_denied() {
+    let mut o = HashMap::new();
+    let result = decide_backend_auth(
+        Some(&role()),
+        false,
+        true,
+        true,
+        &["read".to_string()], // no "write"
+        "conn-1",
+        "s3",
+        &mut o,
+    );
+    assert!(matches!(result, Err(ProxyError::AccessDenied)));
+    assert!(o.is_empty());
+}
+
+#[test]
+fn authorized_write_populates_options() {
+    let mut o = HashMap::new();
+    decide_backend_auth(
+        Some(&role()),
+        false,
+        true,
+        true,
+        &["read".to_string(), "WRITE".to_string()], // case-insensitive match
+        "conn-1",
+        "s3",
+        &mut o,
+    )
+    .unwrap();
+    assert_eq!(
+        o.get("oidc_role_arn").map(String::as_str),
+        Some("arn:aws:iam::1:role/r")
+    );
+    assert!(!o.contains_key("skip_signature"));
+}