//! `POST /api/v1/email-inbound/{app_id}/{trigger_id}` — the inbound-email //! webhook receiver (v1.1.7). //! //! A configured provider (Mailgun / Postmark / SendGrid / SES) POSTs a //! normalized JSON message here; the receiver verifies the optional HMAC //! signature, builds a `TriggerEvent::Email`, and enqueues an outbox row //! the dispatcher picks up like any other async trigger. //! //! This is a PUBLIC endpoint (no admin auth) — the trigger URL itself, //! plus the per-trigger HMAC secret, are the security boundary. It is //! mounted OUTSIDE the `require_authenticated` layer. //! //! Status codes: //! * 202 — accepted + enqueued //! * 401 — HMAC required but missing/invalid //! * 404 — trigger missing, disabled, not `kind=email`, or app mismatch //! * 422 — body is not the expected JSON shape //! //! Only the generic provider-agnostic JSON shape is accepted in v1.1.7 //! (see [`InboundPayload`]); provider-specific unmarshallers are v1.2. use std::collections::HashMap; use std::sync::{Arc, Mutex}; use std::time::Instant; use axum::body::Bytes; use axum::extract::{Path, State}; use axum::http::{HeaderMap, StatusCode}; use axum::response::{IntoResponse, Json, Response}; use axum::routing::post; use axum::Router; use hmac::{Hmac, Mac}; use picloud_shared::{AppId, MasterKey, TriggerEvent, TriggerId}; use serde::Deserialize; use serde_json::json; use sha2::{Digest, Sha256}; use crate::outbox_repo::{NewOutboxRow, OutboxRepo, OutboxSourceKind}; use crate::secrets_repo::StoredSecret; use crate::secrets_service::open; use crate::trigger_repo::TriggerRepo; type HmacSha256 = Hmac; /// 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 { Router::new() .route( "/email-inbound/{app_id}/{trigger_id}", post(receive_inbound_email), ) .with_state(state) } /// The generic provider-agnostic inbound shape. Users configure their /// provider's webhook templating to POST this. `from` is required; /// everything else defaults. #[derive(Debug, Deserialize)] struct InboundPayload { from: String, #[serde(default)] to: Vec, #[serde(default)] cc: Vec, #[serde(default)] subject: String, #[serde(default)] text: Option, #[serde(default)] html: Option, #[serde(default)] message_id: Option, } async fn receive_inbound_email( State(s): State, Path((app_id, trigger_id)): Path<(AppId, TriggerId)>, headers: HeaderMap, body: Bytes, ) -> Result { // Resolve the trigger. 404 covers missing / wrong-kind / cross-app / // disabled — all "this URL doesn't address a live email trigger". let target = s .triggers .email_inbound_target(trigger_id) .await .map_err(|e| EmailInboundError::Backend(e.to_string()))? .ok_or(EmailInboundError::NotFound)?; if target.app_id != app_id || !target.enabled { return Err(EmailInboundError::NotFound); } // HMAC verification (only when the trigger has a secret configured). if let (Some(ct), Some(nonce)) = ( target.inbound_secret_encrypted.as_ref(), target.inbound_secret_nonce.as_ref(), ) { let secret = decrypt_secret(&s.master_key, ct, nonce)?; verify_signature(&headers, &body, secret.as_bytes(), &s.nonce_dedup)?; } // Parse the generic JSON shape. Malformed → 422. let payload: InboundPayload = serde_json::from_slice(&body).map_err(|e| EmailInboundError::Malformed(e.to_string()))?; let event = TriggerEvent::Email { from: payload.from, to: payload.to, cc: payload.cc, subject: payload.subject, text: payload.text, html: payload.html, received_at: chrono::Utc::now(), message_id: payload.message_id, }; let payload_json = serde_json::to_value(&event) .map_err(|e| EmailInboundError::Backend(format!("serialize event: {e}")))?; s.outbox .insert(NewOutboxRow { app_id, source_kind: OutboxSourceKind::Email, trigger_id: Some(trigger_id), script_id: Some(target.script_id), reply_to: None, payload: payload_json, origin_principal: Some(target.registered_by_principal), // Inbound email is the root of a trigger chain (depth 1). trigger_depth: 1, root_execution_id: None, }) .await .map_err(|e| EmailInboundError::Backend(e.to_string()))?; Ok(StatusCode::ACCEPTED) } /// Decrypt the stored inbound secret back to its raw string. It was /// sealed as a JSON string by the admin layer, so `open` yields a /// `Value::String`. fn decrypt_secret( master_key: &MasterKey, ciphertext: &[u8], nonce: &[u8], ) -> Result { let stored = StoredSecret { encrypted_value: ciphertext.to_vec(), nonce: nonce.to_vec(), }; let value = open(master_key, &stored).map_err(|_| { // Corrupted secret means we can't verify — fail closed (401). EmailInboundError::Unauthorized })?; value .as_str() .map(str::to_string) .ok_or(EmailInboundError::Unauthorized) } /// 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)?; 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)] pub enum EmailInboundError { #[error("trigger not found")] NotFound, #[error("invalid signature")] Unauthorized, #[error("malformed body: {0}")] Malformed(String), #[error("backend: {0}")] Backend(String), } impl IntoResponse for EmailInboundError { fn into_response(self) -> Response { let (status, body) = match &self { Self::NotFound => ( StatusCode::NOT_FOUND, json!({ "error": "trigger not found" }), ), Self::Unauthorized => ( StatusCode::UNAUTHORIZED, json!({ "error": "invalid or missing signature" }), ), Self::Malformed(m) => ( StatusCode::UNPROCESSABLE_ENTITY, json!({ "error": format!("malformed inbound email body: {m}") }), ), Self::Backend(e) => { tracing::error!(error = %e, "inbound email receiver backend error"); ( StatusCode::INTERNAL_SERVER_ERROR, json!({ "error": "internal error" }), ) } }; (status, Json(body)).into_response() } } #[cfg(test)] mod tests { //! Unit tests for the security-critical helpers (HMAC verify, secret //! round-trip, payload parsing). The full request flow — 202 / 401 / //! 404 / 422 / cross-app — is exercised end-to-end against a real //! Postgres in `crates/picloud/tests/email_inbound.rs`. use super::*; use crate::secrets_service::seal; use crate::secrets_service::DEFAULT_SECRET_MAX_VALUE_BYTES; 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, 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 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 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 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 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)); } #[test] fn secret_round_trips_through_seal_open() { let key = MasterKey::from_bytes([3u8; 32]); let (ct, nonce) = seal( &key, &serde_json::Value::String("provider-secret".into()), DEFAULT_SECRET_MAX_VALUE_BYTES, ) .unwrap(); let recovered = decrypt_secret(&key, &ct, &nonce).unwrap(); assert_eq!(recovered, "provider-secret"); let body = br#"{"from":"x@y.com"}"#; 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] fn payload_requires_from_but_defaults_rest() { let ok: Result = serde_json::from_slice(br#"{"from":"a@b.com"}"#); let p = ok.expect("from-only payload parses"); assert_eq!(p.from, "a@b.com"); assert!(p.to.is_empty() && p.cc.is_empty() && p.text.is_none()); // Missing `from` → malformed. let bad: Result = serde_json::from_slice(br#"{"subject":"hi"}"#); assert!(bad.is_err()); } }