fix(stage-1): audit High-severity backend correctness + CI unblocker

Closes 4 High-severity audit findings:

- describe_event broken for HTTP-async payloads: was reading source/op
  JSON fields that HttpDispatchPayload doesn't carry, producing empty
  source on DL rows. from_wire("") then fell back to Kv on replay, the
  dispatcher tried to resolve a non-existent trigger, and replay no-op'd
  silently. Now branches on OutboxSourceKind: HTTP rows emit
  "<method> <path>" + source="http"; Invoke rows emit "invoke_async" +
  source="invoke"; TriggerEvent payloads keep top-level field extraction.

- HTTP outbound Authorization leaked across cross-origin redirects.
  Manual redirect loop (Policy::none) reused header_map on every hop
  and only scrubbed Content-Type for POST->GET. Now compares
  url::Url::origin() across hops and strips Authorization,
  Proxy-Authorization, Cookie when origin changes — matching reqwest's
  default policy.

- F-S-010 inbound email HMAC had no replay protection. Signature was
  computed over body only with no timestamp or nonce, so captured POSTs
  were replayable indefinitely. Adds X-Picloud-Timestamp header bound
  into the HMAC input (signed string is ts || "." || body), 300s
  tolerance window, and a process-local LRU keyed on
  (ts, sha256(body)) with 600s TTL to catch within-window replays.
  Breaking change: webhook senders must include the timestamp header.

- expected_schema.txt re-blessed for migrations 0036-0039 (the previous
  snapshot stopped at 0035 so CI would have gone red on first run).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

crates/manager-core/src/dispatcher.rs

@@ -836,7 +836,7 @@ impl Dispatcher {
         }
 
         // Exhausted retries → dead-letter.
-        let (op, source) = describe_event(&row.payload);
+        let (op, source) = describe_event(row.source_kind, &row.payload);
         let now = Utc::now();
         let dl_id = match self
             .dead_letters
