fix(audit-2026-06-11/H-H1): rate-limit email::send + per-message recipient cap

The Rhai SDK email::send path went straight to the SMTP relay with no
admission control. EmailRateLimiter lived in users_service.rs (gating
verification + password-reset emails) but was unreachable from the SDK
surface. Any anonymous-callable HTTP-route script could loop email::send
to burn the operator's SMTP quota or BCC-bomb thousands of recipients
per request.

Two new caps on EmailServiceImpl, both fire BEFORE message assembly +
SMTP connect:

1. Per-message recipient cap (to+cc+bcc combined). Default 20, override
   with PICLOUD_EMAIL_MAX_RECIPIENTS. New EmailError::TooManyRecipients
   variant. Closes the BCC-bomb amplification path.

2. EmailRateLimiter inlined into EmailServiceImpl:
   * Per-(app, recipient): RECIPIENT_BURST=5 / RECIPIENT_WINDOW=60min.
   * Per-app daily: APP_DAILY_CAP=200 / 24h.
   The structure mirrors users_service's private limiter; defense-in-
   depth redundancy is fine since the gates are identical and never
   conflict. New EmailError::RateLimited(&'static str) variant; the
   string names which bucket tripped so the operator can act.

users_service::map_email_error grows two cases (mapped to
EmailTransport for the script-facing error shape).

Tests:
* too_many_recipients_rejected — 30 recipients, default cap 20, rejected
  with TooManyRecipients.
* per_recipient_burst_caps_repeated_send_to_same_address — 6 sends to
  the same address; 6th returns RateLimited("per-recipient burst").

Audit ref: security_audit/09_external_integrations.md#f-ext-h-001.

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

crates/manager-core/src/email_service.rs

@@ -40,13 +40,20 @@ const ADDRESS_MAX_LEN: usize = 320;
 #[derive(Debug, Clone, Copy)]
 pub struct EmailConfig {
     pub max_message_bytes: usize,
+    /// Audit 2026-06-11 H-H1 — per-message recipient cap (to+cc+bcc).
+    /// Default 20; override via `PICLOUD_EMAIL_MAX_RECIPIENTS`.
+    pub max_recipients_per_message: usize,
 }
 
+/// Audit 2026-06-11 H-H1 — default per-message recipient cap.
+pub const DEFAULT_EMAIL_MAX_RECIPIENTS: usize = 20;
+
 impl EmailConfig {
     #[must_use]
     pub const fn conservative() -> Self {
         Self {
             max_message_bytes: DEFAULT_EMAIL_MAX_MESSAGE_BYTES,
+            max_recipients_per_message: DEFAULT_EMAIL_MAX_RECIPIENTS,
         }
     }
 
@@ -62,6 +69,15 @@ impl EmailConfig {
                 ),
             }
         }
+        if let Ok(v) = std::env::var("PICLOUD_EMAIL_MAX_RECIPIENTS") {
+            match v.trim().parse::<usize>() {
+                Ok(n) if n > 0 => c.max_recipients_per_message = n,
+                _ => tracing::warn!(
+                    value = %v,
+                    "ignoring invalid PICLOUD_EMAIL_MAX_RECIPIENTS (want a positive integer)"
+                ),
+            }
+        }
         c
     }
 }
@@ -72,6 +88,97 @@ impl Default for EmailConfig {
     }
 }
 
