diff --git a/crates/manager-core/src/dispatcher.rs b/crates/manager-core/src/dispatcher.rs index 2f873d2..467f71e 100644 --- a/crates/manager-core/src/dispatcher.rs +++ b/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())); diff --git a/crates/manager-core/src/email_inbound_api.rs b/crates/manager-core/src/email_inbound_api.rs index 7ad1ac7..587f321 100644 --- a/crates/manager-core/src/email_inbound_api.rs +++ b/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; -/// 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>, +} + +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, pub outbox: Arc, 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, } 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] diff --git a/crates/manager-core/src/http_service.rs b/crates/manager-core/src/http_service.rs index 818c5f3..ecd4cfe 100644 --- a/crates/manager-core/src/http_service.rs +++ b/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>>) { + let captured: Arc>> = + 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>> = + 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" + ); + } } diff --git a/crates/manager-core/src/lib.rs b/crates/manager-core/src/lib.rs index 6439015..8e40413 100644 --- a/crates/manager-core/src/lib.rs +++ b/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, diff --git a/crates/manager-core/tests/expected_schema.txt b/crates/manager-core/tests/expected_schema.txt index ad16f93..68bc180 100644 --- a/crates/manager-core/tests/expected_schema.txt +++ b/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 diff --git a/crates/picloud/src/lib.rs b/crates/picloud/src/lib.rs index 8e7ab45..ce07f4f 100644 --- a/crates/picloud/src/lib.rs +++ b/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, diff --git a/crates/picloud/tests/email_inbound.rs b/crates/picloud/tests/email_inbound.rs index 5135c6e..a61c152 100644 --- a/crates/picloud/tests/email_inbound.rs +++ b/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::::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 { 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);