3/6: Async IDC Receiver side.

Cargo.toml

@@ -51,6 +51,7 @@ members = [
   "modules/perf-test",
   "modules/module-test",
   "modules/environment-test",
+  "modules/idc-auth-test",
   "templates/basic-rs/spacetimedb",
   "templates/chat-console-rs/spacetimedb",
   "modules/sdk-test",

crates/auth/src/idc.rs

@@ -0,0 +1,70 @@
+//! Host-signed transport credentials. These are never exposed as module JWTs.
+
+use crate::token_signing::TokenSigner;
+use jsonwebtoken::{Algorithm, DecodingKey, Validation};
+use serde::{Deserialize, Serialize};
+use spacetimedb_lib::Identity;
+use std::time::{SystemTime, UNIX_EPOCH};
+
+const ISSUER: &str = "spacetimedb-idc-v1";
+// Keep leaked credentials short-lived. The actor signs a fresh token for each
+// delivery attempt, including retries, so queued messages do not need long-lived tokens.
+// TODO(idc): Consider caching tokens per sender/receiver pair and extending their lifetime
+// to reduce signing overhead, refreshing them before expiry.
+const TOKEN_LIFETIME_SECS: u64 = 60;
+
+/// Signs transport credentials for calls between databases.
+pub trait IdcTokenSigner: Send + Sync {
+    /// Attest the sending database's identity to the receiver.
+    fn sign(&self, sender: Identity, receiver: Identity) -> anyhow::Result<String>;
+}
+
+impl<T: TokenSigner> IdcTokenSigner for T {
+    fn sign(&self, sender: Identity, receiver: Identity) -> anyhow::Result<String> {
+        sign(self, sender, receiver)
+    }
+}
+
+/// We can't use [`crate::identity::IncomingClaims`] here, client auth derives
+/// identity from `(iss, sub)`, and we don't store that pair for databases.
+/// Using the database identity as a new subject would give us a different identity.
+/// Instead, a trusted host signs the database's existing identity directly.
+/// The token is only used between hosts, not exposed through `sender_auth().jwt()`.
+#[derive(Serialize, Deserialize)]
+struct Claims {
+    iss: String,
+    aud: String,
+    sender: Identity,
+    iat: u64,
+    exp: u64,
+}
+
+/// Mint credentials from trusted host state, not a caller-supplied identity.
+pub fn sign(signer: &impl TokenSigner, sender: Identity, receiver: Identity) -> anyhow::Result<String> {
+    let iat = SystemTime::now().duration_since(UNIX_EPOCH)?.as_secs();
+    let claims = Claims {
+        iss: ISSUER.into(),
+        aud: receiver.to_hex().to_string(),
+        sender,
+        iat,
+        exp: iat + TOKEN_LIFETIME_SECS,
+    };
+    Ok(signer.sign(&claims)?)
+}
+
+/// Authenticate only with an explicitly trusted host key, never OIDC discovery.
+pub fn verify(key: &DecodingKey, token: &str, receiver: Identity) -> anyhow::Result<Identity> {
+    let mut validation = Validation::new(Algorithm::ES256);
+    validation.set_required_spec_claims(&["iss", "aud", "exp", "iat"]);
+    validation.set_issuer(&[ISSUER]);
+    validation.set_audience(&[receiver.to_hex().to_string()]);
+    validation.leeway = 0;
+    let claims = jsonwebtoken::decode::<Claims>(token, key, &validation)?.claims;
+    let now = SystemTime::now().duration_since(UNIX_EPOCH)?.as_secs();
+    anyhow::ensure!(claims.iat <= now && claims.exp > now, "invalid IDC token timestamps");
+    anyhow::ensure!(
+        claims.exp > claims.iat && claims.exp - claims.iat <= TOKEN_LIFETIME_SECS,
+        "invalid IDC token lifetime"
+    );
+    Ok(claims.sender)
+}

crates/auth/src/identity.rs

@@ -9,17 +9,18 @@ use std::time::SystemTime;
 
 #[derive(Debug, Clone)]
 pub struct ConnectionAuthCtx {
-    pub claims: SpacetimeIdentityClaims,
-    pub jwt_payload: Box<str>,
+    pub identity: Identity,
+    /// Client JWT claims, absent for host-authenticated database calls.
+    pub jwt_payload: Option<Box<str>>,
 }
 
 impl TryFrom<SpacetimeIdentityClaims> for ConnectionAuthCtx {
     type Error = anyhow::Error;
     fn try_from(claims: SpacetimeIdentityClaims) -> Result<Self, Self::Error> {
         let payload = serde_json::to_string(&claims).map_err(|e| anyhow::anyhow!("Failed to serialize claims: {e}"))?;
         Ok(ConnectionAuthCtx {
-            claims,
-            jwt_payload: payload.into(),
+            identity: claims.identity,
+            jwt_payload: Some(payload.into()),
         })
     }
 }

crates/auth/src/lib.rs

@@ -1 +1,3 @@
+pub mod idc;
 pub mod identity;
+pub mod token_signing;

crates/auth/src/token_signing.rs

@@ -0,0 +1,14 @@
+use jsonwebtoken::{errors::Error as JwtError, Algorithm, EncodingKey, Header};
+use serde::Serialize;
+
+/// Signs tokens without validating their claims.
+pub trait TokenSigner: Sync + Send {
+    /// Serialize the claims and sign a JWT with them as the payload.
+    fn sign<T: Serialize>(&self, claims: &T) -> Result<String, JwtError>;
+}
+
+impl TokenSigner for EncodingKey {
+    fn sign<T: Serialize>(&self, claims: &T) -> Result<String, JwtError> {
+        jsonwebtoken::encode(&Header::new(Algorithm::ES256), claims, self)
+    }
+}

crates/client-api-messages/src/idc.rs

@@ -0,0 +1,124 @@
+//! HTTP responses shared by the IDC receiver and sender.
+//!
+//! Only recorded outcomes finish a delivery. Authentication, routing, and other
+//! failures leave it pending; their response bodies are diagnostics, not outcomes.
+
+use bytes::Bytes;
+use http::StatusCode;
+use std::fmt;
+
+/// A receiver result or an HTTP error that leaves delivery pending.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum IdcResponse {
+    /// Empty for a newly committed call; replays may contain BSATN return bytes.
+    Success(Bytes),
+    /// A reducer or validation failure stored by the receiver. Retrying the same
+    /// message returns the same error. Hence, should be considered terminal.
+    /// The payload is UTF-8 error text, not BSATN.
+    TerminalError(Bytes),
+    /// A recorded energy failure, with a UTF-8 description, not BSATN.
+    BudgetExceeded(Bytes),
+    /// The sender already acknowledged this sequence. There is no result to replay.
+    AlreadyAcked,
+
+    /// A delivery failure the sender should retry. It may resolve on a later attempt.
+    DeliveryError(IdcDeliveryError),
+}
+
+impl IdcResponse {
+    pub fn into_http_parts(self) -> (StatusCode, Bytes) {
+        match self {
+            Self::Success(body) => (StatusCode::OK, body),
+            Self::TerminalError(body) => (StatusCode::UNPROCESSABLE_ENTITY, body),
+            Self::BudgetExceeded(body) => (StatusCode::PAYMENT_REQUIRED, body),
+            Self::AlreadyAcked => (StatusCode::ALREADY_REPORTED, Bytes::new()),
+            Self::DeliveryError(error) => error.into_http_parts(),
+        }
+    }
+
+    /// Decode a complete response body. Anything outside the IDC outcome contract
+    /// is a delivery failure, including authentication failures and redirects.
+    pub fn from_http_parts(status: StatusCode, body: Bytes) -> Self {
+        match status {
+            StatusCode::OK => Self::Success(body),
+            StatusCode::UNPROCESSABLE_ENTITY => Self::TerminalError(body),
+            StatusCode::PAYMENT_REQUIRED => Self::BudgetExceeded(body),
+            StatusCode::ALREADY_REPORTED => Self::AlreadyAcked,
+            _ => Self::DeliveryError(IdcDeliveryError { status, body }),
+        }
+    }
+}
+
+/// An HTTP response without a usable IDC outcome. Retry without acknowledging
+/// the sequence or calling `on_result`; the body must not control retry policy.
+#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
+pub struct IdcDeliveryError {
+    pub status: StatusCode,
+    /// Uninterpreted HTTP body, displayed as lossy UTF-8 for diagnostics only.
+    pub body: Bytes,
+}
+
+impl IdcDeliveryError {
+    pub fn into_http_parts(self) -> (StatusCode, Bytes) {
+        (self.status, self.body)
+    }
+}
+
+impl fmt::Display for IdcDeliveryError {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "HTTP {}: {}", self.status, String::from_utf8_lossy(&self.body))
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn outcomes_round_trip_without_changing_payloads() {
+        for (response, status) in [
+            (IdcResponse::Success(Bytes::new()), StatusCode::OK),
+            (IdcResponse::Success(Bytes::from_static(b"\x00\xff")), StatusCode::OK),
+            (
+                IdcResponse::TerminalError(Bytes::from_static(b"unauthorized sender")),
+                StatusCode::UNPROCESSABLE_ENTITY,
+            ),
+            (
+                IdcResponse::BudgetExceeded(Bytes::from_static(b"out of energy")),
+                StatusCode::PAYMENT_REQUIRED,
+            ),
+            (IdcResponse::AlreadyAcked, StatusCode::ALREADY_REPORTED),
+        ] {
+            let (actual_status, body) = response.clone().into_http_parts();
+            assert_eq!(actual_status, status);
+            assert_eq!(IdcResponse::from_http_parts(actual_status, body), response);
+        }
+    }
+
+    #[test]
+    fn other_responses_leave_delivery_pending_regardless_of_body() {
+        for status in [
+            201, 204, 301, 307, 400, 401, 403, 404, 408, 415, 425, 429, 500, 502, 503, 504, 530,
+        ] {
+            for body in [
+                Bytes::from_static(b"Ok"),
+                Bytes::from_static(b"permanent error"),
+                Bytes::new(),
+            ] {
+                let status = StatusCode::from_u16(status).unwrap();
+                let response = IdcResponse::from_http_parts(status, body.clone());
+                assert_eq!(
+                    response,
+                    IdcResponse::DeliveryError(IdcDeliveryError {
+                        status,
+                        body: body.clone()
+                    })
+                );
+                let (encoded_status, encoded_body) = response.clone().into_http_parts();
+                assert_eq!(encoded_status, status);
+                assert_eq!(encoded_body, body);
+                assert_eq!(IdcResponse::from_http_parts(encoded_status, encoded_body), response);
+            }
+        }
+    }
+}