+/// Audit 2026-06-11 H-H1 — burst limiter on outbound email.
+///
+/// Two caps:
+/// 1. Per-`(app, recipient)`: `RECIPIENT_BURST` per `RECIPIENT_WINDOW`.
+/// 2. Per-app daily total: `APP_DAILY_CAP` per 24h.
+///
+/// Same shape as `users_service::EmailRateLimiter` but installed at the
+/// SMTP-relay boundary so it gates SDK `email::send` (the gap the audit
+/// flagged) AND the verification / password-reset paths (which already
+/// have their own limiter, so they bear a tiny redundant check —
+/// acceptable).
+#[derive(Debug)]
+struct EmailRateLimiter {
+    state: std::sync::Mutex<EmailRateState>,
+}
+
+const RECIPIENT_BURST: u32 = 5;
+const RECIPIENT_WINDOW: std::time::Duration = std::time::Duration::from_secs(3600);
+const APP_DAILY_CAP: u32 = 200;
+const APP_DAILY_WINDOW: std::time::Duration = std::time::Duration::from_secs(24 * 3600);
+
+#[derive(Debug, Default)]
+struct EmailRateState {
+    per_recipient: std::collections::HashMap<(picloud_shared::AppId, String), RateBucket>,
+    per_app: std::collections::HashMap<picloud_shared::AppId, RateBucket>,
+}
+
+#[derive(Debug, Clone, Copy)]
+struct RateBucket {
+    window_started_at: std::time::Instant,
+    count: u32,
+}
+
+impl EmailRateLimiter {
+    fn new() -> Self {
+        Self {
+            state: std::sync::Mutex::new(EmailRateState::default()),
+        }
+    }
+
+    /// Returns `Err(reason)` when either the per-recipient or per-app
+    /// bucket would overflow. `reason` is a static string identifying
+    /// which bucket so the operator can act.
+    fn try_consume(
+        &self,
+        app_id: picloud_shared::AppId,
+        recipient: &str,
+    ) -> Result<(), &'static str> {
+        let now = std::time::Instant::now();
+        let mut state = self.state.lock().expect("email rate state poisoned");
+        // Per-app daily cap.
+        {
+            let b = state.per_app.entry(app_id).or_insert(RateBucket {
+                window_started_at: now,
+                count: 0,
+            });
+            if now.duration_since(b.window_started_at) >= APP_DAILY_WINDOW {
+                *b = RateBucket {
+                    window_started_at: now,
+                    count: 0,
+                };
+            }
+            if b.count >= APP_DAILY_CAP {
+                return Err("per-app daily cap");
+            }
+        }
+        // Per-(app, recipient) burst.
+        let key = (app_id, recipient.to_string());
+        {
+            let b = state.per_recipient.entry(key).or_insert(RateBucket {
+                window_started_at: now,
+                count: 0,
+            });
+            if now.duration_since(b.window_started_at) >= RECIPIENT_WINDOW {
+                *b = RateBucket {
+                    window_started_at: now,
+                    count: 0,
+                };
+            }
+            if b.count >= RECIPIENT_BURST {
+                return Err("per-recipient burst");
+            }
+            b.count += 1;
+        }
+        if let Some(app_b) = state.per_app.get_mut(&app_id) {
+            app_b.count += 1;
+        }
+        Ok(())
+    }
+}
+
 /// TLS mode for the SMTP relay connection.
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub enum SmtpTls {
@@ -197,6 +304,7 @@ pub struct EmailServiceImpl {
     transport: Option<Arc<dyn EmailTransport>>,
     authz: Arc<dyn AuthzRepo>,
     config: EmailConfig,
+    rate_limiter: EmailRateLimiter,
 }
 
 impl EmailServiceImpl {
@@ -210,6 +318,7 @@ impl EmailServiceImpl {
             transport,
             authz,
             config,
+            rate_limiter: EmailRateLimiter::new(),
         }
     }
 
@@ -259,6 +368,37 @@ impl EmailService for EmailServiceImpl {
         let Some(transport) = self.transport.as_ref() else {
             return Err(EmailError::NotConfigured);
         };
+
+        // Audit 2026-06-11 H-H1 — per-message recipient cap. Counts the
+        // union of to+cc+bcc (script can stuff any of the three) and
+        // rejects amplification attempts before we even Argon2-build
+        // the MIME body or hit the SMTP relay.
+        let recipients: Vec<&str> = email
+            .to
+            .iter()
+            .chain(email.cc.iter())
+            .chain(email.bcc.iter())
+            .map(String::as_str)
+            .filter(|s| !s.trim().is_empty())
+            .collect();
+        if recipients.len() > self.config.max_recipients_per_message {
+            return Err(EmailError::TooManyRecipients {
+                limit: self.config.max_recipients_per_message,
+                actual: recipients.len(),
+            });
+        }
+
+        // Audit 2026-06-11 H-H1 — per-(app, recipient) + per-app-daily
+        // limit. Applied BEFORE message assembly + SMTP connect so a
+        // script in a tight loop can't burn the operator's SMTP quota
+        // or BCC-bomb thousands of recipients per request. One slot
+        // consumed per recipient.
+        for r in &recipients {
+            if let Err(reason) = self.rate_limiter.try_consume(cx.app_id, r) {
+                return Err(EmailError::RateLimited(reason));
+            }
+        }
+
         let message = build_message(&email)?;
         let formatted = message.formatted();
         if formatted.len() > self.config.max_message_bytes {
@@ -546,6 +686,7 @@ mod tests {
             Arc::new(DenyAuthz),
             EmailConfig {
                 max_message_bytes: 64,
+                max_recipients_per_message: DEFAULT_EMAIL_MAX_RECIPIENTS,
             },
         );
         let err = svc
@@ -594,4 +735,41 @@ mod tests {
         let err = svc.send(&cx, base_email()).await.unwrap_err();
         assert!(matches!(err, EmailError::Forbidden));
     }
+
+    #[tokio::test]
+    async fn too_many_recipients_rejected() {
+        // Audit 2026-06-11 H-H1 closure.
+        let (svc, _) = recording();
+        let many = (0..30)
+            .map(|i| format!("u{i}@example.com"))
+            .collect::<Vec<_>>();
+        let email = OutboundEmail {
+            to: many,
+            from: "alerts@myapp.com".into(),
+            subject: "x".into(),
+            text: Some("x".into()),
+            ..Default::default()
+        };
+        let err = svc.send(&anon(AppId::new()), email).await.unwrap_err();
+        assert!(
+            matches!(err, EmailError::TooManyRecipients { limit: 20, actual: 30 }),
+            "expected TooManyRecipients, got {err:?}"
+        );
+    }
+
+    #[tokio::test]
+    async fn per_recipient_burst_caps_repeated_send_to_same_address() {
+        // 5 sends to the same (app, recipient) → 5th is the last allowed;
+        // the 6th hits the burst limit. RECIPIENT_BURST = 5.
+        let (svc, _) = recording();
+        let app = AppId::new();
+        for _ in 0..5 {
+            svc.send(&anon(app), base_email()).await.unwrap();
+        }
+        let err = svc.send(&anon(app), base_email()).await.unwrap_err();
+        assert!(
+            matches!(err, EmailError::RateLimited("per-recipient burst")),
+            "expected RateLimited(per-recipient burst), got {err:?}"
+        );
+    }
 }

crates/manager-core/src/users_service.rs

@@ -427,7 +427,9 @@ fn map_email_error(e: EmailError) -> UsersError {
         EmailError::Forbidden => UsersError::Forbidden,
         EmailError::MissingField(_)
         | EmailError::InvalidAddress(_)
-        | EmailError::TooLarge { .. } => UsersError::EmailTransport(e.to_string()),
+        | EmailError::TooLarge { .. }
+        | EmailError::TooManyRecipients { .. }
+        | EmailError::RateLimited(_) => UsersError::EmailTransport(e.to_string()),
         EmailError::Transport(s) => UsersError::EmailTransport(s),
     }
 }

crates/shared/src/email.rs

@@ -74,6 +74,19 @@ pub enum EmailError {
     /// The SMTP relay rejected the message or the connection failed.
     #[error("email transport error: {0}")]
     Transport(String),
+
+    /// Audit 2026-06-11 H-H1 — the message recipient count exceeded
+    /// the per-message cap (default 20, override via
+    /// `PICLOUD_EMAIL_MAX_RECIPIENTS`).
+    #[error("too many recipients: {actual} exceeds the per-message cap of {limit}")]
+    TooManyRecipients { limit: usize, actual: usize },
+
+    /// Audit 2026-06-11 H-H1 — per-(app, recipient) or per-app burst
+    /// limit hit. The string names which bucket so operators can
+    /// distinguish "this user is being spammed" from "this app is
+    /// blowing through its SMTP quota."
+    #[error("email rate limit exceeded ({0})")]
+    RateLimited(&'static str),
 }
 
 /// Stub used by test harnesses that build a `Services` bundle without an