PiCloud/crates/manager-core/src/email_inbound_api.rs

//! `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<Sha256>;

/// 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 {
    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<String>,
    #[serde(default)]
    cc: Vec<String>,
    #[serde(default)]
    subject: String,
    #[serde(default)]
    text: Option<String>,
    #[serde(default)]
    html: Option<String>,
    #[serde(default)]
    message_id: Option<String>,
}

async fn receive_inbound_email(
    State(s): State<EmailInboundState>,
    Path((app_id, trigger_id)): Path<(AppId, TriggerId)>,
    headers: HeaderMap,
    body: Bytes,
) -> Result<StatusCode, EmailInboundError> {
    // 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<String, EmailInboundError> {
    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<InboundPayload, _> = 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<InboundPayload, _> = serde_json::from_slice(br#"{"subject":"hi"}"#);
        assert!(bad.is_err());
    }
}