@@ -1051,18 +1051,57 @@ fn failure_kind_to_status(k: InboxFailureKind) -> u16 {
 
 /// `(op, source)` extracted from the outbox payload. Used to seed the
 /// `dead_letters` row when retries exhaust.
-fn describe_event(payload: &serde_json::Value) -> (String, String) {
-    let source = payload
-        .get("source")
-        .and_then(|v| v.as_str())
-        .unwrap_or("")
-        .to_string();
-    let op = payload
-        .get("op")
-        .and_then(|v| v.as_str())
-        .unwrap_or("")
-        .to_string();
-    (op, source)
+///
+/// The shape of the payload depends on `source_kind`:
+///   * `Http` rows carry an `HttpDispatchPayload` (`method`/`path`).
+///   * `Invoke` rows carry an ad-hoc invoke payload (`script_id`/`args`).
+///   * Every other kind carries a serialized [`TriggerEvent`], which has
+///     `source` and `op` fields at the JSON top level.
+///
+/// F-S-009 (audit fix): previously this function unconditionally read
+/// `source`/`op` from the JSON, which silently produced empty strings
+/// for HTTP rows. The DL row then got written with `source = ""` and
+/// `from_wire("")` returned `None` on replay, defaulting to
+/// `OutboxSourceKind::Kv` — replay rerouted through `resolve_trigger`,
+/// which failed because HTTP rows carry no `trigger_id`. Net result:
+/// HTTP-async DL replays no-op'd silently.
+fn describe_event(source_kind: OutboxSourceKind, payload: &serde_json::Value) -> (String, String) {
+    match source_kind {
+        OutboxSourceKind::Http => {
+            let method = payload
+                .get("method")
+                .and_then(|v| v.as_str())
+                .unwrap_or("?");
+            let path = payload.get("path").and_then(|v| v.as_str()).unwrap_or("?");
+            (
+                format!("{method} {path}"),
+                OutboxSourceKind::Http.as_str().to_string(),
+            )
+        }
+        OutboxSourceKind::Invoke => (
+            "invoke_async".to_string(),
+            OutboxSourceKind::Invoke.as_str().to_string(),
+        ),
+        OutboxSourceKind::Kv
+        | OutboxSourceKind::Docs
+        | OutboxSourceKind::DeadLetter
+        | OutboxSourceKind::Cron
+        | OutboxSourceKind::Files
+        | OutboxSourceKind::Pubsub
+        | OutboxSourceKind::Email => {
+            let source = payload
+                .get("source")
+                .and_then(|v| v.as_str())
+                .unwrap_or("")
+                .to_string();
+            let op = payload
+                .get("op")
+                .and_then(|v| v.as_str())
+                .unwrap_or("")
+                .to_string();
+            (op, source)
+        }
+    }
 }
 
 /// Compute backoff (ms) for the given attempt + policy + jitter.
@@ -1138,6 +1177,59 @@ mod tests {
         }
     }
 
+    #[test]
+    fn describe_event_http_uses_method_and_path() {
+        let payload = serde_json::json!({
+            "method": "POST",
+            "path": "/webhook",
+            "headers": {},
+            "body": null,
+            "params": {},
+            "query": {},
+            "rest": "",
+            "script_name": "hook",
+            "timeout_seconds": 30,
+        });
+        let (op, source) = describe_event(OutboxSourceKind::Http, &payload);
+        assert_eq!(op, "POST /webhook");
+        assert_eq!(source, "http");
+        // Round-trip via from_wire so the DL replay path resolves to the
+        // HTTP branch instead of falling back to Kv.
+        assert_eq!(
+            OutboxSourceKind::from_wire(&source),
+            Some(OutboxSourceKind::Http)
+        );
+    }
+
+    #[test]
+    fn describe_event_invoke_uses_constant_op() {
+        let payload = serde_json::json!({
+            "script_id": "00000000-0000-0000-0000-000000000000",
+            "args": {},
+        });
+        let (op, source) = describe_event(OutboxSourceKind::Invoke, &payload);
+        assert_eq!(op, "invoke_async");
+        assert_eq!(source, "invoke");
+        assert_eq!(
+            OutboxSourceKind::from_wire(&source),
+            Some(OutboxSourceKind::Invoke)
+        );
+    }
+
+    #[test]
+    fn describe_event_trigger_event_reads_top_level_fields() {
+        let payload = serde_json::json!({
+            "source": "kv",
+            "op": "set",
+            "collection": "users",
+            "key": "alice",
+            "value": {"name": "Alice"},
+        });
+        let (op, source) = describe_event(OutboxSourceKind::Kv, &payload);
+        assert_eq!(op, "set");
+        assert_eq!(source, "kv");
+    }
+
     #[test]
     fn classify_exec_error_covers_every_variant() {
         let parse = classify_exec_error(&ExecError::Parse("nope".into()));

crates/manager-core/src/email_inbound_api.rs

@@ -19,7 +19,9 @@
 //! Only the generic provider-agnostic JSON shape is accepted in v1.1.7
 //! (see [`InboundPayload`]); provider-specific unmarshallers are v1.2.
 
-use std::sync::Arc;
+use std::collections::HashMap;
+use std::sync::{Arc, Mutex};
+use std::time::Instant;
 
 use axum::body::Bytes;
 use axum::extract::{Path, State};
@@ -31,7 +33,7 @@ use hmac::{Hmac, Mac};
 use picloud_shared::{AppId, MasterKey, TriggerEvent, TriggerId};
 use serde::Deserialize;
 use serde_json::json;
-use sha2::Sha256;
+use sha2::{Digest, Sha256};
 
 use crate::outbox_repo::{NewOutboxRow, OutboxRepo, OutboxSourceKind};
 use crate::secrets_repo::StoredSecret;
@@ -40,15 +42,61 @@ use crate::trigger_repo::TriggerRepo;
 
 type HmacSha256 = Hmac<Sha256>;
 
-/// Header the provider's HMAC signature is read from. The signature is
-/// the lowercase hex of `HMAC-SHA256(inbound_secret, raw_body)`.
+/// Header the provider's HMAC signature is read from. Signature is
+/// the lowercase hex of `HMAC-SHA256(inbound_secret, timestamp || "." || raw_body)`.
 const SIGNATURE_HEADER: &str = "x-picloud-signature";
 
+/// Header the provider posts the request unix-seconds timestamp on. The
+/// timestamp is bound into the signature input (F-S-010) so captured
+/// POSTs can't be replayed indefinitely.
+const TIMESTAMP_HEADER: &str = "x-picloud-timestamp";
+
+/// Reject signatures whose timestamp is further than this from now. 5 min
+/// gives ample provider+network slack while bounding the replay window.
+const SIGNATURE_TIMESTAMP_TOLERANCE_SECS: i64 = 300;
+
+/// How long an accepted (timestamp, body-hash) pair is remembered to
+/// catch within-window replays. Slightly longer than tolerance so a
+/// replay near the edge of the window is still seen as duplicate.
+const NONCE_DEDUP_TTL_SECS: u64 = 600;
+
+/// Tiny in-memory dedup for accepted `(timestamp, sha256(body))` pairs.
+/// Bounded by [`NONCE_DEDUP_TTL_SECS`] — entries older than the TTL are
+/// pruned on each `record_or_reject` call.
+#[derive(Default)]
+pub struct InboundNonceDedup {
+    inner: Mutex<HashMap<(i64, [u8; 32]), Instant>>,
+}
+
+impl InboundNonceDedup {
+    #[must_use]
+    pub fn new() -> Self {
+        Self::default()
+    }
+
+    /// Return `Ok(())` if the pair was new; `Err(())` if already seen
+    /// within the TTL window.
+    fn record_or_reject(&self, ts: i64, body_hash: [u8; 32]) -> Result<(), ()> {
+        let now = Instant::now();
+        let ttl = std::time::Duration::from_secs(NONCE_DEDUP_TTL_SECS);
+        let mut map = self.inner.lock().expect("nonce dedup mutex poisoned");
+        map.retain(|_, recorded_at| now.duration_since(*recorded_at) < ttl);
+        if map.insert((ts, body_hash), now).is_some() {
+            return Err(());
+        }
+        Ok(())
+    }
+}
+
 #[derive(Clone)]
 pub struct EmailInboundState {
     pub triggers: Arc<dyn TriggerRepo>,
     pub outbox: Arc<dyn OutboxRepo>,
     pub master_key: MasterKey,
+    /// F-S-010: replay-dedup for accepted signed payloads. Process-local
+    /// — cluster mode (v1.3+) will need a shared store. Per-process is
+    /// fine for single-node MVP and dev.
+    pub nonce_dedup: Arc<InboundNonceDedup>,
 }
 
 pub fn email_inbound_router(state: EmailInboundState) -> Router {
@@ -104,7 +152,7 @@ async fn receive_inbound_email(
         target.inbound_secret_nonce.as_ref(),
     ) {
         let secret = decrypt_secret(&s.master_key, ct, nonce)?;
-        verify_signature(&headers, &body, secret.as_bytes())?;
+        verify_signature(&headers, &body, secret.as_bytes(), &s.nonce_dedup)?;
     }
 
     // Parse the generic JSON shape. Malformed → 422.
@@ -165,23 +213,51 @@ fn decrypt_secret(
         .ok_or(EmailInboundError::Unauthorized)
 }
 
-/// Constant-time HMAC-SHA256 verification of the body against the
-/// `X-Picloud-Signature` header (lowercase hex).
+/// Constant-time HMAC-SHA256 verification of `timestamp || "." || body`
+/// against the `X-Picloud-Signature` header (lowercase hex), with
+/// replay protection.
+///
+/// F-S-010: binding `X-Picloud-Timestamp` into the signed input and
+/// rejecting `|now - ts| > SIGNATURE_TIMESTAMP_TOLERANCE_SECS` bounds
+/// the replay window; an in-process LRU keyed on
+/// `(ts, sha256(body))` catches within-window replays. This is a
+/// breaking change versus pre-v1.1.10 — webhook senders must include
+/// the timestamp header and sign `ts || "." || body`.
 fn verify_signature(
     headers: &HeaderMap,
     body: &[u8],
     secret: &[u8],
+    nonce_dedup: &InboundNonceDedup,
 ) -> Result<(), EmailInboundError> {
     let provided_hex = headers
         .get(SIGNATURE_HEADER)
         .and_then(|h| h.to_str().ok())
         .ok_or(EmailInboundError::Unauthorized)?;
     let provided = hex::decode(provided_hex.trim()).map_err(|_| EmailInboundError::Unauthorized)?;
+    let ts_raw = headers
+        .get(TIMESTAMP_HEADER)
+        .and_then(|h| h.to_str().ok())
+        .ok_or(EmailInboundError::Unauthorized)?
+        .trim();
+    let ts: i64 = ts_raw
+        .parse()
+        .map_err(|_| EmailInboundError::Unauthorized)?;
+    let now = chrono::Utc::now().timestamp();
+    if (now - ts).abs() > SIGNATURE_TIMESTAMP_TOLERANCE_SECS {
+        return Err(EmailInboundError::Unauthorized);
+    }
     let mut mac =
         HmacSha256::new_from_slice(secret).map_err(|_| EmailInboundError::Unauthorized)?;
+    mac.update(ts_raw.as_bytes());
+    mac.update(b".");
     mac.update(body);
     mac.verify_slice(&provided)
-        .map_err(|_| EmailInboundError::Unauthorized)
+        .map_err(|_| EmailInboundError::Unauthorized)?;
+
+    let body_hash: [u8; 32] = Sha256::digest(body).into();
+    nonce_dedup
+        .record_or_reject(ts, body_hash)
+        .map_err(|()| EmailInboundError::Unauthorized)
 }
 
 #[derive(Debug, thiserror::Error)]
@@ -234,45 +310,96 @@ mod tests {
     use crate::secrets_service::seal;
     use crate::secrets_service::DEFAULT_SECRET_MAX_VALUE_BYTES;
 
-    fn sign(secret: &[u8], body: &[u8]) -> String {
+    fn sign(secret: &[u8], ts: i64, body: &[u8]) -> String {
         let mut mac = HmacSha256::new_from_slice(secret).unwrap();
+        mac.update(ts.to_string().as_bytes());
+        mac.update(b".");
         mac.update(body);
         hex::encode(mac.finalize().into_bytes())
     }
 
-    fn headers_with_sig(sig: &str) -> HeaderMap {
+    fn headers_with_sig(sig: &str, ts: i64) -> HeaderMap {
         let mut h = HeaderMap::new();
         h.insert(SIGNATURE_HEADER, sig.parse().unwrap());
+        h.insert(TIMESTAMP_HEADER, ts.to_string().parse().unwrap());
         h
     }
 
+    fn now_ts() -> i64 {
+        chrono::Utc::now().timestamp()
+    }
+
     #[test]
     fn valid_signature_verifies() {
         let secret = b"shhh";
         let body = br#"{"from":"a@b.com"}"#;
-        let sig = sign(secret, body);
-        assert!(verify_signature(&headers_with_sig(&sig), body, secret).is_ok());
+        let ts = now_ts();
+        let sig = sign(secret, ts, body);
+        let dedup = InboundNonceDedup::new();
+        assert!(verify_signature(&headers_with_sig(&sig, ts), body, secret, &dedup).is_ok());
     }
 
     #[test]
     fn wrong_signature_rejected() {
         let body = br#"{"from":"a@b.com"}"#;
-        let sig = sign(b"shhh", body);
-        let err = verify_signature(&headers_with_sig(&sig), body, b"different").unwrap_err();
+        let ts = now_ts();
+        let sig = sign(b"shhh", ts, body);
+        let dedup = InboundNonceDedup::new();
+        let err =
+            verify_signature(&headers_with_sig(&sig, ts), body, b"different", &dedup).unwrap_err();
         assert!(matches!(err, EmailInboundError::Unauthorized));
     }
 
     #[test]
     fn missing_signature_header_rejected() {
-        let err = verify_signature(&HeaderMap::new(), b"body", b"secret").unwrap_err();
+        let mut h = HeaderMap::new();
+        h.insert(TIMESTAMP_HEADER, now_ts().to_string().parse().unwrap());
+        let dedup = InboundNonceDedup::new();
+        let err = verify_signature(&h, b"body", b"secret", &dedup).unwrap_err();
+        assert!(matches!(err, EmailInboundError::Unauthorized));
+    }
+
+    #[test]
+    fn missing_timestamp_header_rejected() {
+        let mut h = HeaderMap::new();
+        h.insert(SIGNATURE_HEADER, "deadbeef".parse().unwrap());
+        let dedup = InboundNonceDedup::new();
+        let err = verify_signature(&h, b"body", b"secret", &dedup).unwrap_err();
+        assert!(matches!(err, EmailInboundError::Unauthorized));
+    }
+
+    #[test]
+    fn timestamp_outside_tolerance_rejected() {
+        let secret = b"shhh";
+        let body = br#"{"from":"a@b.com"}"#;
+        let stale_ts = now_ts() - (SIGNATURE_TIMESTAMP_TOLERANCE_SECS + 60);
+        let sig = sign(secret, stale_ts, body);
+        let dedup = InboundNonceDedup::new();
+        let err =
+            verify_signature(&headers_with_sig(&sig, stale_ts), body, secret, &dedup).unwrap_err();
+        assert!(matches!(err, EmailInboundError::Unauthorized));
+    }
+
+    #[test]
+    fn replay_within_window_rejected() {
+        let secret = b"shhh";
+        let body = br#"{"from":"a@b.com"}"#;
+        let ts = now_ts();
+        let sig = sign(secret, ts, body);
+        let dedup = InboundNonceDedup::new();
+        assert!(verify_signature(&headers_with_sig(&sig, ts), body, secret, &dedup).is_ok());
+        let err = verify_signature(&headers_with_sig(&sig, ts), body, secret, &dedup).unwrap_err();
         assert!(matches!(err, EmailInboundError::Unauthorized));
     }
 
     #[test]
     fn tampered_body_fails_verification() {
         let secret = b"shhh";
-        let sig = sign(secret, b"original");
-        let err = verify_signature(&headers_with_sig(&sig), b"tampered", secret).unwrap_err();
+        let ts = now_ts();
+        let sig = sign(secret, ts, b"original");
+        let dedup = InboundNonceDedup::new();
+        let err =
+            verify_signature(&headers_with_sig(&sig, ts), b"tampered", secret, &dedup).unwrap_err();
         assert!(matches!(err, EmailInboundError::Unauthorized));
     }
 
@@ -287,10 +414,17 @@ mod tests {
         .unwrap();
         let recovered = decrypt_secret(&key, &ct, &nonce).unwrap();
         assert_eq!(recovered, "provider-secret");
-        // And a signature made with the recovered secret verifies.
         let body = br#"{"from":"x@y.com"}"#;
-        let sig = sign(recovered.as_bytes(), body);
-        assert!(verify_signature(&headers_with_sig(&sig), body, recovered.as_bytes()).is_ok());
+        let ts = now_ts();
+        let sig = sign(recovered.as_bytes(), ts, body);
+        let dedup = InboundNonceDedup::new();
+        assert!(verify_signature(
+            &headers_with_sig(&sig, ts),
+            body,
+            recovered.as_bytes(),
+            &dedup
+        )
+        .is_ok());
     }
 
     #[test]

crates/manager-core/src/http_service.rs

@@ -29,7 +29,10 @@ use std::time::Duration;
 
 use async_trait::async_trait;
 use picloud_shared::{HttpError, HttpRequest, HttpResponse, HttpService, SdkCallCx};
-use reqwest::header::{HeaderMap, HeaderName, HeaderValue, CONTENT_TYPE, LOCATION, USER_AGENT};
+use reqwest::header::{
+    HeaderMap, HeaderName, HeaderValue, AUTHORIZATION, CONTENT_TYPE, COOKIE, LOCATION,
+    PROXY_AUTHORIZATION, USER_AGENT,
+};
 use reqwest::{Client, Method, StatusCode};
 
 use crate::authz::{self, AuthzRepo, Capability};
@@ -244,10 +247,24 @@ impl HttpServiceImpl {
                     let loc_str = loc.to_str().map_err(|_| {
                         HttpError::Backend("redirect Location not valid UTF-8".into())
                     })?;
-                    current = current
+                    let next = current
                         .join(loc_str)
                         .map_err(|e| HttpError::InvalidUrl(format!("redirect target: {e}")))?;
 
+                    // Cross-origin sensitive-header scrub. reqwest's
+                    // default redirect policy drops Authorization,
+                    // Proxy-Authorization, and Cookie when a redirect
+                    // crosses origin (scheme/host/port). Manual follow
+                    // has to do the same; without this, a script's
+                    // bearer token leaks to whatever target a 302
+                    // points at on another origin.
+                    if next.origin() != current.origin() {
+                        header_map.remove(AUTHORIZATION);
+                        header_map.remove(PROXY_AUTHORIZATION);
+                        header_map.remove(COOKIE);
+                    }
+                    current = next;
+
                     // 303 always → GET; 301/302 historically downgrade
                     // POST→GET (matches browsers). 307/308 preserve.
                     if matches!(status.as_u16(), 301..=303) {
@@ -792,4 +809,137 @@ mod tests {
             .unwrap();
         assert_eq!(resp.status, 200);
     }
+
+    /// Test fixture: capture every received Authorization header value
+    /// across hops. Server B (the redirect target) is the one we care
+    /// about — Authorization on B means the bearer leaked across origin.
+    async fn spawn_pair_redirect(
+        b_addr_template: impl Fn(SocketAddr) -> String + Send + Sync + 'static,
+    ) -> (SocketAddr, SocketAddr, Arc<std::sync::Mutex<Vec<String>>>) {
+        let captured: Arc<std::sync::Mutex<Vec<String>>> =
+            Arc::new(std::sync::Mutex::new(Vec::new()));
+
+        // Server B: records Authorization header values, replies 200.
+        let captured_b = captured.clone();
+        let listener_b = TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let addr_b = listener_b.local_addr().unwrap();
+        tokio::spawn(async move {
+            loop {
+                let Ok((mut sock, _)) = listener_b.accept().await else {
+                    break;
+                };
+                let mut buf = vec![0u8; 65536];
+                let n = sock.read(&mut buf).await.unwrap_or(0);
+                let request = String::from_utf8_lossy(&buf[..n]).to_string();
+                let mut auth = String::new();
+                for line in request.lines() {
+                    if line.to_ascii_lowercase().starts_with("authorization:") {
+                        auth = line[14..].trim().to_string();
+                        break;
+                    }
+                }
+                captured_b.lock().unwrap().push(auth);
+                let _ = sock.write_all(&ok_response("on-b", "text/plain")).await;
+            }
+        });
+
+        // Server A: redirects to B (template chooses same-origin "/b" or
+        // a full cross-origin URL).
+        let target = b_addr_template(addr_b);
+        let listener_a = TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let addr_a = listener_a.local_addr().unwrap();
+        tokio::spawn(async move {
+            loop {
+                let Ok((mut sock, _)) = listener_a.accept().await else {
+                    break;
+                };
+                let mut buf = vec![0u8; 4096];
+                let _ = sock.read(&mut buf).await;
+                let body = format!(
+                    "HTTP/1.1 302 Found\r\nLocation: {target}\r\nContent-Length: 0\r\nConnection: close\r\n\r\n"
+                );
+                let _ = sock.write_all(body.as_bytes()).await;
+            }
+        });
+
+        (addr_a, addr_b, captured)
+    }
+
+    #[tokio::test]
+    async fn authorization_scrubbed_on_cross_origin_redirect() {
+        // A → B where A != B (different port = different origin). The
+        // request carries an Authorization header; after the redirect,
+        // Server B must NOT see Authorization.
+        let (addr_a, _addr_b, captured) =
+            spawn_pair_redirect(|b| format!("http://{b}/landing")).await;
+        let svc = dev_service(Arc::new(AllowAuthz));
+        let mut r = req("GET", format!("http://{addr_a}/start"));
+        r.headers
+            .insert("Authorization".into(), "Bearer secret-token".into());
+        let resp = svc.request(&anon_cx(), r).await.unwrap();
+        assert_eq!(resp.status, 200);
+        let captured = captured.lock().unwrap();
+        assert_eq!(
+            captured.len(),
+            1,
+            "server B should have been hit exactly once"
+        );
+        assert!(
+            captured[0].is_empty(),
+            "Authorization leaked to cross-origin redirect target: {:?}",
+            captured[0]
+        );
+    }
+
+    #[tokio::test]
+    async fn authorization_preserved_on_same_origin_redirect() {
+        // Server that 302s to itself preserves origin → Authorization
+        // must survive the hop.
+        let captured: Arc<std::sync::Mutex<Vec<String>>> =
+            Arc::new(std::sync::Mutex::new(Vec::new()));
+        let captured_clone = captured.clone();
+        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let addr = listener.local_addr().unwrap();
+        tokio::spawn(async move {
+            let mut hits = 0u32;
+            loop {
+                let Ok((mut sock, _)) = listener.accept().await else {
+                    break;
+                };
+                let mut buf = vec![0u8; 4096];
+                let n = sock.read(&mut buf).await.unwrap_or(0);
+                let request = String::from_utf8_lossy(&buf[..n]).to_string();
+                let mut auth = String::new();
+                for line in request.lines() {
+                    if line.to_ascii_lowercase().starts_with("authorization:") {
+                        auth = line[14..].trim().to_string();
+                        break;
+                    }
+                }
+                captured_clone.lock().unwrap().push(auth);
+                hits += 1;
+                let body = if hits == 1 {
+                    // Same-origin redirect (relative Location).
+                    "HTTP/1.1 302 Found\r\nLocation: /next\r\nContent-Length: 0\r\nConnection: close\r\n\r\n".to_string()
+                } else {
+                    String::from_utf8(ok_response("done", "text/plain")).unwrap()
+                };
+                let _ = sock.write_all(body.as_bytes()).await;
+            }
+        });
+
+        let svc = dev_service(Arc::new(AllowAuthz));
+        let mut r = req("GET", format!("http://{addr}/start"));
+        r.headers
+            .insert("Authorization".into(), "Bearer keep-me".into());
+        let resp = svc.request(&anon_cx(), r).await.unwrap();
+        assert_eq!(resp.status, 200);
+        let captured = captured.lock().unwrap();
+        assert_eq!(captured.len(), 2);
+        assert_eq!(captured[0], "Bearer keep-me", "first hop missing auth");
+        assert_eq!(
+            captured[1], "Bearer keep-me",
+            "same-origin hop should retain auth"
+        );
+    }
 }

crates/manager-core/src/lib.rs

@@ -145,7 +145,9 @@ pub use dead_letters_api::{dead_letters_router, DeadLettersApiError, DeadLetters
 pub use dispatcher::{compute_backoff, Dispatcher, DispatcherError};
 pub use docs_repo::{DocsRepo, DocsRepoError, PostgresDocsRepo};
 pub use docs_service::DocsServiceImpl;
-pub use email_inbound_api::{email_inbound_router, EmailInboundError, EmailInboundState};
+pub use email_inbound_api::{
+    email_inbound_router, EmailInboundError, EmailInboundState, InboundNonceDedup,
+};
 pub use email_service::{
     EmailConfig, EmailServiceImpl, EmailTransport, LettreEmailTransport, SmtpConfig, SmtpTls,
     DEFAULT_EMAIL_MAX_MESSAGE_BYTES,

crates/manager-core/tests/expected_schema.txt

@@ -371,6 +371,7 @@ indexes on app_user_invitations:
   app_user_invitations_pkey: public.app_user_invitations USING btree (id)
   app_user_invitations_token_hash_key: public.app_user_invitations USING btree (token_hash)
   idx_app_user_invitations_app_pending: public.app_user_invitations USING btree (app_id) WHERE (accepted_at IS NULL)
+  idx_app_user_invitations_unique_pending: public.app_user_invitations USING btree (app_id, lower(email)) WHERE (accepted_at IS NULL)
 
 indexes on app_user_password_resets:
   app_user_password_resets_pkey: public.app_user_password_resets USING btree (token_hash)
@@ -396,7 +397,6 @@ indexes on apps:
 
 indexes on cron_trigger_details:
   cron_trigger_details_pkey: public.cron_trigger_details USING btree (trigger_id)
-  idx_cron_triggers_due: public.cron_trigger_details USING btree (last_fired_at)
 
 indexes on dead_letter_trigger_details:
   dead_letter_trigger_details_pkey: public.dead_letter_trigger_details USING btree (trigger_id)
@@ -482,6 +482,7 @@ indexes on topics:
 indexes on triggers:
   idx_triggers_app_kind_enabled: public.triggers USING btree (app_id, kind) WHERE (enabled = true)
   idx_triggers_app_pubsub_enabled: public.triggers USING btree (app_id, kind) WHERE ((enabled = true) AND (kind = 'pubsub'::text))
+  idx_triggers_kind_enabled: public.triggers USING btree (kind) WHERE (enabled = true)
   triggers_pkey: public.triggers USING btree (id)
 
 ## constraints
@@ -518,6 +519,7 @@ constraints on app_members:
   [PRIMARY KEY] app_members_pkey: PRIMARY KEY (app_id, user_id)
 
 constraints on app_secrets:
+  [CHECK] app_secrets_realtime_signing_key_pair: CHECK (((realtime_signing_key_encrypted IS NULL) = (realtime_signing_key_nonce IS NULL)))
   [FOREIGN KEY] app_secrets_app_id_fkey: FOREIGN KEY (app_id) REFERENCES apps(id) ON DELETE CASCADE
   [PRIMARY KEY] app_secrets_pkey: PRIMARY KEY (app_id)
 
@@ -580,6 +582,7 @@ constraints on docs_trigger_details:
   [PRIMARY KEY] docs_trigger_details_pkey: PRIMARY KEY (trigger_id)
 
 constraints on email_trigger_details:
+  [CHECK] email_trigger_details_inbound_secret_pair: CHECK (((inbound_secret_encrypted IS NULL) = (inbound_secret_nonce IS NULL)))
   [FOREIGN KEY] email_trigger_details_trigger_id_fkey: FOREIGN KEY (trigger_id) REFERENCES triggers(id) ON DELETE CASCADE
   [PRIMARY KEY] email_trigger_details_pkey: PRIMARY KEY (trigger_id)
 
@@ -698,3 +701,7 @@ constraints on triggers:
   0033: topics auth mode session
   0034: queue messages
   0035: queue triggers
+  0036: index triggers kind enabled
+  0037: drop idx cron triggers due
+  0038: email realtime secret check
+  0039: app user invitations unique pending

crates/picloud/src/lib.rs

@@ -18,19 +18,20 @@ use picloud_manager_core::{
     ApiKeyRepository, ApiKeysState, AppDomainRepository, AppMembersRepository, AppMembersState,
     AppRepository, AppsState, AuthState, AuthzRepo, DeadLetterRepo, DeadLettersState, Dispatcher,
     DocsServiceImpl, EmailInboundState, EmailServiceImpl, FilesAdminState, FilesConfig,
-    FilesServiceImpl, FsFilesRepo, HttpConfig, HttpServiceImpl, KvServiceImpl, OutboxEventEmitter,
-    OutboxRepo, PostgresAbandonedRepo, PostgresAdminSessionRepository, PostgresAdminUserRepository,
-    PostgresApiKeyRepository, PostgresAppDomainRepository, PostgresAppMembersRepository,
-    PostgresAppRepository, PostgresAppSecretsRepo, PostgresAppUserInvitationRepo,
-    PostgresAppUserPasswordResetRepo, PostgresAppUserRepository, PostgresAppUserRoleRepo,
-    PostgresAppUserSessionRepository, PostgresAppUserVerificationRepo, PostgresDeadLetterRepo,
-    PostgresDeadLetterService, PostgresDocsRepo, PostgresExecutionLogRepository,
-    PostgresExecutionLogSink, PostgresKvRepo, PostgresOutboxRepo, PostgresPubsubRepo,
-    PostgresRouteRepository, PostgresScriptRepository, PostgresSecretsRepo, PostgresTopicRepo,
-    PostgresTriggerRepo, PrincipalResolver, PubsubServiceImpl, RealtimeAuthorityImpl, RepoResolver,
-    RouteAdminState, RouteRepository, SandboxCeiling, ScriptRepository, SecretsConfig,
-    SecretsServiceImpl, SecretsState, SubscriberTokenConfig, TopicRepo, TopicsState, TriggerConfig,
-    TriggerRepo, TriggersState, UsersServiceConfig, UsersServiceImpl,
+    FilesServiceImpl, FsFilesRepo, HttpConfig, HttpServiceImpl, InboundNonceDedup, KvServiceImpl,
+    OutboxEventEmitter, OutboxRepo, PostgresAbandonedRepo, PostgresAdminSessionRepository,
+    PostgresAdminUserRepository, PostgresApiKeyRepository, PostgresAppDomainRepository,
+    PostgresAppMembersRepository, PostgresAppRepository, PostgresAppSecretsRepo,
+    PostgresAppUserInvitationRepo, PostgresAppUserPasswordResetRepo, PostgresAppUserRepository,
+    PostgresAppUserRoleRepo, PostgresAppUserSessionRepository, PostgresAppUserVerificationRepo,
+    PostgresDeadLetterRepo, PostgresDeadLetterService, PostgresDocsRepo,
+    PostgresExecutionLogRepository, PostgresExecutionLogSink, PostgresKvRepo, PostgresOutboxRepo,
+    PostgresPubsubRepo, PostgresRouteRepository, PostgresScriptRepository, PostgresSecretsRepo,
+    PostgresTopicRepo, PostgresTriggerRepo, PrincipalResolver, PubsubServiceImpl,
+    RealtimeAuthorityImpl, RepoResolver, RouteAdminState, RouteRepository, SandboxCeiling,
+    ScriptRepository, SecretsConfig, SecretsServiceImpl, SecretsState, SubscriberTokenConfig,
+    TopicRepo, TopicsState, TriggerConfig, TriggerRepo, TriggersState, UsersServiceConfig,
+    UsersServiceImpl,
 };
 use picloud_orchestrator_core::realtime::DEFAULT_GC_INTERVAL_SECS;
 use picloud_orchestrator_core::routing::{AppDomainTable, RouteTable};
@@ -453,6 +454,7 @@ pub async fn build_app(
         triggers: trigger_repo,
         outbox: outbox_repo.clone(),
         master_key: master_key.clone(),
+        nonce_dedup: Arc::new(InboundNonceDedup::new()),
     };
     let dead_letters_state = DeadLettersState {
         repo: dl_repo,

crates/picloud/tests/email_inbound.rs

@@ -112,12 +112,18 @@ async fn create_email_trigger(
     created["id"].as_str().expect("trigger id").to_string()
 }
 
-fn sign(secret: &str, body: &str) -> String {
+fn sign(secret: &str, ts: i64, body: &str) -> String {
     let mut mac = Hmac::<Sha256>::new_from_slice(secret.as_bytes()).expect("hmac key");
+    mac.update(ts.to_string().as_bytes());
+    mac.update(b".");
     mac.update(body.as_bytes());
     hex::encode(mac.finalize().into_bytes())
 }
 
+fn now_ts() -> i64 {
+    chrono::Utc::now().timestamp()
+}
+
 async fn poll_marker(pool: &PgPool, app_id: &str) -> Option<Value> {
     let app_uuid = Uuid::parse_str(app_id).expect("app uuid");
     for _ in 0..100 {
@@ -148,10 +154,12 @@ async fn signed_post_accepts_and_fires_handler() {
     let handler = create_script(&server, &app_id, "eml-handler", MARKER_HANDLER).await;
     let trigger = create_email_trigger(&server, &app_id, &handler, Some("topsecret")).await;
 
-    let sig = sign("topsecret", BODY);
+    let ts = now_ts();
+    let sig = sign("topsecret", ts, BODY);
     server
         .post(&format!("/api/v1/email-inbound/{app_id}/{trigger}"))
         .add_header("x-picloud-signature", sig)
+        .add_header("x-picloud-timestamp", ts.to_string())
         .text(BODY)
         .await
         .assert_status(axum::http::StatusCode::ACCEPTED);
@@ -197,9 +205,11 @@ async fn wrong_signature_is_401() {
     let handler = create_script(&server, &app_id, "h", MARKER_HANDLER).await;
     let trigger = create_email_trigger(&server, &app_id, &handler, Some("topsecret")).await;
 
+    let ts = now_ts();
     server
         .post(&format!("/api/v1/email-inbound/{app_id}/{trigger}"))
-        .add_header("x-picloud-signature", sign("WRONG", BODY))
+        .add_header("x-picloud-signature", sign("WRONG", ts, BODY))
+        .add_header("x-picloud-timestamp", ts.to_string())
         .text(BODY)
         .await
         .assert_status(axum::http::StatusCode::UNAUTHORIZED);