crates/client-api-messages/src/lib.rs

@@ -2,6 +2,7 @@
 
 pub mod energy;
 pub mod http;
+pub mod idc;
 pub mod name;
 pub mod publish;
 pub mod websocket;

crates/client-api/src/auth.rs

@@ -107,8 +107,8 @@ impl SpacetimeAuth {
 impl From<SpacetimeAuth> for ConnectionAuthCtx {
     fn from(auth: SpacetimeAuth) -> Self {
         ConnectionAuthCtx {
-            claims: auth.claims,
-            jwt_payload: auth.jwt_payload.clone(),
+            identity: auth.claims.identity,
+            jwt_payload: Some(auth.jwt_payload),
         }
     }
 }
@@ -259,8 +259,7 @@ impl<TV: TokenValidator + Send + Sync> JwtKeyAuthProvider<TV> {
 
 impl<TV: TokenValidator + Send + Sync> TokenSigner for JwtKeyAuthProvider<TV> {
     fn sign<T: Serialize>(&self, claims: &T) -> Result<String, JwtError> {
-        let header = jsonwebtoken::Header::new(jsonwebtoken::Algorithm::ES256);
-        jsonwebtoken::encode(&header, &claims, &self.keys.private)
+        self.keys.sign(claims)
     }
 }
 
@@ -562,3 +561,33 @@ pub async fn anon_auth_middleware<S: ControlStateDelegate + NodeDelegate>(
     let resp = next.run(req).await;
     Ok((auth.into_headers(), resp))
 }
+
+/// The sending database identity verified by the IDC authentication middleware.
+#[derive(Clone, Copy)]
+pub struct IdcCallerIdentity(pub Identity);
+
+/// IDC authenticates a database, but must not expose its transport token to module code.
+pub async fn idc_auth_middleware<S: NodeDelegate>(
+    State(worker_ctx): State<S>,
+    mut req: Request,
+    next: Next,
+) -> axum::response::Result<impl IntoResponse> {
+    use crate::routes::database::ResolvedDatabase;
+    let receiver = req
+        .extensions()
+        .get::<ResolvedDatabase>()
+        .ok_or(http::StatusCode::INTERNAL_SERVER_ERROR)?
+        .0
+        .database_identity;
+    let bearer = req
+        .headers()
+        .typed_try_get::<headers::Authorization<authorization::Bearer>>()
+        .map_err(|_| http::StatusCode::UNAUTHORIZED)?
+        .ok_or(http::StatusCode::UNAUTHORIZED)?;
+    let key = jsonwebtoken::DecodingKey::from_ec_pem(worker_ctx.jwt_auth_provider().public_key_bytes())
+        .map_err(log_and_500)?;
+    let identity =
+        spacetimedb::auth::idc::verify(&key, bearer.token(), receiver).map_err(|_| http::StatusCode::UNAUTHORIZED)?;
+    req.extensions_mut().insert(IdcCallerIdentity(identity));
+    Ok(next.run(req).await)
+}